From 75efd73961a88a531276d0e2e1cd30d1ec18019e Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Wed, 8 Jul 2026 07:54:53 -0700 Subject: [PATCH] fix(gateway): never resurrect ended sessions for delegation completions; /new severs in-flight delegations Completes the session-binding class on the gateway surface (#55578), matching the TUI rules: 1. Fail-closed pinning: switch_session() re-opens ended sessions, so pinning a completion to a spawning session that has since ENDED (user /new, closed rotation) would resurrect a conversation the user explicitly ended and inject into it. The injection path now checks the pinned row's ended_at first and drops the injection with a WARNING when the spawning session is dead or unknown - the result stays in the delegation records. 2. /new ends the old conversation's delegations: _handle_reset_command calls interrupt_for_session() with the expiring durable session id (matching the parent_session_id pin stamped at dispatch) plus the routing key as fallback, so a reset can't leave dangling subagents whose completions have no live owner. interrupt_for_session() gains the parent_session_id selector because a gateway chat's session_key (the platform conversation key) survives a reset while the session id rotates - key-based matching alone could never sever a gateway conversation's delegations. --- gateway/run.py | 25 ++++ gateway/slash_commands.py | 18 +++ .../test_async_delegation_session_binding.py | 130 ++++++++++++++++++ tools/async_delegation.py | 16 ++- 4 files changed, 185 insertions(+), 4 deletions(-) create mode 100644 tests/gateway/test_async_delegation_session_binding.py diff --git a/gateway/run.py b/gateway/run.py index 252836ada9f..e71fe26ba8e 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -10637,6 +10637,31 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew (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: diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index a299e8e9b35..3d7bed924c7 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -167,6 +167,24 @@ class GatewaySlashCommandsMixin: 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() diff --git a/tests/gateway/test_async_delegation_session_binding.py b/tests/gateway/test_async_delegation_session_binding.py new file mode 100644 index 00000000000..c48b37e47df --- /dev/null +++ b/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 diff --git a/tools/async_delegation.py b/tools/async_delegation.py index ce4b094beb4..9b4b0e9646a 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -539,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. @@ -549,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: @@ -563,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: