diff --git a/gateway/session_context.py b/gateway/session_context.py index cdd1a8bfafef..2214c2a56883 100644 --- a/gateway/session_context.py +++ b/gateway/session_context.py @@ -79,6 +79,13 @@ def session_context_engaged() -> bool: _SESSION_USER_NAME: ContextVar = ContextVar("HERMES_SESSION_USER_NAME", default=_UNSET) _SESSION_KEY: ContextVar = ContextVar("HERMES_SESSION_KEY", default=_UNSET) _SESSION_ID: ContextVar = ContextVar("HERMES_SESSION_ID", default=_UNSET) +# In-process UI session/window id for multi-session desktop/TUI hosts. This is +# intentionally separate from HERMES_SESSION_ID: the latter is the durable +# conversation/session-db id, while the UI id is the live frontend tab/window +# that commissioned a detached completion. Background completions use it as a +# precise return address so a stale/rotated durable session key cannot be +# consumed by whichever desktop poller wakes first. +_SESSION_UI_SESSION_ID: ContextVar = ContextVar("HERMES_UI_SESSION_ID", default=_UNSET) # ID of the message that triggered the current turn. Used as a reply anchor # so background-process notifications stay inside the originating Telegram # private-chat topic (those lanes route only with thread id + reply anchor). @@ -123,6 +130,7 @@ def session_context_engaged() -> bool: "HERMES_SESSION_USER_NAME": _SESSION_USER_NAME, "HERMES_SESSION_KEY": _SESSION_KEY, "HERMES_SESSION_ID": _SESSION_ID, + "HERMES_UI_SESSION_ID": _SESSION_UI_SESSION_ID, "HERMES_SESSION_MESSAGE_ID": _SESSION_MESSAGE_ID, "HERMES_SESSION_PROFILE": _SESSION_PROFILE, "HERMES_CRON_AUTO_DELIVER_PLATFORM": _CRON_AUTO_DELIVER_PLATFORM, @@ -160,6 +168,7 @@ def set_session_vars( profile: str = "", cwd: str = "", async_delivery: bool = True, + ui_session_id: str = "", ) -> list: """Set all session context variables and return reset tokens. @@ -191,6 +200,7 @@ def set_session_vars( _SESSION_USER_NAME.set(user_name), _SESSION_KEY.set(session_key), _SESSION_ID.set(session_id), + _SESSION_UI_SESSION_ID.set(ui_session_id), _SESSION_MESSAGE_ID.set(message_id), _SESSION_PROFILE.set(profile), _SESSION_ASYNC_DELIVERY.set(bool(async_delivery)), @@ -225,6 +235,7 @@ def clear_session_vars(tokens: list) -> None: _SESSION_USER_NAME, _SESSION_KEY, _SESSION_ID, + _SESSION_UI_SESSION_ID, _SESSION_MESSAGE_ID, _SESSION_PROFILE, ): diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 50a9fa8d5ce6..cc6ff4f5b723 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -2112,14 +2112,85 @@ def test_notification_event_routing_by_session_key(monkeypatch): monkeypatch.setattr(server, "_sessions", {"a": mine, "b": other}) # My own event → handle it. - assert server._notification_event_belongs_elsewhere(mine, {"session_key": "mine"}) is False + assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": "mine"}) is False # Global/system event with no owner → handle it. - assert server._notification_event_belongs_elsewhere(mine, {"session_key": ""}) is False - assert server._notification_event_belongs_elsewhere(mine, {}) is False + assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": ""}) is False + assert server._notification_event_belongs_elsewhere("a", mine, {}) is False # Owned by another *live* session → defer to that session's poller. - assert server._notification_event_belongs_elsewhere(mine, {"session_key": "other"}) is True + assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": "other"}) is True # Owner is gone (not in _sessions) → handle as fallback so it isn't lost. - assert server._notification_event_belongs_elsewhere(mine, {"session_key": "ghost"}) is False + assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": "ghost"}) is False + + +def test_async_delegation_event_prefers_origin_ui_session(monkeypatch): + """Detached subagent completions return to the commissioning TUI tab. + + Regression: when the durable session key was stale/orphaned, whichever + desktop poller woke first could consume the async result and inject it into + an unrelated session. + """ + mine = _session(session_key="current-key") + other = _session(session_key="unrelated-key") + monkeypatch.setattr(server, "_sessions", {"origin-sid": mine, "other-sid": other}) + monkeypatch.setattr(server, "_get_db", lambda: None) + evt = { + "type": "async_delegation", + "session_key": "stale-or-rotated-key", + "origin_ui_session_id": "origin-sid", + } + + assert server._notification_event_belongs_elsewhere("other-sid", other, evt) is True + assert server._notification_event_belongs_elsewhere("origin-sid", mine, evt) is False + + +def test_notification_event_follows_compression_continuation(monkeypatch): + """Events keyed to a compressed parent route to the live continuation.""" + old_parent = _session(session_key="old-parent") + live_tip = _session(session_key="new-tip") + monkeypatch.setattr(server, "_sessions", {"old-sid": old_parent, "tip-sid": live_tip}) + + class _DB: + def resolve_resume_session_id(self, session_id): + return "new-tip" if session_id == "old-parent" else session_id + + monkeypatch.setattr(server, "_get_db", lambda: _DB()) + evt = {"type": "async_delegation", "session_key": "old-parent"} + + assert server._notification_event_belongs_elsewhere("old-sid", old_parent, evt) is True + assert server._notification_event_belongs_elsewhere("tip-sid", live_tip, evt) is False + # A third session must leave it alone for the continuation's poller. + third = _session(session_key="third") + monkeypatch.setattr( + server, + "_sessions", + {"old-sid": old_parent, "tip-sid": live_tip, "third-sid": third}, + ) + assert server._notification_event_belongs_elsewhere("third-sid", third, evt) is True + + +def test_finalized_origin_ui_session_falls_back_to_live_continuation(monkeypatch): + """A closed origin tab must not steal its resumed continuation's result.""" + finalized_origin = _session(session_key="old-parent", _finalized=True) + live_tip = _session(session_key="new-tip") + monkeypatch.setattr( + server, + "_sessions", + {"origin-sid": finalized_origin, "tip-sid": live_tip}, + ) + + class _DB: + def resolve_resume_session_id(self, session_id): + return "new-tip" if session_id == "old-parent" else session_id + + monkeypatch.setattr(server, "_get_db", lambda: _DB()) + evt = { + "type": "async_delegation", + "session_key": "old-parent", + "origin_ui_session_id": "origin-sid", + } + + assert server._notification_event_belongs_elsewhere("origin-sid", finalized_origin, evt) is True + assert server._notification_event_belongs_elsewhere("tip-sid", live_tip, evt) is False def test_prompt_submit_rejects_negative_truncate_ordinal(monkeypatch): diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index 0cbd9313cfb4..9a473394b514 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -289,6 +289,68 @@ def slow_child(task_index, goal, child=None, parent_agent=None, **kw): assert "the real task" in text +def test_delegate_task_background_uses_live_tui_agent_session_id(monkeypatch): + """TUI async delegation must route to the live/compressed agent id. + + Regression: delegate_task captured the stale approval/session context key + after compression rotated parent_agent.session_id. The resulting completion + was orphaned and could be consumed by an unrelated desktop session poller. + """ + import json + from unittest.mock import MagicMock + import tools.delegate_tool as dt + from gateway.session_context import clear_session_vars, set_session_vars + from tools.approval import reset_current_session_key, set_current_session_key + + parent = MagicMock() + parent._delegate_depth = 0 + parent.session_id = "post-compress-tip" + parent._interrupt_requested = False + parent._active_children = [] + parent._active_children_lock = None + fake_child = MagicMock() + fake_child._delegate_role = "leaf" + + creds = { + "model": "m", "provider": None, "base_url": None, "api_key": None, + "api_mode": None, "command": None, "args": None, + } + monkeypatch.setattr(dt, "_build_child_agent", lambda **kw: fake_child) + monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: creds) + monkeypatch.setattr( + dt, + "_run_single_child", + lambda *a, **k: { + "task_index": 0, + "status": "completed", + "summary": "done", + "api_calls": 1, + "duration_seconds": 0.1, + "model": "m", + "exit_reason": "completed", + }, + ) + + approval_token = set_current_session_key("pre-compress-parent") + session_tokens = set_session_vars( + source="tui", + session_key="pre-compress-parent", + ui_session_id="origin-tab", + ) + try: + out = dt.delegate_task(goal="bg task", background=True, parent_agent=parent) + assert json.loads(out)["status"] == "dispatched" + evt = _drain_one() + finally: + reset_current_session_key(approval_token) + clear_session_vars(session_tokens) + + assert evt is not None + assert evt["type"] == "async_delegation" + assert evt["session_key"] == "post-compress-tip" + assert evt["origin_ui_session_id"] == "origin-tab" + + def test_delegate_task_background_batch_runs_as_one_unit(monkeypatch): """A multi-item batch with background=True dispatches the WHOLE fan-out as ONE background unit (one handle, one async slot). The children run in diff --git a/tests/tui_gateway/test_delegation_session_lifecycle.py b/tests/tui_gateway/test_delegation_session_lifecycle.py new file mode 100644 index 000000000000..57a96469e230 --- /dev/null +++ b/tests/tui_gateway/test_delegation_session_lifecycle.py @@ -0,0 +1,158 @@ +"""Fail-closed ownership + session-scoped delegation lifecycle (#55578). + +Covers the two hardening rules layered on top of the origin-routing salvage: + +1. ``_session_owns_notification_event`` — positive-proof ownership. An + async-delegation completion may only be injected into a session that + PROVABLY commissioned it (origin UI id, or session-key/lineage match). + Orphans are never adopted by a foreign chat. + +2. ``interrupt_for_session`` — a session's in-flight async delegations end + with the session. ``_finalize_session`` interrupts delegations owned by + the closing session (by origin UI id always; by durable key only when the + TUI owns the lifecycle). +""" + +import threading +from unittest.mock import MagicMock, patch + +import pytest + +import tools.async_delegation as ad +from tui_gateway.server import ( + _finalize_session, + _session_owns_notification_event, +) + + +@pytest.fixture(autouse=True) +def _reset_async_delegation(): + ad._reset_for_tests() + yield + ad._reset_for_tests() + + +class TestSessionOwnsNotificationEvent: + def _session(self, key="sess_key_1"): + return {"session_key": key, "_finalized": False} + + def test_origin_ui_match_owns(self): + evt = {"type": "async_delegation", "origin_ui_session_id": "tab1", "session_key": "other"} + assert _session_owns_notification_event("tab1", self._session(), evt) is True + + def test_session_key_match_owns(self): + evt = {"type": "async_delegation", "origin_ui_session_id": "", "session_key": "sess_key_1"} + assert _session_owns_notification_event("tabX", self._session("sess_key_1"), evt) is True + + def test_orphan_is_not_owned(self): + """No origin match, no key match, owner gone → NOT ours (fail closed).""" + evt = {"type": "async_delegation", "origin_ui_session_id": "dead_tab", "session_key": "gone_key"} + assert _session_owns_notification_event("tab1", self._session(), evt) is False + + def test_empty_key_and_origin_not_owned(self): + """A delegation event with no return address at all is never adopted.""" + evt = {"type": "async_delegation", "origin_ui_session_id": "", "session_key": ""} + assert _session_owns_notification_event("tab1", self._session(), evt) is False + + def test_finalized_session_owns_nothing(self): + evt = {"type": "async_delegation", "origin_ui_session_id": "tab1", "session_key": "sess_key_1"} + sess = self._session() + sess["_finalized"] = True + assert _session_owns_notification_event("tab1", sess, evt) is False + + def test_compression_chain_resolution_owns(self): + evt = {"type": "async_delegation", "origin_ui_session_id": "", "session_key": "parent_key"} + db = MagicMock() + db.resolve_resume_session_id.return_value = "child_key" + with patch("tui_gateway.server._get_db", return_value=db): + assert _session_owns_notification_event("tabX", self._session("child_key"), evt) is True + + +class TestInterruptForSession: + def _seed_record(self, delegation_id, session_key="", origin_ui_session_id="", status="running"): + fn = MagicMock() + with ad._records_lock: + ad._records[delegation_id] = { + "delegation_id": delegation_id, + "status": status, + "session_key": session_key, + "origin_ui_session_id": origin_ui_session_id, + "interrupt_fn": fn, + } + return fn + + def test_interrupts_only_matching_session(self): + mine = self._seed_record("d1", session_key="sess_A") + other = self._seed_record("d2", session_key="sess_B") + n = ad.interrupt_for_session(session_key="sess_A") + assert n == 1 + mine.assert_called_once() + other.assert_not_called() + + def test_matches_by_origin_ui_session_id(self): + mine = self._seed_record("d1", origin_ui_session_id="tab1") + other = self._seed_record("d2", origin_ui_session_id="tab2") + n = ad.interrupt_for_session(origin_ui_session_id="tab1") + assert n == 1 + mine.assert_called_once() + other.assert_not_called() + + def test_no_selector_is_noop(self): + fn = self._seed_record("d1", session_key="sess_A") + assert ad.interrupt_for_session() == 0 + fn.assert_not_called() + + def test_completed_records_untouched(self): + fn = self._seed_record("d1", session_key="sess_A", status="completed") + assert ad.interrupt_for_session(session_key="sess_A") == 0 + fn.assert_not_called() + + +class TestFinalizeInterruptsOwnDelegations: + def _make_session(self, session_key="sess_A", sid="tab1"): + agent = MagicMock() + agent.session_id = session_key + agent._session_messages = None + agent.model = "m" + agent.platform = "tui" + return { + "agent": agent, + "history": [{"role": "user", "content": "x"}], + "history_lock": threading.Lock(), + "session_key": session_key, + "_finalized": False, + "_sid": sid, + } + + @patch("tui_gateway.server._get_db") + def test_finalize_interrupts_sessions_delegations(self, mock_get_db): + mock_db = MagicMock() + mock_db.get_session.return_value = {"source": "tui"} + mock_get_db.return_value = mock_db + + with patch("tools.async_delegation.interrupt_for_session") as mock_int: + _finalize_session(self._make_session(), end_reason="tui_close") + + mock_int.assert_called_once() + kwargs = mock_int.call_args.kwargs + assert kwargs["session_key"] == "sess_A" + assert kwargs["origin_ui_session_id"] == "tab1" + + @patch("tui_gateway.server._get_db") + def test_viewer_of_gateway_session_only_interrupts_by_origin(self, mock_get_db): + """Closing a TUI viewer tab on a live gateway session must not kill + the gateway's own background work — key-based interrupt is skipped, + origin-id interrupt (this tab's own dispatches) still applies.""" + mock_db = MagicMock() + mock_db.get_session.return_value = {"source": "telegram"} + mock_get_db.return_value = mock_db + + with patch("tools.async_delegation.interrupt_for_session") as mock_int: + _finalize_session( + self._make_session(session_key="agent:main:telegram:dm:123", sid="tab9"), + end_reason="ws_orphan_reap", + ) + + kwargs = mock_int.call_args.kwargs + assert kwargs["session_key"] == "" + assert kwargs["origin_ui_session_id"] == "tab9" diff --git a/tools/async_delegation.py b/tools/async_delegation.py index f28156e2f57a..cc22a7f6e9ab 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -130,6 +130,7 @@ def dispatch_async_delegation( model: Optional[str], session_key: str, runner: Callable[[], Dict[str, Any]], + origin_ui_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, ) -> Dict[str, Any]: @@ -172,6 +173,7 @@ def dispatch_async_delegation( "role": role, "model": model, "session_key": session_key, + "origin_ui_session_id": origin_ui_session_id, "status": "running", "dispatched_at": dispatched_at, "completed_at": None, @@ -282,6 +284,7 @@ def _push_completion_event( # session_key routes the completion back to the originating gateway # session; empty string => CLI (single-session) path. "session_key": record.get("session_key", ""), + "origin_ui_session_id": record.get("origin_ui_session_id", ""), "goal": record.get("goal", ""), "context": record.get("context"), "toolsets": record.get("toolsets"), @@ -317,6 +320,7 @@ def dispatch_async_delegation_batch( model: Optional[str], session_key: str, runner: Callable[[], Dict[str, Any]], + origin_ui_session_id: str = "", interrupt_fn: Optional[Callable[[], None]] = None, max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN, ) -> Dict[str, Any]: @@ -356,6 +360,7 @@ def dispatch_async_delegation_batch( "role": role, "model": model, "session_key": session_key, + "origin_ui_session_id": origin_ui_session_id, "status": "running", "dispatched_at": dispatched_at, "completed_at": None, @@ -453,6 +458,7 @@ def _finalize_batch( "type": "async_delegation", "delegation_id": delegation_id, "session_key": event_record.get("session_key", ""), + "origin_ui_session_id": event_record.get("origin_ui_session_id", ""), "goal": event_record.get("goal", ""), "goals": event_record.get("goals"), "context": event_record.get("context"), @@ -519,6 +525,54 @@ def interrupt_all(reason: str = "shutdown") -> int: return count +def interrupt_for_session( + session_key: str = "", + origin_ui_session_id: str = "", + reason: str = "session_end", +) -> int: + """Signal running async delegations owned by ONE session to stop. + + A delegation's lifecycle is bound to the session that spawned it: when + that session ends, its in-flight background subagents must end with it — + a completed orphan would otherwise sit on the shared completion queue + with no live owner, either leaking into another chat or burning tokens + with no one listening (#55578). + + Matches on ``origin_ui_session_id`` (the live UI session that + commissioned the work) and/or the durable ``session_key``; either + matching field claims the record. Returns how many were interrupted. + """ + if not session_key and not origin_ui_session_id: + return 0 + count = 0 + with _records_lock: + targets = [ + r for r in _records.values() + if r.get("status") == "running" + 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) + ) + ] + for r in targets: + fn = r.get("interrupt_fn") + if callable(fn): + try: + fn() + count += 1 + except Exception as exc: + logger.debug( + "interrupt_for_session: %s interrupt failed: %s", + r.get("delegation_id"), exc, + ) + if count: + logger.info( + "Interrupted %d async delegation(s) for ending session (%s)", + count, reason, + ) + return count + + def _reset_for_tests() -> None: """Test-only: clear all state and tear down the executor.""" global _executor, _executor_max_workers diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index b989b7983b19..4179926091a3 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -2796,6 +2796,25 @@ def _execute_and_aggregate() -> dict: return json.dumps(_sync_result, ensure_ascii=False) _session_key = get_current_session_key(default="") + _origin_ui_session_id = "" + try: + from gateway.session_context import get_session_env + + _source = get_session_env("HERMES_SESSION_SOURCE", "") + _origin_ui_session_id = get_session_env("HERMES_UI_SESSION_ID", "") + # In desktop/TUI, the routable session key is the durable + # AIAgent.session_id. Context compression can rotate that id during + # the same turn before the TUI-side session dict is re-anchored; + # if we capture the stale approval/session context key here, the + # async completion becomes an orphan and any desktop poller may + # consume it. Gateway chats are different: their session_key is the + # platform conversation key (agent:main:...), so keep it there. + if _source == "tui": + _agent_session_id = str(getattr(parent_agent, "session_id", "") or "") + if _agent_session_id: + _session_key = _agent_session_id + except Exception: + _origin_ui_session_id = "" _child_agents = [c for (_, _, c) in children] # Detach every child from the parent's interrupt-propagation list — the @@ -2836,6 +2855,7 @@ def _batch_interrupt(): role=top_role, model=creds["model"], session_key=_session_key, + origin_ui_session_id=_origin_ui_session_id, runner=_batch_runner, interrupt_fn=_batch_interrupt, max_async_children=_get_max_async_children(), diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 8092147bc79c..80fa9cea95ed 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -625,6 +625,7 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No # Use session_id (from agent.session_id) not session_key — after compression, # session_key may be stale (the ended parent) while session_id is the live # continuation. Fix for #20001. + _tui_owns_lifecycle = True if session_id: try: db = _get_db() @@ -638,11 +639,40 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No # repeats on every inbound message. (#60609) row = db.get_session(session_id) source = (row or {}).get("source", "") - if not _is_gateway_owned_source(source): + _tui_owns_lifecycle = not _is_gateway_owned_source(source) + if _tui_owns_lifecycle: db.end_session(session_id, end_reason) except Exception: pass + # A session's in-flight async delegations end WITH the session (#55578): + # once nobody owns the return address, a still-running background subagent + # can only burn tokens and park an orphaned completion on the shared + # queue. Always interrupt delegations commissioned by THIS live UI session + # (its sid); additionally interrupt by durable session_key, but only when + # the TUI owns the lifecycle — closing a viewer tab on a live gateway + # session must not kill the gateway's own background work. + 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 = "" + interrupt_for_session( + session_key=str(session_key or "") if _tui_owns_lifecycle else "", + origin_ui_session_id=_own_sid, + reason=end_reason, + ) + except Exception: + pass + # Close the slash-worker subprocess as part of finalize itself, not just # in the callers. Defense-in-depth: every session-end path goes through # _finalize_session (it's the single ``_finalized``-guarded chokepoint), so @@ -712,6 +742,10 @@ def _close_session_by_id(sid: str, *, end_reason: str = "tui_close") -> bool: session = _sessions.pop(sid, None) if session is None: return False + # The session is already out of _sessions here, so downstream teardown + # (e.g. _finalize_session's per-session async-delegation interrupt) can't + # recover its live id by scanning the dict — stamp it on the record. + session["_sid"] = sid _teardown_session(session, end_reason=end_reason) return True @@ -1946,7 +1980,12 @@ def _cwd_for_session_key(session_key: str) -> str: return "" -def _set_session_context(session_key: str, cwd: str | None = None) -> list: +def _set_session_context( + session_key: str, + cwd: str | None = None, + *, + ui_session_id: str = "", +) -> list: try: from gateway.session_context import set_session_vars @@ -1961,7 +2000,12 @@ def _set_session_context(session_key: str, cwd: str | None = None) -> list: if sess.get("session_key") == session_key: source = _session_source(sess) break - return set_session_vars(session_key=session_key, source=source, cwd=resolved) + return set_session_vars( + session_key=session_key, + source=source, + cwd=resolved, + ui_session_id=ui_session_id, + ) except Exception: return [] @@ -8451,22 +8495,76 @@ def run_after_agent_ready() -> None: return _ok(rid, {"status": "streaming"}) -def _notification_event_belongs_elsewhere(session: dict, evt: dict) -> bool: +def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) -> bool: """True if ``evt`` is owned by a *different* live session. - Background-process events carry the ``session_key`` of the session that - started the process. Since all desktop sessions share one process-wide - completion queue, each poller must skip events it doesn't own so a - background job's completion surfaces in the session that launched it — not - whichever poller happened to dequeue first. Orphaned events (owner gone) - and global/system events (empty ``session_key``) return False so the - current poller still handles them rather than losing them. + Background completions carry the ``session_key`` of the session that started + the work. Async delegation completions from the desktop also carry + ``origin_ui_session_id``: the live TUI tab/window that commissioned them. + Since all desktop sessions share one process-wide completion queue, each + poller must skip events it doesn't own so a detached result surfaces in the + launching session, not whichever poller happened to dequeue first. """ + evt_ui_sid = str(evt.get("origin_ui_session_id") or "") + if evt_ui_sid: + if evt_ui_sid == str(sid or "") and not session.get("_finalized"): + return False + try: + with _sessions_lock: + owner_live = evt_ui_sid in _sessions and not _sessions[evt_ui_sid].get("_finalized") + except Exception: + owner_live = False + if owner_live: + return True + # If the exact UI tab is gone, fall through to durable session_key + # routing. That avoids wrong-session delivery while still allowing a + # resumed continuation with the same durable key/lineage to claim it. + evt_key = str(evt.get("session_key") or "") if not evt_key: return False - if evt_key == str(session.get("session_key") or ""): + + current_keys = { + str(session.get("session_key") or ""), + _session_lookup_key(session, fallback=sid), + } + + # Compression can rotate AIAgent.session_id while the detached child is + # still running. Resolve the event's original key to its continuation tip so + # an event captured before or after compression still maps to the same live + # desktop session instead of becoming an orphan that any poller may consume. + resolved_key = evt_key + try: + db = _get_db() + if db is not None: + resolved_key = db.resolve_resume_session_id(evt_key) or evt_key + except Exception: + resolved_key = evt_key + + # If the key has a live continuation, prefer that continuation over the + # compressed parent. Otherwise a stale parent tab could consume the event + # before the real current conversation sees it. + if resolved_key != evt_key: + if resolved_key in current_keys: + return False + try: + with _sessions_lock: + continuation_live = any( + not s.get("_finalized") + and ( + str(s.get("session_key") or "") == resolved_key + or _session_lookup_key(s, fallback="") == resolved_key + ) + for s in _sessions.values() + ) + except Exception: + continuation_live = False + if continuation_live: + return True + + if evt_key in current_keys: return False + try: with _sessions_lock: snapshot = list(_sessions.values()) @@ -8476,11 +8574,50 @@ def _notification_event_belongs_elsewhere(session: dict, evt: dict) -> bool: return False return any( - s is not session and str(s.get("session_key") or "") == evt_key + s is not session + and not s.get("_finalized") + and ( + str(s.get("session_key") or "") in {evt_key, resolved_key} + or _session_lookup_key(s, fallback="") in {evt_key, resolved_key} + ) for s in snapshot ) +def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool: + """True iff *this* session PROVABLY owns ``evt``. + + Positive ownership — the mirror of ``_notification_event_belongs_elsewhere`` + minus its orphan-adoption fallback. An event owns-matches when its + ``origin_ui_session_id`` is this live session, or its ``session_key`` + (raw or resolved through the compression chain) matches this session's + key/lineage. Used as a fail-closed gate for async-delegation payloads: + "not provably elsewhere" is NOT good enough to inject a conversation + payload into this chat (#55578). + """ + if session.get("_finalized"): + return False + if str(evt.get("origin_ui_session_id") or "") == str(sid or ""): + return True + evt_key = str(evt.get("session_key") or "") + if not evt_key: + return False + current_keys = { + str(session.get("session_key") or ""), + _session_lookup_key(session, fallback=sid), + } + if evt_key in current_keys: + return True + try: + db = _get_db() + resolved_key = ( + db.resolve_resume_session_id(evt_key) if db is not None else evt_key + ) or evt_key + except Exception: + resolved_key = evt_key + return resolved_key in current_keys + + def _notification_event_dedup_key(evt: dict) -> tuple: """Return the UI-emission identity for a process notification event. @@ -8546,11 +8683,37 @@ def _notification_poller_loop( # process started in session A would surface its completion in whichever # session's poller happened to wake first (Ben's "reported in a # different session" bug). Leave foreign events for their owner. - if _notification_event_belongs_elsewhere(session, evt): + if _notification_event_belongs_elsewhere(sid, session, evt): process_registry.completion_queue.put(evt) time.sleep(0.1) continue + # Fail closed for async-delegation results (#55578): these carry a + # conversation payload, and injecting one into any chat other than the + # one that commissioned it is a hard cross-session leak. The + # belongs-elsewhere check above already re-queued events owned by + # another LIVE session; what reaches here is either ours or an + # orphan whose owner is gone. Orphaned delegation payloads are + # DROPPED, not adopted — the subagent's summary is already persisted + # in the delegation records/output store, so nothing is lost, whereas + # a wrong-chat injection is unrecoverable. Non-delegation events + # (background process completions etc.) keep the historical + # adopt-orphans behavior. + if evt.get("type") == "async_delegation" and not _session_owns_notification_event( + sid, session, evt + ): + logger.warning( + "async-delegation completion %s has no live owner " + "(origin=%r key=%r); dropping from injection instead of " + "delivering to session %s (#55578 fail-closed; result " + "remains in the delegation records)", + evt.get("delegation_id", "?"), + str(evt.get("origin_ui_session_id") or ""), + str(evt.get("session_key") or ""), + sid, + ) + continue + _evt_sid = evt.get("session_id", "") if evt.get("type") == "completion" and process_registry.is_completion_consumed(_evt_sid): continue @@ -8604,7 +8767,16 @@ def _notification_poller_loop( evt = process_registry.completion_queue.get_nowait() except Exception: break - if _notification_event_belongs_elsewhere(session, evt): + if _notification_event_belongs_elsewhere(sid, session, evt): + deferred.append(evt) + continue + # Same fail-closed rule as the live loop: an orphaned async-delegation + # payload is never adopted by a foreign session — defer it (a later + # resume of the owner's lineage can still claim it) rather than + # injecting another chat's conversation here (#55578). + if evt.get("type") == "async_delegation" and not _session_owns_notification_event( + sid, session, evt + ): deferred.append(evt) continue _evt_sid = evt.get("session_id", "") @@ -8729,7 +8901,10 @@ def run(): ) approval_token = set_current_session_key(session["session_key"]) - session_tokens = _set_session_context(session["session_key"]) + session_tokens = _set_session_context( + session["session_key"], + ui_session_id=sid, + ) _profile_home_str = session.get("profile_home") if _profile_home_str: home_token = set_hermes_home_override(_profile_home_str)