Gateway silently drops responses after /stop interrupt

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #52084 fix(gateway): surface retry hint instead of silently dropping turn after /stop (#31884)

  merged to main

  files: gateway/run.py, tests/test_lazy_session_regressions.py

  Show the diff

  • gateway/run.py

    @@ -2332,6 +2332,12 @@ def _normalize_empty_agent_response(
         Consolidates the existing ``failed`` handler and adds a catch-all for
         the case where the agent did work (api_calls > 0) but returned no text.
         Fix for #18765.
    +
    +    Also surfaces a retry hint when the agent never ran at all
    +    (api_calls == 0) for a non-interrupted, non-failed turn -- this is the
    +    silent-drop pattern observed after ``/stop`` where the next user
    +    message hits a stale generation token and returns an empty result,
    +    leaving the platform with nothing to send. (#31884)
         """
         if response:
             return response
    @@ -2364,6 +2370,22 @@ def _normalize_empty_agent_response(
                 "This may be a transient error — try sending your message again."
             )
     
    +    # api_calls == 0, not failed, not interrupted: the agent never ran for
    +    # this turn. This is the post-/stop generation-race pattern where the
    +    # gateway would otherwise silently drop the turn (response=0 chars) and
    +    # the user sees no reply at all. Surface a short retry hint so the
    +    # message isn't lost in silence. (#31884)
    +    if (
    +        api_calls == 0
    +        and not agent_result.get("interrupted")
    +        and not agent_result.get("failed")
    +        and not agent_result.get("partial")
    +    ):
    +        return (
    +            "⚠️ Your message wasn't processed (the previous turn was still "
    +            "being cleaned up). Please send it again."
    +        )
    +
         return response
     
     

  • tests/test_lazy_session_regressions.py

    @@ -415,6 +415,31 @@ def test_nonempty_response_passes_through(self):
     
             assert result == "Hello!"
     
    +    def test_silent_drop_after_stop_surfaces_hint(self):
    +        """Regression for #31884: after /stop, the next user message hits a
    +        stale generation token in _run_agent and returns with api_calls=0,
    +        no failure, no interruption. Without normalization the gateway
    +        silently drops the turn (response=0 chars). Surface a retry hint
    +        so the user knows the message was lost."""
    +        from gateway.run import _normalize_empty_agent_response
    +
    +        agent_result = {
    +            "final_response": "",
    +            "api_calls": 0,
    +            "failed": False,
    +            "interrupted": False,
    +            "partial": False,
    +        }
    +
    +        response = agent_result.get("final_response") or ""
    +        result = _normalize_empty_agent_response(
    +            agent_result, response, history_len=10,
    +        )
    +
    +        assert result, "Silent-drop turn must surface a user-facing hint"
    +        lowered = result.lower()
    +        assert "send it again" in lowered or "try again" in lowered
    +
     
     # ===========================================================================
     # Prune: finalize_orphaned_compression_sessions

• #59552 fix(gateway): stop post-/stop stale interrupt from swallowing the next message (salvage of #44222 by @AIalliAI)

  merged to main

  files: gateway/run.py, scripts/release.py, tests/gateway/test_tool_response_drop_recovery.py

  Show the diff

  • 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 @@ async def _interrupt_and_clear_session(
             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]

  • scripts/release.py

    @@ -282,6 +282,7 @@
         "[redacted:email]": "synapsesx",
         "[redacted:email]": "itsflownium",
         "[redacted:email]": "dirtyren",
    +    "[redacted:email]": "AIalliAI",
         "[redacted:email]": "devatnull",
         "[redacted:email]": "Marxb85",
         "[redacted:email]": "yyzquwu",  # PR #47567 salvage (Matrix: register inbound handlers with wait_sync=True so _dispatch_sync's gather awaits them; without it mautrix fire-and-forgets and inbound intake has no completion point)

  • tests/gateway/test_tool_response_drop_recovery.py

    @@ -256,3 +256,124 @@ async def handler(_event):
                 "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_surfaces_retry_hint(self):
    +        """No interrupt and no work — #31884 (landed after this PR was
    +        written) surfaces a retry hint instead of silence for the
    +        generation-race drop."""
    +        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 "send it again" in 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)"
    +        )