diff --git a/src/agent_loop.py b/src/agent_loop.py index 052d92c49..4843f28a1 100644 --- a/src/agent_loop.py +++ b/src/agent_loop.py @@ -21,7 +21,7 @@ from src.settings import get_setting from src.prompt_security import untrusted_context_message from src.tool_security import blocked_tools_for_owner, plan_mode_disabled_tools from src.tool_policy import GUIDE_ONLY_DIRECTIVE, ToolPolicy -from src.tool_utils import get_mcp_manager +from src.tool_utils import _truncate, get_mcp_manager from src.agent_tools import ( parse_tool_blocks, strip_tool_blocks, @@ -2751,18 +2751,20 @@ async def stream_agent_loop( # On a bash/python timeout the result carries error + (often # empty) stdout/stderr; fall back to the error so the "timed # out" reason reaches the UI instead of a blank result. - output_text = (result["stdout"] or result["stderr"] or result.get("error", ""))[:2000] + raw = result["stdout"] or result["stderr"] or result.get("error", "") + output_text = _truncate(raw) elif "output" in result: # bash / python canonical result: {"output": ..., "exit_code": ...} - output_text = (result["output"] or "")[:2000] + raw = result["output"] or "" + output_text = _truncate(raw) elif "response" in result: # AI interaction tools (chat_with_model, send_to_session) label = result.get("model", result.get("session_name", "AI")) - output_text = f"{label}: {result['response']}"[:4000] + output_text = _truncate(f"{label}: {result['response']}") elif "content" in result: - output_text = result["content"][:2000] + output_text = _truncate(result["content"]) elif "results" in result: - output_text = result["results"][:4000] + output_text = _truncate(result["results"]) elif "session_id" in result and "name" in result: output_text = f"Session created: {result['name']} (id: {result['session_id']})" elif "success" in result: @@ -2772,7 +2774,7 @@ async def stream_agent_loop( else f"Error: {result.get('error', '')}" ) elif "error" in result: - output_text = result["error"][:2000] + output_text = _truncate(result["error"]) # Emit tool_output (include ui_event data if present) tool_output_data = {"type": "tool_output", "tool": block.tool_type, "command": cmd_display, "output": output_text, "exit_code": result.get("exit_code")} diff --git a/tests/test_agent_loop_tool_output_truncation.py b/tests/test_agent_loop_tool_output_truncation.py new file mode 100644 index 000000000..35e33e88f --- /dev/null +++ b/tests/test_agent_loop_tool_output_truncation.py @@ -0,0 +1,43 @@ +"""Tool-output display truncation uses _truncate with an indicator. + +Previously agent_loop sliced tool output to a hard character limit ([:2000] +or [:4000]) with no signal to the UI that data was lost. Now it delegates to +tool_utils._truncate which caps at MAX_OUTPUT_CHARS (10 000) and appends +a ``... (truncated, N chars total)`` suffix so the frontend can show a +truncation indicator in the tool bubble. +""" +from src.tool_utils import _truncate, MAX_OUTPUT_CHARS + + +def test_short_output_unchanged(): + """Outputs within the limit pass through verbatim.""" + text = "hello world" + assert _truncate(text) == text + + +def test_long_output_truncated_with_indicator(): + """Outputs exceeding MAX_OUTPUT_CHARS are truncated with a suffix.""" + text = "x" * (MAX_OUTPUT_CHARS + 500) + result = _truncate(text) + assert len(result) > MAX_OUTPUT_CHARS # includes suffix + assert result.startswith("x" * MAX_OUTPUT_CHARS) + assert "truncated" in result + assert str(len(text)) in result # original length reported + + +def test_exact_limit_unchanged(): + """An output exactly at the limit is not truncated.""" + text = "a" * MAX_OUTPUT_CHARS + assert _truncate(text) == text + + +def test_default_limit_matches_constant(): + """_truncate default limit equals MAX_OUTPUT_CHARS (10 000).""" + assert MAX_OUTPUT_CHARS == 10_000 + text = "y" * 10_001 + result = _truncate(text) + assert "truncated" in result + + +def test_empty_string(): + assert _truncate("") == ""