diff --git a/agent/context_compressor.py b/agent/context_compressor.py index aaa7c0ddb57..fd0fdc7078f 100644 --- a/agent/context_compressor.py +++ b/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( diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 2e6f6a67a76..b7b3af88280 100644 --- a/agent/conversation_compression.py +++ b/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) diff --git a/contributors/emails/lanyusea@gmail.com b/contributors/emails/lanyusea@gmail.com new file mode 100644 index 00000000000..f4aca247c3b --- /dev/null +++ b/contributors/emails/lanyusea@gmail.com @@ -0,0 +1 @@ +lanyusea diff --git a/hermes_state.py b/hermes_state.py index b5d5421184a..a424591d998 100644 --- a/hermes_state.py +++ b/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 # ────────────────────────────────────────────────────────────────────── diff --git a/tests/agent/test_compression_anti_thrash_persistence.py b/tests/agent/test_compression_anti_thrash_persistence.py new file mode 100644 index 00000000000..7f1324d4c90 --- /dev/null +++ b/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 diff --git a/tests/agent/test_compression_rotation_state.py b/tests/agent/test_compression_rotation_state.py index a5d7b58b962..0ba525d6253 100644 --- a/tests/agent/test_compression_rotation_state.py +++ b/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: diff --git a/tests/agent/test_idle_compaction_lock_and_guards.py b/tests/agent/test_idle_compaction_lock_and_guards.py index a5deaf9e709..2142905434b 100644 --- a/tests/agent/test_idle_compaction_lock_and_guards.py +++ b/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 diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index fe984e10e90..9e87ff501f7 100644 --- a/tests/test_hermes_state.py +++ b/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")