diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 955e648917..4a592cf2ee 100644 --- a/agent/context_compressor.py +++ b/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 # ------------------------------------------------------------------ diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 77e502ee48..50d49be442 100644 --- a/agent/conversation_loop.py +++ b/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, diff --git a/tests/agent/test_proactive_prune_rearm_threshold.py b/tests/agent/test_proactive_prune_rearm_threshold.py new file mode 100644 index 0000000000..2e3c0766d4 --- /dev/null +++ b/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() + ] diff --git a/tests/agent/test_proactive_tool_result_pruning.py b/tests/agent/test_proactive_tool_result_pruning.py index bbb01e4b16..2a963f7f01 100644 --- a/tests/agent/test_proactive_tool_result_pruning.py +++ b/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 diff --git a/tests/run_agent/test_proactive_prune_loop_wiring.py b/tests/run_agent/test_proactive_prune_loop_wiring.py index 957799f466..486865ad90 100644 --- a/tests/run_agent/test_proactive_prune_loop_wiring.py +++ b/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."""