refactor(compression): name the split-failure cooldown; drop duplicate tests

Review folds from the formal gate battery:
- _SPLIT_FAILURE_COOLDOWN_SECONDS = 60 replaces the bare literal, with a
  comment pinning WHY it is the timeout ladder's first rung (transient
  lease/DB condition) rather than the 600s summary-provider cooldown.
- publish_compression_child docstring now states the compression_lock_holder
  condition on the refresh guard.
- Dropped 2 of 3 extracted unit tests as duplicates of existing coverage in
  test_compression_rotation_state.py / test_context_compressor.py; kept the
  force-bypass test (only site pinning that behavior for split failures) and
  the E2E test (now asserting the named constant).
This commit is contained in:
kshitijk4poor
2026-08-31 12:01:29 +05:30
committed by kshitij
parent dc71fb37e1
commit 3aee290899
4 changed files with 20 additions and 35 deletions
+9 -1
View File
@@ -92,6 +92,13 @@ _TERMINAL_COMPRESSION_PROVENANCES = frozenset(
} }
) )
# Cooldown armed when a compression SPLIT fails (session_split_failed /
# rotation rollback, #97948 symptom B). Deliberately the FIRST rung of the
# timeout ladder (60/300/900 in context_compressor.py), not the 600s
# _SUMMARY_FAILURE_COOLDOWN_SECONDS: a split failure is usually a transient
# lease/DB condition, unlike a persistent summary-provider fault.
_SPLIT_FAILURE_COOLDOWN_SECONDS = 60
# Stable marker the gateway matches on to re-tag the auto-compaction lifecycle # Stable marker the gateway matches on to re-tag the auto-compaction lifecycle
# status as ``kind="compacting"`` (tui_gateway/server.py::_status_update), so # status as ``kind="compacting"`` (tui_gateway/server.py::_status_update), so
# drivers like the desktop app can show an explicit "Summarizing…" indicator # drivers like the desktop app can show an explicit "Summarizing…" indicator
@@ -5001,7 +5008,8 @@ def compress_context(
# a stub compressor must not mask the original error. # a stub compressor must not mask the original error.
try: try:
agent.context_compressor._record_compression_failure_cooldown( agent.context_compressor._record_compression_failure_cooldown(
60, f"session_split_failed: {e}", _SPLIT_FAILURE_COOLDOWN_SECONDS,
f"session_split_failed: {e}",
) )
except Exception: except Exception:
logger.debug( logger.debug(
+3 -2
View File
@@ -7072,8 +7072,9 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
rows in ``(watermark, watermark_ceiling]`` are foreign concurrent rows in ``(watermark, watermark_ceiling]`` are foreign concurrent
tail. ``None`` = unbounded (no internal flush happened). tail. ``None`` = unbounded (no internal flush happened).
When *require_lease_refresh* is True, the lease is refreshed inside When *require_lease_refresh* is True and *compression_lock_holder* is
the same transaction before the expiry check. This gives a refresher set, the lease is refreshed inside the same transaction before the
expiry check. This gives a refresher
that stopped due to transient DB failures one final chance to extend that stopped due to transient DB failures one final chance to extend
the lease, preventing wasted compression work. The refresh uses the the lease, preventing wasted compression work. The refresh uses the
same ``conn`` as the publication, so there is no TOCTOU window. same ``conn`` as the publication, so there is no TOCTOU window.
@@ -2233,6 +2233,8 @@ def test_failed_split_arms_failure_cooldown(tmp_path: Path) -> None:
"""Regression #97948 symptom B: a failed split/archive must arm the """Regression #97948 symptom B: a failed split/archive must arm the
compression failure cooldown so the next automatic turn cannot compression failure cooldown so the next automatic turn cannot
immediately re-run the identical doomed compression.""" immediately re-run the identical doomed compression."""
from agent.conversation_compression import _SPLIT_FAILURE_COOLDOWN_SECONDS
db = SessionDB(db_path=tmp_path / "state.db") db = SessionDB(db_path=tmp_path / "state.db")
session_id = "SPLIT_FAIL_COOLDOWN_TEST" session_id = "SPLIT_FAIL_COOLDOWN_TEST"
db.create_session(session_id, source="test") db.create_session(session_id, source="test")
@@ -2255,5 +2257,5 @@ def test_failed_split_arms_failure_cooldown(tmp_path: Path) -> None:
"split failure must arm the failure cooldown (#97948 symptom B)" "split failure must arm the failure cooldown (#97948 symptom B)"
) )
seconds, error = cooldown_calls[0].args seconds, error = cooldown_calls[0].args
assert seconds == 60 assert seconds == _SPLIT_FAILURE_COOLDOWN_SECONDS
assert "session_split_failed" in str(error) assert "session_split_failed" in str(error)
@@ -1,8 +1,10 @@
"""Regression tests: a failed compression split must arm the failure cooldown. """Regression tests: a failed compression split must arm the failure cooldown.
Extracted verbatim from PR #98137's test file (author: vsd2807); the sibling Extracted from PR #98137's test file (author: vsd2807); two sibling unit
timeout-reconciliation tests were not carried because that production path was tests were dropped as duplicates of existing coverage in
not salvaged (blocking review on #98137). test_compression_rotation_state.py / test_context_compressor.py, and the
timeout-reconciliation tests were not carried because that production path
was not salvaged (blocking review on #98137).
Issue #97948 symptom B: without the cooldown, the turn after a Issue #97948 symptom B: without the cooldown, the turn after a
session_split_failed abort immediately re-runs the identical doomed session_split_failed abort immediately re-runs the identical doomed
@@ -12,34 +14,6 @@ compression.
import time import time
from unittest.mock import MagicMock from unittest.mock import MagicMock
def test_split_failure_records_cooldown():
"""After a session split failure, compression failure cooldown is recorded."""
from agent.context_compressor import ContextCompressor
compressor = ContextCompressor.__new__(ContextCompressor)
compressor._summary_failure_cooldown_until = 0.0
compressor._last_summary_error = None
compressor._cooldown_persist_failed = False
compressor._session_db = None
compressor._session_id = ""
compressor._record_compression_failure_cooldown(60, "session_split_failed: test")
assert compressor._summary_failure_cooldown_until > time.monotonic()
assert compressor._last_summary_error == "session_split_failed: test"
def test_cooldown_blocks_automatic_compression():
from agent.context_compressor import ContextCompressor
compressor = ContextCompressor.__new__(ContextCompressor)
compressor._summary_failure_cooldown_until = time.monotonic() + 60.0
compressor._last_summary_error = "session_split_failed"
compressor._cooldown_persist_failed = False
compressor.quiet_mode = True
assert compressor._automatic_compression_blocked_locally() is True
def test_manual_compress_bypasses_cooldown(): def test_manual_compress_bypasses_cooldown():
"""Manual /compress (force=True) bypasses cooldown — existing behavior preserved.""" """Manual /compress (force=True) bypasses cooldown — existing behavior preserved."""