fix(compression): reset blocked-overflow dedup on every compression path + noise-filter survival pins

Follow-up fixes for the #62625 salvage:

- Dedup-reset gap (sweeper review): when the block clears while the
  context is STILL over threshold, execution enters the compression
  branch — the PR's 'else' reset never ran, so the warning stayed
  suppressed forever after the first block. _clear_context_overflow_warn()
  now fires on every automatic compression path: turn-context preflight,
  conversation_loop pre-API gate, and the post-tool loop-compaction gate.
- should_compress_info on current main: main refactored should_compress
  into _automatic_compression_blocked()/_locally(); the tuple variant now
  derives its reason from the same in-memory state via
  _compression_block_reason(), keeping cooldown:<s>/ineffective shapes.
- ContextEngine.should_compress_info ABC default now actually returns
  (should_compress(tokens), None) — the PR's default had a docstring but
  no return (returned None, would crash tuple-unpacking call sites).
- Below-threshold guard: the turn-context persisted-cooldown branch and
  the conversation_loop pre-API cooldown branch no longer warn when the
  estimate is under threshold (should_compress_info returns a None
  reason; the preflight pre-check is not a threshold guarantee). The
  pre-API guard also honors compression.max_attempts instead of a
  hardcoded 3, and no longer fabricates a cooldown reason.
- Noise-filter survival (#69550 composition): warning text is now a
  template constant (CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE) marked
  FAILURE-CLASS, pinned un-swallowed in VISIBLE_COMPRESSION_MESSAGES and
  in new tests that execute the real _TELEGRAM_NOISY_STATUS_RE +
  _prepare_gateway_status_message.
- Contributor mapping for stanislav@local -> sl4m3.

agent/context_engine.py

@@ -157,6 +157,7 @@ class ContextEngine(ABC):
         of silently skipping compression. Added for the silent-overflow
         warning fix (#62625) so plugin engines don't raise AttributeError.
         """
+        return self.should_compress(prompt_tokens), None
 
     @abstractmethod
     def compress(

agent/conversation_compression.py

@@ -109,6 +109,22 @@ COMPRESSION_RETRY_CONTEXT_REDUCED_STATUS_TEMPLATE = (
     "🗜️ Context reduced to {new_ctx:,} tokens (was {old_ctx:,}), retrying..."
 )
 
+# FAILURE-CLASS notice — a deliberate carve-out from routine-compression
+# silence (#16775 class): the context is over the compression threshold but
+# compression is blocked (summary-LLM cooldown / anti-thrash breaker), so the
+# session will keep growing until the hard provider token limit kills it.
+# This MUST stay visible on chat gateways. Do NOT add it to
+# ROUTINE_COMPRESSION_STATUS_SAMPLES or the gateway noise regex
+# (_TELEGRAM_NOISY_STATUS_RE); it is pinned un-swallowed in
+# tests/gateway/test_telegram_noise_filter.py::VISIBLE_COMPRESSION_MESSAGES.
+CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE = (
+    "⚠ Context is over the compression threshold "
+    "(~{tokens:,} tokens >= {threshold:,}) "
+    "but compression is currently blocked ({reason}). "
+    "The model may stop responding. Run /new to start a fresh "
+    "session or /compress to retry immediately."
+)
+
 # Sample-formatted instances of every routine compression status line, for
 # behavioral tests that iterate the ACTUAL emitted wording (formatted from the
 # same constants the emission sites use) through the gateway noise filter.

agent/conversation_loop.py

@@ -1308,6 +1308,11 @@ def run_conversation(
             if _moa_prepared_request is not None:
                 pending_moa_prepared_request = _moa_prepared_request
             compression_attempts += 1
+            # Compression is actually running (block cleared / was never
+            # blocked) — reset the blocked-overflow warning dedup so a future
+            # blocked-over-threshold turn can warn again. Mirrors the
+            # turn-context preflight reset (silent-overflow fix #62625).
+            agent._clear_context_overflow_warn()
             logger.info(
                 "Pre-API compression: ~%s request tokens >= %s threshold "
                 "(context=%s, attempt=%s/%s)",
@@ -1371,14 +1376,16 @@ def run_conversation(
         elif (
             agent.compression_enabled
             and len(messages) > 1
-            and compression_attempts < 3
+            and compression_attempts < max_compression_attempts
             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).
+            # Blocked by the summary-LLM cooldown. Surface a deduped warning
+            # (only when actually over threshold — should_compress_info
+            # returns a None reason below threshold) 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(
@@ -1386,15 +1393,12 @@ def run_conversation(
                 )[1]
             except Exception:
                 _block_reason = None
-            if not _block_reason:
-                _block_reason = (
-                    f"cooldown:{int(_compression_cooldown.get('remaining_seconds', 0.0))}"
+            if _block_reason:
+                agent._warn_context_overflow_blocked(
+                    _block_reason,
+                    request_pressure_tokens,
+                    int(getattr(_compressor, "threshold_tokens", 0) or 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
@@ -5463,6 +5467,11 @@ def run_conversation(
                     and _compressor.should_compress(_real_tokens)
                 ):
                     compression_attempts += 1
+                    # Compression is actually running (block cleared / was
+                    # never blocked) — reset the blocked-overflow warning
+                    # dedup so a future blocked-over-threshold turn can warn
+                    # again (silent-overflow fix #62625).
+                    agent._clear_context_overflow_warn()
                     agent._safe_print("  ⟳ compacting context…")
                     messages, active_system_prompt = agent._compress_context(
                         messages, system_message,

agent/turn_context.py

@@ -767,10 +767,11 @@ def build_turn_context(
                 int(_compression_cooldown.get("remaining_seconds", 0.0)),
                 agent.session_id or "none",
             )
-            # Context is over threshold but compression is blocked by the
-            # summary-LLM cooldown — surface a warning (see block below).
-            _cooldown_secs = _compression_cooldown.get("remaining_seconds", 0.0)
-            _compress_block_reason = f"cooldown:{_cooldown_secs:.0f}"
+            if _preflight_tokens >= _compressor.threshold_tokens:
+                # Context is over threshold but compression is blocked by the
+                # summary-LLM cooldown — surface a warning (see block below).
+                _cooldown_secs = _compression_cooldown.get("remaining_seconds", 0.0)
+                _compress_block_reason = f"cooldown:{_cooldown_secs:.0f}"
         elif _codex_native_auto:
             logger.info(
                 "Skipping Hermes preflight compression for codex app-server "

contributors/emails/stanislav@local

@@ -0,0 +1 @@
+sl4m3

run_agent.py

@@ -954,12 +954,15 @@ class AIAgent:
         _warn_key = ("ctx_overflow_blocked", _warn_kind)
         if getattr(self, "_last_ctx_overflow_warn", None) != _warn_key:
             self._last_ctx_overflow_warn = _warn_key
+            from agent.conversation_compression import (
+                CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE,
+            )
             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."
+                CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE.format(
+                    tokens=preflight_tokens,
+                    threshold=threshold_tokens,
+                    reason=reason,
+                )
             )
 
     def _clear_context_overflow_warn(self) -> None:

tests/agent/test_turn_context_overflow_warning.py

@@ -207,3 +207,141 @@ class TestTurnContextOverflowWarning:
         _run_build(agent)
         assert len(agent._warnings) == 2
         assert "blocked (ineffective)" in agent._warnings[1]
+
+    def test_dedup_resets_when_block_clears_while_over_threshold(self):
+        """The dedup reset must fire when the block clears while pressure is
+        still high (the sweeper-review gap): execution enters the compression
+        branch — not the ``else`` reset — so the reset must live on the
+        compression path itself. No manual state clearing here; only the
+        cooldown timer moves.
+        """
+        comp = _make_compressor()
+        comp.last_prompt_tokens = 73_000
+        comp._summary_failure_cooldown_until = time.monotonic() + 30
+        agent = _build_warn_agent(comp)
+        # Turn 1: over threshold + cooldown -> warn.
+        _run_build(agent)
+        assert len(agent._warnings) == 1
+        # Turn 2: cooldown expires while STILL over threshold -> compression
+        # branch runs; the reset must happen there (not in the else branch).
+        comp._summary_failure_cooldown_until = 0.0
+        _run_build(agent)
+        assert agent._compress_calls > 0
+        assert agent._last_ctx_overflow_warn is None
+        # Turn 3: cooldown re-arms -> the warning must re-fire.
+        comp._summary_failure_cooldown_until = time.monotonic() + 30
+        _run_build(agent)
+        assert len(agent._warnings) == 2
+
+    def test_no_warning_below_threshold_with_persisted_cooldown(self):
+        """A live cooldown with the context BELOW threshold must not warn —
+        there is no overflow to warn about (the cooldown branch is reached
+        via the cheap preflight pre-check, which is not a threshold
+        guarantee)."""
+        comp = _make_compressor()
+        comp.last_prompt_tokens = 10_000
+        comp._summary_failure_cooldown_until = time.monotonic() + 30
+        agent = _build_warn_agent(comp)
+        with patch("agent.auxiliary_client.set_runtime_main", lambda *a, **k: None), \
+             patch("agent.turn_context._should_run_preflight_estimate", return_value=True), \
+             patch("agent.turn_context.estimate_request_tokens_rough", return_value=10_000):
+            build_turn_context(
+                agent=agent,
+                user_message="hello",
+                system_message=None,
+                conversation_history=None,
+                task_id=None,
+                stream_callback=None,
+                persist_user_message=None,
+                restore_or_build_system_prompt=lambda *a, **k: None,
+                install_safe_stdio=lambda: None,
+                sanitize_surrogates=lambda s: s,
+                summarize_user_message_for_log=lambda s: s,
+                set_session_context=lambda _sid: None,
+                set_current_write_origin=lambda _o: None,
+                ra=lambda: type("R", (), {"_set_interrupt": lambda *a, **k: None})(),
+            )
+        assert agent._warnings == []
+
+
+# ---------------------------------------------------------------------------
+# Plugin context engines: backward-compatible should_compress_info default
+# ---------------------------------------------------------------------------
+
+class TestPluginEngineDefault:
+    def test_abc_default_returns_tuple(self):
+        """Engines that only implement should_compress() get the tuple shape
+        for free from the ContextEngine base class — the call sites in
+        turn_context.py / conversation_loop.py must not raise
+        AttributeError on plugin engines (sweeper review, #62625)."""
+        from tests.run_agent.test_plugin_context_engine_init import _StubEngine
+
+        engine = _StubEngine()
+        result = engine.should_compress_info(123_456)
+        assert result == (False, None)
+
+    def test_abc_default_delegates_to_should_compress(self):
+        from agent.context_engine import ContextEngine
+
+        class _TrueEngine(ContextEngine):
+            @property
+            def name(self):
+                return "true-stub"
+
+            def update_from_response(self, usage):
+                pass
+
+            def should_compress(self, prompt_tokens=None):
+                return True
+
+            def compress(self, messages, current_tokens=None, focus_topic=None):
+                return messages
+
+        assert _TrueEngine().should_compress_info(1) == (True, None)
+
+
+# ---------------------------------------------------------------------------
+# Gateway noise filter: the warning is FAILURE-CLASS and must survive
+# ---------------------------------------------------------------------------
+
+class TestWarningSurvivesNoiseFilter:
+    """The blocked-overflow warning is a deliberate carve-out from
+    routine-compression silence (#16775 class). The gateway noise regex
+    (#69550 just widened it) must NOT swallow it, or the fix is dead on
+    every chat platform. Executes the REAL compiled regex — never eyeball
+    a regex (noise-regex salvage rule).
+    """
+
+    def _emitted_warning(self, reason: str) -> str:
+        from agent.conversation_compression import (
+            CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE,
+        )
+
+        return CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE.format(
+            tokens=85_000, threshold=72_000, reason=reason
+        )
+
+    def test_cooldown_warning_not_matched_by_noise_regex(self):
+        from gateway.run import _TELEGRAM_NOISY_STATUS_RE
+
+        assert not _TELEGRAM_NOISY_STATUS_RE.search(
+            self._emitted_warning("cooldown:30")
+        )
+
+    def test_ineffective_warning_not_matched_by_noise_regex(self):
+        from gateway.run import _TELEGRAM_NOISY_STATUS_RE
+
+        assert not _TELEGRAM_NOISY_STATUS_RE.search(
+            self._emitted_warning("ineffective")
+        )
+
+    def test_warning_delivered_on_chat_platform(self):
+        """End-to-end through the fail-closed gateway status preparer."""
+        from gateway.config import Platform
+        from gateway.run import _prepare_gateway_status_message
+
+        message = self._emitted_warning("cooldown:30")
+        assert (
+            _prepare_gateway_status_message(Platform.TELEGRAM, "warn", message)
+            == message
+        )

tests/gateway/test_telegram_noise_filter.py

@@ -2,7 +2,10 @@
 
 import pytest
 
-from agent.conversation_compression import ROUTINE_COMPRESSION_STATUS_SAMPLES
+from agent.conversation_compression import (
+    CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE,
+    ROUTINE_COMPRESSION_STATUS_SAMPLES,
+)
 from gateway.config import Platform
 from gateway.run import (
     _prepare_gateway_status_message,
@@ -94,6 +97,18 @@ VISIBLE_COMPRESSION_MESSAGES = [
         "compression lock. Another compression may still be running, or "
         "the lock check failed — try again shortly."
     ),
+    # Blocked-overflow warning (#62625/#62708): the context is over the
+    # compression threshold but compression is blocked (summary-LLM cooldown
+    # or the anti-thrash breaker). FAILURE-CLASS — must reach chat users so
+    # they can /new or /compress before the session dies at the hard token
+    # limit. Formatted from the SAME template the emit site uses, so a
+    # rewording that drifts into the noise regex fails here.
+    CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE.format(
+        tokens=85_000, threshold=72_000, reason="cooldown:30"
+    ),
+    CONTEXT_OVERFLOW_BLOCKED_WARNING_TEMPLATE.format(
+        tokens=85_000, threshold=72_000, reason="ineffective"
+    ),
 ]