fix(agent): preserve full command output in temp logs

Return only a 10KB preview to the agent so large command results do not overwhelm conversations while keeping the full output available for follow-up reads. Add pytest to the project dependencies to make the regression tests runnable in the project venv.
This commit is contained in:
jxxghp
2026-05-06 20:04:17 +08:00
parent caf615f3bd
commit c762628217
3 changed files with 144 additions and 69 deletions

View File

@@ -1,5 +1,6 @@
import asyncio
import os
import re
import shlex
import subprocess
import sys
@@ -8,7 +9,7 @@ import unittest
from app.agent.tools.impl.execute_command import (
ExecuteCommandTool,
MAX_OUTPUT_CHARS,
MAX_OUTPUT_PREVIEW_BYTES,
)
@@ -21,6 +22,11 @@ def _python_command(code: str) -> str:
class TestExecuteCommandTool(unittest.TestCase):
def _temp_file_path_from_result(self, result: str) -> str:
match = re.search(r"临时文件: (.+)", result)
self.assertIsNotNone(match)
return match.group(1).strip()
def _run_command(self, command: str, timeout: int = 60) -> str:
tool = ExecuteCommandTool(session_id="session-1", user_id="10001")
return asyncio.run(tool.run(command=command, timeout=timeout))
@@ -31,9 +37,19 @@ class TestExecuteCommandTool(unittest.TestCase):
)
result = self._run_command(command)
temp_file_path = self._temp_file_path_from_result(result)
self.assertIn("输出内容过长,已截断", result)
self.assertLess(len(result), MAX_OUTPUT_CHARS + 500)
self.addCleanup(lambda: os.path.exists(temp_file_path) and os.unlink(temp_file_path))
self.assertIn("命令输出超过 10KB", result)
self.assertIn("仅展示前 10KB 内容", result)
self.assertIn("如需完整内容,请继续读取该文件", result)
self.assertLess(len(result), MAX_OUTPUT_PREVIEW_BYTES + 600)
with open(temp_file_path, encoding="utf-8") as file_handle:
file_content = file_handle.read()
self.assertIn("[标准输出]", file_content)
self.assertGreater(len(file_content), 100000)
def test_timeout_returns_partial_output_promptly(self):
command = _python_command(
@@ -48,6 +64,24 @@ class TestExecuteCommandTool(unittest.TestCase):
self.assertIn("命令执行超时", result)
self.assertIn("started", result)
def test_timeout_with_large_output_writes_partial_full_log_to_temp_file(self):
command = _python_command(
"import sys, time; sys.stdout.write('x' * 20000); sys.stdout.flush(); time.sleep(5)"
)
result = self._run_command(command, timeout=1)
temp_file_path = self._temp_file_path_from_result(result)
self.addCleanup(lambda: os.path.exists(temp_file_path) and os.unlink(temp_file_path))
self.assertIn("命令执行超时", result)
self.assertIn("截至命令终止前的完整输出已写入临时文件", result)
with open(temp_file_path, encoding="utf-8") as file_handle:
file_content = file_handle.read()
self.assertIn("[标准输出]", file_content)
self.assertGreaterEqual(file_content.count("x"), 20000)
def test_timeout_is_capped(self):
command = _python_command("print('ok')")