commit f3bf6c0dc429e9fca1095fb8206bf90fa32ce0e6
parent dae249d647b1c5ae2ca480f312df0ceb9632c01b
Author: triesap <tyson@radroots.org>
Date: Tue, 22 Sep 2026 21:42:19 +0000
test(hyf): close C002C fixture ownership, result truth and bounded I/O
- Add scope ownership (__enter__/__exit__ plus a body view) to both provider
handles, the stdio runner and every affected caller so assertion, exception,
early-return and partial-startup failures close pipes and reap the child
- Require a strictly parsed bounded report and verified exit/accounting in
reap(), with a real EINTR/ECHILD/other wait taxonomy and cached status
- Enforce line caps inside each read chunk, retain surplus for the next line,
bound concurrent stdio stdout/stderr draining under one total budget and
reject diagnostics overflow with its own cause
- Fix coalesced-body header-cap accounting, raw header-value validation before
legal-OWS trim, malformed script-header rejection and replace the descriptor
census with a non-opening numeric F_GETFD scan; 55/55 provider-helper, 18/18
provider-adapter, 15/15 jev, 14/14 journeys, 5/5 repo-local-process and 51/51
unit tests pass with format/architecture green and the same 24 D16 failures
Diffstat:
11 files changed, 2496 insertions(+), 1127 deletions(-)
diff --git a/tests/jev_provider_helper.mojo b/tests/jev_provider_helper.mojo
@@ -14,13 +14,15 @@ from flare.utils import usleep
from parent_lifecycle import (
FIXTURE_DEFAULT_DEADLINE_MS,
TERMINATION_GRACE_MS,
+ PipedChildState,
+ PipeFds,
ProcessStatus,
child_exit,
close_fd,
dup2_fd,
fork_pid,
make_pipe,
- read_line_bounded,
+ parse_ready_line,
set_alarm,
terminate_owned,
wait_bounded,
@@ -38,6 +40,7 @@ from strict_fixture import (
parse_report,
render_response,
report_line,
+ report_status_matches_exit,
serve_scripts,
validate_convenience,
verify_exchange,
@@ -255,136 +258,226 @@ def serve_jev(port: Int, mode: String, requests: Int) raises -> ServeReport:
struct SpawnedJevStub(Movable):
+ """Single-owner Jev fixture handle; the body receives a view."""
+
var pid: Int
- var _report_fd: Int
- var _deadline_ms: 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, deadline_ms: Int):
+ var state: PipedChildState
+
+ def __init__(out self, pid: Int, var state: PipedChildState):
self.pid = pid
- self._report_fd = report_fd
- self._deadline_ms = deadline_ms
- self._reaped = False
- self._ok = False
- self._phase = "pending"
- self._case = "-"
- self._reason = "not_reaped"
- self._requests = 0
- self._connections = 0
+ self.state = state^
+
+ def __enter__(mut self) -> SpawnedJevStubView:
+ return SpawnedJevStubView(self.pid, UnsafePointer(to=self))
+
+ def __exit__(mut self):
+ self.cleanup()
+
+ def cleanup(mut self):
+ """Fast, non-raising owned cleanup for assertion/error/early return."""
+ if self.state.reaped:
+ return
+ var status = terminate_owned(self.pid, TERMINATION_GRACE_MS)
+ self.state.status = status.copy()
+ self.state.reaped = True
+ self.state.close_reader()
+ if not status.cleanup_proved():
+ self.state.cleanup_error = "unreaped:" + status.describe()
def ok(self) -> Bool:
- return self._ok
+ return self.state.ok
def phase(self) -> String:
- return String(self._phase)
+ return String(self.state.phase)
def failure_case(self) -> String:
- return String(self._case)
+ return String(self.state.case_label)
def reason(self) -> String:
- return String(self._reason)
+ return String(self.state.reason)
def request_count(self) -> Int:
- return self._requests
+ return self.state.requests
def connection_count(self) -> Int:
- return self._connections
+ return self.state.connections
+
+ def cleanup_error(self) -> String:
+ return String(self.state.cleanup_error)
def describe(self) -> String:
return (
"phase="
- + self._phase
+ + self.state.phase
+ " case="
- + self._case
+ + self.state.case_label
+ " reason="
- + self._reason
+ + self.state.reason
+ " requests="
- + String(self._requests)
+ + String(self.state.requests)
+ " connections="
- + String(self._connections)
+ + String(self.state.connections)
)
def status(self) -> ProcessStatus:
+ if self.state.reaped:
+ return self.state.status.copy()
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:
+ """Strictly reap the owned child and decode its bounded report."""
+ if self.state.reaped:
return
- var st = wait_bounded(self.pid, self._deadline_ms)
- var report_text = ""
- if st.state == "running":
+ var status = wait_bounded(self.pid, self.state.deadline_ms)
+ self.state.status = status.copy()
+ if status.state == "running" or status.state == "interrupted":
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)
+ self.state.status = term.copy()
+ self.state.store(False, "watchdog", "-", "timeout", 0, 0)
+ if not term.cleanup_proved():
+ self.state.cleanup_error = "unreaped:" + term.describe()
+ self.state.reason = "timeout_unreaped"
+ self.state.reaped = True
+ self.state.close_reader()
+ return
+ if status.state == "gone" or status.state == "wait_error":
+ self.state.store(False, "watchdog", "-", status.state, 0, 0)
+ if status.state == "wait_error":
+ self.state.cleanup_error = "wait_error:" + status.error
+ self.state.reaped = True
+ self.state.close_reader()
return
- if self._report_fd >= 0:
- try:
- report_text = read_line_bounded(
- self._report_fd, STRICT_MAX_REPORT_BYTES, 1000
+ var report_text = ""
+ var read_error = ""
+ try:
+ report_text = self.state.read_line(STRICT_MAX_REPORT_BYTES, 1000)
+ except e:
+ read_error = String(e)
+ self.state.close_reader()
+ if report_text == "":
+ if status.exited and status.exit_code == 0:
+ self.state.store(False, "startup", "-", "missing_report", 0, 0)
+ elif status.exited:
+ self.state.store(
+ False,
+ "startup",
+ "-",
+ "exit_" + String(status.exit_code),
+ 0,
+ 0,
)
- 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)
+ else:
+ self.state.store(
+ False,
+ "watchdog",
+ "-",
+ "signal_" + String(status.signal),
+ 0,
+ 0,
+ )
+ if read_error != "":
+ self.state.cleanup_error = "report_read:" + read_error
+ self.state.reaped = True
+ return
+ var parsed = parse_report(report_text)
+ if parsed.phase == "parse":
+ self.state.store(False, "parse", "-", parsed.reason, -1, -1)
+ if read_error != "":
+ self.state.cleanup_error = "report_read:" + read_error
+ self.state.reaped = True
+ return
+ self.state.store(
+ parsed.ok,
+ parsed.phase,
+ parsed.case_label,
+ parsed.reason,
+ parsed.requests,
+ parsed.connections,
+ )
+ if not report_status_matches_exit(
+ status.exited, status.exit_code, parsed.ok
+ ):
+ self.state.ok = False
+ self.state.phase = "startup"
+ self.state.reason = "report_status_mismatch_" + status.describe()
+ elif parsed.ok and parsed.requests != self.state.expected_requests:
+ self.state.ok = False
+ self.state.phase = "accounting"
+ self.state.reason = "request_count_mismatch"
+ elif parsed.ok and (
+ parsed.connections < 1
+ or parsed.connections > self.state.expected_requests
+ ):
+ self.state.ok = False
+ self.state.phase = "accounting"
+ self.state.reason = "connection_count_invalid"
+ self.state.reaped = True
def wait(mut self) raises:
self.reap()
- if not self._ok:
+ if not self.state.ok:
raise Error("fixture-failure " + self.describe())
def terminate(mut self) raises:
- if self._reaped:
+ if self.state.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)
+ var status = terminate_owned(self.pid, TERMINATION_GRACE_MS)
+ self.state.status = status.copy()
+ self.state.reaped = True
+ self.state.close_reader()
+ if not status.cleanup_proved():
+ raise Error(
+ "lifecycle: owned child not reaped: " + status.describe()
+ )
+
+
+struct SpawnedJevStubView(Movable):
+ """Body-scope view of an owned Jev fixture handle."""
+
+ var pid: Int
+ var target: UnsafePointer[SpawnedJevStub, MutAnyOrigin]
+
+ def __init__(
+ out self, pid: Int, target: UnsafePointer[SpawnedJevStub, MutAnyOrigin]
+ ):
+ self.pid = pid
+ self.target = target
+
+ def ok(self) -> Bool:
+ return self.target[].ok()
+
+ def phase(self) -> String:
+ return self.target[].phase()
+
+ def failure_case(self) -> String:
+ return self.target[].failure_case()
+
+ def reason(self) -> String:
+ return self.target[].reason()
+
+ def request_count(self) -> Int:
+ return self.target[].request_count()
+
+ def connection_count(self) -> Int:
+ return self.target[].connection_count()
+
+ def cleanup_error(self) -> String:
+ return self.target[].cleanup_error()
+
+ def describe(self) -> String:
+ return self.target[].describe()
+
+ def status(self) -> ProcessStatus:
+ return self.target[].status()
+
+ def reap(mut self):
+ self.target[].reap()
+
+ def wait(mut self) raises:
+ self.target[].wait()
+
+ def terminate(mut self) raises:
+ self.target[].terminate()
@fieldwise_init
@@ -392,6 +485,26 @@ struct SpawnedJevStubAuto(Movable):
var port: Int
var stub: SpawnedJevStub
+ def __enter__(mut self) -> SpawnedJevStubAutoView:
+ return SpawnedJevStubAutoView(
+ self.port,
+ SpawnedJevStubView(self.stub.pid, UnsafePointer(to=self.stub)),
+ )
+
+ def __exit__(mut self):
+ self.stub.cleanup()
+
+
+struct SpawnedJevStubAutoView(Movable):
+ """Body-scope view of an auto-port Jev fixture handle."""
+
+ var port: Int
+ var stub: SpawnedJevStubView
+
+ def __init__(out self, port: Int, var stub: SpawnedJevStubView):
+ self.port = port
+ self.stub = stub^
+
def reserve_jev_port() raises -> Int:
var listener = TcpListener.bind(SocketAddr.localhost(0))
@@ -425,13 +538,6 @@ def serve_jev_scripted(
return serve_scripts(listener, scripts^, "scripted")
-def _read_ready_line(fd: Int, deadline_ms: Int) -> String:
- try:
- return read_line_bounded(fd, 256, deadline_ms)
- except:
- return ""
-
-
def spawn_jev_scripted_auto(
var scripts: List[ExchangeScript],
deadline_ms: Int = FIXTURE_DEFAULT_DEADLINE_MS,
@@ -439,12 +545,78 @@ def spawn_jev_scripted_auto(
return _spawn_jev_scripted(0, scripts^, deadline_ms)
+def _spawn_child_or_cleanup(
+ pipe: PipeFds, pid: Int, mode: String, deadline_ms: Int, requests: Int
+) raises -> SpawnedJevStubAuto:
+ """Build the owned state, read exact readiness, or clean up and raise."""
+ var state = PipedChildState(
+ pid=pid,
+ report_fd=pipe.read_fd,
+ pending="",
+ eof=False,
+ closed=False,
+ deadline_ms=deadline_ms,
+ expected_requests=requests,
+ reaped=False,
+ ok=False,
+ phase="pending",
+ case_label="-",
+ reason="not_reaped",
+ requests=0,
+ connections=0,
+ cleanup_error="",
+ status=ProcessStatus("pending", False, -1, 0, 0, ""),
+ )
+ var ready_line = ""
+ try:
+ ready_line = state.read_line(STRICT_MAX_REPORT_BYTES, deadline_ms)
+ except e:
+ var st = terminate_owned(pid, TERMINATION_GRACE_MS)
+ state.status = st.copy()
+ state.reaped = True
+ state.close_reader()
+ raise Error(
+ "jev stub readiness failed ("
+ + String(e)
+ + " / "
+ + st.describe()
+ + ")"
+ )
+ var reported_port = 0
+ try:
+ reported_port = parse_ready_line(ready_line, 256)
+ except e:
+ var st = terminate_owned(pid, TERMINATION_GRACE_MS)
+ state.status = st.copy()
+ state.reaped = True
+ state.close_reader()
+ raise Error(
+ "jev stub malformed readiness ("
+ + String(e)
+ + " / report="
+ + ready_line
+ + " / "
+ + st.describe()
+ + ")"
+ )
+ _ = mode
+ return SpawnedJevStubAuto(
+ port=reported_port, stub=SpawnedJevStub(pid, state^)
+ )
+
+
def _spawn_jev_scripted(
port: Int, var scripts: List[ExchangeScript], deadline_ms: Int
) raises -> SpawnedJevStubAuto:
var total = len(scripts)
var pipe = make_pipe()
- var pid = fork_pid()
+ var pid = 0
+ try:
+ pid = fork_pid()
+ except e:
+ close_fd(pipe.read_fd)
+ close_fd(pipe.write_fd)
+ raise Error("jev stub fork failed: " + String(e))
if pid == 0:
if dup2_fd(pipe.write_fd, 1) < 0:
child_exit(126)
@@ -462,32 +634,20 @@ def _spawn_jev_scripted(
write_raw(1, report_line(failed) + "\n")
child_exit(125)
close_fd(pipe.write_fd)
- var ready_line = _read_ready_line(pipe.read_fd, deadline_ms)
- 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, deadline_ms)
- )
+ return _spawn_child_or_cleanup(pipe, pid, "scripted", deadline_ms, total)
def _spawn_jev_stub(
port: Int, mode: String, requests: Int, deadline_ms: Int
) raises -> SpawnedJevStubAuto:
var pipe = make_pipe()
- var pid = fork_pid()
+ var pid = 0
+ try:
+ pid = fork_pid()
+ except e:
+ close_fd(pipe.read_fd)
+ close_fd(pipe.write_fd)
+ raise Error("jev stub fork failed: " + String(e))
if pid == 0:
if dup2_fd(pipe.write_fd, 1) < 0:
child_exit(126)
@@ -505,21 +665,4 @@ def _spawn_jev_stub(
write_raw(1, report_line(failed) + "\n")
child_exit(125)
close_fd(pipe.write_fd)
- var ready_line = _read_ready_line(pipe.read_fd, deadline_ms)
- 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 :]))
- return SpawnedJevStubAuto(
- port=reported_port, stub=SpawnedJevStub(pid, pipe.read_fd, deadline_ms)
- )
+ return _spawn_child_or_cleanup(pipe, pid, mode, deadline_ms, requests)
diff --git a/tests/max_local_process_helper.mojo b/tests/max_local_process_helper.mojo
@@ -8,6 +8,7 @@ failures (FX07).
"""
from std.collections import List
+from std.memory import ArcPointer
from flare.net import SocketAddr
from flare.tcp import TcpListener
@@ -16,13 +17,14 @@ from flare.utils import usleep
from parent_lifecycle import (
FIXTURE_DEFAULT_DEADLINE_MS,
TERMINATION_GRACE_MS,
+ PipedChildState,
ProcessStatus,
child_exit,
close_fd,
dup2_fd,
fork_pid,
make_pipe,
- read_line_bounded,
+ parse_ready_line,
set_alarm,
terminate_owned,
wait_bounded,
@@ -41,6 +43,7 @@ from strict_fixture import (
parse_report,
render_response,
report_line,
+ report_status_matches_exit,
serve_scripts,
validate_convenience,
verify_exchange,
@@ -341,146 +344,245 @@ def serve_max_local(
struct SpawnedMaxLocalStub(Movable):
+ """Single-owner MaxLocal fixture handle.
+
+ The body receives a :class:`SpawnedMaxLocalView` from ``__enter__`` while
+ ``__exit__`` on this manager performs owned cleanup, so parent assertion,
+ exception and early-return paths all reap and close the child without any
+ shared heap state. Use ``with spawn_max_local_stub(...) as stub:``.
+ """
+
var pid: Int
var port: Int
- var _report_fd: Int
- var _deadline_ms: Int
- var _reaped: Bool
- var _ok: Bool
- var _phase: String
- var _case: String
- var _reason: String
- var _requests: Int
- var _connections: Int
+ var state: PipedChildState
- def __init__(
- out self, pid: Int, port: Int, report_fd: Int, deadline_ms: Int
- ):
+ def __init__(out self, pid: Int, port: Int, var state: PipedChildState):
self.pid = pid
self.port = port
- self._report_fd = report_fd
- self._deadline_ms = deadline_ms
- self._reaped = False
- self._ok = False
- self._phase = "pending"
- self._case = "-"
- self._reason = "not_reaped"
- self._requests = 0
- self._connections = 0
+ self.state = state^
+
+ def __enter__(mut self) -> SpawnedMaxLocalView:
+ return SpawnedMaxLocalView(self.pid, self.port, UnsafePointer(to=self))
+
+ def __exit__(mut self):
+ self.cleanup()
+
+ def cleanup(mut self):
+ """Fast, non-raising owned cleanup for assertion/error/early return."""
+ if self.state.reaped:
+ return
+ var status = terminate_owned(self.pid, TERMINATION_GRACE_MS)
+ self.state.status = status.copy()
+ self.state.reaped = True
+ self.state.close_reader()
+ if not status.cleanup_proved():
+ self.state.cleanup_error = "unreaped:" + status.describe()
def ok(self) -> Bool:
- return self._ok
+ return self.state.ok
def phase(self) -> String:
- return String(self._phase)
+ return String(self.state.phase)
def failure_case(self) -> String:
- return String(self._case)
+ return String(self.state.case_label)
def reason(self) -> String:
- return String(self._reason)
+ return String(self.state.reason)
def request_count(self) -> Int:
- return self._requests
+ return self.state.requests
def connection_count(self) -> Int:
- return self._connections
+ return self.state.connections
+
+ def cleanup_error(self) -> String:
+ return String(self.state.cleanup_error)
def describe(self) -> String:
return (
"phase="
- + self._phase
+ + self.state.phase
+ " case="
- + self._case
+ + self.state.case_label
+ " reason="
- + self._reason
+ + self.state.reason
+ " requests="
- + String(self._requests)
+ + String(self.state.requests)
+ " connections="
- + String(self._connections)
+ + String(self.state.connections)
)
def status(self) -> ProcessStatus:
+ if self.state.reaped:
+ return self.state.status.copy()
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:
+ """Reap the owned child and strictly decode its bounded report.
+
+ Never raises (so ``__exit__`` cannot mask a body error). Success
+ requires a complete strictly parsed report, matching child exit status
+ and exact request accounting; missing/truncated/mismatched reports and
+ nonzero exit or signals fail explicitly.
+ """
+ if self.state.reaped:
return
- var st = wait_bounded(self.pid, self._deadline_ms)
- var report_text = ""
- if st.state == "running":
+ var status = wait_bounded(self.pid, self.state.deadline_ms)
+ self.state.status = status.copy()
+ if status.state == "running" or status.state == "interrupted":
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)
+ self.state.status = term.copy()
+ self.state.store(False, "watchdog", "-", "timeout", 0, 0)
+ if not term.cleanup_proved():
+ self.state.cleanup_error = "unreaped:" + term.describe()
+ self.state.reason = "timeout_unreaped"
+ self.state.reaped = True
+ self.state.close_reader()
+ return
+ if status.state == "gone" or status.state == "wait_error":
+ self.state.store(False, "watchdog", "-", status.state, 0, 0)
+ if status.state == "wait_error":
+ self.state.cleanup_error = "wait_error:" + status.error
+ self.state.reaped = True
+ self.state.close_reader()
return
- if self._report_fd >= 0:
- try:
- report_text = read_line_bounded(
- self._report_fd, STRICT_MAX_REPORT_BYTES, 1000
+ var report_text = ""
+ var read_error = ""
+ try:
+ report_text = self.state.read_line(STRICT_MAX_REPORT_BYTES, 1000)
+ except e:
+ read_error = String(e)
+ self.state.close_reader()
+ if report_text == "":
+ if status.exited and status.exit_code == 0:
+ self.state.store(False, "startup", "-", "missing_report", 0, 0)
+ elif status.exited:
+ self.state.store(
+ False,
+ "startup",
+ "-",
+ "exit_" + String(status.exit_code),
+ 0,
+ 0,
)
- 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)
+ else:
+ self.state.store(
+ False,
+ "watchdog",
+ "-",
+ "signal_" + String(status.signal),
+ 0,
+ 0,
+ )
+ if read_error != "":
+ self.state.cleanup_error = "report_read:" + read_error
+ self.state.reaped = True
+ return
+ var parsed = parse_report(report_text)
+ if parsed.phase == "parse":
+ self.state.store(False, "parse", "-", parsed.reason, -1, -1)
+ if read_error != "":
+ self.state.cleanup_error = "report_read:" + read_error
+ self.state.reaped = True
+ return
+ self.state.store(
+ parsed.ok,
+ parsed.phase,
+ parsed.case_label,
+ parsed.reason,
+ parsed.requests,
+ parsed.connections,
+ )
+ if not report_status_matches_exit(
+ status.exited, status.exit_code, parsed.ok
+ ):
+ self.state.ok = False
+ self.state.phase = "startup"
+ self.state.reason = "report_status_mismatch_" + status.describe()
+ elif parsed.ok and parsed.requests != self.state.expected_requests:
+ self.state.ok = False
+ self.state.phase = "accounting"
+ self.state.reason = "request_count_mismatch"
+ elif parsed.ok and (
+ parsed.connections < 1
+ or parsed.connections > self.state.expected_requests
+ ):
+ self.state.ok = False
+ self.state.phase = "accounting"
+ self.state.reason = "connection_count_invalid"
+ self.state.reaped = True
def wait(mut self) raises:
self.reap()
- if not self._ok:
+ if not self.state.ok:
raise Error("fixture-failure " + self.describe())
def terminate(mut self) raises:
- if self._reaped:
+ if self.state.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)
+ var status = terminate_owned(self.pid, TERMINATION_GRACE_MS)
+ self.state.status = status.copy()
+ self.state.reaped = True
+ self.state.close_reader()
+ if not status.cleanup_proved():
+ raise Error(
+ "lifecycle: owned child not reaped: " + status.describe()
+ )
+
+
+struct SpawnedMaxLocalView(Movable):
+ """Body-scope view of an owned MaxLocal fixture handle."""
+
+ var pid: Int
+ var port: Int
+ var target: UnsafePointer[SpawnedMaxLocalStub, MutAnyOrigin]
+
+ def __init__(
+ out self,
+ pid: Int,
+ port: Int,
+ target: UnsafePointer[SpawnedMaxLocalStub, MutAnyOrigin],
+ ):
+ self.pid = pid
+ self.port = port
+ self.target = target
+
+ def ok(self) -> Bool:
+ return self.target[].ok()
+
+ def phase(self) -> String:
+ return self.target[].phase()
+
+ def failure_case(self) -> String:
+ return self.target[].failure_case()
+
+ def reason(self) -> String:
+ return self.target[].reason()
+
+ def request_count(self) -> Int:
+ return self.target[].request_count()
+
+ def connection_count(self) -> Int:
+ return self.target[].connection_count()
+
+ def cleanup_error(self) -> String:
+ return self.target[].cleanup_error()
+
+ def describe(self) -> String:
+ return self.target[].describe()
+
+ def status(self) -> ProcessStatus:
+ return self.target[].status()
+
+ def reap(mut self):
+ self.target[].reap()
+
+ def wait(mut self) raises:
+ self.target[].wait()
+
+ def terminate(mut self) raises:
+ self.target[].terminate()
def reserve_loopback_port() raises -> Int:
@@ -502,13 +604,6 @@ def _serve_max_local_for(
return serve_max_local(port, mode, requests)
-def _read_ready_line(fd: Int, deadline_ms: Int) -> String:
- try:
- return read_line_bounded(fd, 256, deadline_ms)
- except:
- return ""
-
-
def serve_max_local_scripted(
port: Int, var scripts: List[ExchangeScript]
) raises -> ServeReport:
@@ -547,7 +642,13 @@ def _spawn_max_local(
deadline_ms: Int,
) raises -> SpawnedMaxLocalStub:
var pipe = make_pipe()
- var pid = fork_pid()
+ var pid = 0
+ try:
+ pid = fork_pid()
+ except e:
+ close_fd(pipe.read_fd)
+ close_fd(pipe.write_fd)
+ raise Error("max_local stub fork failed: " + String(e))
if pid == 0:
if dup2_fd(pipe.write_fd, 1) < 0:
child_exit(126)
@@ -567,19 +668,54 @@ def _spawn_max_local(
write_raw(1, report_line(failed) + "\n")
child_exit(125)
close_fd(pipe.write_fd)
- var ready_line = _read_ready_line(pipe.read_fd, deadline_ms)
- if not ready_line.startswith("ready"):
+ var state = PipedChildState(
+ pid=pid,
+ report_fd=pipe.read_fd,
+ pending="",
+ eof=False,
+ closed=False,
+ deadline_ms=deadline_ms,
+ expected_requests=requests,
+ reaped=False,
+ ok=False,
+ phase="pending",
+ case_label="-",
+ reason="not_reaped",
+ requests=0,
+ connections=0,
+ cleanup_error="",
+ status=ProcessStatus("pending", False, -1, 0, 0, ""),
+ )
+ var ready_line = ""
+ try:
+ ready_line = state.read_line(STRICT_MAX_REPORT_BYTES, deadline_ms)
+ except e:
var st = terminate_owned(pid, TERMINATION_GRACE_MS)
- close_fd(pipe.read_fd)
+ state.status = st.copy()
+ state.reaped = True
+ state.close_reader()
+ raise Error(
+ "max_local stub readiness failed ("
+ + String(e)
+ + " / "
+ + st.describe()
+ + ")"
+ )
+ var reported_port = 0
+ try:
+ reported_port = parse_ready_line(ready_line, 256)
+ except e:
+ var st = terminate_owned(pid, TERMINATION_GRACE_MS)
+ state.status = st.copy()
+ state.reaped = True
+ state.close_reader()
raise Error(
- "max_local stub failed to report ready ("
+ "max_local stub malformed readiness ("
+ + String(e)
+ + " / report="
+ 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 :]))
- return SpawnedMaxLocalStub(pid, reported_port, pipe.read_fd, deadline_ms)
+ return SpawnedMaxLocalStub(pid, reported_port, state^)
diff --git a/tests/parent_lifecycle.mojo b/tests/parent_lifecycle.mojo
@@ -1,8 +1,10 @@
"""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
+ADR-0012 D29, ADR-0014 D33 FX06/FX08 and ADR-0015 D35 LC01-LC06 require
+*parent-enforced*, finite startup/read/write/wait deadlines, exception-safe
+cleanup/reaping for provider fixture children and ``run_stdio_entrypoint``, a
+truthful wait-error taxonomy, byte caps enforced inside each read chunk and a
+non-opening numeric descriptor census. 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
@@ -14,10 +16,27 @@ 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.
+* ``PipedChildState`` is the single shared mutable lifecycle record for one
+ owned fixture child; every copy of a provider handle shares it, so cleanup is
+ idempotent across the ``with`` manager, the body handle and repeated calls.
* No broad ``pkill``/name matching is performed anywhere.
+
+This toolchain exposes ``std.ffi.get_errno``/``ErrNo``, so the wait-error
+taxonomy classifies a negative ``waitpid`` by real ``EINTR`` (retain and
+retry ownership), ``ECHILD`` (no waitable owned child remains) and every other
+errno (``wait_error``, never completed cleanup).
"""
-from std.ffi import c_int, c_uint, c_ssize_t, c_size_t, external_call
+from std.collections import List
+from std.ffi import (
+ ErrNo,
+ c_int,
+ c_uint,
+ c_ssize_t,
+ c_size_t,
+ external_call,
+ get_errno,
+)
from std.sys._libc import close
from std.time import perf_counter_ns
@@ -29,8 +48,12 @@ comptime POLLERR: Int = 8
comptime POLLHUP: Int = 16
comptime POLLNVAL: Int = 32
comptime SIGALRM: Int = 14
+comptime SIGPIPE: Int = 13
comptime SIGKILL: Int = 9
comptime SIGTERM: Int = 15
+comptime SIG_IGN: Int = 1
+comptime F_GETFD: Int = 1
+comptime CENSUS_MAX_FDS: Int = 65536
comptime FIXTURE_DEFAULT_DEADLINE_MS: Int = 20000
comptime TERMINATION_GRACE_MS: Int = 2000
@@ -50,7 +73,17 @@ struct PipeFds(Movable):
var write_fd: Int
+def ignore_sigpipe():
+ """Ignore SIGPIPE so a peer-close race surfaces as EPIPE, not parent death.
+
+ Inherited across ``fork``, so owned fixture children get the same bounded
+ write-failure behaviour instead of dying on an interrupted report write.
+ """
+ _ = external_call["signal", Int](c_int(SIGPIPE), c_int(SIG_IGN))
+
+
def make_pipe() raises -> PipeFds:
+ ignore_sigpipe()
var fds = InlineArray[c_int, 2](fill=0)
if Int(external_call["pipe", c_int](fds.unsafe_ptr())) != 0:
raise Error("lifecycle: pipe failed")
@@ -108,12 +141,66 @@ def poll_fd(fd: Int, events: Int, timeout_ms: Int) -> Int:
return (Int(cell[1]) >> 16) & 0xFFFF
+@fieldwise_init
+struct PollThree(Movable):
+ """Bounded three-descriptor ``poll(2)`` result; ``count`` is -1 on error."""
+
+ var count: Int
+ var r0: Int
+ var r1: Int
+ var r2: Int
+
+
+def poll_three(
+ fd0: Int,
+ events0: Int,
+ fd1: Int,
+ events1: Int,
+ fd2: Int,
+ events2: Int,
+ timeout_ms: Int,
+) -> PollThree:
+ """Poll exactly three descriptors with one finite timeout."""
+ var cell = InlineArray[Int32, 12](fill=0)
+ cell[0] = Int32(fd0)
+ cell[1] = Int32(events0)
+ cell[2] = Int32(fd1)
+ cell[3] = Int32(events1)
+ cell[4] = Int32(fd2)
+ cell[5] = Int32(events2)
+ var n = Int(
+ external_call["poll", c_int](
+ cell.unsafe_ptr(), c_uint(3), c_int(timeout_ms)
+ )
+ )
+ if n < 0:
+ return PollThree(-1, 0, 0, 0)
+ if n == 0:
+ return PollThree(0, 0, 0, 0)
+ return PollThree(
+ n,
+ (Int(cell[1]) >> 16) & 0xFFFF,
+ (Int(cell[3]) >> 16) & 0xFFFF,
+ (Int(cell[5]) >> 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)))
+ while True:
+ var n = Int(
+ external_call["read", c_ssize_t](fd, buf, c_size_t(max_bytes))
+ )
+ if n >= 0 or get_errno() != ErrNo.EINTR:
+ return n
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)))
+ while True:
+ var written = Int(
+ external_call["write", c_ssize_t](fd, ptr, c_size_t(n))
+ )
+ if written >= 0 or get_errno() != ErrNo.EINTR:
+ return written
def write_raw(fd: Int, text: String) -> Int:
@@ -122,11 +209,16 @@ def write_raw(fd: Int, text: String) -> Int:
return n
+comptime WRITE_CHUNK_BYTES: Int = 512
+
+
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``.
+ Chunks are capped at ``PIPE_BUF``-safe size so a ``POLLOUT`` readiness
+ never lets a blocking write stall past the deadline. Returns ``""`` on
+ success or a bounded reason such as ``write_deadline_expired`` /
+ ``write_pipe_closed``.
"""
var total = data.byte_length()
var sent = 0
@@ -139,7 +231,7 @@ def write_fd_bounded(fd: Int, data: String, deadline_ms: Int) -> String:
continue
if (ev & (POLLERR | POLLHUP | POLLNVAL)) != 0:
return "write_pipe_closed"
- var chunk = min(4096, total - sent)
+ var chunk = min(WRITE_CHUNK_BYTES, total - sent)
var slice = data[byte = sent : sent + chunk]
var n = _write_fd(fd, slice.as_bytes().unsafe_ptr(), chunk)
if n <= 0:
@@ -148,33 +240,130 @@ def write_fd_bounded(fd: Int, data: String, deadline_ms: Int) -> String:
return ""
+@fieldwise_init
+struct ChunkWrite(Movable):
+ var reason: String
+ var written: Int
+
+
+def write_fd_chunk(fd: Int, data: String, offset: Int) -> ChunkWrite:
+ """Write one ``PIPE_BUF``-safe chunk after a readiness poll.
+
+ Returns the bounded failure reason and the bytes actually written so a
+ caller can interleave writing with draining other descriptors.
+ """
+ var total = data.byte_length()
+ if offset >= total:
+ return ChunkWrite("", 0)
+ var chunk = min(WRITE_CHUNK_BYTES, total - offset)
+ var slice = data[byte = offset : offset + chunk]
+ var n = _write_fd(fd, slice.as_bytes().unsafe_ptr(), chunk)
+ if n <= 0:
+ return ChunkWrite("write_failed", 0)
+ return ChunkWrite("", n)
+
+
+# ── Bounded line reader with surplus retention ──────────────────────────────
+
+
+@fieldwise_init
+struct BoundedLineReader(Movable):
+ """Byte-bounded, deadline-bounded line reader that never discards surplus.
+
+ The byte cap is enforced *inside* every read chunk, so a coalesced chunk of
+ ``ready`` + ``report`` lines cannot smuggle an oversized line past the cap
+ and bytes after a returned newline stay available to the next reader call.
+ """
+
+ var fd: Int
+ var max_bytes: Int
+ var _pending: List[UInt8]
+ var _pos: Int
+ var _eof: Bool
+ var _closed: Bool
+
+ def __init__(out self, fd: Int, max_bytes: Int):
+ self.fd = fd
+ self.max_bytes = max_bytes
+ self._pending = List[UInt8]()
+ self._pos = 0
+ self._eof = False
+ self._closed = False
+
+ def close(mut self):
+ if not self._closed:
+ close_fd(self.fd)
+ self._closed = True
+
+ def has_pending(self) -> Bool:
+ return self._pos < len(self._pending)
+
+ def _compact(mut self):
+ if self._pos == 0:
+ return
+ if self._pos >= len(self._pending):
+ self._pending = List[UInt8]()
+ self._pos = 0
+ return
+ var rest = List[UInt8]()
+ for index in range(self._pos, len(self._pending)):
+ rest.append(self._pending[index])
+ self._pending = rest^
+ self._pos = 0
+
+ def _take_available(mut self, mut out: List[UInt8], stop: Int):
+ for index in range(self._pos, stop):
+ out.append(self._pending[index])
+
+ def read_line(mut self, deadline_ms: Int) raises -> String:
+ """Read one newline-terminated line with bounded size and deadline.
+
+ Raises ``ready_output_overflow`` once the line exceeds ``max_bytes``
+ and ``read_deadline_expired`` when the deadline elapses first. EOF
+ before a newline returns the bytes read so far (possibly empty).
+ """
+ var out = List[UInt8]()
+ var start = now_ms()
+ while True:
+ var found = -1
+ for index in range(self._pos, len(self._pending)):
+ if Int(self._pending[index]) == 10:
+ found = index
+ break
+ if found >= 0:
+ self._take_available(out, found)
+ self._pos = found + 1
+ self._compact()
+ if len(out) > self.max_bytes:
+ raise Error("ready_output_overflow")
+ return bytes_to_string(out)
+ self._take_available(out, len(self._pending))
+ self._pos = len(self._pending)
+ self._compact()
+ if len(out) > self.max_bytes:
+ raise Error("ready_output_overflow")
+ if self._eof:
+ return bytes_to_string(out)
+ if now_ms() - start >= deadline_ms:
+ raise Error("read_deadline_expired")
+ var ev = poll_fd(self.fd, POLLIN, LIFECYCLE_POLL_SLICE_MS)
+ if ev == 0:
+ continue
+ var buf = InlineArray[Byte, 512](fill=0)
+ var n = read_fd(self.fd, buf.unsafe_ptr(), 512)
+ if n <= 0:
+ self._eof = True
+ continue
+ for index in range(n):
+ self._pending.append(UInt8(Int(buf[index])))
+
+
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)
+ """One-shot bounded line read for callers without surplus to preserve."""
+ var reader = BoundedLineReader(fd, max_bytes)
+ return reader.read_line(deadline_ms)
def drain_fd_bounded(
@@ -193,9 +382,11 @@ def drain_fd_bounded(
var n = read_fd(fd, buf.unsafe_ptr(), 1024)
if n <= 0:
break
- for index in range(n):
+ var room = max_bytes - len(out)
+ var take = min(Int(n), room)
+ for index in range(take):
out.append(UInt8(Int(buf[index])))
- return _bytes_to_string(out)
+ return bytes_to_string(out)
def read_all_bounded(
@@ -222,10 +413,10 @@ def read_all_bounded(
raise Error("stdout_overflow")
for index in range(n):
out.append(UInt8(Int(buf[index])))
- return _bytes_to_string(out)
+ return bytes_to_string(out)
-def _bytes_to_string(bytes: List[UInt8]) raises -> String:
+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)))
@@ -235,18 +426,33 @@ def _bytes_to_string(bytes: List[UInt8]) raises -> String:
@fieldwise_init
-struct ProcessStatus(Movable):
+struct ProcessStatus(Copyable, Movable):
var state: String
var exited: Bool
var exit_code: Int
var signal: Int
var raw: Int
+ var error: String
+
+ def __copyinit__(out self, existing: Self):
+ self.state = existing.state
+ self.exited = existing.exited
+ self.exit_code = existing.exit_code
+ self.signal = existing.signal
+ self.raw = existing.raw
+ self.error = existing.error
def reaped(self) -> Bool:
return self.state == "reaped"
+ def cleanup_proved(self) -> Bool:
+ """True only when no waitable owned child can remain for this pid."""
+ return self.state == "reaped" or self.state == "gone"
+
def describe(self) -> String:
if self.state != "reaped":
+ if self.error != "":
+ return self.state + ":" + self.error
return self.state
if self.exited:
return "exited=" + String(self.exit_code)
@@ -256,13 +462,37 @@ struct ProcessStatus(Movable):
def _decode_status(raw: Int) -> ProcessStatus:
var low = raw & 0x7F
if low == 0:
- return ProcessStatus("reaped", True, (raw >> 8) & 0xFF, 0, raw)
+ 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)
+ return ProcessStatus("stopped", False, -1, 0, raw, "")
+ return ProcessStatus("reaped", False, -1, low, raw, "")
+
+
+def classify_wait_errno(errno_value: Int) -> String:
+ """Map a real ``waitpid`` errno to the ownership taxonomy.
+
+ ``interrupted`` (EINTR) retains ownership and is retried; ``gone``
+ (ECHILD) proves no waitable child remains; anything else is
+ ``wait_error`` and must never be reported as completed cleanup.
+ """
+ if errno_value == Int(ErrNo.EINTR.value):
+ return "interrupted"
+ if errno_value == Int(ErrNo.ECHILD.value):
+ return "gone"
+ return "wait_error"
def wait_nohang(pid: Int) -> ProcessStatus:
+ """Non-blocking wait with an exact errno ownership taxonomy.
+
+ ``reaped``/``running`` are exact. A negative ``waitpid`` is classified by
+ the real errno: ``EINTR`` retains ownership and is retried by callers;
+ ``ECHILD`` proves no waitable child remains; any other error is a
+ ``wait_error`` that must not be reported as completed cleanup. A
+ non-positive pid is never waited on.
+ """
+ if pid <= 0:
+ return ProcessStatus("wait_error", False, -1, 0, -1, "invalid_pid")
var status = InlineArray[c_int, 1](fill=0)
var r = Int(
external_call["waitpid", c_int](
@@ -271,16 +501,22 @@ def wait_nohang(pid: Int) -> ProcessStatus:
)
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)
+ if r == 0:
+ return ProcessStatus("running", False, -1, 0, 0, "")
+ var errno_value = Int(get_errno().value)
+ var klass = classify_wait_errno(errno_value)
+ if klass == "gone":
+ return ProcessStatus("gone", False, -1, -1, -1, "")
+ return ProcessStatus(
+ klass, False, -1, 0, -1, "errno_" + String(errno_value)
+ )
def wait_bounded(pid: Int, deadline_ms: Int) -> ProcessStatus:
var start = now_ms()
while True:
var st = wait_nohang(pid)
- if st.state != "running":
+ if st.state != "running" and st.state != "interrupted":
return st^
if now_ms() - start >= deadline_ms:
return st^
@@ -291,14 +527,20 @@ 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.
+ reused PID from an unrelated process can never be targeted. An
+ ``interrupted``/``wait_error`` status still owns a live child, so it is
+ signaled and reaped; the returned status is only ``reaped`` when the child
+ was actually collected.
"""
var st = wait_nohang(pid)
- if st.state != "running":
+ if st.cleanup_proved():
+ return st^
+ if st.state == "wait_error":
+ # Identity/ownership is unproved; never signal and never claim cleanup.
return st^
_ = kill_pid(pid, SIGTERM)
st = wait_bounded(pid, grace_ms)
- if st.state == "running":
+ if st.state == "running" or st.state == "interrupted":
_ = kill_pid(pid, SIGKILL)
st = wait_bounded(pid, grace_ms)
return st^
@@ -314,38 +556,180 @@ def pid_not_waitable(pid: Int) -> Bool:
return wait_nohang(pid).state == "gone"
-def open_fd_count() -> Int:
- """Count this process's open descriptors by probing ``/dev/fd/N``.
+# ── Shared owned-child lifecycle state ──────────────────────────────────────
+
- A bounded, read-only descriptor census used to evidence that repeated
- teardown leaks no descriptors. Returns -1 if the platform probe is
- unavailable (never treated as a pass).
+@fieldwise_init
+struct PipedChildState(Movable):
+ """Single mutable lifecycle record shared by every copy of one handle.
+
+ Holds no ``List``: in-place container mutation through a shared reference
+ is avoided so the record stays safe to share across handle copies. The
+ retained line surplus is an immutable ``String`` reassigned in place.
"""
- var probe = String("/dev/fd")
- var dir = Int(
- external_call["open", c_int](
- probe.as_c_string_slice().unsafe_ptr(), c_int(0)
- )
- )
- if dir < 0:
- probe = "/proc/self/fd"
- dir = Int(
- external_call["open", c_int](
- probe.as_c_string_slice().unsafe_ptr(), c_int(0)
- )
- )
- if dir < 0:
- return -1
- close_fd(dir)
+
+ var pid: Int
+ var report_fd: Int
+ var pending: String
+ var eof: Bool
+ var closed: Bool
+ var deadline_ms: Int
+ var expected_requests: Int
+ var reaped: Bool
+ var ok: Bool
+ var phase: String
+ var case_label: String
+ var reason: String
+ var requests: Int
+ var connections: Int
+ var cleanup_error: String
+ var status: ProcessStatus
+
+ 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_label = String(case_label)
+ self.reason = String(reason)
+ self.requests = requests
+ self.connections = connections
+
+ def close_reader(mut self):
+ if not self.closed:
+ close_fd(self.report_fd)
+ self.closed = True
+
+ def read_line(mut self, max_bytes: Int, deadline_ms: Int) raises -> String:
+ """Bounded line read that retains surplus after each newline.
+
+ The byte cap is enforced inside every read chunk (including a newline
+ in the same chunk), and bytes after the returned newline are retained
+ for the next consumer. Raises ``ready_output_overflow`` past the cap
+ and ``read_deadline_expired`` when the deadline elapses first.
+ """
+ var out = List[UInt8]()
+ var start = now_ms()
+ while True:
+ var nl = self.pending.find("\n")
+ if nl >= 0:
+ var line = String(self.pending[byte=0:nl])
+ var rest = String(self.pending[byte = nl + 1 :])
+ if rest.byte_length() > max_bytes:
+ raise Error("ready_output_overflow")
+ self.pending = rest^
+ if line.byte_length() > max_bytes:
+ raise Error("ready_output_overflow")
+ return line^
+ for byte in self.pending.as_bytes():
+ out.append(UInt8(Int(byte)))
+ self.pending = ""
+ if len(out) > max_bytes:
+ raise Error("ready_output_overflow")
+ if self.eof:
+ return bytes_to_string(out)
+ if now_ms() - start >= deadline_ms:
+ raise Error("read_deadline_expired")
+ var ev = poll_fd(self.report_fd, POLLIN, LIFECYCLE_POLL_SLICE_MS)
+ if ev == 0:
+ continue
+ var buf = InlineArray[Byte, 512](fill=0)
+ var n = read_fd(self.report_fd, buf.unsafe_ptr(), 512)
+ if n <= 0:
+ self.eof = True
+ continue
+ var newline_at = -1
+ for index in range(n):
+ if Int(buf[index]) == 10:
+ newline_at = index
+ break
+ if newline_at < 0:
+ for index in range(n):
+ out.append(UInt8(Int(buf[index])))
+ if len(out) > max_bytes:
+ raise Error("ready_output_overflow")
+ continue
+ for index in range(newline_at):
+ out.append(UInt8(Int(buf[index])))
+ if len(out) > max_bytes:
+ raise Error("ready_output_overflow")
+ var rest = List[UInt8]()
+ for index in range(newline_at + 1, n):
+ rest.append(UInt8(Int(buf[index])))
+ var surplus = bytes_to_string(rest)
+ if surplus.byte_length() > max_bytes:
+ raise Error("ready_output_overflow")
+ self.pending = surplus^
+ return bytes_to_string(out)
+
+
+def parse_ready_line(line: String, max_bytes: Int) raises -> Int:
+ """Parse the exact ``ready <port>`` grammar with a valid TCP port range."""
+ if line.byte_length() == 0:
+ raise Error("ready_empty")
+ if line.byte_length() > max_bytes:
+ raise Error("ready_too_large")
+ if not line.startswith("ready "):
+ raise Error("ready_grammar")
+ var digits = String(line[byte=6:])
+ if digits.byte_length() == 0:
+ raise Error("ready_missing_port")
+ if digits.byte_length() > 5:
+ raise Error("ready_port_range")
+ for byte in digits.as_bytes():
+ var b = Int(byte)
+ if b < 48 or b > 57:
+ raise Error("ready_non_digit")
+ var port = Int(digits)
+ if port < 1 or port > 65535:
+ raise Error("ready_port_range")
+ return port
+
+
+# ── Non-opening descriptor census ───────────────────────────────────────────
+
+
+def descriptor_census(limit: Int) -> Int:
+ """Count open descriptors numerically via ``fcntl(F_GETFD)``.
+
+ Never opens a target path, so device nodes cannot block it and sockets and
+ high descriptors are counted the same as regular files. Returns -1 when the
+ census cannot be established (invalid bound or no standard descriptors),
+ which callers must treat as a failed unavailable census, never a pass.
+ """
+ if limit <= 0:
+ return -1
var count = 0
- for n in range(3, 1024):
- var path = probe + "/" + String(n)
- var fd = Int(
- external_call["open", c_int](
- path.as_c_string_slice().unsafe_ptr(), c_int(0)
- )
- )
- if fd >= 0:
+ for fd in range(0, limit):
+ if Int(external_call["fcntl", c_int](c_int(fd), c_int(F_GETFD))) >= 0:
count += 1
- close_fd(fd)
+ if count == 0:
+ return -1
+ return count
+
+
+def fd_scan_limit() -> Int:
+ var n = Int(external_call["getdtablesize", c_int]())
+ if n <= 0:
+ return -1
+ if n > CENSUS_MAX_FDS:
+ n = CENSUS_MAX_FDS
+ return n
+
+
+def open_fd_count() -> Int:
+ """Numeric open-descriptor census for this process (-1 if unavailable)."""
+ return descriptor_census(fd_scan_limit())
+
+
+def open_fd_count_checked() raises -> Int:
+ var count = open_fd_count()
+ if count < 0:
+ raise Error("descriptor_census_unavailable")
return count
diff --git a/tests/stdio_process_helper.mojo b/tests/stdio_process_helper.mojo
@@ -1,29 +1,40 @@
"""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.
+ADR-0014 D33 FX06/FX08 and ADR-0015 D35 LC04: the *parent* enforces one finite
+startup/write/read/wait budget, bounds stdout/stderr/ready output, drains
+stdout and stderr concurrently with writing the request so a chatty child
+cannot deadlock, ignores SIGPIPE so a peer-close race cannot kill the parent,
+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.collections import List
from std.ffi import CStringSlice, c_int, external_call
from parent_lifecycle import (
+ LIFECYCLE_POLL_SLICE_MS,
+ POLLERR,
+ POLLHUP,
+ POLLIN,
+ POLLNVAL,
+ POLLOUT,
TERMINATION_GRACE_MS,
- FIXTURE_DEFAULT_DEADLINE_MS,
+ bytes_to_string,
child_exit,
close_fd,
dup2_fd,
fork_pid,
make_pipe,
- read_all_bounded,
+ now_ms,
+ poll_three,
+ read_fd,
set_alarm,
terminate_owned,
wait_bounded,
- write_fd_bounded,
+ write_fd_chunk,
)
from safe_tempdir import SafeTempDir
@@ -34,7 +45,9 @@ 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.
+# build/run lane budget, not a fixture lifetime and not a product SLO. One
+# budget covers compile, write, drain and wait; only the bounded cleanup grace
+# is added for reaping.
comptime STDIO_ENTRYPOINT_DEADLINE_MS = 120000
comptime STDIO_CHILD_ALARM_SECONDS = 180
comptime STDIO_MAX_STDOUT_BYTES = 2097152
@@ -63,6 +76,29 @@ struct ScopedEnvVar:
_ = std.os.unsetenv(self.name)
+@fieldwise_init
+struct DrainOutcome(Movable):
+ var eof: Bool
+ var reason: String
+
+
+def drain_ready(
+ fd: Int, mut out: List[UInt8], cap: Int, revents: Int
+) -> DrainOutcome:
+ """Read one ready descriptor into a capped buffer; preserve the cause."""
+ if (revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL)) == 0:
+ return DrainOutcome(False, "")
+ var buf = InlineArray[Byte, 4096](fill=0)
+ var n = read_fd(fd, buf.unsafe_ptr(), 4096)
+ if n <= 0:
+ return DrainOutcome(True, "")
+ if len(out) + n > cap:
+ return DrainOutcome(True, "stream_overflow")
+ for index in range(n):
+ out.append(UInt8(Int(buf[index])))
+ return DrainOutcome(False, "")
+
+
def run_stdio_entrypoint(
entrypoint: String, request_json: String
) raises -> Value:
@@ -78,13 +114,25 @@ def run_stdio_entrypoint(
def _terminate_and_raise(
- pid: Int, stdin_fd: Int, stdout_fd: Int, stderr_fd: Int, reason: String
+ 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() + ")")
+ raise Error(
+ "stdio-entrypoint "
+ + reason
+ + " (child "
+ + st.describe()
+ + " cleanup_error="
+ + String("" if st.cleanup_proved() else "unreaped")
+ + ")"
+ )
def run_stdio_entrypoint_with_2_args(
@@ -142,7 +190,17 @@ def run_stdio_entrypoint_with_deadline(
var command_ptr = command.as_c_string_slice().unsafe_ptr()
var argv_ptr = argv.unsafe_ptr()
- var pid = fork_pid()
+ var pid = 0
+ try:
+ pid = fork_pid()
+ except e:
+ 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)
+ raise Error("stdio-entrypoint fork failed: " + String(e))
if pid == 0:
if dup2_fd(stdin_read_fd, 0) < 0:
child_exit(126)
@@ -164,37 +222,100 @@ def run_stdio_entrypoint_with_deadline(
close_fd(stdout_write_fd)
close_fd(stderr_write_fd)
- 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 request = request_json + "\n"
+ var sent = 0
+ var stdout = List[UInt8]()
+ var stderr_bytes = List[UInt8]()
+ var stdout_eof = False
+ var stderr_eof = False
+ var stdin_done = False
+ var write_reason = ""
+ var read_reason = ""
+ var start = now_ms()
+ var budget = deadline_ms
+ if budget <= 0:
+ budget = 1
+
+ while not (stdin_done and stdout_eof and stderr_eof):
+ var elapsed = now_ms() - start
+ if elapsed >= budget:
+ read_reason = "read_deadline_expired"
+ break
+ var remaining = budget - elapsed
+ var slice_ms = min(LIFECYCLE_POLL_SLICE_MS, remaining)
+ if slice_ms < 1:
+ slice_ms = 1
+ var ev_stdin = 0 if stdin_done else POLLOUT
+ var ev_out = 0 if stdout_eof else POLLIN
+ var ev_err = 0 if stderr_eof else POLLIN
+ var pr = poll_three(
+ stdin_write_fd,
+ ev_stdin,
+ stdout_read_fd,
+ ev_out,
+ stderr_read_fd,
+ ev_err,
+ slice_ms,
)
+ if pr.count < 0:
+ read_reason = "poll_failed"
+ break
+ if pr.count == 0:
+ continue
+ if not stdin_done:
+ if (pr.r0 & (POLLERR | POLLHUP | POLLNVAL)) != 0:
+ stdin_done = True
+ elif (pr.r0 & POLLOUT) != 0:
+ var cw = write_fd_chunk(stdin_write_fd, request, sent)
+ if cw.reason != "":
+ write_reason = cw.reason
+ stdin_done = True
+ else:
+ sent += cw.written
+ if sent >= request.byte_length():
+ stdin_done = True
+ close_fd(stdin_write_fd)
+ stdin_write_fd = -1
+ if not stdout_eof:
+ var d = drain_ready(
+ stdout_read_fd, stdout, STDIO_MAX_STDOUT_BYTES, pr.r1
+ )
+ stdout_eof = d.eof
+ if d.reason != "":
+ read_reason = "stdout_" + d.reason
+ break
+ if not stderr_eof:
+ var d = drain_ready(
+ stderr_read_fd, stderr_bytes, STDIO_MAX_STDERR_BYTES, pr.r2
+ )
+ stderr_eof = d.eof
+ if d.reason != "":
+ read_reason = "stderr_" + d.reason
+ break
- var output = ""
- try:
- output = read_all_bounded(
- stdout_read_fd, STDIO_MAX_STDOUT_BYTES, deadline_ms
- )
- except:
+ close_fd(stdin_write_fd)
+ if write_reason != "":
_terminate_and_raise(
pid,
-1,
stdout_read_fd,
stderr_read_fd,
- "stdout_bounded_read_failed",
+ "write_" + write_reason,
+ )
+ if read_reason != "":
+ _terminate_and_raise(
+ pid, -1, stdout_read_fd, stderr_read_fd, read_reason
)
- close_fd(stdout_read_fd)
- var diagnostics = _diagnostics_or_empty(stderr_read_fd)
+ var st = wait_bounded(pid, TERMINATION_GRACE_MS)
+ if not st.reaped():
+ _terminate_and_raise(pid, -1, stdout_read_fd, stderr_read_fd, "timeout")
+ close_fd(stdout_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() + ")")
+ var output = bytes_to_string(stdout)
+ var diagnostics = bytes_to_string(stderr_bytes)
+
if st.exited and st.exit_code == 127:
raise Error(
"stdio-entrypoint exec_failed (stdout="
@@ -216,15 +337,6 @@ def run_stdio_entrypoint_with_deadline(
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_stdout_flood_entrypoint.mojo b/tests/stdio_stdout_flood_entrypoint.mojo
@@ -0,0 +1,20 @@
+"""Test-only stdio entrypoint that floods stdout before reading stdin.
+
+Exercises LC04: the parent must drain stdout concurrently with writing a large
+request instead of deadlocking on a full stdout pipe. The flood is whitespace,
+which JSON parsing tolerates, so a correctly interleaved run succeeds.
+"""
+
+from std.io.io import _fdopen
+from std.sys import stdin
+
+
+def main() raises:
+ var spaces = String("")
+ for _ in range(1000):
+ spaces += " "
+ for _ in range(100):
+ print(spaces)
+ with _fdopen["r"](stdin) as input_file:
+ _ = input_file.readline()
+ print('{"ok":true}')
diff --git a/tests/strict_fixture.mojo b/tests/strict_fixture.mojo
@@ -25,6 +25,7 @@ distinct and verified.
from std.collections import List
+from flare.net import Timeout
from flare.tcp import TcpListener
from flare.tcp import TcpStream
from flare.utils import usleep
@@ -216,6 +217,22 @@ def _valid_header_value(value: String) -> Bool:
return True
+def _trim_ows(value: String) -> String:
+ """Trim only legal HTTP optional whitespace (SP / HTAB)."""
+ var start = 0
+ var end = value.byte_length()
+ var bytes = value.as_bytes()
+ while start < end and (Int(bytes[start]) == 32 or Int(bytes[start]) == 9):
+ start += 1
+ while end > start and (
+ Int(bytes[end - 1]) == 32 or Int(bytes[end - 1]) == 9
+ ):
+ end -= 1
+ if start == 0 and end == value.byte_length():
+ return String(value)
+ return String(value[byte=start:end])
+
+
def _ascii_digits(value: String) -> Bool:
if value.byte_length() == 0:
return False
@@ -294,10 +311,13 @@ struct ConnectionReader(Movable):
outcome.error = "header_too_large"
return outcome^
_ = self._read_more()
- if len(self._buffer) > STRICT_MAX_HEADER_BYTES + 4:
+ # Recompute the terminator before any total-length check: a single
+ # coalesced chunk may carry a small header plus a large body, and
+ # only the header bytes count against the header cap.
+ header_end = self._header_end()
+ if header_end < 0 and len(self._buffer) > STRICT_MAX_HEADER_BYTES:
outcome.error = "header_too_large"
return outcome^
- header_end = self._header_end()
if header_end > STRICT_MAX_HEADER_BYTES:
outcome.error = "header_too_large"
return outcome^
@@ -341,8 +361,12 @@ struct ConnectionReader(Movable):
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):
+ var raw_value = String(line[byte = colon + 1 :])
+ if not _valid_header_name(name):
+ outcome.error = "malformed_header"
+ return outcome^
+ var value = _trim_ows(raw_value)
+ if not _valid_header_value(value):
outcome.error = "malformed_header"
return outcome^
var lower = name.lower()
@@ -411,13 +435,19 @@ struct ConnectionReader(Movable):
try:
self._stream.set_recv_timeout(grace_ms)
except:
- return ""
+ return "completion_probe_config_error"
try:
- var n = self._read_more()
- if n > 0:
- return "extra_exchange_after_completion"
- except:
- return ""
+ try:
+ var n = self._read_more()
+ if n > 0:
+ return "extra_exchange_after_completion"
+ except Timeout:
+ # A bounded grace elapsed with no extra bytes: success.
+ return ""
+ except e:
+ # An unexpected I/O error is not a successful completion.
+ _ = String(e)
+ return "completion_probe_error"
return ""
@@ -499,9 +529,11 @@ def verify_exchange(script: ExchangeScript, framed: FramedRequest) -> String:
continue
var split = expected.find(":")
if split <= 0:
- continue
+ return "malformed_script_header"
var name = String(expected[byte=0:split])
- var value = String(expected[byte = split + 1 :].strip())
+ if name.byte_length() == 0 or not _valid_header_name(name):
+ return "malformed_script_header"
+ var value = _trim_ows(String(expected[byte = split + 1 :]))
var values = header_values(framed.headers_raw, name)
if len(values) == 0:
return "header_missing:" + name
@@ -720,42 +752,113 @@ struct ServeReport(Movable):
)
+def report_status_matches_exit(
+ exited: Bool, exit_code: Int, report_ok: Bool
+) -> Bool:
+ """The fixture child exits 0 for a successful report and 125 for a failed
+ one; any other exit (including a signal) contradicts the report."""
+ if not exited:
+ return False
+ if report_ok:
+ return exit_code == 0
+ return exit_code == 125
+
+
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 _strict_nonneg(value: String) -> Int:
+ """Parse a nonnegative decimal count; -1 for empty/non-digit/overflow."""
+ if value.byte_length() == 0 or value.byte_length() > 12:
+ return -1
+ var total = 0
+ for byte in value.as_bytes():
+ var b = Int(byte)
+ if b < 48 or b > 57:
+ return -1
+ total = total * 10 + (b - 48)
+ return total
+
+
+def _report_failure(reason: String) -> ServeReport:
+ return ServeReport(False, "parse", "-", reason, -1, -1)
def parse_report(line: String) -> ServeReport:
- var report = ServeReport(False, "unknown", "-", "missing_report", 0, 0)
+ """Strictly parse one bounded ``result`` report line.
+
+ Missing/truncated/duplicate/malformed/unknown/missing-count fields fail
+ with a bounded cause instead of defaulting counts to zero, and only the
+ exact ``ok``/``fail`` status is accepted.
+ """
if not line.startswith("result "):
- return report^
- var remainder = String(line[byte=7:])
- var parts = remainder.split(" ")
+ return _report_failure("missing_result_prefix")
+ var parts = String(line[byte=7:]).split(" ")
if len(parts) == 0:
- return report^
- report.ok = String(parts[0]) == "ok"
+ return _report_failure("missing_status")
+ var status = String(parts[0])
+ if status != "ok" and status != "fail":
+ return _report_failure("unknown_status")
+ var phase = ""
+ var case_label = ""
+ var reason = ""
+ var requests = -1
+ var connections = -1
+ var have_phase = False
+ var have_case = False
+ var have_reason = False
+ var have_requests = False
+ var have_connections = False
for index in range(1, len(parts)):
var field = String(parts[index])
+ if field.byte_length() == 0:
+ return _report_failure("empty_field")
var eq = field.find("=")
if eq <= 0:
- continue
+ return _report_failure("malformed_field")
var key = String(field[byte=0:eq])
var value = String(field[byte = eq + 1 :])
if key == "phase":
- report.phase = value
+ if have_phase:
+ return _report_failure("duplicate_field")
+ have_phase = True
+ phase = value
elif key == "case":
- report.case_label = value
+ if have_case:
+ return _report_failure("duplicate_field")
+ have_case = True
+ case_label = value
elif key == "reason":
- report.reason = value
+ if have_reason:
+ return _report_failure("duplicate_field")
+ have_reason = True
+ reason = value
elif key == "requests":
- report.requests = _safe_int(value)
+ if have_requests:
+ return _report_failure("duplicate_field")
+ have_requests = True
+ requests = _strict_nonneg(value)
+ if requests < 0:
+ return _report_failure("invalid_count")
elif key == "connections":
- report.connections = _safe_int(value)
- return report^
+ if have_connections:
+ return _report_failure("duplicate_field")
+ have_connections = True
+ connections = _strict_nonneg(value)
+ if connections < 0:
+ return _report_failure("invalid_count")
+ else:
+ return _report_failure("unknown_field")
+ if not (
+ have_phase
+ and have_case
+ and have_reason
+ and have_requests
+ and have_connections
+ ):
+ return _report_failure("missing_field")
+ return ServeReport(
+ status == "ok", phase, case_label, reason, requests, connections
+ )
diff --git a/tests/test_jev.mojo b/tests/test_jev.mojo
@@ -281,46 +281,52 @@ from jev_provider_helper import (
def test_local_provider_server_serves_scripted_jev() raises:
- var started = spawn_jev_stub_auto("ok", 1)
- var port = started.port
- var url = "http://127.0.0.1:" + String(port) + "/v1/systemone"
- with HttpClient(timeout_ms=5000, max_redirects=0) as client:
- var response = client.post(
- url, '{"model":"jev-1.13.0","state":"s","questions":{}}'
- )
- assert_true(response.ok())
- var body = response.json()
- assert_equal(body["model"].string_value(), "jev-1.13.0")
- started.stub.wait()
+ with spawn_jev_stub_auto("ok", 1) as started:
+ var port = started.port
+ var url = "http://127.0.0.1:" + String(port) + "/v1/systemone"
+ with HttpClient(timeout_ms=5000, max_redirects=0) as client:
+ var response = client.post(
+ url, '{"model":"jev-1.13.0","state":"s","questions":{}}'
+ )
+ assert_true(response.ok())
+ var body = response.json()
+ assert_equal(body["model"].string_value(), "jev-1.13.0")
+ started.stub.wait()
def test_local_provider_server_scripts_transport_failures() raises:
- var rate_port_started = spawn_jev_stub_auto("rate_limit", 1)
- var rate_port = rate_port_started.port
- with HttpClient(timeout_ms=5000, max_redirects=0) as client:
- var response = client.post(
- "http://127.0.0.1:" + String(rate_port) + "/v1/systemone", "{}"
- )
- assert_equal(response.status, 429)
- rate_port_started.stub.wait()
-
- var malformed_port_started = spawn_jev_stub_auto("malformed_json", 1)
- var malformed_port = malformed_port_started.port
- with HttpClient(timeout_ms=5000, max_redirects=0) as client:
- var response = client.post(
- "http://127.0.0.1:" + String(malformed_port) + "/v1/systemone", "{}"
- )
- assert_equal(response.text(), "not json")
- malformed_port_started.stub.wait()
-
- var err_port_started = spawn_jev_stub_auto("server_error", 1)
- var err_port = err_port_started.port
- with HttpClient(timeout_ms=5000, max_redirects=0) as client:
- var response = client.post(
- "http://127.0.0.1:" + String(err_port) + "/v1/systemone", "{}"
- )
- assert_equal(response.status, 500)
- err_port_started.stub.wait()
+ with spawn_jev_stub_auto("rate_limit", 1) as rate_port_started:
+ var rate_port = rate_port_started.port
+ with HttpClient(timeout_ms=5000, max_redirects=0) as client:
+ var response = client.post(
+ "http://127.0.0.1:" + String(rate_port) + "/v1/systemone", "{}"
+ )
+ assert_equal(response.status, 429)
+ rate_port_started.stub.wait()
+
+ with spawn_jev_stub_auto("malformed_json", 1) as malformed_port_started:
+ var malformed_port = malformed_port_started.port
+ with HttpClient(timeout_ms=5000, max_redirects=0) as client:
+ var response = client.post(
+ "http://127.0.0.1:"
+ + String(malformed_port)
+ + "/v1/systemone",
+ "{}",
+ )
+ assert_equal(response.text(), "not json")
+ malformed_port_started.stub.wait()
+
+ with spawn_jev_stub_auto("server_error", 1) as err_port_started:
+ var err_port = err_port_started.port
+ with HttpClient(timeout_ms=5000, max_redirects=0) as client:
+ var response = client.post(
+ "http://127.0.0.1:"
+ + String(err_port)
+ + "/v1/systemone",
+ "{}",
+ )
+ assert_equal(response.status, 500)
+ err_port_started.stub.wait()
from hyf_provider.jev_client import post_jev_systemone, validate_jev_base_url
@@ -345,16 +351,16 @@ def test_jev_endpoint_policy_and_loopback_client() raises:
with assert_raises():
_ = validate_jev_base_url("ftp://api.typesafe.ai")
- var started = spawn_jev_stub_auto("ok", 1)
- var port = started.port
- var outcome = post_jev_systemone(
- "http://127.0.0.1:" + String(port),
- _loads('{"model":"jev-1.13.0","state":"s","questions":{}}'),
- 5000,
- )
- assert_equal(outcome.status, 200)
- assert_true(outcome.body_text.find("jev-1.13.0") >= 0)
- started.stub.wait()
+ with spawn_jev_stub_auto("ok", 1) as started:
+ var port = started.port
+ var outcome = post_jev_systemone(
+ "http://127.0.0.1:" + String(port),
+ _loads('{"model":"jev-1.13.0","state":"s","questions":{}}'),
+ 5000,
+ )
+ assert_equal(outcome.status, 200)
+ assert_true(outcome.body_text.find("jev-1.13.0") >= 0)
+ started.stub.wait()
from flare.tls import TlsVerify
@@ -374,14 +380,14 @@ def test_tls_and_redirect_policy() raises:
assert_tls_verification_required(TlsConfig.insecure())
assert_true(not redirects_forward_credentials())
- var started = spawn_jev_stub_auto("redirect", 1)
- var port = started.port
- with assert_raises():
- with HttpClient(timeout_ms=5000, max_redirects=0) as client:
- _ = client.post(
- "http://127.0.0.1:" + String(port) + "/v1/systemone", "{}"
- )
- started.stub.wait()
+ with spawn_jev_stub_auto("redirect", 1) as started:
+ var port = started.port
+ with assert_raises():
+ with HttpClient(timeout_ms=5000, max_redirects=0) as client:
+ _ = client.post(
+ "http://127.0.0.1:" + String(port) + "/v1/systemone", "{}"
+ )
+ started.stub.wait()
from hyf_provider.jev_client import failure_kind_for_status, retry_decision
@@ -404,21 +410,21 @@ def test_bounded_retry_and_budget_behavior() raises:
def test_transport_cleanup_and_local_cancellation() raises:
- var started = spawn_jev_stub_auto("ok", 1)
- var port = started.port
- var outcome = post_jev_systemone(
- "http://127.0.0.1:" + String(port),
- _loads('{"model":"jev-1.13.0","state":"s","questions":{}}'),
- 5000,
- )
- assert_equal(outcome.status, 200)
- started.stub.wait()
-
- # A refused connection is a bounded local transport failure (no listener).
- var dead_port = reserve_jev_port()
- with assert_raises():
- _ = post_jev_systemone(
- "http://127.0.0.1:" + String(dead_port),
+ with spawn_jev_stub_auto("ok", 1) as started:
+ var port = started.port
+ var outcome = post_jev_systemone(
+ "http://127.0.0.1:" + String(port),
_loads('{"model":"jev-1.13.0","state":"s","questions":{}}'),
- 150,
+ 5000,
)
+ assert_equal(outcome.status, 200)
+ started.stub.wait()
+
+ # A refused connection is a bounded local transport failure (no listener).
+ var dead_port = reserve_jev_port()
+ with assert_raises():
+ _ = post_jev_systemone(
+ "http://127.0.0.1:" + String(dead_port),
+ _loads('{"model":"jev-1.13.0","state":"s","questions":{}}'),
+ 150,
+ )
diff --git a/tests/test_provider_adapter.mojo b/tests/test_provider_adapter.mojo
@@ -261,33 +261,35 @@ def test_max_local_transport_boundary_rejects_invalid_health_url() raises:
def test_max_local_transport_boundary_reports_unknown_chat_transport() raises:
- var provider_stub = spawn_max_local_stub(
+ with spawn_max_local_stub(
0, "query_rewrite_malformed_http", 1
- )
- var provider_port = provider_stub.port
- var outcome = post_max_local_chat_completion(
- _provider_config_for_port(provider_port), loads("{}")
- )
+ ) as provider_stub:
+ var provider_port = provider_stub.port
+ var outcome = post_max_local_chat_completion(
+ _provider_config_for_port(provider_port), loads("{}")
+ )
- assert_true(outcome.failure)
- assert_true(not outcome.response)
- assert_equal(outcome.failure.value().kind, "transport")
- assert_equal(outcome.failure.value().reason, "unknown_transport")
+ assert_true(outcome.failure)
+ assert_true(not outcome.response)
+ assert_equal(outcome.failure.value().kind, "transport")
+ assert_equal(outcome.failure.value().reason, "unknown_transport")
- provider_stub.wait()
+ provider_stub.wait()
def test_max_local_transport_boundary_reports_unknown_health_transport() raises:
- var provider_stub = spawn_max_local_stub(0, "health_malformed_http", 1)
- var provider_port = provider_stub.port
- var outcome = get_max_local_health(_provider_config_for_port(provider_port))
+ with spawn_max_local_stub(0, "health_malformed_http", 1) as provider_stub:
+ var provider_port = provider_stub.port
+ var outcome = get_max_local_health(
+ _provider_config_for_port(provider_port)
+ )
- assert_true(outcome.failure)
- assert_true(not outcome.response)
- assert_equal(outcome.failure.value().kind, "transport")
- assert_equal(outcome.failure.value().reason, "unknown_transport")
+ assert_true(outcome.failure)
+ assert_true(not outcome.response)
+ assert_equal(outcome.failure.value().kind, "transport")
+ assert_equal(outcome.failure.value().reason, "unknown_transport")
- provider_stub.wait()
+ provider_stub.wait()
def test_query_rewrite_request_body_sets_schema_contract() raises:
diff --git a/tests/test_provider_helpers.mojo b/tests/test_provider_helpers.mojo
@@ -6,17 +6,26 @@ exception, signal or unrelated startup failure.
"""
from std.testing import TestSuite, assert_true, assert_equal
+from std.ffi import ErrNo
from flare.net import SocketAddr
from flare.tcp import TcpListener, TcpStream
from parent_lifecycle import (
+ PipedChildState,
+ ProcessStatus,
+ child_exit,
+ classify_wait_errno,
close_fd,
+ descriptor_census,
+ dup2_fd,
+ fork_pid,
make_pipe,
open_fd_count,
pid_not_waitable,
read_all_bounded,
read_line_bounded,
+ wait_nohang,
write_fd_bounded,
write_raw,
)
@@ -26,6 +35,8 @@ from strict_fixture import (
authorization_reason,
exchange_script,
json_escape,
+ parse_report,
+ report_status_matches_exit,
verify_exchange,
)
from max_local_process_helper import (
@@ -131,72 +142,97 @@ def _default_script() -> ExchangeScript:
)
+@fieldwise_init
+struct FramingFailure(Movable):
+ """Bounded snapshot of a fixture read failure for post-scope asserts."""
+
+ var ok: Bool
+ var phase_value: String
+ var case_value: String
+ var reason_value: String
+ var requests: Int
+ var connections: Int
+
+ def phase(self) -> String:
+ return String(self.phase_value)
+
+ def reason(self) -> String:
+ return String(self.reason_value)
+
+ def failure_case(self) -> String:
+ return String(self.case_value)
+
+
# ── 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 = _request(stub.port, "POST", "/v1/chat/completions", body)
- assert_true(response.find('"received_bytes":9000') >= 0)
- stub.wait()
+ with spawn_max_local_stub(0, "echo_body_bytes", 1) as stub:
+ var body = String("")
+ for _ in range(9000):
+ body += "x"
+ var response = _request(stub.port, "POST", "/v1/chat/completions", body)
+ assert_true(response.find('"received_bytes":9000') >= 0)
+ stub.wait()
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 = _request(stub.port, "POST", "/v1/chat/completions", "{}")
- assert_true(response.find('"request_index":' + String(index + 1)) >= 0)
- stub.wait()
+ with spawn_max_local_stub(0, "count_requests", requests) as stub:
+ for index in range(requests):
+ 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 = _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")
+ with spawn_max_local_stub(0, "query_rewrite_ok", 1) as stub:
+ 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 = _request(stub.port, "POST", "/v1/chat/completions", "{}")
- assert_true(response.find('"request_index":1') >= 0)
- stub.wait()
+ with spawn_max_local_stub(0, "count_requests", 1) as stub:
+ assert_true(stub.port > 0)
+ 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 = _request(
- started.port,
- "POST",
- "/v1/systemone",
- "{}",
- "authorization: Bearer hyf-sentinel-token\r\n",
- )
- assert_true(response.find("hyf-sentinel-token") >= 0)
- started.stub.wait()
+ with spawn_jev_stub_auto("echo_authorization", 1) as started:
+ 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 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{}"
- ),
- )
- stub.terminate()
- assert_true(pid_not_waitable(stub.pid))
+ with spawn_max_local_stub(0, "stall", 1) as stub:
+ _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{}"
+ ),
+ )
+ stub.terminate()
+ assert_true(pid_not_waitable(stub.pid))
# ── FX01: explicit ordered scripted exchanges ───────────────────────────────
@@ -213,52 +249,52 @@ def test_max_local_scripted_matches_explicit_exchange() raises:
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)
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ 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_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,
- (
- "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{}"
- ),
- )
- stub.reap()
- assert_equal(stub.phase(), "exchange")
- assert_equal(stub.reason(), "method_mismatch")
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ _raw_send_only(
+ stub.port,
+ (
+ "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{}"
+ ),
+ )
+ stub.reap()
+ assert_equal(stub.phase(), "exchange")
+ assert_equal(stub.reason(), "method_mismatch")
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")
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ _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:
@@ -266,18 +302,18 @@ def test_max_local_scripted_rejects_wrong_selected_header() raises:
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"
- "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")
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ _raw_send_only(
+ stub.port,
+ (
+ "POST /v1/chat/completions HTTP/1.1\r\nhost:"
+ " 127.0.0.1\r\nx-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:
@@ -286,17 +322,17 @@ def test_max_local_scripted_rejects_wrong_body() raises:
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} '
- ),
- )
- stub.reap()
- assert_equal(stub.phase(), "exchange")
- assert_equal(stub.reason(), "body_mismatch")
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ _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} '
+ ),
+ )
+ stub.reap()
+ assert_equal(stub.phase(), "exchange")
+ assert_equal(stub.reason(), "body_mismatch")
# ── FX02: unexpected/extra/missing/unconsumed accounting ────────────────────
@@ -307,28 +343,28 @@ def test_scripted_rejects_extra_pipelined_exchange() raises:
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")
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ 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)
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ 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:
@@ -337,33 +373,40 @@ def test_scripted_reports_unconsumed_remaining_scripts() raises:
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: 2\r\nconnection: keep-alive\r\n\r\n{}"
- ),
- )
- 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)
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ 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: 2\r\nconnection: keep-alive\r\n\r\n{}"
+ ),
+ )
+ 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)
# ── FX03: strict lexical framing ────────────────────────────────────────────
-def _framing_failure(raw: String) raises -> SpawnedMaxLocalStub:
+def _framing_failure(raw: String) raises -> FramingFailure:
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^
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ _raw_send_only(stub.port, raw)
+ stub.reap()
+ return FramingFailure(
+ stub.ok(),
+ stub.phase(),
+ stub.failure_case(),
+ stub.reason(),
+ stub.request_count(),
+ stub.connection_count(),
+ )
def test_strict_framing_lexical_content_length() raises:
@@ -479,17 +522,17 @@ def test_strict_framing_header_cap_exceeded() raises:
def test_jev_echo_authorization_ignores_x_authorization() raises:
- var started = spawn_jev_stub_auto("echo_authorization", 1)
- var response = _request(
- started.port,
- "POST",
- "/v1/systemone",
- "{}",
- "x-authorization: Bearer spoof\r\n",
- )
- assert_true(response.find("401") >= 0)
- assert_true(response.find("spoof") < 0)
- started.stub.wait()
+ with spawn_jev_stub_auto("echo_authorization", 1) as started:
+ var response = _request(
+ started.port,
+ "POST",
+ "/v1/systemone",
+ "{}",
+ "x-authorization: Bearer spoof\r\n",
+ )
+ assert_true(response.find("401") >= 0)
+ assert_true(response.find("spoof") < 0)
+ started.stub.wait()
def test_strict_framing_split_utf8_body() raises:
@@ -503,27 +546,27 @@ def test_strict_framing_split_utf8_body() raises:
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()
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ 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_strict_framing_surplus_retained_for_second_frame() raises:
@@ -537,18 +580,18 @@ def test_strict_framing_surplus_retained_for_second_frame() raises:
"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)
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ 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 ────────────
@@ -561,18 +604,18 @@ def test_jev_scripted_wrong_route_auth_not_bypassed() raises:
)
script.require_bearer = True
scripts.append(script^)
- var started = spawn_jev_scripted_auto(scripts^)
- _raw_send_only(
- started.port,
- (
- "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")
+ with spawn_jev_scripted_auto(scripts^) as started:
+ _raw_send_only(
+ started.port,
+ (
+ "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:
@@ -582,18 +625,18 @@ def test_scripted_rejects_duplicate_authorization() raises:
)
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{}"
- ),
- )
- stub.reap()
- assert_equal(stub.phase(), "exchange")
- assert_equal(stub.reason(), "auth_duplicate")
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ _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{}"
+ ),
+ )
+ stub.reap()
+ assert_equal(stub.phase(), "exchange")
+ assert_equal(stub.reason(), "auth_duplicate")
def test_json_escape_control_characters() raises:
@@ -714,21 +757,21 @@ def test_scripted_persistent_counters_and_close_semantics() raises:
)
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)
+ with spawn_jev_scripted_auto(scripts^) as started:
+ 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)
# ── FX06/FX07/FX08: parent lifecycle and cause-specific failures ────────────
@@ -742,8 +785,8 @@ def test_startup_failure_distinct_from_exchange_failure() raises:
var port = Int(blocker.local_addr().port)
var message = ""
try:
- var stub = spawn_max_local_stub(port, "count_requests", 1)
- stub.terminate()
+ with spawn_max_local_stub(port, "count_requests", 1) as stub:
+ stub.terminate()
except e:
message = String(e)
blocker.close()
@@ -755,21 +798,21 @@ def test_startup_failure_distinct_from_exchange_failure() raises:
def test_provider_stub_parent_deadline_watchdog() raises:
# No client connects, so the child blocks in accept until the parent's own
# finite deadline fires and the owned child is terminated and reaped.
- var stub = spawn_max_local_stub(0, "count_requests", 1, 800)
- stub.reap()
- assert_true(not stub.ok())
- assert_equal(stub.phase(), "watchdog")
- assert_equal(stub.reason(), "timeout")
- assert_true(pid_not_waitable(stub.pid))
+ with spawn_max_local_stub(0, "count_requests", 1, 800) as stub:
+ stub.reap()
+ assert_true(not stub.ok())
+ assert_equal(stub.phase(), "watchdog")
+ assert_equal(stub.reason(), "timeout")
+ assert_true(pid_not_waitable(stub.pid))
def test_jev_stub_parent_deadline_watchdog() raises:
- var started = spawn_jev_stub_auto("ok", 1, 800)
- started.stub.reap()
- assert_true(not started.stub.ok())
- assert_equal(started.stub.phase(), "watchdog")
- assert_equal(started.stub.reason(), "timeout")
- assert_true(pid_not_waitable(started.stub.pid))
+ with spawn_jev_stub_auto("ok", 1, 800) as started:
+ started.stub.reap()
+ assert_true(not started.stub.ok())
+ assert_equal(started.stub.phase(), "watchdog")
+ assert_equal(started.stub.reason(), "timeout")
+ assert_true(pid_not_waitable(started.stub.pid))
def test_bounded_read_caps_fail_for_intended_cause() raises:
@@ -819,52 +862,410 @@ def test_write_deadline_and_closed_pipe_causes() raises:
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))
+ with spawn_max_local_stub(0, "count_requests", 1) as stub:
+ 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))
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ _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_repeated_teardown_does_not_leak_descriptors() raises:
var before = open_fd_count()
assert_true(before > 0)
for _ in range(5):
- var stub = spawn_max_local_stub(0, "count_requests", 1)
- stub.terminate()
- assert_true(pid_not_waitable(stub.pid))
+ with spawn_max_local_stub(0, "count_requests", 1) as stub:
+ stub.terminate()
+ assert_true(pid_not_waitable(stub.pid))
var after = open_fd_count()
assert_true(after > 0)
assert_true(after <= before)
def test_timeout_terminates_and_reaps_stalled_child() raises:
- var stub = spawn_max_local_stub(0, "stall", 1)
- _raw_send_only(
- stub.port,
+ with spawn_max_local_stub(0, "stall", 1) as stub:
+ _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{}"
+ ),
+ )
+ stub.terminate()
+ assert_true(pid_not_waitable(stub.pid))
+
+
+def _owned_report_child(
+ exit_code: Int, report: String
+) raises -> SpawnedMaxLocalStub:
+ """Fork a test-owned child that writes ``report`` to stdout and exits.
+
+ Lets LC02 prove real forged/empty/mismatched reports fail for their cause,
+ independent of the fixture serve loop.
+ """
+ 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)
+ if report != "":
+ _ = write_raw(1, report)
+ child_exit(exit_code)
+ close_fd(pipe.write_fd)
+ var state = PipedChildState(
+ pid=pid,
+ report_fd=pipe.read_fd,
+ pending="",
+ eof=False,
+ closed=False,
+ deadline_ms=2000,
+ expected_requests=1,
+ reaped=False,
+ ok=False,
+ phase="pending",
+ case_label="-",
+ reason="not_reaped",
+ requests=0,
+ connections=0,
+ cleanup_error="",
+ status=ProcessStatus("pending", False, -1, 0, 0, ""),
+ )
+ return SpawnedMaxLocalStub(pid, 0, state^)
+
+
+# ── LC01: automatic scope ownership ─────────────────────────────────────────
+
+
+def test_scope_cleanup_on_assertion_failure() raises:
+ var held_pid = 0
+ var caught = False
+ try:
+ with spawn_max_local_stub(0, "count_requests", 1) as stub:
+ held_pid = stub.pid
+ assert_true(False)
+ except:
+ caught = True
+ assert_true(caught)
+ assert_true(held_pid > 0)
+ assert_true(pid_not_waitable(held_pid))
+
+
+def test_scope_cleanup_on_generic_error() raises:
+ var held_pid = 0
+ var message = ""
+ try:
+ with spawn_max_local_stub(0, "count_requests", 1) as stub:
+ held_pid = stub.pid
+ raise Error("intentional scope error")
+ except e:
+ message = String(e)
+ assert_equal(message, "intentional scope error")
+ assert_true(pid_not_waitable(held_pid))
+
+
+def _early_return_owner() raises -> Int:
+ with spawn_max_local_stub(0, "count_requests", 1) as stub:
+ return stub.pid
+ return 0
+
+
+def test_scope_cleanup_on_early_return() raises:
+ var held_pid = _early_return_owner()
+ assert_true(held_pid > 0)
+ assert_true(pid_not_waitable(held_pid))
+
+
+def test_jev_scope_cleanup_on_assertion_failure() raises:
+ var held_pid = 0
+ var caught = False
+ try:
+ with spawn_jev_stub_auto("ok", 1) as started:
+ held_pid = started.stub.pid
+ assert_true(False)
+ except:
+ caught = True
+ assert_true(caught)
+ assert_true(held_pid > 0)
+ assert_true(pid_not_waitable(held_pid))
+
+
+def test_wait_error_taxonomy_distinguishes_causes() raises:
+ assert_equal(classify_wait_errno(Int(ErrNo.EINTR.value)), "interrupted")
+ assert_equal(classify_wait_errno(Int(ErrNo.ECHILD.value)), "gone")
+ assert_equal(classify_wait_errno(9999), "wait_error")
+ assert_equal(wait_nohang(0).state, "wait_error")
+ var stub = spawn_max_local_stub(0, "count_requests", 1)
+ var live_pid = stub.pid
+ assert_equal(wait_nohang(live_pid).state, "running")
+ stub.terminate()
+ assert_equal(wait_nohang(live_pid).state, "gone")
+ assert_true(pid_not_waitable(live_pid))
+
+
+def test_repeated_reap_and_terminate_are_owned_and_idempotent() raises:
+ with spawn_max_local_stub(0, "count_requests", 1) as stub:
+ var owned_pid = stub.pid
+ stub.terminate()
+ stub.terminate()
+ stub.reap()
+ assert_true(pid_not_waitable(owned_pid))
+ var cached = stub.status()
+ assert_true(cached.cleanup_proved())
+ assert_true(not stub.ok())
+
+
+# ── LC02: strict result truth ───────────────────────────────────────────────
+
+
+def test_result_truth_rejects_empty_exit_zero_report() raises:
+ var stub = _owned_report_child(0, "")
+ stub.reap()
+ assert_true(not stub.ok())
+ assert_equal(stub.reason(), "missing_report")
+
+
+def test_result_truth_rejects_forged_success_with_nonzero_exit() raises:
+ var stub = _owned_report_child(
+ 7,
+ "result ok phase=complete case=- reason=ok requests=1 connections=1\n",
+ )
+ stub.reap()
+ assert_true(not stub.ok())
+ assert_true(stub.reason().startswith("report_status_mismatch"))
+
+
+def test_result_truth_accepts_matching_report_and_exit() raises:
+ var stub = _owned_report_child(
+ 0,
+ "result ok phase=complete case=- reason=ok requests=1 connections=1\n",
+ )
+ stub.reap()
+ assert_true(stub.ok())
+ assert_equal(stub.phase(), "complete")
+ assert_equal(stub.request_count(), 1)
+ assert_equal(stub.connection_count(), 1)
+
+
+def test_parse_report_rejects_malformed_inputs() raises:
+ assert_equal(parse_report("").phase, "parse")
+ assert_equal(
+ parse_report(
+ "result maybe phase=x case=- reason=y requests=1 connections=1"
+ ).reason,
+ "unknown_status",
+ )
+ assert_equal(
+ parse_report(
+ "result ok phase=complete case=- reason=ok requests=1 connections=1"
+ " requests=1"
+ ).reason,
+ "duplicate_field",
+ )
+ assert_equal(
+ parse_report(
+ "result ok phase=complete case=- reason=ok requests=x connections=1"
+ ).reason,
+ "invalid_count",
+ )
+ assert_equal(
+ parse_report(
+ "result ok phase=complete case=- reason=ok requests=1"
+ ).reason,
+ "missing_field",
+ )
+ assert_equal(
+ parse_report(
+ "result ok phase=complete case=- reason=ok requests=1 connections=1"
+ " extra=z"
+ ).reason,
+ "unknown_field",
+ )
+ assert_equal(
+ parse_report(
+ "result ok phase=complete case=- reason=ok requests=1 connections=1"
+ ).phase,
+ "complete",
+ )
+
+
+def test_report_status_matches_exit() raises:
+ assert_true(report_status_matches_exit(True, 0, True))
+ assert_true(report_status_matches_exit(True, 125, False))
+ assert_true(not report_status_matches_exit(True, 7, True))
+ assert_true(not report_status_matches_exit(True, 0, False))
+ assert_true(not report_status_matches_exit(False, 0, True))
+
+
+# ── LC03: byte caps and surplus retention ───────────────────────────────────
+
+
+def _pipe_line(text: String) raises -> String:
+ var pipe = make_pipe()
+ _ = write_raw(pipe.write_fd, text)
+ var result = ""
+ var raised = ""
+ try:
+ result = read_line_bounded(pipe.read_fd, 8, 500)
+ except e:
+ raised = String(e)
+ close_fd(pipe.read_fd)
+ close_fd(pipe.write_fd)
+ if raised != "":
+ return "raised:" + raised
+ return result^
+
+
+def test_line_cap_boundaries() raises:
+ assert_equal(_pipe_line("1234567\n"), "1234567")
+ assert_equal(_pipe_line("12345678\n"), "12345678")
+ assert_equal(_pipe_line("123456789\n"), "raised:ready_output_overflow")
+
+
+def test_coalesced_ready_and_report_lines_retain_surplus() raises:
+ # One write carries both lines; the report line must survive the ready read
+ # rather than being discarded with the chunk.
+ var pipe = make_pipe()
+ _ = write_raw(
+ pipe.write_fd,
(
- "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{}"
+ "ready 4242\nresult ok phase=complete case=- reason=ok requests=1"
+ " connections=1\n"
),
)
- stub.terminate()
- assert_true(pid_not_waitable(stub.pid))
+ var state = PipedChildState(
+ pid=0,
+ report_fd=pipe.read_fd,
+ pending="",
+ eof=False,
+ closed=False,
+ deadline_ms=500,
+ expected_requests=1,
+ reaped=False,
+ ok=False,
+ phase="pending",
+ case_label="-",
+ reason="not_reaped",
+ requests=0,
+ connections=0,
+ cleanup_error="",
+ status=ProcessStatus("pending", False, -1, 0, 0, ""),
+ )
+ var ready = state.read_line(2048, 500)
+ var report = state.read_line(2048, 500)
+ state.close_reader()
+ close_fd(pipe.write_fd)
+ assert_equal(ready, "ready 4242")
+ assert_true(report.startswith("result ok"))
+
+
+# ── LC05: framing and descriptor census ─────────────────────────────────────
+
+
+def test_verify_exchange_rejects_malformed_script_header() raises:
+ var script = exchange_script(
+ "bad_decl", "POST", "/v1/chat/completions", 200, "{}"
+ )
+ script.headers = "not-a-header"
+ var framed = FramedRequest(
+ ok=True,
+ error="",
+ method="POST",
+ path="/v1/chat/completions",
+ version="HTTP/1.1",
+ headers_raw="host: h",
+ body="",
+ content_length=0,
+ keep_alive=False,
+ total_bytes=0,
+ )
+ assert_equal(verify_exchange(script, framed), "malformed_script_header")
+
+
+def test_header_value_rejects_control_bytes_and_trims_ows() raises:
+ # Raw control byte inside a value is rejected as malformed framing.
+ var bad = _framing_failure(
+ "POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\n"
+ "x-ctl: a\x01b\r\ncontent-length: 2\r\nconnection: close\r\n\r\n{}"
+ )
+ assert_equal(bad.phase(), "read")
+ assert_equal(bad.reason(), "malformed_header")
+ # Legal surrounding OWS on a selected header value is accepted.
+ var scripts = List[ExchangeScript]()
+ var script = exchange_script(
+ "ows", "POST", "/v1/chat/completions", 200, "{}"
+ )
+ script.headers = "x-ows:value"
+ scripts.append(script^)
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ var response = _request(
+ stub.port,
+ "POST",
+ "/v1/chat/completions",
+ "{}",
+ "x-ows: value \r\n",
+ )
+ assert_true(response.find("200") >= 0)
+ stub.wait()
+
+
+def test_descriptor_census_detects_planted_high_fd() raises:
+ var before = open_fd_count()
+ assert_true(before > 0)
+ assert_equal(descriptor_census(0), -1)
+ assert_true(descriptor_census(3) > 0)
+ var pipe = make_pipe()
+ var planted = Int(dup2_fd(pipe.read_fd, 900))
+ var with_pipe = open_fd_count()
+ assert_true(planted >= 0)
+ assert_true(with_pipe > before)
+ close_fd(planted)
+ close_fd(pipe.write_fd)
+ var after = open_fd_count()
+ assert_true(after <= with_pipe)
+
+
+# ── LC05: coalesced header cap accounting ───────────────────────────────────
+
+
+def test_coalesced_large_body_does_not_charge_header_cap() raises:
+ var scripts = List[ExchangeScript]()
+ scripts.append(_default_script())
+ with spawn_max_local_scripted(0, scripts^) as stub:
+ var header_filler = String("")
+ for _ in range(20000):
+ header_filler += "a"
+ var body_filler = String("")
+ for _ in range(50000):
+ body_filler += "b"
+ var raw = (
+ "POST /v1/chat/completions HTTP/1.1\r\nhost:"
+ " 127.0.0.1\r\nx-filler: "
+ + header_filler
+ + "\r\ncontent-length: "
+ + String(body_filler.byte_length())
+ + "\r\nconnection: close\r\n\r\n"
+ + body_filler
+ )
+ var response = _raw_exchange(stub.port, raw)
+ assert_true(response.find("200") >= 0)
+ stub.wait()
def main() raises:
diff --git a/tests/test_repo_local_process_contract.mojo b/tests/test_repo_local_process_contract.mojo
@@ -1,12 +1,15 @@
from std.testing import TestSuite, assert_equal, assert_true
+from std.collections import List
from safe_tempdir import SafeTempDir
from json import Value
from fixture_assertions import load_scenario_request_json
+from parent_lifecycle import POLLIN, close_fd, make_pipe, write_raw
from stdio_process_helper import (
HYF_PATHS_PROFILE_ENV,
HYF_PATHS_REPO_LOCAL_ROOT_ENV,
ScopedEnvVar,
+ drain_ready,
run_stdio_entrypoint,
run_stdio_entrypoint_with_deadline,
)
@@ -116,5 +119,50 @@ def test_run_stdio_entrypoint_classifies_loader_failure() raises:
assert_true(message.find("timeout") < 0)
+def test_run_stdio_entrypoint_drains_stdout_concurrently() raises:
+ # LC04: a child that floods stdout before reading stdin must not deadlock
+ # the parent's large request write; a correct interleaving succeeds.
+ var request = String("")
+ for _ in range(150000):
+ request += "r"
+ var response = Value(None)
+ var message = ""
+ var failed = False
+ try:
+ response = run_stdio_entrypoint_with_deadline(
+ "tests/stdio_stdout_flood_entrypoint.mojo",
+ request,
+ "",
+ "",
+ 30000,
+ )
+ except e:
+ failed = True
+ message = String(e)
+ assert_true(not failed)
+ assert_equal(message, "")
+ assert_true(response["ok"].bool_value())
+
+
+def test_diagnostics_overflow_is_bounded_with_cause() raises:
+ # LC04: stderr diagnostics are capped and an overflow is a distinct cause.
+ var pipe = make_pipe()
+ var chunk = String("")
+ for _ in range(4096):
+ chunk += "e"
+ var out = List[UInt8]()
+ var overflow_reason = ""
+ for _ in range(20):
+ _ = write_raw(pipe.write_fd, chunk)
+ var d = drain_ready(pipe.read_fd, out, 65536, POLLIN)
+ if d.reason != "":
+ overflow_reason = d.reason
+ break
+ close_fd(pipe.read_fd)
+ close_fd(pipe.write_fd)
+ assert_equal(overflow_reason, "stream_overflow")
+ assert_true(len(out) <= 65536)
+
+
def main() raises:
TestSuite.discover_tests[__functions_in_module()]().run()
diff --git a/tests/test_stdio_contract.mojo b/tests/test_stdio_contract.mojo
@@ -198,58 +198,60 @@ def _assert_query_rewrite_provider_fallback_with_deadline(
requests: Int,
) raises:
with SafeTempDir() as temp_dir:
- var provider_stub = spawn_max_local_stub(0, mode, requests)
- var provider_port = provider_stub.port
- var startup_config_path = Path(temp_dir) / "explicit-hyf-config.toml"
- startup_config_path.write_text(
- _max_local_runtime_config_toml_with_urls(
- "http://127.0.0.1:" + String(provider_port) + "/v1",
- "http://127.0.0.1:" + String(provider_port) + "/health",
- request_timeout_ms,
+ with spawn_max_local_stub(0, mode, requests) as provider_stub:
+ var provider_port = provider_stub.port
+ var startup_config_path = (
+ Path(temp_dir) / "explicit-hyf-config.toml"
)
- )
- with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
- with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
- var response = run_stdio_entrypoint(
- "src/main.mojo",
- _query_rewrite_assisted_request_json_with_deadline(
- "rewrite-assisted-" + mode, deadline_ms
- ),
- "--config",
- startup_config_path.__fspath__(),
+ startup_config_path.write_text(
+ _max_local_runtime_config_toml_with_urls(
+ "http://127.0.0.1:" + String(provider_port) + "/v1",
+ "http://127.0.0.1:" + String(provider_port) + "/health",
+ request_timeout_ms,
)
+ )
+ with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
+ with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
+ var response = run_stdio_entrypoint(
+ "src/main.mojo",
+ _query_rewrite_assisted_request_json_with_deadline(
+ "rewrite-assisted-" + mode, deadline_ms
+ ),
+ "--config",
+ startup_config_path.__fspath__(),
+ )
- assert_true(response["ok"].bool_value())
- assert_equal(
- response["meta"]["execution_mode"].string_value(),
- "deterministic",
- )
- assert_equal(
- response["meta"]["backend"].string_value(),
- "heuristic",
- )
- assert_true(not _has_key(response["meta"], "provider"))
- _assert_provider_runtime_fallback_meta(
- response, expected_reason
- )
- assert_equal(
- response["meta"]["provenance"]["fallback"][
- "fallback_kind"
- ].string_value(),
- "provider_runtime",
- )
- assert_equal(
- response["meta"]["provenance"]["fallback"][
- "reason"
- ].string_value(),
- expected_reason,
- )
- assert_equal(
- response["output"]["rewritten_text"].string_value(),
- "apples",
- )
+ assert_true(response["ok"].bool_value())
+ assert_equal(
+ response["meta"]["execution_mode"].string_value(),
+ "deterministic",
+ )
+ assert_equal(
+ response["meta"]["backend"].string_value(),
+ "heuristic",
+ )
+ assert_true(not _has_key(response["meta"], "provider"))
+ _assert_provider_runtime_fallback_meta(
+ response, expected_reason
+ )
+ assert_equal(
+ response["meta"]["provenance"]["fallback"][
+ "fallback_kind"
+ ].string_value(),
+ "provider_runtime",
+ )
+ assert_equal(
+ response["meta"]["provenance"]["fallback"][
+ "reason"
+ ].string_value(),
+ expected_reason,
+ )
+ assert_equal(
+ response["output"]["rewritten_text"].string_value(),
+ "apples",
+ )
- provider_stub.wait()
+ provider_stub.wait()
def _assert_query_rewrite_provider_fallback(
@@ -1001,176 +1003,182 @@ def test_status_reports_unconfigured_assisted_runtime_truthfully() raises:
def test_status_reports_non_2xx_max_local_health_truthfully() raises:
with SafeTempDir() as temp_dir:
- var provider_stub = spawn_max_local_stub(0, "health_non_2xx", 1)
- var provider_port = provider_stub.port
- var startup_config_path = Path(temp_dir) / "explicit-hyf-config.toml"
- startup_config_path.write_text(
- _max_local_runtime_config_toml_with_urls(
- "http://127.0.0.1:" + String(provider_port) + "/v1",
- "http://127.0.0.1:" + String(provider_port) + "/health",
- 15000,
+ with spawn_max_local_stub(0, "health_non_2xx", 1) as provider_stub:
+ var provider_port = provider_stub.port
+ var startup_config_path = (
+ Path(temp_dir) / "explicit-hyf-config.toml"
)
- )
- with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
- with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
- var response = run_stdio_entrypoint(
- "src/main.mojo",
- load_scenario_request_json("scenarios/status_ok.json"),
- "--config",
- startup_config_path.__fspath__(),
+ startup_config_path.write_text(
+ _max_local_runtime_config_toml_with_urls(
+ "http://127.0.0.1:" + String(provider_port) + "/v1",
+ "http://127.0.0.1:" + String(provider_port) + "/health",
+ 15000,
)
+ )
+ with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
+ with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
+ var response = run_stdio_entrypoint(
+ "src/main.mojo",
+ load_scenario_request_json("scenarios/status_ok.json"),
+ "--config",
+ startup_config_path.__fspath__(),
+ )
- assert_true(response["ok"].bool_value())
- assert_equal(
- response["output"]["assisted_runtime"][
- "state"
- ].string_value(),
- "unavailable",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "reason"
- ].string_value(),
- "non_2xx",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "reachable"
- ].bool_value(),
- False,
- )
+ assert_true(response["ok"].bool_value())
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "state"
+ ].string_value(),
+ "unavailable",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "reason"
+ ].string_value(),
+ "non_2xx",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "reachable"
+ ].bool_value(),
+ False,
+ )
- provider_stub.wait()
+ provider_stub.wait()
def test_status_reports_ready_max_local_provider_truthfully() raises:
with SafeTempDir() as temp_dir:
- var provider_stub = spawn_max_local_stub(0, "query_rewrite_ok", 1)
- var provider_port = provider_stub.port
- var startup_config_path = Path(temp_dir) / "explicit-hyf-config.toml"
- startup_config_path.write_text(
- _max_local_runtime_config_toml_with_urls(
- "http://127.0.0.1:" + String(provider_port) + "/v1",
- "http://127.0.0.1:" + String(provider_port) + "/health",
- 15000,
+ with spawn_max_local_stub(0, "query_rewrite_ok", 1) as provider_stub:
+ var provider_port = provider_stub.port
+ var startup_config_path = (
+ Path(temp_dir) / "explicit-hyf-config.toml"
)
- )
- with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
- with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
- var response = run_stdio_entrypoint(
- "src/main.mojo",
- load_scenario_request_json("scenarios/status_ok.json"),
- "--config",
- startup_config_path.__fspath__(),
+ startup_config_path.write_text(
+ _max_local_runtime_config_toml_with_urls(
+ "http://127.0.0.1:" + String(provider_port) + "/v1",
+ "http://127.0.0.1:" + String(provider_port) + "/health",
+ 15000,
)
+ )
+ with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
+ with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
+ var response = run_stdio_entrypoint(
+ "src/main.mojo",
+ load_scenario_request_json("scenarios/status_ok.json"),
+ "--config",
+ startup_config_path.__fspath__(),
+ )
- assert_true(response["ok"].bool_value())
- assert_equal(
- response["output"]["execution_mode_request_behavior"][
- "assisted"
- ].string_value(),
- "execute",
- )
- assert_equal(
- response["output"]["backend_reachability"][
- "assisted_backend"
- ].string_value(),
- "ready",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "state"
- ].string_value(),
- "ready",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "reason"
- ].string_value(),
- "ready",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "transport"
- ].string_value(),
- "http",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "backend_kind"
- ].string_value(),
- "max_local",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "provider"
- ].string_value(),
- "max_local",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "route"
- ].string_value(),
- "provider_runtime.query_rewrite.max_local",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "model"
- ].string_value(),
- "max-local-query-rewrite",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "reachable"
- ].bool_value(),
- True,
- )
+ assert_true(response["ok"].bool_value())
+ assert_equal(
+ response["output"]["execution_mode_request_behavior"][
+ "assisted"
+ ].string_value(),
+ "execute",
+ )
+ assert_equal(
+ response["output"]["backend_reachability"][
+ "assisted_backend"
+ ].string_value(),
+ "ready",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "state"
+ ].string_value(),
+ "ready",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "reason"
+ ].string_value(),
+ "ready",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "transport"
+ ].string_value(),
+ "http",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "backend_kind"
+ ].string_value(),
+ "max_local",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "provider"
+ ].string_value(),
+ "max_local",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "route"
+ ].string_value(),
+ "provider_runtime.query_rewrite.max_local",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "model"
+ ].string_value(),
+ "max-local-query-rewrite",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "reachable"
+ ].bool_value(),
+ True,
+ )
- provider_stub.wait()
+ provider_stub.wait()
def test_status_bounds_max_local_health_probe_timeout() raises:
with SafeTempDir() as temp_dir:
- var provider_stub = spawn_max_local_stub(0, "health_timeout", 1)
- var provider_port = provider_stub.port
- var startup_config_path = Path(temp_dir) / "explicit-hyf-config.toml"
- startup_config_path.write_text(
- _max_local_runtime_config_toml_with_urls(
- "http://127.0.0.1:" + String(provider_port) + "/v1",
- "http://127.0.0.1:" + String(provider_port) + "/health",
- 15000,
+ with spawn_max_local_stub(0, "health_timeout", 1) as provider_stub:
+ var provider_port = provider_stub.port
+ var startup_config_path = (
+ Path(temp_dir) / "explicit-hyf-config.toml"
)
- )
- with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
- with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
- var response = run_stdio_entrypoint(
- "src/main.mojo",
- load_scenario_request_json("scenarios/status_ok.json"),
- "--config",
- startup_config_path.__fspath__(),
+ startup_config_path.write_text(
+ _max_local_runtime_config_toml_with_urls(
+ "http://127.0.0.1:" + String(provider_port) + "/v1",
+ "http://127.0.0.1:" + String(provider_port) + "/health",
+ 15000,
)
+ )
+ with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
+ with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
+ var response = run_stdio_entrypoint(
+ "src/main.mojo",
+ load_scenario_request_json("scenarios/status_ok.json"),
+ "--config",
+ startup_config_path.__fspath__(),
+ )
- assert_true(response["ok"].bool_value())
- assert_equal(
- response["output"]["assisted_runtime"][
- "state"
- ].string_value(),
- "unavailable",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "reason"
- ].string_value(),
- "timeout",
- )
- assert_equal(
- response["output"]["assisted_runtime"][
- "reachable"
- ].bool_value(),
- False,
- )
+ assert_true(response["ok"].bool_value())
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "state"
+ ].string_value(),
+ "unavailable",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "reason"
+ ].string_value(),
+ "timeout",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime"][
+ "reachable"
+ ].bool_value(),
+ False,
+ )
- provider_stub.wait()
+ provider_stub.wait()
def test_status_rejects_invalid_max_local_runtime_config() raises:
@@ -1509,118 +1517,122 @@ def test_capabilities_reports_configured_provider_runtime_truthfully() raises:
def test_capabilities_reports_ready_max_local_provider_truthfully() raises:
with SafeTempDir() as temp_dir:
- var provider_stub = spawn_max_local_stub(0, "query_rewrite_ok", 1)
- var provider_port = provider_stub.port
- var startup_config_path = Path(temp_dir) / "explicit-hyf-config.toml"
- startup_config_path.write_text(
- _max_local_runtime_config_toml_with_urls(
- "http://127.0.0.1:" + String(provider_port) + "/v1",
- "http://127.0.0.1:" + String(provider_port) + "/health",
- 15000,
+ with spawn_max_local_stub(0, "query_rewrite_ok", 1) as provider_stub:
+ var provider_port = provider_stub.port
+ var startup_config_path = (
+ Path(temp_dir) / "explicit-hyf-config.toml"
)
- )
- with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
- with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
- var response = run_stdio_entrypoint(
- "src/main.mojo",
- load_scenario_request_json(
- "scenarios/capabilities_ok.json"
- ),
- "--config",
- startup_config_path.__fspath__(),
+ startup_config_path.write_text(
+ _max_local_runtime_config_toml_with_urls(
+ "http://127.0.0.1:" + String(provider_port) + "/v1",
+ "http://127.0.0.1:" + String(provider_port) + "/health",
+ 15000,
)
+ )
+ with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
+ with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
+ var response = run_stdio_entrypoint(
+ "src/main.mojo",
+ load_scenario_request_json(
+ "scenarios/capabilities_ok.json"
+ ),
+ "--config",
+ startup_config_path.__fspath__(),
+ )
- assert_true(response["ok"].bool_value())
- assert_equal(
- response["output"]["business_capabilities"][0][
- "assisted_execution"
- ].string_value(),
- "ready",
- )
- assert_equal(
- response["output"]["business_capabilities"][0][
- "assisted_backend_available"
- ].bool_value(),
- True,
- )
- assert_equal(
- response["output"]["business_capabilities"][2][
- "assisted_execution"
- ].string_value(),
- "unsupported_capability",
- )
- assert_equal(
- response["output"]["assisted_runtime_capabilities"][0][
- "state"
- ].string_value(),
- "ready",
- )
- assert_equal(
- response["output"]["assisted_runtime_capabilities"][0][
- "reason"
- ].string_value(),
- "ready",
- )
- assert_equal(
- response["output"]["assisted_runtime_capabilities"][0][
- "backend_kind"
- ].string_value(),
- "max_local",
- )
+ assert_true(response["ok"].bool_value())
+ assert_equal(
+ response["output"]["business_capabilities"][0][
+ "assisted_execution"
+ ].string_value(),
+ "ready",
+ )
+ assert_equal(
+ response["output"]["business_capabilities"][0][
+ "assisted_backend_available"
+ ].bool_value(),
+ True,
+ )
+ assert_equal(
+ response["output"]["business_capabilities"][2][
+ "assisted_execution"
+ ].string_value(),
+ "unsupported_capability",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime_capabilities"][0][
+ "state"
+ ].string_value(),
+ "ready",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime_capabilities"][0][
+ "reason"
+ ].string_value(),
+ "ready",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime_capabilities"][0][
+ "backend_kind"
+ ].string_value(),
+ "max_local",
+ )
- provider_stub.wait()
+ provider_stub.wait()
def test_capabilities_bounds_max_local_health_probe_timeout() raises:
with SafeTempDir() as temp_dir:
- var provider_stub = spawn_max_local_stub(0, "health_timeout", 1)
- var provider_port = provider_stub.port
- var startup_config_path = Path(temp_dir) / "explicit-hyf-config.toml"
- startup_config_path.write_text(
- _max_local_runtime_config_toml_with_urls(
- "http://127.0.0.1:" + String(provider_port) + "/v1",
- "http://127.0.0.1:" + String(provider_port) + "/health",
- 15000,
+ with spawn_max_local_stub(0, "health_timeout", 1) as provider_stub:
+ var provider_port = provider_stub.port
+ var startup_config_path = (
+ Path(temp_dir) / "explicit-hyf-config.toml"
)
- )
- with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
- with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
- var response = run_stdio_entrypoint(
- "src/main.mojo",
- load_scenario_request_json(
- "scenarios/capabilities_ok.json"
- ),
- "--config",
- startup_config_path.__fspath__(),
+ startup_config_path.write_text(
+ _max_local_runtime_config_toml_with_urls(
+ "http://127.0.0.1:" + String(provider_port) + "/v1",
+ "http://127.0.0.1:" + String(provider_port) + "/health",
+ 15000,
)
+ )
+ with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
+ with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
+ var response = run_stdio_entrypoint(
+ "src/main.mojo",
+ load_scenario_request_json(
+ "scenarios/capabilities_ok.json"
+ ),
+ "--config",
+ startup_config_path.__fspath__(),
+ )
- assert_true(response["ok"].bool_value())
- assert_equal(
- response["output"]["business_capabilities"][0][
- "assisted_execution"
- ].string_value(),
- "unavailable",
- )
- assert_equal(
- response["output"]["business_capabilities"][0][
- "assisted_backend_available"
- ].bool_value(),
- False,
- )
- assert_equal(
- response["output"]["assisted_runtime_capabilities"][0][
- "state"
- ].string_value(),
- "unavailable",
- )
- assert_equal(
- response["output"]["assisted_runtime_capabilities"][0][
- "reason"
- ].string_value(),
- "timeout",
- )
+ assert_true(response["ok"].bool_value())
+ assert_equal(
+ response["output"]["business_capabilities"][0][
+ "assisted_execution"
+ ].string_value(),
+ "unavailable",
+ )
+ assert_equal(
+ response["output"]["business_capabilities"][0][
+ "assisted_backend_available"
+ ].bool_value(),
+ False,
+ )
+ assert_equal(
+ response["output"]["assisted_runtime_capabilities"][0][
+ "state"
+ ].string_value(),
+ "unavailable",
+ )
+ assert_equal(
+ response["output"]["assisted_runtime_capabilities"][0][
+ "reason"
+ ].string_value(),
+ "timeout",
+ )
- provider_stub.wait()
+ provider_stub.wait()
def test_query_rewrite_falls_back_deterministically_when_provider_is_unavailable() raises:
@@ -1748,85 +1760,87 @@ def test_assisted_semantic_rank_falls_back_as_unsupported_provider_capability()
def test_query_rewrite_uses_max_local_provider_when_ready() raises:
with SafeTempDir() as temp_dir:
- var provider_stub = spawn_max_local_stub(0, "query_rewrite_ok", 2)
- var provider_port = provider_stub.port
- var startup_config_path = Path(temp_dir) / "explicit-hyf-config.toml"
- startup_config_path.write_text(
- _max_local_runtime_config_toml_with_urls(
- "http://127.0.0.1:" + String(provider_port) + "/v1",
- "http://127.0.0.1:" + String(provider_port) + "/health",
- 15000,
+ with spawn_max_local_stub(0, "query_rewrite_ok", 2) as provider_stub:
+ var provider_port = provider_stub.port
+ var startup_config_path = (
+ Path(temp_dir) / "explicit-hyf-config.toml"
)
- )
- with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
- with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
- var response = run_stdio_entrypoint(
- "src/main.mojo",
- (
- '{"version":1,"request_id":"rewrite-assisted-max-local-1","trace_id":"rewrite-assisted-max-local-1","capability":"query_rewrite","context":{"execution_mode_preference":"assisted","return_provenance":true},"input":{"query":"local'
- ' apples pickup weekend"}}'
- ),
- "--config",
- startup_config_path.__fspath__(),
+ startup_config_path.write_text(
+ _max_local_runtime_config_toml_with_urls(
+ "http://127.0.0.1:" + String(provider_port) + "/v1",
+ "http://127.0.0.1:" + String(provider_port) + "/health",
+ 15000,
)
+ )
+ with ScopedEnvVar(HYF_PATHS_PROFILE_ENV, "repo_local"):
+ with ScopedEnvVar(HYF_PATHS_REPO_LOCAL_ROOT_ENV, temp_dir):
+ var response = run_stdio_entrypoint(
+ "src/main.mojo",
+ (
+ '{"version":1,"request_id":"rewrite-assisted-max-local-1","trace_id":"rewrite-assisted-max-local-1","capability":"query_rewrite","context":{"execution_mode_preference":"assisted","return_provenance":true},"input":{"query":"local'
+ ' apples pickup weekend"}}'
+ ),
+ "--config",
+ startup_config_path.__fspath__(),
+ )
- assert_true(response["ok"].bool_value())
- assert_equal(
- response["meta"]["execution_mode"].string_value(),
- "assisted",
- )
- assert_equal(
- response["meta"]["backend"].string_value(),
- "provider_runtime",
- )
- assert_equal(
- response["meta"]["provider"].string_value(),
- "max_local",
- )
- assert_equal(
- response["meta"]["route"].string_value(),
- "provider_runtime.query_rewrite.max_local",
- )
- assert_equal(
- response["meta"]["model"].string_value(),
- "max-local-query-rewrite",
- )
- assert_true(
- Int(response["meta"]["latency_ms"].int_value()) >= 0
- )
- assert_equal(
- Int(response["meta"]["schema_version"].int_value()), 1
- )
- assert_equal(
- response["meta"]["prompt_version"].string_value(),
- "max_local_query_rewrite_v1",
- )
- assert_equal(
- response["meta"]["provenance"]["kind"].string_value(),
- "assisted",
- )
- assert_true(
- response["meta"]["provenance"]["fallback"].is_null()
- )
- _assert_no_top_level_fallback_meta(response)
- assert_equal(
- response["output"]["rewritten_text"].string_value(),
- "apples pickup weekend",
- )
- assert_equal(
- response["output"]["query_terms"][0].string_value(),
- "apples",
- )
- assert_equal(
- response["output"]["query_terms"][1].string_value(),
- "pickup",
- )
- assert_equal(
- response["output"]["query_terms"][2].string_value(),
- "weekend",
- )
+ assert_true(response["ok"].bool_value())
+ assert_equal(
+ response["meta"]["execution_mode"].string_value(),
+ "assisted",
+ )
+ assert_equal(
+ response["meta"]["backend"].string_value(),
+ "provider_runtime",
+ )
+ assert_equal(
+ response["meta"]["provider"].string_value(),
+ "max_local",
+ )
+ assert_equal(
+ response["meta"]["route"].string_value(),
+ "provider_runtime.query_rewrite.max_local",
+ )
+ assert_equal(
+ response["meta"]["model"].string_value(),
+ "max-local-query-rewrite",
+ )
+ assert_true(
+ Int(response["meta"]["latency_ms"].int_value()) >= 0
+ )
+ assert_equal(
+ Int(response["meta"]["schema_version"].int_value()), 1
+ )
+ assert_equal(
+ response["meta"]["prompt_version"].string_value(),
+ "max_local_query_rewrite_v1",
+ )
+ assert_equal(
+ response["meta"]["provenance"]["kind"].string_value(),
+ "assisted",
+ )
+ assert_true(
+ response["meta"]["provenance"]["fallback"].is_null()
+ )
+ _assert_no_top_level_fallback_meta(response)
+ assert_equal(
+ response["output"]["rewritten_text"].string_value(),
+ "apples pickup weekend",
+ )
+ assert_equal(
+ response["output"]["query_terms"][0].string_value(),
+ "apples",
+ )
+ assert_equal(
+ response["output"]["query_terms"][1].string_value(),
+ "pickup",
+ )
+ assert_equal(
+ response["output"]["query_terms"][2].string_value(),
+ "weekend",
+ )
- provider_stub.wait()
+ provider_stub.wait()
def test_query_rewrite_falls_back_on_provider_non_2xx() raises:
