test(compaction): move the bench probe rebuild test into its own file

Code-health ratchet: tests/agent/test_context_compressor.py grew to
3847 > 3829. New tests go in a topical test_<stem>_<topic>.py.
This commit is contained in:
kshitijk4poor
2026-10-06 09:48:31 +05:30
committed by kshitij
parent d6daf4bdd2
commit 164d024bb3
2 changed files with 43 additions and 24 deletions

View File

@@ -3685,30 +3685,6 @@ class TestPreLlmFeasibilityCheck:
mock_gen.assert_called_once()
assert compressor._prellm_skip_count == 0
def test_bench_probe_deadline_survives_a_compressor_rebuild(self, compressor, tmp_path):
"""The gateway rebuilds the agent on cache eviction: a fresh compressor bound to the same
session must resume the persisted bench window, not restart it (else the probe starves)."""
db = SessionDB(db_path=tmp_path / "state.db")
db.create_session("s1", "cli")
db.set_compression_fallback_streak("s1", 2)
compressor.bind_session_state(db, "s1")
window = compressor._ANTI_THRASH_RECOVERY_SECONDS
with patch("agent.context_compressor.time.time", return_value=1000.0), \
patch.object(compressor, "_generate_summary", return_value="LLM summary") as mock_gen:
compressor.compress(self._make_messages(), force=False) # benched, arms the window
mock_gen.assert_not_called()
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
rebuilt = ContextCompressor(
model="test/model", threshold_percent=0.85, protect_first_n=2, protect_last_n=2, quiet_mode=True,
)
rebuilt.bind_session_state(db, "s1")
with patch("agent.context_compressor.time.time", return_value=1000.0 + window + 1), \
patch.object(rebuilt, "_generate_summary", return_value="LLM summary") as mock_gen:
rebuilt.compress(self._make_messages(), force=False)
mock_gen.assert_called_once()
def test_skip_fires_on_fat_tail_small_middle(self, compressor):
"""The target scenario from #60451: a tool-heavy transcript whose
protected tail already holds most of the tokens, leaving a tiny

View File

@@ -0,0 +1,43 @@
"""The summary-model bench (fallback streak >= 2) and its persisted probe window."""
from unittest.mock import patch
from agent.context_compressor import ContextCompressor
from hermes_state import SessionDB
def _compressor():
with patch("agent.context_compressor.get_model_context_length", return_value=100000):
c = ContextCompressor(model="test/model", threshold_percent=0.85, protect_first_n=2, protect_last_n=2, quiet_mode=True)
_ = c.context_length
return c
def _messages(n_pairs=10):
msgs = [{"role": "system", "content": "system prompt"}]
for i in range(n_pairs):
msgs += [{"role": "user", "content": f"question {i}"}, {"role": "assistant", "content": "short reply"}]
return msgs
def test_bench_probe_deadline_survives_a_compressor_rebuild(tmp_path):
"""The gateway rebuilds the agent on cache eviction: a fresh compressor bound to the same
session must resume the persisted bench window, not restart it (else the probe starves)."""
db = SessionDB(db_path=tmp_path / "state.db")
db.create_session("s1", "cli")
db.set_compression_fallback_streak("s1", 2)
compressor = _compressor()
compressor.bind_session_state(db, "s1")
window = compressor._ANTI_THRASH_RECOVERY_SECONDS
with patch("agent.context_compressor.time.time", return_value=1000.0), \
patch.object(compressor, "_generate_summary", return_value="LLM summary") as mock_gen:
compressor.compress(_messages(), force=False) # benched, arms the window
mock_gen.assert_not_called()
rebuilt = _compressor()
rebuilt.bind_session_state(db, "s1")
with patch("agent.context_compressor.time.time", return_value=1000.0 + window + 1), \
patch.object(rebuilt, "_generate_summary", return_value="LLM summary") as mock_gen:
rebuilt.compress(_messages(), force=False)
mock_gen.assert_called_once()