[Bob] feat(context): lossless tool-result tray before first send

agent/agent_init.py

@@ -1873,6 +1873,25 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
                 compression_threshold_tokens = None
         except (TypeError, ValueError):
             compression_threshold_tokens = None
+    # Prune-first phase (issue #513): an independent, LLM-free tool-output
+    # elision that fires on an absolute token budget, decoupled from the
+    # window-relative summarization threshold. Disabled unless
+    # prune_protect_tokens is set to a positive int. Keeps the recent
+    # prune_protect_tokens of tool output verbatim; older tool RESULTS (never
+    # tool calls / user / assistant text) are stubbed. Only acts when it would
+    # reclaim >= prune_minimum_tokens.
+    def _opt_pos_int(_key: str) -> int | None:
+        _raw = _compression_cfg.get(_key)
+        if _raw is None:
+            return None
+        try:
+            _iv = int(_raw)
+        except (TypeError, ValueError):
+            return None
+        return _iv if _iv > 0 else None
+
+    compression_prune_protect_tokens = _opt_pos_int("prune_protect_tokens")
+    compression_prune_minimum_tokens = _opt_pos_int("prune_minimum_tokens")
     # In-place compaction: when True, compress_context() rewrites the message
     # list + rebuilds the system prompt WITHOUT rotating the session id (no
     # parent_session_id chain, no `name #N` renumber). See #38763 and
@@ -2312,6 +2331,8 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
             max_tokens=agent.max_tokens,
             model_thresholds=compression_model_thresholds,
             threshold_tokens_cap=compression_threshold_tokens,
+            prune_protect_tokens=compression_prune_protect_tokens,
+            prune_minimum_tokens=compression_prune_minimum_tokens,
         )
     _bind_session_state = getattr(agent.context_compressor, "bind_session_state", None)
     if callable(_bind_session_state):

agent/context_compressor.py

@@ -1422,6 +1422,25 @@ def update_model(
     # rationale as the gpt-5.5/Codex 85% autoraise.
     _MIN_CTX_TRIGGER_RATIO = 0.85
 
+    # -- Tool-output prune-first phase (issue #513) ------------------------
+    # Large-context models resolve a high summarization threshold
+    # (context_length * threshold_percent). On a 1M-window model that is
+    # ~800K tokens, so a real coding session that plateaus at ~260K of
+    # context NEVER crosses it — and because the cheap tool-output prune
+    # (_prune_old_tool_results) only runs INSIDE compress(), it stays
+    # dormant too. The bulky, re-sent-every-turn tool results (measured at
+    # 58-86% of a long session's re-read weight) are never reclaimed.
+    #
+    # These constants gate an independent prune-only phase that fires on an
+    # ABSOLUTE token budget, decoupled from the summarization trigger. The
+    # LLM-based conversation summary still waits for threshold_tokens; only
+    # the free (no-LLM) tool-result elision runs early. Both default to a
+    # generous protect window so small/normal sessions are untouched.
+    #
+    # Disabled by default (prune_protect_tokens=None) to preserve historical
+    # behavior; opt in via compression.prune_protect_tokens in config.yaml.
+    _DEFAULT_PRUNE_MINIMUM_TOKENS = 20_000  # only prune if >=20K reclaimable
+
     @staticmethod
     def _coerce_max_tokens(value: Any) -> int | None:
         """Normalize a max_tokens value to a positive int or None.
@@ -1545,6 +1564,8 @@ def __init__(
         max_tokens: int | None = None,
         model_thresholds: dict[str, float] | None = None,
         threshold_tokens_cap: Any = None,
+        prune_protect_tokens: int | None = None,
+        prune_minimum_tokens: int | None = None,
     ):
         self.model = model
         self.base_url = base_url
@@ -1589,6 +1610,32 @@ def __init__(
         # deterministic "summary unavailable" handoff and drop the middle window.
         self.abort_on_summary_failure = abort_on_summary_failure
 
+        # -- Prune-first phase (issue #513) --
+        # prune_protect_tokens: keep the most recent N tokens of tool output
+        # untouched; elide older tool results. None disables the early phase
+        # entirely (historical behavior). A positive int arms it.
+        self.prune_protect_tokens = (
+            int(prune_protect_tokens)
+            if isinstance(prune_protect_tokens, (int, float))
+            and prune_protect_tokens
+            and int(prune_protect_tokens) > 0
+            else None
+        )
+        # Only run the early prune when it would reclaim at least this many
+        # tokens (avoids churning the prompt-cache prefix for a trivial gain).
+        _pmin = (
+            prune_minimum_tokens
+            if isinstance(prune_minimum_tokens, (int, float))
+            and prune_minimum_tokens
+            and int(prune_minimum_tokens) > 0
+            else self._DEFAULT_PRUNE_MINIMUM_TOKENS
+        )
+        self.prune_minimum_tokens = int(_pmin)
+        # Rough running estimate of the tokens the last early-prune reclaimed,
+        # surfaced in logs/status. Not authoritative (uses the same rough
+        # estimator as preflight), just observability.
+        self.last_prune_saved_tokens = 0
+
         self.context_length = get_model_context_length(
             model, base_url=base_url, api_key=api_key,
             config_context_length=config_context_length,
@@ -1900,13 +1947,84 @@ def _automatic_compression_blocked_locally(self) -> bool:
             return True
         return False
 
+    def should_prune_tools(self, prompt_tokens: Optional[int] = None) -> bool:
+        """Return True if the early, LLM-free tool-output prune should fire.
+
+        This is INDEPENDENT of :meth:`should_compress`. The summarization
+        trigger scales with the model window (``threshold_tokens`` =
+        ``context_length * threshold_percent``), which on a large-context
+        model can sit at hundreds of thousands of tokens — high enough that a
+        real session's re-sent tool output never triggers it. The prune-first
+        phase instead fires on an ABSOLUTE budget (``prune_protect_tokens``)
+        so bulky, already-seen tool results get elided long before the
+        window-relative summary threshold is reached (issue #513).
+
+        Returns False when the phase is disarmed (``prune_protect_tokens is
+        None``) so historical behavior is byte-for-byte preserved unless the
+        user opts in via ``compression.prune_protect_tokens``.
+        """
+        if self.prune_protect_tokens is None:
+            return False
+        tokens = prompt_tokens if prompt_tokens is not None else self.last_prompt_tokens
+        # Fire once the transcript exceeds the protected window PLUS the
+        # minimum reclaimable budget — below that there is nothing old enough
+        # to prune, or too little to be worth churning the cache prefix.
+        trigger_at = self.prune_protect_tokens + self.prune_minimum_tokens
+        return tokens >= trigger_at
+
+    def prune_tools_only(
+        self, messages: List[Dict[str, Any]],
+    ) -> tuple[List[Dict[str, Any]], int]:
+        """Run the cheap tool-output prune WITHOUT any LLM summarization.
+
+        Reuses :meth:`_prune_old_tool_results` in ``result_only`` mode — the
+        same routine :meth:`compress` runs in its Phase 1 — but protects a
+        token-budget tail sized by ``prune_protect_tokens`` rather than the
+        summary tail budget. No conversation turns are summarized or dropped;
+        only old tool *results* (never tool *calls*, user, or assistant text)
+        are replaced with informative one-line stubs. Tool-call arguments are
+        left byte-for-byte intact (``result_only=True``).
+
+        After pruning, the actual token savings are measured and compared
+        against ``prune_minimum_tokens``. If the measured savings fall below
+        the threshold, the original messages are returned unchanged — this
+        prevents churning the prompt-cache prefix for a trivial gain.
+
+        Returns ``(messages, pruned_count)``. When the phase is disarmed,
+        nothing qualifies, or measured savings are insufficient, returns the
+        input unchanged with ``0``.
+        """
+        if self.prune_protect_tokens is None or not messages:
+            return messages, 0
+        pruned_messages, pruned_count = self._prune_old_tool_results(
+            messages,
+            protect_tail_count=self.protect_last_n,
+            protect_tail_tokens=self.prune_protect_tokens,
+            result_only=True,
+        )
+        if not pruned_count:
+            return messages, 0
+        # Measured savings gate: estimate token delta and only commit
+        # when it meets the prune_minimum_tokens threshold.
+        pre_tokens = estimate_messages_tokens_rough(messages)
+        post_tokens = estimate_messages_tokens_rough(pruned_messages)
+        saved = max(0, pre_tokens - post_tokens)
+        if saved < self.prune_minimum_tokens:
+            logger.debug(
+                "Prune-first: measured savings %d < minimum %d — rollback",
+                saved, self.prune_minimum_tokens,
+            )
+            return messages, 0
+        return pruned_messages, pruned_count
+
     # ------------------------------------------------------------------
     # Tool output pruning (cheap pre-pass, no LLM call)
     # ------------------------------------------------------------------
 
     def _prune_old_tool_results(
         self, messages: List[Dict[str, Any]], protect_tail_count: int,
         protect_tail_tokens: int | None = None,
+        result_only: bool = False,
     ) -> tuple[List[Dict[str, Any]], int]:
         """Replace old tool result contents with informative 1-line summaries.
 
@@ -1916,8 +2034,11 @@ def _prune_old_tool_results(
             [read_file] read config.py from line 1 (3,400 chars)
 
         Also deduplicates identical tool results (e.g. reading the same file
-        5x keeps only the newest full copy) and truncates large tool_call
-        arguments in assistant messages outside the protected tail.
+        5x keeps only the newest full copy) and, unless ``result_only`` is
+        True, truncates large tool_call arguments in assistant messages
+        outside the protected tail. When ``result_only=True``, only tool
+        *result* content blocks are pruned — tool-call arguments, user, and
+        assistant text are left byte-for-byte intact.
 
         Walks backward from the end, protecting the most recent messages that
         fall within ``protect_tail_tokens`` (when provided) OR the last
@@ -2086,14 +2207,20 @@ def _truncate_tool_call_args_at(idx: int) -> bool:
         # outside the protected tail. write_file with 50KB content, for
         # example, survives pruning entirely without this.
         #
+        # Skipped in result_only mode (prune-first phase) — the result-only
+        # contract guarantees tool-call arguments are never mutated.
+        #
         # The shrinking is done inside the parsed JSON structure so the
         # result remains valid JSON — otherwise downstream providers 400
         # on every subsequent turn until the broken call falls out of
         # the window. See ``_truncate_tool_call_args_json`` docstring.
-        for i in range(max(0, prune_boundary)):
-            _truncate_tool_call_args_at(i)
+        if not result_only:
+            for i in range(max(0, prune_boundary)):
+                _truncate_tool_call_args_at(i)
 
-        # Pass 4 (issue #61932): protected-tail pressure demotion.
+        # Pass 4 (issue #61932): protected-tail pressure demotion. This is part
+        # of full compression only; result-only early pruning must not mutate
+        # protected recent results or tool-call arguments.
         # After multiple in-place compactions the transcript can be short
         # enough that nearly every remaining message sits inside the
         # protected floor, yet those messages are huge completed tool /
@@ -2102,7 +2229,12 @@ def _truncate_tool_call_args_at(idx: int) -> bool:
         # bodies *inside* the protected region until the protected tail
         # fits the soft budget, always keeping a short recent floor
         # verbatim so the active ask stays readable.
-        if protect_tail_tokens is not None and protect_tail_tokens > 0 and result:
+        if (
+            not result_only
+            and protect_tail_tokens is not None
+            and protect_tail_tokens > 0
+            and result
+        ):
             soft_ceiling = int(protect_tail_tokens * 1.5)
             keep_recent = min(_PRESSURE_KEEP_RECENT_MESSAGES, len(result))
             demote_end = len(result) - keep_recent

agent/prompt_builder.py

@@ -207,10 +207,10 @@ def _strip_yaml_frontmatter(content: str) -> str:
     "## Lifecycle\n"
     "\n"
     "1. **Orient.** Call `kanban_show()` first (no args — it defaults to your "
-    "task). The response includes title, body, parent-task handoffs (summary + "
-    "metadata), any prior attempts on this task if you're a retry, the full "
-    "comment thread, and a pre-formatted `worker_context` you can treat as "
-    "ground truth.\n"
+    "task). It returns bounded task/history slices with explicit truncation and "
+    "omitted counts, plus attachment metadata and parent "
+    "handoffs. Use `kanban_attachments` for all files or `hermes kanban context "
+    "<task-id>` for canonical context.\n"
     "2. **Work inside the workspace.** `cd $HERMES_KANBAN_WORKSPACE` before "
     "any file operations. The workspace is yours for this run. Don't modify "
     "files outside it unless the task explicitly asks.\n"

agent/tool_executor.py

@@ -48,7 +48,7 @@
     maybe_persist_tool_result,
     enforce_turn_budget,
 )
-from tools.budget_config import BudgetConfig, DEFAULT_BUDGET, budget_for_context_window
+from tools.budget_config import BudgetConfig, DEFAULT_BUDGET, load_budget_config
 
 logger = logging.getLogger(__name__)
 
@@ -86,10 +86,41 @@ def _budget_for_agent(agent) -> BudgetConfig:
     """
     try:
         ctx = getattr(getattr(agent, "context_compressor", None), "context_length", None)
-        return budget_for_context_window(int(ctx)) if ctx else DEFAULT_BUDGET
+        return load_budget_config(int(ctx) if ctx else None)
     except Exception:
         return DEFAULT_BUDGET
 
+
+def _current_turn_tool_messages(messages: list[dict]) -> list[dict]:
+    """Return the contiguous tool-result tail for the active assistant turn."""
+    tail: list[dict] = []
+    for message in reversed(messages):
+        if message.get("role") != "tool":
+            break
+        tail.append(message)
+    tail.reverse()
+    return tail
+
+
+def _allocated_tool_result_budget(
+    messages: list[dict],
+    pending_tool_names: list[str],
+    config: BudgetConfig,
+) -> int | float:
+    """Allocate remaining turn room before append/SessionDB persistence."""
+    used = sum(
+        len(message.get("content", ""))
+        for message in _current_turn_tool_messages(messages)
+        if isinstance(message.get("content"), str)
+    )
+    eligible = [
+        name for name in pending_tool_names
+        if config.resolve_threshold(name) != float("inf")
+    ]
+    if not eligible:
+        return float("inf")
+    return max(0, config.turn_budget - used) // len(eligible)
+
 # Maximum number of concurrent worker threads for parallel tool execution.
 # Mirrors the constant in ``run_agent`` for tests/imports that look here.
 _MAX_TOOL_WORKERS = 8
@@ -979,6 +1010,14 @@ def _run_tool(index, tool_call, function_name, function_args, middleware_trace):
             tool_use_id=tc.id,
             env=get_active_env(effective_task_id),
             config=_tool_budget,
+            threshold=min(
+                _tool_budget.resolve_threshold(name),
+                _allocated_tool_result_budget(
+                    messages,
+                    [pending[1] for pending in parsed_calls[i:]],
+                    _tool_budget,
+                ),
+            ),
         ) if not _is_multimodal_tool_result(function_result) else function_result
 
         subdir_hints = agent._subdirectory_hints.check_tool_call(name, args)
@@ -1675,6 +1714,15 @@ def _execute(next_args: dict) -> Any:
             tool_use_id=tool_call.id,
             env=get_active_env(effective_task_id),
             config=_tool_budget,
+            threshold=min(
+                _tool_budget.resolve_threshold(function_name),
+                _allocated_tool_result_budget(
+                    messages,
+                    [function_name]
+                    + [pending.function.name for pending in assistant_message.tool_calls[i:]],
+                    _tool_budget,
+                ),
+            ),
         ) if not _is_multimodal_tool_result(function_result) else function_result
 
         # Discover subdirectory context files from tool arguments