diff --git a/agent/agent_init.py b/agent/agent_init.py index 1f77240b0017..2f87e9aa7800 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -2020,6 +2020,40 @@ def _parse_prune_int(raw, default): _compression_cfg.get("proactive_prune_min_reclaim_tokens", 4096), 4096 ), ) + # In-place tool-result prune (head+tail+marker) — opt-in, default OFF. + # Runs before the summarization region is selected in the in-loop + # compression pre-pass and rewrites the persisted transcript through the + # same archive_and_compact mechanism as in-place compaction. Only tool + # results over threshold_chars are touched, and only down to + # head_chars + marker + tail_chars. The compressor re-validates the + # budget combination (head + marker + tail must fit threshold) and + # disables the feature if it is unsatisfiable. + _tool_result_prune_cfg = _compression_cfg.get("tool_result_prune", {}) + if not isinstance(_tool_result_prune_cfg, dict): + _tool_result_prune_cfg = {} + compression_tool_result_prune = { + "enabled": is_truthy_value( + _tool_result_prune_cfg.get("enabled"), default=False + ), + "threshold_chars": max( + 1, + _parse_prune_int( + _tool_result_prune_cfg.get("threshold_chars", 8192), 8192 + ), + ), + "head_chars": max( + 0, + _parse_prune_int( + _tool_result_prune_cfg.get("head_chars", 4096), 4096 + ), + ), + "tail_chars": max( + 0, + _parse_prune_int( + _tool_result_prune_cfg.get("tail_chars", 1024), 1024 + ), + ), + } # protect_first_n is the number of non-system messages to protect at # the head, in addition to the system prompt (which is always # implicitly protected by the compressor). Floor at 0 — a value of @@ -2563,6 +2597,7 @@ def _parse_prune_int(raw, default): proactive_prune_min_result_chars=compression_proactive_prune_min_chars, proactive_prune_min_reclaim_tokens=compression_proactive_prune_min_reclaim, min_tail_user_messages=compression_min_tail_users, + tool_result_prune=compression_tool_result_prune, ) _bind_session_state = getattr(agent.context_compressor, "bind_session_state", None) if callable(_bind_session_state): diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 3eb05b6b9713..46a1e8ed2496 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -521,6 +521,190 @@ def _prune_stale_reasoning_replay(messages: List[Dict[str, Any]]) -> int: # is never re-summarized away on a later prune pass). _PRUNE_MIN_CHARS = 200 +# Marker substituted for the removed middle span of an in-place-pruned tool +# result (compression.tool_result_prune). A standalone line between blank +# lines — model-visible text, pinned verbatim: consumers (tests, future +# renderers) may rely on the exact bytes. Must never change; it also doubles +# as the idempotence/immunity marker that keeps the deterministic demote +# pass (_prune_old_tool_results) from re-replacing an already-pruned result. +PRUNE_MARKER = "\n\n[... tool result middle pruned ...]\n\n" + +# Conservative defaults for compression.tool_result_prune. A pruned result +# is head_chars + marker + tail_chars — validation (see +# resolve_tool_result_prune_config) requires that sum to stay at or below +# threshold_chars so a prune can never emit MORE than it removes. +_TOOL_RESULT_PRUNE_DEFAULTS = { + "enabled": False, + "threshold_chars": 8192, + "head_chars": 4096, + "tail_chars": 1024, +} + + +def _coerce_prune_budget(value: Any, default: int, *, positive: bool) -> int: + """Coerce one numeric prune budget with the hardened parse semantics. + + Booleans are rejected (bool subclasses int — YAML `true` would coerce to + 1), fractional floats are rejected rather than truncated, integral floats + and numeric strings are accepted; anything else falls back to ``default``. + ``positive=True`` requires >= 1 (threshold), otherwise >= 0 (head/tail). + """ + if isinstance(value, bool): + return default + if isinstance(value, int): + ival = value + elif isinstance(value, float): + if not value.is_integer(): + return default + ival = int(value) + else: + try: + ival = int(str(value).strip()) + except (TypeError, ValueError): + return default + if positive: + return ival if ival >= 1 else default + return ival if ival >= 0 else default + + +def resolve_tool_result_prune_config( + raw: Optional[Dict[str, Any]], +) -> tuple[bool, int, int, int]: + """Resolve and validate ``compression.tool_result_prune``. + + Returns ``(enabled, threshold_chars, head_chars, tail_chars)``. + Conservative by design: any malformed or unsatisfiable combination + disables the feature (with a logged warning) rather than risking a prune + that emits more characters than it removes or produces an empty body. + ``head_chars + len(PRUNE_MARKER) + tail_chars`` must be at most + ``threshold_chars`` — otherwise the pruned result could exceed the very + budget it is meant to enforce. + """ + if not isinstance(raw, dict): + return ( + _TOOL_RESULT_PRUNE_DEFAULTS["enabled"], + _TOOL_RESULT_PRUNE_DEFAULTS["threshold_chars"], + _TOOL_RESULT_PRUNE_DEFAULTS["head_chars"], + _TOOL_RESULT_PRUNE_DEFAULTS["tail_chars"], + ) + enabled = bool(raw.get("enabled", _TOOL_RESULT_PRUNE_DEFAULTS["enabled"])) + threshold = _coerce_prune_budget( + raw.get("threshold_chars", _TOOL_RESULT_PRUNE_DEFAULTS["threshold_chars"]), + _TOOL_RESULT_PRUNE_DEFAULTS["threshold_chars"], + positive=True, + ) + head = _coerce_prune_budget( + raw.get("head_chars", _TOOL_RESULT_PRUNE_DEFAULTS["head_chars"]), + _TOOL_RESULT_PRUNE_DEFAULTS["head_chars"], + positive=False, + ) + tail = _coerce_prune_budget( + raw.get("tail_chars", _TOOL_RESULT_PRUNE_DEFAULTS["tail_chars"]), + _TOOL_RESULT_PRUNE_DEFAULTS["tail_chars"], + positive=False, + ) + if head + len(PRUNE_MARKER) + tail > threshold: + logger.warning( + "compression.tool_result_prune: head_chars (%d) + marker (%d) + " + "tail_chars (%d) exceeds threshold_chars (%d) — disabling the " + "feature (conservative default)", + head, len(PRUNE_MARKER), tail, threshold, + ) + return ( + False, + _TOOL_RESULT_PRUNE_DEFAULTS["threshold_chars"], + _TOOL_RESULT_PRUNE_DEFAULTS["head_chars"], + _TOOL_RESULT_PRUNE_DEFAULTS["tail_chars"], + ) + return enabled, threshold, head, tail + + +def prune_tool_result_content( + content: Any, + threshold_chars: int, + head_chars: int, + tail_chars: int, + marker: str = PRUNE_MARKER, +) -> Any: + """Prune an oversized tool-result body to head + marker + tail. + + In-place by contract: the caller keeps the message node (same position, + same role, same ``tool_call_id``) and swaps only ``content``. Returns the + INPUT object unchanged when the content is within budget, so callers can + detect a rewrite by identity (``result is content``). + + String bodies are sliced by Unicode code point: Python ``str`` indexes + code points (no surrogate pairs), so a retained boundary can never split + a surrogate pair — matching the TS reference implementation. + + List bodies (multimodal parts) prune only ``type == "text"`` parts, + preserving every non-text part and the relative order of all parts. Text + spans across parts are measured and removed as one contiguous span, with + the marker inserted at the first part that intersects the removed span + (same semantics as the TS pruner). Any other body shape is never pruned. + + Requires ``head_chars + len(marker) + tail_chars <= threshold_chars`` + (enforced by ``resolve_tool_result_prune_config``) so the result is + strictly smaller than the input and within budget; the pruned output is + also idempotent — a second pass finds it within budget and leaves it + byte-identical. + """ + if isinstance(content, str): + total = len(content) + if total <= threshold_chars: + return content + removed_start = head_chars + removed_end = total - tail_chars + return content[:removed_start] + marker + content[removed_end:] + + if isinstance(content, list): + text_parts = [ + part + for part in content + if isinstance(part, dict) + and part.get("type") == "text" + and isinstance(part.get("text"), str) + ] + total = sum(len(part["text"]) for part in text_parts) + if total <= threshold_chars: + return content + removed_start = head_chars + removed_end = total - tail_chars + pruned: List[Any] = [] + consumed = 0 + marker_inserted = False + for part in content: + if not ( + isinstance(part, dict) + and part.get("type") == "text" + and isinstance(part.get("text"), str) + ): + pruned.append(part) + continue + text = part["text"] + block_start = consumed + block_end = block_start + len(text) + head_end = min(len(text), max(0, removed_start - block_start)) + tail_start = min(len(text), max(0, removed_end - block_start)) + intersects = block_start < removed_end and block_end > removed_start + m = marker if (intersects and not marker_inserted) else "" + if m: + marker_inserted = True + new_text = text[:head_end] + m + text[tail_start:] + if new_text: + pruned.append({**part, "text": new_text}) + consumed = block_end + # Validated budgets (head + marker + tail <= threshold < total) + # guarantee a non-empty removed text span, so marker_inserted is + # always True here; the guard mirrors the TS implementation's + # fail-safe and must never fire in practice. + if not marker_inserted: + return content + return pruned + + return content # unknown body shape — never pruned + + # Non-response sentinels the clarify callbacks embed as ``user_response`` when # the user never actually answered (timeout / no-user contexts). These must # not be quoted as a user answer during compaction. Sources: @@ -1082,6 +1266,57 @@ def _last_assistant_index(messages: "List[Dict[str, Any]]") -> int: return -1 +def _tool_prune_boundary( + result: List[Dict[str, Any]], + protect_tail_count: int, + protect_tail_tokens: int | None, +) -> int: + """Compute the index below which tool messages are prunable. + + Shared by ``_prune_old_tool_results`` and ``_prune_tool_results_in_place`` + so every deterministic tool-result pre-pass protects the same recent tail. + + Token-budget approach when ``protect_tail_tokens`` is given: walk + backward accumulating tokens, capping the message-count floor the same + way tail-cut does so a default ``protect_last_n=20`` cannot lock a bulky + recent tool run outside the compressible / prunable window (#61932). + Same newest-turn-only thinking charge as the tail-cut walk (#73624) — + this boundary decides which tool results stay prunable, and overcharging + stale thinking shrinks that window. + + The budget walk is translated into a "protected count", the floor is + applied in count-space (where ``max`` reads naturally: protect at least + ``min_protect`` messages or whatever the budget reserved, whichever is + more), then converted back to a prune boundary. Doing this in index-space + with ``max`` would invert the direction (smaller index = MORE protected), + so a generous budget would silently get truncated back down to + ``min_protect``. + """ + if protect_tail_tokens is not None and protect_tail_tokens > 0: + accumulated = 0 + boundary = len(result) + min_protect = min( + protect_tail_count, + len(result), + _MAX_TAIL_MESSAGE_FLOOR, + ) + _newest_asst_idx = _last_assistant_index(result) + for i in range(len(result) - 1, -1, -1): + msg = result[i] + msg_tokens = _estimate_msg_budget_tokens( + msg, charge_stale_thinking=(i == _newest_asst_idx) + ) + if accumulated + msg_tokens > protect_tail_tokens and (len(result) - i) >= min_protect: + boundary = i + break + accumulated += msg_tokens + boundary = i + budget_protect_count = len(result) - boundary + protected_count = max(budget_protect_count, min_protect) + return len(result) - protected_count + return len(result) - protect_tail_count + + def _content_text_for_contains(content: Any) -> str: """Return a best-effort text view of message content. @@ -2532,6 +2767,7 @@ def __init__( proactive_prune_min_result_chars: int = 8000, proactive_prune_min_reclaim_tokens: int = 4096, min_tail_user_messages: int = 1, + tool_result_prune: Optional[Dict[str, Any]] = None, ): self.model = model self.base_url = base_url @@ -2590,6 +2826,22 @@ def __init__( # one until the prompt has regrown the tokens just reclaimed. self._proactive_prune_rearm_tokens: int = 0 self.min_tail_user_messages = min_tail_user_messages + # In-place tool-result prune (compression.tool_result_prune): + # oversized tool results (role="tool", content > threshold_chars) are + # pruned in their OWN message node — same position, role and + # tool_call_id — down to head + marker + tail, as a no-LLM pre-pass + # that runs BEFORE the summarization region is selected. Default OFF + # (opt-in): each commit rewrites already-sent history, a prompt-cache + # break that is only sanctioned at a compression boundary — which is + # exactly where this runs. See resolve_tool_result_prune_config for + # the conservative validation (an unsatisfiable budget disables the + # feature rather than risking a prune that grows the message). + ( + self._tool_result_prune_enabled, + self._tool_result_prune_threshold_chars, + self._tool_result_prune_head_chars, + self._tool_result_prune_tail_chars, + ) = resolve_tool_result_prune_config(tool_result_prune) self.summary_target_ratio = max(0.10, min(summary_target_ratio, 0.80)) self.quiet_mode = quiet_mode # Output-token reservation: the provider carves max_tokens out of the @@ -3096,6 +3348,113 @@ def _automatic_compression_blocked_locally(self) -> bool: # Tool output pruning (cheap pre-pass, no LLM call) # ------------------------------------------------------------------ + def _prune_tool_results_in_place( + self, + messages: List[Dict[str, Any]], + protect_tail_count: int, + protect_tail_tokens: int | None = None, + ) -> tuple[List[Dict[str, Any]], int]: + """Prune oversized tool results in place: head + marker + tail. + + Config-gated (``compression.tool_result_prune.enabled``, default + off). Runs BEFORE the summarization region is selected: shrinking + oversized tool bodies often makes the whole transcript fit the tail + budget, so compression returns at the no-compressible-window check + without ever calling the summarizer. + + Unlike ``_prune_old_tool_results`` (which REPLACES an old tool result + with a 1-line summary), this keeps the node in place — same position, + same role, same ``tool_call_id`` — and rewrites only ``content`` to + head + marker + tail (see ``prune_tool_result_content``). It never + touches non-tool messages and never breaks tool-call/result pairing: + the result row itself is rewritten, not removed. + + Tail protection mirrors the deterministic demote pass + (``_tool_prune_boundary``): only tool results older than the + protected tail (``protect_tail_count`` / ``protect_tail_tokens``) + are pruned, so the active turn's recent tool output stays verbatim. + + A committed prune rewrites message bodies the provider has already + seen — a prompt-cache break exactly like a compression boundary. + Compression is the one sanctioned cache break, and this pass only + ever runs at a compression event, so no extra hysteresis gate is + needed. The commit is durable through the same mechanism as in-place + compaction (``archive_and_compact``), mirroring + ``prune_tool_results_only``: the pruned dicts are stamped + ``_DB_PERSISTED_MARKER`` so the next append-only flush skips them + instead of re-inserting them on top of the archived originals. When + the session store is bound but the atomic rewrite is unavailable or + fails, the prune is NOT committed — the input object is returned + unchanged so the in-memory transcript can never drift from the DB + (a failed rewrite followed by an append-only flush would duplicate + the transcript on resume). Without a bound store there is no flush to + desync, so the in-memory prune still lands for the current turn. + + Returns ``(messages, 0)`` — the input object — when disabled or when + nothing was pruned (standard no-op caller contract). + """ + if not self._tool_result_prune_enabled: + return messages, 0 + if not messages: + return messages, 0 + # Capability gate BEFORE the scan (mirrors prune_tool_results_only): + # a bound store that can't persist the prune atomically (duck-typed / + # plugin session store without archive_and_compact) makes every prune + # either a duplicate-on-resume hazard or a permanent no-op — don't + # pay the scan for it. + session_db = getattr(self, "_session_db", None) + session_id = getattr(self, "_session_id", "") + if ( + session_db + and session_id + and not callable(getattr(session_db, "archive_and_compact", None)) + ): + return messages, 0 + + result = [m.copy() for m in messages] + boundary = _tool_prune_boundary( + result, protect_tail_count, protect_tail_tokens + ) + pruned = 0 + for i in range(max(0, boundary)): + msg = result[i] + if msg.get("role") != "tool": + continue + content = msg.get("content") + new_content = prune_tool_result_content( + content, + self._tool_result_prune_threshold_chars, + self._tool_result_prune_head_chars, + self._tool_result_prune_tail_chars, + ) + if new_content is content: + continue + new_msg = {**msg, "content": new_content} + # Content rewritten → the api_content sidecar (the exact bytes + # previously sent) is stale; drop it so replay can't resend the + # pre-prune bytes. Same rule every content-rewrite path follows + # (drop_stale_api_content) — cost is one cache boundary miss, + # never wrong content. + drop_stale_api_content(new_msg) + result[i] = new_msg + pruned += 1 + if not pruned: + return messages, 0 + if session_db and session_id: + try: + session_db.archive_and_compact(session_id, result) + except Exception as exc: + logger.warning( + "In-place tool-result prune DB commit failed; keeping the " + "original transcript: %s", + exc, + ) + return messages, 0 + for msg in result: + if isinstance(msg, dict): + msg[_DB_PERSISTED_MARKER] = True + return result, pruned + def _prune_old_tool_results( self, messages: List[Dict[str, Any]], protect_tail_count: int, protect_tail_tokens: int | None = None, @@ -3150,44 +3509,9 @@ def _prune_old_tool_results( call_id_to_tool[cid] = (name, args_str) # Determine the prune boundary - if protect_tail_tokens is not None and protect_tail_tokens > 0: - # Token-budget approach: walk backward accumulating tokens. - # Cap the message-count floor the same way tail-cut does so a - # default protect_last_n=20 cannot lock a bulky recent tool run - # outside the compressible / prunable window (#61932). - accumulated = 0 - boundary = len(result) - min_protect = min( - protect_tail_count, - len(result), - _MAX_TAIL_MESSAGE_FLOOR, - ) - # Same newest-turn-only thinking charge as the tail-cut walk - # (#73624) — this boundary decides which tool results stay - # prunable, and overcharging stale thinking shrinks that window. - _newest_asst_idx = _last_assistant_index(result) - for i in range(len(result) - 1, -1, -1): - msg = result[i] - msg_tokens = _estimate_msg_budget_tokens( - msg, charge_stale_thinking=(i == _newest_asst_idx) - ) - if accumulated + msg_tokens > protect_tail_tokens and (len(result) - i) >= min_protect: - boundary = i - break - accumulated += msg_tokens - boundary = i - # Translate the budget walk into a "protected count", apply the - # floor in count-space (where `max` reads naturally: protect at - # least `min_protect` messages or whatever the budget reserved, - # whichever is more), then convert back to a prune boundary. - # Doing this in index-space with `max` would invert the direction - # (smaller index = MORE protected), so a generous budget would - # silently get truncated back down to `min_protect`. - budget_protect_count = len(result) - boundary - protected_count = max(budget_protect_count, min_protect) - prune_boundary = len(result) - protected_count - else: - prune_boundary = len(result) - protect_tail_count + prune_boundary = _tool_prune_boundary( + result, protect_tail_count, protect_tail_tokens + ) # Pass 1: Deduplicate identical tool results. # When the same file is read multiple times, keep only the most recent @@ -3246,6 +3570,12 @@ def _demote_tool_result_at(idx: int, *, spare_protected_skills: bool = True) -> return True if not isinstance(content, str): return False + # In-place-pruned results (head+tail+marker, opt-in + # compression.tool_result_prune) are already shrunk + # deterministically while keeping their content; re-demoting them + # to a 1-line summary would defeat the opt-in retention contract. + if PRUNE_MARKER in content: + return False if not content or content == _PRUNED_TOOL_PLACEHOLDER: return False if content.startswith("[Duplicate tool output"): @@ -6431,7 +6761,12 @@ def compress( """Compress conversation messages by summarizing middle turns. Algorithm: - 1. Prune old tool results (cheap pre-pass, no LLM call) + 1. In-place tool-result prune (opt-in, compression.tool_result_prune, + no LLM call): oversized tool results are shrunk in their own node + to head + marker + tail, BEFORE region selection — which often + makes the whole transcript fit the tail budget, skipping the + summarizer entirely (also rewrites the persisted history). + 1b. Prune old tool results (cheap pre-pass, no LLM call) 2. Protect head messages (system prompt + first exchange) 3. Find tail boundary by token budget (~20K tokens of recent context) 4. Summarize middle turns with structured LLM prompt (skipped @@ -6519,7 +6854,30 @@ def compress( display_tokens = current_tokens if current_tokens else self.last_prompt_tokens or estimate_messages_tokens_rough(messages) - # Phase 1: Prune old tool results (cheap, no LLM call) + # Phase 1: In-place tool-result prune (head+tail+marker, no LLM call). + # Config-gated (compression.tool_result_prune, default off). Runs + # BEFORE the deterministic demote pass AND before region selection: + # an oversized tool result is shrunk in its own node (same position, + # role, tool_call_id) to head + marker + tail, which often makes the + # whole transcript fit the tail budget — so compression returns at + # the no-compressible-window check below without ever calling the + # summarizer. The commit is durable through the same + # archive_and_compact mechanism as in-place compaction, so the + # reclaimed state survives resume even when no summarization follows. + if self._tool_result_prune_enabled: + messages, in_place_pruned = self._prune_tool_results_in_place( + messages, + protect_tail_count=self.protect_last_n, + protect_tail_tokens=self.tail_token_budget, + ) + if in_place_pruned and not self.quiet_mode: + logger.info( + "Pre-compression: pruned %d oversized tool result(s) in " + "place (head+tail+marker)", + in_place_pruned, + ) + + # Phase 1b: Prune old tool results (cheap, no LLM call) messages, pruned_count = self._prune_old_tool_results( messages, protect_tail_count=self.protect_last_n, protect_tail_tokens=self.tail_token_budget, diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 18ac01785358..03992c12f42f 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -794,6 +794,34 @@ # session_search and recoverable, not deleted. # Default True since 2107b86024; set False to # restore the legacy rotating-compaction path. + "tool_result_prune": { # opt-in, default OFF: oversized tool results + # (role="tool", content > threshold_chars) are + # pruned in their OWN message node — same + # position, role and tool_call_id — down to + # head_chars + marker + tail_chars, as a + # no-LLM pre-pass that runs BEFORE the + # summarization region is selected in the + # in-loop compression. Shrinking oversized + # tool bodies often makes the transcript fit + # the tail budget, avoiding the summarizer + # entirely. Tail protection mirrors the + # region selection (recent tool output stays + # verbatim); the commit rewrites the + # persisted transcript via the same + # archive_and_compact mechanism as in-place + # compaction, so the reclaimed state is + # durable even when no summarization + # follows. Each commit is a prompt-cache + # break — sanctioned because it only ever + # runs at a compression event. head_chars + + # marker + tail_chars must stay <= + # threshold_chars; an unsatisfiable budget + # disables the feature. + "enabled": False, + "threshold_chars": 8192, + "head_chars": 4096, + "tail_chars": 1024, + }, "model_thresholds": {}, # Per-model threshold overrides. Keys are # substring-matched against the model name # (longest match wins); values replace the diff --git a/tests/agent/test_tool_result_prune.py b/tests/agent/test_tool_result_prune.py new file mode 100644 index 000000000000..a23f3e71deda --- /dev/null +++ b/tests/agent/test_tool_result_prune.py @@ -0,0 +1,553 @@ +"""Tests for the in-place tool-result prune (``compression.tool_result_prune``). + +Oversized tool results (``role="tool"``, content > threshold_chars) are +pruned in their OWN message node — same position, same role, same +``tool_call_id`` — down to head + marker + tail, as a no-LLM pre-pass that +runs BEFORE the summarization region is selected in the in-loop compression +(``ContextCompressor.compress``). The commit rewrites the persisted +transcript through the same ``archive_and_compact`` mechanism as in-place +compaction, so the reclaimed state is durable even when no summarization +follows. + +Mirrors the construction/patching conventions in +test_proactive_tool_result_pruning.py. +""" + +from unittest.mock import patch + +from agent.context_compressor import ( + PRUNE_MARKER, + _DB_PERSISTED_MARKER, + _TOOL_RESULT_PRUNE_DEFAULTS, + ContextCompressor, + prune_tool_result_content, + resolve_tool_result_prune_config, +) + +LARGE_WINDOW = 1_000_000 +BIG_CHARS = 9200 # > default threshold 8192; ~2300 rough tokens when ASCII + + +def _compressor(**kw): + defaults = dict( + model="test", + quiet_mode=True, + threshold_percent=0.50, + protect_first_n=2, + protect_last_n=2, + ) + defaults.update(kw) + with patch( + "agent.context_compressor.get_model_context_length", + return_value=LARGE_WINDOW, + ): + c = ContextCompressor(**defaults) + # Context length is resolved lazily on first access (outside the + # patch context); pre-set it so every derived budget is deterministic. + c._resolved_context_length = LARGE_WINDOW + return c + + +def _assistant_call(cid, name="terminal", args='{"cmd":"ls"}'): + return { + "role": "assistant", + "content": "", + "tool_calls": [ + {"id": cid, "type": "function", + "function": {"name": name, "arguments": args}} + ], + } + + +def _tool_msg(cid, content): + return {"role": "tool", "tool_call_id": cid, "content": content} + + +def _build(n_big, tail_pairs=2, big_chars=BIG_CHARS): + """system + n_big oversized tool-result pairs + tail pairs + final user.""" + msgs = [{"role": "system", "content": "sys"}] + for i in range(n_big): + cid = f"big_{i}" + msgs.append(_assistant_call(cid)) + msgs.append(_tool_msg(cid, chr(65 + (i % 26)) * big_chars)) + for i in range(tail_pairs): + cid = f"tail_{i}" + msgs.append(_assistant_call(cid)) + msgs.append(_tool_msg(cid, "ok")) + msgs.append({"role": "user", "content": "final question"}) + return msgs + + +def _tool_by_id(msgs, cid): + return [ + m for m in msgs + if m.get("role") == "tool" and m.get("tool_call_id") == cid + ][0] + + +class _FakeSessionDB: + """Minimal session store recording archive_and_compact calls.""" + + def __init__(self, raise_on_compact=False): + self.calls = [] + self.raise_on_compact = raise_on_compact + + def archive_and_compact(self, session_id, messages, **kwargs): + if self.raise_on_compact: + raise RuntimeError("boom") + self.calls.append((session_id, list(messages), kwargs)) + + +# --------------------------------------------------------------------------- +# Pure function: prune_tool_result_content +# --------------------------------------------------------------------------- + + +def test_within_budget_returns_input_object(): + content = "x" * 100 + assert prune_tool_result_content(content, 8192, 4096, 1024) is content + parts = [{"type": "text", "text": "y" * 50}] + assert prune_tool_result_content(parts, 8192, 4096, 1024) is parts + + +def test_exact_threshold_unchanged(): + content = "x" * 8192 + assert prune_tool_result_content(content, 8192, 4096, 1024) is content + + +def test_head_tail_marker_shape(): + head, tail = "HEAD" * 1024, "TAIL" * 256 + content = head + ("M" * 10_000) + tail + out = prune_tool_result_content(content, 8192, 4096, 1024) + assert out.startswith(head) + assert out.endswith(tail) + assert out.count(PRUNE_MARKER) == 1 + # Marker is a standalone line between blank lines. + assert PRUNE_MARKER.startswith("\n\n") and PRUNE_MARKER.endswith("\n\n") + assert out[len(head):].startswith(PRUNE_MARKER) + removed = "M" * 10_000 + assert removed not in out + assert len(out) == 4096 + len(PRUNE_MARKER) + 1024 + + +def test_unicode_astral_code_point_slicing(): + # Each emoji is ONE Python code point (no surrogate pairs in str). + content = "😀" * 10_000 + out = prune_tool_result_content(content, 8192, 4096, 1024) + assert out == ("😀" * 4096) + PRUNE_MARKER + ("😀" * 1024) + # Slicing by code point can never split a surrogate pair: every retained + # boundary is a full astral character. + assert set(out.replace(PRUNE_MARKER, "")) == {"😀"} + # Mixed CJK + ASCII keeps the same code-point semantics. + mixed = "漢" * 5000 + "a" * 5000 + out2 = prune_tool_result_content(mixed, 8192, 4096, 1024) + assert out2 == ("漢" * 4096) + PRUNE_MARKER + ("a" * 1024) + + +def test_zero_head_and_tail_budgets(): + content = "x" * 10_000 + out = prune_tool_result_content(content, 8192, 0, 0) + assert out == PRUNE_MARKER + out2 = prune_tool_result_content(content, 8192, 0, 1024) + assert out2 == PRUNE_MARKER + ("x" * 1024) + + +def test_list_content_prunes_text_keeps_other_parts(): + parts = [ + {"type": "text", "text": "A" * 6000}, + {"type": "image_url", "image_url": {"url": "data:img"}}, + {"type": "text", "text": "B" * 6000}, + {"type": "text", "text": "C" * 6000}, + ] + out = prune_tool_result_content(parts, 8192, 4096, 1024) + # Non-text part survives byte-identical, in the same position. + assert out[1] is parts[1] + # Text total (18_000) > threshold → span [4096, 16_976) removed; the + # marker lands in the first intersecting text part. + total_text = "".join( + p["text"] for p in out if isinstance(p, dict) and p.get("type") == "text" + ) + assert total_text == ("A" * 4096) + PRUNE_MARKER + ("C" * 1024) + # Pruned output is within budget and smaller than the input. + text_after = sum( + len(p["text"]) for p in out + if isinstance(p, dict) and p.get("type") == "text" + ) + assert text_after == 4096 + len(PRUNE_MARKER) + 1024 <= 8192 + + +def test_list_content_within_budget_unchanged(): + parts = [{"type": "text", "text": "x" * 100}, {"type": "image_url", "image_url": {"url": "u"}}] + assert prune_tool_result_content(parts, 8192, 4096, 1024) is parts + + +def test_unknown_shapes_never_pruned(): + for shape in (None, 42, {"_multimodal": True, "content": "x" * 20_000}): + assert prune_tool_result_content(shape, 8192, 4096, 1024) is shape + + +def test_idempotent(): + content = "x" * 20_000 + once = prune_tool_result_content(content, 8192, 4096, 1024) + twice = prune_tool_result_content(once, 8192, 4096, 1024) + assert twice is once # already within budget → byte-identical, no rewrite + + +# --------------------------------------------------------------------------- +# Config resolution +# --------------------------------------------------------------------------- + + +def test_config_defaults_disabled(): + assert resolve_tool_result_prune_config(None) == ( + False, 8192, 4096, 1024, + ) + assert resolve_tool_result_prune_config({}) == (False, 8192, 4096, 1024) + + +def test_config_custom_values(): + assert resolve_tool_result_prune_config( + {"enabled": True, "threshold_chars": 4096, "head_chars": 2048, "tail_chars": 512} + ) == (True, 4096, 2048, 512) + + +def test_config_unsatisfiable_budget_disables(): + # head + marker + tail would exceed threshold → feature disabled. + resolved = resolve_tool_result_prune_config( + {"enabled": True, "threshold_chars": 100, "head_chars": 200, "tail_chars": 50} + ) + assert resolved == (False, 8192, 4096, 1024) + + +def test_config_boolean_and_fractional_rejected(): + # bool subclasses int — YAML `threshold_chars: true` must not coerce to 1. + assert resolve_tool_result_prune_config( + {"enabled": True, "threshold_chars": True} + )[1] == 8192 + # Fractional floats are rejected, not truncated. + assert resolve_tool_result_prune_config( + {"enabled": True, "head_chars": 1.5} + )[2] == 4096 + # Integral floats and numeric strings are accepted. + assert resolve_tool_result_prune_config( + {"enabled": True, "threshold_chars": 4096.0, "head_chars": "2048"} + )[1:3] == (4096, 2048) + + +def test_compressor_defaults_disabled(): + c = _compressor() + assert c._tool_result_prune_enabled is False + assert c._tool_result_prune_threshold_chars == _TOOL_RESULT_PRUNE_DEFAULTS["threshold_chars"] + + +def test_compressor_honors_config(): + c = _compressor(tool_result_prune={"enabled": True, "tail_chars": 256}) + assert c._tool_result_prune_enabled is True + assert c._tool_result_prune_tail_chars == 256 + + +def test_default_config_has_conservative_prune_defaults(): + from hermes_cli.config_defaults import DEFAULT_CONFIG + + trp = DEFAULT_CONFIG["compression"]["tool_result_prune"] + assert trp["enabled"] is False + assert trp["threshold_chars"] == 8192 + assert trp["head_chars"] == 4096 + assert trp["tail_chars"] == 1024 + + +def test_agent_init_plumbs_tool_result_prune(monkeypatch, tmp_path): + """End-to-end config seam: compression.tool_result_prune reaches the + built-in compressor through agent_init (mirrors + test_proactive_prune_config.py).""" + import contextlib + import io + + from hermes_cli import config as config_mod + from hermes_state import SessionDB + from run_agent import AIAgent + + compression = { + "enabled": True, + "threshold": 0.50, + "target_ratio": 0.20, + "protect_first_n": 3, + "protect_last_n": 20, + "tool_result_prune": {"enabled": True, "tail_chars": 256}, + } + cfg = { + "compression": compression, + "prompt_caching": {"cache_ttl": "5m"}, + "sessions": {}, + "bedrock": {}, + } + monkeypatch.setattr(config_mod, "load_config", lambda: cfg) + monkeypatch.setattr(config_mod, "load_config_readonly", lambda: cfg) + 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="tool-result-prune-config-test", + ) + cc = agent.context_compressor + assert cc._tool_result_prune_enabled is True + assert cc._tool_result_prune_tail_chars == 256 + assert cc._tool_result_prune_threshold_chars == 8192 + + +# --------------------------------------------------------------------------- +# Method: _prune_tool_results_in_place +# --------------------------------------------------------------------------- + + +def test_in_place_keeps_node_identity_and_tail_protection(): + c = _compressor(tool_result_prune={"enabled": True}) + msgs = [{"role": "system", "content": "sys"}] + for i in range(3): + cid = f"old_{i}" + msgs.append(_assistant_call(cid)) + msgs.append(_tool_msg(cid, chr(65 + i) * 9000)) + # Protected tail: a recent BIG result must stay verbatim. + msgs.append(_assistant_call("recent")) + msgs.append(_tool_msg("recent", "R" * 9000)) + msgs.append({"role": "user", "content": "final"}) + + snapshot = [dict(m) for m in msgs] + result, pruned = c._prune_tool_results_in_place(msgs, protect_tail_count=2) + assert pruned == 3 + assert len(result) == len(msgs) + for cid in ("old_0", "old_1", "old_2"): + m = _tool_by_id(result, cid) + assert m.get("role") == "tool" + assert m.get("tool_call_id") == cid + assert PRUNE_MARKER in m["content"] + assert m["content"].startswith(chr(65 + int(cid[-1])) * 4096) + assert m["content"].endswith(chr(65 + int(cid[-1])) * 1024) + # Recent tail result untouched. + assert _tool_by_id(result, "recent")["content"] == "R" * 9000 + # Non-tool messages untouched. + assert result[0] == snapshot[0] + assert result[1]["content"] == snapshot[1]["content"] + assert result[-1] == snapshot[-1] + # Input never mutated. + assert msgs == snapshot + + +def test_in_place_drops_stale_api_content_sidecar(): + c = _compressor(tool_result_prune={"enabled": True}) + msgs = [ + {"role": "system", "content": "sys"}, + _assistant_call("c1"), + {**_tool_msg("c1", "A" * 9000), "api_content": "pre-rewrite bytes"}, + _assistant_call("c2"), + _tool_msg("c2", "ok"), + ] + result, pruned = c._prune_tool_results_in_place(msgs, protect_tail_count=2) + assert pruned == 1 + pruned_msg = _tool_by_id(result, "c1") + assert "api_content" not in pruned_msg + + +def test_in_place_disabled_noop_contract(): + c = _compressor() # default: disabled + msgs = _build(3) + result, pruned = c._prune_tool_results_in_place(msgs, protect_tail_count=2) + assert pruned == 0 + assert result is msgs + + +def test_in_place_persists_via_archive_and_compact(): + c = _compressor(tool_result_prune={"enabled": True}) + db = _FakeSessionDB() + c.bind_session_state(session_db=db, session_id="s1") + msgs = _build(3, tail_pairs=1) + result, pruned = c._prune_tool_results_in_place( + msgs, protect_tail_count=2, protect_tail_tokens=2000, + ) + assert pruned >= 1 + assert len(db.calls) == 1 + sid, persisted, _ = db.calls[0] + assert sid == "s1" + assert any( + isinstance(m.get("content"), str) and PRUNE_MARKER in m["content"] + for m in persisted + ) + # Persisted rows are stamped so the next append-only flush skips them. + assert any(m.get(_DB_PERSISTED_MARKER) is True for m in result) + + +def test_in_place_persist_failure_rolls_back_to_input(): + c = _compressor(tool_result_prune={"enabled": True}) + db = _FakeSessionDB(raise_on_compact=True) + c.bind_session_state(session_db=db, session_id="s1") + msgs = _build(3, tail_pairs=1) + result, pruned = c._prune_tool_results_in_place( + msgs, protect_tail_count=2, protect_tail_tokens=10_000, + ) + assert pruned == 0 + assert result is msgs # never commit an in-memory prune the DB can't back + + +def test_in_place_store_without_capability_is_noop(): + c = _compressor(tool_result_prune={"enabled": True}) + c.bind_session_state(session_db=object(), session_id="s1") # no archive_and_compact + msgs = _build(3, tail_pairs=1) + result, pruned = c._prune_tool_results_in_place( + msgs, protect_tail_count=2, protect_tail_tokens=10_000, + ) + assert pruned == 0 + assert result is msgs + + +def test_in_place_without_session_db_commits_in_memory(): + # No bound store → no flush exists to desync → the in-memory prune lands. + c = _compressor(tool_result_prune={"enabled": True}) + msgs = _build(3, tail_pairs=1) + result, pruned = c._prune_tool_results_in_place( + msgs, protect_tail_count=2, protect_tail_tokens=2000, + ) + assert pruned >= 1 + assert result is not msgs + + +# --------------------------------------------------------------------------- +# Integration: compress() runs the prune BEFORE region selection/summarization +# --------------------------------------------------------------------------- + + +def test_prune_runs_before_region_selection_and_summarizer(): + c = _compressor(tool_result_prune={"enabled": True}) + c.tail_token_budget = 12_000 + db = _FakeSessionDB() + c.bind_session_state(session_db=db, session_id="s1") + msgs = _build(22) + + region_inputs = [] + orig_tail_cut = c._find_tail_cut_by_tokens + def spy_tail_cut(messages, head_end, token_budget=None): + region_inputs.append(list(messages)) + return orig_tail_cut(messages, head_end, token_budget) + c._find_tail_cut_by_tokens = spy_tail_cut + + summary_inputs = [] + def fake_summary(turns, focus_topic=None, memory_context=""): + summary_inputs.append(list(turns)) + return "the summary" + c._generate_summary = fake_summary + + out = c.compress(msgs, current_tokens=400_000, force=False) + + # Prune landed and persisted before anything else. + assert db.calls, "prune must rewrite the persisted history" + assert any( + isinstance(m.get("content"), str) and PRUNE_MARKER in m["content"] + for m in db.calls[0][1] + ) + # Region selection saw the pruned transcript (prune BEFORE region selection). + assert region_inputs, "region selection must run after the prune" + assert any( + isinstance(m.get("content"), str) and PRUNE_MARKER in m["content"] + for m in region_inputs[0] + ) + # Summarizer received the pruned turns (prune BEFORE summarization). + assert len(summary_inputs) == 1 + assert any( + isinstance(m.get("content"), str) and PRUNE_MARKER in m["content"] + for m in summary_inputs[0] + ) + # Summary present in the compressed output; a pruned row survives in the + # protected tail with its node identity intact. + assert any("the summary" in str(m.get("content")) for m in out) + survivor = next( + m for m in out + if isinstance(m.get("content"), str) and PRUNE_MARKER in m["content"] + ) + assert survivor.get("role") == "tool" + assert survivor.get("tool_call_id", "").startswith("big_") + # No persistence marker may leave compress() (the terminal sweep strips it). + assert all(m.get(_DB_PERSISTED_MARKER) is not True for m in out) + + +def test_prune_can_skip_summarization_via_feasibility_guard(): + """With a prior ineffectiveness strike, a pruned middle below 10% of the + threshold skips the LLM summarizer entirely (deterministic drop).""" + c = _compressor(tool_result_prune={"enabled": True}) + c.tail_token_budget = 12_000 + db = _FakeSessionDB() + c.bind_session_state(session_db=db, session_id="s1") + # bind_session_state resets the strike counter; arm the guard AFTER it. + c._ineffective_compression_count = 1 + msgs = _build(22) + + calls = [] + def fake_summary(turns, focus_topic=None, memory_context=""): + calls.append(1) + return "the summary" + c._generate_summary = fake_summary + + out = c.compress(msgs, current_tokens=400_000, force=False) + + assert calls == [] # summarization avoided + assert c._last_feasibility_skip is True + assert db.calls # the prune itself still persisted + assert any( + isinstance(m.get("content"), str) and PRUNE_MARKER in m["content"] + for m in out + ) + assert len(out) < len(msgs) # deterministic drop still reclaimed + + +def test_prune_then_summarize_without_strike_still_runs(): + """No prior strike → feasibility guard off → summarization still runs, + on the pruned transcript.""" + c = _compressor(tool_result_prune={"enabled": True}) + c.tail_token_budget = 12_000 + c._ineffective_compression_count = 0 + db = _FakeSessionDB() + c.bind_session_state(session_db=db, session_id="s1") + msgs = _build(22) + + calls = [] + def fake_summary(turns, focus_topic=None, memory_context=""): + calls.append(1) + return "the summary" + c._generate_summary = fake_summary + + out = c.compress(msgs, current_tokens=400_000, force=False) + + assert len(calls) == 1 + assert c._last_feasibility_skip is False + assert db.calls + assert any("the summary" in str(m.get("content")) for m in out) + + +def test_default_off_compress_unchanged_and_no_persist(): + """Default-off pin: without tool_result_prune, compress() runs the + historical path — no marker, no prune-originated archive_and_compact.""" + c = _compressor() # nothing configured + c.tail_token_budget = 12_000 + db = _FakeSessionDB() + c.bind_session_state(session_db=db, session_id="s1") + msgs = _build(22) + + calls = [] + def fake_summary(turns, focus_topic=None, memory_context=""): + calls.append(1) + return "the summary" + c._generate_summary = fake_summary + + out = c.compress(msgs, current_tokens=400_000, force=False) + + assert len(calls) == 1 + assert db.calls == [] # the in-place prune never ran → no persist + for m in out: + if isinstance(m.get("content"), str): + assert PRUNE_MARKER not in m["content"]