fix(agent): offload oversized tool output at the chat seam, where the model actually sees it (#13997)

autobot-backend/agent_loop/tool_output_spill.py

@@ -34,7 +34,7 @@
 import os
 from contextvars import ContextVar
 from pathlib import Path
-from typing import Any, Dict, Tuple
+from typing import Any, Dict, List, Tuple
 
 from autobot_shared.logging_manager import get_logger
 
@@ -389,6 +389,90 @@ def spill_results(task_id: str, tool_results: Dict[str, Any]) -> Tuple[Dict[str,
     return rewritten, spilled_count
 
 
+#: Fields a chat execution result may carry its payload in.
+#:
+#: `output` is what the builtin handlers write. **`stdout` is what shell results
+#: carry** — `_create_execution_result` emits
+#: `{command, host, stdout, stderr, return_code, status, approved}` with no
+#: `output`, no `result` and no `tool` key at all. Omitting it meant
+#: `execute_command`, the single likeliest source of a window-consuming result,
+#: was the one thing this never offloaded (#13997).
+_EXECUTION_RESULT_TEXT_KEYS = ("output", "result", "stdout", "stderr")
+
+
+def spill_execution_results(task_id: str, results: List[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], int]:
+    """Spill oversized output in a chat *execution-results list* (#13997).
+
+    The agent loop passes a ``{name: result}`` mapping; the chat seam collects a
+    **list of envelopes** — ``{"tool", "status", "output"}`` — and extends
+    ``execution_history`` with it, which is what the model then sees. So this is
+    the shape that needs offloading on the live path.
+
+    Only the payload field is replaced, never the envelope: ``status`` decides
+    error handling downstream (`tool_handler` counts ``status == "success"``,
+    delegation checks ``== "error"``), and swapping the whole dict for an
+    excerpt would drop it. The field stays a **str** because
+    ``_as_output_text`` and ``tool_handler`` read it as one.
+
+    Reuses :func:`spill_if_oversized`, so the measured
+    smaller-than-what-it-replaced guard, the artifact cap and the flag all apply
+    unchanged — this adds a shape adapter, not a second mechanism.
+    """
+    if not SPILL_ENABLED or not results:
+        return results, 0
+
+    rewritten: List[Dict[str, Any]] = []
+    spilled_count = 0
+    for entry in results:
+        if not isinstance(entry, dict):
+            rewritten.append(entry)
+            continue
+        keys = [k for k in _EXECUTION_RESULT_TEXT_KEYS if isinstance(entry.get(k), str) and entry[k]]
+        if not keys:
+            rewritten.append(entry)
+            continue
+
+        # Shell results carry no `tool` key; `command` is their identity. Without
+        # this every spilled shell result would bucket under the literal "tool",
+        # making anchors collide across unrelated commands in one session.
+        tool_name = str(entry.get("tool") or entry.get("command") or "tool")
+        updated = dict(entry)
+        anchors: List[str] = []
+        for key in keys:
+            value, was_spilled = spill_if_oversized(task_id, tool_name, entry[key])
+            if not was_spilled:
+                continue
+            # The note carries the anchor and names the read tool, so it must
+            # reach the model with the excerpt — it is the only place the
+            # capability is advertised on this path.
+            updated[key] = f"{value['excerpt']}\n\n{value['note']}"
+            anchors.append(value["anchor"])
+
+        if not anchors:
+            rewritten.append(entry)
+            continue
+
+        # A shell result can have both stdout and stderr oversized, so this is a
+        # list rather than a single key.
+        updated["anchors"] = anchors
+        rewritten.append(updated)
+        spilled_count += 1
+    return rewritten, spilled_count
+
+
+async def spill_execution_results_async(
+    task_id: str, results: List[Dict[str, Any]]
+) -> Tuple[List[Dict[str, Any]], int]:
+    """:func:`spill_execution_results` off the event loop.
+
+    Same reason as :func:`spill_results_async`: the write is large by definition
+    and the chat seam is async.
+    """
+    if not SPILL_ENABLED or not results:
+        return results, 0
+    return await asyncio.to_thread(spill_execution_results, task_id, results)
+
+
 async def spill_results_async(task_id: str, tool_results: Dict[str, Any]) -> Tuple[Dict[str, Any], int]:
     """:func:`spill_results` off the event loop.
 
@@ -413,6 +497,8 @@ async def spill_results_async(task_id: str, tool_results: Dict[str, Any]) -> Tup
     "sweepable_spill_root",
     "read_spilled_window",
     "spill_if_oversized",
+    "spill_execution_results",
+    "spill_execution_results_async",
     "spill_results",
     "spill_results_async",
 ]

autobot-backend/chat_workflow/manager.py

@@ -2296,7 +2296,88 @@ def _validate_tool_message(self, tool_msg: Any) -> bool:
 
         return True
 
-    def _handle_execution_summary(
+    def _bind_spill_run(self, session_id: str) -> None:
+        """Declare which conversation the current Task is serving (#13997).
+
+        Never fatal: if the agent-loop package cannot be imported, the spill
+        simply does not fire and output passes through whole.
+        """
+        try:
+            from agent_loop.tool_output_spill import bind_task
+
+            bind_task(session_id)
+        except Exception:  # pragma: no cover - the turn must not fail over this
+            logger.warning("Could not bind the tool-output spill run", exc_info=True)
+
+    def _spill_run_id(self) -> str | None:
+        """The conversation this Task is bound to, or None when unbound.
+
+        Read rather than passed: `_handle_execution_summary` has no session_id in
+        scope, and threading one through four call layers to reach it would be a
+        worse coupling than a ContextVar the dispatch already relies on.
+        """
+        try:
+            from agent_loop.tool_output_spill import current_task_id
+
+            return current_task_id()
+        except Exception:  # pragma: no cover
+            return None
+
+    async def _offload_oversized_output(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """Write oversized tool output aside before it enters the model's context (#13997).
+
+        #13692 built this and wired it only into ``AgentLoop``, which is
+        documented as never instantiated in production — so on the live chat path
+        it never fired, and its re-read tool could never resolve an anchor. This
+        is the seam that actually assembles what the model sees.
+
+        Worth knowing what it replaces: the chat path has **no general cap** on
+        tool output today. ``execute_command`` — the likeliest source of a
+        window-consuming result — is unbounded, and the per-tool caps that do
+        exist (``web_search`` at 3000 chars, subagent results at 1000) discard
+        the remainder permanently. The spill is non-lossy: the excerpt carries an
+        anchor and the full output stays retrievable.
+
+        Off by default and non-fatal: with the flag unset this returns *results*
+        unchanged, so a turn is byte-identical to one without this call.
+        """
+        try:
+            from agent_loop.tool_output_spill import _EXECUTION_RESULT_TEXT_KEYS as _SPILL_PAYLOAD_KEYS
+            from agent_loop.tool_output_spill import (
+                spill_execution_results_async,
+            )
+
+            run_id = self._spill_run_id()
+            if not run_id:
+                # No run bound: an artifact written now could never be re-read,
+                # so writing it would be pure cost. The agent loop makes the
+                # same call for the same reason.
+                return results
+            rewritten, spilled = await spill_execution_results_async(run_id, results)
+        except Exception:
+            # Losing the offload is always better than losing the observation.
+            logger.warning("Tool-output offload failed; keeping full output", exc_info=True)
+            return results
+
+        if spilled:
+            # Summed over the SAME keys the adapter offloads, not just `output`.
+            # Reading only `output` reported 0 -> 0 for every shell result, since
+            # those carry `stdout`/`stderr` — so the one measurement this change
+            # exists to produce was blind to its most common case.
+            def _payload_chars(entries: List[Dict[str, Any]]) -> int:
+                return sum(len(v) for e in entries for k in _SPILL_PAYLOAD_KEYS if isinstance((v := e.get(k)), str))
+
+            before = _payload_chars(results)
+            after = _payload_chars(rewritten)
+            logger.info(
+                "Offloaded %d oversized tool result(s): %d -> %d chars (#13997)",
+                spilled,
+                before,
+                after,
+            )
+        return rewritten
+
+    async def _handle_execution_summary(
         self,
         tool_msg: WorkflowMessage,
         new_execution_results: List[Dict[str, Any]],
@@ -2311,6 +2392,7 @@ def _handle_execution_summary(
         """
         if tool_msg.type == "execution_summary":
             new_results = tool_msg.metadata.get("execution_results", [])
+            new_results = await self._offload_oversized_output(new_results)
             new_execution_results.extend(new_results)
             execution_history.extend(new_results)
             logger.info(
@@ -2400,7 +2482,7 @@ async def _process_tool_results(
             if not self._validate_tool_message(tool_msg):
                 continue
 
-            if self._handle_execution_summary(tool_msg, new_execution_results, execution_history):
+            if await self._handle_execution_summary(tool_msg, new_execution_results, execution_history):
                 continue
 
             if tool_msg.metadata is not None:
@@ -2806,6 +2888,13 @@ async def _run_continuation_loop_iteration(
         # MVA-1993 / #11216 / #11612: store lightweight_mode_used in a task-local
         # ContextVar (not on the shared singleton) for the response-metadata badge.
         _lw_token = _current_lightweight_mode.set(ctx.context.get("lightweight_mode_used", False))
+        # #13997: bind the spill run at the SAME seam, and for the same reason
+        # #11612 gives above — the LangGraph path (the production default) calls
+        # this method directly and bypasses `_run_llm_iterations` entirely. A
+        # bind in that outer wrapper never runs in production, which is how the
+        # cost badge came to always read False. Both are ContextVars, so both
+        # must be set here or not at all.
+        self._bind_spill_run(ctx.session_id)
         try:
             llm_response = None
             should_continue = False