diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 1dba861fa4e..214ed013f57 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -113,6 +113,15 @@ def _record_codex_app_server_usage(agent, turn) -> dict[str, Any]: usage = getattr(turn, "token_usage_last", None) if not isinstance(usage, dict) or not usage: + compressor = getattr(agent, "context_compressor", None) + if ( + compressor is not None + and getattr(compressor, "awaiting_real_usage_after_compression", False) + ): + # No usage means this turn cannot adjudicate the pending compaction. + # Consume the marker so a later unrelated reading is not charged to + # it and preflight deferral cannot stay latched indefinitely. + compressor.update_from_response({}) if agent._session_db and agent.session_id: try: if not agent._session_db_created: diff --git a/agent/context_compressor.py b/agent/context_compressor.py index ed5fc646a42..d8f661987b9 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -737,6 +737,7 @@ class ContextCompressor(ContextEngine): self._last_compression_savings_pct = 100.0 self._ineffective_compression_count = 0 self._verify_compaction_cleared_threshold = False + self._last_compression_made_progress = False self._summary_failure_cooldown_until = 0.0 # transient errors must not block a fresh session self._last_summary_error = None self._last_compress_aborted = False @@ -773,6 +774,7 @@ class ContextCompressor(ContextEngine): self._last_compression_savings_pct = 100.0 self._ineffective_compression_count = 0 self._verify_compaction_cleared_threshold = False + self._last_compression_made_progress = False self._summary_failure_cooldown_until = 0.0 self._last_compress_aborted = False self._context_probed = False @@ -947,6 +949,7 @@ class ContextCompressor(ContextEngine): self.awaiting_real_usage_after_compression = False self._ineffective_compression_count = 0 self._verify_compaction_cleared_threshold = False + self._last_compression_made_progress = False # When the MINIMUM_CONTEXT_LENGTH floor meets/exceeds a small context # window, compacting at the percentage (50% → 32K of a 64K window) wastes @@ -1134,9 +1137,12 @@ class ContextCompressor(ContextEngine): # Anti-thrashing: track whether last compression was effective self._last_compression_savings_pct: float = 100.0 self._ineffective_compression_count: int = 0 - # Set by compress(); consumed by should_compress() to check the next - # real reading against the threshold. See should_compress(). + # Set after a completed compression boundary; consumed by the next + # provider-reported prompt count in update_from_response(). self._verify_compaction_cleared_threshold: bool = False + # Lets the boundary wrapper distinguish a completed rewrite from a + # no-op/abort without inferring progress from message-list length. + self._last_compression_made_progress: bool = False self._summary_failure_cooldown_until: float = 0.0 self._last_summary_error: Optional[str] = None # When summary generation fails and a static fallback is inserted, @@ -1183,6 +1189,10 @@ class ContextCompressor(ContextEngine): if self.last_prompt_tokens < self.threshold_tokens: if self.awaiting_real_usage_after_compression and self.last_compression_rough_tokens > 0: self.last_rough_tokens_when_real_prompt_fit = self.last_compression_rough_tokens + # Any real provider reading below the trigger proves the prompt + # fits again. Clear the episode latch even when this response was + # not the one immediately following compaction. + self._ineffective_compression_count = 0 else: self.last_rough_tokens_when_real_prompt_fit = 0 @@ -1296,9 +1306,9 @@ class ContextCompressor(ContextEngine): if self._ineffective_compression_count >= 2: if not self.quiet_mode: logger.warning( - "Compression skipped — last %d compressions saved <10%% each. " - "Consider /new to start a fresh session, or /compress " - "for focused compression.", + "Compression skipped — last %d compaction attempts did not " + "restore enough context headroom. Consider /new to start a " + "fresh session, or /compress for focused compression.", self._ineffective_compression_count, ) return False @@ -2862,6 +2872,7 @@ This compaction should PRIORITISE preserving all information related to the focu self._last_aux_model_failure_error = None self._last_aux_model_failure_model = None self._last_compress_aborted = False + self._last_compression_made_progress = False # NOTE: do NOT reset _last_summary_auth_failure or # _last_summary_network_failure here. These flags are set by # _generate_summary() on a terminal failure and are already cleared on @@ -3207,11 +3218,11 @@ This compaction should PRIORITISE preserving all information related to the focu savings_pct = (saved_estimate / pre_estimate * 100) if pre_estimate > 0 else 0 self._last_compression_savings_pct = savings_pct - # Only ever INCREMENT here. The reset lives in should_compress(), which - # is the one place that sees the same measure the trigger uses; see - # ``_verify_compaction_cleared_threshold``. - if savings_pct < 10: - self._ineffective_compression_count += 1 + # Message-only savings are diagnostic. The anti-thrashing verdict is + # owned by the next provider-reported prompt count, which answers the + # actual question: did this completed boundary get under the threshold? + # Counting a low message-savings estimate here as well would give one + # compaction two strikes when that real reading remains over threshold. if not self.quiet_mode: logger.info( @@ -3228,5 +3239,6 @@ This compaction should PRIORITISE preserving all information related to the focu # are positional; this single terminal sweep makes it structural so a # future copy site cannot re-leak the marker into the child-session flush. _strip_persistence_markers(compressed) + self._last_compression_made_progress = True return compressed diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 886e04795c4..22decb591a2 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -673,6 +673,20 @@ def compress_context( finally: _release_lock() + # A compressor that returns the exact input object made no structural + # progress. Do not rotate/rewrite the session or arm post-compression + # deferral in that case; its own anti-thrash counter records the no-op. + if compressed is messages: + logger.info( + "Compression made no progress (session=%s) — skipping boundary rewrite.", + agent.session_id or "none", + ) + _existing_sp = getattr(agent, "_cached_system_prompt", None) + if not _existing_sp: + _existing_sp = agent._build_system_prompt(system_message) + _release_lock() + return messages, _existing_sp + try: summary_error = getattr(agent.context_compressor, "_last_summary_error", None) if summary_error: @@ -961,11 +975,11 @@ def compress_context( agent.context_compressor.last_prompt_tokens = -1 agent.context_compressor.last_completion_tokens = 0 agent.context_compressor.awaiting_real_usage_after_compression = True - # Arm the effectiveness verdict only after the full compaction boundary - # succeeds. Exceptions and aborted/no-op attempts return before here, so - # unrelated later usage cannot be charged to an attempt that never took - # effect. External engines without the private flag stay untouched. - if hasattr(agent.context_compressor, "_verify_compaction_cleared_threshold"): + # Arm the effectiveness verdict only after a completed rewrite crosses + # the full compaction boundary. Exceptions, aborts, and no-op attempts + # leave this false, so unrelated later usage cannot be charged to an + # attempt that never changed the transcript. + if getattr(agent.context_compressor, "_last_compression_made_progress", False): agent.context_compressor._verify_compaction_cleared_threshold = True # Clear the file-read dedup cache. After compression the original diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index f3b2bc0d39e..75c07540d88 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -2119,7 +2119,19 @@ def run_conversation( "reasoning_tokens": canonical_usage.reasoning_tokens, } agent.context_compressor.update_from_response(usage_dict) + elif getattr( + agent.context_compressor, + "awaiting_real_usage_after_compression", + False, + ): + # A response with no usage cannot adjudicate whether the + # prior compaction cleared the threshold. Consume the pending + # verdict now so a much later, unrelated reading is not + # charged to that old compaction, and so preflight deferral + # does not remain latched indefinitely. + agent.context_compressor.update_from_response({}) + if hasattr(response, 'usage') and response.usage: # Cache discovered context length after successful call. # Only persist limits confirmed by the provider (parsed # from the error message), not guessed probe tiers. diff --git a/tests/agent/test_compaction_anti_thrash.py b/tests/agent/test_compaction_anti_thrash.py index cd9c6d46220..2eb4ea66108 100644 --- a/tests/agent/test_compaction_anti_thrash.py +++ b/tests/agent/test_compaction_anti_thrash.py @@ -162,6 +162,7 @@ class TestFutilityGuard: assert len(out) < before, "a long transcript must still compact" assert cc._last_compression_savings_pct > 10 + assert cc._last_compression_made_progress is True assert cc._ineffective_compression_count == 0 def test_no_false_positive_under_tokenizer_skew(self): @@ -189,6 +190,26 @@ class TestFutilityGuard: "tokenizer skew must not be mistaken for an incompressible floor" ) + def test_latched_counter_resets_after_any_real_prompt_fits(self): + cc = _compressor(threshold_tokens=24_576) + cc._ineffective_compression_count = 2 + + cc.update_from_response({"prompt_tokens": 20_000}) + + assert cc._ineffective_compression_count == 0 + assert cc.should_compress(33_564) + + def test_usage_less_response_consumes_pending_verdict(self): + cc = _compressor(threshold_tokens=24_576) + cc._verify_compaction_cleared_threshold = True + cc.awaiting_real_usage_after_compression = True + + cc.update_from_response({}) + + assert cc._verify_compaction_cleared_threshold is False + assert cc.awaiting_real_usage_after_compression is False + assert cc._ineffective_compression_count == 0 + def test_a_failed_pass_records_exactly_one_strike(self): """A compaction that leaves the real prompt over the threshold: one strike. @@ -226,4 +247,5 @@ class TestMinimumMessagesBranch: out = cc.compress(msgs, current_tokens=100_000) assert len(out) == len(msgs), "nothing should have been compressed" + assert cc._last_compression_made_progress is False assert cc._ineffective_compression_count == before + 1