fix(delegate): pin gateway async completions to the spawning session (#57498)

gateway/run.py

@@ -10633,6 +10633,46 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
 
         session_entry = self.session_store.get_or_create_session(source)
         session_key = session_entry.session_key
+        pinned_session_id = str(
+            (getattr(event, "metadata", None) or {}).get("gateway_session_id") or ""
+        ).strip()
+        if pinned_session_id and pinned_session_id != session_entry.session_id:
+            # Fail closed (#55578): the spawning session may have ENDED since
+            # dispatch (user /new-reset, compression rotation whose parent was
+            # closed). switch_session() re-opens ended sessions, so pinning
+            # blindly would RESURRECT a conversation the user explicitly
+            # ended and inject into it — the same illicit-revival class as
+            # the ws_orphan_reap loop (#60609). A completion whose spawning
+            # session is dead is dropped from injection; the subagent's
+            # output remains in the delegation records.
+            pinned_row = None
+            try:
+                if self._session_db is not None:
+                    # AsyncSessionDB already offloads to a thread.
+                    pinned_row = await self._session_db.get_session(pinned_session_id)
+            except Exception:
+                pinned_row = None
+            if pinned_row is None or pinned_row.get("ended_at"):
+                logger.warning(
+                    "Async-delegation completion pinned to session %s, which is "
+                    "%s — dropping injection instead of resurrecting it "
+                    "(#55578 fail-closed; result remains in the delegation "
+                    "records).",
+                    pinned_session_id,
+                    "unknown" if pinned_row is None else "ended",
+                )
+                return
+            prior_session_id = session_entry.session_id
+            switched = self.session_store.switch_session(session_key, pinned_session_id)
+            if switched is not None:
+                session_entry = switched
+                logger.info(
+                    "Pinned async-delegation completion to spawning session %s "
+                    "(was %s) for routing key %s (#57498)",
+                    pinned_session_id,
+                    prior_session_id,
+                    session_key,
+                )
         self._cache_session_source(session_key, source)
         if await asyncio.to_thread(self._is_telegram_topic_lane, source):
             try:
@@ -15202,12 +15242,17 @@ async def _inject_watch_notification(self, synth_text: str, evt: dict) -> None:
         if not adapter:
             return
         try:
+            metadata = {}
+            parent_session_id = str(evt.get("parent_session_id") or "").strip()
+            if parent_session_id:
+                metadata["gateway_session_id"] = parent_session_id
             synth_event = MessageEvent(
                 text=synth_text,
                 message_type=MessageType.TEXT,
                 source=source,
                 internal=True,
                 message_id=str(evt.get("message_id") or "").strip() or None,
+                metadata=metadata,
             )
             logger.info(
                 "Watch pattern notification — injecting for %s chat=%s thread=%s",

gateway/slash_commands.py

@@ -167,6 +167,24 @@ async def _handle_reset_command(self, event: MessageEvent) -> Union[str, Ephemer
         if _qe is not None:
             _qe.pop(session_key, None)
 
+        # The old conversation's in-flight async delegations end WITH it
+        # (#55578): after the reset rotates the session id, their completions
+        # would have no live owner — a dangling subagent can only burn tokens
+        # and park an orphaned payload on the shared queue. Interrupt by the
+        # expiring durable session id (delegations dispatched from gateway
+        # chats are pinned to it via parent_session_id) and by the routing
+        # key as a fallback for older records.
+        try:
+            from tools.async_delegation import interrupt_for_session
+
+            interrupt_for_session(
+                session_key=session_key,
+                parent_session_id=str(getattr(old_entry, "session_id", "") or ""),
+                reason="session_reset",
+            )
+        except Exception:
+            pass
+
         try:
             from tools.env_passthrough import clear_env_passthrough
             clear_env_passthrough()

tests/gateway/test_async_delegation_session_binding.py

@@ -0,0 +1,130 @@
+"""Gateway-side session binding for async delegations (#57498, #55578).
+
+Three invariants on the messaging-gateway surface, mirroring the TUI rules:
+
+1. Completions are pinned to the spawning session (contributor commit).
+2. A dead/ended spawning session is never resurrected: the injection is
+   dropped, fail-closed (never rerouted to the peer's current session).
+3. /new interrupts the old conversation's in-flight async delegations.
+"""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+import tools.async_delegation as ad
+
+
+@pytest.fixture(autouse=True)
+def _reset_async_delegation():
+    ad._reset_for_tests()
+    yield
+    ad._reset_for_tests()
+
+
+def _seed_record(delegation_id, session_key="", parent_session_id="", status="running"):
+    fn = MagicMock()
+    with ad._records_lock:
+        ad._records[delegation_id] = {
+            "delegation_id": delegation_id,
+            "status": status,
+            "session_key": session_key,
+            "parent_session_id": parent_session_id,
+            "interrupt_fn": fn,
+        }
+    return fn
+
+
+class TestInterruptForSessionByParentId:
+    def test_parent_session_id_selector(self):
+        mine = _seed_record("d1", session_key="agent:main:telegram:dm:1", parent_session_id="sess_old")
+        other = _seed_record("d2", session_key="agent:main:telegram:dm:2", parent_session_id="sess_other")
+        n = ad.interrupt_for_session(parent_session_id="sess_old")
+        assert n == 1
+        mine.assert_called_once()
+        other.assert_not_called()
+
+    def test_reset_interrupts_by_key_and_parent(self):
+        """A /new reset passes both selectors — either match claims the record."""
+        by_key = _seed_record("d1", session_key="agent:main:telegram:dm:1", parent_session_id="")
+        by_parent = _seed_record("d2", session_key="", parent_session_id="sess_old")
+        unrelated = _seed_record("d3", session_key="other", parent_session_id="other")
+        n = ad.interrupt_for_session(
+            session_key="agent:main:telegram:dm:1",
+            parent_session_id="sess_old",
+            reason="session_reset",
+        )
+        assert n == 2
+        by_key.assert_called_once()
+        by_parent.assert_called_once()
+        unrelated.assert_not_called()
+
+
+class TestGatewayPinningFailsClosed:
+    """The gateway injection path must never resurrect an ended session."""
+
+    def _make_runner(self, pinned_row):
+        from gateway.run import GatewayRunner
+
+        runner = object.__new__(GatewayRunner)
+        db = MagicMock()
+        db.get_session = AsyncMock(return_value=pinned_row)
+        runner._session_db = db
+
+        entry = MagicMock()
+        entry.session_key = "agent:main:telegram:dm:1"
+        entry.session_id = "sess_current"
+        runner.session_store = MagicMock()
+        runner.session_store.get_or_create_session.return_value = entry
+        runner.session_store.switch_session.return_value = entry
+        return runner, entry
+
+    def _run_pinning_prefix(self, runner, pinned_session_id):
+        """Execute the pinning guard logic exactly as _handle_message does."""
+
+        async def _go():
+            event = MagicMock()
+            event.metadata = {"gateway_session_id": pinned_session_id}
+            session_entry = runner.session_store.get_or_create_session(MagicMock())
+            pinned = str((getattr(event, "metadata", None) or {}).get("gateway_session_id") or "").strip()
+            if pinned and pinned != session_entry.session_id:
+                pinned_row = None
+                try:
+                    if runner._session_db is not None:
+                        pinned_row = await runner._session_db.get_session(pinned)
+                except Exception:
+                    pinned_row = None
+                if pinned_row is None or pinned_row.get("ended_at"):
+                    return "dropped"
+                switched = runner.session_store.switch_session(session_entry.session_key, pinned)
+                if switched is not None:
+                    return "pinned"
+            return "default"
+
+        return asyncio.run(_go())
+
+    def test_live_spawning_session_pins(self):
+        runner, _ = self._make_runner({"id": "sess_old", "ended_at": None})
+        assert self._run_pinning_prefix(runner, "sess_old") == "pinned"
+
+    def test_ended_spawning_session_drops(self):
+        runner, _ = self._make_runner({"id": "sess_old", "ended_at": "2026-07-08T00:00:00"})
+        assert self._run_pinning_prefix(runner, "sess_old") == "dropped"
+        runner.session_store.switch_session.assert_not_called()
+
+    def test_unknown_spawning_session_drops(self):
+        runner, _ = self._make_runner(None)
+        assert self._run_pinning_prefix(runner, "sess_gone") == "dropped"
+        runner.session_store.switch_session.assert_not_called()
+
+
+class TestResetHandlerInterruptsDelegations:
+    def test_reset_command_calls_interrupt_for_session(self):
+        """The /new handler must sever the old conversation's delegations."""
+        import inspect
+        from gateway import slash_commands
+
+        src = inspect.getsource(slash_commands.GatewaySlashCommandsMixin._handle_reset_command)
+        assert "interrupt_for_session" in src
+        assert "session_reset" in src

tests/tools/test_async_delegation.py

@@ -99,6 +99,7 @@ def runner():
     res = ad.dispatch_async_delegation(
         goal="compute X", context="some context", toolsets=["web", "file"],
         role="leaf", model="test-model", session_key="agent:main:cli:dm:local",
+        parent_session_id="20260703_parent_sid",
         runner=runner, max_async_children=3,
     )
     assert res["status"] == "dispatched"
@@ -108,6 +109,7 @@ def runner():
     assert evt["type"] == "async_delegation"
     assert evt["summary"] == "the result"
     assert evt["session_key"] == "agent:main:cli:dm:local"
+    assert evt["parent_session_id"] == "20260703_parent_sid"
     assert evt["delegation_id"] == res["delegation_id"]
 
 

tools/async_delegation.py

@@ -129,6 +129,7 @@ def dispatch_async_delegation(
     role: str,
     model: Optional[str],
     session_key: str,
+    parent_session_id: Optional[str] = None,
     runner: Callable[[], Dict[str, Any]],
     origin_ui_session_id: str = "",
     interrupt_fn: Optional[Callable[[], None]] = None,
@@ -146,6 +147,11 @@ def dispatch_async_delegation(
         captured on the parent thread BEFORE dispatch, because the daemon
         worker thread won't carry the contextvar. Used to route the
         completion back to the originating session.
+    parent_session_id
+        The durable ``state.db`` session id of the parent agent that spawned
+        the delegation. Carried on the completion event so the gateway can
+        pin routing to the spawning session instead of recovering the latest
+        ``ended_at IS NULL`` row for the peer tuple (#57498).
     runner
         Zero-arg callable that builds + runs the child and returns the same
         result dict ``_run_single_child`` produces. Runs on the worker thread.
@@ -174,6 +180,7 @@ def dispatch_async_delegation(
         "model": model,
         "session_key": session_key,
         "origin_ui_session_id": origin_ui_session_id,
+        "parent_session_id": parent_session_id,
         "status": "running",
         "dispatched_at": dispatched_at,
         "completed_at": None,
@@ -285,6 +292,7 @@ def _push_completion_event(
         # session; empty string => CLI (single-session) path.
         "session_key": record.get("session_key", ""),
         "origin_ui_session_id": record.get("origin_ui_session_id", ""),
+        "parent_session_id": record.get("parent_session_id"),
         "goal": record.get("goal", ""),
         "context": record.get("context"),
         "toolsets": record.get("toolsets"),
@@ -319,6 +327,7 @@ def dispatch_async_delegation_batch(
     role: str,
     model: Optional[str],
     session_key: str,
+    parent_session_id: Optional[str] = None,
     runner: Callable[[], Dict[str, Any]],
     origin_ui_session_id: str = "",
     interrupt_fn: Optional[Callable[[], None]] = None,
@@ -361,6 +370,7 @@ def dispatch_async_delegation_batch(
         "model": model,
         "session_key": session_key,
         "origin_ui_session_id": origin_ui_session_id,
+        "parent_session_id": parent_session_id,
         "status": "running",
         "dispatched_at": dispatched_at,
         "completed_at": None,
@@ -459,6 +469,7 @@ def _finalize_batch(
         "delegation_id": delegation_id,
         "session_key": event_record.get("session_key", ""),
         "origin_ui_session_id": event_record.get("origin_ui_session_id", ""),
+        "parent_session_id": event_record.get("parent_session_id"),
         "goal": event_record.get("goal", ""),
         "goals": event_record.get("goals"),
         "context": event_record.get("context"),
@@ -528,6 +539,7 @@ def interrupt_all(reason: str = "shutdown") -> int:
 def interrupt_for_session(
     session_key: str = "",
     origin_ui_session_id: str = "",
+    parent_session_id: str = "",
     reason: str = "session_end",
 ) -> int:
     """Signal running async delegations owned by ONE session to stop.
@@ -538,11 +550,17 @@ def interrupt_for_session(
     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.
+    Selectors (any matching field claims the record):
+    - ``origin_ui_session_id``: the live TUI tab/window that commissioned it.
+    - ``session_key``: the durable routing key captured at dispatch.
+    - ``parent_session_id``: the spawning agent's durable session-db id —
+      the right selector for gateway chats, whose ``session_key`` (the
+      platform conversation key) SURVIVES a ``/new`` reset while the
+      session id rotates.
+
+    Returns how many were interrupted.
     """
-    if not session_key and not origin_ui_session_id:
+    if not session_key and not origin_ui_session_id and not parent_session_id:
         return 0
     count = 0
     with _records_lock:
@@ -552,6 +570,7 @@ def interrupt_for_session(
             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)
             )
         ]
     for r in targets:

tools/delegate_tool.py

@@ -2815,6 +2815,7 @@ def _execute_and_aggregate() -> dict:
                     _session_key = _agent_session_id
         except Exception:
             _origin_ui_session_id = ""
+        _parent_session_id = getattr(parent_agent, "session_id", None)
         _child_agents = [c for (_, _, c) in children]
 
         # Detach every child from the parent's interrupt-propagation list — the
@@ -2856,6 +2857,7 @@ def _batch_interrupt():
             model=creds["model"],
             session_key=_session_key,
             origin_ui_session_id=_origin_ui_session_id,
+            parent_session_id=_parent_session_id,
             runner=_batch_runner,
             interrupt_fn=_batch_interrupt,
             max_async_children=_get_max_async_children(),