cli: prettier tool call summaries for read_file and todo tools

This commit is contained in:
2026-08-11 15:38:07 +08:00
parent e4ed89b391
commit 65349b3c07
2 changed files with 139 additions and 7 deletions
+65 -4
View File
@@ -1,10 +1,11 @@
from __future__ import annotations
import contextlib
import json
import os
import sys
from contextvars import ContextVar
from typing import TYPE_CHECKING, Literal
from typing import TYPE_CHECKING, Literal, cast
import click
@@ -28,6 +29,10 @@ if TYPE_CHECKING:
_TOOL_RESULT_PREVIEW = 120
_TOOL_ARGS_PREVIEW = 80
# Tool calls that render as a single pretty summary line instead of
# ``[tool]`` / ``[tool result]``.
_PRETTY_TOOLS = frozenset({"read_file", "todo_push", "todo_update"})
# Process/session display flags (set from ReplState / slash).
_verbose_tool_results: ContextVar[bool] = ContextVar("verbose_tool_results", default=False)
_markdown_enabled: ContextVar[bool] = ContextVar("markdown_enabled", default=True)
@@ -69,6 +74,49 @@ def _preview(text: str, limit: int) -> str:
return text[:limit] + "…"
def _json_str_arg(args_json: str, key: str) -> str | None:
"""Extract a string argument from a tool-call arguments JSON blob."""
try:
raw = json.loads(args_json)
except ValueError:
return None
if isinstance(raw, dict):
value = cast("dict[str, object]", raw).get(key)
if isinstance(value, str):
return value
return None
def _read_file_pretty(args_json: str, result: str) -> str:
"""One-line summary for a ``read_file`` call: ``* Read 'path' L1-4 (done)``."""
path = _json_str_arg(args_json, "path") or "?"
if result.startswith("error: file not found"):
return f"* Read '{path}' (file not found)"
if result.startswith("error: not a file"):
return f"* Read '{path}' (not a file)"
if result.startswith("error:"):
return f"* Read '{path}' (error)"
first = result.splitlines()[0] if result.splitlines() else ""
if first.startswith("L") and "-" in first:
return f"* Read '{path}' {first} (done)"
return f"* Read '{path}' (done)"
def _todo_pretty(name: str, result: str) -> str:
"""Summary for todo push/update: header + the rendered todo stack."""
label = "Todo Push" if name == "todo_push" else "Todo Update"
return f"* {label}:\n{result}"
def _pretty_tool_result(name: str, args_json: str, result: str) -> str | None:
"""Pretty summary for a known tool call + result; None keeps the old style."""
if name == "read_file":
return _read_file_pretty(args_json, result)
if name in {"todo_push", "todo_update"}:
return _todo_pretty(name, result)
return None
def _echo_stream(text: str) -> None:
"""Write without newline and flush so assistant text appears token-by-token."""
click.echo(text, nl=False)
@@ -144,6 +192,9 @@ async def render_events( # noqa: C901, PLR0912, PLR0915
assistant_buf: list[str] = []
printed_reasoning = False
printed_assistant = False
# Tool calls buffer so prettified tools can render once their result lands
# (the summary line needs the status/range). FIFO matches loop event order.
pending_tools: list[tuple[str, str, bool]] = []
def flush_assistant() -> None:
nonlocal source, assistant_buf, printed_assistant
@@ -192,14 +243,24 @@ async def render_events( # noqa: C901, PLR0912, PLR0915
flush_assistant()
call = event.tool_call
if isinstance(call, AssistantFunctionToolCall):
args = _preview(call.function.arguments, _TOOL_ARGS_PREVIEW)
click.secho(f"\n[tool] {call.function.name}({args})", fg="yellow")
name = call.function.name
args = call.function.arguments
pretty_tool = name in _PRETTY_TOOLS
pending_tools.append((name, args, pretty_tool))
if not pretty_tool:
preview = _preview(args, _TOOL_ARGS_PREVIEW)
click.secho(f"\n[tool] {name}({preview})", fg="yellow")
else:
pending_tools.append(("custom", call.id, False))
click.secho(f"\n[tool] custom id={call.id}", fg="yellow")
elif isinstance(event, ToolResultEvent):
flush_assistant()
content = event.message.content
if show_full:
name, args, pretty_tool = pending_tools.pop(0) if pending_tools else ("", "", False)
pretty_line = _pretty_tool_result(name, args, content) if pretty_tool else None
if pretty_line is not None:
click.secho(f"\n{pretty_line}", fg="yellow")
elif show_full:
click.secho(f"[tool ok]\n{content}", fg="magenta")
else:
preview = _preview(content, _TOOL_RESULT_PREVIEW)
+74 -3
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING
from plyngent.agent import ReasoningDeltaEvent, TextDeltaEvent, ToolResultEvent
from plyngent.agent import ReasoningDeltaEvent, TextDeltaEvent, ToolCallEvent, ToolResultEvent
from plyngent.cli.display import (
get_markdown_enabled,
markdown_render_available,
@@ -11,7 +11,11 @@ from plyngent.cli.display import (
set_markdown_enabled,
set_verbose_tool_results,
)
from plyngent.lmproto.openai_compatible.model import ToolChatMessage
from plyngent.lmproto.openai_compatible.model import (
AssistantFunctionTool,
AssistantFunctionToolCall,
ToolChatMessage,
)
if TYPE_CHECKING:
from collections.abc import AsyncIterator
@@ -61,7 +65,7 @@ async def test_flush_markdown_on_source_change(
call = AssistantFunctionToolCall(
id="1",
function=AssistantFunctionTool(name="read_file", arguments="{}"),
function=AssistantFunctionTool(name="run_argv", arguments='{"argv": ["echo", "x"]}'),
)
await render_events(
_aiter(
@@ -78,6 +82,73 @@ async def test_flush_markdown_on_source_change(
assert "after" in out
def _read_call(args_json: str) -> ToolCallEvent:
return ToolCallEvent(
tool_call=AssistantFunctionToolCall(
id="1",
function=AssistantFunctionTool(name="read_file", arguments=args_json),
)
)
def _result(content: str) -> ToolResultEvent:
return ToolResultEvent(message=ToolChatMessage(content=content, tool_call_id="1"))
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 "[tool]" not in out
assert "[tool ok]" not in out
async def test_pretty_read_file_statuses(capsys: pytest.CaptureFixture[str]) -> None:
cases = [
("error: file not found: a.txt", "(file not found)"),
("error: not a file: a.txt", "(not a file)"),
("error: something else", "(error)"),
("no header content", "(done)"),
]
for content, status in cases:
await render_events(_aiter([_read_call('{"path": "a.txt"}'), _result(content)]))
out = capsys.readouterr().out
assert f"* Read 'a.txt' {status}" in out, f"status {status} not rendered for {content!r}"
async def test_pretty_todo_push(capsys: pytest.CaptureFixture[str]) -> None:
call = ToolCallEvent(
tool_call=AssistantFunctionToolCall(
id="2",
function=AssistantFunctionTool(name="todo_push", arguments='{"titles": ["T1"]}'),
)
)
result = (
"pushed group (depth=1) items=[a1]\n"
"(LIFO of groups: depth=1; TOP group = current breakdown level)\n"
"group d=0 TOP:\n"
" [ ] a1: T1"
)
await render_events(_aiter([call, _result(result)]))
out = capsys.readouterr().out
assert "* Todo Push:" in out
assert " [ ] a1: T1" in out
assert "[tool]" not in out
async def test_non_pretty_tool_keeps_old_style(capsys: pytest.CaptureFixture[str]) -> None:
call = ToolCallEvent(
tool_call=AssistantFunctionToolCall(
id="3",
function=AssistantFunctionTool(name="run_argv", arguments='{"argv": ["ls"]}'),
)
)
await render_events(_aiter([call, _result("exit_code=0")]))
out = capsys.readouterr().out
assert "[tool] run_argv(" in out
assert "[tool ok]" in out
async def test_tool_result_preview_vs_verbose(capsys: pytest.CaptureFixture[str]) -> None:
long = "x" * 200
msg = ToolChatMessage(content=long, tool_call_id="1")