Limit tool output to last 10k characters

README.md

@@ -10,6 +10,8 @@ conversations can be resumed with context. One example tool is included:
   web content via tools like ``curl`` or run any other commands. The assistant
   must invoke this tool to search online when unsure about a response. Output
   from ``stdout`` and ``stderr`` is captured when each command finishes.
+  The output string is capped at the last 10,000 characters so very long
+  results are truncated. A short notice is prepended whenever data is hidden.
   Execution happens asynchronously so the assistant can continue responding
   while the command runs.
   The VM is created when a chat session starts and reused for all subsequent

src/__init__.py

@@ -1,5 +1,6 @@
 from .chat import ChatSession
 from .tools import execute_terminal, execute_terminal_async, set_vm
+from .utils import limit_chars
 from .vm import LinuxVM
 
 __all__ = [
@@ -8,4 +9,6 @@ __all__ = [
     "execute_terminal_async",
     "set_vm",
     "LinuxVM",
+    "limit_chars",
 ]
+

src/tools.py

@@ -7,6 +7,8 @@ import os
 from typing import Optional
 import asyncio
 
+from .utils import limit_chars
+
 from .vm import LinuxVM
 
 _VM: Optional[LinuxVM] = None
@@ -37,7 +39,8 @@ def execute_terminal(command: str) -> str:
 
     if _VM:
         try:
-            return _VM.execute(command, timeout=None)
+            output = _VM.execute(command, timeout=None)
+            return limit_chars(output)
         except Exception as exc:  # pragma: no cover - unforeseen errors
             return f"Failed to execute command in VM: {exc}"
 
@@ -53,7 +56,7 @@ def execute_terminal(command: str) -> str:
         output = completed.stdout
         if completed.stderr:
             output = f"{output}\n{completed.stderr}" if output else completed.stderr
-        return output.strip()
+        return limit_chars(output)
     except Exception as exc:  # pragma: no cover - unforeseen errors
         return f"Failed to execute command: {exc}"
 
@@ -62,3 +65,4 @@ async def execute_terminal_async(command: str) -> str:
     """Asynchronously execute a shell command."""
     loop = asyncio.get_running_loop()
     return await loop.run_in_executor(None, execute_terminal, command)
+

src/utils.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+__all__ = ["limit_chars"]
+
+
+def limit_chars(text: str, limit: int = 10_000) -> str:
+    """Return at most ``limit`` characters from ``text``.
+
+    Earlier characters are stripped when the text exceeds the limit.
+    A short notice is prepended indicating the amount removed.
+    """
+    text = text.strip()
+    if len(text) <= limit:
+        return text
+
+    truncated = len(text) - limit
+    return f"(output truncated, {truncated} characters hidden)\n{text[-limit:]}"
+

src/vm.py

@@ -9,6 +9,7 @@ from pathlib import Path
 from threading import Lock
 
 from .config import UPLOAD_DIR, VM_IMAGE, PERSIST_VMS, VM_STATE_DIR
+from .utils import limit_chars
 
 from .log import get_logger
 
@@ -147,7 +148,7 @@ class LinuxVM:
         output = completed.stdout
         if completed.stderr:
             output = f"{output}\n{completed.stderr}" if output else completed.stderr
-        return output.strip()
+        return limit_chars(output)
 
     async def execute_async(
         self, command: str, *, timeout: int | None = 3, detach: bool = False