From af7dceaf77bbcd5dcafe4f65982c7cd7df5f4c4a Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:50:20 +0530 Subject: [PATCH] fix(context): persist fallback compaction breaker --- agent/codex_runtime.py | 9 +- agent/context_compressor.py | 142 ++++++++++++++---- agent/conversation_compression.py | 41 ++++- hermes_state.py | 40 +++++ tests/agent/test_compaction_anti_thrash.py | 64 ++++++++ .../agent/test_compression_rotation_state.py | 94 ++++++++++++ tests/agent/test_context_compressor.py | 56 ------- .../test_context_engine_host_contract.py | 3 + .../test_codex_app_server_compaction.py | 28 ++++ tests/test_hermes_state.py | 8 + 10 files changed, 394 insertions(+), 91 deletions(-) diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 664e4fbe9b14..c1e46e8a4604 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -282,7 +282,14 @@ def _record_codex_app_server_compaction( # The app server has already completed a real compaction boundary. Its # usage update (when supplied) is therefore the same real-vs-real # effectiveness verdict used by the normal compression path. - if hasattr(compressor, "_verify_compaction_cleared_threshold"): + record_boundary = getattr( + type(compressor), "record_completed_compaction", None + ) + if callable(record_boundary): + # Codex owns this summary. A prior Hermes deterministic-fallback + # flag must not leak into the native boundary's quality verdict. + record_boundary(compressor, used_fallback=False) + elif hasattr(compressor, "_verify_compaction_cleared_threshold"): compressor._verify_compaction_cleared_threshold = True if not getattr(turn, "token_usage_last", None): compressor.last_prompt_tokens = -1 diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 2ffff5ef9c15..ec4314ab40b0 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -736,6 +736,7 @@ class ContextCompressor(ContextEngine): self._last_aux_model_failure_model = None self._last_compression_savings_pct = 100.0 self._ineffective_compression_count = 0 + self._fallback_compression_streak = 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 @@ -773,6 +774,7 @@ class ContextCompressor(ContextEngine): self._last_aux_model_failure_model = None self._last_compression_savings_pct = 100.0 self._ineffective_compression_count = 0 + self._fallback_compression_streak = 0 self._verify_compaction_cleared_threshold = False self._last_compression_made_progress = False self._summary_failure_cooldown_until = 0.0 @@ -790,12 +792,84 @@ class ContextCompressor(ContextEngine): self._session_id = session_id or "" self._summary_failure_cooldown_until = 0.0 self._last_summary_error = None + self._fallback_compression_streak = 0 self.get_active_compression_failure_cooldown() + self._load_fallback_compression_streak() def on_session_start(self, session_id: str, **kwargs) -> None: """Bind session-scoped compression state for a new or resumed session.""" super().on_session_start(session_id, **kwargs) - self.bind_session_state(kwargs.get("session_db", getattr(self, "_session_db", None)), session_id) + boundary_reason = kwargs.get("boundary_reason") + 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 + if boundary_reason == "compression" and old_session_id: + getter = getattr(session_db, "get_compression_fallback_streak", None) + if callable(getter): + try: + stored_streak = getter(old_session_id) + if isinstance(stored_streak, (int, float, str)): + previous_fallback_streak = max(0, int(stored_streak)) + except (TypeError, ValueError, sqlite3.Error) as exc: + logger.debug("compression parent fallback streak lookup failed: %s", exc) + except Exception as exc: + logger.debug( + "compression parent fallback streak 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 + + def _load_fallback_compression_streak(self) -> None: + session_db = getattr(self, "_session_db", None) + session_id = getattr(self, "_session_id", "") + getter = getattr(session_db, "get_compression_fallback_streak", None) + if not session_id or not callable(getter): + return + try: + stored_streak = getter(session_id) + self._fallback_compression_streak = max( + 0, + int(stored_streak) + if isinstance(stored_streak, (int, float, str)) + else 0, + ) + except (TypeError, ValueError, sqlite3.Error) as exc: + logger.debug("compression fallback streak lookup failed: %s", exc) + except Exception as exc: + logger.debug("compression fallback streak lookup failed (non-sqlite): %s", exc) + + def _persist_fallback_compression_streak(self) -> None: + session_db = getattr(self, "_session_db", None) + session_id = getattr(self, "_session_id", "") + setter = getattr(session_db, "set_compression_fallback_streak", None) + if not session_id or not callable(setter): + return + try: + setter(session_id, self._fallback_compression_streak) + except sqlite3.Error as exc: + logger.debug("compression fallback streak persist failed: %s", exc) + except Exception as exc: + logger.debug("compression fallback streak persist failed (non-sqlite): %s", exc) + + def record_completed_compaction(self, *, used_fallback: bool = False) -> None: + """Record one completed boundary and its summary quality.""" + self._verify_compaction_cleared_threshold = True + if used_fallback: + self._fallback_compression_streak += 1 + if not self.quiet_mode: + logger.warning( + "Compaction completed with a deterministic fallback summary. " + "fallback_compression_streak=%d", + self._fallback_compression_streak, + ) + elif self._fallback_compression_streak: + self._fallback_compression_streak = 0 + self._persist_fallback_compression_streak() def get_active_compression_failure_cooldown(self) -> Optional[Dict[str, Any]]: """Return the live compression-failure cooldown for the bound session.""" @@ -893,6 +967,12 @@ class ContextCompressor(ContextEngine): max_tokens: int | None = None, ) -> None: """Update model info after a model switch or fallback activation.""" + runtime_changed = any(( + model != self.model, + provider != self.provider, + base_url != self.base_url, + api_mode != self.api_mode, + )) self.model = model self.base_url = base_url self.api_key = api_key @@ -948,6 +1028,9 @@ class ContextCompressor(ContextEngine): self.last_compression_rough_tokens = 0 self.awaiting_real_usage_after_compression = False self._ineffective_compression_count = 0 + if runtime_changed: + self._fallback_compression_streak = 0 + self._persist_fallback_compression_streak() self._verify_compaction_cleared_threshold = False self._last_compression_made_progress = False @@ -1137,6 +1220,10 @@ 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 + # Consecutive completed deterministic-fallback boundaries. Unlike the + # real-usage effectiveness counter, ordinary fitting responses must not + # reset this breaker; only a healthy completed summary does. + self._fallback_compression_streak: int = 0 # 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 @@ -1186,23 +1273,15 @@ class ContextCompressor(ContextEngine): self.last_total_tokens = usage.get("total_tokens", self.last_prompt_tokens + self.last_completion_tokens) if self.last_prompt_tokens > 0: self.last_real_prompt_tokens = self.last_prompt_tokens - fallback_compaction = ( - self._verify_compaction_cleared_threshold - and self._last_summary_fallback_used - ) 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, UNLESS the just-finished compaction had to fall - # back to the deterministic placeholder summary. That path - # does shrink the prompt, but it is still a degraded compaction - # attempt and must count toward the anti-thrashing strike limit. - # Otherwise repeated empty-summary failures can rotate through - # fallback markers forever while each smaller prompt reading - # resets the counter back to zero (#63008 / R1). - if not fallback_compaction: - self._ineffective_compression_count = 0 + # fits again. Clear the real-usage effectiveness latch even + # 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 else: self.last_rough_tokens_when_real_prompt_fit = 0 @@ -1223,17 +1302,7 @@ class ContextCompressor(ContextEngine): # Keying on real usage compares like with like and fires exactly once # per compaction. if self._verify_compaction_cleared_threshold: - if fallback_compaction: - self._ineffective_compression_count += 1 - if not self.quiet_mode: - logger.warning( - "Compaction completed with a deterministic fallback " - "summary. Counting this as a degraded attempt to " - "avoid repeated fallback-only compaction loops. " - "ineffective_compression_count=%d", - self._ineffective_compression_count, - ) - elif self.last_prompt_tokens >= self.threshold_tokens: + if self.last_prompt_tokens >= self.threshold_tokens: self._ineffective_compression_count += 1 if not self.quiet_mode: logger.warning( @@ -1305,6 +1374,10 @@ class ContextCompressor(ContextEngine): tokens = prompt_tokens if prompt_tokens is not None else self.last_prompt_tokens if tokens < self.threshold_tokens: return False + return not self._automatic_compression_blocked() + + def _automatic_compression_blocked(self) -> bool: + """Return whether automatic compaction is in cooldown or tripped.""" # 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 @@ -1321,18 +1394,23 @@ class ContextCompressor(ContextEngine): "Compression deferred — summary LLM in cooldown for %.0fs more", _cooldown_remaining, ) - return False + return True # Anti-thrashing: back off if recent compressions were ineffective - if self._ineffective_compression_count >= 2: + if ( + self._ineffective_compression_count >= 2 + or self._fallback_compression_streak >= 2 + ): if not self.quiet_mode: logger.warning( - "Compression skipped — last %d compaction attempts did not " - "restore enough context headroom. Consider /new to start a " - "fresh session, or /compress for focused compression.", + "Compression skipped — repeated compaction attempts did not " + "restore healthy context. ineffective=%d fallback=%d. " + "Consider /new to start fresh, or /compress for " + "focused compression.", self._ineffective_compression_count, + self._fallback_compression_streak, ) - return False - return True + return True + return False # ------------------------------------------------------------------ # Tool output pruning (cheap pre-pass, no LLM call) diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index d24e7d88c6ce..6f2c5eb57576 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -504,6 +504,21 @@ def compress_context( force=force, ) + # Every automatic entrypoint must honor compressor-owned cooldown and + # breaker state. Gateway hygiene constructs a fresh AIAgent, so the + # persisted fallback streak is loaded by bind_session_state() before this. + if not force: + blocked = getattr( + type(agent.context_compressor), + "_automatic_compression_blocked", + None, + ) + if callable(blocked) and blocked(agent.context_compressor): + existing_prompt = getattr(agent, "_cached_system_prompt", None) + if not existing_prompt: + existing_prompt = agent._build_system_prompt(system_message) + return messages, existing_prompt + # Lazy feasibility check — run the auxiliary-provider probe + context # length lookup just-in-time on the first compression attempt instead of # at AIAgent.__init__. Saves ~400ms cold off every short session that @@ -718,6 +733,17 @@ def compress_context( _release_lock() raise + # Capture boundary quality before session-rotation callbacks run. Built-in + # and plugin lifecycle hooks may reset per-session compressor fields while + # rebinding to the child id; the completed attempt's verdict must survive + # that rebind and be recorded only after the full boundary commits. + _compression_made_progress = bool( + getattr(agent.context_compressor, "_last_compression_made_progress", False) + ) + _compression_used_fallback = bool( + getattr(agent.context_compressor, "_last_summary_fallback_used", False) + ) + # If compression aborted (aux LLM failed to produce a usable summary) # the compressor returns the input messages unchanged. Surface the # error to the user, skip the session-rotation work entirely (no @@ -1046,8 +1072,19 @@ def compress_context( # 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 + if _compression_made_progress: + record_boundary = getattr( + type(agent.context_compressor), + "record_completed_compaction", + None, + ) + if callable(record_boundary): + record_boundary( + agent.context_compressor, + used_fallback=_compression_used_fallback, + ) + else: + agent.context_compressor._verify_compaction_cleared_threshold = True # Clear the file-read dedup cache. After compression the original # read content is summarised away — if the model re-reads the same diff --git a/hermes_state.py b/hermes_state.py index 11d1507bb318..67dc9d5d8cb9 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -758,6 +758,7 @@ CREATE TABLE IF NOT EXISTS sessions ( handoff_error TEXT, compression_failure_cooldown_until REAL, compression_failure_error TEXT, + compression_fallback_streak INTEGER NOT NULL DEFAULT 0, rewind_count INTEGER NOT NULL DEFAULT 0, archived INTEGER NOT NULL DEFAULT 0, FOREIGN KEY (parent_session_id) REFERENCES sessions(id) @@ -2295,6 +2296,45 @@ class SessionDB: "clear_compression_failure_cooldown(%s) failed: %s", session_id, exc, ) + + def get_compression_fallback_streak(self, session_id: str) -> int: + """Return the persisted deterministic-fallback streak.""" + if not session_id: + return 0 + with self._lock: + conn = self._conn + if conn is None: + return 0 + row = conn.execute( + "SELECT compression_fallback_streak FROM sessions WHERE id = ?", + (session_id,), + ).fetchone() + if row is None: + return 0 + value = ( + row["compression_fallback_streak"] + if isinstance(row, sqlite3.Row) + else row[0] + ) + try: + return max(0, int(value or 0)) + except (TypeError, ValueError): + return 0 + + def set_compression_fallback_streak(self, session_id: str, streak: int) -> None: + """Persist the deterministic-fallback streak for one session.""" + if not session_id: + return + normalized = max(0, int(streak)) + + def _do(conn): + conn.execute( + "UPDATE sessions SET compression_fallback_streak = ? WHERE id = ?", + (normalized, session_id), + ) + + self._execute_write(_do) + # ────────────────────────────────────────────────────────────────────── # Compression locks # ────────────────────────────────────────────────────────────────────── diff --git a/tests/agent/test_compaction_anti_thrash.py b/tests/agent/test_compaction_anti_thrash.py index 2eb4ea66108d..b7803c44c1e5 100644 --- a/tests/agent/test_compaction_anti_thrash.py +++ b/tests/agent/test_compaction_anti_thrash.py @@ -210,6 +210,70 @@ class TestFutilityGuard: assert cc.awaiting_real_usage_after_compression is False assert cc._ineffective_compression_count == 0 + def test_fallback_streak_survives_ordinary_fitting_responses(self): + cc = _compressor(threshold_tokens=24_576) + + cc.record_completed_compaction(used_fallback=True) + cc.update_from_response({"prompt_tokens": 20_000}) + assert cc._fallback_compression_streak == 1 + + # Context regrows through ordinary successful turns before the next + # fallback boundary. Those turns reset real-usage effectiveness, not + # the independent summary-quality breaker. + cc.update_from_response({"prompt_tokens": 20_000}) + cc.record_completed_compaction(used_fallback=True) + cc.update_from_response({"prompt_tokens": 20_000}) + + assert cc._fallback_compression_streak == 2 + assert not cc.should_compress(33_564) + + def test_usage_less_fallback_boundary_still_counts(self): + cc = _compressor(threshold_tokens=24_576) + + cc.record_completed_compaction(used_fallback=True) + cc.awaiting_real_usage_after_compression = True + cc.update_from_response({}) + + assert cc._fallback_compression_streak == 1 + assert cc._verify_compaction_cleared_threshold is False + assert cc.awaiting_real_usage_after_compression is False + + def test_healthy_boundary_resets_only_fallback_streak(self): + cc = _compressor(threshold_tokens=24_576) + cc.record_completed_compaction(used_fallback=True) + cc.record_completed_compaction(used_fallback=False) + + assert cc._fallback_compression_streak == 0 + assert cc._verify_compaction_cleared_threshold is True + + def test_model_switch_resets_and_persists_fallback_streak(self, tmp_path): + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session("s1", source="cli") + cc = _compressor(threshold_tokens=24_576) + cc.bind_session_state(db, "s1") + cc.record_completed_compaction(used_fallback=True) + + cc.update_model("next-model", 100_000) + + assert cc._fallback_compression_streak == 0 + assert db.get_compression_fallback_streak("s1") == 0 + + def test_same_runtime_context_recalibration_preserves_fallback_streak(self, tmp_path): + from hermes_state import SessionDB + + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session("s1", source="cli") + cc = _compressor(threshold_tokens=24_576) + cc.bind_session_state(db, "s1") + cc.record_completed_compaction(used_fallback=True) + + cc.update_model(cc.model, 64_000, provider=cc.provider) + + assert cc._fallback_compression_streak == 1 + assert db.get_compression_fallback_streak("s1") == 1 + def test_a_failed_pass_records_exactly_one_strike(self): """A compaction that leaves the real prompt over the threshold: one strike. diff --git a/tests/agent/test_compression_rotation_state.py b/tests/agent/test_compression_rotation_state.py index 83ab63e2a699..a88db5f364b5 100644 --- a/tests/agent/test_compression_rotation_state.py +++ b/tests/agent/test_compression_rotation_state.py @@ -21,6 +21,7 @@ import os from pathlib import Path from unittest.mock import MagicMock, patch +from agent.context_compressor import ContextCompressor from hermes_state import SessionDB @@ -130,3 +131,96 @@ class TestPlatformForwardedAtBoundary: kwargs = calls[-1].kwargs assert kwargs.get("platform") == "telegram" assert kwargs.get("boundary_reason") == "compression" + + +class TestFallbackStreakFollowsRotation: + def test_fallback_boundary_persists_on_child_session(self, tmp_path: Path): + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_FALLBACK_ROT" + db.create_session(parent, source="telegram") + with patch( + "agent.context_compressor.get_model_context_length", + return_value=100_000, + ): + compressor = ContextCompressor( + model="test/model", + threshold_percent=0.85, + protect_first_n=2, + protect_last_n=2, + quiet_mode=True, + ) + compressor.bind_session_state(db, parent) + + # A fallback streak must survive the session-id rotation itself. The + # boundary then records the just-completed fallback on the child row. + compressor.record_completed_compaction(used_fallback=True) + assert db.get_compression_fallback_streak(parent) == 1 + db.create_session( + "CHILD_FALLBACK_ROT", + source="telegram", + parent_session_id=parent, + ) + compressor.on_session_start( + "CHILD_FALLBACK_ROT", + session_db=db, + boundary_reason="compression", + old_session_id=parent, + ) + assert compressor._fallback_compression_streak == 1 + + compressor.record_completed_compaction(used_fallback=True) + assert compressor._fallback_compression_streak == 2 + assert db.get_compression_fallback_streak("CHILD_FALLBACK_ROT") == 2 + + resumed = ContextCompressor( + model="test/model", + threshold_percent=0.85, + protect_first_n=2, + protect_last_n=2, + quiet_mode=True, + ) + resumed.bind_session_state(db, "CHILD_FALLBACK_ROT") + assert resumed._fallback_compression_streak == 2 + + def test_real_rotation_records_fallback_after_lifecycle_rebind(self, tmp_path: Path): + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_REAL_FALLBACK_ROT" + db.create_session(parent, source="telegram") + agent = _build_agent_with_db(db, parent, platform="telegram") + + with patch( + "agent.context_compressor.get_model_context_length", + return_value=100_000, + ): + compressor = ContextCompressor( + model="test/model", + threshold_percent=0.85, + protect_first_n=2, + protect_last_n=2, + quiet_mode=True, + ) + compressor.bind_session_state(db, parent) + compressed = [ + {"role": "user", "content": "[CONTEXT COMPACTION] fallback"}, + {"role": "assistant", "content": "tail"}, + ] + + def _fallback_compress(*_args, **_kwargs): + compressor._last_summary_error = "empty summary" + compressor._last_summary_fallback_used = True + compressor._last_compression_made_progress = True + return compressed + + with patch.object( + compressor, + "compress", + side_effect=_fallback_compress, + ): + compressor.compression_count = 1 + setattr(agent, "context_compressor", compressor) + agent._compress_context(_msgs(), "sys", approx_tokens=120_000) + child = getattr(agent, "session_id") + + assert child != parent + assert compressor._fallback_compression_streak == 1 + assert db.get_compression_fallback_streak(child) == 1 diff --git a/tests/agent/test_context_compressor.py b/tests/agent/test_context_compressor.py index 46cbff8bfa15..230090fd6f95 100644 --- a/tests/agent/test_context_compressor.py +++ b/tests/agent/test_context_compressor.py @@ -65,62 +65,6 @@ class TestUpdateFromResponse: compressor.update_from_response({}) assert compressor.last_prompt_tokens == 0 - def test_fallback_compaction_counts_as_ineffective_even_when_prompt_fits(self, compressor): - compressor.threshold_tokens = 85_000 - compressor.awaiting_real_usage_after_compression = True - compressor.last_compression_rough_tokens = 90_000 - compressor._verify_compaction_cleared_threshold = True - compressor._last_summary_fallback_used = True - - compressor.update_from_response({ - "prompt_tokens": 5_000, - "completion_tokens": 100, - "total_tokens": 5_100, - }) - - assert compressor.last_real_prompt_tokens == 5_000 - assert compressor.last_rough_tokens_when_real_prompt_fit == 90_000 - assert compressor._ineffective_compression_count == 1 - assert compressor._verify_compaction_cleared_threshold is False - assert compressor.awaiting_real_usage_after_compression is False - - def test_successful_compaction_fit_still_clears_prior_ineffective_count(self, compressor): - compressor.threshold_tokens = 85_000 - compressor.awaiting_real_usage_after_compression = True - compressor.last_compression_rough_tokens = 90_000 - compressor._verify_compaction_cleared_threshold = True - compressor._last_summary_fallback_used = False - compressor._ineffective_compression_count = 1 - - compressor.update_from_response({ - "prompt_tokens": 5_000, - "completion_tokens": 100, - "total_tokens": 5_100, - }) - - assert compressor._ineffective_compression_count == 0 - - def test_second_fallback_strike_blocks_future_compression(self, compressor): - compressor.threshold_tokens = 85_000 - compressor.last_prompt_tokens = 90_000 - assert compressor.should_compress() is True - - for expected in (1, 2): - compressor.awaiting_real_usage_after_compression = True - compressor.last_compression_rough_tokens = 90_000 - compressor._verify_compaction_cleared_threshold = True - compressor._last_summary_fallback_used = True - compressor.update_from_response({ - "prompt_tokens": 5_000, - "completion_tokens": 100, - "total_tokens": 5_100, - }) - assert compressor._ineffective_compression_count == expected - compressor.last_prompt_tokens = 90_000 - - assert compressor.should_compress() is False - - class TestPreflightDeferral: def test_defers_when_recent_real_usage_fit_and_rough_growth_is_small(self, compressor): compressor.threshold_tokens = 85_000 diff --git a/tests/agent/test_context_engine_host_contract.py b/tests/agent/test_context_engine_host_contract.py index 9bcde04cc901..d8033d53b7ab 100644 --- a/tests/agent/test_context_engine_host_contract.py +++ b/tests/agent/test_context_engine_host_contract.py @@ -165,6 +165,7 @@ def test_reset_session_state_rebinds_builtin_compressor_after_session_switch(tmp db.create_session("old-sid", source="cli") db.create_session("new-sid", source="cli") db.record_compression_failure_cooldown("old-sid", 4_000_000_000.0, "old-timeout") + db.set_compression_fallback_streak("old-sid", 2) monkeypatch.setattr( "agent.context_compressor.get_model_context_length", @@ -188,7 +189,9 @@ def test_reset_session_state_rebinds_builtin_compressor_after_session_switch(tmp assert compressor._session_id == "new-sid" assert compressor.get_active_compression_failure_cooldown() is None + assert compressor._fallback_compression_streak == 0 assert db.get_compression_failure_cooldown("old-sid") is not None + assert db.get_compression_fallback_streak("old-sid") == 2 compressor._record_compression_failure_cooldown(30.0, "new-timeout") diff --git a/tests/run_agent/test_codex_app_server_compaction.py b/tests/run_agent/test_codex_app_server_compaction.py index e5e7d668e744..93bbe0eadf22 100644 --- a/tests/run_agent/test_codex_app_server_compaction.py +++ b/tests/run_agent/test_codex_app_server_compaction.py @@ -194,3 +194,31 @@ def test_codex_app_server_native_compaction_notice_emits_status_and_event(): }, ) ] + + +def test_codex_native_boundary_clears_stale_hermes_fallback_streak(): + from unittest.mock import patch + + from agent.context_compressor import ContextCompressor + + with patch( + "agent.context_compressor.get_model_context_length", + return_value=100_000, + ): + compressor = ContextCompressor(model="test-model", quiet_mode=True) + compressor._fallback_compression_streak = 1 + compressor._last_summary_fallback_used = True + + agent = DummyAgent( + TurnResult(thread_id="thread-1", turn_id="normal-turn-1") + ) + agent.context_compressor = compressor + turn = TurnResult( + thread_id="thread-1", + turn_id="normal-turn-1", + compacted=True, + ) + + assert _record_codex_app_server_compaction(agent, turn) is True + assert compressor._fallback_compression_streak == 0 + assert compressor._verify_compaction_cleared_threshold is True diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index be21f71859fb..c45ad6344ccb 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -5791,6 +5791,14 @@ def test_expired_compression_failure_cooldown_is_ignored(db): assert db.get_compression_failure_cooldown("s1") is None +def test_compression_fallback_streak_round_trips(db): + db.create_session("s1", "cli") + + assert db.get_compression_fallback_streak("s1") == 0 + db.set_compression_fallback_streak("s1", 2) + assert db.get_compression_fallback_streak("s1") == 2 + + def test_refresh_compression_lock_requires_holder_and_preserves_reclaimability(db, monkeypatch): db.create_session("s1", "cli")