fix(tui-gateway): preserve active isolated turns after disconnect

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])

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)

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

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: