fix(compression): report total ceiling expiry accurately
This commit is contained in:
@@ -1102,6 +1102,7 @@ def _retry_compression_on_fallback_chain(
|
||||
idle_timeout_seconds: float,
|
||||
total_ceiling_seconds: float,
|
||||
on_commit_overrun: Optional[Callable[[float, float], None]] = None,
|
||||
on_timeout_cause: Optional[Callable[[bool, bool], None]] = None,
|
||||
telemetry_agent: Any = None,
|
||||
new_fence: Optional[Callable[[], CompressionCommitFence]] = None,
|
||||
) -> Optional[Tuple[list, str]]:
|
||||
@@ -1176,6 +1177,7 @@ def _retry_compression_on_fallback_chain(
|
||||
idle_timeout_seconds=idle,
|
||||
total_ceiling_seconds=ceiling,
|
||||
on_commit_overrun=on_commit_overrun,
|
||||
on_timeout_cause=on_timeout_cause,
|
||||
fence=retry_fence,
|
||||
telemetry_agent=telemetry_agent,
|
||||
stall_fallback=False,
|
||||
@@ -1213,6 +1215,7 @@ def run_compress_context_with_progress_timeout(
|
||||
idle_timeout_seconds: float,
|
||||
total_ceiling_seconds: float,
|
||||
on_timeout: Optional[Callable[[float, float, float], None]] = None,
|
||||
on_timeout_cause: Optional[Callable[[bool, bool], None]] = None,
|
||||
on_commit_overrun: Optional[Callable[[float, float], None]] = None,
|
||||
fence: Optional[CompressionCommitFence] = None,
|
||||
telemetry_agent: Any = None,
|
||||
@@ -1247,7 +1250,10 @@ def run_compress_context_with_progress_timeout(
|
||||
|
||||
``system_prompt_fallback`` may be a string or a zero-arg callable resolved
|
||||
only on the timeout path, so successful compression never pays for (or
|
||||
fails on) an eager prompt rebuild.
|
||||
fails on) an eager prompt rebuild. ``on_timeout_cause`` receives whether
|
||||
the total ceiling expired and whether provider progress was observed before
|
||||
``on_timeout`` runs, allowing hosts to report the timeout accurately while
|
||||
preserving the existing three-argument timeout callback contract.
|
||||
|
||||
``stall_fallback`` (default on) makes an aborted stall attempt the
|
||||
configured ``auxiliary.compression.fallback_chain`` once — pinned onto a
|
||||
@@ -1395,6 +1401,15 @@ def run_compress_context_with_progress_timeout(
|
||||
# another automatic attempt cannot overlap the unchanged source.
|
||||
fence.retain_compression_lock_until_worker_done()
|
||||
|
||||
if on_timeout_cause is not None:
|
||||
try:
|
||||
on_timeout_cause(total_exhausted, fence.progress_observed)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"compress_context timeout-cause callback failed",
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
cancelled: Optional[bool] = None
|
||||
while cancelled is None:
|
||||
# F1: ``begin_commit`` retains the fence lock until
|
||||
@@ -1493,6 +1508,7 @@ def run_compress_context_with_progress_timeout(
|
||||
idle_timeout_seconds=idle,
|
||||
total_ceiling_seconds=ceiling,
|
||||
on_commit_overrun=on_commit_overrun,
|
||||
on_timeout_cause=on_timeout_cause,
|
||||
telemetry_agent=telemetry_agent,
|
||||
new_fence=new_fence,
|
||||
)
|
||||
|
||||
+56
-17
@@ -8179,15 +8179,35 @@ class AIAgent:
|
||||
)
|
||||
return system_message or ""
|
||||
|
||||
timeout_cause = {
|
||||
"total_exhausted": False,
|
||||
"progress_observed": False,
|
||||
}
|
||||
|
||||
def _on_timeout_cause(total_exhausted, progress_observed):
|
||||
timeout_cause["total_exhausted"] = total_exhausted
|
||||
timeout_cause["progress_observed"] = progress_observed
|
||||
|
||||
def _on_timeout(idle, waited, since_progress):
|
||||
logger.warning(
|
||||
"Context compression made no progress for %.1fs "
|
||||
"(total wait %.1fs, ceiling %.1fs); continuing without "
|
||||
"compression",
|
||||
since_progress,
|
||||
waited,
|
||||
total_ceiling,
|
||||
)
|
||||
total_exhausted = timeout_cause["total_exhausted"]
|
||||
progress_observed = timeout_cause["progress_observed"]
|
||||
if total_exhausted:
|
||||
logger.warning(
|
||||
"Context compression reached its total ceiling "
|
||||
"after %.1fs (progress observed=%s); continuing "
|
||||
"without compression",
|
||||
waited,
|
||||
progress_observed,
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
"Context compression made no progress for %.1fs "
|
||||
"(total wait %.1fs, ceiling %.1fs); continuing "
|
||||
"without compression",
|
||||
since_progress,
|
||||
waited,
|
||||
total_ceiling,
|
||||
)
|
||||
touch = getattr(self, "_touch_activity", None)
|
||||
if callable(touch):
|
||||
try:
|
||||
@@ -8207,10 +8227,14 @@ class AIAgent:
|
||||
record = getattr(compressor, "record_timeout_failure", None)
|
||||
if callable(record):
|
||||
try:
|
||||
record(
|
||||
"host compress_context timeout "
|
||||
reason = (
|
||||
"host compress_context total ceiling "
|
||||
"exhausted"
|
||||
if total_exhausted
|
||||
else "host compress_context timeout "
|
||||
"(no summary progress)"
|
||||
)
|
||||
record(reason)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"failed to record compress_context timeout "
|
||||
@@ -8219,13 +8243,27 @@ class AIAgent:
|
||||
)
|
||||
emit = getattr(self, "_emit_warning", None)
|
||||
if callable(emit):
|
||||
emit(
|
||||
"⚠ Context compression timed out "
|
||||
f"after {idle:.1f}s with no output from the summary "
|
||||
"model. No messages were dropped — continuing without "
|
||||
"compression. Run /compress to retry, /new for a clean "
|
||||
"session, or check auxiliary.compression."
|
||||
)
|
||||
if total_exhausted:
|
||||
progress = (
|
||||
" after summary output was observed"
|
||||
if progress_observed
|
||||
else ""
|
||||
)
|
||||
emit(
|
||||
"⚠ Context compression reached its total ceiling "
|
||||
f"after {waited:.1f}s{progress}. No messages were "
|
||||
"dropped — continuing without compression. Run "
|
||||
"/compress to retry or /new for a clean session."
|
||||
)
|
||||
else:
|
||||
emit(
|
||||
"⚠ Context compression timed out "
|
||||
f"after {idle:.1f}s with no output from the summary "
|
||||
"model. No messages were dropped — continuing "
|
||||
"without compression. Run /compress to retry, /new "
|
||||
"for a clean session, or check "
|
||||
"auxiliary.compression."
|
||||
)
|
||||
|
||||
def _on_commit_overrun(waited, ceiling):
|
||||
# Commit-phase ceiling breach: the SessionDB mutation is in
|
||||
@@ -8259,6 +8297,7 @@ class AIAgent:
|
||||
idle_timeout_seconds=idle_timeout,
|
||||
total_ceiling_seconds=total_ceiling,
|
||||
on_timeout=_on_timeout,
|
||||
on_timeout_cause=_on_timeout_cause,
|
||||
on_commit_overrun=_on_commit_overrun,
|
||||
fence=active_fence,
|
||||
telemetry_agent=self,
|
||||
|
||||
@@ -483,6 +483,73 @@ class TestCompressContextForwarderOwnsTimeout:
|
||||
provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT,
|
||||
)
|
||||
|
||||
def test_owned_total_ceiling_reports_progress_accurately(self, monkeypatch):
|
||||
from run_agent import AIAgent
|
||||
from agent.context_compressor import ContextCompressor
|
||||
|
||||
agent = object.__new__(AIAgent)
|
||||
agent.session_id = "s1"
|
||||
agent._cached_system_prompt = "sys"
|
||||
agent._emit_warning = MagicMock()
|
||||
agent._touch_activity = MagicMock()
|
||||
agent._build_system_prompt = MagicMock(return_value="sys")
|
||||
agent._conversation_root_id = MagicMock(return_value=None)
|
||||
agent.context_compressor = MagicMock()
|
||||
agent.context_compressor._consecutive_timeout_failures = 0
|
||||
agent.context_compressor.record_timeout_failure = (
|
||||
ContextCompressor.record_timeout_failure.__get__(
|
||||
agent.context_compressor, MagicMock
|
||||
)
|
||||
)
|
||||
agent.context_compressor._record_compression_failure_cooldown = MagicMock()
|
||||
|
||||
release = threading.Event()
|
||||
|
||||
def streaming_compress(agent_obj, messages, system_message, **kwargs):
|
||||
fence = kwargs["commit_fence"]
|
||||
while not fence.deadline_exceeded:
|
||||
fence.touch_progress()
|
||||
time.sleep(0.005)
|
||||
release.wait(timeout=2)
|
||||
return messages, "sys"
|
||||
|
||||
monkeypatch.setattr(
|
||||
"agent.conversation_compression.compress_context",
|
||||
streaming_compress,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"agent.conversation_compression.resolve_context_compression_timeouts",
|
||||
lambda compression_cfg=None: (0.05, 0.15),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"agent.conversation_compression.resolve_compression_fallback_route",
|
||||
lambda: None,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"agent.portal_tags.get_conversation_context",
|
||||
lambda: object(),
|
||||
)
|
||||
|
||||
original = [{"role": "user", "content": "stay"}]
|
||||
try:
|
||||
out_msgs, out_prompt = AIAgent._compress_context(
|
||||
agent, original, "sys"
|
||||
)
|
||||
finally:
|
||||
release.set()
|
||||
|
||||
assert out_msgs is original
|
||||
assert out_prompt == "sys"
|
||||
warning = agent._emit_warning.call_args.args[0]
|
||||
assert "total ceiling" in warning
|
||||
assert "summary output was observed" in warning
|
||||
assert "no output" not in warning
|
||||
cooldown_error = (
|
||||
agent.context_compressor._record_compression_failure_cooldown
|
||||
.call_args.args[1]
|
||||
)
|
||||
assert "total ceiling exhausted" in cooldown_error
|
||||
|
||||
def test_fallback_prompt_resolved_lazily_on_timeout(self, monkeypatch):
|
||||
"""Eager prompt rebuild must not run before compression starts."""
|
||||
from run_agent import AIAgent
|
||||
|
||||
Reference in New Issue
Block a user