diff --git a/core/agent_runtime/microcompact.py b/core/agent_runtime/microcompact.py new file mode 100644 index 00000000..0d1aa13c --- /dev/null +++ b/core/agent_runtime/microcompact.py @@ -0,0 +1,256 @@ +"""Microcompact — model-free clearing of old tool results. + +When the runner's context-pressure gate fires, the cheapest first rung on the +compaction ladder is to replace *stale tool results* with a placeholder text +instead of deleting them, clearing real token pressure **before** the +summarization compactor (and its model round-trip) is ever considered. + +Two different jobs, two different modules: + +* :class:`~core.agent_runtime.pruner.ToolResultPruner` **middle-prunes + oversized** results that a single turn blew up, keeping head + tail — the + fix for "this one read was 800KB". +* microcompact **fully clears old, already-consumed** results once the turn + has moved on — the fix for "40 tool results are still resident because the + loop never compacted". + +The runner owns the trigger decision: ``AgentRunner._maybe_compact`` runs only +*after* its own token-meter pressure gate passes, so this module does no +estimation of its own. An internal chars/4 estimate over message text would +miss tool-call arguments, tool schemas, and the provider's real token count — +exactly the Chinese / write-heavy sessions that most need clearing. The caller +decides *when*; this module only applies the clear. + +Three invariants the clear must keep: + +* **The latest tool batch is never touched** — the tool results the model has + not yet read (those issued by the most recent assistant ``tool_calls``) are + kept verbatim regardless of how many parallel calls there were, so the model + always sees the full batch it is about to act on. +* **Recency is kept** — the last ``keep_recent_tool_results`` results are also + never touched, so older-but-still-fresh evidence survives. +* **Pair integrity is kept** — a tool ``result`` message keeps its + ``tool_call_id`` and role; only the *content* becomes the placeholder, so the + assistant→tool pairing the providers require is never broken, and a media + (image) block is passed through untouched. + +A result that was spilled to disk keeps its ``Full output saved to:`` line +appended to the placeholder, so the model can still find the full output again. + +The pass defaults to **off** (``None`` policy) so existing runs are +byte-for-byte unchanged until a host opts in via the user-facing +``agents.defaults.microcompact`` switch. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Any + +__all__ = [ + "MICROCOMPACT_PLACEHOLDER", + "MicrocompactPolicy", + "MicrocompactResult", + "apply_microcompact", + "count_tool_results", + "estimate_tokens", +] + +# A cleared result is recognizable and never mistaken for real content. +MICROCOMPACT_PLACEHOLDER = "[Old tool result content cleared]" + +# ~4 chars/token, used only to score the *savings* of a clear (never to decide +# whether to clear — that gate lives in the runner, which has the real count). +_CHARS_PER_TOKEN = 4.0 + +# The persisted-result reference head (kept as plain literals so this module +# stays import-cycle free; see ``core.agent_runtime.helpers``): a spilled tool +# result's content carries this marker and a ``Full output saved to: `` +# line pointing at the full output on disk. +_PERSISTED_MARKER = "[tool output persisted]" +_SAVED_OUTPUT_PREFIX = "Full output saved to:" + + +def estimate_tokens(text: str) -> float: + """Approximate token count for ``text`` (~4 chars/token).""" + if not text: + return 0.0 + return len(text) / _CHARS_PER_TOKEN + + +def _content_text(content: Any) -> str: + """Flatten a message ``content`` (str or block list) to plain text.""" + if isinstance(content, str): + return content + if isinstance(content, list): + parts: list[str] = [] + for block in content: + if isinstance(block, dict): + text = block.get("text") + if isinstance(text, str): + parts.append(text) + elif isinstance(block, str): + parts.append(block) + return "\n".join(parts) + return "" if content is None else str(content) + + +def _is_tool_result(msg: dict[str, Any]) -> bool: + return msg.get("role") == "tool" + + +def _has_media(msg: dict[str, Any]) -> bool: + """True when the content carries a media (image) block that must survive.""" + content = msg.get("content") + if not isinstance(content, list): + return False + for block in content: + if isinstance(block, dict) and block.get("type") in {"image", "input_image"}: + return True + return False + + +def count_tool_results(messages: Sequence[dict[str, Any]]) -> int: + """How many tool-result messages the policy would consider clearable.""" + return sum(1 for msg in messages if _is_tool_result(msg)) + + +def _latest_tool_batch_ids(messages: Sequence[dict[str, Any]]) -> set[str]: + """``tool_call_id``s of the most recent assistant message that issued calls. + + Those results form the batch the model has *not* read yet — + ``_maybe_compact`` runs before it does — so the whole batch must survive a + clear no matter how many parallel calls it contains. + """ + for msg in reversed(messages): + if msg.get("role") != "assistant": + continue + calls = msg.get("tool_calls") + if isinstance(calls, list): + return { + str(call.get("id")) + for call in calls + if isinstance(call, dict) and call.get("id") + } + # The most recent assistant replied with plain text: its tool batch was + # already consumed, so there is no pending batch to protect. + return set() + return set() + + +def _preserved_reference(content: Any) -> str: + """Pull the on-disk reference lines out of a spilled result's content. + + A result saved to disk carries a ``Full output saved to: `` line; + clearing the preview must not lose the pointer to the full output. + """ + if not isinstance(content, str): + return "" + kept = [ + line.rstrip() + for line in content.splitlines() + if line.strip() == _PERSISTED_MARKER + or line.strip().startswith(_SAVED_OUTPUT_PREFIX) + ] + return "\n".join(kept) + + +@dataclass(frozen=True, slots=True) +class MicrocompactPolicy: + """Microcompact parameters. ``None`` (the runner default) disables the pass. + + The trigger (token pressure) lives entirely in ``AgentRunner._maybe_compact``; + this policy only carries the *shape* of the clear — how much recency to keep + and how small a saving is not worth the bookkeeping. + """ + + # How many of the most recent tool results are kept verbatim (on top of the + # entire latest tool batch, which is always kept). + keep_recent_tool_results: int = 5 + # Never spend a pass to clear less than this — the saving must repay the + # bookkeeping (scored against the ~4 chars/token estimate). + min_savings_tokens: float = 256.0 + # Placeholder substituted for each cleared result's content. + placeholder: str = MICROCOMPACT_PLACEHOLDER + + def __post_init__(self) -> None: + if self.keep_recent_tool_results < 0: + raise ValueError("keep_recent_tool_results must be >= 0") + if self.min_savings_tokens < 0: + raise ValueError("min_savings_tokens must be >= 0") + + +@dataclass(frozen=True, slots=True) +class MicrocompactResult: + """Outcome of one policy application — ``messages`` is a fresh list.""" + + messages: list[dict[str, Any]] + cleared: int + savings_tokens: float + + @property + def applied(self) -> bool: + return self.cleared > 0 + + +def apply_microcompact( + messages: Sequence[dict[str, Any]], + policy: MicrocompactPolicy, +) -> MicrocompactResult: + """Clear old tool results to the placeholder, keeping the recent ``N`` and + the entire latest tool batch, and preserving any saved-output path line. + + The caller (``AgentRunner._maybe_compact``) has already decided there is + token pressure using the provider's real token count, so this function does + no estimation of its own and always applies the clear (subject to the + ``min_savings_tokens`` floor). Non-tool messages and media blocks pass + through untouched; the caller's history is never mutated in place. + """ + tool_index = [i for i, msg in enumerate(messages) if _is_tool_result(msg)] + if not tool_index: + return MicrocompactResult(list(messages), 0, 0.0) + + # Protect the whole latest tool batch (the model is about to read it) plus + # the most recent ``keep_recent`` results (still-fresh evidence). + protected: set[int] = set() + if policy.keep_recent_tool_results > 0: + protected.update(tool_index[-policy.keep_recent_tool_results :]) + latest_ids = _latest_tool_batch_ids(messages) + if latest_ids: + for i in tool_index: + if str(messages[i].get("tool_call_id")) in latest_ids: + protected.add(i) + + clearable_positions = {i for i in tool_index if i not in protected} + if not clearable_positions: + return MicrocompactResult(list(messages), 0, 0.0) + + new_messages: list[dict[str, Any]] = [] + cleared = 0 + savings = 0.0 + for i, msg in enumerate(messages): + if i not in clearable_positions or _has_media(msg): + new_messages.append(msg) + continue + original = _content_text(msg.get("content")) + # Keep the on-disk pointer so the model can still find the full output. + reference = _preserved_reference(msg.get("content")) + # Copy so the caller's history (and any host still holding the old + # message) is never mutated in place — microcompact is persisted by the + # runner reassigning ``messages[:]``, not by side effect here. + copy = dict(msg) + copy["content"] = ( + f"{policy.placeholder}\n{reference}" if reference else policy.placeholder + ) + new_messages.append(copy) + cleared += 1 + savings += estimate_tokens(original) - estimate_tokens(copy["content"]) + + if cleared == 0 or savings < policy.min_savings_tokens: + # Not worth the pass — hand back untouched copies so the caller sees + # no change (convergence: a second pass over an already-cleared history + # clears nothing and saves nothing). + return MicrocompactResult(list(messages), 0, 0.0) + + return MicrocompactResult(new_messages, cleared, savings) diff --git a/core/agent_runtime/runner.py b/core/agent_runtime/runner.py index 353705c5..f9e9b0ad 100644 --- a/core/agent_runtime/runner.py +++ b/core/agent_runtime/runner.py @@ -50,6 +50,10 @@ EvidenceLedger, ) from core.agent_runtime.hook import AgentHook, AgentHookContext +from core.agent_runtime.microcompact import ( + MicrocompactPolicy, + apply_microcompact as _apply_microcompact, +) from core.agent_runtime.pruner import ToolResultPruner from core.agent_runtime.repeat_guard import ( DEFAULT_THRESHOLDS as DEFAULT_REPEAT_THRESHOLDS, @@ -247,6 +251,14 @@ class AgentRunSpec: default_factory=ToolResultPruner ) compaction_strategy: CompactionStrategy = DEFAULT_COMPACTION_STRATEGY + # Optional model-free microcompact pass. When armed, old tool results (all + # but the most recent N and the entire latest tool batch) are replaced by a + # placeholder under the runner's own pressure gate, before the summarizer is + # ever considered — the cheap first rung the ladder was missing. The policy + # carries the recency window and the min-savings floor. ``None`` — the + # default — disables it (existing runs unchanged); the user-facing + # ``agents.defaults.microcompact`` switch arms it. + microcompact: MicrocompactPolicy | None = None # Per-run: the anchored meter carries this conversation's last # reported prompt size, which must not leak into another Session. token_meter: TokenMeter = field(default_factory=DEFAULT_TOKEN_METER_FACTORY) @@ -1803,10 +1815,11 @@ async def _maybe_compact( ) -> list[dict[str, Any]]: """Relieve context pressure with the cheapest sufficient measure. - The ladder (dsh's ordering): pressure gate → model-free prune of + The ladder (dsh's ordering): pressure gate → model-free clear of old + tool results (microcompact, when armed) → model-free prune of oversized tool results → remeasure → only if still over pressure, a summarization round-trip that replaces old turns (C4a). Both - effects are persisted in ``messages``; the drop-based + model-free effects are persisted in ``messages``; the drop-based ``_snip_history`` fallback still follows for anything left over. PreCompact/PostCompact hooks fire around it; a PreCompact ``block`` skips @@ -1825,6 +1838,28 @@ async def _maybe_compact( if estimate is None or estimate <= trigger: return messages + # Cheapest model-free rung first: clear OLD tool results to a + # placeholder, keeping only the most recent N plus the entire latest + # tool batch (the one the model has not read yet). This fires ahead of + # the middle-pruner because it is strictly cheaper (no per-block + # head/tail slicing) and clears the common "40 stale results still + # resident" case whole. The gate above already decided to compact using + # the provider's real token count, so no second estimate is made here. + # A landed clear is durable and never triggers a model round-trip. + if spec.microcompact is not None: + mc = _apply_microcompact(messages, spec.microcompact) + if mc.applied: + messages = mc.messages + logger.info( + "Micro-cleared {} stale tool result(s) for {} (~{} tok)", + mc.cleared, + spec.session_key or "default", + int(mc.savings_tokens), + ) + estimate = self._estimate_prompt(spec, messages) + if estimate is None or estimate <= trigger: + return messages + # Model-free pass first (the dsh ladder): middle-prune oversized tool # results in the persisted history, remeasure, and skip the model # round-trip entirely when pruning alone clears pressure. A landed diff --git a/core/agent_setup.py b/core/agent_setup.py index f1217d6b..85ea0a9c 100644 --- a/core/agent_setup.py +++ b/core/agent_setup.py @@ -480,6 +480,14 @@ def _permission_checker(name: str, arguments: dict[str, Any]): ) return engine.evaluate(name, arguments) + # User-facing microcompact switch (agents.defaults.microcompact): arms the + # model-free stale-tool-result clear in the runner's pressure gate. Read + # defensively so a config without the field still constructs the session. + agents_cfg = getattr(getattr(active_runtime, "config", None), "agents", None) + microcompact_enabled = bool( + getattr(getattr(agents_cfg, "defaults", None), "microcompact", False) + ) + session = AgentSession( provider, tool_registry, @@ -507,6 +515,7 @@ def _permission_checker(name: str, arguments: dict[str, Any]): streaming_transport=streaming_transport, skill_runtime=skill_runtime, context_window_tokens=resolved_execution.context_window, + microcompact=microcompact_enabled, workspace=workspace, execution_profile=resolved_execution, tool_filter=_compose_tool_filters( diff --git a/core/config.py b/core/config.py index 0a53023d..7d8745ea 100644 --- a/core/config.py +++ b/core/config.py @@ -116,6 +116,11 @@ class AgentDefaults(_Base): max_tool_iterations: int = 200 max_tool_result_chars: int = 16_000 context_window_tokens: int = 65_536 + # Model-free microcompact pass: when True, stale tool results are replaced + # by a placeholder under the runner's pressure gate before the summarizer + # runs (see core.agent_runtime.microcompact). Off by default — existing + # runs are unchanged. + microcompact: bool = False class AgentPhase(_Base): diff --git a/core/events/session.py b/core/events/session.py index 6f29066b..fec85a16 100644 --- a/core/events/session.py +++ b/core/events/session.py @@ -32,6 +32,7 @@ from core.agent_runtime.compaction import message_chars from core.agent_runtime.context import EnvironmentContext from core.agent_runtime.hook import AgentHook, AgentHookContext +from core.agent_runtime.microcompact import MicrocompactPolicy from core.agent_runtime.runner import AgentRunner, AgentRunSpec from core.agent_runtime.token_meter import ProviderAnchoredTokenMeter from core.agent_runtime.tools.base import ToolResult @@ -394,6 +395,7 @@ def __init__( hooks_engine: Any | None = None, agent_context: tuple[str, str] | None = None, context_window_tokens: int | None = None, + microcompact: bool = False, workspace: str | Path | None = None, streaming: bool = False, streaming_transport: bool | None = None, @@ -437,6 +439,10 @@ def __init__( # model catalog is what makes "long sessions don't crash" (P2 exit # criterion) true for every AgentSession frontend — exec, TUI, web. self._context_window_tokens = context_window_tokens or context_window_for(model) + # User-facing microcompact switch: arms the model-free stale-tool-result + # clear in the runner's pressure gate (see core.agent_runtime. + # microcompact). Off by default so existing runs are unchanged. + self._microcompact = microcompact # The execution root is model-visible Turn context, not just an # implementation detail of file and shell tools. This keeps Skill # resource paths from being mistaken for the task workspace. @@ -1042,6 +1048,7 @@ def visible_tool_names() -> tuple[str, ...] | None: transient_context_messages=tuple(turn_context_messages), workspace=self._workspace, context_window_tokens=self._context_window_tokens, + microcompact=MicrocompactPolicy() if self._microcompact else None, hook=event_hook, permission_checker=self._permission_checker, approval_callback=self._approval_callback, diff --git a/tests/test_microcompact.py b/tests/test_microcompact.py new file mode 100644 index 00000000..bc769899 --- /dev/null +++ b/tests/test_microcompact.py @@ -0,0 +1,323 @@ +"""Microcompact — unit tests for the clear policy, plus runner integration. + +The unit tests exercise ``apply_microcompact`` directly: what is kept (the +recency window and the whole latest tool batch), what is cleared, and the +copy-on-write / media / pair-integrity invariants. The integration tests drive +``AgentRunner._maybe_compact`` end to end: with the switch on, the summarizer is +skipped and tool-call pairs stay intact; with it off, the history is unchanged. +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import asyncio # noqa: E402 + +from core.agent_runtime.microcompact import ( # noqa: E402 + MICROCOMPACT_PLACEHOLDER, + MicrocompactPolicy, + apply_microcompact, + count_tool_results, + estimate_tokens, +) + + +def _tool_result(tool_call_id: str, text: str) -> dict[str, Any]: + return {"role": "tool", "tool_call_id": tool_call_id, "content": text} + + +def _assistant_calls(ids: list[str]) -> dict[str, Any]: + return { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": cid, + "type": "function", + "function": {"name": "read", "arguments": "{}"}, + } + for cid in ids + ], + } + + +def _history(n_tool: int, chars: int = 4000) -> list[dict[str, Any]]: + """A user turn then n_tool sequential call+result PAIRS (each its own + assistant turn), so earlier results are stale and only the last is the + 'latest batch' the model has not read yet.""" + messages: list[dict[str, Any]] = [{"role": "user", "content": "go"}] + for i in range(n_tool): + messages.append(_assistant_calls([f"call_{i}"])) + messages.append(_tool_result(f"call_{i}", "x" * chars)) + return messages + + +# -- policy validation ------------------------------------------------------- + + +def test_policy_rejects_negative_knobs() -> None: + for kwargs in ( + {"keep_recent_tool_results": -1}, + {"min_savings_tokens": -1.0}, + ): + try: + MicrocompactPolicy(**kwargs) + except ValueError: + pass + else: + raise AssertionError(f"expected ValueError for {kwargs}") + + +def test_placeholder_constant_is_stable() -> None: + assert MICROCOMPACT_PLACEHOLDER == "[Old tool result content cleared]" + + +# -- what is kept / cleared -------------------------------------------------- + + +def test_clears_the_stale_middle_but_keeps_the_recent_tail() -> None: + # keep_recent=2 with 6 results: the last 2 stay, the first 4 clear. + policy = MicrocompactPolicy(keep_recent_tool_results=2, min_savings_tokens=1.0) + messages = _history(6) + result = apply_microcompact(messages, policy) + assert result.applied and result.cleared == 4 + contents = [m.get("content") for m in result.messages if m.get("role") == "tool"] + assert contents[-2:] == ["x" * 4000, "x" * 4000] # recent tail kept verbatim + assert contents[:4] == [MICROCOMPACT_PLACEHOLDER] * 4 # stale middle cleared + + +def test_giant_parallel_batch_is_fully_protected() -> None: + # The point-3 regression: 6 parallel calls in the LATEST batch exceed the + # default keep_recent=5. Without latest-batch protection, the first of the + # six would be cleared before the model read it. All six must survive. + policy = MicrocompactPolicy(min_savings_tokens=1.0) # keep_recent=5 default + messages: list[dict[str, Any]] = [ + {"role": "user", "content": "go"}, + _assistant_calls(["old"]), + _tool_result("old", "x" * 4000), # stale, clearable + _assistant_calls([f"b{j}" for j in range(6)]), # latest batch of 6 + ] + messages.extend(_tool_result(f"b{j}", f"batch result {j}") for j in range(6)) + result = apply_microcompact(messages, policy) + assert result.cleared == 1 # only the stale "old" result + batch = [ + m.get("content") + for m in result.messages + if str(m.get("tool_call_id", "")).startswith("b") + ] + assert batch == [f"batch result {j}" for j in range(6)] # whole batch intact + old = next(m for m in result.messages if m.get("tool_call_id") == "old") + assert old.get("content") == MICROCOMPACT_PLACEHOLDER + + +def test_pair_integrity_and_roles_survive() -> None: + policy = MicrocompactPolicy(keep_recent_tool_results=1, min_savings_tokens=1.0) + messages = _history(3) + result = apply_microcompact(messages, policy) + # Same number of messages, same roles, same tool_call_ids — only content changed. + assert [m.get("role") for m in result.messages] == [m.get("role") for m in messages] + assert [ + m.get("tool_call_id") for m in result.messages if m.get("role") == "tool" + ] == [m.get("tool_call_id") for m in messages if m.get("role") == "tool"] + # Every result still pairs with the assistant that issued it. + declared = {tc["id"] for m in result.messages for tc in (m.get("tool_calls") or [])} + results = { + m.get("tool_call_id") for m in result.messages if m.get("role") == "tool" + } + assert results == declared + + +def test_media_blocks_survive_verbatim() -> None: + policy = MicrocompactPolicy(keep_recent_tool_results=0, min_savings_tokens=1.0) + media = { + "role": "tool", + "tool_call_id": "c0", + "content": [ + {"type": "text", "text": "caption"}, + {"type": "image", "source": {"data": "AAAA"}}, + ], + } + messages = [_assistant_calls(["c0"]), media] + result = apply_microcompact(messages, policy) + assert result.messages[1] is media # untouched + + +def test_original_history_is_never_mutated_in_place() -> None: + policy = MicrocompactPolicy(keep_recent_tool_results=0, min_savings_tokens=1.0) + messages = _history(2) + before = [dict(m) for m in messages] + apply_microcompact(messages, policy) + assert messages == before # caller's list is byte-for-byte unchanged + + +def test_saved_output_path_line_is_preserved() -> None: + # The point-4 regression: a result spilled to disk keeps its + # "Full output saved to:" pointer even after the preview is cleared. The + # spill is the STALE result; a newer call+result follows so it is not the + # protected latest batch. + policy = MicrocompactPolicy(keep_recent_tool_results=0, min_savings_tokens=1.0) + spilled = ( + "[tool output persisted]\n" + "Full output saved to: .deepcode/tool-results/abc.txt\n" + "PREVIEW" + "x" * 400 + ) + messages = [ + _assistant_calls(["c0"]), + _tool_result("c0", spilled), + _assistant_calls(["c1"]), + _tool_result("c1", "recent body"), + ] + result = apply_microcompact(messages, policy) + content = result.messages[1].get("content") + assert "Full output saved to: .deepcode/tool-results/abc.txt" in content + assert "[tool output persisted]" in content + assert "PREVIEW" not in content # the bulky preview is gone + assert content.startswith(MICROCOMPACT_PLACEHOLDER) + # The newest result (the model's pending batch) is untouched. + assert result.messages[3].get("content") == "recent body" + + +def test_below_min_savings_is_a_no_op_and_a_second_pass_converges() -> None: + # Tiny results: clearing saves less than the floor, so nothing happens. + policy = MicrocompactPolicy(keep_recent_tool_results=0, min_savings_tokens=10_000.0) + small = _history(2, chars=20) + first = apply_microcompact(small, policy) + assert not first.applied and first.messages == small + + # An aggressive pass over a big history converges: the second pass clears + # nothing and reports no further saving. + aggressive = MicrocompactPolicy(keep_recent_tool_results=1, min_savings_tokens=1.0) + big = _history(4, chars=4000) + once = apply_microcompact(big, aggressive) + twice = apply_microcompact(once.messages, aggressive) + assert once.applied and once.savings_tokens > 0 + assert not twice.applied and twice.savings_tokens == 0.0 + + +def test_no_tool_results_is_a_no_op() -> None: + messages = [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "yo"}, + ] + result = apply_microcompact(messages, MicrocompactPolicy()) + assert not result.applied and result.messages == messages + + +def test_count_and_estimate_basics() -> None: + assert count_tool_results(_history(3)) == 3 + assert estimate_tokens("") == 0.0 + assert estimate_tokens("x" * 8) == 2.0 + + +# -- integration through AgentRunner._maybe_compact -------------------------- +# +# The maintainer's acceptance bar: one test proves the summarizer is SKIPPED and +# tool-call pairs stay intact when the switch is on; one proves the history is +# UNCHANGED when it is off. + +from core.agent_runtime.runner import AgentRunner, AgentRunSpec # noqa: E402 +from core.agent_runtime.tools.registry import ToolRegistry # noqa: E402 + + +class _CountingProvider: + """Counts chat calls (the only thing the summarizer would make).""" + + def __init__(self) -> None: + self.calls = 0 + + def get_default_model(self) -> str: + return "gpt-5.4" + + async def chat_with_retry(self, **kwargs: Any): # pragma: no cover - not reached + self.calls += 1 + raise AssertionError("the summarizer should not be called in these tests") + + +def _pressure_history(*, old: int = 6, batch: int = 6) -> list[dict[str, Any]]: + """A history over the pressure trigger: `old` huge stale results + a latest + parallel batch of `batch` small results the model has not read yet.""" + big = "lorem ipsum dolor sit amet " * 1500 # ~42K chars each + messages: list[dict[str, Any]] = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "do the work"}, + ] + for i in range(old): + messages.append(_assistant_calls([f"old_{i}"])) + messages.append(_tool_result(f"old_{i}", f"OLD{i} " + big)) + batch_ids = [f"b{j}" for j in range(batch)] + messages.append(_assistant_calls(batch_ids)) + messages.extend( + _tool_result(bid, f"BATCH{j} result body") for j, bid in enumerate(batch_ids) + ) + return messages + + +def _spec(**kw: Any) -> AgentRunSpec: + base: dict[str, Any] = dict( + initial_messages=[], + tools=ToolRegistry(), + model="gpt-5.4", + max_iterations=3, + max_tool_result_chars=100_000, + context_window_tokens=8_000, + # Isolate microcompact: the middle-pruner must not get the credit. + tool_result_pruner=None, + ) + base.update(kw) + return AgentRunSpec(**base) + + +def test_microcompact_on_skips_summarizer_and_keeps_tool_pairs() -> None: + provider = _CountingProvider() + messages = _pressure_history() + spec = _spec(microcompact=MicrocompactPolicy(min_savings_tokens=1.0)) + out = asyncio.run(AgentRunner(provider)._maybe_compact(spec, messages)) + + # The summarizer never ran — clearing the stale results alone relieved the + # pressure the gate had detected. + assert provider.calls == 0 + + # The stale (already-read) results are replaced by the placeholder... + old = [ + m.get("content") + for m in out + if str(m.get("tool_call_id", "")).startswith("old_") + ] + assert len(old) == 6 and all(c == MICROCOMPACT_PLACEHOLDER for c in old) + # ...while the latest batch (the model is about to read it) survives whole. + batch = [ + m.get("content") for m in out if str(m.get("tool_call_id", "")).startswith("b") + ] + assert batch == [f"BATCH{j} result body" for j in range(6)] + + # Tool-call pairs stay intact: every result still has its id and matches an + # assistant tool_call — nothing dropped, nothing orphaned. + declared = {tc["id"] for m in out for tc in (m.get("tool_calls") or [])} + results = {m.get("tool_call_id") for m in out if m.get("role") == "tool"} + assert results == declared + + +def test_microcompact_off_leaves_history_unchanged() -> None: + from core.harness.hooks.engine import ContextOutcome + + provider = _CountingProvider() + messages = _pressure_history() + + # Block the summarizer so the ONLY thing that could mutate history is the + # microcompact rung (which runs before the PreCompact hook). With the + # switch off it must not fire, so the history comes back byte-for-byte. + async def _block(trigger: Any) -> ContextOutcome: + return ContextOutcome(block=True, block_reason="skip") + + spec = _spec(microcompact=None, pre_compact_hook=_block) + out = asyncio.run(AgentRunner(provider)._maybe_compact(spec, messages)) + + assert out is messages # unchanged, same object + assert not any(m.get("content") == MICROCOMPACT_PLACEHOLDER for m in out) + assert provider.calls == 0