From 697a953d4dca9b37366ae338ffb143e2959b387f Mon Sep 17 00:00:00 2001 From: ethernet Date: Tue, 22 Sep 2026 01:54:54 -0400 Subject: [PATCH] test(tui): synchronize compute-host protocol frames Compute-host protocol tests polled a thread-unsafe StringIO on a five-second wall-clock deadline while unrelated provider discovery and persistence work competed under CI load. Isolate those side paths and wait causally for complete JSON-line frames so strict turn ordering remains observable without scheduling races. --- tests/tui_gateway/_compute_host_frames.py | 59 ++++++++++++++ .../test_compute_host_borrowed_lease.py | 34 ++++----- .../test_compute_host_turn_protocol.py | 76 ++++++++++--------- 3 files changed, 113 insertions(+), 56 deletions(-) create mode 100644 tests/tui_gateway/_compute_host_frames.py diff --git a/tests/tui_gateway/_compute_host_frames.py b/tests/tui_gateway/_compute_host_frames.py new file mode 100644 index 0000000000..fce2744570 --- /dev/null +++ b/tests/tui_gateway/_compute_host_frames.py @@ -0,0 +1,59 @@ +"""Causal, thread-safe compute-host frame capture for protocol tests.""" + +from __future__ import annotations + +import json +import threading +import time +from collections.abc import Callable + + +class FrameSink: + """Capture complete JSON-line frames and wake waiters when one is published.""" + + def __init__(self) -> None: + self._condition = threading.Condition() + self._pending = "" + self._frames: list[dict] = [] + + def write(self, data: str) -> int: + with self._condition: + self._pending += data + published = False + while "\n" in self._pending: + line, self._pending = self._pending.split("\n", 1) + if line.strip(): + self._frames.append(json.loads(line)) + published = True + if published: + self._condition.notify_all() + return len(data) + + def flush(self) -> None: + pass + + def frames(self) -> list[dict]: + with self._condition: + return list(self._frames) + + def wait_for(self, predicate: Callable[[dict], bool], timeout: float = 20.0) -> dict: + """Wait on frame publication; ``timeout`` is only a deadlock guard.""" + deadline = time.monotonic() + timeout + with self._condition: + while True: + for frame in self._frames: + if predicate(frame): + return frame + remaining = deadline - time.monotonic() + if remaining <= 0: + raise AssertionError(f"timed out; saw={self._frames}") + self._condition.wait(remaining) + + +def start_test_work(target, *, name: str, session: dict | None = None) -> threading.Thread: + """Start the real turn thread without process-retirement/import side paths.""" + thread = threading.Thread(target=target, name=name) + if session is not None: + session["_run_thread"] = thread + thread.start() + return thread diff --git a/tests/tui_gateway/test_compute_host_borrowed_lease.py b/tests/tui_gateway/test_compute_host_borrowed_lease.py index 7b643fa6c0..1e3ab8efcd 100644 --- a/tests/tui_gateway/test_compute_host_borrowed_lease.py +++ b/tests/tui_gateway/test_compute_host_borrowed_lease.py @@ -14,33 +14,21 @@ rotation and held past ``session.close`` until the child's turn settles. from __future__ import annotations -import io +import contextlib import json import os +import sys import threading import time import types import pytest +from tests.tui_gateway._compute_host_frames import FrameSink, start_test_work from tui_gateway import server from tui_gateway.compute_host import ComputeHost -def _frames(out: io.StringIO) -> list[dict]: - return [json.loads(line) for line in out.getvalue().splitlines() if line.strip()] - - -def _wait(out: io.StringIO, predicate, timeout: float = 5.0) -> dict: - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - for frame in _frames(out): - if predicate(frame): - return frame - time.sleep(0.01) - raise AssertionError(f"timed out; saw={_frames(out)}") - - def _stub_agent(deltas: list[str]) -> types.SimpleNamespace: def run_conversation(prompt, *, conversation_history=None, stream_callback=None, **_kw): final = "".join(deltas) @@ -108,11 +96,19 @@ def isolated_env(monkeypatch, tmp_path): monkeypatch.setattr(server, "_session_cwd", lambda session: str(tmp_path)) monkeypatch.setattr(server, "_register_session_cwd", lambda session: None) monkeypatch.setattr(server, "_tts_stream_begin", lambda: None) + monkeypatch.setattr(server, "_ensure_session_db_row", lambda session: True) + monkeypatch.setattr(server, "_persist_branch_seed", lambda session: None) + monkeypatch.setattr(server, "_routing_provenance_db", lambda session: contextlib.nullcontext(None)) + monkeypatch.setattr(server, "_reopen_routed_session_row", lambda *a, **k: None) + monkeypatch.setattr(server, "_record_turn_marker", lambda *a, **k: "marker") + monkeypatch.setattr(server, "_retire_turn_marker", lambda *a, **k: None) + monkeypatch.setattr(server, "_start_session_work", start_test_work) monkeypatch.setattr(server, "_get_usage", lambda agent_: {}) monkeypatch.setattr(server, "_hydrate_session_cwd", lambda *a, **k: None) monkeypatch.setattr(server, "_wire_session_agent", lambda *a, **k: None) monkeypatch.setattr(server, "_start_session_services", lambda *a, **k: None) monkeypatch.setattr(server, "_schedule_mcp_late_refresh", lambda *a, **k: None) + monkeypatch.setitem(sys.modules, "hermes_undo", types.SimpleNamespace(on_user_message_appended=lambda key: None)) import tui_gateway.prompt_turn as prompt_turn for mod in (server, prompt_turn): @@ -123,16 +119,16 @@ def isolated_env(monkeypatch, tmp_path): server._sessions.pop(sid, None) -def _run_turn(frame: dict, timeout: float = 5.0) -> tuple[list[dict], dict | None]: +def _run_turn(frame: dict, timeout: float = 20.0) -> tuple[list[dict], dict | None]: """Run one turn.start through the real child path; return (all frames, turn.end frame).""" - out = io.StringIO() + out = FrameSink() host = ComputeHost(stdout=out, heartbeat_secs=0) try: host.handle_frame(frame) - end = _wait(out, lambda f: f["type"] == "turn.end", timeout=timeout) + end = out.wait_for(lambda f: f["type"] == "turn.end", timeout=timeout) finally: host.close() - return _frames(out), end + return out.frames(), end # ── Fix 1: the child borrows instead of re-claiming ───────────────────────── diff --git a/tests/tui_gateway/test_compute_host_turn_protocol.py b/tests/tui_gateway/test_compute_host_turn_protocol.py index 92e85bbd5b..98fe456204 100644 --- a/tests/tui_gateway/test_compute_host_turn_protocol.py +++ b/tests/tui_gateway/test_compute_host_turn_protocol.py @@ -9,44 +9,46 @@ a ``server._sessions`` entry whose agent runs on the turn worker. from __future__ import annotations -import io -import json +import contextlib +import sys import threading import time import types import pytest +from tests.tui_gateway._compute_host_frames import FrameSink, start_test_work from tui_gateway import server from tui_gateway.compute_host import ComputeHost -def _frames(out: io.StringIO) -> list[dict]: - return [json.loads(line) for line in out.getvalue().splitlines() if line.strip()] - - -def _wait(out: io.StringIO, predicate, timeout: float = 5.0) -> dict: - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - for frame in _frames(out): - if predicate(frame): - return frame - time.sleep(0.01) - raise AssertionError(f"timed out; saw={_frames(out)}") - - @pytest.fixture() def turn_env(monkeypatch, tmp_path): """Neutralize the turn pipeline's environment-heavy side paths (same set as the prompt.submit tests) so the frame protocol is what's under test. Threads stay REAL: ``_run_real_turn`` joins ``session["_run_thread"]`` itself before emitting ``turn.end``.""" monkeypatch.setattr(server, "_wire_callbacks", lambda sid: None) + monkeypatch.setattr(server, "_apply_pending_model_switch", lambda *a, **k: None) monkeypatch.setattr(server, "_sync_agent_model_with_config", lambda sid, session: None) + monkeypatch.setattr(server, "_sync_agent_compression_with_config", lambda *a, **k: None) + monkeypatch.setattr(server, "_sync_agent_fallback_with_config", lambda *a, **k: None) + monkeypatch.setattr(server, "_sync_bot_capabilities", lambda *a, **k: None) + monkeypatch.setattr(server, "_adopt_out_of_band_turns", lambda *a, **k: None) monkeypatch.setattr(server, "_session_cwd", lambda session: str(tmp_path)) monkeypatch.setattr(server, "_register_session_cwd", lambda session: None) monkeypatch.setattr(server, "_tts_stream_begin", lambda: None) + monkeypatch.setattr(server, "_ensure_session_db_row", lambda session: True) + monkeypatch.setattr(server, "_persist_branch_seed", lambda session: None) + monkeypatch.setattr(server, "_routing_provenance_db", lambda session: contextlib.nullcontext(None)) + monkeypatch.setattr(server, "_reopen_routed_session_row", lambda *a, **k: None) + monkeypatch.setattr(server, "_record_turn_marker", lambda *a, **k: "marker") + monkeypatch.setattr(server, "_retire_turn_marker", lambda *a, **k: None) + monkeypatch.setattr(server, "_start_session_work", start_test_work) monkeypatch.setattr(server, "_sync_session_key_after_compress", lambda *a, **k: None) monkeypatch.setattr(server, "_get_usage", lambda agent: {}) + monkeypatch.setattr(server, "_start_turn_voice", lambda: (None, False)) + monkeypatch.setattr(server, "_start_usage_ticker", lambda *a, **k: (threading.Event(), start_test_work(lambda: None, name="usage"))) + monkeypatch.setitem(sys.modules, "hermes_undo", types.SimpleNamespace(on_user_message_appended=lambda key: None)) def _agent(deltas: list[str], *, delay_s: float = 0.0, interrupt: threading.Event | None = None): @@ -80,18 +82,18 @@ def _session(agent) -> dict: def test_turn_start_streams_deltas_then_turn_end_with_history_identity(turn_env): - out = io.StringIO() + out = FrameSink() host = ComputeHost(stdout=out, heartbeat_secs=0) sid = "s1" server._sessions[sid] = _session(_agent(["a ", "b ", "c "])) try: host.handle_frame({"type": "turn.start", "sid": sid, "request_id": "turn", "prompt": "hello"}) - end = _wait(out, lambda f: f["type"] == "turn.end") + end = out.wait_for(lambda f: f["type"] == "turn.end") finally: server._sessions.pop(sid, None) host.close() - frames = _frames(out) + frames = out.frames() kinds = [f["type"] for f in frames] assert kinds[0] == "turn.started" assert kinds[-1] == "turn.end" @@ -117,18 +119,18 @@ def test_turn_start_streams_deltas_then_turn_end_with_history_identity(turn_env) def test_turn_start_without_sid_is_a_turn_error(turn_env): - out = io.StringIO() + out = FrameSink() host = ComputeHost(stdout=out, heartbeat_secs=0) try: host.handle_frame({"type": "turn.start", "request_id": "nosid", "prompt": "x"}) - err = _wait(out, lambda f: f["type"] == "turn.error") + err = out.wait_for(lambda f: f["type"] == "turn.error") finally: host.close() assert err["request_id"] == "nosid" and err["message"] == "sid required" def test_second_turn_start_while_running_is_session_busy(turn_env): - out = io.StringIO() + out = FrameSink() host = ComputeHost(stdout=out, heartbeat_secs=0) sid = "s1" session = _session(_agent(["x"])) @@ -136,7 +138,7 @@ def test_second_turn_start_while_running_is_session_busy(turn_env): server._sessions[sid] = session try: host.handle_frame({"type": "turn.start", "sid": sid, "request_id": "t2", "prompt": "hi"}) - err = _wait(out, lambda f: f["type"] == "turn.error") + err = out.wait_for(lambda f: f["type"] == "turn.error") finally: server._sessions.pop(sid, None) host.close() @@ -145,7 +147,7 @@ def test_second_turn_start_while_running_is_session_busy(turn_env): def test_stale_queued_prompt_generation_ends_turn_as_interrupted(turn_env): """A queued prompt whose generation was bumped by an interrupt must not run.""" - out = io.StringIO() + out = FrameSink() host = ComputeHost(stdout=out, heartbeat_secs=0) sid = "s1" session = _session(_agent(["x"])) @@ -154,27 +156,27 @@ def test_stale_queued_prompt_generation_ends_turn_as_interrupted(turn_env): try: host.handle_frame({"type": "turn.start", "sid": sid, "request_id": "q", "prompt": "hi", "queued_prompt_generation": 2}) - end = _wait(out, lambda f: f["type"] == "turn.end") + end = out.wait_for(lambda f: f["type"] == "turn.end") finally: server._sessions.pop(sid, None) host.close() assert end["interrupted"] is True and end["request_id"] == "q" - assert not any(f["type"] == "turn.started" for f in _frames(out)) + assert not any(f["type"] == "turn.started" for f in out.frames()) def test_interrupt_frame_acks_and_marks_turn_interrupted(turn_env): """The turn runs on the host worker while ``interrupt`` arrives on the control path.""" - out = io.StringIO() + out = FrameSink() host = ComputeHost(stdout=out, heartbeat_secs=0) sid = "s1" stop = threading.Event() server._sessions[sid] = _session(_agent([f"{i:03d} " for i in range(200)], delay_s=0.01, interrupt=stop)) try: host.handle_frame({"type": "turn.start", "sid": sid, "request_id": "turn", "prompt": "go"}) - _wait(out, lambda f: f["type"] == "rpc" and f["message"]["params"]["type"] == "message.delta") + out.wait_for(lambda f: f["type"] == "rpc" and f["message"]["params"]["type"] == "message.delta") host.handle_frame({"type": "interrupt", "sid": sid, "request_id": "stop"}) - ack = _wait(out, lambda f: f["type"] == "interrupt.ack") - end = _wait(out, lambda f: f["type"] == "turn.end") + ack = out.wait_for(lambda f: f["type"] == "interrupt.ack") + end = out.wait_for(lambda f: f["type"] == "turn.end") finally: stop.set() server._sessions.pop(sid, None) @@ -182,19 +184,19 @@ def test_interrupt_frame_acks_and_marks_turn_interrupted(turn_env): assert ack["applied"] is True and ack["request_id"] == "stop" and "applied_ns" in ack assert end["interrupted"] is True - deltas = sum(1 for f in _frames(out) if f["type"] == "rpc" and f["message"]["params"]["type"] == "message.delta") + deltas = sum(1 for f in out.frames() if f["type"] == "rpc" and f["message"]["params"]["type"] == "message.delta") assert 0 < deltas < 200 def test_unknown_frame_type_is_an_error(): - out = io.StringIO() + out = FrameSink() host = ComputeHost(stdout=out, heartbeat_secs=0) try: host.handle_frame({"type": "bogus", "request_id": "b"}) finally: host.close() - assert _frames(out) == [{"type": "error", "request_id": "b", "message": "unknown frame type: bogus", - "host_ns": _frames(out)[0]["host_ns"]}] + assert out.frames() == [{"type": "error", "request_id": "b", "message": "unknown frame type: bogus", + "host_ns": out.frames()[0]["host_ns"]}] @pytest.mark.parametrize("kind", ["legacy", "hard-only", "dynamic-getattr"]) @@ -224,7 +226,7 @@ def test_compute_host_interrupt_uses_explicit_stop_compatibility(monkeypatch, ki agent = {"legacy": _Legacy(), "hard-only": _HardOnly(), "dynamic-getattr": _Dynamic()}[kind] # The child never routes back to a supervisor (HERMES_COMPUTE_HOST_CHILD=1 in production). monkeypatch.setenv("HERMES_COMPUTE_HOST_CHILD", "1") - out = io.StringIO() + out = FrameSink() host = ComputeHost(stdout=out, heartbeat_secs=0) sid = "s1" session = _session(agent) @@ -237,7 +239,7 @@ def test_compute_host_interrupt_uses_explicit_stop_compatibility(monkeypatch, ki host.close() assert calls == ["hard" if kind == "hard-only" else "legacy"] - ack = _frames(out)[-1] + ack = out.frames()[-1] assert ack["type"] == "interrupt.ack" and ack["applied"] is True assert session["_turn_cancel_requested"] is True @@ -245,7 +247,7 @@ def test_compute_host_interrupt_uses_explicit_stop_compatibility(monkeypatch, ki def test_host_builds_the_session_agent_with_the_frame_login(monkeypatch): """The host process has no record for a first turn, and its pipe names no login, so the agent is built from the login the frame carries and the new record keeps it for later rebuilds.""" - host = ComputeHost(stdout=io.StringIO(), heartbeat_secs=0) + host = ComputeHost(stdout=FrameSink(), heartbeat_secs=0) captured = {} def fake_make_agent(sid, key, **kwargs):