cli: reduce pretty summaries to minimal done/error semantics

This commit is contained in:
2026-08-15 18:36:21 +08:00
parent 5344a20b67
commit 8e8a10a05e
2 changed files with 82 additions and 126 deletions
+52 -105
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import contextlib
import json
import os
import re
import shlex
import sys
from contextvars import ContextVar
@@ -150,7 +151,7 @@ def _pretty_line(prefix: str, *segments: tuple[str, str | None]) -> str:
def _read_file_pretty(args_json: str, result: str) -> str:
"""One-line summary for a ``read_file`` call: ``* Read 'path' L1-4 (done)``."""
"""One-line summary for a ``read_file`` call: ``* Read 'path' (done)``."""
path = _json_str_arg(args_json, "path") or "?"
if result.startswith("error: file not found"):
return _pretty_line(f"* Read '{path}' ", ("(file not found)", "red"))
@@ -158,9 +159,6 @@ def _read_file_pretty(args_json: str, result: str) -> str:
return _pretty_line(f"* Read '{path}' ", ("(not a file)", "red"))
if result.startswith("error:"):
return _pretty_line(f"* Read '{path}' ", ("(error)", "red"))
first = result.splitlines()[0] if result.splitlines() else ""
if first.startswith("L") and "-" in first:
return _pretty_line(f"* Read '{path}' ", (f"{first} ", "dim"), ("(done)", "green"))
return _pretty_line(f"* Read '{path}' ", ("(done)", "green"))
@@ -273,23 +271,12 @@ 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)``."""
"""One-line summary for a ``run_argv`` call: ``* Run $ git status --short (exit code 0)``."""
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"))
return _pretty_line(f"* Run $ {cmd} ", (f"({result})", "red"))
fields: dict[str, str] = {}
for line in result.splitlines():
if line.startswith("--- "):
@@ -297,63 +284,26 @@ def _run_argv_pretty(args_json: str, result: str) -> str:
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"))
return _pretty_line(f"* Run $ {cmd} ", ("(timed out)", "red"))
code = fields.get("exit_code", "")
fg = "green" if code == "0" else "red"
return _pretty_line(f"* Run $ {cmd} ", (f"(exit code {code or 'killed'})", fg))
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)``."""
"""One-line summary for a ``run_argv_batch`` call: ``* Run batch (done)``."""
if result.startswith("error:"):
return _pretty_line("* Batch ", (f"({result})", "red"))
return _pretty_line("* Run batch ", (f"({result})", "red"))
lines = result.splitlines()
head = lines[0] if lines else ""
fields: dict[str, str] = {}
stopped_early = False
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 part == "stopped_early=true":
stopped_early = True
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))
def _human_bytes(value: str) -> str:
"""Compact byte-count rendering: ``900B`` / ``1.2KB`` / ``3.4MB``."""
kib = 1024
try:
n = int(value)
except ValueError:
return ""
if n < kib:
return f"{n}B"
kb = n / kib
if kb < kib:
return f"{kb:.1f}KB"
mb = kb / kib
if mb < kib:
return f"{mb:.1f}MB"
return f"{mb / kib:.1f}GB"
return _pretty_line("* Run batch ", ("(stopped early)", "yellow"))
return _pretty_line("* Run batch ", ("(done)", "green"))
def _http_status_fg(status: str) -> str | None:
@@ -364,53 +314,50 @@ def _http_status_fg(status: str) -> str | None:
def _fetch_pretty(args_json: str, result: str) -> str:
"""One-line summary for a ``fetch`` call: ``* GET 200 https://… (json, 1.2KB)``."""
"""One-line summary for a ``fetch`` call: ``* Fetch GET https://… (200)``."""
method = _json_str_arg(args_json, "method") or "GET"
url = _json_str_arg(args_json, "url") or "?"
if result.startswith("error:"):
return _pretty_line(f"* {method} ", (f"({result})", "red"))
fields: dict[str, str] = {}
return _pretty_line(f"* Fetch {method} {url} ", (f"({result})", "red"))
status = ""
for line in result.splitlines():
if line.startswith("--- "):
break
if "=" in line:
key, _, value = line.partition("=")
fields[key] = value
status = fields.get("status", "?")
url = fields.get("final_url", "?")
detail_parts: list[str] = []
kind = fields.get("body_kind", "")
if kind:
detail_parts.append(kind)
size = _human_bytes(fields.get("bytes", ""))
if size:
detail_parts.append(size)
if fields.get("truncated") == "true":
detail_parts.append("truncated")
detail = f" ({', '.join(detail_parts)})" if detail_parts else ""
return _pretty_line(f"* {method} {status} {url}", (detail, _http_status_fg(status)))
if line.startswith("status="):
status = line.partition("=")[2]
break
return _pretty_line(f"* Fetch {method} {url} ", (f"({status or 'done'})", _http_status_fg(status)))
_MUTATOR_VERBS: dict[str, str] = {
"edit_replace": "Edited",
"edit_lineno": "Edited",
"write_file": "Wrote",
"copy_path": "Copied",
"move_path": "Moved",
"delete_path": "Deleted",
"edit_replace": "Edit",
"edit_lineno": "Edit",
"write_file": "Write",
"copy_path": "Copy",
"move_path": "Move",
"delete_path": "Delete",
}
_MUTATOR_LOWER_PREFIXES = ("wrote ", "copied ", "moved ", "deleted ")
def _mutator_pretty(name: str, result: str) -> str:
"""One-line summary for file-mutation tools: ``* Wrote 120 characters to src/x.py``."""
def _mutator_pretty(name: str, args_json: str, result: str) -> str:
"""One-line summary for a file mutation: ``* Wrote 'path' (120 chars)`` / ``* Edit 'path' (done)``."""
verb = _MUTATOR_VERBS[name]
if name in {"copy_path", "move_path"}:
src = _json_str_arg(args_json, "src") or "?"
dst = _json_str_arg(args_json, "dst") or "?"
target = f"'{src}' → '{dst}'"
else:
path = _json_str_arg(args_json, "path") or "?"
target = f"'{path}'"
if result.startswith("error:"):
return _pretty_line(f"* {verb} ", (f"({result})", "red"))
for prefix in _MUTATOR_LOWER_PREFIXES:
if result.startswith(prefix):
return _pretty_line(f"* {verb} {result[len(prefix) :]}")
return _pretty_line(f"* {verb}: {result}")
return _pretty_line(f"* {verb} {target} ", (f"({result})", "red"))
if name == "write_file":
# write_file keeps its brief detail: ``wrote 120 characters to path``.
match = re.match(r"wrote (\d+) characters to (.+)$", result)
if match:
chars, path = match.groups()
return _pretty_line(f"* Write '{path}' ", (f"({chars} chars)", None))
return _pretty_line(f"* {verb} {target} ", ("(done)", "green"))
_PRETTY_BUILDERS: dict[str, Callable[[str, str], str]] = {
@@ -424,12 +371,12 @@ _PRETTY_BUILDERS: dict[str, Callable[[str, str], str]] = {
"run_argv": _run_argv_pretty,
"run_argv_batch": _run_argv_batch_pretty,
"fetch": _fetch_pretty,
"edit_replace": lambda _args, result: _mutator_pretty("edit_replace", result),
"edit_lineno": lambda _args, result: _mutator_pretty("edit_lineno", result),
"write_file": lambda _args, result: _mutator_pretty("write_file", result),
"copy_path": lambda _args, result: _mutator_pretty("copy_path", result),
"move_path": lambda _args, result: _mutator_pretty("move_path", result),
"delete_path": lambda _args, result: _mutator_pretty("delete_path", result),
"edit_replace": lambda args, result: _mutator_pretty("edit_replace", args, result),
"edit_lineno": lambda args, result: _mutator_pretty("edit_lineno", args, result),
"write_file": lambda args, result: _mutator_pretty("write_file", args, result),
"copy_path": lambda args, result: _mutator_pretty("copy_path", args, result),
"move_path": lambda args, result: _mutator_pretty("move_path", args, result),
"delete_path": lambda args, result: _mutator_pretty("delete_path", args, result),
}
+30 -21
View File
@@ -98,7 +98,7 @@ def _result(content: str) -> ToolResultEvent:
async def test_pretty_read_file_done(capsys: pytest.CaptureFixture[str]) -> None:
await render_events(_aiter([_read_call('{"path": "a.txt"}'), _result("L1-4\none\ntwo\n")]))
out = capsys.readouterr().out
assert "* Read 'a.txt' L1-4 (done)" in out
assert "* Read 'a.txt' (done)" in out
assert "[tool]" not in out
assert "[tool ok]" not in out
@@ -357,7 +357,7 @@ async def test_pretty_run_argv_success(capsys: pytest.CaptureFixture[str]) -> No
)
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 "* Run $ git status --short (exit code 0)" in out
assert "[tool]" not in out
assert "[tool ok]" not in out
@@ -368,14 +368,14 @@ async def test_pretty_run_argv_failure(capsys: pytest.CaptureFixture[str]) -> No
)
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
assert "* Run $ ls /nope (exit code 1)" 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
assert "* Run $ sleep 10 (timed out)" in out
async def test_pretty_run_argv_error(capsys: pytest.CaptureFixture[str]) -> None:
@@ -383,7 +383,7 @@ async def test_pretty_run_argv_error(capsys: pytest.CaptureFixture[str]) -> None
_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
assert "* Run $ nope (error: executable not found: 'nope')" in out
async def test_pretty_run_argv_batch_stopped(capsys: pytest.CaptureFixture[str]) -> None:
@@ -397,7 +397,7 @@ async def test_pretty_run_argv_batch_stopped(capsys: pytest.CaptureFixture[str])
)
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
assert "* Run batch (stopped early)" in out
async def test_pretty_run_argv_batch_complete(capsys: pytest.CaptureFixture[str]) -> None:
@@ -409,7 +409,7 @@ async def test_pretty_run_argv_batch_complete(capsys: pytest.CaptureFixture[str]
)
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
assert "* Run batch (done)" in out
async def test_pretty_fetch(capsys: pytest.CaptureFixture[str]) -> None:
@@ -419,7 +419,7 @@ async def test_pretty_fetch(capsys: pytest.CaptureFixture[str]) -> None:
)
await render_events(_aiter([_pretty_call("fetch", '{"url": "https://example.com"}'), _result(result)]))
out = capsys.readouterr().out
assert "* GET 200 https://example.com (json, 1.2KB)" in out
assert "* Fetch GET https://example.com (200)" in out
assert "[tool]" not in out
@@ -430,7 +430,7 @@ async def test_pretty_fetch_error_status(capsys: pytest.CaptureFixture[str]) ->
)
await render_events(_aiter([_pretty_call("fetch", '{"url": "https://example.com/nope"}'), _result(result)]))
out = capsys.readouterr().out
assert "* GET 404 https://example.com/nope (html, 45B)" in out
assert "* Fetch GET https://example.com/nope (404)" in out
async def test_pretty_fetch_truncated(capsys: pytest.CaptureFixture[str]) -> None:
@@ -440,7 +440,7 @@ async def test_pretty_fetch_truncated(capsys: pytest.CaptureFixture[str]) -> Non
)
await render_events(_aiter([_pretty_call("fetch", '{"url": "https://example.com"}'), _result(result)]))
out = capsys.readouterr().out
assert "* GET 200 https://example.com (text, 48.8KB, truncated)" in out
assert "* Fetch GET https://example.com (200)" in out
async def test_pretty_fetch_error(capsys: pytest.CaptureFixture[str]) -> None:
@@ -448,34 +448,43 @@ async def test_pretty_fetch_error(capsys: pytest.CaptureFixture[str]) -> None:
_aiter([_pretty_call("fetch", '{"url": "https://example.com"}'), _result("error: fetch failed: boom")])
)
out = capsys.readouterr().out
assert "* GET (error: fetch failed: boom)" in out
assert "* Fetch GET https://example.com (error: fetch failed: boom)" in out
async def test_pretty_mutators(capsys: pytest.CaptureFixture[str]) -> None:
cases = [
("write_file", "wrote 120 characters to src/x.py", "* Wrote 120 characters to src/x.py"),
("write_file", '{"path": "src/x.py"}', "wrote 120 characters to src/x.py", "* Write 'src/x.py' (120 chars)"),
(
"edit_replace",
'{"path": "src/a.py"}',
"replaced 1 occurrence in src/a.py ('x' → 'y')",
"* Edited: replaced 1 occurrence in src/a.py ('x' → 'y')",
"* Edit 'src/a.py' (done)",
),
(
"edit_lineno",
'{"path": "src/a.py"}',
"replaced lines 3-5 (3 lines; first: 'pass') with 2 lines in src/a.py",
"* Edited: replaced lines 3-5 (3 lines; first: 'pass') with 2 lines in src/a.py",
"* Edit 'src/a.py' (done)",
),
("copy_path", "copied file a.txt -> b.txt", "* Copied file a.txt -> b.txt"),
("move_path", "moved directory src -> dst", "* Moved directory src -> dst"),
("delete_path", "deleted file tmp/x.txt", "* Deleted file tmp/x.txt"),
(
"copy_path",
'{"src": "a.txt", "dst": "b.txt"}',
"copied file a.txt -> b.txt",
"* Copy 'a.txt' → 'b.txt' (done)",
),
("move_path", '{"src": "src", "dst": "dst"}', "moved directory src -> dst", "* Move 'src' → 'dst' (done)"),
("delete_path", '{"path": "tmp/x.txt"}', "deleted file tmp/x.txt", "* Delete 'tmp/x.txt' (done)"),
]
for name, content, expected in cases:
await render_events(_aiter([_pretty_call(name, "{}"), _result(content)]))
for name, args, content, expected in cases:
await render_events(_aiter([_pretty_call(name, args), _result(content)]))
out = capsys.readouterr().out
assert expected in out, f"{name}: expected {expected!r} in {out!r}"
assert "[tool ok]" not in out
async def test_pretty_mutator_error(capsys: pytest.CaptureFixture[str]) -> None:
await render_events(_aiter([_pretty_call("copy_path", "{}"), _result("error: source does not exist: x")]))
await render_events(
_aiter([_pretty_call("copy_path", '{"src": "x", "dst": "y"}'), _result("error: source does not exist: x")])
)
out = capsys.readouterr().out
assert "* Copied (error: source does not exist: x)" in out
assert "* Copy 'x' → 'y' (error: source does not exist: x)" in out