From de4155b1bb7f67f3f31f01dcda5ce6de6a30326a Mon Sep 17 00:00:00 2001 From: fangliquanflq Date: Mon, 31 Aug 2026 01:25:57 +0800 Subject: [PATCH] fix(compression): report total ceiling expiry accurately --- agent/conversation_compression.py | 18 ++++- run_agent.py | 73 ++++++++++++++----- .../test_compress_context_progress_timeout.py | 67 +++++++++++++++++ 3 files changed, 140 insertions(+), 18 deletions(-) diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 4d4f086a59..a6b0968cfb 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -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, ) diff --git a/run_agent.py b/run_agent.py index 4958b50cf9..642986f700 100644 --- a/run_agent.py +++ b/run_agent.py @@ -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, diff --git a/tests/agent/test_compress_context_progress_timeout.py b/tests/agent/test_compress_context_progress_timeout.py index 1a81a1d495..f0e52986d1 100644 --- a/tests/agent/test_compress_context_progress_timeout.py +++ b/tests/agent/test_compress_context_progress_timeout.py @@ -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