fix(terminal): bound foreground output capture

tests/tools/test_base_environment.py

@@ -6,7 +6,7 @@ init_session() failure handling, and the CWD marker contract.
 
 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 @@ class _TestableEnv(BaseEnvironment):
         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,44 @@ class TestBackgroundChildDoesNotHang:
         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)
+
+        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_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)
+        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 @@ from unittest.mock import MagicMock
 
 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")

tools/environments/base.py

@@ -17,6 +17,7 @@ import threading
 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,100 @@ if _DEBUG_INTERRUPT:
 _activity_callback_local = threading.local()
 
 
+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
@@ -594,7 +689,13 @@ class BaseEnvironment(ABC):
         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] = []
+        try:
+            from tools.tool_output_limits import get_max_bytes
+
+            capture_limit = get_max_bytes()
+        except Exception:
+            capture_limit = 50_000
+        output = _BoundedOutputCollector(capture_limit)
 
         # Non-blocking drain via select().
         #
@@ -635,16 +736,16 @@ class BaseEnvironment(ABC):
                     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 +776,14 @@ class BaseEnvironment(ABC):
                         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 +801,7 @@ class BaseEnvironment(ABC):
                             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 +817,7 @@ class BaseEnvironment(ABC):
                 try:
                     tail = decoder.decode(b"", final=True)
                     if tail:
-                        output_chunks.append(tail)
+                        output.append(tail)
                 except Exception:
                     pass
 
@@ -762,7 +863,7 @@ class BaseEnvironment(ABC):
                     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 +875,11 @@ class BaseEnvironment(ABC):
                         )
                     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 +955,7 @@ class BaseEnvironment(ABC):
                 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."""

tools/terminal_tool.py

@@ -2708,9 +2708,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