fix(gateway): stop post-/stop stale interrupt from silently swallowing the next message

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

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

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)"
+        )