Bremen
/
emma-hermes
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(code-exec): expose truncated stdout metadata

fix/verification-admin-route-recovery
nima20002000 2026-05-31 21:59:24 +03:30 committed by Teknium
parent fce298f700
commit 193871f1a6
4 changed files with 159 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
acp_adapter/tools.py
View File

@ -387,6 +387,24 @@ def _format_execute_code_result(result: Optional[str]) -> Optional[str]:
error = str(data.get("error") or "")
exit_code = data.get("exit_code")
parts = [f"Exit code: {exit_code}" if exit_code is not None else "Execution complete"]
if data.get("stdout_truncated"):
total = data.get("stdout_bytes_total")
captured = data.get("stdout_bytes_captured")
omitted = data.get("stdout_bytes_omitted")
if all(isinstance(v, int) for v in (captured, total, omitted)):
parts.extend([
"",
(
"Output truncated: "
f"captured {captured:,} of {total:,} bytes "
f"({omitted:,} omitted)."
),
])
else:
parts.extend(["", "Output truncated."])
warning = str(data.get("warning") or "").strip()
if warning:
parts.extend(["", "Warning:", warning])
if output:
parts.extend(["", "Output:", output])
if error:

21
tests/acp/test_tools.py
View File

@ -376,6 +376,27 @@ class TestBuildToolComplete:
assert "hello" in text
assert result.raw_output is None
def test_build_tool_complete_for_execute_code_shows_truncation_metadata(self):
result = build_tool_complete(
"tc-code-truncated",
"execute_code",
(
'{"output":"HEAD\\n... [OUTPUT TRUNCATED - 10 bytes omitted out of 60 total] ...\\nTAIL",'
'"exit_code":0,'
'"stdout_truncated":true,'
'"stdout_bytes_captured":50,'
'"stdout_bytes_total":60,'
'"stdout_bytes_omitted":10,'
'"warning":"execute_code stdout was truncated; the script did run."}'
),
)
text = result.content[0].content.text
assert "Exit code: 0" in text
assert "Output truncated: captured 50 of 60 bytes (10 omitted)." in text
assert "Warning:" in text
assert "the script did run" in text
assert result.raw_output is None
def test_build_tool_complete_marks_success_false_as_failed(self):
result = build_tool_complete("tc-fail", "skill_manage", '{"success": false, "error": "boom"}')
assert result.status == "failed"

53
tests/tools/test_code_execution.py
View File

@ -167,6 +167,9 @@ class TestExecuteCodeRemoteTempDir(unittest.TestCase):
result = json.loads(_execute_remote("print('hello')", "task-1", ["terminal"]))
self.assertEqual(result["status"], "success")
self.assertEqual(result["exit_code"], 0)
self.assertFalse(result["stdout_truncated"])
self.assertEqual(result["stdout_bytes_total"], len("hello\n".encode("utf-8")))
mkdir_cmd = env.commands[1][0]
run_cmd = next(cmd for cmd, _, _ in env.commands if "python3 script.py" in cmd)
cleanup_cmd = env.commands[-1][0]
@ -993,9 +996,13 @@ print("TAIL_MARKER_END")
self.assertIn("TAIL_MARKER_END", output)
# Truncation notice should be present
self.assertIn("TRUNCATED", output)
self.assertTrue(result["stdout_truncated"])
self.assertGreater(result["stdout_bytes_total"], result["stdout_bytes_captured"])
self.assertGreater(result["stdout_bytes_omitted"], 0)
self.assertIn("execute_code stdout was truncated", result["warning"])
def test_truncation_notice_format(self):
"""Truncation notice includes character counts."""
"""Truncation notice includes byte counts."""
code = '''
for i in range(15000):
print(f"padding_line_{i:06d}_xxxxxxxxxxxxxxxxxxxxxxxxxx")
@ -1003,9 +1010,51 @@ for i in range(15000):
result = self._run(code)
output = result["output"]
if "TRUNCATED" in output:
self.assertIn("chars omitted", output)
self.assertIn("bytes omitted", output)
self.assertIn("total", output)
def test_short_output_has_explicit_non_truncated_metadata(self):
"""Even non-truncated output exposes unambiguous truncation metadata."""
result = self._run('print("small output")')
self.assertFalse(result["stdout_truncated"])
self.assertEqual(result["stdout_bytes_omitted"], 0)
self.assertEqual(result["stdout_bytes_total"], result["stdout_bytes_captured"])
self.assertEqual(result["exit_code"], 0)
def test_remote_large_output_gets_truncation_metadata(self):
"""Remote backend output capping is explicit in the JSON result."""
class FakeEnv:
def __init__(self):
self.commands = []
def get_temp_dir(self):
return "/tmp"
def execute(self, command, cwd=None, timeout=None):
self.commands.append((command, cwd, timeout))
if "command -v python3" in command:
return {"output": "OK\n"}
if "python3 script.py" in command:
return {"output": "HEAD\n" + ("x" * 80_000) + "\nTAIL\n", "returncode": 0}
return {"output": ""}
fake_thread = MagicMock()
with patch("tools.code_execution_tool._load_config", return_value={"timeout": 30, "max_tool_calls": 5}), \
patch("tools.code_execution_tool._get_or_create_env", return_value=(FakeEnv(), "ssh")), \
patch("tools.code_execution_tool._ship_file_to_remote"), \
patch("tools.code_execution_tool.threading.Thread", return_value=fake_thread):
result = json.loads(_execute_remote("print('large')", "task-1", ["terminal"]))
self.assertEqual(result["status"], "success")
self.assertEqual(result["exit_code"], 0)
self.assertTrue(result["stdout_truncated"])
self.assertIn("HEAD", result["output"])
self.assertIn("TAIL", result["output"])
self.assertGreater(result["stdout_bytes_total"], result["stdout_bytes_captured"])
self.assertGreater(result["stdout_bytes_omitted"], 0)
self.assertIn("execute_code stdout was truncated", result["warning"])
class TestRpcTokenAuthorization(unittest.TestCase):
"""The per-session RPC token must gate socket dispatch (fail-closed).

95
tools/code_execution_tool.py
View File

@ -45,7 +45,7 @@ import time
import uuid
_IS_WINDOWS = platform.system() == "Windows"
from typing import Any, Dict, List, Optional
from typing import Any, Dict, List, Optional, Tuple
from tools.thread_context import propagate_context_to_thread
@ -75,6 +75,63 @@ DEFAULT_MAX_TOOL_CALLS = 50
MAX_STDOUT_BYTES = 50_000 # 50 KB
MAX_STDERR_BYTES = 10_000 # 10 KB
def _assemble_stdout_result(
head: bytes,
tail: bytes = b"",
*,
total_bytes: Optional[int] = None,
) -> Tuple[str, Dict[str, Any]]:
"""Build display stdout plus explicit truncation metadata.
The agent receives execute_code results as JSON. A textual truncation
marker can be missed or later re-truncated by a client layer, so keep the
marker for humans and also expose byte counts for deterministic handling.
"""
captured = head + tail
total = len(captured) if total_bytes is None else max(total_bytes, len(captured))
truncated = total > len(captured)
omitted = max(0, total - len(captured))
if truncated:
stdout_text = (
head.decode("utf-8", errors="replace")
+ f"\n\n... [OUTPUT TRUNCATED - {omitted:,} bytes omitted "
f"out of {total:,} total] ...\n\n"
+ tail.decode("utf-8", errors="replace")
)
else:
stdout_text = captured.decode("utf-8", errors="replace")
metadata: Dict[str, Any] = {
"stdout_truncated": truncated,
"stdout_bytes_captured": len(captured),
"stdout_bytes_total": total,
"stdout_bytes_omitted": omitted,
}
if truncated:
metadata["warning"] = (
"execute_code stdout was truncated; the script did run, but only "
"the captured head/tail output is included. Re-run only with "
"narrower output if the omitted data is required."
)
return stdout_text, metadata
def _truncate_stdout_text(stdout_text: str) -> Tuple[str, Dict[str, Any]]:
"""Cap a complete stdout string by bytes using the same head/tail policy."""
stdout_bytes = stdout_text.encode("utf-8", errors="replace")
if len(stdout_bytes) <= MAX_STDOUT_BYTES:
return _assemble_stdout_result(stdout_bytes)
head_bytes = int(MAX_STDOUT_BYTES * 0.4)
tail_bytes = MAX_STDOUT_BYTES - head_bytes
return _assemble_stdout_result(
stdout_bytes[:head_bytes],
stdout_bytes[-tail_bytes:],
total_bytes=len(stdout_bytes),
)
# Environment variable scrubbing rules (shared between the local + remote
# backends). Secret-substring block is applied first; anything left must
# match a safe prefix, the operational HERMES_ allowlist, or (on Windows) an
@ -1010,7 +1067,7 @@ def _execute_remote(
timeout=timeout,
)
stdout_text = script_result.get("output", "")
stdout_text = script_result.get("output", "") or ""
exit_code = script_result.get("returncode", -1)
status = "success"
@ -1052,19 +1109,7 @@ def _execute_remote(
# --- Post-process output (same as local path) ---
# Truncate stdout to cap
if len(stdout_text) > MAX_STDOUT_BYTES:
head_bytes = int(MAX_STDOUT_BYTES * 0.4)
tail_bytes = MAX_STDOUT_BYTES - head_bytes
head = stdout_text[:head_bytes]
tail = stdout_text[-tail_bytes:]
omitted = len(stdout_text) - len(head) - len(tail)
stdout_text = (
head
+ f"\n\n... [OUTPUT TRUNCATED - {omitted:,} chars omitted "
f"out of {len(stdout_text):,} total] ...\n\n"
+ tail
)
stdout_text, stdout_metadata = _truncate_stdout_text(stdout_text)
# Strip ANSI escape sequences
from tools.ansi_strip import strip_ansi
@ -1080,9 +1125,11 @@ def _execute_remote(
result: Dict[str, Any] = {
"status": status,
"output": stdout_text,
"exit_code": exit_code,
"tool_calls_made": tool_call_counter[0],
"duration_seconds": duration,
}
result.update(stdout_metadata)
if status == "timeout":
timeout_msg = f"Script timed out after {timeout}s and was killed."
@ -1465,21 +1512,13 @@ def execute_code(
stdout_reader.join(timeout=3)
stderr_reader.join(timeout=3)
stdout_head = b"".join(stdout_head_chunks).decode("utf-8", errors="replace")
stdout_tail = b"".join(stdout_tail_chunks).decode("utf-8", errors="replace")
stderr_text = b"".join(stderr_chunks).decode("utf-8", errors="replace")
# Assemble stdout with head+tail truncation
total_stdout = stdout_total_bytes[0]
if total_stdout > MAX_STDOUT_BYTES and stdout_tail:
omitted = total_stdout - len(stdout_head) - len(stdout_tail)
truncated_notice = (
f"\n\n... [OUTPUT TRUNCATED - {omitted:,} chars omitted "
f"out of {total_stdout:,} total] ...\n\n"
stdout_text, stdout_metadata = _assemble_stdout_result(
b"".join(stdout_head_chunks),
b"".join(stdout_tail_chunks),
total_bytes=stdout_total_bytes[0],
)
stdout_text = stdout_head + truncated_notice + stdout_tail
else:
stdout_text = stdout_head + stdout_tail
exit_code = proc.returncode if proc.returncode is not None else -1
duration = round(time.monotonic() - exec_start, 2)
@ -1510,9 +1549,11 @@ def execute_code(
result: Dict[str, Any] = {
"status": status,
"output": stdout_text,
"exit_code": exit_code,
"tool_calls_made": tool_call_counter[0],
"duration_seconds": duration,
}
result.update(stdout_metadata)
if status == "timeout":
timeout_msg = f"Script timed out after {timeout}s and was killed."