fix(compression): close unblock-direction gaps in durable guard refresh

Follow-up on the salvaged #64511 commit:

- _automatic_compression_blocked() now refreshes durable guard state
  (cooldown + fallback streak) when — and only when — the in-memory
  snapshot says blocked, then re-evaluates. The should_compress()
  pre-gates (preflight/turn paths) consult this before ever reaching
  compress_context, and a stale fallback streak has no expiry timer, so
  without a gate-level refresh a cleared durable row could never unblock
  a prebound agent. The unblocked hot path pays no DB reads.
- A refresh that finds no durable cooldown row no longer clears a live
  local cooldown whose DB persist FAILED (_cooldown_persist_failed):
  an empty row is not evidence another agent cleared it, and honouring
  it would reopen the #11529 thrash window. A successful durable
  round-trip (record or read) makes the DB authoritative again.
- Guard tests for both directions (red on the pre-fix code), including
  a hot-path test asserting the unblocked gate never touches the DB.

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

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