cli/display: list a batch's commands above the '* Run batch' row

src/plyngent/cli/display.py

@@ -33,6 +33,9 @@ if TYPE_CHECKING:
 _TOOL_RESULT_PREVIEW = 120
 _TOOL_ARGS_PREVIEW = 80
 _ASK_SUBJECT_PREVIEW = 60
+# ``run_argv_batch`` lists the commands it runs above its ``* Run batch `` row — one
+# line per command, a ``pipe_out`` chain joined into one — elided past this width.
+_RUN_BATCH_CMD_LIMIT = 80
 _MCP_TOOL_PREFIX = "mcp__"
 _SKILL_MAIN_FILE = "SKILL.md"
 
@@ -117,8 +120,8 @@ 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)
+def _str_list(value: object) -> list[str] | None:
+    """*value* as a list of strings, or None when it is not one."""
     if isinstance(value, list):
         raw_items = cast("list[object]", value)
         items = [item for item in raw_items if isinstance(item, str)]
@@ -127,6 +130,10 @@ def _json_list_arg(args_json: str, key: str) -> list[str] | None:
     return None
 
 
+def _json_list_arg(args_json: str, key: str) -> list[str] | None:
+    return _str_list(_json_arg(args_json, key))
+
+
 def _json_int_arg(args_json: str, key: str) -> int | None:
     value = _json_arg(args_json, key)
     if isinstance(value, bool):
@@ -387,8 +394,42 @@ def _run_argv_detail(_args_json: str, result: str) -> str:
     return _pretty_segments((f"(exit code {code or 'killed'})", fg))
 
 
-def _run_argv_batch_prefix(_args_json: str) -> str:
-    return _pretty_prefix("* Run batch ")
+def _batch_steps(args_json: str) -> list[dict[str, object]]:
+    """Steps of a batch call: a JSON array, or a JSON string holding one."""
+    value = _json_arg(args_json, "steps")
+    if isinstance(value, str):
+        try:
+            value = json.loads(value)
+        except ValueError:
+            return []
+    if not isinstance(value, list):
+        return []
+    raw_steps = cast("list[object]", value)
+    return [cast("dict[str, object]", step) for step in raw_steps if isinstance(step, dict)]
+
+
+def _batch_command_lines(args_json: str) -> list[str]:
+    """One command per line, a ``pipe_out`` chain joined into one line with ``|``."""
+    lines: list[str] = []
+    piped_from_previous = False
+    for step in _batch_steps(args_json):
+        argv = _str_list(step.get("argv"))
+        command = shlex.join(argv) if argv else "?"
+        if piped_from_previous and lines:
+            lines[-1] = f"{lines[-1]} | {command}"
+        else:
+            lines.append(command)
+        # Same flag spellings the tool itself accepts (bool, or "true"/"1"/…).
+        piped_from_previous = str(step.get("pipe_out") or "").strip().lower() in {"1", "true", "yes", "on"}
+    return lines
+
+
+def _run_argv_batch_prefix(args_json: str) -> str:
+    """``* Run batch `` with the commands it runs listed above it."""
+    commands = _batch_command_lines(args_json)
+    body = "\n".join(_pretty_prefix(f"$ {_preview(command, _RUN_BATCH_CMD_LIMIT)}") for command in commands)
+    header = _pretty_prefix("* Run batch ")
+    return f"{body}\n{header}" if body else header
 
 
 def _run_argv_batch_detail(_args_json: str, result: str) -> str:

tests/test_cli/test_display.py

@@ -4,6 +4,7 @@ from typing import TYPE_CHECKING
 
 from plyngent.agent import ErrorEvent, ReasoningDeltaEvent, TextDeltaEvent, ToolCallEvent, ToolResultEvent
 from plyngent.cli.display import (
+    _RUN_BATCH_CMD_LIMIT,
     _clear_streamed_lines,
     _line_count_for_clear,
     _pretty_line_for,
@@ -470,6 +471,67 @@ async def test_pretty_run_argv_batch_complete(capsys: pytest.CaptureFixture[str]
     assert "* Run batch (done)" in out
 
 
+_BATCH_DONE_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"
+)
+
+
+async def test_pretty_run_argv_batch_lists_commands(capsys: pytest.CaptureFixture[str]) -> None:
+    """The batch row leads with one ``$ command`` line per step."""
+    args = '{"steps": [{"argv": ["git", "status"]}, {"argv": ["git", "diff", "--stat"]}]}'
+    await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)]))
+    out = capsys.readouterr().out
+    assert "$ git status\n$ git diff --stat\n* Run batch (done)" in out
+
+
+async def test_pretty_run_argv_batch_joins_pipe_chain(capsys: pytest.CaptureFixture[str]) -> None:
+    """``pipe_out`` steps share one line; a step they do not feed keeps its own."""
+    args = (
+        '{"steps": [{"argv": ["grep", "-n", "foo", "a.py"], "pipe_out": true},'
+        ' {"argv": ["wc", "-l"]}, {"argv": ["git", "status"]}]}'
+    )
+    await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)]))
+    out = capsys.readouterr().out
+    assert "$ grep -n foo a.py | wc -l\n$ git status\n* Run batch (done)" in out
+
+
+async def test_pretty_run_argv_batch_elides_a_long_command(capsys: pytest.CaptureFixture[str]) -> None:
+    long_arg = "x" * 200
+    args = f'{{"steps": [{{"argv": ["echo", "{long_arg}"]}}]}}'
+    await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)]))
+    out = capsys.readouterr().out
+    command = f"echo {long_arg}"
+    line = next(text for text in out.splitlines() if text.startswith("$ "))
+    assert line == f"$ {command[:_RUN_BATCH_CMD_LIMIT]}…"
+
+
+async def test_pretty_run_argv_batch_steps_as_json_string(capsys: pytest.CaptureFixture[str]) -> None:
+    """``steps`` may arrive as a JSON string holding the array."""
+    args = '{"steps": "[{\\"argv\\": [\\"git\\", \\"status\\"]}]"}'
+    await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)]))
+    out = capsys.readouterr().out
+    assert "$ git status\n* Run batch (done)" in out
+
+
+async def test_pretty_run_argv_batch_unreadable_steps_hide_commands(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    await render_events(
+        _aiter(
+            [
+                _pretty_call("run_argv_batch", '{"steps": "nope"}'),
+                _result("error: steps must be a JSON array: Expecting value"),
+            ]
+        )
+    )
+    out = capsys.readouterr().out
+    assert "$ " not in out
+    assert "* Run batch (error: steps must be a JSON array" in out
+
+
 async def test_pretty_fetch(capsys: pytest.CaptureFixture[str]) -> None:
     result = (
         "status=200\nmethod=GET\nfinal_url=https://example.com\ncontent_type=application/json\n"