From 259dd87c43d3f28f9b16d75a821f1b4460a21aa5 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Sat, 15 Aug 2026 18:08:28 +0800 Subject: [PATCH] cli: pretty printed summaries for run_argv and run_argv_batch --- src/plyngent/cli/display.py | 80 ++++++++++++++++++++++++++++++++++ tests/test_cli/test_display.py | 69 +++++++++++++++++++++++++++-- 2 files changed, 145 insertions(+), 4 deletions(-) diff --git a/src/plyngent/cli/display.py b/src/plyngent/cli/display.py index 0bcd202..5c308f5 100644 --- a/src/plyngent/cli/display.py +++ b/src/plyngent/cli/display.py @@ -3,6 +3,7 @@ from __future__ import annotations import contextlib import json import os +import shlex import sys from contextvars import ContextVar from typing import TYPE_CHECKING, Literal, cast @@ -40,6 +41,8 @@ _PRETTY_TOOLS = frozenset( "listdir", "glob_paths", "grep_files", + "run_argv", + "run_argv_batch", } ) @@ -113,6 +116,16 @@ def _json_str_arg(args_json: str, key: str) -> str | None: return value if isinstance(value, str) else None +def _json_list_arg(args_json: str, key: str) -> list[str] | None: + value = _json_arg(args_json, key) + if isinstance(value, list): + raw_items = cast("list[object]", value) + items = [item for item in raw_items if isinstance(item, str)] + if len(items) == len(raw_items): + return items + return None + + def _pretty_line(prefix: str, *segments: tuple[str, str | None]) -> str: """Style a pretty tool summary: yellow ``* Verb`` prefix + colored detail segments. @@ -253,6 +266,71 @@ def _grep_pretty(args_json: str, result: str) -> str: ) +def _section_line_count(result: str, section: str) -> int: + """Count non-empty lines in a ``--- section ---`` block of a tool result.""" + lines = result.splitlines() + try: + start = lines.index(f"--- {section} ---") + 1 + except ValueError: + return 0 + end = next((i for i in range(start, len(lines)) if lines[i].startswith("--- ")), len(lines)) + return sum(1 for line in lines[start:end] if line.strip()) + + +def _run_argv_pretty(args_json: str, result: str) -> str: + """One-line summary for a ``run_argv`` call: ``* Run 'git status' → exit 0 (3 lines)``.""" + argv = _json_list_arg(args_json, "argv") + cmd = shlex.join(argv) if argv else "?" + if result.startswith("error:"): + return _pretty_line(f"* Run '{cmd}' ", (f"({result})", "red")) + fields: dict[str, str] = {} + for line in result.splitlines(): + if line.startswith("--- "): + break + if "=" in line: + key, _, value = line.partition("=") + fields[key] = value + out_lines = _section_line_count(result, "stdout") + _section_line_count(result, "stderr") + unit = "line" if out_lines == 1 else "lines" + if fields.get("timed_out") == "true": + status, fg = "timed out", "red" + else: + code = fields.get("exit_code", "") + status = f"exit {code or 'killed'}" + fg = "green" if code == "0" else "red" + return _pretty_line(f"* Run '{cmd}' ", (f"→ {status} ", fg), (f"({out_lines} {unit})", "dim")) + + +def _run_argv_batch_pretty(_args_json: str, result: str) -> str: + """One-line summary for a ``run_argv_batch`` call: ``* Batch (2/3 steps ran, stopped early)``.""" + if result.startswith("error:"): + return _pretty_line("* Batch ", (f"({result})", "red")) + lines = result.splitlines() + head = lines[0] if lines else "" + fields: dict[str, str] = {} + for part in head.split(): + if "=" in part: + key, _, value = part.partition("=") + fields[key] = value + steps = fields.get("steps", "?") + ran = fields.get("ran", "?") + stopped_early = fields.get("stopped_early") == "true" + last_exit = "" + in_summary = False + for line in lines: + if line == "--- summary ---": + in_summary = True + elif in_summary and line.startswith("last_exit="): + last_exit = line.partition("=")[2] + if stopped_early: + tail = "stopped early" + fg = "yellow" + else: + tail = f"last exit {last_exit or 'killed'}" + fg = "green" if last_exit == "0" else "red" + return _pretty_line("* Batch ", (f"({ran}/{steps} steps ran, {tail})", fg)) + + _PRETTY_BUILDERS: dict[str, Callable[[str, str], str]] = { "read_file": _read_file_pretty, "tree": _tree_pretty, @@ -261,6 +339,8 @@ _PRETTY_BUILDERS: dict[str, Callable[[str, str], str]] = { "listdir": _listdir_pretty, "glob_paths": _glob_pretty, "grep_files": _grep_pretty, + "run_argv": _run_argv_pretty, + "run_argv_batch": _run_argv_batch_pretty, } diff --git a/tests/test_cli/test_display.py b/tests/test_cli/test_display.py index 804992e..609fa91 100644 --- a/tests/test_cli/test_display.py +++ b/tests/test_cli/test_display.py @@ -65,7 +65,7 @@ async def test_flush_markdown_on_source_change( call = AssistantFunctionToolCall( id="1", - function=AssistantFunctionTool(name="run_argv", arguments='{"argv": ["echo", "x"]}'), + function=AssistantFunctionTool(name="vcs_diff", arguments='{"path": ""}'), ) await render_events( _aiter( @@ -178,12 +178,12 @@ async def test_non_pretty_tool_keeps_old_style(capsys: pytest.CaptureFixture[str call = ToolCallEvent( tool_call=AssistantFunctionToolCall( id="3", - function=AssistantFunctionTool(name="run_argv", arguments='{"argv": ["ls"]}'), + function=AssistantFunctionTool(name="vcs_diff", arguments='{"path": ""}'), ) ) - await render_events(_aiter([call, _result("exit_code=0")])) + await render_events(_aiter([call, _result("diff --git a/x b/x")])) out = capsys.readouterr().out - assert "[tool] run_argv(" in out + assert "[tool] vcs_diff(" in out assert "[tool ok]" in out @@ -349,3 +349,64 @@ async def test_pretty_grep_no_matches(capsys: pytest.CaptureFixture[str]) -> Non await render_events(_aiter([_pretty_call("grep_files", '{"pattern": "zzz"}'), _result("(no matches)")])) out = capsys.readouterr().out assert "* Grep 'zzz' in '.' (no matches)" in out + + +async def test_pretty_run_argv_success(capsys: pytest.CaptureFixture[str]) -> None: + result = ( + "exit_code=0\ntimed_out=false\ncwd=.\ncmd=git status --short\n--- stdout ---\n M src/foo.py\n--- stderr ---\n" + ) + await render_events(_aiter([_pretty_call("run_argv", '{"argv": ["git", "status", "--short"]}'), _result(result)])) + out = capsys.readouterr().out + assert "* Run 'git status --short' → exit 0 (1 line)" in out + assert "[tool]" not in out + assert "[tool ok]" not in out + + +async def test_pretty_run_argv_failure(capsys: pytest.CaptureFixture[str]) -> None: + result = ( + "exit_code=1\ntimed_out=false\ncwd=.\ncmd=ls /nope\n--- stdout ---\n--- stderr ---\nls: cannot access '/nope'\n" + ) + await render_events(_aiter([_pretty_call("run_argv", '{"argv": ["ls", "/nope"]}'), _result(result)])) + out = capsys.readouterr().out + assert "* Run 'ls /nope' → exit 1 (1 line)" in out + + +async def test_pretty_run_argv_timed_out(capsys: pytest.CaptureFixture[str]) -> None: + result = "exit_code=\ntimed_out=true\ncwd=.\ncmd=sleep 10\n--- stdout ---\n--- stderr ---\n" + await render_events(_aiter([_pretty_call("run_argv", '{"argv": ["sleep", "10"]}'), _result(result)])) + out = capsys.readouterr().out + assert "* Run 'sleep 10' → timed out (0 lines)" in out + + +async def test_pretty_run_argv_error(capsys: pytest.CaptureFixture[str]) -> None: + await render_events( + _aiter([_pretty_call("run_argv", '{"argv": ["nope"]}'), _result("error: executable not found: 'nope'")]) + ) + out = capsys.readouterr().out + assert "* Run 'nope' (error: executable not found: 'nope')" in out + + +async def test_pretty_run_argv_batch_stopped(capsys: pytest.CaptureFixture[str]) -> None: + result = ( + "steps=3 ran=2 stop_on_error=true stopped_early=true\n" + "--- step 1 ---\nexit_code=1\ntimed_out=false\ncwd=.\ncmd=git status\n" + "pipe_out=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n" + "--- step 2 ---\nexit_code=0\ntimed_out=false\ncwd=.\ncmd=git diff\n" + "pipe_out=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n" + "--- summary ---\nlast_exit=1\n" + ) + await render_events(_aiter([_pretty_call("run_argv_batch", '{"steps": []}'), _result(result)])) + out = capsys.readouterr().out + assert "* Batch (2/3 steps ran, stopped early)" in out + + +async def test_pretty_run_argv_batch_complete(capsys: pytest.CaptureFixture[str]) -> None: + result = ( + "steps=2 ran=2 stop_on_error=true stopped_early=false\n" + "--- step 1 ---\nexit_code=0\n--- stdout ---\n--- stderr ---\n" + "--- step 2 ---\nexit_code=0\n--- stdout ---\n--- stderr ---\n" + "--- summary ---\nlast_exit=0\n" + ) + await render_events(_aiter([_pretty_call("run_argv_batch", '{"steps": []}'), _result(result)])) + out = capsys.readouterr().out + assert "* Batch (2/2 steps ran, last exit 0)" in out