Fix #2805: cap egg-owned MCP tool output at the tool layer

docs/reference/agent-tools.md

@@ -212,6 +212,26 @@ sized to keep a worst-case page under the SDK's 60 s MCP timeout; if
 you know your dataset is small, raise `limit` to skip the second
 round-trip.
 
+### Output-size cap (`EGG_TOOL_OUTPUT_CAP_BYTES`, #2805)
+
+Every egg-owned tool result is bounded before it crosses the Claude
+Agent SDK's hard 1 MB JSON reader buffer — an oversized result kills the
+agent with exit 255 (#2804). The cap is applied at two chokepoints: the
+orchestrator MCP server (`handle_tool_call` → `cap_result_dict`) and
+every sandbox `@tool` wrapper (`invoke_handler` → `cap_text`), both via
+the shared `shared/egg_tool_output.py` helper.
+
+| Variable | Default | Effect |
+|----------|---------|--------|
+| `EGG_TOOL_OUTPUT_CAP_BYTES` | `102400` (100 KB) | Max serialized size of a single tool result. Output above the cap is replaced with a structured head-preview marker (`_egg_truncated`) that names how to narrow the call, or — for unpaginated content like a full checkpoint transcript — spilled to a temp file (`_egg_output_spilled`) the agent can `Read`/`grep`, with a small inline preview. |
+
+The default leaves ~10× headroom under the 1 MB buffer. A non-positive
+or non-integer value is **ignored with a logged warning** (the operator
+is not left believing a cap is in effect when it isn't); the helper
+falls back to the 100 KB default. The orchestrator measures the cap
+against `indent=2`-serialized JSON (matching what its MCP server ships),
+so raising the cap toward the buffer size stays safe.
+
 ### `cli_command=None` rationale pattern (decision-13)
 
 A `ToolRegistration` declares `cli_command=None` for verbs that have

orchestrator/mcp_tools.py

@@ -49,6 +49,36 @@ def get_logger(name: str, **kwargs) -> logging.Logger:
 except ImportError:
     _SLICE_ID_PATTERN = re.compile(r"^slice-[0-9]+$")
 
+try:
+    from egg_tool_output import cap_result_dict
+except ImportError:  # pragma: no cover - shared path always present in prod
+
+    def cap_result_dict(result, **_kwargs):  # type: ignore[no-redef]
+        return result
+
+
+# Per-tool "how to narrow this call" hints, surfaced in the truncation
+# marker when a result exceeds the tool-output cap (#2805). These tools'
+# output scales with cluster/repo state rather than the caller's params, so
+# they are the at-risk set; anything not listed gets a generic hint and, in
+# practice, never trips the cap because its output is bounded by design.
+_TOOL_NARROW_HINTS: dict[str, str] = {
+    "get_service_logs": (
+        "lower `lines` or set a smaller `since_seconds` window, or fetch a single service at a time"
+    ),
+    "get_container_logs": (
+        "lower `lines` or set a smaller `since_seconds` window, or target a single container"
+    ),
+    "list_containers": (
+        "request a single pipeline/phase if the API supports it; otherwise "
+        "this result is proportional to the running container count"
+    ),
+    "list_tasks": "use `limit` (and any status/phase filter) to page results",
+    "list_checkpoints": "use `limit` + `cursor` pagination or add filters",
+    "search_checkpoints": "use `limit` + `cursor` pagination or a narrower query",
+    "list_agent_local_commits": "target a single agent/branch or use `limit`",
+}
+
 
 logger = get_logger("orchestrator.mcp_tools")
 
@@ -1147,10 +1177,24 @@ def handle_tool_call(self, tool_name: str, arguments: dict[str, Any]) -> dict[st
             return {"error": f"Unknown tool: {tool_name}"}
 
         try:
-            return handler(arguments)
+            result = handler(arguments)
         except Exception as e:
             logger.error("Tool call failed", tool=tool_name, error=str(e))
             return {"error": str(e)}
+        # Bound the result before mcp_server.py serializes it across the
+        # operator's Agent SDK 1 MB reader buffer (#2805). Oversized output
+        # is replaced with a head-preview marker naming how to narrow the
+        # call; bounded tools never trip this. ``indent=2`` matches
+        # mcp_server.py's ``json.dumps(result, indent=2)`` so the cap is
+        # measured against the real on-wire size, not compact JSON.
+        if isinstance(result, dict):
+            return cap_result_dict(
+                result,
+                tool=tool_name,
+                narrow_hint=_TOOL_NARROW_HINTS.get(tool_name),
+                indent=2,
+            )
+        return result
 
     def _make_request(
         self,

orchestrator/tests/test_mcp_tools.py

@@ -3043,3 +3043,39 @@ def test_rebuild_and_rollout_wait_default_false(self):
         tools_by_name = {t["name"]: t for t in PIPELINE_TOOLS}
         props = tools_by_name["rebuild_and_rollout"]["inputSchema"]["properties"]
         assert props["wait"]["default"] is False
+
+
+class TestToolOutputCap:
+    """Output-size caps (#2805): handle_tool_call must bound any result
+    before mcp_server.py serializes it across the operator's 1 MB SDK
+    reader buffer."""
+
+    def test_oversized_result_truncated_to_marker(self, handler):
+        # A list_containers result proportional to a huge cluster.
+        huge = {"containers": [{"id": f"c{i}", "log": "x" * 200} for i in range(20000)]}
+        handler._handle_list_containers = lambda args: huge  # type: ignore[assignment]
+
+        result = handler.handle_tool_call("list_containers", {})
+
+        assert result["_egg_truncated"] is True
+        assert result["tool"] == "list_containers"
+        # Measure against the indent=2 serialization mcp_server.py actually
+        # ships, and hold the marker to the real 100 KB cap (not a loose 2×).
+        assert len(json.dumps(result, indent=2).encode("utf-8")) <= 100 * 1024
+        # The narrow-hint for this tool is surfaced in the marker note.
+        assert "container" in result["note"]
+
+    def test_small_result_passes_through(self, handler):
+        small = {"containers": [{"id": "c1"}]}
+        handler._handle_list_containers = lambda args: small  # type: ignore[assignment]
+
+        result = handler.handle_tool_call("list_containers", {})
+        assert result == small
+
+    def test_error_dict_not_inflated(self, handler):
+        def boom(args):
+            raise RuntimeError("kaboom")
+
+        handler._handle_list_containers = boom  # type: ignore[assignment]
+        result = handler.handle_tool_call("list_containers", {})
+        assert result == {"error": "kaboom"}

sandbox/egg_agent_tools/tools/_common.py

@@ -20,6 +20,8 @@
 from collections.abc import Callable
 from typing import Any
 
+from egg_tool_output import cap_text, spill_to_file
+
 from egg_agent_tools.handlers.errors import GatewayError, HandlerError
 
 # Prefer the structured logger used by shared/egg_agent/client.py so
@@ -53,31 +55,48 @@ def _format_error(exc: BaseException) -> str:
     return f"{type(exc).__name__}: {exc}"
 
 
-def _success_payload(response: dict[str, Any]) -> dict[str, Any]:
+def _success_payload(response: dict[str, Any], tool_name: str | None = None) -> dict[str, Any]:
     try:
         text = json.dumps(response, default=str)
     except Exception:
         text = str(response)
+    # Defense-in-depth (#2805): bound the payload before it crosses the
+    # Agent SDK's 1 MB JSON reader buffer. Oversized output is replaced with
+    # a structured head-preview marker telling the agent how to narrow the
+    # call. Bounded/paginated tools never trip this.
+    text = cap_text(text, tool=tool_name)
     return {"content": [{"type": "text", "text": text}]}
 
 
 def _error_payload(exc: BaseException) -> dict[str, Any]:
+    # Defense-in-depth (#2805): _format_error clamps GatewayError.details but
+    # not message/hint, so a large upstream error body could still cross the
+    # 1 MB SDK reader buffer. Cap the rendered error text too.
     return {
-        "content": [{"type": "text", "text": _format_error(exc)}],
+        "content": [{"type": "text", "text": cap_text(_format_error(exc))}],
         "is_error": True,
     }
 
 
 async def invoke_handler(
     handler: Callable[[dict[str, Any]], dict[str, Any]],
     args: dict[str, Any],
+    *,
+    tool_name: str | None = None,
+    spill: bool = False,
 ) -> dict[str, Any]:
     """Run ``handler(args)`` in a thread and wrap the result for the SDK.
 
     The handler is expected to be synchronous (most of ours are thin
     wrappers around urllib).  If the handler raises, the error is
     serialised into an ``is_error`` SDK tool-result rather than
     propagated — a gateway flake must never crash the agent loop.
+
+    Output is bounded before it crosses the Agent SDK's 1 MB JSON reader
+    buffer (#2805). When ``spill`` is set (large, unpaginated content such
+    as a full checkpoint transcript), an oversized result is written to a
+    file the agent can re-``Read``/``grep`` and replaced with a small
+    preview descriptor; otherwise it is truncated to a head-preview marker.
     """
     try:
         response = await asyncio.to_thread(handler, args or {})
@@ -95,4 +114,16 @@ async def invoke_handler(
             exc,
         )
         return _error_payload(exc)
-    return _success_payload(response)
+    name = tool_name or getattr(handler, "__name__", None)
+    if spill:
+        # Serialize with indent=2 so the spilled file has real line breaks —
+        # the descriptor tells the agent to navigate it with Read's line-based
+        # offset/limit, which is useless against one giant physical line.
+        try:
+            text = json.dumps(response, indent=2, default=str)
+        except Exception:
+            text = str(response)
+        descriptor = spill_to_file(text, tool=name or "tool")
+        if descriptor is not None:
+            return _success_payload(descriptor, name)
+    return _success_payload(response, name)

sandbox/egg_agent_tools/tools/checkpoint.py

@@ -112,7 +112,11 @@ async def checkpoint_list(args: dict[str, Any]) -> dict[str, Any]:
     _SHOW_SCHEMA,
 )
 async def checkpoint_show(args: dict[str, Any]) -> dict[str, Any]:
-    return await invoke_handler(handlers.checkpoint_show, args)
+    # A single checkpoint is a full agent transcript — unpaginated and often
+    # MB-scale, with the relevant detail anywhere in it. Spill an oversized
+    # result to a file the agent can Read/grep rather than truncating the
+    # tail away (#2805).
+    return await invoke_handler(handlers.checkpoint_show, args, spill=True)
 
 
 @tool(