fix(desktop): isolate process notifications by session

tests/test_tui_gateway_server.py

@@ -7799,14 +7799,9 @@ def start(self):
     monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
     monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
 
-    # Isolate the completion queue for the duration of this test. The poller
-    # reads process_registry.completion_queue by attribute at runtime; the
-    # event below carries no session_key, so any *other* poller (a leaked
-    # daemon thread from another test, or a concurrent one in the same xdist
-    # worker) is allowed to dequeue and dispatch it to its own session — whose
-    # agent may be a fixture double without run_conversation. A fresh Queue
-    # here fully isolates this test; monkeypatch restores the original on
-    # teardown. (Same pattern as test_notification_poller_requeues_when_busy.)
+    # Isolate the completion queue for the duration of this test. The event
+    # carries the matching session key because ownerless notification events
+    # must fail closed instead of being adopted by whichever poller wakes first.
     isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
     monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
     process_registry._completion_consumed.discard("proc_poller_test")
@@ -7818,6 +7813,7 @@ def start(self):
     isolated_queue.put({
         "type": "completion",
         "session_id": "proc_poller_test",
+        "session_key": sess["session_key"],
         "command": "echo hello",
         "exit_code": 0,
         "output": "hello",
@@ -7867,17 +7863,16 @@ def start(self):
     monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
     monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
 
-    # Isolate the completion queue so a concurrent/leaked poller in the same
-    # xdist worker can't dequeue this session_key-less event before our poller
-    # does. monkeypatch restores the shared singleton on teardown. (Same
-    # pattern as test_notification_poller_requeues_when_busy.)
+    # Isolate the completion queue and bind the event to this session so this
+    # test exercises consumed-event suppression rather than orphan rejection.
     isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
     monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
 
     process_registry._completion_consumed.add("proc_already_done")
     isolated_queue.put({
         "type": "completion",
         "session_id": "proc_already_done",
+        "session_key": sess["session_key"],
         "command": "echo x",
         "exit_code": 0,
         "output": "x",
@@ -7920,6 +7915,7 @@ def test_notification_poller_requeues_when_busy(monkeypatch):
     evt = {
         "type": "completion",
         "session_id": "proc_busy_test",
+        "session_key": sess["session_key"],
         "command": "make build",
         "exit_code": 0,
         "output": "ok",
@@ -8051,6 +8047,7 @@ def _fake_run_prompt_submit(rid, sid, session, text):
     base = {
         "type": "watch_match",
         "session_id": "proc_watch_dedup",
+        "session_key": sess["session_key"],
         "command": "tail -f app.log",
         "pattern": "READY",
         "output": "READY on port 8000",

tests/tools/test_process_registry.py

@@ -1404,6 +1404,80 @@ def test_drain_notifications_filters_async_delegation_by_session_key():
             process_registry.completion_queue.get_nowait()
 
 
+def test_drain_notifications_filters_process_completions_by_session_key():
+    """A terminal completion must only become a turn in its owning session.
+
+    Regression for the cross-project context switch where a frontend chat
+    drained a model-router review process completion from the process-global
+    queue. The old filter only covered async_delegation events.
+    """
+    from tools.process_registry import process_registry
+
+    while not process_registry.completion_queue.empty():
+        process_registry.completion_queue.get_nowait()
+
+    try:
+        process_registry.completion_queue.put({
+            "type": "completion",
+            "session_id": "proc_session_a",
+            "session_key": "desktop-session-a",
+            "command": "review project A",
+            "exit_code": 0,
+            "output": "A done",
+        })
+        process_registry.completion_queue.put({
+            "type": "completion",
+            "session_id": "proc_session_b",
+            "session_key": "desktop-session-b",
+            "command": "build project B",
+            "exit_code": 0,
+            "output": "B done",
+        })
+
+        results_a = process_registry.drain_notifications(
+            session_key="desktop-session-a"
+        )
+        assert [r[0]["session_id"] for r in results_a] == ["proc_session_a"]
+        assert "A done" in results_a[0][1]
+
+        results_b = process_registry.drain_notifications(
+            session_key="desktop-session-b"
+        )
+        assert [r[0]["session_id"] for r in results_b] == ["proc_session_b"]
+        assert "B done" in results_b[0][1]
+        assert process_registry.completion_queue.empty()
+    finally:
+        while not process_registry.completion_queue.empty():
+            process_registry.completion_queue.get_nowait()
+
+
+def test_drain_notifications_ownership_callback_filters_process_completion():
+    """The desktop's chain-aware callback must gate terminal completions too."""
+    from tools.process_registry import process_registry
+
+    while not process_registry.completion_queue.empty():
+        process_registry.completion_queue.get_nowait()
+
+    try:
+        process_registry.completion_queue.put({
+            "type": "completion",
+            "session_id": "proc_foreign",
+            "session_key": "foreign-key",
+            "command": "foreign review",
+            "exit_code": 1,
+            "output": "failed",
+        })
+        results = process_registry.drain_notifications(
+            session_key="current-key",
+            owns_event=lambda e: e.get("session_key") == "current-key",
+        )
+        assert results == []
+        assert process_registry.completion_queue.get_nowait()["session_id"] == "proc_foreign"
+    finally:
+        while not process_registry.completion_queue.empty():
+            process_registry.completion_queue.get_nowait()
+
+
 def test_drain_notifications_no_filter_passes_all_async_delegation():
     """Without a session_key filter, all async-delegation events are consumed.
 

tools/process_registry.py

@@ -1168,13 +1168,13 @@ def drain_notifications(
         Skips completion events the agent already consumed via wait/log or
         observed inline via poll() (see ``_drain_should_skip``).
 
-        Async-delegation events carry a conversation payload, so draining one
-        into the wrong session is a cross-chat leak (#58684, #55578). Two
-        filter modes, strongest wins:
+        Every notification event becomes a synthetic conversation turn, so
+        draining even an ordinary process completion into the wrong session is
+        a cross-chat context leak. Two filter modes, strongest wins:
 
         - ``owns_event(evt) -> bool``: positive-proof ownership callback.
-          When provided, an async-delegation event is consumed ONLY if the
-          callback returns True; everything else is re-queued for its owner.
+          When provided, an event is consumed ONLY if the callback returns
+          True; everything else is re-queued for its owner.
           The TUI passes its compression-chain-aware ownership check here so
           a post-compression session still claims its own pre-compression
           dispatches.
@@ -1194,23 +1194,25 @@ def drain_notifications(
             _evt_sid = evt.get("session_id", "")
             if evt.get("type") == "completion" and self._drain_should_skip(_evt_sid):
                 continue
-            # Filter async-delegation events so they are not delivered to the
-            # wrong session/thread (#58684). Positive-proof callback beats
-            # bare key equality when the caller can provide one.
-            if evt.get("type") == "async_delegation":
-                if owns_event is not None:
-                    try:
-                        owned = bool(owns_event(evt))
-                    except Exception:
-                        owned = False  # fail closed — never leak on a broken check
-                    if not owned:
-                        requeue.append(evt)
-                        continue
-                elif session_key:
-                    evt_session_key = evt.get("session_key", "") or ""
-                    if evt_session_key != session_key:
-                        requeue.append(evt)
-                        continue
+            # Filter ALL notification events, not only async delegations.
+            # Process completions and watch matches are also injected as
+            # synthetic user turns; letting whichever chat drains this global
+            # queue first consume them can replace that chat's active task with
+            # another project's background work. Positive-proof ownership beats
+            # bare key equality when the caller can provide it.
+            if owns_event is not None:
+                try:
+                    owned = bool(owns_event(evt))
+                except Exception:
+                    owned = False  # fail closed — never leak on a broken check
+                if not owned:
+                    requeue.append(evt)
+                    continue
+            elif session_key:
+                evt_session_key = evt.get("session_key", "") or ""
+                if evt_session_key != session_key:
+                    requeue.append(evt)
+                    continue
             text = format_process_notification(evt)
             if text:
                 results.append((evt, text))

tui_gateway/server.py

@@ -8638,9 +8638,9 @@ def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool
     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).
+    key/lineage. Used as a fail-closed gate for every notification payload:
+    "not provably elsewhere" is NOT good enough to inject a synthetic
+    conversation turn into this chat (#55578).
     """
     if session.get("_finalized"):
         return False
@@ -8735,26 +8735,23 @@ def _notification_poller_loop(
             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
+        # Fail closed for every notification result (#55578): process
+        # completions and watch matches also become synthetic conversation
+        # turns, so injecting any of them into a chat other than the one that
+        # commissioned it is a hard cross-session context 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
-        ):
+        # another LIVE session; what reaches here is either ours or an orphan
+        # whose owner is gone. Orphaned events are DROPPED from auto-injection,
+        # not adopted. Delegation results remain in the delegation records;
+        # process output remains in the process registry/status terminal. A
+        # missed notification is recoverable, while a wrong-chat injection is
+        # not.
+        if not _session_owns_notification_event(sid, session, evt):
             logger.warning(
-                "async-delegation completion %s has no live owner "
+                "notification %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", "?"),
+                "delivering to session %s (#55578 fail-closed)",
+                evt.get("delegation_id") or evt.get("session_id", "?"),
                 str(evt.get("origin_ui_session_id") or ""),
                 str(evt.get("session_key") or ""),
                 sid,
@@ -8825,13 +8822,10 @@ def _notification_poller_loop(
         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
-        ):
+        # Same fail-closed rule as the live loop: an orphaned notification is
+        # never adopted by a foreign session. Defer it rather than injecting
+        # another chat's synthetic turn during shutdown (#55578).
+        if not _session_owns_notification_event(sid, session, evt):
             deferred.append(evt)
             continue
         _evt_sid = evt.get("session_id", "")