Gateway compression exhaustion handling and auto-reset

Fixes on record

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

• #4750 fix: prevent compression death spiral from API disconnects (#2153)

  merged to main

  files: gateway/run.py, run_agent.py

• #10063 fix(gateway): break compression-exhaustion infinite loop and auto-reset session

  merged to main

  files: gateway/run.py, run_agent.py, tests/run_agent/test_1630_context_overflow_loop.py

  Show the diff

  • gateway/run.py

    @@ -3898,14 +3898,11 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str):
                 # intermediate reasoning) so sessions can be resumed with full context
                 # and transcripts are useful for debugging and training data.
                 #
    -            # IMPORTANT: When the agent failed before producing any response
    -            # (e.g. context-overflow 400), do NOT persist the user's message.
    +            # IMPORTANT: When the agent failed (e.g. context-overflow 400,
    +            # compression exhausted), do NOT persist the user's message.
                 # Persisting it would make the session even larger, causing the
    -            # same failure on the next attempt — an infinite loop. (#1630)
    -            agent_failed_early = (
    -                agent_result.get("failed")
    -                and not agent_result.get("final_response")
    -            )
    +            # same failure on the next attempt — an infinite loop. (#1630, #9893)
    +            agent_failed_early = bool(agent_result.get("failed"))
                 if agent_failed_early:
                     logger.info(
                         "Skipping transcript persistence for failed request in "
    @@ -3913,6 +3910,24 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str):
                         session_entry.session_id,
                     )
     
    +            # When compression is exhausted, the session is permanently too
    +            # large to process.  Auto-reset it so the next message starts
    +            # fresh instead of replaying the same oversized context in an
    +            # infinite fail loop.  (#9893)
    +            if agent_result.get("compression_exhausted") and session_entry and session_key:
    +                logger.info(
    +                    "Auto-resetting session %s after compression exhaustion.",
    +                    session_entry.session_id,
    +                )
    +                self.session_store.reset_session(session_key)
    +                self._evict_cached_agent(session_key)
    +                self._session_model_overrides.pop(session_key, None)
    +                response = (response or "") + (
    +                    "\n\n🔄 Session auto-reset — the conversation exceeded the "
    +                    "maximum context size and could not be compressed further. "
    +                    "Your next message will start a fresh session."
    +                )
    +
                 ts = datetime.now().isoformat()
                 
                 # If this is a fresh session (no history), write the full tool
    @@ -8624,6 +8639,8 @@ def _approval_notify_sync(approval_data: dict) -> None:
                         "final_response": error_msg,
                         "messages": result.get("messages", []),
                         "api_calls": result.get("api_calls", 0),
    +                    "failed": result.get("failed", False),
    +                    "compression_exhausted": result.get("compression_exhausted", False),
                         "tools": tools_holder[0] or [],
                         "history_offset": len(agent_history),
                         "last_prompt_tokens": _last_prompt_toks,

  • run_agent.py

    @@ -9312,7 +9312,9 @@ def _stop_spinner():
                                     "completed": False,
                                     "api_calls": api_call_count,
                                     "error": f"Request payload too large: max compression attempts ({max_compression_attempts}) reached.",
    -                                "partial": True
    +                                "partial": True,
    +                                "failed": True,
    +                                "compression_exhausted": True,
                                 }
                             self._emit_status(f"⚠️  Request payload too large (413) — compression attempt {compression_attempts}/{max_compression_attempts}...")
     
    @@ -9341,7 +9343,9 @@ def _stop_spinner():
                                     "completed": False,
                                     "api_calls": api_call_count,
                                     "error": "Request payload too large (413). Cannot compress further.",
    -                                "partial": True
    +                                "partial": True,
    +                                "failed": True,
    +                                "compression_exhausted": True,
                                 }
     
                         # Check for context-length errors BEFORE generic 4xx handler.
    @@ -9392,7 +9396,9 @@ def _stop_spinner():
                                         "completed": False,
                                         "api_calls": api_call_count,
                                         "error": f"Context length exceeded: max compression attempts ({max_compression_attempts}) reached.",
    -                                    "partial": True
    +                                    "partial": True,
    +                                    "failed": True,
    +                                    "compression_exhausted": True,
                                     }
                                 restart_with_compressed_messages = True
                                 break
    @@ -9442,7 +9448,9 @@ def _stop_spinner():
                                     "completed": False,
                                     "api_calls": api_call_count,
                                     "error": f"Context length exceeded: max compression attempts ({max_compression_attempts}) reached.",
    -                                "partial": True
    +                                "partial": True,
    +                                "failed": True,
    +                                "compression_exhausted": True,
                                 }
                             self._emit_status(f"🗜️ Context too large (~{approx_tokens:,} tokens) — compressing ({compression_attempts}/{max_compression_attempts})...")
     
    @@ -9473,7 +9481,9 @@ def _stop_spinner():
                                     "completed": False,
                                     "api_calls": api_call_count,
                                     "error": f"Context length exceeded ({approx_tokens:,} tokens). Cannot compress further.",
    -                                "partial": True
    +                                "partial": True,
    +                                "failed": True,
    +                                "compression_exhausted": True,
                                 }
     
                         # Check for non-retryable client errors.  The classifier

  • tests/run_agent/test_1630_context_overflow_loop.py

    @@ -136,33 +136,29 @@ class TestGatewaySkipsPersistenceOnFailure:
         the gateway should NOT persist messages to the transcript."""
     
         def test_agent_failed_early_detected(self):
    -        """The agent_failed_early flag is True when failed=True and
    -        no final_response."""
    +        """The agent_failed_early flag is True when failed=True,
    +        regardless of final_response."""
             agent_result = {
                 "failed": True,
                 "final_response": None,
                 "messages": [],
                 "error": "Non-retryable client error",
             }
    -        agent_failed_early = (
    -            agent_result.get("failed")
    -            and not agent_result.get("final_response")
    -        )
    +        agent_failed_early = bool(agent_result.get("failed"))
             assert agent_failed_early
     
    -    def test_agent_with_response_not_failed_early(self):
    -        """When the agent has a final_response, it's not a failed-early
    -        scenario even if failed=True."""
    +    def test_agent_failed_with_error_response_still_detected(self):
    +        """When _run_agent_blocking converts an error to final_response,
    +        the failed flag should still trigger agent_failed_early.  This
    +        was the core bug in #9893 — the old guard checked
    +        ``not final_response`` which was always truthy after conversion."""
             agent_result = {
                 "failed": True,
    -            "final_response": "Here is a partial response",
    +            "final_response": "⚠️ Request payload too large: max compression attempts reached.",
                 "messages": [],
             }
    -        agent_failed_early = (
    -            agent_result.get("failed")
    -            and not agent_result.get("final_response")
    -        )
    -        assert not agent_failed_early
    +        agent_failed_early = bool(agent_result.get("failed"))
    +        assert agent_failed_early
     
         def test_successful_agent_not_failed_early(self):
             """A successful agent result should not trigger skip."""
    @@ -170,13 +166,41 @@ def test_successful_agent_not_failed_early(self):
                 "final_response": "Hello!",
                 "messages": [{"role": "assistant", "content": "Hello!"}],
             }
    -        agent_failed_early = (
    -            agent_result.get("failed")
    -            and not agent_result.get("final_response")
    -        )
    +        agent_failed_early = bool(agent_result.get("failed"))
             assert not agent_failed_early
     
     
    +class TestCompressionExhaustedFlag:
    +    """When compression is exhausted, the agent should set both
    +    failed=True and compression_exhausted=True so the gateway can
    +    auto-reset the session.  (#9893)"""
    +
    +    def test_compression_exhausted_returns_carry_flag(self):
    +        """Simulate the return dict from a compression-exhausted agent."""
    +        agent_result = {
    +            "messages": [],
    +            "completed": False,
    +            "api_calls": 3,
    +            "error": "Request payload too large: max compression attempts (3) reached.",
    +            "partial": True,
    +            "failed": True,
    +            "compression_exhausted": True,
    +        }
    +        assert agent_result.get("failed")
    +        assert agent_result.get("compression_exhausted")
    +
    +    def test_normal_failure_not_compression_exhausted(self):
    +        """Non-compression failures should not have compression_exhausted."""
    +        agent_result = {
    +            "messages": [],
    +            "completed": False,
    +            "failed": True,
    +            "error": "Invalid API response after 3 retries",
    +        }
    +        assert agent_result.get("failed")
    +        assert not agent_result.get("compression_exhausted")
    +
    +
     # ---------------------------------------------------------------------------
     # Test 3: Context-overflow error messages
     # ---------------------------------------------------------------------------

• #45613 fix(gateway): rebind topic after compression auto-reset

  merged to main

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

  Show the diff

  • gateway/run.py

    @@ -8815,12 +8815,29 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
                         "Auto-resetting session %s after compression exhaustion.",
                         session_entry.session_id,
                     )
    -                self.session_store.reset_session(session_key)
    +                new_entry = self.session_store.reset_session(session_key)
                     self._evict_cached_agent(session_key)
                     self._session_model_overrides.pop(session_key, None)
                     self._set_session_reasoning_override(session_key, None)
                     if hasattr(self, "_pending_model_notes"):
                         self._pending_model_notes.pop(session_key, None)
    +                if new_entry is not None:
    +                    # Drop the stale reference to the bloated compressed child and
    +                    # re-point the Telegram topic binding at the fresh session.
    +                    # Compression rotated session_entry.session_id to the oversized
    +                    # compressed child earlier this turn (the agent-result sync
    +                    # above), and that _sync also rewrote the (chat_id, thread_id)
    +                    # -> bloated-child binding. reset_session swaps in a clean,
    +                    # parentless session, but without re-syncing the binding the
    +                    # next inbound message in this topic gets switch_session'd back
    +                    # onto the bloated child by the binding-heal walk, reloads the
    +                    # oversized transcript, and re-triggers compression exhaustion
    +                    # forever (#35809 — regression of the #9893/#10063 auto-reset).
    +                    # No-op on non-topic lanes.
    +                    session_entry = new_entry
    +                    self._sync_telegram_topic_binding(
    +                        source, session_entry, reason="compression-exhausted-reset",
    +                    )
                     response = (response or "") + (
                         "\n\n🔄 Session auto-reset — the conversation exceeded the "
                         "maximum context size and could not be compressed further. "

  • tests/gateway/test_35809_auto_reset_clean_context.py

    @@ -0,0 +1,196 @@
    +"""Regression tests for #35809 — compression-exhaustion auto-reset loop.
    +
    +After compression is exhausted the gateway auto-resets the session so the
    +next message starts on a fresh, empty conversation (#9893 / #10063). That
    +guarantee regressed once the Telegram topic-binding heal landed
    +(#20470 / #29712 / #33414):
    +
    +    1. Compression rotates ``session_entry.session_id`` to an oversized
    +       compressed *child* session mid-turn and the agent-result sync rewrites
    +       the ``(chat_id, thread_id) -> child`` topic binding.
    +    2. ``reset_session`` swaps in a clean, parentless session — but its return
    +       value was discarded and the topic binding was left pointing at the
    +       bloated child.
    +    3. On the next inbound message in that topic, the binding-heal walk
    +       ``switch_session``'d the freshly-reset lane *back* onto the bloated
    +       child, ``load_transcript`` reloaded the oversized transcript, and
    +       compression exhaustion re-fired — a new session id every loop.
    +
    +The fix captures the fresh entry from ``reset_session`` and re-syncs the
    +topic binding to it (a no-op on non-topic lanes).
    +
    +Two tests:
    +
    +* ``TestAutoResetBlockReSyncsBinding`` — an AST invariant on
    +  ``gateway/run.py`` (mirrors ``test_compression_session_id_persistence.py``):
    +  the compression-exhausted auto-reset block must capture
    +  ``reset_session(...)`` and call ``_sync_telegram_topic_binding`` afterward.
    +  This is the load-bearing regression pin.
    +* ``TestAutoResetLoadsCleanContext`` — a behavioral contract on the real
    +  ``SessionStore``: after ``reset_session`` the next turn loads an EMPTY
    +  transcript for the new session_id, never the bloated child's transcript.
    +"""
    +
    +from __future__ import annotations
    +
    +import ast
    +import inspect
    +
    +from gateway import run as gateway_run
    +from gateway.config import GatewayConfig, Platform
    +from gateway.session import SessionSource, SessionStore
    +from hermes_state import SessionDB
    +
    +
    +# ---------------------------------------------------------------------------
    +# AST invariant: the auto-reset block re-syncs the topic binding
    +# ---------------------------------------------------------------------------
    +def _find_compression_exhausted_reset_block() -> ast.If:
    +    """Return the ``if agent_result.get('compression_exhausted') ...`` block."""
    +    tree = ast.parse(inspect.getsource(gateway_run))
    +
    +    for node in ast.walk(tree):
    +        if not isinstance(node, ast.If):
    +            continue
    +        consts = [
    +            n.value
    +            for n in ast.walk(node.test)
    +            if isinstance(n, ast.Constant) and isinstance(n.value, str)
    +        ]
    +        # Identify the auto-reset branch by the literal passed to .get(...).
    +        if "compression_exhausted" in consts:
    +            # Only the branch that actually performs the reset, not the
    +            # earlier classifier that merely reads the flag into a bool.
    +            calls = {
    +                sub.func.attr
    +                for sub in ast.walk(node)
    +                if isinstance(sub, ast.Call) and isinstance(sub.func, ast.Attribute)
    +            }
    +            if "reset_session" in calls:
    +                return node
    +    raise AssertionError(
    +        "Could not locate the compression-exhausted auto-reset block "
    +        "(if agent_result.get('compression_exhausted') ... reset_session) "
    +        "in gateway/run.py — the structure changed or the AST walker is stale."
    +    )
    +
    +
    +class TestAutoResetBlockReSyncsBinding:
    +    def test_reset_session_return_is_captured(self):
    +        """``reset_session`` must be assigned, not called-and-discarded —
    +        the fresh entry is needed to re-point the binding and drop the stale
    +        reference to the bloated compressed child (#35809)."""
    +        block = _find_compression_exhausted_reset_block()
    +        captured = False
    +        for stmt in ast.walk(block):
    +            if isinstance(stmt, ast.Assign):
    +                val = stmt.value
    +                if (
    +                    isinstance(val, ast.Call)
    +                    and isinstance(val.func, ast.Attribute)
    +                    and val.func.attr == "reset_session"
    +                ):
    +                    captured = True
    +        assert captured, (
    +            "gateway/run.py auto-reset block calls reset_session() but discards "
    +            "its return value. The fresh SessionEntry must be captured so the "
    +            "topic binding can be re-pointed at it; otherwise the next message "
    +            "resolves back to the bloated compressed child (#35809)."
    +        )
    +
    +    def test_topic_binding_is_resynced_after_reset(self):
    +        """The block must re-sync the topic binding so the next inbound message
    +        cannot ``switch_session`` back onto the bloated compressed child."""
    +        block = _find_compression_exhausted_reset_block()
    +        sync_calls = [
    +            sub
    +            for sub in ast.walk(block)
    +            if isinstance(sub, ast.Call)
    +            and isinstance(sub.func, ast.Attribute)
    +            and sub.func.attr == "_sync_telegram_topic_binding"
    +        ]
    +        assert sync_calls, (
    +            "gateway/run.py auto-reset block does not call "
    +            "_sync_telegram_topic_binding after reset_session. Without it the "
    +            "(chat_id, thread_id) -> bloated-child binding survives the reset "
    +            "and the binding-heal walk re-anchors the fresh lane onto the "
    +            "oversized compressed transcript, re-triggering the loop (#35809)."
    +        )
    +
    +
    +# ---------------------------------------------------------------------------
    +# Behavioral contract: reset yields a clean next-turn transcript
    +# ---------------------------------------------------------------------------
    +def _make_store(tmp_path):
    +    store = SessionStore(sessions_dir=tmp_path, config=GatewayConfig())
    +    # Isolate the SQLite transcript store so we exercise per-session_id
    +    # transcripts without touching the developer's real state.db.
    +    store._db = SessionDB(db_path=tmp_path / "state.db")
    +    return store
    +
    +
    +def _make_source():
    +    return SessionSource(platform=Platform.TELEGRAM, chat_id="123", user_id="u1")
    +
    +
    +def _bloat(n):
    +    # Stand-in for the oversized, post-compression "child" transcript that
    +    # could not be compressed any further (#35809).
    +    return [{"role": "user", "content": "x" * 2000} for _ in range(n)]
    +
    +
    +class TestAutoResetLoadsCleanContext:
    +    """#35809: after the gateway auto-resets a session because compression
    +    was exhausted, the NEXT turn must load an EMPTY transcript for the new
    +    session_id — never the bloated compressed-child transcript."""
    +
    +    def test_next_turn_transcript_is_empty_after_auto_reset(self, tmp_path):
    +        store = _make_store(tmp_path)
    +        source = _make_source()
    +
    +        entry = store.get_or_create_session(source)
    +        session_key = entry.session_key
    +        bloated_sid = entry.session_id
    +        store._db.create_session(
    +            session_id=bloated_sid, source="telegram", user_id="u1"
    +        )
    +        store._db.replace_messages(bloated_sid, _bloat(120))
    +        assert len(store.load_transcript(bloated_sid)) == 120  # precondition
    +
    +        new_entry = store.reset_session(session_key)
    +        assert new_entry is not None
    +        assert new_entry.session_id != bloated_sid
    +
    +        resolved = store.get_or_create_session(source)
    +        assert resolved.session_id == new_entry.session_id
    +        loaded = store.load_transcript(resolved.session_id)
    +
    +        assert loaded == [], (
    +            f"Auto-reset must yield an empty context, got {len(loaded)} "
    +            f"messages — the bloated compressed child leaked into the new session."
    +        )
    +        # The old transcript is still searchable, not destroyed.
    +        assert len(store.load_transcript(bloated_sid)) == 120
    +
    +    def test_clean_context_survives_gateway_restart(self, tmp_path):
    +        """The fresh, empty session must still be the one loaded after a
    +        gateway restart (sessions.json + state.db round-trip)."""
    +        store = _make_store(tmp_path)
    +        source = _make_source()
    +        entry = store.get_or_create_session(source)
    +        bloated_sid = entry.session_id
    +        store._db.create_session(
    +            session_id=bloated_sid, source="telegram", user_id="u1"
    +        )
    +        store._db.replace_messages(bloated_sid, _bloat(120))
    +
    +        new_entry = store.reset_session(entry.session_key)
    +        new_sid = new_entry.session_id
    +
    +        # Simulate restart: drop in-memory index, reload from disk.
    +        store._loaded = False
    +        store._entries.clear()
    +
    +        reloaded = store.get_or_create_session(source)
    +        assert reloaded.session_id == new_sid
    +        assert store.load_transcript(reloaded.session_id) == []