From a14caf7759e2571b23b44ef95ebae75e088ec475 Mon Sep 17 00:00:00 2001 From: AIalliAI <285906080+AIalliAI@users.noreply.github.com> Date: Thu, 18 Jun 2026 22:29:57 +0000 Subject: [PATCH] fix(gateway): stop post-/stop stale interrupt from silently swallowing the next message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A /stop sets _interrupt_requested on the session's cached agent, but the flag is only cleared by the turn finalizer. When the stopped run is hung or still draining, the flag survives the forced lock release and the session's NEXT user message is killed at the top of the tool loop (conversation_loop.py interrupt check): the run completes with interrupted=True, api_calls=0 and an empty response, which _normalize_empty_agent_response passed through as pure silence — the user's message was swallowed with no trace except a 'response ready: ... api_calls=0 response=0 chars' log line. Two-layer fix: - _interrupt_and_clear_session now evicts the cached agent whenever it releases the running state. The next message rebuilds the agent from session history (mirroring the /new and /model paths), while the old agent object keeps its interrupt flag so a hung drain still dies when it unblocks. This intentionally does NOT clear the flag in place: turn_context deliberately preserves a pending interrupt across turn start (it carries interrupt-message delivery), and clearing it could revive a hung run the user just stopped. - _normalize_empty_agent_response distinguishes a drain from a swallowed turn: an interrupted run that did work (api_calls > 0) stays silent as before (deliberate stop/steer; queued messages are delivered by the recursive drain inside _run_agent), but an interrupted run with ZERO api_calls never processed the user's message at all and now surfaces a 'send it again' notice instead of nothing. Same silent-delivery class as a1f76ba7e (#29346), which covered the extract-stripped case; regression tests added next to that coverage. Fixes #44212 --- gateway/run.py | 27 +++- .../test_tool_response_drop_recovery.py | 119 ++++++++++++++++++ 2 files changed, 145 insertions(+), 1 deletion(-) diff --git a/gateway/run.py b/gateway/run.py index 51e33ceb499..4b2b45b1559 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2580,7 +2580,22 @@ def _normalize_empty_agent_response( ) api_calls = int(agent_result.get("api_calls", 0) or 0) - if api_calls > 0 and not agent_result.get("interrupted"): + if agent_result.get("interrupted"): + # An interrupted run that did work (api_calls > 0) is the drain of a + # run the user deliberately stopped or steered — its silence is + # intentional, and any queued/interrupting message is delivered by + # the recursive drain inside _run_agent before this result is seen. + # An interrupted run with ZERO api_calls never processed the user's + # message at all: it was killed at the top of the tool loop by an + # interrupt flag left over from a recent /stop (#44212). Pure + # silence there swallows a real user message, so surface it. + if api_calls == 0: + return ( + "⚠️ Your message was interrupted before processing started " + "(likely by a recent /stop). Please send it again." + ) + return response + if api_calls > 0: if agent_result.get("partial"): err = agent_result.get("error", "processing incomplete") return f"⚠️ Processing stopped: {str(err)[:200]}. Try again." @@ -15650,6 +15665,16 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew self._pending_messages.pop(session_key, None) if release_running_state: self._release_running_agent_state(session_key) + # Evict the cached agent: ``_interrupt_requested`` is only + # cleared by the turn finalizer, so on a hung or still-draining + # run the flag survives the lock release and kills the session's + # NEXT message at the top of the tool loop (interrupted=True, + # api_calls=0, empty response — silently swallowed, #44212). + # Evicting mirrors the /new and /model paths: the next message + # rebuilds the agent from session history, while the old agent + # object keeps its interrupt flag so a hung drain still dies + # when it unblocks. + self._evict_cached_agent(session_key) async def _refresh_agent_cache_message_count( self, session_key: str, session_id: Optional[str] diff --git a/tests/gateway/test_tool_response_drop_recovery.py b/tests/gateway/test_tool_response_drop_recovery.py index f39362b30c8..ba983f36718 100644 --- a/tests/gateway/test_tool_response_drop_recovery.py +++ b/tests/gateway/test_tool_response_drop_recovery.py @@ -256,3 +256,122 @@ class TestUnrecoverableDropIsLoud: "response_delivery_dropped" in r.getMessage() for r in caplog.records if r.levelno == logging.ERROR ), [r.getMessage() for r in caplog.records] + + +# =========================================================================== +# Issue #44212: post-/stop stale interrupt silently swallows the next message +# =========================================================================== + +class TestPostStopInterruptSwallow: + """A `/stop` sets ``_interrupt_requested`` on the session's cached agent, + but the flag is only cleared by the turn finalizer. When the stopped run + is hung or still draining, the flag survives the lock release and the + session's NEXT message is killed at the top of the tool loop — + ``interrupted=True, api_calls=0, final_response=""`` — which + ``_normalize_empty_agent_response`` used to pass through as pure silence. + + Two-layer fix: ``_interrupt_and_clear_session`` evicts the cached agent + (root cause), and the normalizer surfaces a notice for interrupted runs + that never made an API call (a swallowed user turn, not a drain).""" + + def test_interrupted_zero_api_calls_surfaces_notice(self): + """Interrupted before the first API call → the user's message was + never processed; silence here swallows it (the #44212 malign shape: + ``response ready ... api_calls=0 response=0 chars``).""" + from gateway.run import _normalize_empty_agent_response + + agent_result = { + "final_response": None, + "api_calls": 0, + "partial": False, + "interrupted": True, + } + + response = _normalize_empty_agent_response(agent_result, "", history_len=10) + + assert response != "", "A turn killed before doing any work must not be silent" + assert "send it again" in response.lower() + + def test_interrupted_after_work_stays_silent(self): + """Interrupted mid-work → this is the drain of a run the user + deliberately stopped/steered; its silence is intentional (any + queued/interrupting message is delivered by the recursive drain + inside _run_agent).""" + from gateway.run import _normalize_empty_agent_response + + agent_result = { + "final_response": None, + "api_calls": 3, + "partial": False, + "interrupted": True, + } + + response = _normalize_empty_agent_response(agent_result, "", history_len=10) + + assert response == "" + + def test_uninterrupted_zero_api_calls_stays_silent(self): + """No interrupt and no work — unchanged behavior.""" + from gateway.run import _normalize_empty_agent_response + + agent_result = { + "final_response": None, + "api_calls": 0, + "partial": False, + "interrupted": False, + } + + response = _normalize_empty_agent_response(agent_result, "", history_len=10) + + assert response == "" + + @pytest.mark.asyncio + async def test_interrupt_and_clear_session_evicts_cached_agent(self): + """The control-interrupt path must evict the session's cached agent + so its ``_interrupt_requested`` flag cannot leak into the next turn.""" + import threading + + from gateway.run import GatewayRunner, _INTERRUPT_REASON_STOP + + class _RecordingAgent: + def __init__(self): + self.interrupt_reasons = [] + + def interrupt(self, reason=None): + self.interrupt_reasons.append(reason) + + agent = _RecordingAgent() + session_key = "agent:main:telegram:dm:12345" + source = SessionSource( + platform=Platform.TELEGRAM, chat_id="12345", chat_type="dm" + ) + + runner = object.__new__(GatewayRunner) + runner._running_agents = {session_key: agent} + runner._agent_cache = {session_key: (agent, "config-sig")} + runner._agent_cache_lock = threading.Lock() + runner.adapters = {} + runner._pending_messages = {} + + invalidated = [] + runner._invalidate_session_run_generation = ( + lambda key, reason=None: invalidated.append((key, reason)) + ) + released = [] + runner._release_running_agent_state = ( + lambda key, **kw: released.append(key) + ) + + await runner._interrupt_and_clear_session( + session_key, + source, + interrupt_reason=_INTERRUPT_REASON_STOP, + invalidation_reason="stop_command", + ) + + assert agent.interrupt_reasons == [_INTERRUPT_REASON_STOP] + assert released == [session_key] + assert session_key not in runner._agent_cache, ( + "Cached agent with a set interrupt flag must be evicted on /stop " + "so the flag cannot kill the session's next message (#44212)" + )