[verified] fix(desktop): preserve background delegates across session switches

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}

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:

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"):