From 65349b3c076f87ff28c9cba74af8c60b3a14611f Mon Sep 17 00:00:00 2001 From: worldmozara Date: Tue, 11 Aug 2026 15:38:07 +0800 Subject: [PATCH] cli: prettier tool call summaries for read_file and todo tools --- src/plyngent/cli/display.py | 69 ++++++++++++++++++++++++++++-- tests/test_cli/test_display.py | 77 ++++++++++++++++++++++++++++++++-- 2 files changed, 139 insertions(+), 7 deletions(-) diff --git a/src/plyngent/cli/display.py b/src/plyngent/cli/display.py index 348c6f2..abc24f0 100644 --- a/src/plyngent/cli/display.py +++ b/src/plyngent/cli/display.py @@ -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) diff --git a/tests/test_cli/test_display.py b/tests/test_cli/test_display.py index 140f725..0924c8e 100644 --- a/tests/test_cli/test_display.py +++ b/tests/test_cli/test_display.py @@ -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")