diff --git a/gateway/run.py b/gateway/run.py index 4fe3668fc8..5138ad3c5d 100644 --- a/gateway/run.py +++ b/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: diff --git a/gateway/run_inbound.py b/gateway/run_inbound.py index da9770ad93..bd39387ca0 100644 --- a/gateway/run_inbound.py +++ b/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 diff --git a/tests/gateway/test_reaped_eviction_interrupts_run.py b/tests/gateway/test_reaped_eviction_interrupts_run.py new file mode 100644 index 0000000000..1b9b3f1801 --- /dev/null +++ b/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 diff --git a/tests/gateway/test_reaped_session_recovery.py b/tests/gateway/test_reaped_session_recovery.py index 2937d8ca84..c3ef31a419 100644 --- a/tests/gateway/test_reaped_session_recovery.py +++ b/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