feat(compression): prompt-cache reclaim gate + hardened wiring for proactive prune

Follow-ups on top of the cherry-picked #62644 mechanism, porting it to
current main and closing the salvage-review requirements:

- proactive_prune_min_reclaim_tokens (default 4096): a prune only COMMITS
  when it reclaims a meaningful token batch, measured on the pruned output.
  A committed prune rewrites already-sent history and invalidates the
  provider prompt-cache prefix; this hysteresis gate keeps those breaks
  episodic/amortized (like a compression boundary) instead of firing every
  tool iteration. 0 disables the gate. (Design point credited to the
  #62389 review cycle's prune_minimum_tokens.)
- Standard no-op caller contract: every skip path returns the INPUT list
  object; the loop commits only on 'result is not messages' + non-zero count.
- Loop call is getattr+callable guarded (plugin engines predating the hook,
  SimpleNamespace test doubles) and exception-swallowed at debug level.
- Config parse follows the compression.max_attempts hardened semantics:
  booleans rejected, fractional floats rejected, integral floats/numeric
  strings accepted; negative trigger = disabled.
- cli-config.yaml.example documented (all three keys) and gateway
  _CACHE_BUSTING_CONFIG_KEYS extended so hot-reload rebuilds the agent.
- Tests: min-reclaim gate both directions, input-object no-op contract,
  no-orphan tool_call_id pairing in BOTH directions (#69830 pin rule),
  default-off zero-behavior-change pin, config parse seam, and behavioral
  loop-wiring tests (consulted/commit/no-op/absent-method/raising).

cli-config.yaml.example

@@ -484,6 +484,34 @@ compression:
   # summarization on a short idle thread. Example: 1800 = compact after 30 min idle.
   idle_compact_after_seconds: 0
 
+  # Proactive tool-result prune (default: 0 = disabled). Opt-in token trigger
+  # for a deterministic, no-LLM prune of OLD tool-result payloads, run
+  # independently of `threshold` above. On large-window models (512K/1M) the
+  # ratio threshold rarely fires, so bulky tool outputs (terminal dumps, file
+  # reads, web extracts) ride along in history and get re-billed every turn.
+  # When re-sent history exceeds this many tokens, the prune dedupes identical
+  # results, summarizes older oversized ones, and truncates large tool-call
+  # arguments — protecting the most recent `protect_last_n` messages and never
+  # calling the model. Try 48000 to enable. Built-in compressor engine only;
+  # other context engines inherit a safe no-op.
+  # NOTE: a committed prune rewrites already-sent history, which invalidates
+  # the provider's prompt-cache prefix — the min_reclaim gate below keeps
+  # those cache breaks episodic (like a compression boundary) instead of
+  # per-turn.
+  proactive_prune_tokens: 0
+
+  # The prune's summarize pass only touches tool results larger than this many
+  # characters (clamped to >= 200 so a generated summary can't be
+  # re-summarized). Default 8000.
+  proactive_prune_min_result_chars: 8000
+
+  # A proactive prune only COMMITS when it reclaims at least this many tokens
+  # (measured on the pruned output). This is the prompt-cache hysteresis gate:
+  # one meaningful, amortized cache break per batch of stale tool output
+  # instead of a tiny break on every tool iteration. 0 = commit any non-zero
+  # prune. Default 4096.
+  proactive_prune_min_reclaim_tokens: 4096
+
   # To pin a specific model/provider for compression summaries, use the
   # auxiliary section below (auxiliary.compression.provider / model).
 

gateway/run.py

@@ -18122,6 +18122,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         ("compression", "codex_app_server_auto"),
         ("compression", "target_ratio"),
         ("compression", "protect_last_n"),
+        ("compression", "proactive_prune_tokens"),
+        ("compression", "proactive_prune_min_result_chars"),
+        ("compression", "proactive_prune_min_reclaim_tokens"),
         ("agent", "disabled_toolsets"),
         ("memory", "provider"),
         ("checkpoints", "enabled"),

tests/agent/test_proactive_prune_config.py

@@ -0,0 +1,111 @@
+"""compression.proactive_prune_* — config parse seam for the proactive prune.
+
+Mirrors ``test_compression_max_attempts_config.py``: the three knobs are
+parsed in ``agent_init`` with the same hardened semantics (booleans rejected,
+fractional floats rejected — not truncated, integral floats and numeric
+strings accepted) and attached to the built-in compressor.  Default is
+0 / 8000 / 4096, i.e. the feature is OFF and behavior-neutral unless
+``proactive_prune_tokens`` is set above 0.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import io
+from pathlib import Path
+
+from hermes_state import SessionDB
+from run_agent import AIAgent
+
+
+def _config(**prune_keys) -> dict:
+    compression = {
+        "enabled": True,
+        "threshold": 0.50,
+        "target_ratio": 0.20,
+        "protect_first_n": 3,
+        "protect_last_n": 20,
+    }
+    compression.update(prune_keys)
+    return {
+        "compression": compression,
+        "prompt_caching": {"cache_ttl": "5m"},
+        "sessions": {},
+        "bedrock": {},
+    }
+
+
+def _make_agent(monkeypatch, tmp_path: Path, **prune_keys):
+    from hermes_cli import config as config_mod
+
+    monkeypatch.setattr(config_mod, "load_config", lambda: _config(**prune_keys))
+    db = SessionDB(db_path=tmp_path / "state.db")
+    with contextlib.redirect_stdout(io.StringIO()):
+        agent = AIAgent(
+            base_url="https://chatgpt.com/backend-api/codex",
+            api_key="test-key",
+            provider="openai-codex",
+            model="gpt-5.5",
+            enabled_toolsets=[],
+            disabled_toolsets=[],
+            quiet_mode=True,
+            skip_memory=True,
+            session_db=db,
+            session_id="proactive-prune-config-test",
+        )
+    return agent
+
+
+class TestProactivePruneConfig:
+    def test_default_is_disabled_when_unset(self, monkeypatch, tmp_path):
+        agent = _make_agent(monkeypatch, tmp_path)
+        cc = agent.context_compressor
+        assert cc.proactive_prune_tokens == 0
+        assert cc.proactive_prune_min_result_chars == 8000
+        assert cc.proactive_prune_min_reclaim_tokens == 4096
+
+    def test_custom_values_are_honored(self, monkeypatch, tmp_path):
+        agent = _make_agent(
+            monkeypatch,
+            tmp_path,
+            proactive_prune_tokens=48_000,
+            proactive_prune_min_result_chars=12_000,
+            proactive_prune_min_reclaim_tokens=8_192,
+        )
+        cc = agent.context_compressor
+        assert cc.proactive_prune_tokens == 48_000
+        assert cc.proactive_prune_min_result_chars == 12_000
+        assert cc.proactive_prune_min_reclaim_tokens == 8_192
+
+    def test_boolean_is_rejected_not_coerced(self, monkeypatch, tmp_path):
+        # bool subclasses int: YAML `proactive_prune_tokens: true` must fall
+        # back to disabled, never coerce to 1 token.
+        agent = _make_agent(monkeypatch, tmp_path, proactive_prune_tokens=True)
+        assert agent.context_compressor.proactive_prune_tokens == 0
+
+    def test_fractional_float_is_rejected_not_truncated(self, monkeypatch, tmp_path):
+        agent = _make_agent(monkeypatch, tmp_path, proactive_prune_tokens=48_000.7)
+        assert agent.context_compressor.proactive_prune_tokens == 0
+
+    def test_integral_float_and_numeric_string_accepted(self, monkeypatch, tmp_path):
+        agent = _make_agent(monkeypatch, tmp_path, proactive_prune_tokens=48_000.0)
+        assert agent.context_compressor.proactive_prune_tokens == 48_000
+        agent = _make_agent(monkeypatch, tmp_path, proactive_prune_tokens="32000")
+        assert agent.context_compressor.proactive_prune_tokens == 32_000
+
+    def test_negative_trigger_treated_as_disabled(self, monkeypatch, tmp_path):
+        agent = _make_agent(monkeypatch, tmp_path, proactive_prune_tokens=-100)
+        assert agent.context_compressor.proactive_prune_tokens == 0
+
+    def test_garbage_falls_back_to_defaults(self, monkeypatch, tmp_path):
+        agent = _make_agent(
+            monkeypatch,
+            tmp_path,
+            proactive_prune_tokens="lots",
+            proactive_prune_min_result_chars=None,
+            proactive_prune_min_reclaim_tokens="???",
+        )
+        cc = agent.context_compressor
+        assert cc.proactive_prune_tokens == 0
+        assert cc.proactive_prune_min_result_chars == 8000
+        assert cc.proactive_prune_min_reclaim_tokens == 4096

tests/agent/test_proactive_tool_result_pruning.py

@@ -100,7 +100,11 @@ def test_below_trigger_is_noop():
 
 
 def test_recent_tail_is_protected():
-    c = _compressor(proactive_prune_tokens=48_000, proactive_prune_min_result_chars=8_000)
+    c = _compressor(
+        proactive_prune_tokens=48_000,
+        proactive_prune_min_result_chars=8_000,
+        proactive_prune_min_reclaim_tokens=0,  # gate off: this test pins tail semantics
+    )
     # pair 0 tool is old (index 2); pair 7 tool is in the last-4 protected tail (index 16)
     msgs = _build(8, big_indices={0, 7})
     result, pruned = c.prune_tool_results_only(msgs, current_tokens=120_000)
@@ -109,7 +113,11 @@ def test_recent_tail_is_protected():
 
 
 def test_size_floor_spares_small_results():
-    c = _compressor(proactive_prune_tokens=48_000, proactive_prune_min_result_chars=8_000)
+    c = _compressor(
+        proactive_prune_tokens=48_000,
+        proactive_prune_min_result_chars=8_000,
+        proactive_prune_min_reclaim_tokens=0,  # gate off: this test pins the size floor
+    )
     msgs = _build(8, big_indices={1}, big_chars=9000)
     for m in msgs:                      # make pair 0's tool 5000 chars (< 8000 floor), still old
         if m.get("tool_call_id") == "call_0":
@@ -162,3 +170,108 @@ def test_min_result_chars_floor_is_clamped():
     assert _compressor(proactive_prune_min_result_chars=50).proactive_prune_min_result_chars == 200
     assert _compressor(proactive_prune_min_result_chars=-1).proactive_prune_min_result_chars == 200
     assert _compressor(proactive_prune_min_result_chars=8000).proactive_prune_min_result_chars == 8000
+
+
+# ---------------------------------------------------------------------------
+# Salvage follow-ups: no-op caller contract, prompt-cache hysteresis gate,
+# no-orphan pairing invariant, and the default-off behavior pin.
+# ---------------------------------------------------------------------------
+
+
+def test_noop_paths_return_input_object():
+    """Standard caller contract: every no-op path hands back the INPUT list
+    object so callers can gate bookkeeping on ``result is not input``."""
+    msgs = _build(8, big_indices={0, 1, 2})
+    # Disabled (default)
+    c = _compressor()
+    result, pruned = c.prune_tool_results_only(msgs, current_tokens=500_000)
+    assert pruned == 0 and result is msgs
+    # Below trigger
+    c = _compressor(proactive_prune_tokens=48_000)
+    result, pruned = c.prune_tool_results_only(msgs, current_tokens=10_000)
+    assert pruned == 0 and result is msgs
+    # Above trigger but nothing prunable (all results tiny)
+    c = _compressor(proactive_prune_tokens=48_000)
+    tiny = _build(8, big_indices=set())
+    result, pruned = c.prune_tool_results_only(tiny, current_tokens=120_000)
+    assert pruned == 0 and result is tiny
+
+
+def test_min_reclaim_gate_blocks_small_prunes():
+    """Prompt-cache hysteresis: a prune that would reclaim less than
+    ``proactive_prune_min_reclaim_tokens`` must NOT commit (returns the input
+    object) — rewriting already-sent history for a trivial saving would break
+    the provider's cached prefix every tool iteration."""
+    c = _compressor(
+        proactive_prune_tokens=48_000,
+        proactive_prune_min_result_chars=8_000,
+        proactive_prune_min_reclaim_tokens=1_000_000,  # unreachably high
+    )
+    msgs = _build(8, big_indices={0, 1, 2})
+    result, pruned = c.prune_tool_results_only(msgs, current_tokens=120_000)
+    assert pruned == 0
+    assert result is msgs  # input object — caller commits nothing
+
+
+def test_min_reclaim_gate_allows_large_prunes():
+    """A prune reclaiming more than the gate commits normally."""
+    c = _compressor(
+        proactive_prune_tokens=48_000,
+        proactive_prune_min_result_chars=8_000,
+        proactive_prune_min_reclaim_tokens=1_000,  # 3×9000 chars ≈ 6.7K tokens reclaimed
+    )
+    msgs = _build(8, big_indices={0, 1, 2})
+    result, pruned = c.prune_tool_results_only(msgs, current_tokens=120_000)
+    assert pruned >= 3
+    assert result is not msgs
+
+
+def test_min_reclaim_gate_default_and_clamp():
+    """Default 4096; negative/None coerce to disabled (0)."""
+    assert _compressor().proactive_prune_min_reclaim_tokens == 4096
+    assert _compressor(proactive_prune_min_reclaim_tokens=0).proactive_prune_min_reclaim_tokens == 0
+    assert _compressor(proactive_prune_min_reclaim_tokens=-5).proactive_prune_min_reclaim_tokens == 0
+    assert _compressor(proactive_prune_min_reclaim_tokens=None).proactive_prune_min_reclaim_tokens == 0
+
+
+def test_no_orphans_both_directions():
+    """tool_call_id pairing survives the prune in BOTH directions: every
+    surviving tool result has its assistant call, and every assistant tool_call
+    has its result row (the #69830 test-pin rule — never assert exact surviving
+    pair counts, only the pairing invariant)."""
+    c = _compressor(
+        proactive_prune_tokens=48_000,
+        proactive_prune_min_result_chars=8_000,
+        proactive_prune_min_reclaim_tokens=0,
+    )
+    msgs = _build(10, big_indices={0, 1, 2, 3, 4})
+    result, pruned = c.prune_tool_results_only(msgs, current_tokens=120_000)
+    assert pruned >= 1
+    call_ids = set()
+    for m in result:
+        if m.get("role") == "assistant":
+            for tc in m.get("tool_calls") or []:
+                call_ids.add(tc["id"] if isinstance(tc, dict) else tc.id)
+    result_ids = {m["tool_call_id"] for m in result if m.get("role") == "tool"}
+    assert result_ids <= call_ids, "orphan tool results without a matching call"
+    assert call_ids <= result_ids, "orphan tool calls without a matching result"
+
+
+def test_unset_config_zero_behavior_change():
+    """Pin: with the config knobs unset, the compressor behaves byte-identically
+    to pre-feature main — the prune path is dead code and the full-compression
+    Phase-1 caller keeps its 200-char floor."""
+    c = _compressor()  # nothing configured
+    assert c.proactive_prune_tokens == 0
+    msgs = _build(8, big_indices={0, 1, 2})
+    import copy
+    snapshot = copy.deepcopy(msgs)
+    result, pruned = c.prune_tool_results_only(msgs, current_tokens=10_000_000)
+    assert pruned == 0
+    assert result is msgs
+    assert msgs == snapshot  # input never mutated
+    # And the compression-path caller still prunes at the 200-char default floor
+    # (min_prune_chars default unchanged).
+    import inspect
+    sig = inspect.signature(c._prune_old_tool_results)
+    assert sig.parameters["min_prune_chars"].default == 200

tests/run_agent/test_proactive_prune_loop_wiring.py

@@ -0,0 +1,214 @@
+"""Behavioral tests for the post-tool proactive tool-result prune wiring.
+
+The conversation loop's post-tool gate now has a prune arm inside the
+``elif agent.compression_enabled`` branch: when full compression does NOT
+fire (the usual case on a large-window model), the deterministic no-LLM
+prune gets one shot per tool iteration, committing only when the engine
+returns a NEW list object with a non-zero prune count.
+
+These tests drive ``run_conversation()`` through real tool iterations and pin:
+- the prune is consulted when compression stands down;
+- a committed prune replaces ``messages`` for subsequent iterations;
+- a no-op (input object returned) commits nothing;
+- a compressor WITHOUT the method (plugin engine predating the hook /
+  SimpleNamespace test double) does not raise — getattr-guarded;
+- a raising prune is swallowed (debug log), never fails the turn.
+"""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from run_agent import AIAgent
+
+
+def _tool_call(i: int):
+    return SimpleNamespace(
+        id=f"call_{i}",
+        type="function",
+        function=SimpleNamespace(name="web_search", arguments='{"query": "x"}'),
+    )
+
+
+def _tool_response(i: int):
+    msg = SimpleNamespace(
+        content=None,
+        reasoning_content=None,
+        reasoning=None,
+        tool_calls=[_tool_call(i)],
+    )
+    choice = SimpleNamespace(message=msg, finish_reason="tool_calls")
+    return SimpleNamespace(choices=[choice], model="test/model", usage=None)
+
+
+def _stop_response():
+    msg = SimpleNamespace(
+        content="done",
+        reasoning_content=None,
+        reasoning=None,
+        tool_calls=None,
+    )
+    choice = SimpleNamespace(message=msg, finish_reason="stop")
+    return SimpleNamespace(choices=[choice], model="test/model", usage=None)
+
+
+def _make_tool_defs(*names: str) -> list:
+    return [
+        {
+            "type": "function",
+            "function": {
+                "name": n,
+                "description": f"{n} tool",
+                "parameters": {"type": "object", "properties": {}},
+            },
+        }
+        for n in names
+    ]
+
+
+def _quiet_compressor() -> MagicMock:
+    """A compressor that never demands full compression.
+
+    ``should_compress`` False routes the post-tool gate into the ``elif``
+    branch where the proactive prune arm lives.  ``should_compress_info``
+    reports unblocked (no block reason) so the overflow warning stays quiet.
+    """
+    compressor = MagicMock()
+    compressor.protect_first_n = 3
+    compressor.protect_last_n = 20
+    compressor.threshold_tokens = 500_000
+    compressor.context_length = 1_000_000
+    compressor.last_prompt_tokens = 120_000
+    compressor.should_compress.return_value = False
+    compressor.should_compress_info.return_value = (False, None)
+    compressor.should_defer_preflight_to_real_usage.return_value = True
+    compressor.get_active_compression_failure_cooldown.return_value = None
+    return compressor
+
+
+@pytest.fixture()
+def agent():
+    with (
+        patch("run_agent.get_tool_definitions", return_value=_make_tool_defs("web_search")),
+        patch("run_agent.check_toolset_requirements", return_value={}),
+        patch("run_agent.OpenAI"),
+    ):
+        a = AIAgent(
+            api_key="test-key-1234567890",
+            base_url="https://openrouter.ai/api/v1",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+            max_iterations=10,
+        )
+    a.client = MagicMock()
+    a._cached_system_prompt = "You are helpful."
+    a._use_prompt_caching = False
+    a._disable_streaming = True
+    a.tool_delay = 0
+    a.save_trajectories = False
+    a.compression_enabled = True
+    a.context_compressor = _quiet_compressor()
+    return a
+
+
+def _run_tool_loop(agent, n_tool_iterations: int):
+    responses = [_tool_response(i) for i in range(n_tool_iterations)]
+    responses.append(_stop_response())
+    agent.client.chat.completions.create.side_effect = responses
+
+    with (
+        patch.object(agent, "_persist_session"),
+        patch.object(agent, "_save_trajectory"),
+        patch.object(agent, "_cleanup_task_resources"),
+        patch(
+            "run_agent.handle_function_call",
+            lambda name, args, task_id=None, **kwargs: json.dumps({"ok": True}),
+        ),
+    ):
+        result = agent.run_conversation("do a lot of tool work")
+
+    return result
+
+
+class TestProactivePruneLoopWiring:
+    def test_prune_consulted_when_compression_stands_down(self, agent):
+        calls = []
+
+        def _prune(messages, current_tokens=None):
+            calls.append(current_tokens)
+            return messages, 0  # no-op contract: input object back
+
+        agent.context_compressor.prune_tool_results_only = _prune
+        result = _run_tool_loop(agent, n_tool_iterations=3)
+        assert result["completed"] is True
+        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_committed_prune_replaces_messages(self, agent):
+        marker = "[old tool output pruned]"
+
+        def _prune(messages, current_tokens=None):
+            pruned = [dict(m) for m in messages]
+            changed = 0
+            for m in pruned:
+                if m.get("role") == "tool" and m.get("content") != marker:
+                    m["content"] = marker
+                    changed += 1
+            if not changed:
+                return messages, 0
+            return pruned, changed
+
+        agent.context_compressor.prune_tool_results_only = _prune
+        result = _run_tool_loop(agent, n_tool_iterations=2)
+        assert result["completed"] is True
+        tool_rows = [m for m in result["messages"] if m.get("role") == "tool"]
+        assert tool_rows, "expected tool rows in the final transcript"
+        assert all(m["content"] == marker for m in tool_rows)
+
+    def test_noop_input_object_commits_nothing(self, agent):
+        """Engine returns the INPUT object with a (bogus) non-zero count —
+        the caller's ``result is not input`` gate must refuse the commit."""
+        def _prune(messages, current_tokens=None):
+            return messages, 5  # lies about count but returns input object
+
+        agent.context_compressor.prune_tool_results_only = _prune
+        result = _run_tool_loop(agent, n_tool_iterations=2)
+        assert result["completed"] is True
+        tool_rows = [m for m in result["messages"] if m.get("role") == "tool"]
+        # tool output may be wrapped in an untrusted_tool_result envelope —
+        # assert the original payload survived un-pruned.
+        assert all('"ok": true' in m["content"] for m in tool_rows)
+
+    def test_engine_without_method_does_not_raise(self, agent):
+        """Plugin engines predating the hook / minimal doubles lack the
+        method entirely — the getattr guard treats absence as a no-op."""
+        compressor = SimpleNamespace(
+            protect_first_n=3,
+            protect_last_n=20,
+            threshold_tokens=500_000,
+            context_length=1_000_000,
+            last_prompt_tokens=120_000,
+            should_compress=lambda _t: False,
+            should_defer_preflight_to_real_usage=lambda _t: True,
+            get_active_compression_failure_cooldown=lambda: None,
+        )
+        agent.context_compressor = compressor
+        result = _run_tool_loop(agent, n_tool_iterations=2)
+        assert result["completed"] is True
+
+    def test_raising_prune_is_swallowed(self, agent):
+        def _prune(messages, current_tokens=None):
+            raise RuntimeError("boom")
+
+        agent.context_compressor.prune_tool_results_only = _prune
+        result = _run_tool_loop(agent, n_tool_iterations=2)
+        assert result["completed"] is True
+        tool_rows = [m for m in result["messages"] if m.get("role") == "tool"]
+        # tool output may be wrapped in an untrusted_tool_result envelope —
+        # assert the original payload survived un-pruned.
+        assert all('"ok": true' in m["content"] for m in tool_rows)