Address sweeper review: safe should_compress_info + cover all guards

- ContextEngine.should_compress_info() default impl so plugin engines
  (e.g. _StubEngine) don't raise AttributeError at the call site.
- Centralise warning/reset in AIAgent._warn_context_overflow_blocked /
  _clear_context_overflow_warn so turn-context and conversation-loop guards
  share identical dedup logic and reset on the real compression boundary.
- Cover conversation_loop.py pre-API (~L1007) and loop-compaction (~L4774)
  guards, not just the turn-context preflight.
- _FakeAgent mirrors the two helpers; test suite green (219 passed).

Fixes #62708

agent/context_engine.py

@@ -146,6 +146,18 @@ class ContextEngine(ABC):
     def should_compress(self, prompt_tokens: int = None) -> bool:
         """Return True if compaction should fire this turn."""
 
+    def should_compress_info(self, prompt_tokens: int = None) -> "tuple[bool, str | None]":
+        """Return ``(should_compress, reason)``.
+
+        The base implementation is backward-compatible: engines that only
+        implement ``should_compress`` get ``(should_compress(prompt_tokens),
+        None)``. Concrete engines with richer block reasons (e.g. a
+        summary-LLM cooldown or an anti-thrashing guard) override this to
+        surface a human-readable reason so callers can warn the user instead
+        of silently skipping compression. Added for the silent-overflow
+        warning fix (#62625) so plugin engines don't raise AttributeError.
+        """
+
     @abstractmethod
     def compress(
         self,

agent/conversation_loop.py

@@ -1368,6 +1368,33 @@ def run_conversation(
             agent._api_call_count = api_call_count
             agent.iteration_budget.refund()
             continue
+        elif (
+            agent.compression_enabled
+            and len(messages) > 1
+            and compression_attempts < 3
+            and not _defer_preflight(request_pressure_tokens)
+            and _compression_cooldown
+        ):
+            # Over threshold (would compress) but blocked by the summary-LLM
+            # cooldown. Surface a deduped warning so the user isn't left with a
+            # silently growing context. Mirrors the turn-context preflight and
+            # the loop-compaction guards (silent-overflow fix #62625).
+            _block_reason = None
+            try:
+                _block_reason = _compressor.should_compress_info(
+                    request_pressure_tokens
+                )[1]
+            except Exception:
+                _block_reason = None
+            if not _block_reason:
+                _block_reason = (
+                    f"cooldown:{int(_compression_cooldown.get('remaining_seconds', 0.0))}"
+                )
+            agent._warn_context_overflow_blocked(
+                _block_reason,
+                request_pressure_tokens,
+                int(getattr(_compressor, "threshold_tokens", 0) or 0),
+            )
         
         # Thinking spinner for quiet mode (animated during API call)
         thinking_spinner = None
@@ -5445,6 +5472,25 @@ def run_conversation(
                     conversation_history = conversation_history_after_compression(
                         agent, messages, conversation_history
                     )
+                elif agent.compression_enabled:
+                    # Over threshold but compression is blocked (summary-LLM
+                    # cooldown or anti-thrashing). Surface a deduped warning so
+                    # the user isn't left with a silently growing context that
+                    # eventually hits the hard provider limit. Mirrors the
+                    # turn-context preflight guard (silent-overflow fix #62625).
+                    _block_reason = None
+                    _info = getattr(_compressor, "should_compress_info", None)
+                    if _info is not None:
+                        try:
+                            _block_reason = _info(_real_tokens)[1]
+                        except Exception:
+                            _block_reason = None
+                    if _block_reason:
+                        agent._warn_context_overflow_blocked(
+                            _block_reason,
+                            _real_tokens,
+                            int(getattr(_compressor, "threshold_tokens", 0) or 0),
+                        )
                 
                 # Save session log incrementally (so progress is visible even if interrupted)
                 agent._session_messages = messages

agent/turn_context.py

@@ -786,6 +786,10 @@ def build_turn_context(
                 _compress_block_reason = _compressor.should_compress_info(_preflight_tokens)[1]
         if _should_compress_now:
             _preflight_compressed = True
+            # Compression is actually running (block cleared / was never
+            # blocked) — reset the dedup so a future blocked-over-threshold
+            # turn can warn again. Real session boundary.
+            agent._clear_context_overflow_warn()
             logger.info(
                 "Preflight compression: ~%s tokens >= %s threshold (model %s, ctx %s)",
                 f"{_preflight_tokens:,}",
@@ -864,24 +868,19 @@ def build_turn_context(
             # answering — the conversation hits the hard provider token limit
             # with no explanation. Surface a deduped warning so the user can
             # take action (/new or /compress) instead of hitting a silent hang.
-            # Dedup on the *kind* of block (cooldown / ineffective), not the
-            # exact countdown, so a cooldown ticking down 30→29→… doesn't re-fire
-            # the warning every turn.
-            _warn_kind = _compress_block_reason.split(":", 1)[0]
-            _warn_key = ("ctx_overflow_blocked", _warn_kind)
-            if getattr(agent, "_last_ctx_overflow_warn", None) != _warn_key:
-                agent._last_ctx_overflow_warn = _warn_key
-                agent._emit_warning(
-                    f"⚠ Context is over the compression threshold "
-                    f"(~{_preflight_tokens:,} tokens >= {_compressor.threshold_tokens:,}) "
-                    f"but compression is currently blocked ({_compress_block_reason}). "
-                    f"The model may stop responding. Run /new to start a fresh session "
-                    f"or /compress to retry immediately."
-                )
+            agent._warn_context_overflow_blocked(
+                _compress_block_reason,
+                _preflight_tokens,
+                _compressor.threshold_tokens,
+            )
         else:
             # Sub-threshold and unblocked — allow the overflow warning to fire
             # again next time the context is over threshold but blocked.
-            agent._last_ctx_overflow_warn = None
+            # getattr guard: test doubles built via object.__new__ lack the
+            # method (gateway test-double pitfall) — treat absence as no-op.
+            _clear_warn = getattr(agent, "_clear_context_overflow_warn", None)
+            if callable(_clear_warn):
+                _clear_warn()
             # ── Engine-driven sub-threshold preflight maintenance (#20316) ──
             # None of the threshold-path branches fired (not deferred, no
             # failure cooldown, not codex-native, and should_compress() said

run_agent.py

@@ -931,6 +931,46 @@ class AIAgent:
             except Exception:
                 logger.debug("status_callback error in _emit_warning", exc_info=True)
 
+    def _warn_context_overflow_blocked(
+        self, reason: str, preflight_tokens: int, threshold_tokens: int
+    ) -> None:
+        """Surface a deduped warning when the context is over the compression
+        threshold but compression is blocked (summary-LLM cooldown or
+        anti-thrashing).
+
+        Without this signal the session keeps growing until the model silently
+        stops answering — the conversation hits the hard provider token limit
+        with no explanation. Centralised here so every caller that checks
+        ``should_compress_info`` (turn-context preflight, conversation-loop
+        guards) shares identical dedup/reset logic.
+
+        Dedup is on the *kind* of block (``cooldown`` / ``ineffective``), not the
+        exact countdown string, so a cooldown ticking down 30→29→… doesn't
+        re-fire the warning every turn. The dedup key is cleared when the block
+        clears (see ``_clear_context_overflow_warn``), so the warning can fire
+        again on the next blocked-over-threshold turn.
+        """
+        _warn_kind = (reason or "unknown").split(":", 1)[0]
+        _warn_key = ("ctx_overflow_blocked", _warn_kind)
+        if getattr(self, "_last_ctx_overflow_warn", None) != _warn_key:
+            self._last_ctx_overflow_warn = _warn_key
+            self._emit_warning(
+                f"⚠ Context is over the compression threshold "
+                f"(~{preflight_tokens:,} tokens >= {threshold_tokens:,}) "
+                f"but compression is currently blocked ({reason}). "
+                f"The model may stop responding. Run /new to start a fresh "
+                f"session or /compress to retry immediately."
+            )
+
+    def _clear_context_overflow_warn(self) -> None:
+        """Reset the dedup state for the blocked-overflow warning.
+
+        Call this whenever compression is no longer blocked while the context
+        is over threshold (e.g. the cooldown elapsed, or compression ran
+        successfully), so the warning can re-fire on the next blocked turn.
+        """
+        self._last_ctx_overflow_warn = None
+
     def _emit_notice(self, notice) -> None:
         """Fire a structured ``AgentNotice`` to the active driver (TUI / CLI).
 

tests/agent/test_turn_context.py

@@ -58,6 +58,17 @@ class _FakeAgent:
         self.context_compressor = types.SimpleNamespace(
             protect_first_n=2, protect_last_n=2
         )
+        # Make the fake compressor honour the ContextEngine contract that the
+        # real code now relies on (should_compress_info returns a (bool, reason)
+        # tuple). Without it build_turn_context raises AttributeError.
+        def _fake_should_compress(tokens=None):
+            return False
+
+        def _fake_should_compress_info(tokens=None):
+            return (False, None)
+
+        self.context_compressor.should_compress = _fake_should_compress
+        self.context_compressor.should_compress_info = _fake_should_compress_info
         self._cached_system_prompt = "SYSTEM"
         self._memory_store = None
         self._memory_manager = None
@@ -68,6 +79,7 @@ class _FakeAgent:
         self._tool_guardrails = _FakeGuardrails()
         self._compression_warning = None
         self._emit_warning = MagicMock()
+        self._last_ctx_overflow_warn = None
         self._interrupt_requested = False
         self._memory_write_origin = "assistant_tool"
         self._stream_context_scrubber = None
@@ -83,6 +95,21 @@ class _FakeAgent:
         # is called (regression guard for #45499 turn-setup ordering).
         self._ensure_db_prompt_at_call = "<unset>"
 
+    def _warn_context_overflow_blocked(self, reason, preflight_tokens, threshold_tokens):
+        # Mirror the real AIAgent helper so tests can assert the warning fired.
+        _warn_kind = (reason or "unknown").split(":", 1)[0]
+        _warn_key = ("ctx_overflow_blocked", _warn_kind)
+        if self._last_ctx_overflow_warn != _warn_key:
+            self._last_ctx_overflow_warn = _warn_key
+            self._emit_warning(
+                f"⚠ Context is over the compression threshold "
+                f"(~{preflight_tokens:,} tokens >= {threshold_tokens:,}) "
+                f"but compression is currently blocked ({reason})."
+            )
+
+    def _clear_context_overflow_warn(self):
+        self._last_ctx_overflow_warn = None
+
     # --- methods the prologue calls ---
     def _ensure_db_session(self):
         self._ensure_db_prompt_at_call = self._cached_system_prompt