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.
This commit is contained in:
teknium1 2026-07-08 07:54:53 -07:00 committed by Teknium
parent d39c62409b
commit 75efd73961
4 changed files with 185 additions and 4 deletions

View file

@ -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:

View file

@ -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()

View file

@ -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

View file

@ -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: