Bound terminal output to prevent OOM issues

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #64581 fix(terminal): bound foreground output capture during streaming (#64435)

  merged to main

  files: tests/tools/test_base_environment.py, tests/tools/test_local_background_child_hang.py, tests/tools/test_terminal_output_transform_hook.py, tests/tools/test_terminal_task_cwd.py, tools/environments/base.py, tools/environments/modal_utils.py, tools/terminal_tool.py

  Show the diff

  • tests/tools/test_base_environment.py

    @@ -6,7 +6,7 @@
     
     from unittest.mock import MagicMock
     
    -from tools.environments.base import BaseEnvironment
    +from tools.environments.base import BaseEnvironment, _BoundedOutputCollector
     
     
     class _TestableEnv(BaseEnvironment):
    @@ -22,6 +22,41 @@ def cleanup(self):
             pass
     
     
    +class TestBoundedOutputCollector:
    +    def test_large_stream_retains_bounded_head_and_tail(self):
    +        collector = _BoundedOutputCollector(1_000)
    +        collector.append("HEAD-SENTINEL\n")
    +        for _ in range(2_000):
    +            collector.append("x" * 4_096)
    +        collector.append("\nTAIL-SENTINEL")
    +
    +        rendered = collector.render()
    +
    +        assert collector.total_chars > 8_000_000
    +        assert collector.buffered_chars <= 1_000
    +        assert len(rendered) <= 1_000
    +        assert rendered.startswith("HEAD-SENTINEL")
    +        assert rendered.endswith("TAIL-SENTINEL")
    +        assert "[OUTPUT TRUNCATED" in rendered
    +
    +    def test_small_stream_is_unchanged(self):
    +        collector = _BoundedOutputCollector(100)
    +        collector.append("hello ")
    +        collector.append("world")
    +
    +        assert collector.render() == "hello world"
    +
    +    def test_required_status_suffix_stays_inside_limit(self):
    +        collector = _BoundedOutputCollector(120)
    +        collector.append("A" * 10_000)
    +
    +        rendered = collector.render(suffix="\n[Command timed out after 1s]")
    +
    +        assert len(rendered) <= 120
    +        assert rendered.endswith("[Command timed out after 1s]")
    +        assert "[OUTPUT TRUNCATED" in rendered
    +
    +
     class TestWrapCommand:
         def test_basic_shape(self):
             env = _TestableEnv()

  • tests/tools/test_local_background_child_hang.py

    @@ -91,6 +91,69 @@ def test_high_volume_output_complete(self, local_env):
             assert lines[0] == "1"
             assert lines[-1] == "3000"
     
    +    def test_foreground_capture_is_bounded_while_draining(
    +        self, local_env, monkeypatch
    +    ):
    +        monkeypatch.setattr("tools.tool_output_limits.get_max_bytes", lambda: 10_000)
    +        command = (
    +            "python3 -c \"import sys; "
    +            "sys.stdout.write('HEAD-SENTINEL\\n' + 'x' * 2000000 + "
    +            "'\\nTAIL-SENTINEL')\""
    +        )
    +
    +        result = local_env.execute(command, timeout=10, bounded_capture=True)
    +
    +        assert result["returncode"] == 0
    +        assert len(result["output"]) <= 10_000
    +        assert result["output"].startswith("HEAD-SENTINEL")
    +        assert result["output"].endswith("TAIL-SENTINEL")
    +        assert "[OUTPUT TRUNCATED" in result["output"]
    +
    +    def test_default_capture_is_full_fidelity_for_internal_consumers(
    +        self, local_env
    +    ):
    +        """Default execute() (no bounded_capture) must return complete output.
    +
    +        Internal consumers — file-operation ``cat`` reads that feed the patch
    +        engine, code-execution RPC reads, log reads — rely on full-fidelity
    +        capture. Bounding them at tool_output.max_bytes would CORRUPT files
    +        on read-modify-write (#64435 review finding), so only the foreground
    +        terminal tool opts in via bounded_capture=True.
    +        """
    +        # ~200 KB — four times the default 50 KB cap.
    +        command = (
    +            "python3 -c \"import sys; "
    +            "sys.stdout.write('START-MARK\\n' + ('y' * 200000) + '\\nEND-MARK')\""
    +        )
    +
    +        result = local_env.execute(command, timeout=10)
    +
    +        assert result["returncode"] == 0
    +        assert "[OUTPUT TRUNCATED" not in result["output"]
    +        assert result["output"].startswith("START-MARK")
    +        assert result["output"].endswith("END-MARK")
    +        assert len(result["output"]) > 200000
    +
    +    def test_continuous_output_still_honors_foreground_timeout(
    +        self, local_env, monkeypatch
    +    ):
    +        monkeypatch.setattr("tools.tool_output_limits.get_max_bytes", lambda: 5_000)
    +        command = (
    +            "python3 -c \"import sys; "
    +            "chunk = 'x' * 4096; "
    +            "exec('while True: sys.stdout.write(chunk); sys.stdout.flush()')\""
    +        )
    +
    +        started = time.monotonic()
    +        result = local_env.execute(command, timeout=1, bounded_capture=True)
    +        elapsed = time.monotonic() - started
    +
    +        assert elapsed < 4.0
    +        assert result["returncode"] == 124
    +        assert len(result["output"]) <= 5_000
    +        assert "[OUTPUT TRUNCATED" in result["output"]
    +        assert result["output"].endswith("[Command timed out after 1s]")
    +
         def test_timeout_path_still_works(self, local_env):
             """Foreground command exceeding timeout must still be killed."""
             t0 = time.monotonic()

  • tests/tools/test_terminal_output_transform_hook.py

    @@ -5,6 +5,7 @@
     
     import hermes_cli.plugins as plugins_mod
     import tools.terminal_tool as terminal_tool_module
    +from tools.environments.local import LocalEnvironment
     
     
     _UNSET = object()
    @@ -138,6 +139,61 @@ def test_terminal_output_transform_still_runs_strip_and_redact(monkeypatch, tmp_
         assert "abc123def456" not in result["output"]  # secret body is gone
     
     
    +def test_large_process_output_is_bounded_before_sudo_and_plugin_hooks(
    +    monkeypatch, tmp_path
    +):
    +    limit = 10_000
    +    monkeypatch.setattr("tools.tool_output_limits.get_max_bytes", lambda: limit)
    +    monkeypatch.setattr(
    +        terminal_tool_module, "_get_env_config", lambda: _make_env_config(tmp_path)
    +    )
    +    monkeypatch.setattr(terminal_tool_module, "_start_cleanup_thread", lambda: None)
    +    monkeypatch.setattr(
    +        terminal_tool_module,
    +        "_check_all_guards",
    +        lambda *_args, **_kwargs: {"approved": True},
    +    )
    +
    +    sudo_input_lengths = []
    +    hook_inputs = []
    +
    +    def _sudo_spy(output):
    +        sudo_input_lengths.append(len(output))
    +        return False
    +
    +    def _hook_spy(hook_name, **kwargs):
    +        if hook_name == "transform_terminal_output":
    +            hook_inputs.append(kwargs["output"])
    +        return []
    +
    +    monkeypatch.setattr(
    +        terminal_tool_module, "_sudo_wrong_password_failure", _sudo_spy
    +    )
    +    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", _hook_spy)
    +
    +    env = LocalEnvironment(cwd=str(tmp_path), timeout=10)
    +    monkeypatch.setitem(terminal_tool_module._active_environments, "default", env)
    +    monkeypatch.setitem(terminal_tool_module._last_activity, "default", 0.0)
    +    try:
    +        command = (
    +            "python3 -c \"import sys; "
    +            "sys.stdout.write('HEAD-SENTINEL\\n' + 'x' * 2000000 + "
    +            "'\\nTAIL-SENTINEL')\""
    +        )
    +        result = json.loads(terminal_tool_module.terminal_tool(command=command))
    +    finally:
    +        env.cleanup()
    +
    +    assert sudo_input_lengths
    +    assert max(sudo_input_lengths) <= limit
    +    assert len(hook_inputs) == 1
    +    assert len(hook_inputs[0]) <= limit
    +    assert hook_inputs[0].startswith("HEAD-SENTINEL")
    +    assert hook_inputs[0].endswith("TAIL-SENTINEL")
    +    assert "[OUTPUT TRUNCATED" in hook_inputs[0]
    +    assert len(result["output"]) <= limit
    +
    +
     def test_terminal_output_transform_hook_exception_falls_back(monkeypatch, tmp_path):
         def _raise(*_args, **_kwargs):
             raise RuntimeError("boom")

  • tests/tools/test_terminal_task_cwd.py

    @@ -40,7 +40,7 @@ def execute(self, command, **kwargs):
         result = json.loads(terminal_tool.terminal_tool(command="pwd", task_id=task_id))
     
         assert result["exit_code"] == 0
    -    assert calls == [("pwd", {"timeout": 60, "cwd": "/workspace/acp"})]
    +    assert calls == [("pwd", {"timeout": 60, "cwd": "/workspace/acp", "bounded_capture": True})]
     
     
     def test_explicit_workdir_still_wins_over_registered_task_cwd(monkeypatch):
    @@ -73,7 +73,7 @@ def execute(self, command, **kwargs):
         )
     
         assert result["exit_code"] == 0
    -    assert calls == [{"timeout": 60, "cwd": "/explicit/workdir"}]
    +    assert calls == [{"timeout": 60, "cwd": "/explicit/workdir", "bounded_capture": True}]
     
     
     def test_foreground_command_prefers_live_env_cwd_over_init_time_cwd(monkeypatch):
    @@ -104,7 +104,7 @@ def execute(self, command, **kwargs):
         result = json.loads(terminal_tool.terminal_tool(command="pwd", task_id=task_id))
     
         assert result["exit_code"] == 0
    -    assert calls == [("pwd", {"timeout": 60, "cwd": "/workspace/live"})]
    +    assert calls == [("pwd", {"timeout": 60, "cwd": "/workspace/live", "bounded_capture": True})]
     
     
     def test_background_command_prefers_live_env_cwd_over_init_time_cwd(monkeypatch):
    @@ -261,7 +261,7 @@ def execute(self, command, **kwargs):
         assert result["exit_code"] == 0
         # The command must run in the config cwd (hermes-agent), NOT the stale
         # env.cwd left by session A (hermes-desktop-tipc).
    -    assert calls == [("pwd", {"timeout": 60, "cwd": "/home/user/src/hermes-agent"})]
    +    assert calls == [("pwd", {"timeout": 60, "cwd": "/home/user/src/hermes-agent", "bounded_capture": True})]
     
     
     def test_same_session_env_cwd_is_trusted_after_first_claim(monkeypatch):
    @@ -303,7 +303,7 @@ def execute(self, command, **kwargs):
         result = json.loads(terminal_tool.terminal_tool(command="pwd", task_id=task_id))
     
         assert result["exit_code"] == 0
    -    assert calls == [("pwd", {"timeout": 60, "cwd": "/workspace/deep"})]
    +    assert calls == [("pwd", {"timeout": 60, "cwd": "/workspace/deep", "bounded_capture": True})]
     
     
     def test_safe_getcwd_returns_real_cwd(monkeypatch):

  • tools/environments/base.py

    @@ -17,6 +17,7 @@
     import time
     import uuid
     from abc import ABC, abstractmethod
    +from collections import deque
     from pathlib import Path
     from typing import IO, Callable, Protocol
     
    @@ -44,6 +45,105 @@
     _activity_callback_local = threading.local()
     
     
    +# Sentinel capacity for full-fidelity capture (internal consumers). Large
    +# enough that the collector never evicts in practice, keeping a single code
    +# path for both bounded and unbounded modes.
    +_UNBOUNDED_CAPTURE_CHARS = 2**63 - 1
    +
    +
    +class _BoundedOutputCollector:
    +    """Retain a bounded 40/60 head-tail window of streamed text."""
    +    def __init__(self, max_chars: int):
    +        self.max_chars = max(1, int(max_chars))
    +        self._head_limit = int(self.max_chars * 0.4)
    +        self._tail_limit = self.max_chars - self._head_limit
    +        self._head: list[str] = []
    +        self._tail: deque[str] = deque()
    +        self._head_chars = 0
    +        self._tail_chars = 0
    +        self._total_chars = 0
    +        self._lock = threading.Lock()
    +
    +    @property
    +    def buffered_chars(self) -> int:
    +        with self._lock:
    +            return self._head_chars + self._tail_chars
    +
    +    @property
    +    def total_chars(self) -> int:
    +        with self._lock:
    +            return self._total_chars
    +
    +    def append(self, text: str) -> None:
    +        if not text:
    +            return
    +        with self._lock:
    +            text_len = len(text)
    +            self._total_chars += text_len
    +            start = 0
    +
    +            if self._head_chars < self._head_limit:
    +                take = min(self._head_limit - self._head_chars, text_len)
    +                if take:
    +                    self._head.append(text[:take])
    +                    self._head_chars += take
    +                    start = take
    +
    +            remaining = text_len - start
    +            if remaining <= 0 or self._tail_limit <= 0:
    +                return
    +            if remaining >= self._tail_limit:
    +                self._tail.clear()
    +                self._tail.append(text[-self._tail_limit :])
    +                self._tail_chars = self._tail_limit
    +                return
    +
    +            chunk = text[start:]
    +            self._tail.append(chunk)
    +            self._tail_chars += len(chunk)
    +            while self._tail_chars > self._tail_limit:
    +                excess = self._tail_chars - self._tail_limit
    +                first = self._tail[0]
    +                if len(first) <= excess:
    +                    self._tail.popleft()
    +                    self._tail_chars -= len(first)
    +                else:
    +                    self._tail[0] = first[excess:]
    +                    self._tail_chars -= excess
    +
    +    def render(self, *, suffix: str = "") -> str:
    +        """Render within ``max_chars``, preserving a required status suffix."""
    +        with self._lock:
    +            if len(suffix) >= self.max_chars:
    +                return suffix[-self.max_chars :]
    +
    +            head = "".join(self._head)
    +            tail = "".join(self._tail)
    +            available = self.max_chars - len(suffix)
    +            if self._total_chars <= available:
    +                return head + tail + suffix
    +
    +            notice = ""
    +            for _ in range(4):
    +                content_budget = max(0, available - len(notice))
    +                head_chars = int(content_budget * 0.4)
    +                tail_chars = content_budget - head_chars
    +                omitted = max(0, self._total_chars - head_chars - tail_chars)
    +                updated = (
    +                    f"\n\n... [OUTPUT TRUNCATED - {omitted:,} chars omitted "
    +                    f"out of {self._total_chars:,} total] ...\n\n"
    +                )
    +                if updated == notice:
    +                    break
    +                notice = updated
    +
    +            content_budget = max(0, available - len(notice))
    +            head_chars = int(content_budget * 0.4)
    +            tail_chars = content_budget - head_chars
    +            rendered_tail = tail[-tail_chars:] if tail_chars else ""
    +            return head[:head_chars] + notice[:available] + rendered_tail + suffix
    +
    +
     def set_activity_callback(cb: Callable[[str], None] | None) -> None:
         """Register a callback that _wait_for_process fires periodically."""
         _activity_callback_local.callback = cb
    @@ -578,11 +678,21 @@ def _embed_stdin_heredoc(command: str, stdin_data: str) -> str:
         # Process lifecycle
         # ------------------------------------------------------------------
     
    -    def _wait_for_process(self, proc: ProcessHandle, timeout: int = 120) -> dict:
    +    def _wait_for_process(
    +        self, proc: ProcessHandle, timeout: int = 120, *, bounded_capture: bool = False
    +    ) -> dict:
             """Poll-based wait with interrupt checking and stdout draining.
     
             Shared across all backends — not overridden.
     
    +        ``bounded_capture=True`` (foreground terminal-tool path only) retains
    +        at most ``tool_output.max_bytes`` of output in a head/tail window
    +        while draining, so a verbose subprocess cannot OOM the process
    +        (#64435). The default (False) preserves full-fidelity capture for
    +        internal consumers — file-operation ``cat`` reads feeding the patch
    +        engine, code-execution RPC reads, log reads — where truncation would
    +        corrupt data.
    +
             Fires the ``activity_callback`` (if set on this instance) every 10s
             while the process is running so the gateway's inactivity timeout
             doesn't kill long-running commands.
    @@ -594,7 +704,19 @@ def _wait_for_process(self, proc: ProcessHandle, timeout: int = 120) -> dict:
             an orphan with ``PPID=1`` when python is shut down mid-tool — the
             ``sleep 300``-survives-30-min bug Physikal and I both hit.
             """
    -        output_chunks: list[str] = []
    +        if bounded_capture:
    +            try:
    +                from tools.tool_output_limits import get_max_bytes
    +
    +                capture_limit = get_max_bytes()
    +            except Exception:
    +                capture_limit = 50_000
    +        else:
    +            # Full fidelity: effectively unbounded collector (single head
    +            # segment, no eviction) so behavior matches the historical
    +            # accumulate-everything semantics.
    +            capture_limit = _UNBOUNDED_CAPTURE_CHARS
    +        output = _BoundedOutputCollector(capture_limit)
     
             # Non-blocking drain via select().
             #
    @@ -635,16 +757,16 @@ def _drain_iterable(stream):
                         if piece is None:
                             continue
                         if isinstance(piece, bytes):
    -                        output_chunks.append(decoder.decode(piece))
    +                        output.append(decoder.decode(piece))
                         else:
    -                        output_chunks.append(str(piece))
    +                        output.append(str(piece))
                 except Exception:
                     pass
                 finally:
                     try:
                         tail = decoder.decode(b"", final=True)
                         if tail:
    -                        output_chunks.append(tail)
    +                        output.append(tail)
                     except Exception:
                         pass
     
    @@ -675,14 +797,14 @@ def _drain():
                             chunk = os.read(fd, 4096)
                             if not chunk:
                                 break
    -                        output_chunks.append(decoder.decode(chunk))
    +                        output.append(decoder.decode(chunk))
                     except (ValueError, OSError):
                         pass
                     finally:
                         try:
                             tail = decoder.decode(b"", final=True)
                             if tail:
    -                            output_chunks.append(tail)
    +                            output.append(tail)
                         except Exception:
                             pass
                     return
    @@ -700,7 +822,7 @@ def _drain():
                                 break
                             if not chunk:
                                 break  # true EOF — all writers closed
    -                        output_chunks.append(decoder.decode(chunk))
    +                        output.append(decoder.decode(chunk))
                             idle_after_exit = 0
                         elif proc.poll() is not None:
                             # bash is gone and the pipe was idle for ~100ms.  Give
    @@ -716,7 +838,7 @@ def _drain():
                     try:
                         tail = decoder.decode(b"", final=True)
                         if tail:
    -                        output_chunks.append(tail)
    +                        output.append(tail)
                     except Exception:
                         pass
     
    @@ -762,7 +884,7 @@ def _drain():
                         self._kill_process(proc)
                         drain_thread.join(timeout=2)
                         return {
    -                        "output": "".join(output_chunks) + "\n[Command interrupted]",
    +                        "output": output.render(suffix="\n[Command interrupted]"),
                             "returncode": 130,
                         }
                     if time.monotonic() > deadline:
    @@ -774,12 +896,11 @@ def _drain():
                             )
                         self._kill_process(proc)
                         drain_thread.join(timeout=2)
    -                    partial = "".join(output_chunks)
                         timeout_msg = f"\n[Command timed out after {timeout}s]"
                         return {
    -                        "output": partial + timeout_msg
    -                        if partial
    -                        else timeout_msg.lstrip(),
    +                        "output": output.render(suffix=timeout_msg).lstrip()
    +                        if output.total_chars == 0
    +                        else output.render(suffix=timeout_msg),
                             "returncode": 124,
                         }
                     # Periodic activity touch so the gateway knows we're alive
    @@ -855,7 +976,7 @@ def _drain():
                     proc.returncode,
                 )
     
    -        return {"output": "".join(output_chunks), "returncode": proc.returncode}
    +        return {"output": output.render(), "returncode": proc.returncode}
     
         def _kill_process(self, proc: ProcessHandle):
             """Terminate a process. Subclasses may override for process-group kill."""
    @@ -932,8 +1053,19 @@ def execute(
             timeout: int | None = None,
             stdin_data: str | None = None,
             rewrite_compound_background: bool = True,
    +        bounded_capture: bool = False,
         ) -> dict:
    -        """Execute a command, return {"output": str, "returncode": int}."""
    +        """Execute a command, return {"output": str, "returncode": int}.
    +
    +        ``bounded_capture=True`` caps stdout/stderr retention at
    +        ``tool_output.max_bytes`` WHILE the stream is drained (head/tail
    +        window) instead of holding the full output in memory (#64435).
    +        It must only be set by callers whose output is destined for the
    +        model/tool payload (the foreground terminal tool). Internal
    +        full-fidelity consumers — file operations ``cat`` reads that feed
    +        the patch engine, code-execution RPC reads, log reads — MUST leave
    +        it False: truncating those corrupts data, not just display.
    +        """
             self._before_execute()
     
             exec_command, sudo_stdin = self._prepare_command(command)
    @@ -968,7 +1100,9 @@ def execute(
             proc = self._run_bash(
                 wrapped, login=login, timeout=effective_timeout, stdin_data=effective_stdin
             )
    -        result = self._wait_for_process(proc, timeout=effective_timeout)
    +        result = self._wait_for_process(
    +            proc, timeout=effective_timeout, bounded_capture=bounded_capture
    +        )
             self._update_cwd(result)
     
             return result

  • tools/environments/modal_utils.py

    @@ -80,11 +80,17 @@ def execute(
             timeout: int | None = None,
             stdin_data: str | None = None,
             rewrite_compound_background: bool = True,
    +        bounded_capture: bool = False,
         ) -> dict:
             # Managed/remote modal transports execute commands via explicit transport
             # and do not rely on shell background rewriters. Keep parameter for
             # compatibility with BaseEnvironment callers.
             _ = rewrite_compound_background
    +        # bounded_capture: accepted for BaseEnvironment.execute() signature
    +        # parity (the terminal tool passes it). Modal transports return the
    +        # remote function's result in one payload, so streaming-time bounding
    +        # does not apply; the terminal tool's final truncation still caps it.
    +        _ = bounded_capture
             self._before_execute()
             prepared = self._prepare_modal_exec(
                 command,

  • tools/terminal_tool.py

    @@ -2657,6 +2657,12 @@ def terminal_tool(
                         execute_kwargs = {
                             "timeout": effective_timeout,
                             "cwd": command_cwd,
    +                        # Foreground model-facing output: cap retention while
    +                        # streaming (head/tail window) so a verbose command
    +                        # can't OOM the gateway before truncation (#64435).
    +                        # Internal env.execute() consumers (file ops cat
    +                        # reads, RPC reads) intentionally stay unbounded.
    +                        "bounded_capture": True,
                         }
                         result = env.execute(command, **execute_kwargs)
                     except Exception as e:
    @@ -2708,9 +2714,10 @@ def terminal_tool(
                             "command."
                         )
     
    -            # Foreground terminal output canonicalization seam: plugins receive
    -            # the full output string before default truncation and may only
    -            # replace it by returning a string from transform_terminal_output.
    +            # Foreground terminal output canonicalization seam: process capture
    +            # is already bounded by BaseEnvironment before sudo checks and hooks
    +            # run. Plugins may replace that bounded string; replacements are
    +            # still subject to the final output limit below.
                 # The hook is fail-open, and the first valid string return wins.
                 try:
                     from hermes_cli.plugins import invoke_hook