fix(compression): do not let prune rearm lock out over-threshold sessions

Message-only rearm could sit just above the body estimate while provider
prompt_tokens (system + tool schemas) already exceeded threshold_tokens,
so prune no-oped forever with no log. Bypass that rearm short-circuit on
the billed basis, warn once when over-threshold reclamation no-ops, and
name attempts_exhausted when should_compress_info says run but the loop
skips.

Fixes #101889

agent/context_compressor.py

@@ -3548,6 +3548,10 @@ class ContextCompressor(ContextEngine):
         # A committed prune is a prompt-cache boundary. Do not permit the next
         # one until the prompt has regrown the tokens just reclaimed.
         self._proactive_prune_rearm_tokens: int = 0
+        # Dedup key for the over-threshold "reclamation no-oped" warning
+        # (#101889) so a tool loop riding above the threshold warns once per
+        # distinct reason + rearm snapshot instead of every iteration.
+        self._last_reclaim_block_warn: "tuple[str, int] | None" = None
         self.min_tail_user_messages = min_tail_user_messages
         self.summary_target_ratio = max(0.10, min(summary_target_ratio, 0.80))
         self.quiet_mode = quiet_mode
@@ -4442,6 +4446,57 @@ class ContextCompressor(ContextEngine):
 
         return result, pruned
 
+    def _billed_basis_over_threshold(self, current_tokens: "int | None") -> bool:
+        """Whether a provider-billed reading says the session is over threshold.
+
+        ``current_tokens`` is the provider's ``prompt_tokens`` (or the
+        overhead-aware fallback estimate): it counts the system prompt and tool
+        schemas, which the message-only estimate behind
+        ``_proactive_prune_rearm_tokens`` does not. Used to stop schema
+        overhead from parking the prune rearm gate above a real request that is
+        already over ``threshold_tokens`` (#101889).
+        """
+        return (
+            current_tokens is not None
+            and self.threshold_tokens > 0
+            and current_tokens >= self.threshold_tokens
+        )
+
+    def _warn_reclamation_no_op(
+        self,
+        reason: str,
+        current_tokens: "int | None",
+        before: "int | None" = None,
+    ) -> None:
+        """Warn when an over-threshold session's reclamation path no-ops.
+
+        A session sitting above ``threshold_tokens`` with every reclamation
+        path declining is the failure mode from #101889: context keeps growing
+        until the provider's hard limit rejects the request, with nothing in
+        the log to explain it. Silent below the threshold (a declined prune
+        there is ordinary hysteresis, not a lockout). Deduped on
+        ``reason`` + the rearm snapshot so a busy tool loop logs once per
+        distinct state, not once per iteration; the key is cleared whenever a
+        prune commits so a later lockout warns again.
+        """
+        if not self._billed_basis_over_threshold(current_tokens):
+            return
+        key = (reason, int(self._proactive_prune_rearm_tokens))
+        if self._last_reclaim_block_warn == key:
+            return
+        self._last_reclaim_block_warn = key
+        logger.warning(
+            "Context is over the compression threshold (~%s of %s tokens) but "
+            "reclamation did not run: %s (message-token estimate %s, prune "
+            "rearm mark %s). The session may keep growing until the provider "
+            "rejects the request — /compact to compress history now.",
+            f"{int(current_tokens):,}",
+            f"{int(self.threshold_tokens):,}",
+            reason,
+            "n/a" if before is None else f"{int(before):,}",
+            f"{int(self._proactive_prune_rearm_tokens):,}",
+        )
+
     def prune_tool_results_only(
         self, messages: List[Dict[str, Any]], current_tokens: int | None = None,
     ) -> tuple[List[Dict[str, Any]], int]:
@@ -4483,6 +4538,13 @@ class ContextCompressor(ContextEngine):
         object is returned unchanged — the standard no-op caller contract
         (callers gate bookkeeping on ``result is not input``).
 
+        The rearm gate is measured on message bodies only, so it is bypassed
+        (never the reclaim gate) when a provider-billed ``current_tokens``
+        reading already puts the request over ``threshold_tokens``: schema
+        overhead must not park an over-threshold session below the rearm mark
+        forever with no reclamation and no log (#101889). Every no-op taken
+        while over threshold is logged once per distinct reason.
+
         Returns ``(messages, 0)`` — the input object — when disabled, below
         the trigger, or when the reclaim gate rejects the commit.
         """
@@ -4492,10 +4554,17 @@ class ContextCompressor(ContextEngine):
             return messages, 0
         # Nothing to reclaim until there are messages outside the protected tail.
         if len(messages) <= self.protect_last_n + self._protect_head_size(messages) + 1:
+            self._warn_reclamation_no_op("prune:tail_only", current_tokens)
             return messages, 0
         before = sum(_estimate_msg_budget_tokens(m) for m in messages)
         if before < self._proactive_prune_rearm_tokens:
-            return messages, 0
+            # Message-only estimate is short of the runway. Honour it as
+            # prompt-cache hysteresis only while the real (billed) request is
+            # still under threshold — above it, the lockout is the bug. The
+            # under-threshold skip stays silent on purpose: ordinary
+            # hysteresis, not a stuck session.
+            if not self._billed_basis_over_threshold(current_tokens):
+                return messages, 0
         # Capability gate BEFORE the expensive multi-pass scan: a bound store that
         # can't persist the prune atomically (duck-typed/plugin session store
         # without archive_and_compact) makes every prune a permanent no-op, so
@@ -4507,6 +4576,7 @@ class ContextCompressor(ContextEngine):
             and session_id
             and not callable(getattr(session_db, "archive_and_compact", None))
         ):
+            self._warn_reclamation_no_op("prune:store_cannot_persist", current_tokens)
             return messages, 0
         pruned_msgs, pruned_count = self._prune_old_tool_results(
             messages,
@@ -4517,6 +4587,7 @@ class ContextCompressor(ContextEngine):
         if not pruned_count:
             # Standard no-op contract: hand back the INPUT object so callers
             # can gate bookkeeping on `result is not input`.
+            self._warn_reclamation_no_op("prune:nothing_eligible", current_tokens)
             return messages, 0
         # Measured-savings gate (prompt-cache hysteresis): only commit when
         # the prune reclaims a meaningful batch of tokens. Estimated on the
@@ -4524,6 +4595,9 @@ class ContextCompressor(ContextEngine):
         after = sum(_estimate_msg_budget_tokens(m) for m in pruned_msgs)
         reclaimed = max(0, before - after)
         if reclaimed < self.proactive_prune_min_reclaim_tokens:
+            self._warn_reclamation_no_op(
+                "prune:reclaim_below_minimum", current_tokens, before=before
+            )
             return messages, 0
         # ``after`` includes the tool batch appended since the provider's last
         # usage reading, so both the low-water mark and future gate use the
@@ -4556,6 +4630,8 @@ class ContextCompressor(ContextEngine):
             # the micro-compaction sync (#98450) — one stamp site for the class.
             stamp_db_persisted_markers(pruned_msgs)
         self._proactive_prune_rearm_tokens = next_rearm_tokens
+        # Reclamation just ran: let a future lockout warn again.
+        self._last_reclaim_block_warn = None
         return pruned_msgs, pruned_count
 
     # ------------------------------------------------------------------

agent/conversation_loop.py

@@ -8275,9 +8275,19 @@ def run_conversation(
                     _info = getattr(_compressor, "should_compress_info", None)
                     if _info is not None:
                         try:
-                            _block_reason = _info(_real_tokens)[1]
+                            _should_now, _block_reason = _info(_real_tokens)
                         except Exception:
-                            _block_reason = None
+                            _should_now, _block_reason = False, None
+                        if _should_now and not _block_reason:
+                            # The engine says compression SHOULD run, yet this
+                            # branch was taken — the per-turn attempt budget is
+                            # spent. Over threshold with no reclamation left is
+                            # exactly the silent-lockout case, so name it
+                            # instead of dropping the (True, None) on the floor
+                            # (#101889).
+                            _block_reason = (
+                                f"attempts_exhausted:{compression_attempts}"
+                            )
                     if _block_reason:
                         agent._warn_context_overflow_blocked(
                             _block_reason,

tests/agent/test_proactive_prune_rearm_threshold.py

@@ -0,0 +1,189 @@
+"""Proactive-prune rearm must not lock out an over-threshold session (#101889).
+
+``_proactive_prune_rearm_tokens`` is armed from a message-bodies-only estimate,
+but the provider bills the system prompt and tool schemas too. On a schema-heavy
+session the message-only estimate can sit permanently just below the rearm mark
+while the real request rides *above* ``threshold_tokens`` — the prune declines
+every iteration, full compression never gets there, and nothing is logged. The
+session then grows until the provider rejects the request.
+
+Pinned here as invariants (no frozen config literals): the gates are evaluated
+against this compressor's own ``threshold_tokens`` / ``proactive_prune_tokens``.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List
+from unittest.mock import patch
+
+from agent.context_compressor import ContextCompressor, _estimate_msg_budget_tokens
+
+LARGE_WINDOW = 1_000_000
+
+
+def _compressor(**kw: Any) -> ContextCompressor:
+    defaults = dict(
+        model="test",
+        quiet_mode=True,
+        threshold_percent=0.50,
+        protect_first_n=2,
+        protect_last_n=4,
+        proactive_prune_tokens=48_000,
+        proactive_prune_min_result_chars=8_000,
+    )
+    defaults.update(kw)
+    with patch(
+        "agent.context_compressor.get_model_context_length",
+        return_value=LARGE_WINDOW,
+    ):
+        return ContextCompressor(**defaults)
+
+
+def _history(n_pairs: int = 8, big: int = 9_000) -> List[Dict[str, Any]]:
+    msgs: List[Dict[str, Any]] = [{"role": "system", "content": "sys"}]
+    for i in range(n_pairs):
+        cid = f"call_{i}"
+        msgs.append({
+            "role": "assistant",
+            "content": "",
+            "tool_calls": [{
+                "id": cid,
+                "type": "function",
+                "function": {"name": "terminal", "arguments": '{"cmd":"ls"}'},
+            }],
+        })
+        msgs.append({
+            "role": "tool",
+            "tool_call_id": cid,
+            "content": chr(65 + i) * big if i < 3 else "ok",
+        })
+    return msgs
+
+
+def _park_rearm_just_above_messages(
+    compressor: ContextCompressor, messages: List[Dict[str, Any]]
+) -> int:
+    """Reproduce the reporter's state: message-only estimate stuck 913 tokens
+    below the rearm mark (schema overhead makes up the rest of the request)."""
+    before = sum(_estimate_msg_budget_tokens(m) for m in messages)
+    compressor._proactive_prune_rearm_tokens = before + 913
+    assert before < compressor._proactive_prune_rearm_tokens
+    return before
+
+
+def test_billed_basis_over_threshold_defeats_message_only_rearm_lockout() -> None:
+    """Over ``threshold_tokens`` on the provider-billed basis, the rearm gate
+    must not short-circuit the prune on the message-only estimate alone."""
+    c = _compressor()
+    msgs = _history()
+    _park_rearm_just_above_messages(c, msgs)
+    billed = c.threshold_tokens + 1  # provider says: over threshold, now
+
+    scans: List[int] = []
+    # Stand in for the real multi-pass scan: a NEW list whose old tool outputs
+    # are reclaimed, so the (untouched) reclaim gate can commit it.
+    reclaimed = [dict(m) for m in msgs]
+    for m in reclaimed[:-2]:
+        if m.get("role") == "tool":
+            m["content"] = "[pruned]"
+
+    def _scan(*args: Any, **kwargs: Any) -> tuple[List[Dict[str, Any]], int]:
+        scans.append(1)
+        return reclaimed, 3
+
+    with patch.object(c, "_prune_old_tool_results", _scan):
+        result, pruned = c.prune_tool_results_only(msgs, current_tokens=billed)
+
+    assert scans, "rearm gate short-circuited on the message-only estimate"
+    assert pruned == 3
+    assert result is not msgs
+
+
+def test_message_only_rearm_still_holds_below_threshold() -> None:
+    """Prompt-cache hysteresis is intact while the real request is under the
+    compression threshold — the rearm bypass is an overflow escape hatch only."""
+    c = _compressor()
+    msgs = _history()
+    _park_rearm_just_above_messages(c, msgs)
+    under = c.threshold_tokens - 1
+    assert under >= c.proactive_prune_tokens  # above the prune trigger
+
+    with patch.object(
+        c,
+        "_prune_old_tool_results",
+        side_effect=AssertionError("scan must not run below threshold"),
+    ):
+        result, pruned = c.prune_tool_results_only(msgs, current_tokens=under)
+
+    assert result is msgs
+    assert pruned == 0
+
+
+def test_no_op_below_the_prune_trigger() -> None:
+    """Under ``proactive_prune_tokens`` nothing is reclaimed, rearm or not —
+    the bypass must not turn into over-pruning of small sessions."""
+    c = _compressor()
+    msgs = _history()
+    c._proactive_prune_rearm_tokens = 0  # fully rearmed; only the trigger gates
+
+    with patch.object(
+        c,
+        "_prune_old_tool_results",
+        side_effect=AssertionError("scan must not run below the trigger"),
+    ):
+        result, pruned = c.prune_tool_results_only(
+            msgs, current_tokens=c.proactive_prune_tokens - 1
+        )
+
+    assert result is msgs
+    assert pruned == 0
+
+
+def test_over_threshold_reclamation_no_op_warns_once(caplog) -> None:
+    """A session riding above the threshold with every reclamation path
+    declining must be distinguishable in the log — and must not spam the same
+    reason on every tool iteration."""
+    # Reclaim floor above anything this transcript can free: the scan runs,
+    # finds candidates, and the commit gate rejects it — a silent no-op today.
+    c = _compressor(proactive_prune_min_reclaim_tokens=10_000_000)
+    msgs = _history()
+    billed = c.threshold_tokens + 5_000
+
+    with caplog.at_level(logging.WARNING, logger="agent.context_compressor"):
+        result, pruned = c.prune_tool_results_only(msgs, current_tokens=billed)
+    assert (result, pruned) == (msgs, 0)
+
+    warnings = [
+        r for r in caplog.records
+        if r.levelno >= logging.WARNING
+        and "over the compression threshold" in r.getMessage()
+    ]
+    assert warnings, "over-threshold reclamation no-op was silent"
+
+    # Same state on the next tool iteration: deduped, not re-logged.
+    with caplog.at_level(logging.WARNING, logger="agent.context_compressor"):
+        c.prune_tool_results_only(msgs, current_tokens=billed)
+    assert len([
+        r for r in caplog.records
+        if r.levelno >= logging.WARNING
+        and "over the compression threshold" in r.getMessage()
+    ]) == len(warnings)
+
+
+def test_under_threshold_no_op_is_not_warned(caplog) -> None:
+    """Ordinary hysteresis below the threshold stays quiet."""
+    c = _compressor(proactive_prune_min_reclaim_tokens=10_000_000)
+    msgs = _history()
+
+    with caplog.at_level(logging.WARNING, logger="agent.context_compressor"):
+        result, pruned = c.prune_tool_results_only(
+            msgs, current_tokens=c.threshold_tokens - 1
+        )
+
+    assert (result, pruned) == (msgs, 0)
+    assert not [
+        r for r in caplog.records
+        if r.levelno >= logging.WARNING
+        and "over the compression threshold" in r.getMessage()
+    ]

tests/agent/test_proactive_tool_result_pruning.py

@@ -130,7 +130,12 @@ def test_rearms_only_after_reclaimed_token_runway():
         _tool_msg("call_9", "ok"),
     ]
     assert sum(map(_estimate_msg_budget_tokens, grown)) < rearm_tokens
-    blocked, n2 = c.prune_tool_results_only(grown, current_tokens=1_000_000)
+    # Below the full-compression threshold, where the runway is pure
+    # prompt-cache hysteresis. (Above it the runway is bypassed on the
+    # provider-billed reading instead — see
+    # tests/agent/test_proactive_prune_rearm_threshold.py, #101889.)
+    _under_threshold = c.threshold_tokens - 1
+    blocked, n2 = c.prune_tool_results_only(grown, current_tokens=_under_threshold)
     assert n2 == 0
     assert blocked is grown
     assert len(_tool_by_id(blocked, "call_6")["content"]) == 9000
@@ -139,7 +144,7 @@ def test_rearms_only_after_reclaimed_token_runway():
     missing = rearm_tokens - sum(map(_estimate_msg_budget_tokens, grown))
     regrown = grown + [{"role": "user", "content": "x" * (missing * 4)}]
     assert sum(map(_estimate_msg_budget_tokens, regrown)) >= rearm_tokens
-    rearmed, n3 = c.prune_tool_results_only(regrown, current_tokens=1_000_000)
+    rearmed, n3 = c.prune_tool_results_only(regrown, current_tokens=_under_threshold)
     assert n3 >= 2
     assert rearmed is not regrown
 

tests/run_agent/test_proactive_prune_loop_wiring.py

@@ -189,6 +189,27 @@ class TestProactivePruneLoopWiring:
         assert tool_rows, "expected tool rows in the final transcript"
         assert all(m["content"] == marker for m in tool_rows)
 
+    def test_should_compress_true_but_skipped_is_warned(self, agent):
+        """``should_compress_info`` says RUN (``(True, None)``) yet this branch
+        was taken — the per-turn compression budget is spent. Over threshold
+        with no reclamation running must not be swallowed silently (#101889)."""
+        agent.context_compressor.should_compress.return_value = False
+        agent.context_compressor.should_compress_info.return_value = (True, None)
+        agent.context_compressor.prune_tool_results_only = (
+            lambda messages, current_tokens=None: (messages, 0)
+        )
+        warned = []
+        with patch.object(
+            agent,
+            "_warn_context_overflow_blocked",
+            side_effect=lambda reason, tokens, threshold: warned.append(reason),
+        ):
+            result = _run_tool_loop(agent, n_tool_iterations=1)
+
+        assert result["completed"] is True
+        assert warned, "over-threshold turn with no compaction ran silently"
+        assert all(r.startswith("attempts_exhausted") for r in warned)
+
     def test_noop_input_object_commits_nothing(self, agent):
         """Engine returns the INPUT object with a (bogus) non-zero count —
         the caller's ``result is not input`` gate must refuse the commit."""