From 820f63e842d15516e2a55a87b51134bfaa360fa5 Mon Sep 17 00:00:00 2001 From: null-runner Date: Sun, 19 Jul 2026 15:46:39 +0200 Subject: [PATCH] [verified] fix(desktop): preserve background delegates across session switches --- tests/test_tui_gateway_server.py | 70 ++++++++++++++++++++++++++ tools/async_delegation.py | 48 ++++++++++++++++-- tui_gateway/server.py | 86 +++++++++++++++++++++++++------- 3 files changed, 181 insertions(+), 23 deletions(-) diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 12161e09ef841..6883916ed97ea 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -14,6 +14,7 @@ import pytest from hermes_constants import reset_hermes_home_override, set_hermes_home_override from hermes_cli.active_sessions import active_session_registry_snapshot from hermes_cli.browser_connect import ChromeDebugLaunch +from tools import async_delegation as ad from tui_gateway import server @@ -3868,6 +3869,75 @@ def test_ws_orphan_reap_spares_reattached_session(monkeypatch): assert server._ws_session_is_orphaned(done) is False +def test_ws_orphan_reap_spares_detached_session_with_running_async_delegation(monkeypatch): + """A detached desktop session with live background delegation is parked. + + Regression for Desktop session switches / transient WS detaches: the parent + turn is idle, but a background delegate_task still owns the session's + return address. Reaping immediately interrupts the child and turns its + completion into an unowned orphan. + """ + timers = [] + closed = [] + + class _Timer: + def __init__(self, _delay, fn): + self.fn = fn + timers.append(self) + + def start(self): + return None + + class _DB: + def get_session(self, _session_id): + return {"id": "sess_bg", "source": "desktop"} + + monkeypatch.setattr(server, "_WS_ORPHAN_REAP_GRACE_S", 0.01) + monkeypatch.setattr(server.threading, "Timer", _Timer) + monkeypatch.setattr(server, "_get_db", lambda: _DB()) + monkeypatch.setattr( + server, + "_close_session_by_id", + lambda sid, *, end_reason="tui_close": closed.append((sid, end_reason)) or True, + ) + + server._sessions["bg-sid"] = _session( + transport=server._detached_ws_transport, + running=False, + session_key="sess_bg", + ) + ad._reset_for_tests() + try: + with ad._records_lock: + ad._records["deleg_bg"] = { + "delegation_id": "deleg_bg", + "status": "running", + "session_key": "sess_bg", + "origin_ui_session_id": "bg-sid", + "interrupt_fn": lambda: None, + } + + server._schedule_ws_orphan_reap("bg-sid") + assert len(timers) == 1 + + timers.pop(0).fn() + + assert closed == [] + assert "bg-sid" in server._sessions + assert len(timers) == 1 + + with ad._records_lock: + ad._records["deleg_bg"]["status"] = "completed" + ad._records["deleg_bg"]["interrupt_fn"] = None + + timers.pop(0).fn() + + assert closed == [("bg-sid", "ws_orphan_reap")] + finally: + ad._reset_for_tests() + server._sessions.pop("bg-sid", None) + + def test_ws_orphan_reap_disabled_when_grace_zero(monkeypatch): """Grace=0 disables the reaper entirely (pre-fix park-forever behaviour).""" fired = {"timer": False} diff --git a/tools/async_delegation.py b/tools/async_delegation.py index fe7c9ee5318d2..3fd08531e0fd0 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -585,6 +585,45 @@ def active_task_count() -> int: return total +def _matches_session_selectors( + record: Dict[str, Any], + *, + session_key: str = "", + origin_ui_session_id: str = "", + parent_session_id: str = "", +) -> bool: + return ( + (origin_ui_session_id and str(record.get("origin_ui_session_id") or "") == origin_ui_session_id) + or (session_key and str(record.get("session_key") or "") == session_key) + or (parent_session_id and str(record.get("parent_session_id") or "") == parent_session_id) + ) + + +def has_live_for_session( + session_key: str = "", + origin_ui_session_id: str = "", + parent_session_id: str = "", +) -> bool: + """Whether a session still owns any live async delegation. + + Live = running / stalling / finalizing — the same states the reapers' + keepalive treats as active work. + """ + if not session_key and not origin_ui_session_id and not parent_session_id: + return False + with _records_lock: + return any( + r.get("status") in {"running", "stalling", "finalizing"} + and _matches_session_selectors( + r, + session_key=session_key, + origin_ui_session_id=origin_ui_session_id, + parent_session_id=parent_session_id, + ) + for r in _records.values() + ) + + def _new_delegation_id() -> str: return f"deleg_{uuid.uuid4().hex[:8]}" @@ -1432,10 +1471,11 @@ def interrupt_for_session( targets = [ r for r in _records.values() if r.get("status") in ("running", "stalling") - and ( - (origin_ui_session_id and str(r.get("origin_ui_session_id") or "") == origin_ui_session_id) - or (session_key and str(r.get("session_key") or "") == session_key) - or (parent_session_id and str(r.get("parent_session_id") or "") == parent_session_id) + and _matches_session_selectors( + r, + session_key=session_key, + origin_ui_session_id=origin_ui_session_id, + parent_session_id=parent_session_id, ) ] for r in targets: diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 16c10d0583d1d..1a3cbd19976cd 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -761,18 +761,11 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No try: from tools.async_delegation import interrupt_for_session - _own_sid = str(session.get("_sid") or "") - if not _own_sid: - try: - with _sessions_lock: - for _cand_sid, _cand in _sessions.items(): - if _cand is session: - _own_sid = _cand_sid - break - except Exception: - _own_sid = "" + _own_sid, _owned_session_key = _session_async_delegation_selectors( + session, sid_hint=str(session.get("_sid") or "") + ) interrupt_for_session( - session_key=str(session_key or "") if _tui_owns_lifecycle else "", + session_key=_owned_session_key, origin_ui_session_id=_own_sid, reason=end_reason, ) @@ -946,14 +939,69 @@ def _ws_session_is_orphaned(session: dict | None) -> bool: return session.get("transport") is _detached_ws_transport -def _session_has_active_delegations(sid: str) -> bool: - """True when UI session ``sid`` still owns live background work.""" - if not sid: +def _session_owns_durable_lifecycle(session_id: str | None) -> bool: + """Whether this TUI/desktop session may end its durable DB row by key.""" + if not session_id: + return True + try: + db = _get_db() + if db is None: + return True + # Don't end gateway-originated sessions — the gateway owns their + # lifecycle. The TUI is only a viewer there (#60609). + row = db.get_session(session_id) + source = (row or {}).get("source", "") + return not _is_gateway_owned_source(source) + except Exception: + return True + + +def _session_async_delegation_selectors( + session: dict | None, *, sid_hint: str = "" +) -> tuple[str, str]: + """Ownership selectors for async background work tied to one UI session.""" + if not session: + return "", "" + own_sid = str(sid_hint or session.get("_sid") or "") + if not own_sid: + try: + with _sessions_lock: + for _cand_sid, _cand in _sessions.items(): + if _cand is session: + own_sid = _cand_sid + break + except Exception: + own_sid = "" + agent = session.get("agent") + session_key = str(session.get("session_key") or "") + session_id = getattr(agent, "session_id", None) or session_key + owned_session_key = session_key if _session_owns_durable_lifecycle(session_id) else "" + return own_sid, owned_session_key + + +def _session_has_active_delegations(sid: str, session: dict | None = None) -> bool: + """True when UI session ``sid`` still owns live background work. + + Matches by the live UI sid AND — when the TUI owns the durable lifecycle + (never for gateway-viewer tabs, #60609) — by the durable session_key, so a + delegation dispatched from an earlier tab of the same resumed session still + keeps it alive. + """ + if session is None: + with _sessions_lock: + session = _sessions.get(sid) + own_sid, owned_session_key = _session_async_delegation_selectors( + session, sid_hint=sid + ) + if not own_sid and not owned_session_key: return False try: - from tools.async_delegation import active_for_session + from tools.async_delegation import has_live_for_session - return active_for_session(sid) > 0 + return has_live_for_session( + session_key=owned_session_key, + origin_ui_session_id=own_sid, + ) except Exception: logger.debug( "Failed to query active delegations for UI session %s", @@ -993,7 +1041,7 @@ def _schedule_ws_orphan_reap(sid: str) -> None: current = _sessions.get(sid) if not _ws_session_is_orphaned(current): return - if _session_has_active_delegations(sid): + if _session_has_active_delegations(sid, current): reschedule = True else: session = _pop_session_by_id(sid) @@ -1079,7 +1127,7 @@ def _transport_is_dead(transport) -> bool: def _session_is_evictable(sid: str, session: dict, now: float) -> bool: if session.get("running") or _session_pending_kind(sid): return False - if _session_has_active_delegations(sid): + if _session_has_active_delegations(sid, session): return False ready = session.get("agent_ready") # Lazy watch sessions (subagent spectator windows) never start a build, @@ -1172,7 +1220,7 @@ def _session_is_lru_evictable(sid: str, session: dict) -> bool: # moment it loses its client. if session.get("running") or _session_pending_kind(sid): return False - if _session_has_active_delegations(sid): + if _session_has_active_delegations(sid, session): return False ready = session.get("agent_ready") if ready is not None and not ready.is_set() and not session.get("lazy"):