From 77915e344cb0cd8e20661d4a7b393f987a2eef32 Mon Sep 17 00:00:00 2001 From: "Mark S." <274530371+unsupportedpastels@users.noreply.github.com> Date: Sun, 6 Sep 2026 11:54:01 +0000 Subject: [PATCH] fix(tui-gateway): preserve active isolated turns after disconnect --- .../test_isolated_orphan_activity.py | 207 ++++++++++++++++++ tui_gateway/compute_host.py | 24 +- tui_gateway/compute_host_bridge.py | 24 ++ tui_gateway/session_lifecycle.py | 7 + 4 files changed, 261 insertions(+), 1 deletion(-) create mode 100644 tests/tui_gateway/test_isolated_orphan_activity.py diff --git a/tests/tui_gateway/test_isolated_orphan_activity.py b/tests/tui_gateway/test_isolated_orphan_activity.py new file mode 100644 index 0000000000..fa130e011b --- /dev/null +++ b/tests/tui_gateway/test_isolated_orphan_activity.py @@ -0,0 +1,207 @@ +"""Detached Desktop/TUI turns use child-owned activity, not process heartbeats.""" + +from pathlib import Path +import sys +import threading +import time + +import pytest + +from tui_gateway import server +from tui_gateway.host_supervisor import HostSupervisor + + +class _Timer: + def __init__(self, delay, callback): + self.delay, self.callback = delay, callback + + def start(self): + pass + + def cancel(self): + pass + + +def _session(sid): + return dict(agent=None, agent_ready=threading.Event(), session_key=sid, + history=[], history_version=0, history_lock=threading.Lock(), + running=True, transport=server._detached_ws_transport, + attached_images=[], cols=80, source="desktop", inflight_turn=None) + + +@pytest.mark.parametrize("mode", ["fresh", "stale", "missing", "previous"]) +def test_real_child_detached_turn_activity(tmp_path, monkeypatch, mode): + """Real supervisor pipes, child admission/turn thread, bridge and orphan timer. + + Only the agent/provider and environment-heavy UI side effects are stubbed in + the child. Its activity writer and snapshot contract are the production ones. + """ + sid = "detached-turn" + session = _session(sid) + forwarded = [] + monkeypatch.setattr(server, "_sessions", {sid: session}) + monkeypatch.setattr(server, "_pending_ws_reaps", {}) + monkeypatch.setattr(server, "write_json", lambda msg: forwarded.append(msg) or True) + monkeypatch.setattr(server, "_load_dashboard_process_isolation_config", lambda: {"turn_isolation": True}) + monkeypatch.setattr(server, "_WS_ORPHAN_ACTIVITY_STALE_S", 30.0) + monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 20.0) + monkeypatch.setattr(server, "_session_has_active_delegations", lambda *args: False) + monkeypatch.setattr(server, "_session_cwd", lambda s: str(tmp_path)) + home = tmp_path / "home" + home.mkdir() + supervisor = HostSupervisor( + argv=[sys.executable, str(Path(__file__).resolve()), mode, str(tmp_path)], + registry_path=tmp_path / "host.json", env={"HERMES_HOME": str(home)}, + expected_hermes_home=str(home), rpc_sink=server._relay_compute_host_rpc, + heartbeat_secs=1, autostart=False) + monkeypatch.setattr(server, "_get_compute_host_supervisor", lambda *args: supervisor) + try: + response = server._submit_prompt_to_compute_host("request", sid, session, "work") + assert response["result"]["turn_isolation"] is True + deadline = time.monotonic() + 12 + while not (tmp_path / "provider-started").exists() and time.monotonic() < deadline: + time.sleep(0.02) + assert (tmp_path / "provider-started").exists(), supervisor._stderr_tail + # Give the actual child-to-parent sampler a bounded opportunity to arrive. + deadline = time.monotonic() + 3 + while not server._ws_orphan_turn_activity_is_fresh(session) and time.monotonic() < deadline: + time.sleep(0.02) + assert supervisor.is_running() + assert session["agent"] is None + assert server._ws_orphan_turn_activity_is_fresh(session) is (mode == "fresh") + monkeypatch.setattr(server.threading, "Timer", _Timer) + server._schedule_ws_orphan_reap(sid) + server._pending_ws_reaps[sid].callback() + assert bool(session.get("_client_gone_interrupt_requested")) is (mode != "fresh") + assert server._pending_ws_reaps[sid].delay == ( + 20.0 if mode == "fresh" else server._WS_ORPHAN_INTERRUPT_REAP_POLL_S) + assert not any(m.get("method") == "compute_host.activity" for m in forwarded) + if mode != "fresh": + deadline = time.monotonic() + 5 + while session["running"] and time.monotonic() < deadline: + time.sleep(0.02) + assert not session["running"], "stale child must receive and settle the real interrupt" + if mode == "fresh": + old_token = session["_compute_host_turn_id"] + old_request = next(iter(supervisor._pending_turns)) + (tmp_path / "release").touch() + deadline = time.monotonic() + 5 + while session["running"] and time.monotonic() < deadline: + time.sleep(0.02) + assert not session["running"] + assert "_compute_host_activity_ns" not in session + (tmp_path / "release").unlink() + (tmp_path / "provider-started").unlink() + session["running"] = True + # Same sid and caller rid, same child/agent, but NO new activity. + server._submit_prompt_to_compute_host("request", sid, session, "next") + assert session["_compute_host_turn_id"] != old_token + new_token = session["_compute_host_turn_id"] + # A delayed terminal frame cannot resolve the new caller-rid reuse. + supervisor._handle_host_frame({"type": "turn.end", "sid": sid, "request_id": old_request}) + assert session["running"] + assert session["_compute_host_turn_id"] == new_token + deadline = time.monotonic() + 5 + while not (tmp_path / "provider-started").exists() and time.monotonic() < deadline: + time.sleep(0.02) + assert (tmp_path / "provider-started").exists() + # Also replay a delayed sample from the previous dispatch. + server._relay_compute_host_rpc({"method": "compute_host.activity", "params": { + "session_id": sid, "turn_id": old_token, "activity_ns": time.perf_counter_ns()}}) + deadline = time.monotonic() + 3 + while "_compute_host_activity_ns" not in session and time.monotonic() < deadline: + time.sleep(0.02) + assert "_compute_host_activity_ns" in session + assert not server._ws_orphan_turn_activity_is_fresh(session) + server._pending_ws_reaps[sid].callback() + assert session["_client_gone_interrupt_requested"] + finally: + supervisor.shutdown() + + +@pytest.mark.parametrize("change", ["none", "other-session", "old-turn", "not-running", "stale", "missing"]) +def test_activity_relay_is_fenced_and_ages(monkeypatch, change): + session = _session("session") + session.update(_compute_host_active=True, _compute_host_turn_id="new-turn") + monkeypatch.setattr(server, "_sessions", {"session": session}) + monkeypatch.setattr(server, "_WS_ORPHAN_ACTIVITY_STALE_S", 30) + monkeypatch.setattr(server, "write_json", lambda msg: pytest.fail("internal activity leaked to client")) + params: dict = dict(session_id="session", turn_id="new-turn", activity_ns=time.perf_counter_ns()) + if change == "other-session": + params["session_id"] = "other" + elif change == "old-turn": + params["turn_id"] = "old-turn" + elif change == "not-running": + session["running"] = False + elif change == "stale": + params["activity_ns"] -= 31_000_000_000 + elif change == "missing": + params["activity_ns"] = None + server._relay_compute_host_rpc({"jsonrpc": "2.0", "method": "compute_host.activity", "params": params}) + assert server._ws_orphan_turn_activity_is_fresh(session) is (change == "none") + if change == "none": + # Repeated delivery is an observation of the same clock, not a refresh. + monkeypatch.setattr(server.time, "perf_counter_ns", lambda: params["activity_ns"] + 31_000_000_000) + server._relay_compute_host_rpc({"method": "compute_host.activity", "params": params}) + assert not server._ws_orphan_turn_activity_is_fresh(session) + + +def _run_child(mode, directory): + import socket + from agent.activity_tracking import ActivityTrackingMixin + from agent.session_activity import build_activity_snapshot + from tui_gateway.compute_host import run_host + + def no_network(*args, **kwargs): + raise AssertionError("test child must not contact a provider") + socket.socket.connect = no_network + + class Agent(ActivityTrackingMixin): + def __init__(self, sid): + self.session_id = sid + self._interrupt = threading.Event() + if mode == "previous": + self._touch_activity("previous turn") + + def get_activity_summary(self): + return build_activity_snapshot(last_activity_at=getattr(self, "_last_activity_ts", None), + last_activity_description="test provider") + + def clear_interrupt(self): + self._interrupt.clear() + + def interrupt(self, **kwargs): + self._interrupt.set() + + def run_conversation(self, *args, **kwargs): + Path(directory, "provider-started").touch() + deadline = time.monotonic() + 20 + while not self._interrupt.wait(0.05) and time.monotonic() < deadline: + if Path(directory, "release").exists(): + break + if mode == "fresh" and args[0] != "next": + self._touch_activity("provider wait") + elif mode == "stale": + self._last_activity_ts = time.time() - 3600 + return {"final_response": "done", "interrupted": self._interrupt.is_set()} + + def init(sid, key, agent, history, **kwargs): + s = _session(sid) + s.update(agent=agent, running=False, transport=None, + image_counter=0, slash_worker=None, show_reasoning=False, + tool_progress_mode="all") + server._sessions[sid] = s + + server._make_agent = lambda sid, *a, **kw: Agent(sid) + server._init_session = init + server._wire_callbacks = lambda *a: None + server._sync_agent_model_with_config = lambda *a: None + server._register_session_cwd = lambda *a: None + server._tts_stream_begin = lambda: None + server._sync_session_key_after_compress = lambda *a, **kw: None + server._get_usage = lambda *a: {} + run_host(stdout=sys.__stdout__) + + +if __name__ == "__main__": + _run_child(sys.argv[1], sys.argv[2]) diff --git a/tui_gateway/compute_host.py b/tui_gateway/compute_host.py index 48256fc709..2b76aa5b7b 100644 --- a/tui_gateway/compute_host.py +++ b/tui_gateway/compute_host.py @@ -7,6 +7,7 @@ import argparse import concurrent.futures import contextlib import json +import logging import os import signal import subprocess @@ -223,6 +224,7 @@ class ComputeHost: return session.update(running=True, _turn_cancel_requested=False, last_active=time.time()) server._start_inflight_turn(session, inflight) + turn_started_at = time.time() self._reply("turn.started", sid, request_id, started_ns=now_ns()) with contextlib.suppress(Exception): server._ensure_session_db_row(session) @@ -235,7 +237,10 @@ class ComputeHost: request_id, sid, session, text, display_kind=frame.get("display_kind") or None) run_thread = session.get("_run_thread") if run_thread is not None and hasattr(run_thread, "join"): - run_thread.join() + while run_thread.is_alive(): + run_thread.join(timeout=1.0) + if run_thread.is_alive() and frame.get("turn_id"): + self._emit_turn_activity(sid, session, frame["turn_id"], turn_started_at) with session["history_lock"]: meta = _history_meta(session) interrupted = bool(session.get("_turn_cancel_requested")) @@ -255,6 +260,23 @@ class ComputeHost: server._clear_inflight_turn(session) self._reply("turn.error", sid, request_id, reason="exception", message=str(exc)) + def _emit_turn_activity(self, sid: str, session: dict, turn_id: str, started_at: float) -> None: + # Observe the agent clock, never the host heartbeat. A reused agent's last + # turn must not lend its activity to a new turn that has not made progress. + activity_ns = None + try: + summary = session["agent"].get_activity_summary() + stamped_at = summary.get("last_activity_at") + elapsed = summary.get("seconds_since_activity") + if stamped_at is not None and stamped_at >= started_at and elapsed is not None and elapsed >= 0: + activity_ns = now_ns() - int(elapsed * 1_000_000_000) + except Exception: + logging.getLogger(__name__).debug("compute host activity unavailable sid=%s", sid, exc_info=True) + # perf_counter is shared across local processes; queued frames and cached + # samples age without requiring synchronized wall clocks in the parent. + self._transport.write({"jsonrpc": "2.0", "method": "compute_host.activity", "params": { + "session_id": sid, "turn_id": turn_id, "activity_ns": activity_ns}}) + def _ensure_server_session(self, server: Any, frame: dict[str, Any]) -> dict: sid = str(frame.get("sid") or "") session = server._sessions.get(sid) diff --git a/tui_gateway/compute_host_bridge.py b/tui_gateway/compute_host_bridge.py index 1804ce9da4..ab5fd9b022 100644 --- a/tui_gateway/compute_host_bridge.py +++ b/tui_gateway/compute_host_bridge.py @@ -91,6 +91,15 @@ def _compute_host_adopt_frame_meta(session: dict, frame: dict) -> None: def _relay_compute_host_rpc(message: dict) -> bool: """Relay host events while retaining the clarify snapshot needed on resume.""" params = message.get("params") if isinstance(message, dict) else None + if isinstance(message, dict) and message.get("method") == "compute_host.activity": + if isinstance(params, dict): + session = _sessions.get(str(params.get("session_id") or "")) + if session is not None: + with _history_lock(session): + if (session.get("running") and params.get("turn_id") + and session.get("_compute_host_turn_id") == params["turn_id"]): + session["_compute_host_activity_ns"] = params.get("activity_ns") + return True # Internal observation, not a client event or replay entry. kind = params.get("type") if isinstance(params, dict) else None if kind in {"clarify.request", "clarify.expire"}: session = _sessions.get(str(params.get("session_id") or "")) @@ -208,15 +217,30 @@ def _submit_prompt_to_compute_host( frame = _compute_host_turn_frame(rid, sid, session, text, image_paths=image_paths, queued_prompt_generation=queued_prompt_generation, display_kind=display_kind) + # Caller JSON-RPC ids may repeat across sockets and turns. Use an opaque + # dispatch lifetime token, installed before a fast child can send activity. + turn_id = frame["turn_id"] = frame["request_id"] = uuid.uuid4().hex + with session["history_lock"]: + session["_compute_host_turn_id"] = turn_id + session.pop("_compute_host_activity_ns", None) def _complete(done: dict) -> None: # submit_turn reports a synchronous pipe failure via the callback before re-raising; # leave the session untouched so prompt.submit can fail open to the in-process path. if done.get("reason") != "send_failed": + with session["history_lock"]: + if session.get("_compute_host_turn_id") != turn_id: + return + session.pop("_compute_host_turn_id", None) + session.pop("_compute_host_activity_ns", None) _on_compute_host_turn_done(rid, sid, session, done) try: _get_compute_host_supervisor(cfg).submit_turn(frame, on_complete=_complete) except Exception as exc: + with session["history_lock"]: + if session.get("_compute_host_turn_id") == turn_id: + session.pop("_compute_host_turn_id", None) + session.pop("_compute_host_activity_ns", None) return _err(rid, 5019, f"compute-host dispatch failed: {exc}") with session["history_lock"]: session["_compute_host_active"] = True diff --git a/tui_gateway/session_lifecycle.py b/tui_gateway/session_lifecycle.py index e7a77fcb37..0679f1ea28 100644 --- a/tui_gateway/session_lifecycle.py +++ b/tui_gateway/session_lifecycle.py @@ -473,9 +473,16 @@ def _ws_orphan_turn_activity_is_fresh(session: dict) -> bool: Reuses the agent's existing activity summary (``_touch_activity`` is stamped by API waits, stream tokens, and tool heartbeats — the same clock the turn-liveness watchdog samples; see agent/turn_liveness.py). See #100325, #98028. + Isolated turns mirror that clock from the child under a unique dispatch token; + their monotonic samples keep aging even if the child or its pipe stalls. """ if _WS_ORPHAN_ACTIVITY_STALE_S <= 0: return False + if session.get("_compute_host_turn_id"): + with session["history_lock"]: + stamp = session.get("_compute_host_activity_ns") + return (session.get("running", False) and isinstance(stamp, int) + and 0 <= (time.perf_counter_ns() - stamp) / 1_000_000_000 < _WS_ORPHAN_ACTIVITY_STALE_S) if not callable(summary_fn := getattr(session.get("agent"), "get_activity_summary", None)): return False try: