fix(compression): persist anti-thrash state across process restarts (#69872)

The anti-thrash guard (_ineffective_compression_count) was in-memory
only: a fresh compressor bound to a resumed, already-compacted session
started with compression_count=0 and a disarmed guard, so a
near-threshold session could legally re-compact once per process
restart, forever.

Persist the counter through the durable session-state channel,
mirroring the failure-cooldown (#54465) and fallback-streak (af7dceaf7)
pattern:

- hermes_state.py: sessions.compression_ineffective_count column
  (declarative reconciliation adds it on existing DBs) +
  get/set_compression_ineffective_count accessors.
- context_compressor.py: every strike/clear verdict routes through
  _record_ineffective_compression_verdict() which writes through to the
  session row (no-change verdicts skip the DB write);
  bind_session_state() loads the persisted value; the compression
  rotation boundary carries the counter onto the child row;
  update_model()'s reset also clears the durable copy; the
  ineffective-only fast path in _automatic_compression_blocked() is
  removed because the counter is now durable and another agent's clear
  must unblock a stale local snapshot.
- conversation_compression.py: _refresh_persisted_compression_guards
  re-reads the counter alongside cooldown + fallback streak.

Reset semantics are unchanged: any real provider reading below the
threshold still clears the counter — and now clears it durably too.

Resolves the residual gap identified in #54923 by @lanyusea (the
second-threshold mechanism was superseded by persisting the existing
guard state).

Co-authored-by: lanyusea <lanyusea@gmail.com>

agent/context_compressor.py

@@ -1134,8 +1134,10 @@ class ContextCompressor(ContextEngine):
         self._last_summary_error = None
         self._consecutive_timeout_failures = 0
         self._fallback_compression_streak = 0
+        self._ineffective_compression_count = 0
         self.get_active_compression_failure_cooldown()
         self._load_fallback_compression_streak()
+        self._load_ineffective_compression_count()
 
     def on_session_start(self, session_id: str, **kwargs) -> None:
         """Bind session-scoped compression state for a new or resumed session."""
@@ -1144,6 +1146,7 @@ class ContextCompressor(ContextEngine):
         old_session_id = kwargs.get("old_session_id")
         session_db = kwargs.get("session_db", getattr(self, "_session_db", None))
         previous_fallback_streak = self._fallback_compression_streak
+        previous_ineffective_count = self._ineffective_compression_count
         if boundary_reason == "compression" and old_session_id:
             getter = getattr(session_db, "get_compression_fallback_streak", None)
             if callable(getter):
@@ -1158,12 +1161,37 @@ class ContextCompressor(ContextEngine):
                         "compression parent fallback streak lookup failed (non-sqlite): %s",
                         exc,
                     )
+            count_getter = getattr(
+                session_db, "get_compression_ineffective_count", None,
+            )
+            if callable(count_getter):
+                try:
+                    stored_count = count_getter(old_session_id)
+                    if isinstance(stored_count, (int, float, str)):
+                        previous_ineffective_count = max(0, int(stored_count))
+                except (TypeError, ValueError, sqlite3.Error) as exc:
+                    logger.debug(
+                        "compression parent ineffective count lookup failed: %s", exc,
+                    )
+                except Exception as exc:
+                    logger.debug(
+                        "compression parent ineffective count lookup failed (non-sqlite): %s",
+                        exc,
+                    )
         self.bind_session_state(session_db, session_id)
         if boundary_reason == "compression":
             # Rotation creates a fresh child row before this callback. Preserve
             # the logical conversation's streak until boundary bookkeeping
             # persists the updated value onto the child row.
             self._fallback_compression_streak = previous_fallback_streak
+            # Same for the anti-thrash strike counter — but unlike the streak,
+            # no later boundary bookkeeping writes it, so persist the carried
+            # value onto the (fresh) child row now. Otherwise a restart between
+            # rotation and the next real-usage verdict would silently disarm
+            # an armed guard (#54923).
+            if self._ineffective_compression_count != previous_ineffective_count:
+                self._ineffective_compression_count = previous_ineffective_count
+                self._persist_ineffective_compression_count()
 
     def _load_fallback_compression_streak(self) -> None:
         session_db = getattr(self, "_session_db", None)
@@ -1197,6 +1225,59 @@ class ContextCompressor(ContextEngine):
         except Exception as exc:
             logger.debug("compression fallback streak persist failed (non-sqlite): %s", exc)
 
+    def _load_ineffective_compression_count(self) -> None:
+        """Load the durable anti-thrash strike count for the bound session.
+
+        A fresh compressor on a resumed session starts with
+        ``compression_count == 0`` and, historically, an in-memory-only
+        ineffective counter — so a guard armed (1 strike) or tripped
+        (2 strikes) before a process restart silently disarmed, and a
+        near-threshold session could re-compact once per restart forever
+        (#54923). The counter now round-trips through the session row like
+        the failure cooldown and the fallback streak.
+        """
+        session_db = getattr(self, "_session_db", None)
+        session_id = getattr(self, "_session_id", "")
+        getter = getattr(session_db, "get_compression_ineffective_count", None)
+        if not session_id or not callable(getter):
+            return
+        try:
+            stored_count = getter(session_id)
+            self._ineffective_compression_count = max(
+                0,
+                int(stored_count)
+                if isinstance(stored_count, (int, float, str))
+                else 0,
+            )
+        except (TypeError, ValueError, sqlite3.Error) as exc:
+            logger.debug("compression ineffective count lookup failed: %s", exc)
+        except Exception as exc:
+            logger.debug("compression ineffective count lookup failed (non-sqlite): %s", exc)
+
+    def _persist_ineffective_compression_count(self) -> None:
+        session_db = getattr(self, "_session_db", None)
+        session_id = getattr(self, "_session_id", "")
+        setter = getattr(session_db, "set_compression_ineffective_count", None)
+        if not session_id or not callable(setter):
+            return
+        try:
+            setter(session_id, self._ineffective_compression_count)
+        except sqlite3.Error as exc:
+            logger.debug("compression ineffective count persist failed: %s", exc)
+        except Exception as exc:
+            logger.debug("compression ineffective count persist failed (non-sqlite): %s", exc)
+
+    def _record_ineffective_compression_verdict(self, count: int) -> None:
+        """Set the anti-thrash strike counter, keeping the durable copy in sync.
+
+        Persists only on change so the reset issued by every ordinary fitting
+        response (already-zero -> zero) never costs a DB write.
+        """
+        if count == self._ineffective_compression_count:
+            return
+        self._ineffective_compression_count = count
+        self._persist_ineffective_compression_count()
+
     def record_completed_compaction(self, *, used_fallback: bool = False) -> None:
         """Record one completed boundary and its summary quality."""
         self._verify_compaction_cleared_threshold = True
@@ -1405,7 +1486,10 @@ class ContextCompressor(ContextEngine):
         self.last_rough_tokens_when_real_prompt_fit = 0
         self.last_compression_rough_tokens = 0
         self.awaiting_real_usage_after_compression = False
-        self._ineffective_compression_count = 0
+        # Strikes were judged against the PREVIOUS threshold; a recomputed
+        # trigger invalidates them. Keep the durable copy in sync so a
+        # restart doesn't resurrect strikes this recalibration just voided.
+        self._record_ineffective_compression_verdict(0)
         if runtime_changed:
             self._fallback_compression_streak = 0
             self._persist_fallback_compression_streak()
@@ -1729,7 +1813,7 @@ class ContextCompressor(ContextEngine):
                 # when this response was not immediately after compaction. The
                 # independent fallback streak is boundary-scoped and survives
                 # ordinary fitting responses during context regrowth.
-                self._ineffective_compression_count = 0
+                self._record_ineffective_compression_verdict(0)
             else:
                 self.last_rough_tokens_when_real_prompt_fit = 0
 
@@ -1751,7 +1835,9 @@ class ContextCompressor(ContextEngine):
             # per compaction.
             if self._verify_compaction_cleared_threshold:
                 if self.last_prompt_tokens >= self.threshold_tokens:
-                    self._ineffective_compression_count += 1
+                    self._record_ineffective_compression_verdict(
+                        self._ineffective_compression_count + 1,
+                    )
                     if not self.quiet_mode:
                         logger.warning(
                             "Compaction did not clear the threshold: %d real "
@@ -1763,7 +1849,7 @@ class ContextCompressor(ContextEngine):
                             self._ineffective_compression_count,
                         )
                 else:
-                    self._ineffective_compression_count = 0
+                    self._record_ineffective_compression_verdict(0)
         # Consume the pending-verification flag once real usage arrives, whether
         # or not prompt_tokens was reported, so a usage-less response can't leave
         # it armed for a later, unrelated reading.
@@ -1825,13 +1911,15 @@ class ContextCompressor(ContextEngine):
         return not self._automatic_compression_blocked()
 
     def _refresh_durable_guards(self) -> None:
-        """Re-read durable cooldown + fallback-streak state from the DB.
+        """Re-read durable cooldown + breaker 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.
+        durable rows (successful boundary, forced retry, a real usage
+        reading that dipped below the threshold) after this compressor was
+        bound, and neither the fallback streak nor the ineffective-strike
+        counter has a timer — without a re-read the stale in-memory
+        snapshot blocks forever.
         """
         try:
             self.get_active_compression_failure_cooldown(refresh=True)
@@ -1841,26 +1929,22 @@ class ContextCompressor(ContextEngine):
             self._load_fallback_compression_streak()
         except Exception as exc:
             logger.debug("compression fallback-streak refresh failed: %s", exc)
+        try:
+            self._load_ineffective_compression_count()
+        except Exception as exc:
+            logger.debug("compression ineffective-count 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.
-        if (
-            self._summary_failure_cooldown_until <= time.monotonic()
-            and self._fallback_compression_streak < 2
-        ):
-            # Blocked solely by the in-memory ineffective-compression
-            # counter, which is not durable — there is nothing in the DB
-            # that could unblock it, so skip the refresh (otherwise this
-            # branch would re-read the DB on every gate check for the rest
-            # of the session).
-            return True
+        # been cleared by another agent since bind_session_state() — a
+        # successful boundary, a forced retry, or a real usage reading
+        # below the threshold (which zeroes the durable ineffective
+        # counter) — so refresh and re-evaluate before letting a stale
+        # local block 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()
 
@@ -4187,7 +4271,9 @@ This compaction should PRIORITISE preserving all information related to the focu
             # threshold because of the incompressible floor (system prompt +
             # tool schemas), every subsequent turn re-fires a compaction that
             # returns here unchanged, and the CLI appears frozen.
-            self._ineffective_compression_count += 1
+            self._record_ineffective_compression_verdict(
+                self._ineffective_compression_count + 1,
+            )
             self._last_compression_savings_pct = 0.0
             telemetry["failure_class"] = "insufficient_messages"
             if not self.quiet_mode:
@@ -4255,7 +4341,9 @@ This compaction should PRIORITISE preserving all information related to the focu
             # an ineffective compression the anti-thrashing guard in
             # should_compress() never fires and every subsequent turn
             # re-triggers a no-op compression loop.  (#40803)
-            self._ineffective_compression_count += 1
+            self._record_ineffective_compression_verdict(
+                self._ineffective_compression_count + 1,
+            )
             self._last_compression_savings_pct = 0.0
             if not self.quiet_mode:
                 logger.warning(

agent/conversation_compression.py

@@ -277,6 +277,7 @@ def _refresh_persisted_compression_guards(compressor: Any) -> None:
     method_calls = (
         ("get_active_compression_failure_cooldown", {"refresh": True}),
         ("_load_fallback_compression_streak", {}),
+        ("_load_ineffective_compression_count", {}),
     )
     for method_name, kwargs in method_calls:
         method = getattr(type(compressor), method_name, None)

contributors/emails/lanyusea@gmail.com

@@ -0,0 +1 @@
+lanyusea

hermes_state.py

@@ -1040,6 +1040,7 @@ CREATE TABLE IF NOT EXISTS sessions (
     compression_failure_cooldown_until REAL,
     compression_failure_error TEXT,
     compression_fallback_streak INTEGER NOT NULL DEFAULT 0,
+    compression_ineffective_count INTEGER NOT NULL DEFAULT 0,
     profile_name TEXT,
     rewind_count INTEGER NOT NULL DEFAULT 0,
     archived INTEGER NOT NULL DEFAULT 0,
@@ -4040,6 +4041,51 @@ class SessionDB:
 
         self._execute_write(_do)
 
+    def get_compression_ineffective_count(self, session_id: str) -> int:
+        """Return the persisted ineffective-compaction strike count.
+
+        Mirrors ``get_compression_fallback_streak``: this is the durable half
+        of the anti-thrash guard (``_ineffective_compression_count`` on the
+        built-in compressor), persisted so that a fresh compressor bound to a
+        resumed session inherits an armed/tripped guard instead of starting
+        from zero across process restarts (#54923).
+        """
+        if not session_id:
+            return 0
+        with self._lock:
+            conn = self._conn
+            if conn is None:
+                return 0
+            row = conn.execute(
+                "SELECT compression_ineffective_count FROM sessions WHERE id = ?",
+                (session_id,),
+            ).fetchone()
+        if row is None:
+            return 0
+        value = (
+            row["compression_ineffective_count"]
+            if isinstance(row, sqlite3.Row)
+            else row[0]
+        )
+        try:
+            return max(0, int(value or 0))
+        except (TypeError, ValueError):
+            return 0
+
+    def set_compression_ineffective_count(self, session_id: str, count: int) -> None:
+        """Persist the ineffective-compaction strike count for one session."""
+        if not session_id:
+            return
+        normalized = max(0, int(count))
+
+        def _do(conn):
+            conn.execute(
+                "UPDATE sessions SET compression_ineffective_count = ? WHERE id = ?",
+                (normalized, session_id),
+            )
+
+        self._execute_write(_do)
+
     # ──────────────────────────────────────────────────────────────────────
     # Compression locks
     # ──────────────────────────────────────────────────────────────────────

tests/agent/test_compression_anti_thrash_persistence.py

@@ -0,0 +1,232 @@
+"""Anti-thrash state must survive process restarts (#54923).
+
+The guard in ``_automatic_compression_blocked_locally()`` trips only after two
+consecutive compactions that fail to bring the real prompt under the
+threshold. Historically ``_ineffective_compression_count`` was in-memory only:
+a fresh compressor bound to a resumed (already-compacted) session started at
+``compression_count == 0`` with a disarmed guard, so a near-threshold session
+could legally re-compact once per process restart, forever — the exact
+residual @lanyusea identified in #54923.
+
+The counter now round-trips the durable session-state channel exactly like
+``compression_failure_cooldown_until`` (#54465) and the fallback streak
+(af7dceaf7):
+
+* every verdict (strike or clear) writes through to the session row,
+* ``bind_session_state()`` loads the persisted value, so a fresh compressor
+  on a resumed session inherits an armed (1) or tripped (2) guard,
+* the reset semantics are unchanged — a real provider reading below the
+  threshold still clears the counter (update_from_response), and that clear
+  is durable too.
+"""
+from pathlib import Path
+from unittest.mock import patch
+
+from agent.context_compressor import ContextCompressor
+from hermes_state import SessionDB
+
+
+def _compressor(db: SessionDB | None = None, session_id: str = "") -> ContextCompressor:
+    with patch(
+        "agent.context_compressor.get_model_context_length",
+        return_value=100_000,
+    ):
+        cc = ContextCompressor(
+            model="test/model",
+            threshold_percent=0.85,
+            protect_first_n=2,
+            protect_last_n=2,
+            quiet_mode=True,
+        )
+    if db is not None:
+        cc.bind_session_state(db, session_id)
+    return cc
+
+
+def _db(tmp_path: Path) -> SessionDB:
+    return SessionDB(db_path=tmp_path / "state.db")
+
+
+class TestCounterRoundTripsBindSessionState:
+    def test_fresh_compressor_inherits_tripped_guard_after_restart(self, tmp_path):
+        """A restart must not disarm a tripped anti-thrash breaker."""
+        db = _db(tmp_path)
+        db.create_session("s1", source="cli")
+
+        first = _compressor(db, "s1")
+        # Two compactions that failed to clear the threshold — judged on the
+        # provider's real prompt counts, exactly as conversation_loop drives it.
+        for _ in range(2):
+            first._verify_compaction_cleared_threshold = True
+            first.update_from_response({"prompt_tokens": first.threshold_tokens + 1})
+        assert first._ineffective_compression_count == 2
+        assert first.should_compress(10**9) is False
+
+        # Process restart: a brand-new compressor binds the same session.
+        second = _compressor(db, "s1")
+        assert second.compression_count == 0  # the #54923 precondition
+        assert second._ineffective_compression_count == 2
+        assert second.should_compress(10**9) is False, (
+            "a fresh compressor on a resumed session must inherit the "
+            "tripped anti-thrash guard instead of re-compacting"
+        )
+
+    def test_fresh_compressor_inherits_armed_single_strike(self, tmp_path):
+        """One strike before the restart still counts toward the trip."""
+        db = _db(tmp_path)
+        db.create_session("s1", source="cli")
+
+        first = _compressor(db, "s1")
+        first._verify_compaction_cleared_threshold = True
+        first.update_from_response({"prompt_tokens": first.threshold_tokens + 1})
+        assert first._ineffective_compression_count == 1
+
+        second = _compressor(db, "s1")
+        assert second._ineffective_compression_count == 1
+        # One inherited strike does not block yet...
+        assert second.should_compress(10**9) is True
+        # ...but the next ineffective pass trips the guard cross-process.
+        second._verify_compaction_cleared_threshold = True
+        second.update_from_response({"prompt_tokens": second.threshold_tokens + 1})
+        assert second._ineffective_compression_count == 2
+        assert second.should_compress(10**9) is False
+
+    def test_rebind_to_other_session_does_not_leak_counter(self, tmp_path):
+        """The counter is per-session: switching sessions must not carry it."""
+        db = _db(tmp_path)
+        db.create_session("hot", source="cli")
+        db.create_session("cold", source="cli")
+        db.set_compression_ineffective_count("hot", 2)
+
+        cc = _compressor(db, "hot")
+        assert cc._ineffective_compression_count == 2
+
+        cc.bind_session_state(db, "cold")
+        assert cc._ineffective_compression_count == 0
+
+    def test_unbound_compressor_keeps_in_memory_behavior(self):
+        """No session DB bound (plugins/tests): everything still works."""
+        cc = _compressor()
+        cc._verify_compaction_cleared_threshold = True
+        cc.update_from_response({"prompt_tokens": cc.threshold_tokens + 1})
+        assert cc._ineffective_compression_count == 1
+        cc.update_from_response({"prompt_tokens": 1})
+        assert cc._ineffective_compression_count == 0
+
+
+class TestResetSemanticsPreserved:
+    def test_real_dip_below_threshold_clears_counter_durably(self, tmp_path):
+        """The L1466-1474 contract survives: any real provider reading below
+        the threshold clears the latch — and now clears the durable copy, so
+        a restart cannot resurrect voided strikes."""
+        db = _db(tmp_path)
+        db.create_session("s1", source="cli")
+
+        cc = _compressor(db, "s1")
+        cc._verify_compaction_cleared_threshold = True
+        cc.update_from_response({"prompt_tokens": cc.threshold_tokens + 1})
+        assert cc._ineffective_compression_count == 1
+        assert db.get_compression_ineffective_count("s1") == 1
+
+        # An ordinary fitting response (not post-compaction) clears the latch.
+        cc.update_from_response({"prompt_tokens": cc.threshold_tokens - 1})
+        assert cc._ineffective_compression_count == 0
+        assert db.get_compression_ineffective_count("s1") == 0
+
+        # And a restart sees the cleared state.
+        fresh = _compressor(db, "s1")
+        assert fresh._ineffective_compression_count == 0
+        assert fresh.should_compress(10**9) is True
+
+    def test_post_compaction_clearing_reading_resets_durably(self, tmp_path):
+        """The post-compaction success verdict (real tokens under threshold)
+        also zeroes the durable strike count."""
+        db = _db(tmp_path)
+        db.create_session("s1", source="cli")
+
+        cc = _compressor(db, "s1")
+        cc._verify_compaction_cleared_threshold = True
+        cc.update_from_response({"prompt_tokens": cc.threshold_tokens + 1})
+        assert db.get_compression_ineffective_count("s1") == 1
+
+        cc._verify_compaction_cleared_threshold = True
+        cc.update_from_response({"prompt_tokens": cc.threshold_tokens - 1})
+        assert cc._ineffective_compression_count == 0
+        assert db.get_compression_ineffective_count("s1") == 0
+
+    def test_update_model_reset_writes_through(self, tmp_path):
+        """update_model() voids strikes judged against the old threshold; the
+        durable copy must not resurrect them on the next restart."""
+        db = _db(tmp_path)
+        db.create_session("s1", source="cli")
+
+        cc = _compressor(db, "s1")
+        db.set_compression_ineffective_count("s1", 2)
+        cc._ineffective_compression_count = 2
+
+        cc.update_model("other/model", 100_000)
+
+        assert cc._ineffective_compression_count == 0
+        assert db.get_compression_ineffective_count("s1") == 0
+
+
+class TestStrikesPersistFromEveryVerdictSite:
+    def test_no_op_compaction_branches_write_through(self, tmp_path):
+        """The insufficient-messages no-op branch records its strike durably."""
+        db = _db(tmp_path)
+        db.create_session("s1", source="cli")
+
+        cc = _compressor(db, "s1")
+        # 3 tiny messages < minimum window → the #40803 no-op branch.
+        msgs = [
+            {"role": "system", "content": "sys"},
+            {"role": "user", "content": "hi"},
+            {"role": "assistant", "content": "hello"},
+        ]
+        out = cc.compress(msgs, current_tokens=10**9)
+        assert out == msgs
+        assert cc._ineffective_compression_count == 1
+        assert db.get_compression_ineffective_count("s1") == 1
+
+    def test_persist_failure_is_swallowed_and_memory_still_advances(self, tmp_path):
+        """A DB write failure must not break the in-memory guard."""
+        db = _db(tmp_path)
+        db.create_session("s1", source="cli")
+        cc = _compressor(db, "s1")
+
+        with patch.object(
+            db,
+            "set_compression_ineffective_count",
+            side_effect=Exception("disk full"),
+        ):
+            cc._verify_compaction_cleared_threshold = True
+            cc.update_from_response({"prompt_tokens": cc.threshold_tokens + 1})
+
+        assert cc._ineffective_compression_count == 1
+        assert db.get_compression_ineffective_count("s1") == 0
+
+
+class TestCompressionBoundaryCarry:
+    def test_rotation_boundary_carries_counter_onto_child_row(self, tmp_path):
+        """Session-id rotation must not launder an armed guard through the
+        fresh child row (same carry contract as the fallback streak)."""
+        db = _db(tmp_path)
+        db.create_session("parent", source="cli")
+        cc = _compressor(db, "parent")
+        cc._verify_compaction_cleared_threshold = True
+        cc.update_from_response({"prompt_tokens": cc.threshold_tokens + 1})
+        assert db.get_compression_ineffective_count("parent") == 1
+
+        db.create_session("child", source="cli", parent_session_id="parent")
+        cc.on_session_start(
+            "child",
+            boundary_reason="compression",
+            old_session_id="parent",
+            session_db=db,
+        )
+
+        assert cc._session_id == "child"
+        assert cc._ineffective_compression_count == 1
+        # Persisted onto the child row so a restart right after rotation
+        # still inherits the armed guard.
+        assert db.get_compression_ineffective_count("child") == 1

tests/agent/test_compression_rotation_state.py

@@ -618,24 +618,28 @@ class TestCooldownPersistFailureIsNotAClearedRow:
         assert compressor.get_active_compression_failure_cooldown(refresh=True) is None
         assert compressor._summary_failure_cooldown_until == 0.0
 
-    def test_ineffective_count_only_block_skips_durable_refresh(
+    def test_ineffective_count_block_honors_durable_clear_by_another_agent(
         self,
         refresh_state_db: SessionDB,
     ):
-        """A block owed solely to the in-memory ineffective counter (which is
-        not durable) must not re-read the DB on every gate check."""
+        """The ineffective-strike counter is durable (#54923): a block owed to
+        it must re-read the DB so another agent's clear (a real usage reading
+        that dipped below the threshold) unblocks this compressor too."""
         db = refresh_state_db
-        session_id = "INEFFECTIVE_ONLY_BLOCK"
+        session_id = "INEFFECTIVE_DURABLE_BLOCK"
         db.create_session(session_id, source="telegram")
+        db.set_compression_ineffective_count(session_id, 2)
         compressor = _bound_context_compressor(db, session_id)
-        compressor._ineffective_compression_count = 2
+        assert compressor._ineffective_compression_count == 2
 
-        with patch.object(
-            compressor,
-            "_refresh_durable_guards",
-            side_effect=AssertionError("nothing durable to refresh"),
-        ):
-            assert compressor._automatic_compression_blocked() is True
+        assert compressor._automatic_compression_blocked() is True
+
+        # Another agent's real prompt reading dipped below the threshold and
+        # zeroed the durable counter.
+        db.set_compression_ineffective_count(session_id, 0)
+
+        assert compressor._automatic_compression_blocked() is False
+        assert compressor._ineffective_compression_count == 0
 
 
 class TestTodoSnapshotMergedNotDuplicated:

tests/agent/test_idle_compaction_lock_and_guards.py

@@ -221,6 +221,9 @@ def test_idle_compaction_respects_anti_thrash_breaker(tmp_path: Path) -> None:
             quiet_mode=True,
         )
     compressor.bind_session_state(db, sid)
+    # Trip the breaker durably (#54923: the strike counter now round-trips
+    # state.db, and the gate re-reads durable rows before honoring a block).
+    db.set_compression_ineffective_count(sid, 2)
     compressor._ineffective_compression_count = 2  # breaker tripped
     compressor.compress = MagicMock()
     agent.context_compressor = compressor

tests/test_hermes_state.py

@@ -6903,6 +6903,22 @@ def test_compression_fallback_streak_round_trips(db):
     assert db.get_compression_fallback_streak("s1") == 2
 
 
+def test_compression_ineffective_count_round_trips(db):
+    db.create_session("s1", "cli")
+
+    assert db.get_compression_ineffective_count("s1") == 0
+    db.set_compression_ineffective_count("s1", 2)
+    assert db.get_compression_ineffective_count("s1") == 2
+    # Clearing (real usage dipped below the threshold) round-trips too.
+    db.set_compression_ineffective_count("s1", 0)
+    assert db.get_compression_ineffective_count("s1") == 0
+    # Negative and missing-session inputs are normalized/ignored.
+    db.set_compression_ineffective_count("s1", -3)
+    assert db.get_compression_ineffective_count("s1") == 0
+    assert db.get_compression_ineffective_count("nope") == 0
+    assert db.get_compression_ineffective_count("") == 0
+
+
 def test_refresh_compression_lock_requires_holder_and_preserves_reclaimability(db, monkeypatch):
     db.create_session("s1", "cli")