mirror/nanobot
1
0
Fork 0
mirror of https://github.com/HKUDS/nanobot.git synced 2026-08-04 16:38:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(agent): structure long tool output summaries

Browse Source
This commit is contained in:
Ubuntu 2026-06-27 11:26:28 +00:00
parent c2b1453b2e
commit abf930a381
5 changed files with 198 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
nanobot/agent/tools/exec_session.py
View File

@ -18,10 +18,12 @@ from nanobot.agent.tools.schema import (
tool_parameters_schema,
)
from nanobot.agent.verification_state import (
VerificationAnalysis,
analyze_verification_result,
append_verification_feedback,
record_verification_observation,
)
from nanobot.utils.helpers import build_structured_output_summary
DEFAULT_YIELD_MS = 1000
MAX_YIELD_MS = 30_000
@ -42,6 +44,7 @@ class _SessionPoll:
terminated: bool = False
stdin_closed: bool = False
truncated_chars: int = 0
analysis: VerificationAnalysis | None = None
@dataclass(slots=True)
@ -152,7 +155,19 @@ class _ExecSession:
output = "".join(self._chunks)
self._chunks.clear()
output, truncated = _truncate_output(output, max_output_chars)
analysis = analyze_verification_result(
command=self.command,
output=output,
exit_code=self.process.returncode,
timed_out=self._timed_out,
)
output, truncated = _truncate_output(
output,
max_output_chars,
analysis=analysis,
exit_code=self.process.returncode,
elapsed_s=max(0.0, time.monotonic() - self.started_at),
)
return _SessionPoll(
output=output,
done=self.process.returncode is not None,
@ -162,6 +177,7 @@ class _ExecSession:
terminated=terminated,
stdin_closed=stdin_closed,
truncated_chars=truncated,
analysis=analysis,
)
async def kill(self) -> None:
@ -325,15 +341,33 @@ def clamp_session_int(value: int | None, default: int, minimum: int, maximum: in
return min(max(value, minimum), maximum)
def _truncate_output(output: str, max_output_chars: int) -> tuple[str, int]:
def _truncate_output(
output: str,
max_output_chars: int,
*,
analysis: VerificationAnalysis | None = None,
exit_code: int | None = None,
elapsed_s: float | None = None,
) -> tuple[str, int]:
if len(output) <= max_output_chars:
return output, 0
half = max_output_chars // 2
omitted = len(output) - max_output_chars
return (
output[:half]
+ f"\n\n... ({omitted:,} chars truncated) ...\n\n"
+ output[-half:],
build_structured_output_summary(
"[tool output truncated]",
output,
max_chars=max_output_chars,
metadata=[
("original_size_chars", len(output)),
("exit_code", exit_code if exit_code is not None else "running"),
("elapsed_s", f"{elapsed_s:.1f}" if elapsed_s is not None else "unknown"),
],
analysis=analysis,
guidance=(
"Use the structured summary first. Poll again for new output "
"or rerun a narrower command instead of reading broad logs."
),
),
omitted,
)
@ -360,7 +394,7 @@ def _format_poll_with_verification(session_id: str, poll: _SessionPoll) -> str:
result = format_session_poll(session_id, poll)
if not poll.done:
return result
analysis = analyze_verification_result(
analysis = poll.analysis or analyze_verification_result(
command="",
output=result,
exit_code=poll.exit_code,

33
nanobot/agent/tools/shell.py
View File

@ -7,6 +7,7 @@ import os
import re
import shutil
import sys
import time
from contextlib import suppress
from dataclasses import dataclass
from pathlib import Path
@ -42,6 +43,7 @@ from nanobot.config.paths import get_media_dir
from nanobot.config_base import Base
from nanobot.security.workspace_access import current_scope_allows_loopback, current_tool_workspace
from nanobot.security.workspace_policy import is_path_within
from nanobot.utils.helpers import build_structured_output_summary
_IS_WINDOWS = sys.platform == "win32"
@ -273,6 +275,7 @@ class ExecTool(Tool):
return await self._execute_session(prepared, yield_time_ms, max_output_chars)
try:
started_at = time.monotonic()
process = await self._spawn(
prepared.command,
prepared.cwd,
@ -314,21 +317,33 @@ class ExecTool(Tool):
output_parts.append(f"\nExit code: {process.returncode}")
result = "\n".join(output_parts) if output_parts else "(no output)"
max_len = clamp_session_int(max_output_chars, self._MAX_OUTPUT, 1000, MAX_OUTPUT_CHARS)
if len(result) > max_len:
half = max_len // 2
result = (
result[:half]
+ f"\n\n... ({len(result) - max_len:,} chars truncated) ...\n\n"
+ result[-half:]
)
elapsed_s = max(0.0, time.monotonic() - started_at)
analysis = analyze_verification_result(
command=prepared.command,
output=result,
exit_code=process.returncode,
)
max_len = clamp_session_int(max_output_chars, self._MAX_OUTPUT, 1000, MAX_OUTPUT_CHARS)
if len(result) > max_len:
result = build_structured_output_summary(
"[tool output truncated]",
result,
max_chars=max_len,
metadata=[
("original_size_chars", len(result)),
("exit_code", process.returncode),
("duration_s", f"{elapsed_s:.1f}"),
],
analysis=analysis,
guidance=(
"Use the structured summary first. Rerun a narrower "
"command, grep a specific failure, or inspect the "
"named artifact instead of rerunning broad noisy logs."
),
)
record_verification_observation(current_request_session_key(), analysis)
return append_verification_feedback(result, analysis)

105
nanobot/utils/helpers.py
View File

@ -405,37 +405,19 @@ def stringify_text_blocks(content: list[dict[str, Any]]) -> str | None:
return "\n".join(parts)
def _render_tool_result_reference(
filepath: Path,
def build_structured_output_summary(
title: str,
text: str,
*,
original_size: int,
head: str,
tail: str | None,
omitted_middle_chars: int,
max_chars: int,
metadata: list[tuple[str, Any]] | None = None,
analysis: Any | None = None,
guidance: str | None = None,
) -> str:
lines = [
"[tool output persisted]",
f"tool_output_id: {filepath.stem}",
f"original_size_chars: {original_size}",
"storage: internal audit artifact",
(
"guidance: Use this head/tail summary first. Avoid reading "
"persisted tool-output files wholesale; rerun a narrower command "
"when more detail is needed."
),
"head:",
head,
]
if tail is not None:
lines.extend([
f"... omitted_middle_chars: {omitted_middle_chars}",
"tail:",
tail,
])
return "\n".join(lines)
"""Return a compact, structured head/tail summary for oversized tool output."""
def _build_tool_result_reference(filepath: Path, text: str, *, max_chars: int) -> str:
if max_chars <= 0:
return text
edge_chars = min(
_TOOL_RESULT_SUMMARY_MAX_EDGE_CHARS,
max(_TOOL_RESULT_SUMMARY_MIN_EDGE_CHARS, max_chars // 3),
@ -448,9 +430,11 @@ def _build_tool_result_reference(filepath: Path, text: str, *, max_chars: int) -
else:
tail = None
omitted_middle_chars = 0
result = _render_tool_result_reference(
filepath,
original_size=len(text),
result = _render_structured_output_summary(
title,
metadata=metadata or [],
guidance=guidance,
analysis=analysis,
head=head,
tail=tail,
omitted_middle_chars=omitted_middle_chars,
@ -464,6 +448,65 @@ def _build_tool_result_reference(filepath: Path, text: str, *, max_chars: int) -
)
def _render_structured_output_summary(
title: str,
*,
metadata: list[tuple[str, Any]],
guidance: str | None,
analysis: Any | None,
head: str,
tail: str | None,
omitted_middle_chars: int,
) -> str:
lines = [title]
lines.extend(f"{key}: {value}" for key, value in metadata)
if omitted_middle_chars:
lines.append(f"truncation: {omitted_middle_chars:,} chars truncated from the middle")
if guidance:
lines.append(f"guidance: {guidance}")
lines.extend(_verification_summary_lines(analysis))
lines.extend(["head:", head])
if tail is not None:
lines.extend(["tail:", tail])
return "\n".join(lines)
def _verification_summary_lines(analysis: Any | None) -> list[str]:
if analysis is None or getattr(analysis, "status", None) != "failed":
return []
lines = ["verification_status: failed"]
if getattr(analysis, "timed_out", False):
lines.append("failure_type: command timeout")
if getattr(analysis, "failed_tests", ()):
lines.append("failed_tests:")
lines.extend(f"- {item}" for item in analysis.failed_tests)
if getattr(analysis, "primary_errors", ()):
lines.append("primary_errors:")
lines.extend(f"- {item}" for item in analysis.primary_errors)
if getattr(analysis, "missing_artifacts", ()):
lines.append("missing_artifacts:")
lines.extend(f"- {item}" for item in analysis.missing_artifacts)
return lines
def _build_tool_result_reference(filepath: Path, text: str, *, max_chars: int) -> str:
return build_structured_output_summary(
"[tool output persisted]",
text,
max_chars=max_chars,
metadata=[
("tool_output_id", filepath.stem),
("original_size_chars", len(text)),
("storage", "internal audit artifact"),
],
guidance=(
"Use this head/tail summary first. Avoid reading persisted "
"tool-output files wholesale; rerun a narrower command when "
"more detail is needed."
),
)
def _bucket_mtime(path: Path) -> float:
try:
return path.stat().st_mtime

54
tests/tools/test_exec_session_tools.py
View File

@ -104,6 +104,31 @@ def test_exec_one_shot_accepts_max_output_tokens_alias(tmp_path):
assert "Exit code: 0" in result
def test_exec_long_output_summary_includes_failure_signals(tmp_path):
async def run() -> str:
tool = ExecTool(working_dir=str(tmp_path), timeout=5)
command = _python_command(
"print('A' * 3000); "
"print('FAILED ../tests/test_outputs.py::test_artifact - AssertionError: missing output'); "
"print(\"FileNotFoundError: [Errno 2] No such file or directory: '/app/out.txt'\"); "
"print('B' * 3000); "
"raise SystemExit(1)"
)
return await tool.execute(command=command, max_output_tokens=2500)
result = asyncio.run(run())
assert "[tool output truncated]" in result
assert "chars truncated" in result
assert "failed_tests:" in result
assert "../tests/test_outputs.py::test_artifact" in result
assert "missing_artifacts:" in result
assert "/app/out.txt" in result
assert "head:" in result
assert "tail:" in result
assert "[Verification Feedback]" in result
def test_exec_adds_verification_feedback_for_test_failures(tmp_path):
async def run() -> str:
tool = ExecTool(working_dir=str(tmp_path), timeout=5)
@ -252,6 +277,35 @@ def test_write_stdin_accepts_max_output_tokens_alias(tmp_path):
assert "Session terminated." in cleanup
def test_write_stdin_long_output_summary_includes_failure_signals(tmp_path):
async def run() -> str:
manager = ExecSessionManager()
exec_tool = ExecTool(working_dir=str(tmp_path), timeout=5, session_manager=manager)
command = _python_command(
"print('A' * 3000); "
"print('FAILED test_outputs.py::test_file - AssertionError: bad'); "
"print(\"FileNotFoundError: [Errno 2] No such file or directory: '/app/missing.txt'\"); "
"print('B' * 3000); "
"raise SystemExit(1)"
)
return await exec_tool.execute(
command=command,
yield_time_ms=1000,
max_output_tokens=2500,
)
result = asyncio.run(run())
assert "[tool output truncated]" in result
assert "chars truncated" in result
assert "failed_tests:" in result
assert "test_outputs.py::test_file" in result
assert "missing_artifacts:" in result
assert "/app/missing.txt" in result
assert "Exit code: 1" in result
assert "[Verification Feedback]" in result
def test_write_stdin_preserves_completed_session_output_until_polled(tmp_path):
async def run() -> tuple[str, str]:
manager = ExecSessionManager()

8
tests/tools/test_tool_validation.py
View File

@ -660,10 +660,12 @@ async def test_exec_head_tail_truncation(tmp_path) -> None:
else:
command = f"{shlex.quote(sys.executable)} {shlex.quote(str(script_file))}"
result = await tool.execute(command=command)
assert "[tool output truncated]" in result
assert "chars truncated" in result
# Head portion should start with As
assert result.startswith("A")
# Tail portion should end with the exit code which comes after Bs
assert "head:" in result
assert "tail:" in result
assert "A" * 80 in result
assert "B" * 80 in result
assert "Exit code:" in result