360 lines
12 KiB
Python
360 lines
12 KiB
Python
"""Stall-interrupted preflight compression must persist a durable backoff.
|
|
|
|
#96775: an explicit /stop after the summary stream has already crossed the
|
|
no-progress stall window restores the original transcript but must record a
|
|
stall-specific cooldown so the next automatic turn does not re-enter the
|
|
same strategy. An ordinary early /stop stays cooldown-neutral.
|
|
|
|
Native Codex app-server compaction is a sibling path (#75364) and is not
|
|
covered here. Stall classification reads the commit fence's progress clock,
|
|
not Chat Completions vs Responses frame shapes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from agent.auxiliary_client import AuxiliaryExplicitCancellation
|
|
from agent.conversation_compression import (
|
|
STALL_INTERRUPTED_FAILURE_CLASS,
|
|
CompressionCommitFence,
|
|
compress_context,
|
|
compression_attempt_stalled,
|
|
)
|
|
from hermes_state import SessionDB
|
|
|
|
|
|
def _build_agent(tmp_path: Path, session_id: str = "STALL_INTERRUPT_96775"):
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session(session_id, source="cli")
|
|
with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}):
|
|
from run_agent import AIAgent
|
|
|
|
agent = AIAgent(
|
|
api_key="test-key",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
model="test/model",
|
|
quiet_mode=True,
|
|
session_db=db,
|
|
session_id=session_id,
|
|
skip_context_files=True,
|
|
skip_memory=True,
|
|
)
|
|
agent._compression_feasibility_checked = True
|
|
agent.compression_in_place = True
|
|
agent._cached_system_prompt = "sys"
|
|
agent.context_compressor.threshold_tokens = 1_000
|
|
return db, agent
|
|
|
|
|
|
def _messages():
|
|
return [{"role": "user", "content": f"m{i}"} for i in range(20)]
|
|
|
|
|
|
def _age_fence(fence: CompressionCommitFence, idle_seconds: float) -> None:
|
|
fence._last_progress = time.monotonic() - float(idle_seconds)
|
|
|
|
|
|
class TestStallClassificationIsFenceIdle:
|
|
def test_fresh_fence_is_not_stalled(self):
|
|
fence = CompressionCommitFence()
|
|
assert compression_attempt_stalled(
|
|
commit_fence=fence,
|
|
started_at=time.monotonic(),
|
|
idle_timeout_seconds=1.0,
|
|
) is False
|
|
|
|
def test_idle_fence_is_stalled(self):
|
|
fence = CompressionCommitFence()
|
|
_age_fence(fence, 2.0)
|
|
assert compression_attempt_stalled(
|
|
commit_fence=fence,
|
|
started_at=time.monotonic(),
|
|
idle_timeout_seconds=1.0,
|
|
) is True
|
|
|
|
def test_recent_progress_is_not_stalled(self):
|
|
fence = CompressionCommitFence()
|
|
_age_fence(fence, 2.0)
|
|
fence.touch_progress()
|
|
assert compression_attempt_stalled(
|
|
commit_fence=fence,
|
|
started_at=time.monotonic() - 5.0,
|
|
idle_timeout_seconds=1.0,
|
|
) is False
|
|
|
|
def test_chat_and_responses_share_the_fence_clock(self):
|
|
"""Transport-agnostic: both APIs tick the same fence or they don't."""
|
|
chat_fence = CompressionCommitFence()
|
|
responses_fence = CompressionCommitFence()
|
|
_age_fence(chat_fence, 2.0)
|
|
responses_fence.touch_progress()
|
|
assert compression_attempt_stalled(
|
|
commit_fence=chat_fence,
|
|
started_at=time.monotonic(),
|
|
idle_timeout_seconds=1.0,
|
|
) is True
|
|
assert compression_attempt_stalled(
|
|
commit_fence=responses_fence,
|
|
started_at=time.monotonic(),
|
|
idle_timeout_seconds=1.0,
|
|
) is False
|
|
|
|
|
|
class TestEarlyStopStaysNeutral:
|
|
def test_explicit_interrupt_before_stall_does_not_arm_cooldown(
|
|
self, tmp_path: Path
|
|
):
|
|
db, agent = _build_agent(tmp_path, "EARLY_STOP_96775")
|
|
original = _messages()
|
|
live = copy.deepcopy(original)
|
|
fence = CompressionCommitFence()
|
|
|
|
def _early_stop(messages, **_kwargs):
|
|
messages[0]["content"] = "must be rolled back"
|
|
raise AuxiliaryExplicitCancellation()
|
|
|
|
agent.context_compressor.compress = _early_stop
|
|
|
|
compressed, _prompt = compress_context(
|
|
agent,
|
|
live,
|
|
"sys",
|
|
approx_tokens=50_000,
|
|
commit_fence=fence,
|
|
)
|
|
|
|
assert compressed == original
|
|
assert live == original
|
|
assert db.get_compression_lock_holder("EARLY_STOP_96775") is None
|
|
assert db.get_compression_failure_cooldown("EARLY_STOP_96775") is None
|
|
assert agent.context_compressor.should_compress(50_000) is True
|
|
db.append_message("EARLY_STOP_96775", "assistant", "still writable")
|
|
|
|
|
|
class TestStallInterruptedBackoff:
|
|
def test_aux_explicit_cancel_after_stall_persists_backoff(
|
|
self, tmp_path: Path, monkeypatch
|
|
):
|
|
monkeypatch.setattr(
|
|
"agent.conversation_compression.resolve_context_compression_timeouts",
|
|
lambda compression_cfg=None: (1.0, 10.0),
|
|
)
|
|
db, agent = _build_agent(tmp_path, "STALL_AUX_CANCEL")
|
|
original = _messages()
|
|
live = copy.deepcopy(original)
|
|
fence = CompressionCommitFence()
|
|
compress_calls = {"n": 0}
|
|
|
|
def _stalled_then_stop(messages, **_kwargs):
|
|
compress_calls["n"] += 1
|
|
_age_fence(fence, 2.0)
|
|
messages[0]["content"] = "must be rolled back"
|
|
raise AuxiliaryExplicitCancellation()
|
|
|
|
agent.context_compressor.compress = _stalled_then_stop
|
|
|
|
compressed, _prompt = compress_context(
|
|
agent,
|
|
live,
|
|
"sys",
|
|
approx_tokens=50_000,
|
|
commit_fence=fence,
|
|
)
|
|
|
|
assert compressed == original
|
|
assert live == original
|
|
assert db.get_compression_lock_holder("STALL_AUX_CANCEL") is None
|
|
state = db.get_compression_failure_cooldown("STALL_AUX_CANCEL")
|
|
assert state is not None
|
|
assert STALL_INTERRUPTED_FAILURE_CLASS in str(state["error"])
|
|
assert "msgs=20" in str(state["error"])
|
|
assert agent.context_compressor.should_compress(50_000) is False
|
|
|
|
compress_calls["n"] = 0
|
|
again, _ = compress_context(
|
|
agent,
|
|
copy.deepcopy(original),
|
|
"sys",
|
|
approx_tokens=50_000,
|
|
commit_fence=CompressionCommitFence(),
|
|
)
|
|
assert again == original
|
|
assert compress_calls["n"] == 0
|
|
|
|
db.append_message("STALL_AUX_CANCEL", "assistant", "still writable")
|
|
|
|
def test_commit_fence_cancel_after_stall_persists_backoff(
|
|
self, tmp_path: Path, monkeypatch
|
|
):
|
|
monkeypatch.setattr(
|
|
"agent.conversation_compression.resolve_context_compression_timeouts",
|
|
lambda compression_cfg=None: (1.0, 10.0),
|
|
)
|
|
db, agent = _build_agent(tmp_path, "STALL_FENCE_CANCEL")
|
|
original = _messages()
|
|
live = copy.deepcopy(original)
|
|
fence = CompressionCommitFence()
|
|
|
|
def _summary_then_cancel(messages, **_kwargs):
|
|
_age_fence(fence, 2.0)
|
|
fence.cancel_before_commit()
|
|
return [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] summary"},
|
|
{"role": "assistant", "content": "tail"},
|
|
]
|
|
|
|
agent.context_compressor.compress = _summary_then_cancel
|
|
|
|
compressed, _prompt = compress_context(
|
|
agent,
|
|
live,
|
|
"sys",
|
|
approx_tokens=50_000,
|
|
commit_fence=fence,
|
|
)
|
|
|
|
assert compressed == original
|
|
assert live == original
|
|
state = db.get_compression_failure_cooldown("STALL_FENCE_CANCEL")
|
|
assert state is not None
|
|
assert STALL_INTERRUPTED_FAILURE_CLASS in str(state["error"])
|
|
assert agent.context_compressor.should_compress(50_000) is False
|
|
db.append_message("STALL_FENCE_CANCEL", "user", "next turn")
|
|
|
|
def test_commit_fence_cancel_with_fresh_progress_stays_neutral(
|
|
self, tmp_path: Path, monkeypatch
|
|
):
|
|
monkeypatch.setattr(
|
|
"agent.conversation_compression.resolve_context_compression_timeouts",
|
|
lambda compression_cfg=None: (1.0, 10.0),
|
|
)
|
|
db, agent = _build_agent(tmp_path, "FRESH_FENCE_CANCEL")
|
|
original = _messages()
|
|
live = copy.deepcopy(original)
|
|
fence = CompressionCommitFence()
|
|
|
|
def _healthy_then_cancel(messages, **_kwargs):
|
|
fence.touch_progress()
|
|
fence.cancel_before_commit()
|
|
return [
|
|
{"role": "user", "content": "[CONTEXT COMPACTION] summary"},
|
|
{"role": "assistant", "content": "tail"},
|
|
]
|
|
|
|
agent.context_compressor.compress = _healthy_then_cancel
|
|
|
|
compressed, _prompt = compress_context(
|
|
agent,
|
|
live,
|
|
"sys",
|
|
approx_tokens=50_000,
|
|
commit_fence=fence,
|
|
)
|
|
|
|
assert compressed == original
|
|
assert db.get_compression_failure_cooldown("FRESH_FENCE_CANCEL") is None
|
|
assert agent.context_compressor.should_compress(50_000) is True
|
|
|
|
def test_manual_force_bypasses_stall_backoff(
|
|
self, tmp_path: Path, monkeypatch
|
|
):
|
|
monkeypatch.setattr(
|
|
"agent.conversation_compression.resolve_context_compression_timeouts",
|
|
lambda compression_cfg=None: (1.0, 10.0),
|
|
)
|
|
db, agent = _build_agent(tmp_path, "FORCE_BYPASS_STALL")
|
|
original = _messages()
|
|
fence = CompressionCommitFence()
|
|
|
|
def _stalled_then_stop(messages, **_kwargs):
|
|
_age_fence(fence, 2.0)
|
|
raise AuxiliaryExplicitCancellation()
|
|
|
|
agent.context_compressor.compress = _stalled_then_stop
|
|
compress_context(
|
|
agent,
|
|
copy.deepcopy(original),
|
|
"sys",
|
|
approx_tokens=50_000,
|
|
commit_fence=fence,
|
|
)
|
|
assert agent.context_compressor.should_compress(50_000) is False
|
|
|
|
forced_calls = {"n": 0}
|
|
|
|
def _forced(messages, **kwargs):
|
|
forced_calls["n"] += 1
|
|
assert kwargs.get("force") is True
|
|
return messages
|
|
|
|
agent.context_compressor.compress = _forced
|
|
compress_context(
|
|
agent,
|
|
copy.deepcopy(original),
|
|
"sys",
|
|
approx_tokens=50_000,
|
|
force=True,
|
|
commit_fence=CompressionCommitFence(),
|
|
)
|
|
assert forced_calls["n"] == 1
|
|
|
|
def test_stall_backoff_merges_with_longer_active_deadline(
|
|
self, tmp_path: Path, monkeypatch
|
|
):
|
|
monkeypatch.setattr(
|
|
"agent.conversation_compression.resolve_context_compression_timeouts",
|
|
lambda compression_cfg=None: (1.0, 10.0),
|
|
)
|
|
db, agent = _build_agent(tmp_path, "MERGE_MAX_STALL")
|
|
agent.context_compressor._record_compression_failure_cooldown(
|
|
900.0, "prior timeout"
|
|
)
|
|
before = db.get_compression_failure_cooldown("MERGE_MAX_STALL")
|
|
assert before is not None
|
|
prior_until = float(before["cooldown_until"])
|
|
|
|
fence = CompressionCommitFence()
|
|
|
|
def _stalled_then_stop(messages, **_kwargs):
|
|
_age_fence(fence, 2.0)
|
|
raise AuxiliaryExplicitCancellation()
|
|
|
|
agent.context_compressor.compress = _stalled_then_stop
|
|
# force=True so the pre-existing cooldown does not skip the attempt
|
|
# before the stall-interrupt restore/merge path can run.
|
|
compress_context(
|
|
agent,
|
|
_messages(),
|
|
"sys",
|
|
approx_tokens=50_000,
|
|
force=True,
|
|
commit_fence=fence,
|
|
)
|
|
|
|
after = db.get_compression_failure_cooldown("MERGE_MAX_STALL")
|
|
assert after is not None
|
|
assert float(after["cooldown_until"]) >= prior_until - 1.0
|
|
assert STALL_INTERRUPTED_FAILURE_CLASS in str(after["error"])
|
|
|
|
|
|
def test_session_db_cooldown_write_does_not_shorten_longer_deadline(
|
|
tmp_path: Path,
|
|
):
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("merge-max", source="cli")
|
|
later = time.time() + 900.0
|
|
sooner = time.time() + 60.0
|
|
db.record_compression_failure_cooldown("merge-max", later, "timeout")
|
|
db.record_compression_failure_cooldown(
|
|
"merge-max", sooner, STALL_INTERRUPTED_FAILURE_CLASS
|
|
)
|
|
row = db.get_compression_failure_cooldown("merge-max")
|
|
assert row is not None
|
|
assert float(row["cooldown_until"]) >= later - 1.0
|
|
assert row["error"] == STALL_INTERRUPTED_FAILURE_CLASS
|