fix(memory): suppress the proactive tool-result prune under the checkpoint gate

The proactive prune is a lossy rewrite of the durable transcript — the
built-in commits it through archive_and_compact — and it runs BETWEEN
compactions, so no pre-compress checkpoint precedes it. With
compression.checkpoint_required armed, tool evidence could still be
dropped without ever reaching the durable provider: the same authority
class already closed for native compaction, codex app-server, and
post-turn micro-compaction.

- turn_preflight: the prune arm of compress_after_tool_results() moves
  into _proactive_tool_result_prune() and grows an `is True` gate on
  compression_checkpoint_required. Gating the call site rather than
  ContextCompressor's implementation covers plugin context engines that
  override prune_tool_results_only too — both reach the transcript
  through this one seam. The caller keeps the read-dedup reset on a
  committed prune; publish_prune moves with the commit.
- Suppression, not refusal (the prune fires per tool iteration; a raise
  would abort the turn). It is also lossless: the material the prune
  would have dropped stays in context, and the eventual threshold
  compaction — which IS checkpointed — handles it.
- conversation_compression_checkpoint: _record_checkpoint_gate_suppression
  bumps a suppression counter on the agent and logs once per withheld
  authority. The log names the availability trade the operator made:
  with the prune suppressed and compaction fail-closed, a checkpoint
  provider outage leaves no reclamation path at all and the session
  halts instead of compacting unarchived context. That is the gate's
  promise rather than a side effect, so it belongs in the line that
  fires when it becomes true.
- Count and log only a prune that would have run
  (ProactivePruneMixin.would_proactively_prune, _prune_would_have_run).
  A default install ships proactive_prune_tokens: 0, and a configured
  prune below its trigger returns the input untouched, so neither is a
  withheld reclamation path. An engine that overrides
  prune_tool_results_only owns its trigger policy, so occupying the hook
  is the precondition there; the inherited ABC no-op is not an authority.
- Tests (tests/agent/test_proactive_prune_checkpoint_gate.py): the call
  site swallows compressor exceptions with logger.debug, so the doubles
  count calls instead of asserting. Gate off is the sabotage control
  proving the harness reaches the prune; the engine-override case proves
  the seam covers both authorities; the log test pins one warning and its
  wording; five tests pin the default, the threshold (with a
  same-compressor control one token above it), the real compressor's
  predicate, the hookless and raising shapes, and the inherited-vs-
  overridden hook. test_proactive_prune_loop_wiring drives a real
  run_conversation with the gate armed: the prune is never consulted.

(cherry picked from commit 6904b46b55ac125f0b816f705e23f4c4a9c69991)
(cherry picked from commit 9e505a834742df11045da3b809a3a49c6178354c)

agent/context_compressor_prune.py

@@ -25,6 +25,15 @@ def _cc():
 class ProactivePruneMixin:
     """``ContextCompressor`` mixin: the deterministic, no-LLM tool-result prune and its commit."""
 
+    def would_proactively_prune(self, current_tokens: int | None = None) -> bool:
+        """Configured on AND above its trigger: the two state-free preconditions
+        ``prune_tool_results_only`` checks first, published so the host can tell a withheld prune
+        from a no-op. Not a prediction that a prune would commit (tail, runway and reclaim gate
+        need the scan)."""
+        return self.proactive_prune_tokens > 0 and (
+            current_tokens is None or current_tokens >= self.proactive_prune_tokens
+        )
+
     def prune_tool_results_only(
         self, messages: list[dict[str, Any]], current_tokens: int | None = None,
     ) -> tuple[list[dict[str, Any]], int]:
@@ -47,9 +56,7 @@ class ProactivePruneMixin:
         recent-tail protection applies to passes (2) and (3); pass (1) is tail-agnostic by design because
         dedup is lossless.
         """
-        if self.proactive_prune_tokens <= 0 or (
-            current_tokens is not None and current_tokens < self.proactive_prune_tokens
-        ):
+        if not self.would_proactively_prune(current_tokens):
             return messages, 0
         if len(messages) <= self.protect_last_n + self._protect_head_size(messages) + 1:
             self._warn_reclamation_no_op("prune:tail_only", current_tokens)

agent/conversation_compression_checkpoint.py

@@ -1,7 +1,8 @@
 """Fail-closed pre-compress checkpoint gate for ``agent.conversation_compression``.
 
 The refusal type, its remediation text and the startup warning for
-``compression.checkpoint_required``; ``conversation_compression`` re-exports every name.
+``compression.checkpoint_required`` (``conversation_compression`` re-exports the refusal
+helpers), plus the record for lossy authorities outside compression the armed gate withholds.
 """
 
 from __future__ import annotations
@@ -91,3 +92,48 @@ def _warn_checkpoint_required_without_capable_provider(agent: Any) -> None:
         PRE_COMPRESS_CHECKPOINT_API_VERSION,
         _CHECKPOINT_REQUIRED_REMEDIATION,
     )
+
+
+_checkpoint_gate_warned: set[str] = set()
+
+
+def _record_checkpoint_gate_suppression(agent: Any, authority: str) -> None:
+    """Count a lossy authority the armed gate withheld; warn once per process per authority.
+
+    The suppression itself is re-evaluated on every call, only the log is deduplicated (per
+    authority, so an operator sees which path was lost). The line names the availability
+    trade: with every other reclamation path withheld, a checkpoint outage halts the session.
+    """
+    agent._checkpoint_gate_suppression_count = getattr(agent, "_checkpoint_gate_suppression_count", 0) + 1
+    if authority in _checkpoint_gate_warned:
+        return
+    _checkpoint_gate_warned.add(authority)
+    logger.warning(
+        "compression.checkpoint_required is enabled: %s is withheld so the checkpoint-gated "
+        "compressor stays the only lossy authority. If the checkpoint provider is unavailable, "
+        "compaction fails closed and the session halts instead of compacting unarchived context.",
+        authority,
+    )
+
+
+def _prune_would_have_run(compressor: Any, current_tokens: int) -> bool:
+    """Only a prune that would have run is a withheld authority.
+
+    The built-in answers via ``would_proactively_prune()``. An engine that overrides
+    ``prune_tool_results_only`` owns its trigger policy, so occupying the hook is the
+    precondition; the inherited ABC no-op is not an authority. A raising predicate reads as
+    "would have run": the call was withheld either way.
+    """
+    from agent.context_engine import ContextEngine
+
+    prune = getattr(compressor, "prune_tool_results_only", None)
+    if not callable(prune) or getattr(prune, "__func__", None) is ContextEngine.prune_tool_results_only:
+        return False
+    predicate = getattr(compressor, "would_proactively_prune", None)
+    if not callable(predicate):
+        return True
+    try:
+        return bool(predicate(current_tokens))
+    except Exception:
+        logger.debug("would_proactively_prune() failed; counting the withheld prune", exc_info=True)
+        return True

agent/turn_preflight.py

@@ -371,27 +371,48 @@ def compress_after_tool_results(
             agent._warn_context_overflow_blocked(
                 _block_reason, _real_tokens, int(getattr(_compressor, "threshold_tokens", 0) or 0)
             )
-        # Proactive tool-result prune (deterministic, no LLM, keeps tail): no-op unless
-        # proactive_prune_tokens is exceeded; commits only past
-        # proactive_prune_min_reclaim_tokens so cache breaks stay episodic.
-        _prune = getattr(_compressor, "prune_tool_results_only", None)
-        if callable(_prune):
-            try:
-                _pruned_msgs, _pruned_n = _prune(messages, current_tokens=_real_tokens)
-            except Exception:
-                logger.debug("proactive tool-result prune failed; skipping", exc_info=True)
-                _pruned_msgs, _pruned_n = messages, 0
-            # Standard no-op caller contract: only commit when the engine returned a
-            # NEW list object with a non-zero count. Do NOT rebuild
-            # conversation_history: rows already carry _DB_PERSISTED_MARKER, and on a
-            # stale in-place flag the helper could seed unpersisted rows.
-            if _pruned_n and _pruned_msgs is not messages:
-                from agent.compaction_events import publish_prune
-
-                publish_prune(agent, messages, _pruned_msgs, _pruned_n)
-                messages = _pruned_msgs
-                # A committed prune is a content-loss boundary like compaction: demoted skill_view /
-                # read_file bodies survive only as one-line markers, so the repeat-read dedup must
-                # stop answering "unchanged" for them or the reload the marker asks for is refused.
-                _reset_read_dedup_caches(effective_task_id, session_id=agent.session_id or "")
+        _pruned_msgs = _proactive_tool_result_prune(agent, _compressor, messages, _real_tokens)
+        if _pruned_msgs is not messages:
+            messages = _pruned_msgs
+            # A committed prune is a content-loss boundary like compaction: demoted skill_view /
+            # read_file bodies survive only as one-line markers, so the repeat-read dedup must
+            # stop answering "unchanged" for them or the reload the marker asks for is refused.
+            _reset_read_dedup_caches(effective_task_id, session_id=agent.session_id or "")
     return _verdict(False)
+
+
+def _proactive_tool_result_prune(agent: Any, compressor: Any, messages: list, current_tokens: int) -> list:
+    """Proactive tool-result prune (deterministic, no LLM, keeps tail): the pruned list when one
+    committed, else the same ``messages`` object. No-op unless proactive_prune_tokens is exceeded;
+    commits only past proactive_prune_min_reclaim_tokens so cache breaks stay episodic.
+
+    Withheld while compression.checkpoint_required is armed: the prune rewrites the durable
+    transcript between compactions, so no checkpoint precedes it. Gating here, not in the
+    compressor, also covers engines that override the hook. Suppression, not refusal (it runs per
+    tool iteration); the material stays in context for the checkpointed threshold compaction.
+    """
+    if getattr(agent, "compression_checkpoint_required", False) is True:
+        from agent.conversation_compression_checkpoint import (
+            _prune_would_have_run, _record_checkpoint_gate_suppression,
+        )
+
+        if _prune_would_have_run(compressor, current_tokens):
+            _record_checkpoint_gate_suppression(agent, "the proactive tool-result prune")
+        return messages
+    prune = getattr(compressor, "prune_tool_results_only", None)
+    if not callable(prune):
+        return messages
+    try:
+        pruned_msgs, pruned_n = prune(messages, current_tokens=current_tokens)
+    except Exception:
+        logger.debug("proactive tool-result prune failed; skipping", exc_info=True)
+        return messages
+    # Standard no-op caller contract: only commit when the engine returned a NEW list object
+    # with a non-zero count. Do NOT rebuild conversation_history: rows already carry
+    # _DB_PERSISTED_MARKER, and on a stale in-place flag the helper could seed unpersisted rows.
+    if not pruned_n or pruned_msgs is messages:
+        return messages
+    from agent.compaction_events import publish_prune
+
+    publish_prune(agent, messages, pruned_msgs, pruned_n)
+    return pruned_msgs

tests/agent/test_proactive_prune_checkpoint_gate.py

@@ -0,0 +1,210 @@
+"""The proactive tool-result prune under ``compression.checkpoint_required``.
+
+The prune rewrites the durable transcript between compactions, so no pre-compress
+checkpoint precedes it: the armed gate withholds it (suppression, not refusal) and
+reports only a prune that would actually have run.
+"""
+
+import logging
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from agent import conversation_compression_checkpoint as gate
+from agent.conversation_compression_checkpoint import _prune_would_have_run
+from agent.turn_preflight import _proactive_tool_result_prune
+
+
+class _CountingPruneCompressor:
+    """Built-in-shaped double: counts instead of asserting, because the call site swallows
+    exceptions. Publishes ``would_proactively_prune`` with a configured trigger by default."""
+
+    def __init__(self, proactive_prune_tokens: int = 48_000):
+        self.calls = 0
+        self.proactive_prune_tokens = proactive_prune_tokens
+
+    def would_proactively_prune(self, current_tokens=None):
+        return self.proactive_prune_tokens > 0 and (
+            current_tokens is None or current_tokens >= self.proactive_prune_tokens
+        )
+
+    def prune_tool_results_only(self, messages, current_tokens=None):
+        self.calls += 1
+        return list(messages) + [{"role": "system", "content": "pruned"}], 3
+
+
+def _engine_stub():
+    """Built against the currently imported ABC (sibling suites purge ``agent.*``)."""
+    from agent.context_engine import ContextEngine
+
+    class _CompressOnlyEngine(ContextEngine):
+        @property
+        def name(self) -> str:
+            return "compress-only"
+
+        def update_from_response(self, usage):
+            return None
+
+        def should_compress(self, prompt_tokens=None):
+            return False
+
+        def compress(self, messages, current_tokens=None, focus_topic=None, force=False, memory_context=""):
+            return messages
+
+    return _CompressOnlyEngine
+
+
+def _agent(checkpoint_required: bool):
+    return SimpleNamespace(compression_checkpoint_required=checkpoint_required)
+
+
+def _prune_warnings(caplog):
+    return [r for r in caplog.records if r.levelno >= logging.WARNING and "tool-result prune" in r.getMessage()]
+
+
+@pytest.fixture(autouse=True)
+def _fresh_once_log(monkeypatch):
+    monkeypatch.setattr(gate, "_checkpoint_gate_warned", set())
+
+
+def test_proactive_prune_is_suppressed_when_checkpoint_required():
+    """The gate reaches the prune and the transcript survives intact."""
+    compressor = _CountingPruneCompressor()
+    messages = [{"role": "user", "content": "scan"}]
+    agent = _agent(True)
+
+    assert _proactive_tool_result_prune(agent, compressor, messages, 400_000) is messages
+    assert compressor.calls == 0
+    assert agent._checkpoint_gate_suppression_count == 1
+
+
+def test_proactive_prune_still_runs_when_gate_is_off():
+    """Sabotage control: gate off, the harness really reaches the prune."""
+    compressor = _CountingPruneCompressor()
+    messages = [{"role": "user", "content": "scan"}]
+
+    with patch("agent.compaction_events.publish_prune"):
+        result = _proactive_tool_result_prune(_agent(False), compressor, messages, 400_000)
+
+    assert compressor.calls == 1
+    assert result is not messages
+    assert result[-1]["content"] == "pruned"
+
+
+def test_engine_override_of_prune_is_suppressed_too():
+    """One call site covers the built-in and every engine overriding the hook."""
+
+    class _PruningEngine(_engine_stub()):
+        calls = 0
+
+        def prune_tool_results_only(self, messages, current_tokens=None):
+            type(self).calls += 1
+            return list(messages), 3
+
+    engine = _PruningEngine()
+    messages = [{"role": "user", "content": "scan"}]
+    agent = _agent(True)
+
+    assert _proactive_tool_result_prune(agent, engine, messages, 400_000) is messages
+    assert _PruningEngine.calls == 0
+    # An overriding engine owns its trigger policy: the host would have dispatched it.
+    assert agent._checkpoint_gate_suppression_count == 1
+
+    with patch("agent.compaction_events.publish_prune"):
+        _proactive_tool_result_prune(_agent(False), engine, messages, 400_000)
+    assert _PruningEngine.calls == 1
+
+
+def test_prune_suppression_logs_once_and_names_the_availability_consequence(caplog):
+    compressor = _CountingPruneCompressor()
+    agent = _agent(True)
+
+    with caplog.at_level(logging.WARNING, logger="agent.conversation_compression"):
+        for _ in range(2):
+            _proactive_tool_result_prune(agent, compressor, [], 400_000)
+
+    warnings = _prune_warnings(caplog)
+    assert len(warnings) == 1
+    assert "checkpoint_required" in warnings[0].getMessage()
+    assert "halt" in warnings[0].getMessage()
+    # Suppression is re-evaluated every call; only the log is deduplicated.
+    assert agent._checkpoint_gate_suppression_count == 2
+
+
+def test_disabled_prune_is_not_reported_as_a_suppressed_authority(caplog):
+    """The shipping default (``proactive_prune_tokens: 0``) withholds nothing."""
+    compressor = _CountingPruneCompressor(proactive_prune_tokens=0)
+    messages = [{"role": "user", "content": "scan"}]
+    agent = _agent(True)
+
+    with caplog.at_level(logging.WARNING, logger="agent.conversation_compression"):
+        assert _proactive_tool_result_prune(agent, compressor, messages, 400_000) is messages
+
+    assert compressor.calls == 0
+    assert getattr(agent, "_checkpoint_gate_suppression_count", 0) == 0
+    assert _prune_warnings(caplog) == []
+
+
+def test_prune_below_its_trigger_is_not_reported_as_a_suppressed_authority(caplog):
+    compressor = _CountingPruneCompressor(proactive_prune_tokens=400_000)
+    agent = _agent(True)
+
+    with caplog.at_level(logging.WARNING, logger="agent.conversation_compression"):
+        _proactive_tool_result_prune(agent, compressor, [], 399_999)
+    assert getattr(agent, "_checkpoint_gate_suppression_count", 0) == 0
+    assert _prune_warnings(caplog) == []
+
+    # One token more and the same compressor IS a withheld authority.
+    with caplog.at_level(logging.WARNING, logger="agent.conversation_compression"):
+        _proactive_tool_result_prune(agent, compressor, [], 400_000)
+    assert agent._checkpoint_gate_suppression_count == 1
+    assert len(_prune_warnings(caplog)) == 1
+
+
+def test_would_proactively_prune_is_the_compressors_own_precondition():
+    """Answered by the real ``ContextCompressor``, so the report cannot drift from its trigger."""
+    from agent.context_compressor import ContextCompressor
+
+    with patch("agent.context_compressor.get_model_context_length", return_value=1_000_000):
+        default = ContextCompressor(model="test", quiet_mode=True)
+        configured = ContextCompressor(model="test", quiet_mode=True, proactive_prune_tokens=48_000)
+
+    assert default.proactive_prune_tokens == 0
+    assert default.would_proactively_prune(400_000) is False
+    assert _prune_would_have_run(default, 400_000) is False
+    assert configured.would_proactively_prune(47_999) is False
+    assert configured.would_proactively_prune(48_000) is True
+    assert _prune_would_have_run(configured, 47_999) is False
+    assert _prune_would_have_run(configured, 48_000) is True
+    # An unknown token count cannot rule the prune out; the prune proceeds on None.
+    assert configured.would_proactively_prune(None) is True
+
+
+def test_prune_hookless_and_broken_predicate_shapes_are_handled():
+    """No hook = no authority; a raising predicate reads as "would have run"."""
+    assert _prune_would_have_run(SimpleNamespace(), 400_000) is False
+
+    class _BrokenPredicate:
+        def would_proactively_prune(self, current_tokens=None):
+            raise RuntimeError("engine blew up")
+
+        def prune_tool_results_only(self, messages, current_tokens=None):
+            return messages, 0
+
+    assert _prune_would_have_run(_BrokenPredicate(), 400_000) is True
+
+
+def test_engine_that_never_overrode_the_prune_hook_is_not_a_suppressed_authority():
+    """``__func__`` identity tells occupancy of the hook from inheriting the ABC no-op."""
+    compress_only = _engine_stub()
+
+    class _InheritsTheDefault(compress_only):
+        pass
+
+    class _OccupiesTheHook(compress_only):
+        def prune_tool_results_only(self, messages, current_tokens=None):
+            return list(messages), 1
+
+    assert _prune_would_have_run(_InheritsTheDefault(), 400_000) is False
+    assert _prune_would_have_run(_OccupiesTheHook(), 400_000) is True

tests/agent/test_proactive_prune_loop_wiring.py

@@ -195,6 +195,22 @@ class TestProactivePruneLoopWiring:
         assert len(calls) == 3  # one shot per tool iteration
         assert all(t == 120_000 for t in calls)  # fed the real usage reading
 
+    def test_armed_checkpoint_gate_withholds_the_prune(self, agent):
+        """compression.checkpoint_required: no checkpoint precedes the prune's durable
+        rewrite, so the loop never consults it (the control is the test above)."""
+        calls = []
+        agent.compression_checkpoint_required = True
+        # The suppression counter is shared with the on_turn_complete gate; a declared observer
+        # keeps this test about the prune.
+        agent.context_compressor.compacts_outside_compress = False
+        agent.context_compressor.prune_tool_results_only = (
+            lambda messages, current_tokens=None: (calls.append(current_tokens), (messages[:-1], 1))[1]
+        )
+        result = _run_tool_loop(agent, n_tool_iterations=3)
+        assert result["completed"] is True
+        assert calls == []
+        assert agent._checkpoint_gate_suppression_count == 3
+
     def test_committed_prune_replaces_messages(self, agent):
         marker = "[old tool output pruned]"