diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 705cf0fdfe00..490cb1b51fdc 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -889,6 +889,7 @@ class ContextCompressor(ContextEngine): 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._cooldown_persist_failed = False self._last_summary_error = None self._last_compress_aborted = False self.last_real_prompt_tokens = 0 @@ -928,6 +929,7 @@ class ContextCompressor(ContextEngine): self._verify_compaction_cleared_threshold = False self._last_compression_made_progress = False self._summary_failure_cooldown_until = 0.0 + self._cooldown_persist_failed = False self._last_compress_aborted = False self._context_probed = False self._context_probe_persistable = False @@ -941,6 +943,7 @@ class ContextCompressor(ContextEngine): self._session_db = session_db self._session_id = session_id or "" self._summary_failure_cooldown_until = 0.0 + self._cooldown_persist_failed = False self._last_summary_error = None self._consecutive_timeout_failures = 0 self._fallback_compression_streak = 0 @@ -1058,6 +1061,14 @@ class ContextCompressor(ContextEngine): return local_state if not state: if refresh: + if local_state is not None and self._cooldown_persist_failed: + # The live local cooldown never made it to the DB (persist + # failed), so the empty row is not evidence that another + # agent cleared it. Honouring the DB here would re-enable + # auto-compress mid-cooldown and reopen the #11529 thrash + # window. Keep the local timer authoritative until it + # expires or a successful DB read supersedes it. + return local_state self._summary_failure_cooldown_until = 0.0 self._last_summary_error = None return None @@ -1065,12 +1076,15 @@ class ContextCompressor(ContextEngine): remaining_seconds = float(state.get("remaining_seconds") or 0.0) if remaining_seconds <= 0: if refresh: + if local_state is not None and self._cooldown_persist_failed: + return local_state self._summary_failure_cooldown_until = 0.0 self._last_summary_error = None return None self._summary_failure_cooldown_until = now_mono + remaining_seconds self._last_summary_error = state.get("error") + self._cooldown_persist_failed = False return { "cooldown_until": float(state.get("cooldown_until") or 0.0), "remaining_seconds": remaining_seconds, @@ -1093,18 +1107,23 @@ class ContextCompressor(ContextEngine): recorder = getattr(session_db, "record_compression_failure_cooldown", None) if recorder is None: + self._cooldown_persist_failed = True return try: recorder(session_id, cooldown_until, error) + self._cooldown_persist_failed = False except sqlite3.Error as exc: + self._cooldown_persist_failed = True logger.debug("compression failure cooldown persist failed: %s", exc) except Exception as exc: + self._cooldown_persist_failed = True logger.debug("compression failure cooldown persist failed (non-sqlite): %s", exc) def _clear_compression_failure_cooldown(self) -> None: self._summary_failure_cooldown_until = 0.0 self._last_summary_error = None self._consecutive_timeout_failures = 0 + self._cooldown_persist_failed = False session_db = getattr(self, "_session_db", None) session_id = getattr(self, "_session_id", "") @@ -1398,6 +1417,10 @@ class ContextCompressor(ContextEngine): # 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 + # True while the live local cooldown failed to persist to the DB; + # a refresh must then treat an empty durable row as unknown, not + # cleared (see get_active_compression_failure_cooldown). + self._cooldown_persist_failed: bool = False self._last_summary_error: Optional[str] = None # When summary generation fails and a static fallback is inserted, # record how many turns were unrecoverably dropped so callers @@ -1543,8 +1566,38 @@ class ContextCompressor(ContextEngine): return False return not self._automatic_compression_blocked() + def _refresh_durable_guards(self) -> None: + """Re-read durable cooldown + fallback-streak state from the DB. + + Cheap, best-effort, and only called when a gate is about to say + "blocked": another agent on the same session may have cleared the + durable rows (successful boundary, forced retry) after this + compressor was bound, and a fallback streak has no timer — without + a re-read the stale in-memory snapshot blocks forever. + """ + try: + self.get_active_compression_failure_cooldown(refresh=True) + except Exception as exc: + logger.debug("compression cooldown refresh failed: %s", exc) + try: + self._load_fallback_compression_streak() + except Exception as exc: + logger.debug("compression fallback-streak refresh failed: %s", exc) + def _automatic_compression_blocked(self) -> bool: """Return whether automatic compaction is in cooldown or tripped.""" + if not self._automatic_compression_blocked_locally(): + return False + # Blocked on the in-memory snapshot. Durable guard rows may have + # been cleared by another agent since bind_session_state(); refresh + # and re-evaluate so a stale local block cannot outlive the durable + # state that justified it. The unblocked hot path above never pays + # for the DB reads. + self._refresh_durable_guards() + return self._automatic_compression_blocked_locally() + + def _automatic_compression_blocked_locally(self) -> bool: + """Evaluate the automatic-compaction gate on in-memory state only.""" # Do not trigger compression while the summary LLM is in cooldown. # On a 429/transient failure _generate_summary() sets a cooldown and # returns None; compress() then inserts a static fallback marker and diff --git a/tests/agent/test_compression_rotation_state.py b/tests/agent/test_compression_rotation_state.py index 07e714aca31c..7eb46afb09c6 100644 --- a/tests/agent/test_compression_rotation_state.py +++ b/tests/agent/test_compression_rotation_state.py @@ -494,3 +494,86 @@ class TestAutomaticCompressionStateRefreshAfterLock: force=True, ) assert db.get_compression_lock_holder(session_id) is None + + +class TestGateLevelGuardRefresh: + """The unblock direction must work from the should_compress() pre-gates. + + compress_context refreshes durable guards internally, but the automatic + paths (preflight/turn gates) consult should_compress() first — if a stale + in-memory fallback streak (which has no expiry timer) blocks there, the + refresh inside compress_context is never reached and the agent stays + blocked forever. + """ + + def test_should_compress_unblocks_after_another_agent_clears_streak( + self, + refresh_state_db: SessionDB, + ): + db = refresh_state_db + session_id = "GATE_LEVEL_STREAK_CLEAR" + db.create_session(session_id, source="telegram") + db.set_compression_fallback_streak(session_id, 2) + compressor = _bound_context_compressor(db, session_id) + assert compressor._fallback_compression_streak == 2 + + # Another agent's healthy boundary clears the durable breaker. + db.set_compression_fallback_streak(session_id, 0) + + assert compressor.should_compress(10**9) is True + assert compressor._fallback_compression_streak == 0 + + def test_unblocked_gate_does_not_touch_the_db( + self, + refresh_state_db: SessionDB, + ): + db = refresh_state_db + session_id = "GATE_LEVEL_HOT_PATH" + db.create_session(session_id, source="telegram") + compressor = _bound_context_compressor(db, session_id) + + with patch.object( + compressor, + "_refresh_durable_guards", + side_effect=AssertionError("hot path must not refresh"), + ): + assert compressor._automatic_compression_blocked() is False + + +class TestCooldownPersistFailureIsNotAClearedRow: + def test_refresh_keeps_local_cooldown_when_persist_failed( + self, + refresh_state_db: SessionDB, + ): + """An empty durable row is not evidence of a clear when OUR write failed. + + _record_compression_failure_cooldown sets the local timer first and + persists best-effort. If that persist failed, a later refresh=True + finding no DB row must keep the local cooldown (otherwise the #11529 + thrash guard silently re-opens), until it expires or a successful + DB round-trip supersedes it. + """ + db = refresh_state_db + session_id = "PERSIST_FAILED_COOLDOWN" + db.create_session(session_id, source="telegram") + compressor = _bound_context_compressor(db, session_id) + + with patch.object( + db, + "record_compression_failure_cooldown", + side_effect=Exception("disk full"), + ): + compressor._record_compression_failure_cooldown(60, "rate limited") + assert compressor._cooldown_persist_failed is True + + state = compressor.get_active_compression_failure_cooldown(refresh=True) + assert state is not None + assert compressor._summary_failure_cooldown_until > 0 + assert compressor._automatic_compression_blocked() is True + + # Once a durable round-trip succeeds, the DB is authoritative again. + compressor._record_compression_failure_cooldown(30, "retry later") + assert compressor._cooldown_persist_failed is False + db.clear_compression_failure_cooldown(session_id) + assert compressor.get_active_compression_failure_cooldown(refresh=True) is None + assert compressor._summary_failure_cooldown_until == 0.0