cli: pretty printed summaries for run_argv and run_argv_batch

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,
 }
 
 

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