fix(gateway): interrupt evicted in-flight runs

Fixes #106963.

Co-authored-by: KoNit-K <124019182+KoNit-K@users.noreply.github.com>

gateway/run.py

@@ -2567,6 +2567,7 @@ def _dequeue_pending_event(adapter, session_key: str) -> MessageEvent | None:
 _INTERRUPT_REASON_STOP = "Stop requested"
 _INTERRUPT_REASON_RESET = "Session reset requested"
 _INTERRUPT_REASON_TIMEOUT = "Execution timed out (inactivity)"
+_INTERRUPT_REASON_EVICTED = "Session ended while the turn was running"
 _INTERRUPT_REASON_SSE_DISCONNECT = "SSE client disconnected"
 _INTERRUPT_REASON_GATEWAY_SHUTDOWN = "Gateway shutting down"
 _INTERRUPT_REASON_GATEWAY_RESTART = "Gateway restarting"
@@ -2700,7 +2701,8 @@ def _watch_gateway_turn_inactivity(
 _CONTROL_INTERRUPT_MESSAGES = frozenset({
     _INTERRUPT_REASON_STOP.lower(), _INTERRUPT_REASON_RESET.lower(),
     _INTERRUPT_REASON_TIMEOUT.lower(), _INTERRUPT_REASON_SSE_DISCONNECT.lower(),
-    _INTERRUPT_REASON_GATEWAY_SHUTDOWN.lower(), _INTERRUPT_REASON_GATEWAY_RESTART.lower()})
+    _INTERRUPT_REASON_EVICTED.lower(), _INTERRUPT_REASON_GATEWAY_SHUTDOWN.lower(),
+    _INTERRUPT_REASON_GATEWAY_RESTART.lower()})
 
 
 def _is_control_interrupt_message(message: Optional[str]) -> bool:

gateway/run_inbound.py

@@ -489,8 +489,26 @@ class GatewayInboundMixin:
             logger.debug("reaped-session staleness check failed", exc_info=True)
 
     def _hm_evict_running_agent(self, _quick_key: str, reason: str) -> None:
+        from gateway.run import _AGENT_PENDING_SENTINEL, _INTERRUPT_REASON_EVICTED, request_hard_interrupt
+        state = self._peek_session_state(_quick_key)
+        running_agent = state.turn.agent if state else None
+        if running_agent and running_agent is not _AGENT_PENDING_SENTINEL:
+            try:
+                request_hard_interrupt(running_agent, _INTERRUPT_REASON_EVICTED)
+            except Exception:
+                # Eviction must still invalidate and release the slot if a legacy or third-party
+                # agent cannot accept the interrupt request; otherwise the stale session remains
+                # unroutable and the pre-existing cleanup path is lost.
+                logger.warning(
+                    "Failed to interrupt evicted agent for %s; continuing eviction cleanup",
+                    _quick_key,
+                    exc_info=True,
+                )
         self._invalidate_session_run_generation(_quick_key, reason=reason)
         self._release_running_agent_state(_quick_key)
+        # The interrupt flag is cleared only by the turn finalizer. Remove the cached instance after
+        # releasing the slot so a late-finishing orphan cannot poison the replacement turn.
+        self._evict_cached_agent(_quick_key)
 
     def _hm_merge_pending_for_source(
         self, source: SessionSource, _quick_key: str, event: "MessageEvent", *, merge_text: bool = False

tests/gateway/test_reaped_eviction_interrupts_run.py

@@ -0,0 +1,99 @@
+"""Regression tests for interrupting work evicted from a gateway turn slot."""
+
+from __future__ import annotations
+
+import pytest
+
+from gateway.run import (
+    GatewayRunner,
+    _AGENT_PENDING_SENTINEL,
+    _is_control_interrupt_message,
+)
+from gateway.run_inbound import GatewayInboundMixin
+
+
+KEY = "agent:main:telegram:dm:106963"
+EVICTION_REASON = "Session ended while the turn was running"
+
+
+class _RecordingAgent:
+    def __init__(self, events: list[tuple], slot_agent) -> None:
+        self._events = events
+        self._slot_agent = slot_agent
+        self.interrupted = False
+
+    def hard_interrupt(self, message: str | None = None, **_kwargs) -> None:
+        self.interrupted = True
+        self._events.append(("interrupt", message, self._slot_agent() is self))
+
+
+class _RaisingAgent:
+    def hard_interrupt(self, _message: str | None = None, **_kwargs) -> None:
+        raise RuntimeError("interrupt transport failed")
+
+
+class _ReapedStore:
+    def peek_session_id(self, _session_key: str) -> str:
+        return "session-106963"
+
+    def _is_session_ended_in_db(self, session_id: str) -> bool:
+        return session_id == "session-106963"
+
+
+def _build_gateway(agent, events: list[tuple]):
+    gateway = object.__new__(GatewayRunner)
+    gateway._persist_active_agents = lambda: None
+    gateway._agent_cache_lock = None
+    gateway._agent_cache = {KEY: (agent, "signature", 0)}
+    gateway._spawn_release_thread = lambda target, args, name, inline_fallback: events.append(
+        ("cache_release", args[0])
+    )
+    state = gateway._session_state(KEY)
+    state.turn.agent = agent
+    return gateway, state
+
+
+@pytest.mark.parametrize("entrypoint", ("direct", "reaped"))
+def test_eviction_interrupts_before_release_and_drops_cached_agent(entrypoint: str) -> None:
+    events: list[tuple] = []
+    gateway = None
+
+    def current_agent():
+        state = gateway._peek_session_state(KEY)
+        return state.turn.agent if state else None
+
+    agent = _RecordingAgent(events, current_agent)
+    gateway, _state = _build_gateway(agent, events)
+    release = gateway._release_running_agent_state
+
+    def release_with_record(session_key: str, **kwargs) -> bool:
+        events.append(("release", current_agent() is agent))
+        return release(session_key, **kwargs)
+
+    gateway._release_running_agent_state = release_with_record
+    if entrypoint == "reaped":
+        gateway.session_store = _ReapedStore()
+        gateway._hm_evict_reaped_agent(KEY)
+    else:
+        gateway._hm_evict_running_agent(KEY, "stale_running_agent_eviction")
+
+    assert isinstance(gateway, GatewayInboundMixin)
+    assert agent.interrupted
+    assert events[0] == ("interrupt", EVICTION_REASON, True)
+    release_events = [event for event in events if event[0] == "release"]
+    assert release_events == [("release", True)]
+    assert events.index(events[0]) < events.index(release_events[0])
+    assert gateway._peek_session_state(KEY).turn.agent is None
+    assert KEY not in gateway._agent_cache
+    assert _is_control_interrupt_message(EVICTION_REASON)
+
+
+@pytest.mark.parametrize("agent", (None, _AGENT_PENDING_SENTINEL, _RaisingAgent()))
+def test_eviction_cleanup_survives_empty_pending_or_failed_interrupt(agent) -> None:
+    events: list[tuple] = []
+    gateway, state = _build_gateway(agent, events)
+
+    gateway._hm_evict_running_agent(KEY, "reaped_session_eviction")
+
+    assert state.turn.agent is None
+    assert KEY not in gateway._agent_cache

tests/gateway/test_reaped_session_recovery.py

@@ -22,7 +22,7 @@ from gateway.config import (
     PlatformConfig,
 )
 from gateway.platforms.event import MessageEvent, MessageType
-from gateway.run import GatewayRunner
+from gateway.run import GatewayRunner, _INTERRUPT_REASON_EVICTED
 from gateway.session import SessionSource, SessionStore
 
 
@@ -36,7 +36,7 @@ class _FakeAdapter:
 
 
 class _DeadReapedAgent:
-    """Runtime whose turn was reaped: interrupt() lands nowhere."""
+    """Runtime whose durable session was reaped; records eviction interrupts."""
 
     def __init__(self):
         self.interrupts = []
@@ -138,8 +138,9 @@ async def test_reaped_session_message_reaches_cold_path(tmp_path):
 
     assert result == "COLD_PATH_REPLY"
     assert cold_path.await_count == 1
-    # Nothing was interrupt()-delivered into the dead runtime.
-    assert agent.interrupts == []
+    # The stale runtime is interrupted before eviction, then the message still heals through
+    # the cold path instead of being delivered to the ended session.
+    assert agent.interrupts == [_INTERRUPT_REASON_EVICTED]
 
 
 @pytest.mark.asyncio