commit 985663ff1f52ac5daf36ea128371ffa753702b7d
parent d822883e01144db69b0c73f9c3dc013f65991dca
Author: triesap <tyson@radroots.org>
Date: Tue, 22 Sep 2026 16:54:30 +0000
tests: prove strict scripted fixtures and parent-owned lifecycle
- Add explicit ordered exchange scripts with strict method/path/header/body
and scripted status/header/body/delay matching for both providers.
- Enforce bounded lexical framing, exact route/auth validation and bounded
extra-exchange/missing-exchange accounting with case-specific reasons.
- Add parent-enforced deadlines, bounded output and exception-safe reaping in
a shared test-only POSIX lifecycle module for stubs and run_stdio_entrypoint.
- Report bounded phase/case/reason plus owned identity for cause-specific
negative controls; existing 24 D16 stdio signatures remain unchanged.
Diffstat:
8 files changed, 2661 insertions(+), 621 deletions(-)
diff --git a/tests/jev_provider_helper.mojo b/tests/jev_provider_helper.mojo
@@ -1,61 +1,70 @@
-import std.os
-from std.ffi import c_int, c_size_t, c_ssize_t, c_uint, external_call
-from std.os import Pipe, Process
-from std.sys._libc import close
-
-from flare.net import SocketAddr
-from flare.tcp import TcpListener, TcpStream
-from flare.utils import usleep
-
-from strict_fixture import ConnectionReader, FramedRequest, bearer_token
-
-
-def _dup2(oldfd: c_int, newfd: c_int) -> c_int:
- return external_call["dup2", c_int](oldfd, newfd)
-
-
-@always_inline
-def _fork() -> c_int:
- return external_call["fork", c_int]()
-
-
-@always_inline
-def _alarm(seconds: c_uint) -> c_uint:
- return external_call["alarm", c_uint](seconds)
+"""Strict scripted Jev provider HTTP fixture (ADR-0012 D29 / ADR-0014 D33).
+Same parent-owned lifecycle and phase/case/reason reporting as the MaxLocal
+fixture: parent-enforced startup/read/write/wait deadlines (FX06), bounded
+reporting (FX07) and exception-safe reaping (FX08).
+"""
-@always_inline
-def _exit_child(code: c_int):
- _ = external_call["_exit", c_int](code)
-
-
-def _write(fd: Int, text: String):
- _ = external_call["write", c_ssize_t](
- fd, text.as_bytes().unsafe_ptr(), c_size_t(text.byte_length())
- )
-
-
-def _read_pipe_line(mut pipe: Pipe) raises -> String:
- var buffer = InlineArray[Byte, 1](fill=0)
- var output = String("")
- while True:
- var read = pipe.read_bytes(Span(buffer))
- if read == 0:
- break
- var chunk = String(
- from_utf8=Span(ptr=buffer.unsafe_ptr(), length=Int(read))
- )
- if chunk == "\n":
- break
- output += chunk
- return output^
-
-
-def _json_string(value: String) -> String:
- return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
+from std.collections import List
+from flare.net import SocketAddr
+from flare.tcp import TcpListener
+from flare.utils import usleep
-def _analysis() -> String:
+from parent_lifecycle import (
+ FIXTURE_DEFAULT_DEADLINE_MS,
+ TERMINATION_GRACE_MS,
+ ProcessStatus,
+ child_exit,
+ close_fd,
+ dup2_fd,
+ fork_pid,
+ make_pipe,
+ read_line_bounded,
+ set_alarm,
+ terminate_owned,
+ wait_bounded,
+ write_raw,
+)
+from strict_fixture import (
+ STRICT_MAX_REPORT_BYTES,
+ STRICT_COMPLETION_GRACE_MS,
+ ConnectionReader,
+ ExchangeScript,
+ FramedRequest,
+ ServeReport,
+ authorization_reason,
+ exchange_script,
+ parse_report,
+ render_response,
+ report_line,
+ serve_scripts,
+ validate_convenience,
+ verify_exchange,
+)
+
+
+comptime STUB_ALARM_SECONDS = 20
+comptime JEV_INTENDED_PATH = "/v1/systemone"
+
+
+def allowed_jev_paths() -> List[String]:
+ var paths = List[String]()
+ paths.append(JEV_INTENDED_PATH)
+ return paths^
+
+
+def allowed_jev_methods() -> List[String]:
+ var methods = List[String]()
+ methods.append("POST")
+ return methods^
+
+
+def require_bearer_for(mode: String) -> Bool:
+ return False
+
+
+def analysis() -> String:
return (
'{"model":"jev-1.13.0","answers":{'
'"supply_status":{"type":"choice","choice":"offered","probabilities":{"offered":1.0,"forecast":0.0,"unclear":0.0},"confidence":1.0},'
@@ -65,124 +74,315 @@ def _analysis() -> String:
)
-def _response(status: Int, body: String) -> String:
- var reason = "OK"
- if status == 401:
- reason = "Unauthorized"
- elif status == 429:
- reason = "Too Many Requests"
- elif status == 500:
- reason = "Internal Server Error"
- elif status == 529:
- reason = "Overloaded"
- return (
- "HTTP/1.1 "
- + String(status)
- + " "
- + reason
- + "\r\ncontent-type: application/json\r\ncontent-length: "
- + String(body.byte_length())
- + "\r\nconnection: close\r\n\r\n"
- + body
- )
-
-
-def _send(mut reader: ConnectionReader, status: Int, body: String) raises:
- reader.write_all(_response(status, body))
+def _status(mode: String) -> Int:
+ if mode == "rate_limit":
+ return 429
+ if mode == "server_error":
+ return 500
+ if mode == "overloaded":
+ return 529
+ if mode == "auth":
+ return 401
+ return 200
-def _send_raw(mut reader: ConnectionReader, response: String) raises:
- reader.write_all(response)
-
-
-def _handle(
- mut reader: ConnectionReader, mode: String, framed: FramedRequest
-) raises:
- var path = framed.path
- if mode == "echo_authorization":
- var token = bearer_token(framed.headers_raw)
- if token == "":
- _send(reader, 401, '{"error":{"message":"missing authorization"}}')
- else:
- _send(reader, 200, '{"authorization":' + _json_string(token) + "}")
- return
- if path != "/v1/systemone":
- _send(reader, 404, '{"error":{"message":"not found"}}')
- return
-
- if mode == "ok":
- _send(reader, 200, _analysis())
- elif mode == "rate_limit":
- _send(reader, 429, '{"error":{"message":"slow down"}}')
- elif mode == "server_error":
- _send(reader, 500, '{"error":{"message":"boom"}}')
- elif mode == "overloaded":
- _send(reader, 529, '{"error":{"message":"overloaded"}}')
- elif mode == "auth":
- _send(reader, 401, '{"error":{"message":"bad key"}}')
- elif mode == "malformed_json":
- _send(reader, 200, "not json")
- elif mode == "model_mismatch":
- _send(reader, 200, _analysis().replace("jev-1.13.0", "jev-other"))
- elif mode == "truncated":
- var truncated = (
+def _raw_response(mode: String) -> String:
+ if mode == "truncated":
+ return (
"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n"
"content-length: 999\r\nconnection: close\r\n\r\n"
'{"model":"jev'
)
- reader.write_all(truncated)
- elif mode == "redirect":
- reader.write_all(
+ if mode == "redirect":
+ return (
"HTTP/1.1 302 Found\r\nlocation:"
" http://127.0.0.1:1/steal\r\ncontent-length:"
" 0\r\nconnection: close\r\n\r\n"
)
- elif mode == "slow":
- usleep(2_000_000)
- _send(reader, 200, _analysis())
+ return ""
+
+
+def _delay_ms(mode: String) -> Int:
+ if mode == "slow":
+ return 2000
+ return 0
+
+
+def _body(mode: String) -> String:
+ if mode == "ok" or mode == "slow":
+ return analysis()
+ if mode == "rate_limit":
+ return '{"error":{"message":"slow down"}}'
+ if mode == "server_error":
+ return '{"error":{"message":"boom"}}'
+ if mode == "overloaded":
+ return '{"error":{"message":"overloaded"}}'
+ if mode == "auth":
+ return '{"error":{"message":"bad key"}}'
+ if mode == "malformed_json":
+ return "not json"
+ if mode == "model_mismatch":
+ return analysis().replace("jev-1.13.0", "jev-other")
+ return '{"error":{"message":"unsupported_mode"}}'
+
+
+def _build_script(mode: String, framed: FramedRequest) -> ExchangeScript:
+ var script = exchange_script(
+ mode, framed.method, JEV_INTENDED_PATH, _status(mode), ""
+ )
+ script.delay_ms = _delay_ms(mode)
+ script.close_connection = True
+ script.require_bearer = require_bearer_for(mode)
+ var raw = _raw_response(mode)
+ if raw != "":
+ script.raw_response = raw
+ elif mode == "echo_authorization":
+ var auth = authorization_reason(framed.headers_raw, True)
+ if auth != "":
+ script.status = 401
+ script.response_body = '{"error":{"message":"' + auth + '"}}'
+ else:
+ script.echo_authorization = True
else:
- _send(reader, 500, '{"error":{"message":"unsupported_mode"}}')
+ script.response_body = _body(mode)
+ return script^
-def _serve(port: Int, mode: String, requests: Int) raises:
+def serve_jev(port: Int, mode: String, requests: Int) raises -> ServeReport:
+ var allowed = allowed_jev_paths()
+ var methods = allowed_jev_methods()
var listener = TcpListener.bind(SocketAddr.localhost(UInt16(port)))
var actual_port = Int(listener.local_addr().port)
- _write(1, "ready " + String(actual_port) + "\n")
+ write_raw(1, "ready " + String(actual_port) + "\n")
var request_count = 0
- while request_count < requests:
- var stream = listener.accept()
- var reader = ConnectionReader(stream^)
+ var connection_count = 0
+ try:
while request_count < requests:
- var framed = reader.read()
- if not framed.ok:
- if framed.error == "empty":
+ var stream = listener.accept()
+ connection_count += 1
+ var reader = ConnectionReader(stream^)
+ while request_count < requests:
+ var framed = reader.read()
+ if not framed.ok:
+ if framed.error == "empty":
+ if request_count < requests:
+ return ServeReport(
+ False,
+ "accounting",
+ mode,
+ "missing_exchanges",
+ request_count,
+ connection_count,
+ )
+ break
+ return ServeReport(
+ False,
+ "read",
+ mode,
+ framed.error,
+ request_count,
+ connection_count,
+ )
+ var reason = validate_convenience(
+ framed, allowed, methods, False
+ )
+ if reason != "":
+ return ServeReport(
+ False,
+ "exchange",
+ mode,
+ reason,
+ request_count,
+ connection_count,
+ )
+ var next_index = request_count + 1
+ var script = _build_script(mode, framed)
+ var verify = verify_exchange(script, framed)
+ if verify != "":
+ return ServeReport(
+ False,
+ "exchange",
+ mode,
+ verify,
+ request_count,
+ connection_count,
+ )
+ if next_index == requests:
+ var extra = reader.probe_completion(
+ STRICT_COMPLETION_GRACE_MS
+ )
+ if extra != "":
+ return ServeReport(
+ False,
+ "accounting",
+ mode,
+ extra,
+ request_count,
+ connection_count,
+ )
+ request_count = next_index
+ if script.delay_ms > 0:
+ usleep(script.delay_ms * 1000)
+ reader.write_all(
+ render_response(
+ script,
+ framed.headers_raw,
+ request_count,
+ connection_count,
+ )
+ )
+ if script.close_connection:
break
- raise Error("strict fixture framing error: " + framed.error)
- request_count += 1
- _handle(reader, mode, framed)
- if not framed.keep_alive:
- break
- listener.close()
+ if request_count < requests:
+ return ServeReport(
+ False,
+ "accounting",
+ mode,
+ "missing_exchanges",
+ request_count,
+ connection_count,
+ )
+ return ServeReport(
+ True, "complete", mode, "ok", request_count, connection_count
+ )
+ except:
+ return ServeReport(
+ False, "read", mode, "io_error", request_count, connection_count
+ )
+
+
+# ── Parent side ─────────────────────────────────────────────────────────────
struct SpawnedJevStub(Movable):
var pid: Int
-
- def __init__(out self, pid: Int):
+ var _report_fd: Int
+ var _reaped: Bool
+ var _ok: Bool
+ var _phase: String
+ var _case: String
+ var _reason: String
+ var _requests: Int
+ var _connections: Int
+
+ def __init__(out self, pid: Int, report_fd: Int):
self.pid = pid
+ self._report_fd = report_fd
+ self._reaped = False
+ self._ok = False
+ self._phase = "pending"
+ self._case = "-"
+ self._reason = "not_reaped"
+ self._requests = 0
+ self._connections = 0
+
+ def ok(self) -> Bool:
+ return self._ok
+
+ def phase(self) -> String:
+ return String(self._phase)
+
+ def failure_case(self) -> String:
+ return String(self._case)
+
+ def reason(self) -> String:
+ return String(self._reason)
+
+ def request_count(self) -> Int:
+ return self._requests
+
+ def connection_count(self) -> Int:
+ return self._connections
+
+ def describe(self) -> String:
+ return (
+ "phase="
+ + self._phase
+ + " case="
+ + self._case
+ + " reason="
+ + self._reason
+ + " requests="
+ + String(self._requests)
+ + " connections="
+ + String(self._connections)
+ )
- def wait(mut self) raises:
- var process = Process(self.pid)
- var status = process.wait()
- if not status.exit_code or status.exit_code.value() != 0:
- raise Error("jev stub exited unexpectedly")
+ def status(self) -> ProcessStatus:
+ return wait_bounded(self.pid, 0)
+
+ def _store(
+ mut self,
+ ok: Bool,
+ phase: String,
+ case_label: String,
+ reason: String,
+ requests: Int,
+ connections: Int,
+ ):
+ self._ok = ok
+ self._phase = String(phase)
+ self._case = String(case_label)
+ self._reason = String(reason)
+ self._requests = requests
+ self._connections = connections
+
+ def reap(mut self):
+ if self._reaped:
+ return
+ var st = wait_bounded(self.pid, FIXTURE_DEFAULT_DEADLINE_MS)
+ var report_text = ""
+ if st.state == "running":
+ var term = terminate_owned(self.pid, TERMINATION_GRACE_MS)
+ self._store(False, "watchdog", "-", "timeout", 0, 0)
+ if not term.reaped():
+ self._reason = "unreaped:" + term.describe()
+ self._reaped = True
+ close_fd(self._report_fd)
+ return
+ if self._report_fd >= 0:
+ try:
+ report_text = read_line_bounded(
+ self._report_fd, STRICT_MAX_REPORT_BYTES, 1000
+ )
+ except:
+ report_text = ""
+ if report_text.startswith("result "):
+ var parsed = parse_report(report_text)
+ self._store(
+ parsed.ok,
+ parsed.phase,
+ parsed.case_label,
+ parsed.reason,
+ parsed.requests,
+ parsed.connections,
+ )
+ elif st.exited and st.exit_code == 0:
+ self._store(True, "complete", "-", "ok", 0, 0)
+ elif st.exited:
+ self._store(
+ False, "startup", "-", "exit_" + String(st.exit_code), 0, 0
+ )
+ else:
+ self._store(
+ False, "watchdog", "-", "signal_" + String(st.signal), 0, 0
+ )
+ self._reaped = True
+ close_fd(self._report_fd)
+ def wait(mut self) raises:
+ self.reap()
+ if not self._ok:
+ raise Error("fixture-failure " + self.describe())
-def reserve_jev_port() raises -> Int:
- var listener = TcpListener.bind(SocketAddr.localhost(0))
- var port = Int(listener.local_addr().port)
- listener.close()
- return port
+ def terminate(mut self) raises:
+ if self._reaped:
+ return
+ var st = terminate_owned(self.pid, TERMINATION_GRACE_MS)
+ if not st.reaped():
+ raise Error("lifecycle: owned child not reaped: " + st.describe())
+ self._reaped = True
+ close_fd(self._report_fd)
@fieldwise_init
@@ -191,6 +391,13 @@ struct SpawnedJevStubAuto(Movable):
var stub: SpawnedJevStub
+def reserve_jev_port() raises -> Int:
+ var listener = TcpListener.bind(SocketAddr.localhost(0))
+ var port = Int(listener.local_addr().port)
+ listener.close()
+ return port
+
+
def spawn_jev_stub_auto(
mode: String, requests: Int
) raises -> SpawnedJevStubAuto:
@@ -204,36 +411,109 @@ def spawn_jev_stub(
return started.stub^
+def serve_jev_scripted(
+ port: Int, var scripts: List[ExchangeScript]
+) raises -> ServeReport:
+ var listener = TcpListener.bind(SocketAddr.localhost(UInt16(port)))
+ var actual_port = Int(listener.local_addr().port)
+ write_raw(1, "ready " + String(actual_port) + "\n")
+ return serve_scripts(listener, scripts^, "scripted")
+
+
+def _read_ready_line(fd: Int) -> String:
+ try:
+ return read_line_bounded(fd, 256, FIXTURE_DEFAULT_DEADLINE_MS)
+ except:
+ return ""
+
+
+def spawn_jev_scripted_auto(
+ var scripts: List[ExchangeScript],
+) raises -> SpawnedJevStubAuto:
+ return _spawn_jev_scripted(0, scripts^)
+
+
+def _spawn_jev_scripted(
+ port: Int, var scripts: List[ExchangeScript]
+) raises -> SpawnedJevStubAuto:
+ var total = len(scripts)
+ var pipe = make_pipe()
+ var pid = fork_pid()
+ if pid == 0:
+ if dup2_fd(pipe.write_fd, 1) < 0:
+ child_exit(126)
+ close_fd(pipe.read_fd)
+ close_fd(pipe.write_fd)
+ _ = set_alarm(STUB_ALARM_SECONDS)
+ try:
+ var report = serve_jev_scripted(port, scripts^)
+ write_raw(1, report_line(report) + "\n")
+ child_exit(0 if report.ok else 125)
+ except:
+ var failed = ServeReport(
+ False, "startup", "scripted", "serve_failed", 0, 0
+ )
+ write_raw(1, report_line(failed) + "\n")
+ child_exit(125)
+ close_fd(pipe.write_fd)
+ var ready_line = _read_ready_line(pipe.read_fd)
+ if not ready_line.startswith("ready"):
+ var st = terminate_owned(pid, TERMINATION_GRACE_MS)
+ close_fd(pipe.read_fd)
+ raise Error(
+ "jev stub failed to report ready ("
+ + ready_line
+ + " / "
+ + st.describe()
+ + ")"
+ )
+ var reported_port = port
+ var space = ready_line.find(" ")
+ if space >= 0:
+ reported_port = Int(String(ready_line[byte = space + 1 :]))
+ _ = total
+ return SpawnedJevStubAuto(
+ port=reported_port, stub=SpawnedJevStub(pid, pipe.read_fd)
+ )
+
+
def _spawn_jev_stub(
port: Int, mode: String, requests: Int
) raises -> SpawnedJevStubAuto:
- var stdout_pipe = Pipe()
- var stdout_read_fd = c_int(stdout_pipe.fd_in.value().value)
- var stdout_write_fd = c_int(stdout_pipe.fd_out.value().value)
- var pid = _fork()
- if pid < 0:
- raise Error("failed to spawn jev stub")
+ var pipe = make_pipe()
+ var pid = fork_pid()
if pid == 0:
- if _dup2(stdout_write_fd, 1) < 0:
- _exit_child(c_int(126))
- _ = close(stdout_read_fd)
- _ = close(stdout_write_fd)
- _ = _alarm(c_uint(20))
+ if dup2_fd(pipe.write_fd, 1) < 0:
+ child_exit(126)
+ close_fd(pipe.read_fd)
+ close_fd(pipe.write_fd)
+ _ = set_alarm(STUB_ALARM_SECONDS)
try:
- _serve(port, mode, requests)
- _exit_child(c_int(0))
+ var report = serve_jev(port, mode, requests)
+ write_raw(1, report_line(report) + "\n")
+ child_exit(0 if report.ok else 125)
except:
- _exit_child(c_int(125))
- stdout_pipe.set_input_only()
- var ready_line = _read_pipe_line(stdout_pipe)
+ var failed = ServeReport(
+ False, "startup", mode, "serve_failed", 0, 0
+ )
+ write_raw(1, report_line(failed) + "\n")
+ child_exit(125)
+ close_fd(pipe.write_fd)
+ var ready_line = _read_ready_line(pipe.read_fd)
if not ready_line.startswith("ready"):
- stdout_pipe.set_output_only()
- var process = Process(Int(pid))
- _ = process.wait()
- raise Error("jev stub failed to report ready")
+ var st = terminate_owned(pid, TERMINATION_GRACE_MS)
+ close_fd(pipe.read_fd)
+ raise Error(
+ "jev stub failed to report ready ("
+ + ready_line
+ + " / "
+ + st.describe()
+ + ")"
+ )
var reported_port = port
var space = ready_line.find(" ")
if space >= 0:
reported_port = Int(String(ready_line[byte = space + 1 :]))
- stdout_pipe.set_output_only()
- return SpawnedJevStubAuto(port=reported_port, stub=SpawnedJevStub(Int(pid)))
+ return SpawnedJevStubAuto(
+ port=reported_port, stub=SpawnedJevStub(pid, pipe.read_fd)
+ )
diff --git a/tests/max_local_process_helper.mojo b/tests/max_local_process_helper.mojo
@@ -1,66 +1,84 @@
-from std.ffi import c_int, c_size_t, c_ssize_t, c_uint, external_call
-from std.os import Pipe, Process
-from std.sys._libc import close
+"""Strict scripted MaxLocal HTTP fixture (ADR-0012 D29 / ADR-0014 D33).
+
+Owned by the parent test process: startup, read, write and wait all have
+parent-enforced deadlines (FX06) and cleanup is exception-safe (FX08). Fixture
+verification failures are reported with a bounded phase/case/reason so tests
+can distinguish intended rejections from compiler/loader/startup/signal
+failures (FX07).
+"""
+
+from std.collections import List
from flare.net import SocketAddr
from flare.tcp import TcpListener
-from flare.tcp import TcpStream
from flare.utils import usleep
-from strict_fixture import ConnectionReader, FramedRequest, bearer_token
-
-
-def _dup2(oldfd: c_int, newfd: c_int) -> c_int:
- return external_call["dup2", c_int](oldfd, newfd)
-
-
-@always_inline
-def _fork() -> c_int:
- return external_call["fork", c_int]()
-
-
-@always_inline
-def _kill(pid: c_int, sig: c_int) -> c_int:
- return external_call["kill", c_int](pid, sig)
-
-
-@always_inline
-def _alarm(seconds: c_uint) -> c_uint:
- return external_call["alarm", c_uint](seconds)
-
-
-@always_inline
-def _exit_child(code: c_int):
- _ = external_call["_exit", c_int](code)
+from parent_lifecycle import (
+ FIXTURE_DEFAULT_DEADLINE_MS,
+ TERMINATION_GRACE_MS,
+ ProcessStatus,
+ child_exit,
+ close_fd,
+ dup2_fd,
+ fork_pid,
+ make_pipe,
+ read_line_bounded,
+ set_alarm,
+ terminate_owned,
+ wait_bounded,
+ write_raw,
+)
+from strict_fixture import (
+ STRICT_MAX_REPORT_BYTES,
+ json_escape,
+ STRICT_COMPLETION_GRACE_MS,
+ ConnectionReader,
+ ExchangeScript,
+ FramedRequest,
+ ServeReport,
+ authorization_reason,
+ exchange_script,
+ parse_report,
+ render_response,
+ report_line,
+ serve_scripts,
+ validate_convenience,
+ verify_exchange,
+)
+
+
+comptime STUB_ALARM_SECONDS = 20
+
+
+def allowed_max_local_paths() -> List[String]:
+ var paths = List[String]()
+ paths.append("/health")
+ paths.append("/v1/chat/completions")
+ return paths^
+
+
+def allowed_max_local_methods() -> List[String]:
+ var methods = List[String]()
+ methods.append("GET")
+ methods.append("POST")
+ return methods^
+
+
+def require_bearer_for(mode: String) -> Bool:
+ # Convenience modes never bypass route/method validation; the
+ # echo_authorization mode validates the exact Authorization header inside
+ # its own handler and answers 401 when it is missing/duplicated/spoofed.
+ return False
+
+
+# ── Scripted response bodies ────────────────────────────────────────────────
-def _write(fd: Int, text: String):
- _ = external_call["write", c_ssize_t](
- fd, text.as_bytes().unsafe_ptr(), c_size_t(text.byte_length())
- )
-
-
-def _read_pipe_line(mut pipe: Pipe) raises -> String:
- var buffer = InlineArray[Byte, 1](fill=0)
- var output = String("")
- while True:
- var read = pipe.read_bytes(Span(buffer))
- if read == 0:
- break
- var chunk = String(
- from_utf8=Span(ptr=buffer.unsafe_ptr(), length=Int(read))
- )
- if chunk == "\n":
- break
- output += chunk
- return output^
-
-
-def _json_string(value: String) -> String:
- return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
+def _chat_completion(body: String) -> String:
+ return '{"choices":[{"message":{"content":' + json_escape(body) + "}}]}"
-def _query_rewrite_analysis() -> String:
+def query_rewrite_analysis() -> String:
return (
'{"original_text":"local apples pickup weekend",'
'"normalized_text":"local apples pickup weekend",'
@@ -76,182 +94,389 @@ def _query_rewrite_analysis() -> String:
)
-def _chat_completion(body: String) -> String:
- return '{"choices":[{"message":{"content":' + _json_string(body) + "}}]}"
-
-
-def _response(status: Int, body: String) -> String:
- var reason = "OK"
- if status == 401:
- reason = "Unauthorized"
- elif status == 404:
- reason = "Not Found"
- elif status == 500:
- reason = "Internal Server Error"
- elif status == 503:
- reason = "Service Unavailable"
+def _schema_invalid_analysis() -> String:
return (
- "HTTP/1.1 "
- + String(status)
- + " "
- + reason
- + "\r\ncontent-type: application/json\r\ncontent-length: "
- + String(body.byte_length())
- + "\r\nconnection: close\r\n\r\n"
- + body
+ '{"original_text":"local apples pickup weekend",'
+ '"normalized_text":"local apples pickup weekend",'
+ '"query_terms":["apples","pickup","weekend"],'
+ '"normalization_signals":["lowercase","local_intent_detected"],'
+ '"ranking_hints":["prefer_local_results","prefer_pickup"],'
+ '"extracted_filters":{'
+ '"local_intent":true,'
+ '"fulfillment":"pickup",'
+ '"time_window":"weekend"'
+ "}}"
)
-def _send(mut reader: ConnectionReader, status: Int, body: String) raises:
- reader.write_all(_response(status, body))
-
-
-def _send_raw(mut reader: ConnectionReader, response: String) raises:
- reader.write_all(response)
-
-
-def _handle_health(mut reader: ConnectionReader, mode: String) raises:
+def _health_status(mode: String) -> Int:
if mode == "health_non_2xx":
- _send(reader, 503, '{"status":"unavailable"}')
- elif mode == "health_timeout":
- usleep(1_000_000)
- elif mode == "health_malformed_http":
- _send_raw(reader, "not an http response\r\n\r\n")
- elif mode == "query_rewrite_remaining_deadline_timeout":
- usleep(200_000)
- _send(reader, 200, '{"status":"ok"}')
- else:
- _send(reader, 200, '{"status":"ok"}')
-
-
-def _handle_chat_completions(mut reader: ConnectionReader, mode: String) raises:
- if mode == "query_rewrite_ok":
- _send(reader, 200, _chat_completion(_query_rewrite_analysis()))
- elif mode == "query_rewrite_non_2xx":
- _send(reader, 503, '{"error":{"message":"provider unavailable"}}')
- elif mode == "query_rewrite_invalid_json":
- _send(reader, 200, '{"choices":[{"message":{"content":"not json"}}]}')
- elif mode == "query_rewrite_schema_invalid":
- var body = (
- '{"original_text":"local apples pickup weekend",'
- '"normalized_text":"local apples pickup weekend",'
- '"query_terms":["apples","pickup","weekend"],'
- '"normalization_signals":["lowercase","local_intent_detected"],'
- '"ranking_hints":["prefer_local_results","prefer_pickup"],'
- '"extracted_filters":{'
- '"local_intent":true,'
- '"fulfillment":"pickup",'
- '"time_window":"weekend"'
- "}}"
+ return 503
+ return 200
+
+
+def _chat_status(mode: String) -> Int:
+ if mode == "query_rewrite_non_2xx":
+ return 503
+ return 200
+
+
+def _raw_response(mode: String, path: String) -> String:
+ if path == "/health" and mode == "health_malformed_http":
+ return "not an http response\r\n\r\n"
+ if (
+ path == "/v1/chat/completions"
+ and mode == "query_rewrite_malformed_http"
+ ):
+ return "not an http response\r\n\r\n"
+ return ""
+
+
+def _delay_ms(mode: String, path: String) -> Int:
+ if path == "/health" and mode == "health_timeout":
+ return 1000
+ if path == "/health" and mode == "query_rewrite_remaining_deadline_timeout":
+ return 200
+ if path == "/v1/chat/completions":
+ if mode == "query_rewrite_timeout":
+ return 2000
+ if mode == "query_rewrite_remaining_deadline_timeout":
+ return 400
+ if mode == "stall":
+ return 30000
+ return 0
+
+
+def _chat_body(
+ mode: String, request_index: Int, connection_index: Int
+) -> String:
+ if mode == "count_requests":
+ return (
+ '{"request_index":'
+ + String(request_index)
+ + ',"connection_index":'
+ + String(connection_index)
+ + "}"
)
- _send(reader, 200, _chat_completion(body))
- elif mode == "query_rewrite_top_level_string":
- _send(reader, 200, '"not object"')
- elif mode == "query_rewrite_top_level_array":
- _send(reader, 200, "[]")
- elif mode == "query_rewrite_top_level_null":
- _send(reader, 200, "null")
- elif mode == "query_rewrite_empty_choices":
- _send(reader, 200, '{"choices":[]}')
- elif mode == "query_rewrite_missing_content":
- _send(reader, 200, '{"choices":[{"message":{}}]}')
- elif mode == "query_rewrite_error_payload":
- _send(reader, 200, '{"error":{"message":"provider refusal"}}')
- elif mode == "query_rewrite_timeout":
- usleep(2_000_000)
- _send(reader, 200, _chat_completion(_query_rewrite_analysis()))
- elif mode == "query_rewrite_remaining_deadline_timeout":
- usleep(400_000)
- _send(reader, 200, _chat_completion(_query_rewrite_analysis()))
- elif mode == "query_rewrite_malformed_http":
- _send_raw(reader, "not an http response\r\n\r\n")
- else:
- _send(reader, 500, '{"error":"unsupported_mode"}')
+ if mode == "query_rewrite_ok":
+ return _chat_completion(query_rewrite_analysis())
+ if mode == "query_rewrite_non_2xx":
+ return '{"error":{"message":"provider unavailable"}}'
+ if mode == "query_rewrite_invalid_json":
+ return '{"choices":[{"message":{"content":"not json"}}]}'
+ if mode == "query_rewrite_schema_invalid":
+ return _chat_completion(_schema_invalid_analysis())
+ if mode == "query_rewrite_top_level_string":
+ return '"not object"'
+ if mode == "query_rewrite_top_level_array":
+ return "[]"
+ if mode == "query_rewrite_top_level_null":
+ return "null"
+ if mode == "query_rewrite_empty_choices":
+ return '{"choices":[]}'
+ if mode == "query_rewrite_missing_content":
+ return '{"choices":[{"message":{}}]}'
+ if mode == "query_rewrite_error_payload":
+ return '{"error":{"message":"provider refusal"}}'
+ if mode in (
+ "query_rewrite_timeout",
+ "query_rewrite_remaining_deadline_timeout",
+ ):
+ return _chat_completion(query_rewrite_analysis())
+ return '{"error":"unsupported_mode"}'
+
+
+def _health_body(mode: String) -> String:
+ if mode == "health_non_2xx":
+ return '{"status":"unavailable"}'
+ return '{"status":"ok"}'
-def _handle_framed(
- mut reader: ConnectionReader,
+def _build_script(
mode: String,
+ framed: FramedRequest,
request_index: Int,
connection_index: Int,
- framed: FramedRequest,
-) raises:
+) -> ExchangeScript:
var path = framed.path
- if mode == "echo_body_bytes":
- _send(
- reader,
- 200,
- '{"received_bytes":' + String(framed.body.byte_length()) + "}",
- )
- elif mode == "count_requests":
- _send(
- reader,
- 200,
- '{"request_index":'
- + String(request_index)
- + ',"connection_index":'
- + String(connection_index)
- + "}",
- )
- elif mode == "echo_authorization":
- var token = bearer_token(framed.headers_raw)
- if token == "":
- _send(reader, 401, '{"error":"missing_authorization"}')
+ var status = _chat_status(mode)
+ if path == "/health":
+ status = _health_status(mode)
+ var script = exchange_script(mode, framed.method, path, status, "")
+ script.delay_ms = _delay_ms(mode, path)
+ script.close_connection = True
+ script.require_bearer = require_bearer_for(mode)
+ var raw = _raw_response(mode, path)
+ if raw != "":
+ script.raw_response = raw
+ elif mode == "echo_authorization" and path == "/v1/chat/completions":
+ var auth = authorization_reason(framed.headers_raw, True)
+ if auth != "":
+ script.status = 401
+ script.response_body = '{"error":"' + auth + '"}'
else:
- _send(reader, 200, '{"authorization":' + _json_string(token) + "}")
- elif mode == "stall":
- usleep(30_000_000)
+ script.echo_authorization = True
+ elif mode == "echo_body_bytes" and path == "/v1/chat/completions":
+ script.response_body = (
+ '{"received_bytes":' + String(framed.body.byte_length()) + "}"
+ )
elif path == "/health":
- _handle_health(reader, mode)
- elif path == "/v1/chat/completions":
- _handle_chat_completions(reader, mode)
+ script.response_body = _health_body(mode)
else:
- _send(reader, 404, '{"error":"not_found"}')
+ script.response_body = _chat_body(mode, request_index, connection_index)
+ return script^
+
+
+# ── Child serve loop ────────────────────────────────────────────────────────
-def _serve_max_local_stub(port: Int, mode: String, requests: Int) raises:
+def serve_max_local(
+ port: Int, mode: String, requests: Int
+) raises -> ServeReport:
+ var allowed = allowed_max_local_paths()
+ var methods = allowed_max_local_methods()
var listener = TcpListener.bind(SocketAddr.localhost(UInt16(port)))
var actual_port = Int(listener.local_addr().port)
- _write(1, "ready " + String(actual_port) + "\n")
+ write_raw(1, "ready " + String(actual_port) + "\n")
var request_count = 0
var connection_count = 0
- while request_count < requests:
- var stream = listener.accept()
- connection_count += 1
- var reader = ConnectionReader(stream^)
+ try:
while request_count < requests:
- var framed = reader.read()
- if not framed.ok:
- if framed.error == "empty":
+ var stream = listener.accept()
+ connection_count += 1
+ var reader = ConnectionReader(stream^)
+ while request_count < requests:
+ var framed = reader.read()
+ if not framed.ok:
+ if framed.error == "empty":
+ if request_count < requests:
+ return ServeReport(
+ False,
+ "accounting",
+ mode,
+ "missing_exchanges",
+ request_count,
+ connection_count,
+ )
+ break
+ return ServeReport(
+ False,
+ "read",
+ mode,
+ framed.error,
+ request_count,
+ connection_count,
+ )
+ var reason = validate_convenience(
+ framed, allowed, methods, require_bearer_for(mode)
+ )
+ if reason != "":
+ return ServeReport(
+ False,
+ "exchange",
+ mode,
+ reason,
+ request_count,
+ connection_count,
+ )
+ var next_index = request_count + 1
+ var script = _build_script(
+ mode, framed, next_index, connection_count
+ )
+ var verify = verify_exchange(script, framed)
+ if verify != "":
+ return ServeReport(
+ False,
+ "exchange",
+ mode,
+ verify,
+ request_count,
+ connection_count,
+ )
+ if next_index == requests:
+ var extra = reader.probe_completion(
+ STRICT_COMPLETION_GRACE_MS
+ )
+ if extra != "":
+ return ServeReport(
+ False,
+ "accounting",
+ mode,
+ extra,
+ request_count,
+ connection_count,
+ )
+ request_count = next_index
+ if script.delay_ms > 0:
+ usleep(script.delay_ms * 1000)
+ reader.write_all(
+ render_response(
+ script,
+ framed.headers_raw,
+ request_count,
+ connection_count,
+ )
+ )
+ if script.close_connection:
break
- raise Error("strict fixture framing error: " + framed.error)
- request_count += 1
- _handle_framed(
- reader, mode, request_count, connection_count, framed
+ if request_count < requests:
+ return ServeReport(
+ False,
+ "accounting",
+ mode,
+ "missing_exchanges",
+ request_count,
+ connection_count,
)
- if not framed.keep_alive:
- break
- listener.close()
+ return ServeReport(
+ True, "complete", mode, "ok", request_count, connection_count
+ )
+ except:
+ return ServeReport(
+ False, "read", mode, "io_error", request_count, connection_count
+ )
+
+
+# ── Parent side ─────────────────────────────────────────────────────────────
struct SpawnedMaxLocalStub(Movable):
var pid: Int
var port: Int
-
- def __init__(out self, pid: Int, port: Int):
+ var _report_fd: Int
+ var _reaped: Bool
+ var _ok: Bool
+ var _phase: String
+ var _case: String
+ var _reason: String
+ var _requests: Int
+ var _connections: Int
+
+ def __init__(out self, pid: Int, port: Int, report_fd: Int):
self.pid = pid
self.port = port
+ self._report_fd = report_fd
+ self._reaped = False
+ self._ok = False
+ self._phase = "pending"
+ self._case = "-"
+ self._reason = "not_reaped"
+ self._requests = 0
+ self._connections = 0
+
+ def ok(self) -> Bool:
+ return self._ok
+
+ def phase(self) -> String:
+ return String(self._phase)
+
+ def failure_case(self) -> String:
+ return String(self._case)
+
+ def reason(self) -> String:
+ return String(self._reason)
+
+ def request_count(self) -> Int:
+ return self._requests
+
+ def connection_count(self) -> Int:
+ return self._connections
+
+ def describe(self) -> String:
+ return (
+ "phase="
+ + self._phase
+ + " case="
+ + self._case
+ + " reason="
+ + self._reason
+ + " requests="
+ + String(self._requests)
+ + " connections="
+ + String(self._connections)
+ )
+
+ def status(self) -> ProcessStatus:
+ return wait_bounded(self.pid, 0)
+
+ def _store(
+ mut self,
+ ok: Bool,
+ phase: String,
+ case_label: String,
+ reason: String,
+ requests: Int,
+ connections: Int,
+ ):
+ self._ok = ok
+ self._phase = String(phase)
+ self._case = String(case_label)
+ self._reason = String(reason)
+ self._requests = requests
+ self._connections = connections
+
+ def reap(mut self):
+ """Reap the owned child and decode its bounded report (never raises)."""
+ if self._reaped:
+ return
+ var st = wait_bounded(self.pid, FIXTURE_DEFAULT_DEADLINE_MS)
+ var report_text = ""
+ if st.state == "running":
+ var term = terminate_owned(self.pid, TERMINATION_GRACE_MS)
+ self._store(False, "watchdog", "-", "timeout", 0, 0)
+ if not term.reaped():
+ self._reason = "unreaped:" + term.describe()
+ self._reaped = True
+ close_fd(self._report_fd)
+ return
+ if self._report_fd >= 0:
+ try:
+ report_text = read_line_bounded(
+ self._report_fd, STRICT_MAX_REPORT_BYTES, 1000
+ )
+ except:
+ report_text = ""
+ if report_text.startswith("result "):
+ var parsed = parse_report(report_text)
+ self._store(
+ parsed.ok,
+ parsed.phase,
+ parsed.case_label,
+ parsed.reason,
+ parsed.requests,
+ parsed.connections,
+ )
+ elif st.exited and st.exit_code == 0:
+ self._store(True, "complete", "-", "ok", 0, 0)
+ elif st.exited:
+ self._store(
+ False,
+ "startup",
+ "-",
+ "exit_" + String(st.exit_code),
+ 0,
+ 0,
+ )
+ else:
+ self._store(
+ False, "watchdog", "-", "signal_" + String(st.signal), 0, 0
+ )
+ self._reaped = True
+ close_fd(self._report_fd)
def wait(mut self) raises:
- var process = Process(self.pid)
- var status = process.wait()
- if not status.exit_code or status.exit_code.value() != 0:
- raise Error("max_local stub exited unexpectedly")
+ self.reap()
+ if not self._ok:
+ raise Error("fixture-failure " + self.describe())
def terminate(mut self) raises:
- _ = _kill(c_int(self.pid), c_int(15))
- var process = Process(self.pid)
- _ = process.wait()
+ if self._reaped:
+ return
+ var st = terminate_owned(self.pid, TERMINATION_GRACE_MS)
+ if not st.reaped():
+ raise Error("lifecycle: owned child not reaped: " + st.describe())
+ self._reaped = True
+ close_fd(self._report_fd)
def reserve_loopback_port() raises -> Int:
@@ -261,41 +486,88 @@ def reserve_loopback_port() raises -> Int:
return port
+def _serve_max_local_for(
+ port: Int,
+ var scripts: List[ExchangeScript],
+ mode: String,
+ requests: Int,
+ scripted: Bool,
+) raises -> ServeReport:
+ if scripted:
+ return serve_max_local_scripted(port, scripts^)
+ return serve_max_local(port, mode, requests)
+
+
+def _read_ready_line(fd: Int) -> String:
+ try:
+ return read_line_bounded(fd, 256, FIXTURE_DEFAULT_DEADLINE_MS)
+ except:
+ return ""
+
+
+def serve_max_local_scripted(
+ port: Int, var scripts: List[ExchangeScript]
+) raises -> ServeReport:
+ var listener = TcpListener.bind(SocketAddr.localhost(UInt16(port)))
+ var actual_port = Int(listener.local_addr().port)
+ write_raw(1, "ready " + String(actual_port) + "\n")
+ return serve_scripts(listener, scripts^, "scripted")
+
+
def spawn_max_local_stub(
port: Int, mode: String, requests: Int
) raises -> SpawnedMaxLocalStub:
- var stdout_pipe = Pipe()
- var stdout_read_fd = c_int(stdout_pipe.fd_in.value().value)
- var stdout_write_fd = c_int(stdout_pipe.fd_out.value().value)
+ var scripts = List[ExchangeScript]()
+ return _spawn_max_local(port, scripts^, mode, requests, False)
- var pid = _fork()
- if pid < 0:
- raise Error("failed to spawn max_local stub")
+def spawn_max_local_scripted(
+ port: Int, var scripts: List[ExchangeScript]
+) raises -> SpawnedMaxLocalStub:
+ return _spawn_max_local(port, scripts^, "scripted", len(scripts), True)
+
+
+def _spawn_max_local(
+ port: Int,
+ var scripts: List[ExchangeScript],
+ mode: String,
+ requests: Int,
+ scripted: Bool,
+) raises -> SpawnedMaxLocalStub:
+ var pipe = make_pipe()
+ var pid = fork_pid()
if pid == 0:
- if _dup2(stdout_write_fd, 1) < 0:
- _exit_child(c_int(126))
- _ = close(stdout_read_fd)
- _ = close(stdout_write_fd)
- _ = _alarm(c_uint(20))
+ if dup2_fd(pipe.write_fd, 1) < 0:
+ child_exit(126)
+ close_fd(pipe.read_fd)
+ close_fd(pipe.write_fd)
+ _ = set_alarm(STUB_ALARM_SECONDS)
try:
- _serve_max_local_stub(port, mode, requests)
- _exit_child(c_int(0))
+ var report = _serve_max_local_for(
+ port, scripts^, mode, requests, scripted
+ )
+ write_raw(1, report_line(report) + "\n")
+ child_exit(0 if report.ok else 125)
except:
- _exit_child(c_int(125))
-
- stdout_pipe.set_input_only()
- var ready_line = _read_pipe_line(stdout_pipe)
+ var failed = ServeReport(
+ False, "startup", mode, "serve_failed", 0, 0
+ )
+ write_raw(1, report_line(failed) + "\n")
+ child_exit(125)
+ close_fd(pipe.write_fd)
+ var ready_line = _read_ready_line(pipe.read_fd)
if not ready_line.startswith("ready"):
- stdout_pipe.set_output_only()
- var process = Process(Int(pid))
- _ = process.wait()
- raise Error("max_local stub failed to report ready")
-
+ var st = terminate_owned(pid, TERMINATION_GRACE_MS)
+ close_fd(pipe.read_fd)
+ raise Error(
+ "max_local stub failed to report ready ("
+ + ready_line
+ + " / "
+ + st.describe()
+ + ")"
+ )
var reported_port = port
var space = ready_line.find(" ")
if space >= 0:
reported_port = Int(String(ready_line[byte = space + 1 :]))
-
- stdout_pipe.set_output_only()
- return SpawnedMaxLocalStub(Int(pid), reported_port)
+ return SpawnedMaxLocalStub(pid, reported_port, pipe.read_fd)
diff --git a/tests/parent_lifecycle.mojo b/tests/parent_lifecycle.mojo
@@ -0,0 +1,314 @@
+"""Governed test-only POSIX process/pipe lifecycle and deadline helpers.
+
+ADR-0014 D33 FX06/FX08 require *parent-enforced*, finite startup/read/write/
+wait deadlines and exception-safe cleanup/reaping for provider fixture
+children and ``run_stdio_entrypoint``. A child ``alarm(2)`` watchdog is
+defense in depth, never the parent's lifecycle proof.
+
+Mojo's ``std.os.Pipe`` wrapper does not expose stable raw descriptors for
+``poll(2)``-based bounded I/O, so this module owns the required POSIX surface
+directly. It is test-only tooling: it changes no HYF product policy.
+
+Ownership rules:
+
+* ``make_pipe``/``fork_pid`` create resources owned by the calling test.
+* ``terminate_owned`` only signals a pid this process forked and has not yet
+ reaped, so a reused PID can never be targeted by a repeated teardown.
+* No broad ``pkill``/name matching is performed anywhere.
+"""
+
+from std.ffi import c_int, c_uint, c_ssize_t, c_size_t, external_call
+from std.sys._libc import close
+from std.time import perf_counter_ns
+
+
+comptime WNOHANG: Int = 1
+comptime POLLIN: Int = 1
+comptime POLLOUT: Int = 4
+comptime POLLERR: Int = 8
+comptime POLLHUP: Int = 16
+comptime POLLNVAL: Int = 32
+comptime SIGALRM: Int = 14
+comptime SIGKILL: Int = 9
+comptime SIGTERM: Int = 15
+
+comptime FIXTURE_DEFAULT_DEADLINE_MS: Int = 20000
+comptime TERMINATION_GRACE_MS: Int = 2000
+comptime LIFECYCLE_POLL_SLICE_MS: Int = 25
+
+
+def now_ms() -> Int:
+ return Int(perf_counter_ns() // 1_000_000)
+
+
+# ── Raw descriptor helpers ──────────────────────────────────────────────────
+
+
+@fieldwise_init
+struct PipeFds(Movable):
+ var read_fd: Int
+ var write_fd: Int
+
+
+def make_pipe() raises -> PipeFds:
+ var fds = InlineArray[c_int, 2](fill=0)
+ if Int(external_call["pipe", c_int](fds.unsafe_ptr())) != 0:
+ raise Error("lifecycle: pipe failed")
+ return PipeFds(Int(fds[0]), Int(fds[1]))
+
+
+def close_fd(fd: Int):
+ if fd >= 0:
+ _ = close(c_int(fd))
+
+
+def fork_pid() raises -> Int:
+ var pid = Int(external_call["fork", c_int]())
+ if pid < 0:
+ raise Error("lifecycle: fork failed")
+ return pid
+
+
+def child_exit(code: Int):
+ _ = external_call["_exit", c_int](c_int(code))
+
+
+def dup2_fd(oldfd: Int, newfd: Int) -> Int:
+ return Int(external_call["dup2", c_int](c_int(oldfd), c_int(newfd)))
+
+
+def kill_pid(pid: Int, sig: Int) -> Int:
+ return Int(external_call["kill", c_int](c_int(pid), c_int(sig)))
+
+
+def owned_pid() -> Int:
+ return Int(external_call["getpid", c_int]())
+
+
+def set_alarm(seconds: Int) -> Int:
+ return Int(external_call["alarm", c_uint](c_uint(seconds)))
+
+
+def sleep_ms(ms: Int):
+ if ms > 0:
+ _ = external_call["usleep", c_int](c_int(ms * 1000))
+
+
+def poll_fd(fd: Int, events: Int, timeout_ms: Int) -> Int:
+ var cell = InlineArray[Int32, 2](fill=0)
+ cell[0] = Int32(fd)
+ cell[1] = Int32(events)
+ var n = Int(
+ external_call["poll", c_int](
+ cell.unsafe_ptr(), c_uint(1), c_int(timeout_ms)
+ )
+ )
+ if n <= 0:
+ return 0
+ return (Int(cell[1]) >> 16) & 0xFFFF
+
+
+def read_fd(fd: Int, buf: UnsafePointer[Byte, ...], max_bytes: Int) -> Int:
+ return Int(external_call["read", c_ssize_t](fd, buf, c_size_t(max_bytes)))
+
+
+def _write_fd(fd: Int, ptr: UnsafePointer[UInt8, ...], n: Int) -> Int:
+ return Int(external_call["write", c_ssize_t](fd, ptr, c_size_t(n)))
+
+
+def write_raw(fd: Int, text: String) -> Int:
+ """Best-effort blocking write of a small bounded string (no deadline)."""
+ var n = _write_fd(fd, text.as_bytes().unsafe_ptr(), text.byte_length())
+ return n
+
+
+def write_fd_bounded(fd: Int, data: String, deadline_ms: Int) -> String:
+ """Write ``data`` with a parent-enforced deadline.
+
+ Returns ``""`` on success or a bounded reason such as
+ ``write_deadline_expired`` / ``write_pipe_closed``.
+ """
+ var total = data.byte_length()
+ var sent = 0
+ var start = now_ms()
+ while sent < total:
+ if now_ms() - start >= deadline_ms:
+ return "write_deadline_expired"
+ var ev = poll_fd(fd, POLLOUT, LIFECYCLE_POLL_SLICE_MS)
+ if ev == 0:
+ continue
+ if (ev & (POLLERR | POLLHUP | POLLNVAL)) != 0:
+ return "write_pipe_closed"
+ var chunk = min(4096, total - sent)
+ var slice = data[byte = sent : sent + chunk]
+ var n = _write_fd(fd, slice.as_bytes().unsafe_ptr(), chunk)
+ if n <= 0:
+ return "write_failed"
+ sent += n
+ return ""
+
+
+def read_line_bounded(
+ fd: Int, max_bytes: Int, deadline_ms: Int
+) raises -> String:
+ """Read one newline-terminated line with bounded size and deadline.
+
+ Raises on overflow or deadline. An EOF before a newline returns whatever
+ bytes were read (possibly empty).
+ """
+ var out = List[UInt8]()
+ var buf = InlineArray[Byte, 512](fill=0)
+ var start = now_ms()
+ while True:
+ if len(out) >= max_bytes:
+ raise Error("ready_output_overflow")
+ if now_ms() - start >= deadline_ms:
+ raise Error("read_deadline_expired")
+ var ev = poll_fd(fd, POLLIN, LIFECYCLE_POLL_SLICE_MS)
+ if ev == 0:
+ continue
+ var n = read_fd(fd, buf.unsafe_ptr(), 512)
+ if n <= 0:
+ break
+ for index in range(n):
+ if Int(buf[index]) == 10:
+ return _bytes_to_string(out)
+ out.append(UInt8(Int(buf[index])))
+ return _bytes_to_string(out)
+
+
+def drain_fd_bounded(
+ fd: Int, max_bytes: Int, deadline_ms: Int
+) raises -> String:
+ """Read available bytes up to ``max_bytes`` under deadline."""
+ var out = List[UInt8]()
+ var buf = InlineArray[Byte, 1024](fill=0)
+ var start = now_ms()
+ while len(out) < max_bytes:
+ if now_ms() - start >= deadline_ms:
+ break
+ var ev = poll_fd(fd, POLLIN, LIFECYCLE_POLL_SLICE_MS)
+ if ev == 0:
+ continue
+ var n = read_fd(fd, buf.unsafe_ptr(), 1024)
+ if n <= 0:
+ break
+ for index in range(n):
+ out.append(UInt8(Int(buf[index])))
+ return _bytes_to_string(out)
+
+
+def read_all_bounded(
+ fd: Int, max_bytes: Int, deadline_ms: Int
+) raises -> String:
+ """Read until EOF, with a bounded size cap and parent deadline.
+
+ Raises ``stdout_overflow`` when the cap is exceeded and
+ ``read_deadline_expired`` when the deadline elapses first.
+ """
+ var out = List[UInt8]()
+ var buf = InlineArray[Byte, 4096](fill=0)
+ var start = now_ms()
+ while True:
+ if now_ms() - start >= deadline_ms:
+ raise Error("read_deadline_expired")
+ var ev = poll_fd(fd, POLLIN, LIFECYCLE_POLL_SLICE_MS)
+ if ev == 0:
+ continue
+ var n = read_fd(fd, buf.unsafe_ptr(), 4096)
+ if n <= 0:
+ break
+ if len(out) + n > max_bytes:
+ raise Error("stdout_overflow")
+ for index in range(n):
+ out.append(UInt8(Int(buf[index])))
+ return _bytes_to_string(out)
+
+
+def _bytes_to_string(bytes: List[UInt8]) raises -> String:
+ if len(bytes) == 0:
+ return ""
+ return String(from_utf8=Span(ptr=bytes.unsafe_ptr(), length=len(bytes)))
+
+
+# ── Child lifecycle ─────────────────────────────────────────────────────────
+
+
+@fieldwise_init
+struct ProcessStatus(Movable):
+ var state: String
+ var exited: Bool
+ var exit_code: Int
+ var signal: Int
+ var raw: Int
+
+ def reaped(self) -> Bool:
+ return self.state == "reaped"
+
+ def describe(self) -> String:
+ if self.state != "reaped":
+ return self.state
+ if self.exited:
+ return "exited=" + String(self.exit_code)
+ return "signal=" + String(self.signal)
+
+
+def _decode_status(raw: Int) -> ProcessStatus:
+ var low = raw & 0x7F
+ if low == 0:
+ return ProcessStatus("reaped", True, (raw >> 8) & 0xFF, 0, raw)
+ if low == 0x7F:
+ return ProcessStatus("stopped", False, -1, 0, raw)
+ return ProcessStatus("reaped", False, -1, low, raw)
+
+
+def wait_nohang(pid: Int) -> ProcessStatus:
+ var status = InlineArray[c_int, 1](fill=0)
+ var r = Int(
+ external_call["waitpid", c_int](
+ c_int(pid), status.unsafe_ptr(), c_int(WNOHANG)
+ )
+ )
+ if r == pid:
+ return _decode_status(Int(status[0]))
+ if r < 0:
+ return ProcessStatus("gone", False, -1, -1, -1)
+ return ProcessStatus("running", False, -1, 0, 0)
+
+
+def wait_bounded(pid: Int, deadline_ms: Int) -> ProcessStatus:
+ var start = now_ms()
+ while True:
+ var st = wait_nohang(pid)
+ if st.state != "running":
+ return st^
+ if now_ms() - start >= deadline_ms:
+ return st^
+ sleep_ms(5)
+
+
+def terminate_owned(pid: Int, grace_ms: Int) -> ProcessStatus:
+ """Reap a child this test owns, escalating SIGTERM -> SIGKILL.
+
+ A pid already reaped or not waitable (``gone``) is never signaled, so a
+ reused PID from an unrelated process can never be targeted.
+ """
+ var st = wait_nohang(pid)
+ if st.state != "running":
+ return st^
+ _ = kill_pid(pid, SIGTERM)
+ st = wait_bounded(pid, grace_ms)
+ if st.state == "running":
+ _ = kill_pid(pid, SIGKILL)
+ st = wait_bounded(pid, grace_ms)
+ return st^
+
+
+def pid_not_waitable(pid: Int) -> Bool:
+ """True when ``pid`` is neither running nor an unreaped zombie of ours.
+
+ Used only to evidence, after ``terminate_owned`` reaped a specific owned
+ child, that the same pid is no longer waitable. It never reaps a pid this
+ test did not fork and never scans by process name.
+ """
+ return wait_nohang(pid).state == "gone"
diff --git a/tests/stdio_process_helper.mojo b/tests/stdio_process_helper.mojo
@@ -1,7 +1,30 @@
+"""Bounded, parent-owned stdio entrypoint runner for HYF process tests.
+
+ADR-0014 D33 FX06/FX08: the *parent* enforces finite startup/read/write/wait
+deadlines, bounds stdout/ready output, and reaps/closes every owned descriptor
+on success, assertion failure, timeout or early return. The child ``alarm`` is
+defense in depth only.
+
+The request/response API is unchanged so existing call sites keep working.
+"""
+
import std.os
-from std.os import Pipe, Process
from std.ffi import CStringSlice, c_int, external_call
-from std.sys._libc import close
+
+from parent_lifecycle import (
+ TERMINATION_GRACE_MS,
+ FIXTURE_DEFAULT_DEADLINE_MS,
+ child_exit,
+ close_fd,
+ dup2_fd,
+ fork_pid,
+ make_pipe,
+ read_all_bounded,
+ set_alarm,
+ terminate_owned,
+ wait_bounded,
+ write_fd_bounded,
+)
from safe_tempdir import SafeTempDir
from json import Value, loads
@@ -10,6 +33,13 @@ from json import Value, loads
comptime HYF_PATHS_PROFILE_ENV = "HYF_PATHS_PROFILE"
comptime HYF_PATHS_REPO_LOCAL_ROOT_ENV = "HYF_PATHS_REPO_LOCAL_ROOT"
+# The entrypoint compiles and runs a real Mojo program; this is a finite
+# build/run lane budget, not a fixture lifetime and not a product SLO.
+comptime STDIO_ENTRYPOINT_DEADLINE_MS = 120000
+comptime STDIO_CHILD_ALARM_SECONDS = 180
+comptime STDIO_MAX_STDOUT_BYTES = 2097152
+comptime STDIO_MAX_STDERR_BYTES = 65536
+
struct ScopedEnvVar:
var name: String
@@ -33,33 +63,6 @@ struct ScopedEnvVar:
_ = std.os.unsetenv(self.name)
-def _dup2(oldfd: c_int, newfd: c_int) -> c_int:
- return external_call["dup2", c_int](oldfd, newfd)
-
-
-@always_inline
-def _fork() -> c_int:
- return external_call["fork", c_int]()
-
-
-@always_inline
-def _exit_child(code: c_int):
- _ = external_call["_exit", c_int](code)
-
-
-def _read_pipe_to_string(mut pipe: Pipe) raises -> String:
- var buffer = InlineArray[Byte, 4096](fill=0)
- var output = String("")
- while True:
- var read = pipe.read_bytes(Span(buffer))
- if read == 0:
- break
- output += String(
- from_utf8=Span(ptr=buffer.unsafe_ptr(), length=Int(read))
- )
- return output^
-
-
def run_stdio_entrypoint(
entrypoint: String, request_json: String
) raises -> Value:
@@ -74,12 +77,35 @@ def run_stdio_entrypoint(
)
+def _terminate_and_raise(
+ pid: Int, stdin_fd: Int, stdout_fd: Int, stderr_fd: Int, reason: String
+) raises:
+ var st = terminate_owned(pid, TERMINATION_GRACE_MS)
+ close_fd(stdin_fd)
+ close_fd(stdout_fd)
+ close_fd(stderr_fd)
+ raise Error("stdio-entrypoint " + reason + " (child " + st.describe() + ")")
+
+
def run_stdio_entrypoint_with_2_args(
entrypoint: String, request_json: String, arg0: String, arg1: String
) raises -> Value:
- var stdin_pipe = Pipe()
- var stdout_pipe = Pipe()
- var output = String("")
+ return run_stdio_entrypoint_with_deadline(
+ entrypoint, request_json, arg0, arg1, STDIO_ENTRYPOINT_DEADLINE_MS
+ )
+
+
+def run_stdio_entrypoint_with_deadline(
+ entrypoint: String,
+ request_json: String,
+ arg0: String,
+ arg1: String,
+ deadline_ms: Int,
+) raises -> Value:
+ var stdin_pipe = make_pipe()
+ var stdout_pipe = make_pipe()
+ var stderr_pipe = make_pipe()
+
var command = String("mojo")
var include_flag = String("-I")
var include_path = String("src")
@@ -107,48 +133,98 @@ def run_stdio_entrypoint_with_2_args(
process_arg1.as_c_string_slice()
)
- var stdin_read_fd = c_int(stdin_pipe.fd_in.value().value)
- var stdin_write_fd = c_int(stdin_pipe.fd_out.value().value)
- var stdout_read_fd = c_int(stdout_pipe.fd_in.value().value)
- var stdout_write_fd = c_int(stdout_pipe.fd_out.value().value)
+ var stdin_read_fd = stdin_pipe.read_fd
+ var stdin_write_fd = stdin_pipe.write_fd
+ var stdout_read_fd = stdout_pipe.read_fd
+ var stdout_write_fd = stdout_pipe.write_fd
+ var stderr_read_fd = stderr_pipe.read_fd
+ var stderr_write_fd = stderr_pipe.write_fd
var command_ptr = command.as_c_string_slice().unsafe_ptr()
var argv_ptr = argv.unsafe_ptr()
- var pid = _fork()
- if pid < 0:
- raise Error("failed to spawn hyf process test child")
-
+ var pid = fork_pid()
if pid == 0:
- if _dup2(stdin_read_fd, 0) < 0:
- _exit_child(c_int(126))
- if _dup2(stdout_write_fd, 1) < 0:
- _exit_child(c_int(126))
- _ = close(stdin_read_fd)
- _ = close(stdin_write_fd)
- _ = close(stdout_read_fd)
- _ = close(stdout_write_fd)
+ if dup2_fd(stdin_read_fd, 0) < 0:
+ child_exit(126)
+ if dup2_fd(stdout_write_fd, 1) < 0:
+ child_exit(126)
+ if dup2_fd(stderr_write_fd, 2) < 0:
+ child_exit(126)
+ close_fd(stdin_read_fd)
+ close_fd(stdin_write_fd)
+ close_fd(stdout_read_fd)
+ close_fd(stdout_write_fd)
+ close_fd(stderr_read_fd)
+ close_fd(stderr_write_fd)
+ _ = set_alarm(STDIO_CHILD_ALARM_SECONDS)
_ = external_call["execvp", c_int](command_ptr, argv_ptr)
- _exit_child(c_int(127))
-
- stdin_pipe.set_output_only()
- stdout_pipe.set_input_only()
+ child_exit(127)
- stdin_pipe.write_bytes((request_json + "\n").as_bytes())
- stdin_pipe.set_input_only()
+ close_fd(stdin_read_fd)
+ close_fd(stdout_write_fd)
+ close_fd(stderr_write_fd)
- output = _read_pipe_to_string(stdout_pipe)
- stdout_pipe.set_output_only()
-
- var process = Process(Int(pid))
- var status = process.wait()
- if not status.exit_code or status.exit_code.value() != 0:
- raise Error("hyf process exited unexpectedly")
+ var write_reason = write_fd_bounded(
+ stdin_write_fd, request_json + "\n", deadline_ms
+ )
+ close_fd(stdin_write_fd)
+ if write_reason != "":
+ _terminate_and_raise(
+ pid, -1, stdout_read_fd, stderr_read_fd, "write_" + write_reason
+ )
+ var output = ""
+ try:
+ output = read_all_bounded(
+ stdout_read_fd, STDIO_MAX_STDOUT_BYTES, deadline_ms
+ )
+ except:
+ _terminate_and_raise(
+ pid,
+ -1,
+ stdout_read_fd,
+ stderr_read_fd,
+ "stdout_bounded_read_failed",
+ )
+ close_fd(stdout_read_fd)
+
+ var diagnostics = _diagnostics_or_empty(stderr_read_fd)
+ close_fd(stderr_read_fd)
+
+ var st = wait_bounded(pid, deadline_ms)
+ if not st.reaped():
+ var term = terminate_owned(pid, TERMINATION_GRACE_MS)
+ raise Error("stdio-entrypoint timeout (child " + term.describe() + ")")
+ if st.exited and st.exit_code == 127:
+ raise Error(
+ "stdio-entrypoint exec_failed (stdout="
+ + output
+ + " stderr="
+ + diagnostics
+ + ")"
+ )
+ if not st.exited or st.exit_code != 0:
+ raise Error(
+ "stdio-entrypoint child_failed ("
+ + st.describe()
+ + " stderr="
+ + diagnostics
+ + ")"
+ )
if output == "":
raise Error("hyf process returned no stdout payload")
return loads(output)
+def _diagnostics_or_empty(fd: Int) -> String:
+ try:
+ return read_all_bounded(
+ fd, STDIO_MAX_STDERR_BYTES, TERMINATION_GRACE_MS
+ )
+ except:
+ return ""
+
+
def run_hyf_stdio(request_json: String) raises -> Value:
var response = Value(None)
with SafeTempDir() as temp_dir:
diff --git a/tests/stdio_stall_entrypoint.mojo b/tests/stdio_stall_entrypoint.mojo
@@ -0,0 +1,8 @@
+"""Test-only stdio entrypoint that stalls to exercise parent deadlines."""
+
+from std.ffi import c_int, external_call
+
+
+def main():
+ for _ in range(200):
+ _ = external_call["usleep", c_int](c_int(1_000_000))
diff --git a/tests/strict_fixture.mojo b/tests/strict_fixture.mojo
@@ -1,39 +1,229 @@
-"""Strict bounded HTTP request framing for local test fixtures (ADR-0012 D29).
+"""Strict bounded HTTP request framing and scripted exchanges for local test
+fixtures (ADR-0012 D29, ADR-0014 D33 FX01-FX05).
-Bounded byte-oriented accumulation and header parsing *before* UTF-8 decoding,
-validated nonnegative ``Content-Length``, exact header-name matching,
-duplicate/conflicting length rejection, premature-EOF rejection, bounded
-unsupported-transfer rejection and surplus retention for persistent
-connections.
+This module owns the *test fixture* wire contract only. It deliberately does
+not change HYF's product HTTP/TLS policy or timeouts.
+
+Framing contract (FX03):
+
+* bounded byte-oriented accumulation before any UTF-8 decode
+* header cap 65,536 bytes and body cap 1,048,576 bytes (ADR-0010 D21)
+* lexical ``Content-Length``: nonempty ASCII digits only, with an explicit
+ overflow guard; no sign, whitespace or non-digit
+* request line must be ``METHOD SP target SP HTTP/1.1|HTTP/1.0``
+* malformed header syntax (whitespace before ``:``, obs-fold, control bytes,
+ non-token name bytes) is rejected rather than silently stripped
+* duplicate/conflicting framing and unsupported transfer encodings rejected
+* premature EOF and oversize rejected before unbounded accumulation
+* surplus bytes after a frame are retained for the next frame
+
+Exchange contract (FX01/FX02/FX04/FX05): explicit ordered scripts carrying the
+expected method/path/selected headers/body plus the scripted status/headers/
+body/delay and connection semantics. Request and connection counters are
+distinct and verified.
"""
from std.collections import List
+from flare.tcp import TcpListener
from flare.tcp import TcpStream
+from flare.utils import usleep
+
comptime STRICT_MAX_HEADER_BYTES = 65536
comptime STRICT_MAX_BODY_BYTES = 1048576
+comptime STRICT_MAX_READY_BYTES = 256
+comptime STRICT_MAX_REPORT_BYTES = 2048
+comptime STRICT_COMPLETION_GRACE_MS = 100
+
+
+# ── JSON-safe escaping (FX04) ───────────────────────────────────────────────
+
+
+def _hex_digit(value: Int) -> UInt8:
+ if value < 10:
+ return UInt8(48 + value)
+ return UInt8(97 + (value - 10))
+
+
+def json_escape(value: String) -> String:
+ """Return ``value`` as a JSON string literal with full control escaping."""
+ var out = List[UInt8]()
+ out.append(UInt8(34))
+ for byte in value.as_bytes():
+ var b = Int(byte)
+ if b == 34:
+ out.append(UInt8(92))
+ out.append(UInt8(34))
+ elif b == 92:
+ out.append(UInt8(92))
+ out.append(UInt8(92))
+ elif b == 8:
+ out.append(UInt8(92))
+ out.append(UInt8(98))
+ elif b == 12:
+ out.append(UInt8(92))
+ out.append(UInt8(102))
+ elif b == 10:
+ out.append(UInt8(92))
+ out.append(UInt8(110))
+ elif b == 13:
+ out.append(UInt8(92))
+ out.append(UInt8(114))
+ elif b == 9:
+ out.append(UInt8(92))
+ out.append(UInt8(116))
+ elif b < 32 or b == 127:
+ out.append(UInt8(92))
+ out.append(UInt8(117))
+ out.append(UInt8(48))
+ out.append(UInt8(48))
+ out.append(_hex_digit(b // 16))
+ out.append(_hex_digit(b % 16))
+ else:
+ out.append(UInt8(b))
+ out.append(UInt8(34))
+ return String(unsafe_from_utf8=Span(ptr=out.unsafe_ptr(), length=len(out)))
+
+
+# ── Header access (FX03/FX04) ───────────────────────────────────────────────
+
+
+def header_values(headers_raw: String, name: String) -> List[String]:
+ """Return every value for an exactly named header (case-insensitive name).
+
+ The header name is compared byte-exactly after lowercasing; malformed
+ spacing was already rejected by the framer, so no whitespace stripping of
+ the name token occurs here.
+ """
+ var out = List[String]()
+ var lines = headers_raw.split("\r\n")
+ var wanted = name.lower()
+ for i in range(1, len(lines)):
+ var line = String(lines[i])
+ var colon = line.find(":")
+ if colon <= 0:
+ continue
+ if String(line[byte=0:colon]).lower() == wanted:
+ out.append(String(line[byte = colon + 1 :].strip()))
+ return out^
+
+
+def header_value(headers_raw: String, name: String) -> String:
+ var values = header_values(headers_raw, name)
+ if len(values) != 1:
+ return ""
+ return values[0]
+
+
+def bearer_token(headers_raw: String) -> String:
+ var values = header_values(headers_raw, "authorization")
+ if len(values) != 1:
+ return ""
+ if not values[0].startswith("Bearer "):
+ return ""
+ return String(values[0][byte=7:])
+
+
+def authorization_reason(headers_raw: String, required: Bool) -> String:
+ """Validate the exact ``Authorization`` header, rejecting spoofs (FX04)."""
+ if not required:
+ return ""
+ var values = header_values(headers_raw, "authorization")
+ if len(values) == 0:
+ return "auth_missing"
+ if len(values) > 1:
+ return "auth_duplicate"
+ if not values[0].startswith("Bearer ") or values[0].byte_length() <= 7:
+ return "auth_invalid"
+ return ""
+
+# ── Framing ─────────────────────────────────────────────────────────────────
+
+@fieldwise_init
struct FramedRequest(Movable):
var ok: Bool
var error: String
var method: String
var path: String
+ var version: String
var headers_raw: String
var body: String
+ var content_length: Int
var keep_alive: Bool
var total_bytes: Int
- def __init__(out self):
- self.ok = False
- self.error = ""
- self.method = ""
- self.path = ""
- self.headers_raw = ""
- self.body = ""
- self.keep_alive = False
- self.total_bytes = 0
+
+def _token_byte(value: Int) -> Bool:
+ if value >= 48 and value <= 57:
+ return True
+ if value >= 65 and value <= 90:
+ return True
+ if value >= 97 and value <= 122:
+ return True
+ # RFC 7230 token punctuation.
+ return (
+ value == 33
+ or value == 35
+ or value == 36
+ or value == 37
+ or value == 38
+ or value == 39
+ or value == 42
+ or value == 43
+ or value == 45
+ or value == 46
+ or value == 94
+ or value == 95
+ or value == 96
+ or value == 124
+ or value == 126
+ )
+
+
+def _valid_method(method: String) -> Bool:
+ if method.byte_length() == 0:
+ return False
+ for byte in method.as_bytes():
+ var b = Int(byte)
+ if b < 65 or b > 90:
+ return False
+ return True
+
+
+def _valid_version(version: String) -> Bool:
+ return version == "HTTP/1.1" or version == "HTTP/1.0"
+
+
+def _valid_header_name(name: String) -> Bool:
+ if name.byte_length() == 0:
+ return False
+ for byte in name.as_bytes():
+ if not _token_byte(Int(byte)):
+ return False
+ return True
+
+
+def _valid_header_value(value: String) -> Bool:
+ for byte in value.as_bytes():
+ var b = Int(byte)
+ if b == 9:
+ continue
+ if b < 32 or b == 127:
+ return False
+ return True
+
+
+def _ascii_digits(value: String) -> Bool:
+ if value.byte_length() == 0:
+ return False
+ for byte in value.as_bytes():
+ var b = Int(byte)
+ if b < 48 or b > 57:
+ return False
+ return True
def _bytes_string(bytes: List[UInt8], stop: Int) raises -> String:
@@ -50,6 +240,9 @@ struct ConnectionReader(Movable):
self._buffer = List[UInt8]()
self._eof = False
+ def has_buffered(self) -> Bool:
+ return len(self._buffer) > 0
+
def _read_more(mut self) raises -> Int:
var chunk = InlineArray[Byte, 4096](fill=0)
var n = self._stream.read(chunk.unsafe_ptr(), 4096)
@@ -77,7 +270,18 @@ struct ConnectionReader(Movable):
return -1
def read(mut self) raises -> FramedRequest:
- var outcome = FramedRequest()
+ var outcome = FramedRequest(
+ ok=False,
+ error="",
+ method="",
+ path="",
+ version="",
+ headers_raw="",
+ body="",
+ content_length=0,
+ keep_alive=False,
+ total_bytes=0,
+ )
var header_end = self._header_end()
while header_end < 0:
if self._eof:
@@ -90,6 +294,9 @@ struct ConnectionReader(Movable):
outcome.error = "header_too_large"
return outcome^
_ = self._read_more()
+ if len(self._buffer) > STRICT_MAX_HEADER_BYTES + 4:
+ outcome.error = "header_too_large"
+ return outcome^
header_end = self._header_end()
if header_end > STRICT_MAX_HEADER_BYTES:
outcome.error = "header_too_large"
@@ -107,36 +314,63 @@ struct ConnectionReader(Movable):
return outcome^
outcome.method = String(parts[0])
outcome.path = String(parts[1])
+ outcome.version = String(parts[2])
+ if not _valid_method(outcome.method):
+ outcome.error = "malformed_method"
+ return outcome^
+ if not _valid_version(outcome.version):
+ outcome.error = "malformed_version"
+ return outcome^
+ if outcome.path.byte_length() == 0:
+ outcome.error = "malformed_target"
+ return outcome^
- var content_length = -1
+ var content_length = 0
var have_length = False
var have_transfer = False
for i in range(1, len(lines)):
var line = String(lines[i])
if line.byte_length() == 0:
continue
+ var first = Int(line.as_bytes()[0])
+ if first == 32 or first == 9:
+ outcome.error = "malformed_header"
+ return outcome^
var colon = line.find(":")
- if colon < 0:
+ if colon <= 0:
+ outcome.error = "malformed_header"
+ return outcome^
+ var name = String(line[byte=0:colon])
+ var value = String(line[byte = colon + 1 :].strip())
+ if not _valid_header_name(name) or not _valid_header_value(value):
outcome.error = "malformed_header"
return outcome^
- var name = String(line[byte=0:colon]).lower().strip()
- var value = String(line[byte = colon + 1 :]).strip()
- if name == "content-length":
+ var lower = name.lower()
+ if lower == "content-length":
if have_length:
outcome.error = "duplicate_content_length"
return outcome^
+ if not _ascii_digits(value):
+ outcome.error = "malformed_content_length"
+ return outcome^
+ if value.byte_length() > 10:
+ outcome.error = "content_length_overflow"
+ return outcome^
var parsed = Int(value)
- if parsed < 0:
- outcome.error = "negative_content_length"
+ if parsed > STRICT_MAX_BODY_BYTES:
+ outcome.error = "body_too_large"
return outcome^
content_length = parsed
have_length = True
- elif name == "transfer-encoding":
+ elif lower == "transfer-encoding":
+ if have_transfer:
+ outcome.error = "duplicate_transfer_encoding"
+ return outcome^
have_transfer = True
if value.lower() != "identity":
outcome.error = "unsupported_transfer_encoding"
return outcome^
- elif name == "connection" and value.lower() == "keep-alive":
+ elif lower == "connection" and value.lower() == "keep-alive":
outcome.keep_alive = True
if have_transfer and have_length:
@@ -144,9 +378,7 @@ struct ConnectionReader(Movable):
return outcome^
if not have_length:
content_length = 0
- if content_length > STRICT_MAX_BODY_BYTES:
- outcome.error = "body_too_large"
- return outcome^
+ outcome.content_length = content_length
var expected_total = header_end + 4 + content_length
while len(self._buffer) < expected_total:
@@ -167,23 +399,363 @@ struct ConnectionReader(Movable):
outcome.total_bytes = expected_total
return outcome^
+ def probe_completion(mut self, grace_ms: Int) -> String:
+ """Bounded completion handshake after the final expected exchange.
-def header_value(headers_raw: String, name: String) -> String:
- """Return the value of an exactly named header, or "" when absent."""
- var lines = headers_raw.split("\r\n")
- var wanted = name.lower()
- for i in range(1, len(lines)):
- var line = String(lines[i])
- var colon = line.find(":")
- if colon < 0:
+ Any already-buffered or subsequently received bytes mean the client
+ sent an extra exchange. A timeout with no bytes is success; this never
+ waits indefinitely for hypothetical future requests.
+ """
+ if len(self._buffer) > 0:
+ return "extra_exchange_after_completion"
+ try:
+ self._stream.set_recv_timeout(grace_ms)
+ except:
+ return ""
+ try:
+ var n = self._read_more()
+ if n > 0:
+ return "extra_exchange_after_completion"
+ except:
+ return ""
+ return ""
+
+
+# ── Scripted exchanges (FX01/FX02/FX04/FX05) ────────────────────────────────
+
+
+@fieldwise_init
+struct ExchangeScript(Copyable, Movable):
+ var case_label: String
+ var method: String
+ var path: String
+ var headers: String
+ var body: String
+ var check_body: Bool
+ var require_bearer: Bool
+ var status: Int
+ var response_headers: String
+ var response_body: String
+ var raw_response: String
+ var delay_ms: Int
+ var close_connection: Bool
+ var echo_authorization: Bool
+
+ def __copyinit__(out self, existing: Self):
+ self.case_label = existing.case_label
+ self.method = existing.method
+ self.path = existing.path
+ self.headers = existing.headers
+ self.body = existing.body
+ self.check_body = existing.check_body
+ self.require_bearer = existing.require_bearer
+ self.status = existing.status
+ self.response_headers = existing.response_headers
+ self.response_body = existing.response_body
+ self.raw_response = existing.raw_response
+ self.delay_ms = existing.delay_ms
+ self.close_connection = existing.close_connection
+ self.echo_authorization = existing.echo_authorization
+
+
+def exchange_script(
+ case_label: String,
+ method: String,
+ path: String,
+ status: Int,
+ body: String,
+) -> ExchangeScript:
+ return ExchangeScript(
+ case_label=case_label,
+ method=method,
+ path=path,
+ headers="",
+ body="",
+ check_body=False,
+ require_bearer=False,
+ status=status,
+ response_headers="",
+ response_body=body,
+ raw_response="",
+ delay_ms=0,
+ close_connection=True,
+ echo_authorization=False,
+ )
+
+
+def verify_exchange(script: ExchangeScript, framed: FramedRequest) -> String:
+ """Return ``""`` when the request matches the script, else a bounded reason.
+ """
+ if framed.method != script.method:
+ return "method_mismatch"
+ if framed.path != script.path:
+ return "path_mismatch"
+ if script.check_body and framed.body != script.body:
+ return "body_mismatch"
+ var expected_headers = script.headers.split("\n")
+ for entry in expected_headers:
+ var expected = String(entry).strip()
+ if expected.byte_length() == 0:
+ continue
+ var split = expected.find(":")
+ if split <= 0:
continue
- if String(line[byte=0:colon]).lower().strip() == wanted:
- return String(String(line[byte = colon + 1 :]).strip())
+ var name = String(expected[byte=0:split])
+ var value = String(expected[byte = split + 1 :].strip())
+ var values = header_values(framed.headers_raw, name)
+ if len(values) == 0:
+ return "header_missing:" + name
+ if len(values) > 1:
+ return "header_duplicate:" + name
+ if values[0] != value:
+ return "header_mismatch:" + name
+ var auth = authorization_reason(framed.headers_raw, script.require_bearer)
+ if auth != "":
+ return auth
return ""
-def bearer_token(headers_raw: String) -> String:
- var value = header_value(headers_raw, "authorization")
- if not value.startswith("Bearer "):
- return ""
- return String(value[byte=7:])
+def _substitute(
+ text: String, request_index: Int, connection_index: Int
+) -> String:
+ var out = text.replace("{request_index}", String(request_index))
+ return out.replace("{connection_index}", String(connection_index))
+
+
+def render_response(
+ script: ExchangeScript,
+ request_headers_raw: String,
+ request_index: Int,
+ connection_index: Int,
+) -> String:
+ if script.raw_response != "":
+ return String(script.raw_response)
+ var reason = "OK"
+ if script.status == 401:
+ reason = "Unauthorized"
+ elif script.status == 404:
+ reason = "Not Found"
+ elif script.status == 429:
+ reason = "Too Many Requests"
+ elif script.status == 500:
+ reason = "Internal Server Error"
+ elif script.status == 503:
+ reason = "Service Unavailable"
+ elif script.status == 529:
+ reason = "Overloaded"
+ var body = _substitute(
+ script.response_body, request_index, connection_index
+ )
+ if script.echo_authorization:
+ body = (
+ '{"authorization":'
+ + json_escape(bearer_token(request_headers_raw))
+ + "}"
+ )
+ var extra = ""
+ if script.response_headers != "":
+ extra = script.response_headers + "\r\n"
+ var connection = "keep-alive" if not script.close_connection else "close"
+ return (
+ "HTTP/1.1 "
+ + String(script.status)
+ + " "
+ + reason
+ + "\r\ncontent-type: application/json\r\n"
+ + extra
+ + "content-length: "
+ + String(body.byte_length())
+ + "\r\nconnection: "
+ + connection
+ + "\r\n\r\n"
+ + body
+ )
+
+
+# ── Convenience-mode validation + reports ────────────────────────────────────
+
+
+def serve_scripts(
+ listener: TcpListener, var scripts: List[ExchangeScript], label: String
+) raises -> ServeReport:
+ """Serve an ordered list of explicit exchange scripts (FX01/FX02/FX05).
+
+ Every script must be consumed in order on its expected method/path/
+ selected-headers/body; unexpected, extra, missing and unconsumed exchanges
+ fail with a bounded case-specific reason. A 404-then-success outcome is
+ never a pass.
+ """
+ var request_count = 0
+ var connection_count = 0
+ var total = len(scripts)
+ try:
+ while request_count < total:
+ var stream = listener.accept()
+ connection_count += 1
+ var reader = ConnectionReader(stream^)
+ while request_count < total:
+ var framed = reader.read()
+ if not framed.ok:
+ if framed.error == "empty":
+ if request_count < total:
+ return ServeReport(
+ False,
+ "accounting",
+ label,
+ "missing_exchanges",
+ request_count,
+ connection_count,
+ )
+ break
+ return ServeReport(
+ False,
+ "read",
+ label,
+ framed.error,
+ request_count,
+ connection_count,
+ )
+ var script = scripts[request_count].copy()
+ var verify = verify_exchange(script, framed)
+ if verify != "":
+ return ServeReport(
+ False,
+ "exchange",
+ script.case_label,
+ verify,
+ request_count,
+ connection_count,
+ )
+ var next_index = request_count + 1
+ if next_index == total:
+ var extra = reader.probe_completion(
+ STRICT_COMPLETION_GRACE_MS
+ )
+ if extra != "":
+ return ServeReport(
+ False,
+ "accounting",
+ script.case_label,
+ extra,
+ request_count,
+ connection_count,
+ )
+ request_count = next_index
+ if script.delay_ms > 0:
+ usleep(script.delay_ms * 1000)
+ reader.write_all(
+ render_response(
+ script,
+ framed.headers_raw,
+ request_count,
+ connection_count,
+ )
+ )
+ if script.close_connection:
+ break
+ if request_count < total:
+ return ServeReport(
+ False,
+ "accounting",
+ scripts[request_count].case_label,
+ "missing_exchanges",
+ request_count,
+ connection_count,
+ )
+ return ServeReport(
+ True, "complete", label, "ok", request_count, connection_count
+ )
+ except:
+ return ServeReport(
+ False, "read", label, "io_error", request_count, connection_count
+ )
+
+
+def validate_convenience(
+ framed: FramedRequest,
+ allowed_paths: List[String],
+ allowed_methods: List[String],
+ require_bearer: Bool,
+) -> String:
+ """Validate route then method (and auth) before any response.
+
+ ``allowed_methods`` is parallel to ``allowed_paths``. Wrong routes,
+ methods and spoofed auth never count as a successful intended exchange
+ (FX02/FX04).
+ """
+ var matched = -1
+ for index in range(len(allowed_paths)):
+ if framed.path == allowed_paths[index]:
+ matched = index
+ if matched < 0:
+ return "unexpected_path"
+ if matched >= len(allowed_methods):
+ return "unexpected_method"
+ if framed.method != allowed_methods[matched]:
+ return "unexpected_method"
+ return authorization_reason(framed.headers_raw, require_bearer)
+
+
+@fieldwise_init
+struct ServeReport(Movable):
+ var ok: Bool
+ var phase: String
+ var case_label: String
+ var reason: String
+ var requests: Int
+ var connections: Int
+
+ def describe(self) -> String:
+ return (
+ "phase="
+ + self.phase
+ + " case="
+ + self.case_label
+ + " reason="
+ + self.reason
+ + " requests="
+ + String(self.requests)
+ + " connections="
+ + String(self.connections)
+ )
+
+
+def report_line(report: ServeReport) -> String:
+ var state = "ok" if report.ok else "fail"
+ return "result " + state + " " + report.describe()
+
+
+def _safe_int(value: String) -> Int:
+ try:
+ return Int(value)
+ except:
+ return 0
+
+
+def parse_report(line: String) -> ServeReport:
+ var report = ServeReport(False, "unknown", "-", "missing_report", 0, 0)
+ if not line.startswith("result "):
+ return report^
+ var remainder = String(line[byte=7:])
+ var parts = remainder.split(" ")
+ if len(parts) == 0:
+ return report^
+ report.ok = String(parts[0]) == "ok"
+ for index in range(1, len(parts)):
+ var field = String(parts[index])
+ var eq = field.find("=")
+ if eq <= 0:
+ continue
+ var key = String(field[byte=0:eq])
+ var value = String(field[byte = eq + 1 :])
+ if key == "phase":
+ report.phase = value
+ elif key == "case":
+ report.case_label = value
+ elif key == "reason":
+ report.reason = value
+ elif key == "requests":
+ report.requests = _safe_int(value)
+ elif key == "connections":
+ report.connections = _safe_int(value)
+ return report^
diff --git a/tests/test_provider_helpers.mojo b/tests/test_provider_helpers.mojo
@@ -1,37 +1,40 @@
-from std.testing import TestSuite, assert_true
+"""Strict fixture/script verification tests (ADR-0012 D29 / ADR-0014 D33).
-from flare.net import SocketAddr
-from flare.tcp import TcpStream
+Covers FX01-FX08 for both providers plus the in-memory mutation controls that
+prove a negative control fails for the intended cause rather than for any
+exception, signal or unrelated startup failure.
+"""
+
+from std.testing import TestSuite, assert_true, assert_equal
+from flare.net import SocketAddr
+from flare.tcp import TcpListener, TcpStream
+
+from parent_lifecycle import pid_not_waitable
+from strict_fixture import (
+ ExchangeScript,
+ FramedRequest,
+ authorization_reason,
+ exchange_script,
+ json_escape,
+ verify_exchange,
+)
from max_local_process_helper import (
SpawnedMaxLocalStub,
+ reserve_loopback_port,
+ spawn_max_local_scripted,
spawn_max_local_stub,
)
-from jev_provider_helper import spawn_jev_stub_auto
+from jev_provider_helper import (
+ spawn_jev_scripted_auto,
+ spawn_jev_stub_auto,
+)
-def _client_request(
- port: Int, path: String, body: String, auth: String = ""
-) raises -> String:
- var client = TcpStream.connect(SocketAddr.localhost(UInt16(port)))
- var auth_line = "" if auth == "" else "authorization: " + auth + "\r\n"
- var headers = (
- "POST "
- + path
- + " HTTP/1.1\r\nhost: 127.0.0.1\r\n"
- + auth_line
- + "content-type: application/json\r\ncontent-length: "
- + String(body.byte_length())
- + "\r\nconnection: close\r\n\r\n"
- )
- client.write_all(Span[UInt8, _](headers.as_bytes()))
- var sent = 0
- while sent < body.byte_length():
- var end = sent + 3000
- if end > body.byte_length():
- end = body.byte_length()
- client.write_all(Span[UInt8, _](body[byte=sent:end].as_bytes()))
- sent = end
+# ── Client helpers ──────────────────────────────────────────────────────────
+
+
+def _read_all(mut client: TcpStream) raises -> String:
var buffer = InlineArray[Byte, 4096](fill=0)
var response = String("")
while True:
@@ -41,16 +44,93 @@ def _client_request(
response += String(
unsafe_from_utf8=Span(ptr=buffer.unsafe_ptr(), length=Int(n))
)
+ return response^
+
+
+def _write_all(mut client: TcpStream, text: String) raises:
+ client.write_all(Span[UInt8, _](text.as_bytes()))
+
+
+def _raw_exchange(port: Int, raw: String) raises -> String:
+ var client = TcpStream.connect(SocketAddr.localhost(UInt16(port)))
+ _write_all(client, raw)
+ var response = _read_all(client)
client.close()
return response^
+def _raw_send_only(port: Int, raw: String) raises:
+ var client = TcpStream.connect(SocketAddr.localhost(UInt16(port)))
+ _write_all(client, raw)
+ client.close()
+
+
+def _read_one_response(mut client: TcpStream) raises -> String:
+ var data = String("")
+ var buf = InlineArray[Byte, 1024](fill=0)
+ while data.find("\r\n\r\n") < 0:
+ var n = client.read(buf.unsafe_ptr(), 1024)
+ if n <= 0:
+ return data^
+ data += String(
+ unsafe_from_utf8=Span(ptr=buf.unsafe_ptr(), length=Int(n))
+ )
+ var header_end = data.find("\r\n\r\n")
+ var length = 0
+ for line in data[byte=0:header_end].split("\r\n"):
+ var entry = String(line)
+ if entry.lower().startswith("content-length:"):
+ length = Int(String(entry[byte=15:]).strip())
+ var body_start = header_end + 4
+ while data.byte_length() - body_start < length:
+ var n = client.read(buf.unsafe_ptr(), 1024)
+ if n <= 0:
+ break
+ data += String(
+ unsafe_from_utf8=Span(ptr=buf.unsafe_ptr(), length=Int(n))
+ )
+ return data^
+
+
+def _request(
+ port: Int,
+ method: String,
+ path: String,
+ body: String,
+ extra: String = "",
+ connection: String = "close",
+) raises -> String:
+ var raw = (
+ method
+ + " "
+ + path
+ + " HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ + extra
+ + "content-type: application/json\r\ncontent-length: "
+ + String(body.byte_length())
+ + "\r\nconnection: "
+ + connection
+ + "\r\n\r\n"
+ + body
+ )
+ return _raw_exchange(port, raw)
+
+
+def _default_script() -> ExchangeScript:
+ return exchange_script(
+ "expect_ok", "POST", "/v1/chat/completions", 200, '{"ok":true}'
+ )
+
+
+# ── Existing convenience-mode coverage ──────────────────────────────────────
+
+
def test_max_local_stub_reads_fragmented_large_body() raises:
var stub = spawn_max_local_stub(0, "echo_body_bytes", 1)
var body = String("")
for _ in range(9000):
body += "x"
- var response = _client_request(stub.port, "/v1/chat/completions", body)
+ var response = _request(stub.port, "POST", "/v1/chat/completions", body)
assert_true(response.find('"received_bytes":9000') >= 0)
stub.wait()
@@ -59,167 +139,577 @@ def test_max_local_stub_counts_every_wire_attempt() raises:
var requests = 3
var stub = spawn_max_local_stub(0, "count_requests", requests)
for index in range(requests):
- var response = _client_request(stub.port, "/v1/chat/completions", "{}")
+ var response = _request(stub.port, "POST", "/v1/chat/completions", "{}")
assert_true(response.find('"request_index":' + String(index + 1)) >= 0)
stub.wait()
def test_max_local_stub_rejects_unknown_path() raises:
+ # FX02/FX04: an unexpected route must fail fixture verification, not be
+ # answered 404 and then reported as a successful stub run.
var stub = spawn_max_local_stub(0, "query_rewrite_ok", 1)
- var response = _client_request(stub.port, "/not-a-route", "{}")
- assert_true(response.find("404") >= 0)
- assert_true(response.find('"not_found"') >= 0)
- stub.wait()
+ var response = _request(stub.port, "POST", "/not-a-route", "{}")
+ assert_true(response.find("404") < 0)
+ stub.reap()
+ assert_true(not stub.ok())
+ assert_equal(stub.phase(), "exchange")
+ assert_equal(stub.reason(), "unexpected_path")
def test_max_local_stub_binds_and_reports_port() raises:
var stub = spawn_max_local_stub(0, "count_requests", 1)
assert_true(stub.port > 0)
- var response = _client_request(stub.port, "/v1/chat/completions", "{}")
+ var response = _request(stub.port, "POST", "/v1/chat/completions", "{}")
assert_true(response.find('"request_index":1') >= 0)
stub.wait()
def test_jev_stub_observes_bearer_sentinel_at_intended_origin() raises:
var started = spawn_jev_stub_auto("echo_authorization", 1)
- var response = _client_request(
- started.port, "/v1/systemone", "{}", "Bearer hyf-sentinel-token"
+ var response = _request(
+ started.port,
+ "POST",
+ "/v1/systemone",
+ "{}",
+ "authorization: Bearer hyf-sentinel-token\r\n",
)
assert_true(response.find("hyf-sentinel-token") >= 0)
started.stub.wait()
-def _client_send_raw(port: Int, data: String) raises -> String:
- var client = TcpStream.connect(SocketAddr.localhost(UInt16(port)))
- client.write_all(Span[UInt8, _](data.as_bytes()))
- var buffer = InlineArray[Byte, 4096](fill=0)
- var response = String("")
- while True:
- var n = client.read(buffer.unsafe_ptr(), 4096)
- if n <= 0:
- break
- response += String(
- unsafe_from_utf8=Span(ptr=buffer.unsafe_ptr(), length=Int(n))
- )
- client.close()
- return response^
-
-
-def _client_two_on_one(port: Int, path: String) raises -> String:
- var client = TcpStream.connect(SocketAddr.localhost(UInt16(port)))
- var body = "{}"
- var first = (
- "POST "
- + path
- + " HTTP/1.1\r\nhost: 127.0.0.1\r\ncontent-type:"
- " application/json\r\ncontent-length: "
- + String(body.byte_length())
- + "\r\nconnection: keep-alive\r\n\r\n"
- + body
- )
- var second = (
- "POST "
- + path
- + " HTTP/1.1\r\nhost: 127.0.0.1\r\ncontent-type:"
- " application/json\r\ncontent-length: "
- + String(body.byte_length())
- + "\r\nconnection: close\r\n\r\n"
- + body
+def test_max_local_stub_stalled_child_is_reaped() raises:
+ var stub = spawn_max_local_stub(0, "stall", 1)
+ _raw_send_only(
+ stub.port,
+ (
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: 2\r\nconnection: close\r\n\r\n{}"
+ ),
)
- client.write_all(Span[UInt8, _](first.as_bytes()))
- client.write_all(Span[UInt8, _](second.as_bytes()))
- var buffer = InlineArray[Byte, 4096](fill=0)
- var response = String("")
- while True:
- var n = client.read(buffer.unsafe_ptr(), 4096)
- if n <= 0:
- break
- response += String(
- unsafe_from_utf8=Span(ptr=buffer.unsafe_ptr(), length=Int(n))
- )
- client.close()
- return response^
+ stub.terminate()
+ assert_true(pid_not_waitable(stub.pid))
-def _stub_failed(mut stub: SpawnedMaxLocalStub) raises -> Bool:
- try:
- stub.wait()
- return False
- except:
- return True
+# ── FX01: explicit ordered scripted exchanges ───────────────────────────────
-def _client_send_only(port: Int, data: String) raises:
- var client = TcpStream.connect(SocketAddr.localhost(UInt16(port)))
- client.write_all(Span[UInt8, _](data.as_bytes()))
- client.close()
+def test_max_local_scripted_matches_explicit_exchange() raises:
+ var scripts = List[ExchangeScript]()
+ var script = exchange_script(
+ "explicit_ok", "POST", "/v1/chat/completions", 201, '{"made":"yes"}'
+ )
+ script.headers = "x-sentinel:explicit"
+ script.check_body = True
+ script.body = '{"q":"apples"}'
+ script.response_headers = "x-scripted: yes"
+ script.delay_ms = 20
+ scripts.append(script^)
+ var stub = spawn_max_local_scripted(0, scripts^)
+ var response = _request(
+ stub.port,
+ "POST",
+ "/v1/chat/completions",
+ '{"q":"apples"}',
+ "x-sentinel: explicit\r\n",
+ )
+ assert_true(response.find("201") >= 0)
+ assert_true(response.find("x-scripted: yes") >= 0)
+ assert_true(response.find('{"made":"yes"}') >= 0)
+ stub.wait()
+ assert_equal(stub.request_count(), 1)
+ assert_equal(stub.connection_count(), 1)
-def test_max_local_strict_rejects_truncated_frame() raises:
- var stub = spawn_max_local_stub(0, "echo_body_bytes", 1)
- _client_send_only(
+def test_max_local_scripted_rejects_wrong_method() raises:
+ var scripts = List[ExchangeScript]()
+ scripts.append(_default_script())
+ var stub = spawn_max_local_scripted(0, scripts^)
+ _raw_send_only(
stub.port,
(
- "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
- "content-length: 100\r\nconnection: close\r\n\r\nshort"
+ "GET /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: 2\r\nconnection: close\r\n\r\n{}"
),
)
- assert_true(_stub_failed(stub))
+ stub.reap()
+ assert_equal(stub.phase(), "exchange")
+ assert_equal(stub.reason(), "method_mismatch")
-def test_max_local_strict_rejects_duplicate_content_length() raises:
- var stub = spawn_max_local_stub(0, "echo_body_bytes", 1)
- _client_send_only(
+def test_max_local_scripted_rejects_wrong_path() raises:
+ var scripts = List[ExchangeScript]()
+ scripts.append(_default_script())
+ var stub = spawn_max_local_scripted(0, scripts^)
+ _raw_send_only(
+ stub.port,
+ (
+ "POST /v1/other HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: 2\r\nconnection: close\r\n\r\n{}"
+ ),
+ )
+ stub.reap()
+ assert_equal(stub.phase(), "exchange")
+ assert_equal(stub.reason(), "path_mismatch")
+
+
+def test_max_local_scripted_rejects_wrong_selected_header() raises:
+ var scripts = List[ExchangeScript]()
+ var script = _default_script()
+ script.headers = "x-sentinel:expected"
+ scripts.append(script^)
+ var stub = spawn_max_local_scripted(0, scripts^)
+ _raw_send_only(
stub.port,
(
"POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
- "content-length: 2\r\ncontent-length: 2\r\n"
- "connection: close\r\n\r\n{}"
+ "x-sentinel: wrong\r\ncontent-length: 2\r\nconnection: close\r\n"
+ "\r\n{}"
+ ),
+ )
+ stub.reap()
+ assert_equal(stub.phase(), "exchange")
+ assert_equal(stub.reason(), "header_mismatch:x-sentinel")
+
+
+def test_max_local_scripted_rejects_wrong_body() raises:
+ var scripts = List[ExchangeScript]()
+ var script = _default_script()
+ script.check_body = True
+ script.body = '{"expected":true}'
+ scripts.append(script^)
+ var stub = spawn_max_local_scripted(0, scripts^)
+ _raw_send_only(
+ stub.port,
+ (
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ 'content-length: 15\r\nconnection: close\r\n\r\n{"wrong":true} '
),
)
- assert_true(_stub_failed(stub))
+ stub.reap()
+ assert_equal(stub.phase(), "exchange")
+ assert_equal(stub.reason(), "body_mismatch")
-def test_max_local_strict_rejects_oversize_body() raises:
- var stub = spawn_max_local_stub(0, "echo_body_bytes", 1)
- _client_send_only(
- stub.port,
+# ── FX02: unexpected/extra/missing/unconsumed accounting ────────────────────
+
+
+def test_scripted_rejects_extra_pipelined_exchange() raises:
+ var scripts = List[ExchangeScript]()
+ var script = _default_script()
+ script.close_connection = False
+ scripts.append(script^)
+ var stub = spawn_max_local_scripted(0, scripts^)
+ var first_frame = (
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: 2\r\nconnection: keep-alive\r\n\r\n{}"
+ )
+ var second_frame = String(first_frame)
+ _raw_send_only(stub.port, first_frame + second_frame)
+ stub.reap()
+ assert_equal(stub.phase(), "accounting")
+ assert_equal(stub.reason(), "extra_exchange_after_completion")
+
+
+def test_scripted_rejects_missing_exchange() raises:
+ var scripts = List[ExchangeScript]()
+ scripts.append(_default_script())
+ var stub = spawn_max_local_scripted(0, scripts^)
+ var client = TcpStream.connect(SocketAddr.localhost(UInt16(stub.port)))
+ client.close()
+ stub.reap()
+ assert_equal(stub.phase(), "accounting")
+ assert_equal(stub.reason(), "missing_exchanges")
+ assert_equal(stub.request_count(), 0)
+
+
+def test_scripted_reports_unconsumed_remaining_scripts() raises:
+ var scripts = List[ExchangeScript]()
+ var first = _default_script()
+ first.close_connection = False
+ scripts.append(first^)
+ scripts.append(_default_script())
+ var stub = spawn_max_local_scripted(0, scripts^)
+ var client = TcpStream.connect(SocketAddr.localhost(UInt16(stub.port)))
+ _write_all(
+ client,
(
"POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
- "content-length: 2097152\r\nconnection: close\r\n\r\n"
+ "content-length: 2\r\nconnection: keep-alive\r\n\r\n{}"
),
)
- assert_true(_stub_failed(stub))
+ var response = _read_one_response(client)
+ client.close()
+ assert_true(response.find('{"ok":true}') >= 0)
+ stub.reap()
+ assert_equal(stub.reason(), "missing_exchanges")
+ assert_equal(stub.request_count(), 1)
-def test_max_local_strict_persistent_exchanges_distinct_counters() raises:
- var stub = spawn_max_local_stub(0, "count_requests", 2)
- var response = _client_two_on_one(stub.port, "/v1/chat/completions")
- assert_true(response.find('"request_index":1') >= 0)
- assert_true(response.find('"request_index":2') >= 0)
- assert_true(response.find('"connection_index":1') >= 0)
- assert_true(response.find('"connection_index":2') < 0)
+# ── FX03: strict lexical framing ────────────────────────────────────────────
+
+
+def _framing_failure(raw: String) raises -> SpawnedMaxLocalStub:
+ var scripts = List[ExchangeScript]()
+ scripts.append(_default_script())
+ var stub = spawn_max_local_scripted(0, scripts^)
+ _raw_send_only(stub.port, raw)
+ stub.reap()
+ return stub^
+
+
+def test_strict_framing_lexical_content_length() raises:
+ var plus = _framing_failure(
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: +2\r\nconnection: close\r\n\r\n{}"
+ )
+ assert_equal(plus.phase(), "read")
+ assert_equal(plus.reason(), "malformed_content_length")
+ var spaced = _framing_failure(
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: 2 3\r\nconnection: close\r\n\r\n{}"
+ )
+ assert_equal(spaced.reason(), "malformed_content_length")
+ var empty = _framing_failure(
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length:\r\nconnection: close\r\n\r\n"
+ )
+ assert_equal(empty.reason(), "malformed_content_length")
+ var non_digit = _framing_failure(
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: 2x\r\nconnection: close\r\n\r\n{}"
+ )
+ assert_equal(non_digit.reason(), "malformed_content_length")
+ var overflow = _framing_failure(
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: 99999999999999999999\r\nconnection: close\r\n\r\n"
+ )
+ assert_equal(overflow.reason(), "content_length_overflow")
+
+
+def test_strict_framing_versions_and_header_syntax() raises:
+ var version = _framing_failure(
+ "POST /v1/chat/completions NONHTTP\r\nhost: 127.0.0.1\r\n"
+ "content-length: 2\r\nconnection: close\r\n\r\n{}"
+ )
+ assert_equal(version.phase(), "read")
+ assert_equal(version.reason(), "malformed_version")
+ var space_name = _framing_failure(
+ "POST /v1/chat/completions HTTP/1.1\r\nhost : 127.0.0.1\r\n"
+ "content-length: 2\r\nconnection: close\r\n\r\n{}"
+ )
+ assert_equal(space_name.reason(), "malformed_header")
+ var obs_fold = _framing_failure(
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ " content-length: 2\r\nconnection: close\r\n\r\n{}"
+ )
+ assert_equal(obs_fold.reason(), "malformed_header")
+
+
+def test_strict_framing_conflicting_and_transfer_modes() raises:
+ var conflicting = _framing_failure(
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: 2\r\ntransfer-encoding: identity\r\n"
+ "connection: close\r\n\r\n{}"
+ )
+ assert_equal(conflicting.reason(), "conflicting_framing")
+ var chunked = _framing_failure(
+ "POST /v1/chat/completions HTTP/1.1\r\nhost:"
+ " 127.0.0.1\r\ntransfer-encoding: chunked\r\nconnection:"
+ " close\r\n\r\n2\r\n{}\r\n0\r\n\r\n"
+ )
+ assert_equal(chunked.reason(), "unsupported_transfer_encoding")
+ var duplicate = _framing_failure(
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "transfer-encoding: identity\r\ntransfer-encoding: identity\r\n"
+ "connection: close\r\n\r\n"
+ )
+ assert_equal(duplicate.reason(), "duplicate_transfer_encoding")
+
+
+def test_strict_framing_split_utf8_body() raises:
+ var scripts = List[ExchangeScript]()
+ var script = exchange_script(
+ "utf8", "POST", "/v1/chat/completions", 200, '{"ok":true}'
+ )
+ script.check_body = True
+ var payload = String("")
+ for _ in range(800):
+ payload += "é"
+ script.body = payload
+ scripts.append(script^)
+ var stub = spawn_max_local_scripted(0, scripts^)
+ var client = TcpStream.connect(SocketAddr.localhost(UInt16(stub.port)))
+ var head = (
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: "
+ + String(payload.byte_length())
+ + "\r\nconnection: close\r\n\r\n"
+ )
+ _write_all(client, head)
+ var payload_bytes = payload.as_bytes()
+ var sent = 0
+ while sent < payload.byte_length():
+ var end = sent + 3
+ if end > payload.byte_length():
+ end = payload.byte_length()
+ client.write_all(payload_bytes[sent:end])
+ sent = end
+ var response = _read_all(client)
+ client.close()
+ assert_true(response.find("200") >= 0)
stub.wait()
-def test_jev_strict_rejects_x_authorization() raises:
- var started = spawn_jev_stub_auto("echo_authorization", 1)
- var response = _client_send_raw(
+def test_strict_framing_surplus_retained_for_second_frame() raises:
+ var scripts = List[ExchangeScript]()
+ var first = exchange_script(
+ "first", "POST", "/v1/chat/completions", 200, '{"n":1}'
+ )
+ first.close_connection = False
+ scripts.append(first^)
+ var second = exchange_script(
+ "second", "POST", "/v1/chat/completions", 200, '{"n":2}'
+ )
+ scripts.append(second^)
+ var stub = spawn_max_local_scripted(0, scripts^)
+ var first_frame = (
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: 2\r\nconnection: keep-alive\r\n\r\n{}"
+ )
+ var second_frame = String(first_frame)
+ var response = _raw_exchange(stub.port, first_frame + second_frame)
+ assert_true(response.find('{"n":1}') >= 0)
+ assert_true(response.find('{"n":2}') >= 0)
+ stub.wait()
+ assert_equal(stub.request_count(), 2)
+ assert_equal(stub.connection_count(), 1)
+
+
+# ── FX04: route/method before auth, exact headers, safe escaping ────────────
+
+
+def test_jev_scripted_wrong_route_auth_not_bypassed() raises:
+ var scripts = List[ExchangeScript]()
+ var script = exchange_script(
+ "jev_expect", "POST", "/v1/systemone", 200, '{"ok":true}'
+ )
+ script.require_bearer = True
+ scripts.append(script^)
+ var started = spawn_jev_scripted_auto(scripts^)
+ _raw_send_only(
started.port,
(
- "POST /v1/systemone HTTP/1.1\r\nhost: 127.0.0.1\r\nx-authorization:"
- " Bearer spoof\r\ncontent-type: application/json\r\ncontent-length:"
- " 2\r\nconnection: close\r\n\r\n{}"
+ "POST /not-a-route HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "x-authorization: Bearer spoof\r\ncontent-length: 2\r\n"
+ "connection: close\r\n\r\n{}"
+ ),
+ )
+ started.stub.reap()
+ assert_equal(started.stub.phase(), "exchange")
+ assert_equal(started.stub.reason(), "path_mismatch")
+
+
+def test_scripted_rejects_duplicate_authorization() raises:
+ var scripts = List[ExchangeScript]()
+ var script = exchange_script(
+ "dup_auth", "POST", "/v1/chat/completions", 200, '{"ok":true}'
+ )
+ script.require_bearer = True
+ scripts.append(script^)
+ var stub = spawn_max_local_scripted(0, scripts^)
+ _raw_send_only(
+ stub.port,
+ (
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "authorization: Bearer one\r\nauthorization: Bearer two\r\n"
+ "content-length: 2\r\nconnection: close\r\n\r\n{}"
),
)
- assert_true(response.find("401") >= 0)
- assert_true(response.find("spoof") < 0)
+ stub.reap()
+ assert_equal(stub.phase(), "exchange")
+ assert_equal(stub.reason(), "auth_duplicate")
+
+
+def test_json_escape_control_characters() raises:
+ assert_equal(json_escape("a\nb\tc"), '"a\\nb\\tc"')
+ assert_equal(json_escape('q"uote'), '"q\\"uote"')
+ assert_equal(json_escape("back\\slash"), '"back\\\\slash"')
+ assert_equal(json_escape("ctl\x01"), '"ctl\\u0001"')
+
+
+def test_verify_exchange_mutation_controls() raises:
+ # Deliberate in-memory mutations: each must fail for its own bounded
+ # reason rather than any generic exception (FX07).
+ var script = exchange_script(
+ "control", "POST", "/v1/chat/completions", 200, "{}"
+ )
+ script.headers = "x-sel:1"
+ script.check_body = True
+ script.body = '{"b":1}'
+ var ok = FramedRequest(
+ ok=True,
+ error="",
+ method="POST",
+ path="/v1/chat/completions",
+ version="HTTP/1.1",
+ headers_raw="host: h\r\nx-sel:1",
+ body='{"b":1}',
+ content_length=7,
+ keep_alive=False,
+ total_bytes=0,
+ )
+ assert_equal(verify_exchange(script, ok), "")
+ var wrong_method = FramedRequest(
+ ok=True,
+ error="",
+ method="PUT",
+ path="/v1/chat/completions",
+ version="HTTP/1.1",
+ headers_raw="host: h\r\nx-sel:1",
+ body='{"b":1}',
+ content_length=7,
+ keep_alive=False,
+ total_bytes=0,
+ )
+ assert_equal(verify_exchange(script, wrong_method), "method_mismatch")
+ var wrong_path = FramedRequest(
+ ok=True,
+ error="",
+ method="POST",
+ path="/v1/other",
+ version="HTTP/1.1",
+ headers_raw="host: h\r\nx-sel:1",
+ body='{"b":1}',
+ content_length=7,
+ keep_alive=False,
+ total_bytes=0,
+ )
+ assert_equal(verify_exchange(script, wrong_path), "path_mismatch")
+ var missing_header = FramedRequest(
+ ok=True,
+ error="",
+ method="POST",
+ path="/v1/chat/completions",
+ version="HTTP/1.1",
+ headers_raw="host: h",
+ body='{"b":1}',
+ content_length=7,
+ keep_alive=False,
+ total_bytes=0,
+ )
+ assert_equal(
+ verify_exchange(script, missing_header), "header_missing:x-sel"
+ )
+ var wrong_body = FramedRequest(
+ ok=True,
+ error="",
+ method="POST",
+ path="/v1/chat/completions",
+ version="HTTP/1.1",
+ headers_raw="host: h\r\nx-sel:1",
+ body='{"b":2}',
+ content_length=7,
+ keep_alive=False,
+ total_bytes=0,
+ )
+ assert_equal(verify_exchange(script, wrong_body), "body_mismatch")
+ assert_equal(
+ authorization_reason(
+ (
+ "POST /v1/chat/completions HTTP/1.1\r\nx-authorization: Bearer"
+ " spoof"
+ ),
+ True,
+ ),
+ "auth_missing",
+ )
+ assert_equal(
+ authorization_reason(
+ "POST /v1/chat/completions HTTP/1.1\r\nauthorization: Bearer t",
+ True,
+ ),
+ "",
+ )
+
+
+# ── FX05: persistent reusable/close connection semantics ────────────────────
+
+
+def test_scripted_persistent_counters_and_close_semantics() raises:
+ var scripts = List[ExchangeScript]()
+ var first = exchange_script(
+ "reusable", "POST", "/v1/systemone", 200, '{"step":1}'
+ )
+ first.close_connection = False
+ first.response_headers = "x-step: one"
+ scripts.append(first^)
+ var second = exchange_script(
+ "close", "POST", "/v1/systemone", 200, '{"step":2}'
+ )
+ second.response_headers = "x-step: two"
+ scripts.append(second^)
+ var started = spawn_jev_scripted_auto(scripts^)
+ var first_frame = (
+ "POST /v1/systemone HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "authorization: Bearer t\r\ncontent-length: 2\r\n"
+ "connection: keep-alive\r\n\r\n{}"
+ )
+ var second_frame = String(first_frame)
+ var response = _raw_exchange(started.port, first_frame + second_frame)
+ assert_true(response.find("x-step: one") >= 0)
+ assert_true(response.find("x-step: two") >= 0)
+ assert_true(response.find("connection: keep-alive") >= 0)
+ assert_true(response.find("connection: close") >= 0)
started.stub.wait()
+ assert_equal(started.stub.request_count(), 2)
+ assert_equal(started.stub.connection_count(), 1)
-def test_max_local_stub_stalled_child_is_reaped() raises:
+# ── FX06/FX07/FX08: parent lifecycle and cause-specific failures ────────────
+
+
+def test_startup_failure_distinct_from_exchange_failure() raises:
+ # Occupy a port so the fixture child cannot bind; the parent must observe
+ # a startup failure (no ready line), not a script/parser rejection.
+ var blocker = TcpListener.bind(SocketAddr.localhost(0))
+ var port = Int(blocker.local_addr().port)
+ var raised = False
+ try:
+ var stub = spawn_max_local_stub(port, "count_requests", 1)
+ stub.terminate()
+ except:
+ raised = True
+ blocker.close()
+ assert_true(raised)
+
+
+def test_owned_child_reaped_after_early_terminate() raises:
+ var stub = spawn_max_local_stub(0, "count_requests", 1)
+ stub.terminate()
+ assert_true(pid_not_waitable(stub.pid))
+
+
+def test_repeated_failures_leave_no_owned_child() raises:
+ for _ in range(3):
+ var scripts = List[ExchangeScript]()
+ scripts.append(_default_script())
+ var stub = spawn_max_local_scripted(0, scripts^)
+ _raw_send_only(
+ stub.port,
+ (
+ "POST /v1/nope HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "content-length: 2\r\nconnection: close\r\n\r\n{}"
+ ),
+ )
+ stub.reap()
+ assert_true(not stub.ok())
+ assert_equal(stub.reason(), "path_mismatch")
+ assert_true(pid_not_waitable(stub.pid))
+
+
+def test_timeout_terminates_and_reaps_stalled_child() raises:
var stub = spawn_max_local_stub(0, "stall", 1)
- _client_send_only(
+ _raw_send_only(
stub.port,
(
"POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
@@ -227,6 +717,7 @@ def test_max_local_stub_stalled_child_is_reaped() raises:
),
)
stub.terminate()
+ assert_true(pid_not_waitable(stub.pid))
def main() raises:
diff --git a/tests/test_repo_local_process_contract.mojo b/tests/test_repo_local_process_contract.mojo
@@ -8,6 +8,7 @@ from stdio_process_helper import (
HYF_PATHS_REPO_LOCAL_ROOT_ENV,
ScopedEnvVar,
run_stdio_entrypoint,
+ run_stdio_entrypoint_with_deadline,
)
@@ -89,5 +90,31 @@ def test_src_main_consumes_repo_local_env_without_outer_wrapper() raises:
)
+def test_run_stdio_entrypoint_reaps_stalled_child_under_deadline() raises:
+ # FX06/FX08: the parent deadline (not the child alarm) must terminate and
+ # reap the owned child, and the raised error must carry the reap result.
+ var message = ""
+ try:
+ _ = run_stdio_entrypoint_with_deadline(
+ "tests/stdio_stall_entrypoint.mojo", "{}", "", "", 3000
+ )
+ except e:
+ message = String(e)
+ assert_true(message.find("stdio-entrypoint") >= 0)
+ assert_true(message.find("signal=") >= 0 or message.find("exited=") >= 0)
+
+
+def test_run_stdio_entrypoint_classifies_loader_failure() raises:
+ var message = ""
+ try:
+ _ = run_stdio_entrypoint_with_deadline(
+ "tests/does_not_exist_entrypoint.mojo", "{}", "", "", 30000
+ )
+ except e:
+ message = String(e)
+ assert_true(message.find("child_failed") >= 0)
+ assert_true(message.find("timeout") < 0)
+
+
def main() raises:
TestSuite.discover_tests[__functions_in_module()]().run()
