cli/display: show pretty tool calls while they run

src/plyngent/cli/display.py

@@ -7,6 +7,7 @@ import re
 import shlex
 import sys
 from contextvars import ContextVar
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Literal, cast
 
 import click
@@ -31,32 +32,6 @@ 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",
-        "tree",
-        "listdir",
-        "glob_paths",
-        "grep_files",
-        "run_argv",
-        "run_argv_batch",
-        "fetch",
-        "edit_replace",
-        "edit_lineno",
-        "write_file",
-        "copy_path",
-        "move_path",
-        "delete_path",
-        "vcs_status",
-        "vcs_log",
-        "wait",
-    }
-)
-
 # 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)
@@ -82,8 +57,8 @@ def get_markdown_enabled() -> bool:
     return _markdown_enabled.get()
 
 
-def markdown_render_available() -> bool:
+def interactive_terminal() -> bool:
-    """True when stdout is a TTY and plain mode is not forced via env."""
+    """True when stdout is a terminal we may write mid-line and erase on."""
     if os.environ.get("PLYNGENT_PLAIN", "").strip() in {"1", "true", "yes", "on"}:
         return False
     try:
@@ -92,6 +67,11 @@ def markdown_render_available() -> bool:
         return False
 
 
+def markdown_render_available() -> bool:
+    """True when stdout is a TTY and plain mode is not forced via env."""
+    return interactive_terminal()
+
+
 def _preview(text: str, limit: int) -> str:
     if len(text) <= limit:
         return text
@@ -137,14 +117,19 @@ def _json_list_arg(args_json: str, key: str) -> list[str] | None:
     return None
 
 
-def _pretty_line(prefix: str, *segments: tuple[str, str | None]) -> str:
+def _pretty_prefix(text: str) -> str:
-    """Style a pretty tool summary: yellow ``* Verb`` prefix + colored detail segments.
+    """Style the leading ``* Verb 'target' `` segment of a pretty tool line."""
+    return click.style(text, fg="yellow")
+
+
+def _pretty_segments(*segments: tuple[str, str | None]) -> str:
+    """Style pretty-line detail segments.
 
     ``segments`` are ``(text, fg)`` pairs; ``fg=None`` leaves a segment uncolored
     and ``"dim"`` renders it dim. click.style emits ANSI codes only on a TTY,
     so non-TTY output stays plain.
     """
-    parts = [click.style(prefix, fg="yellow")]
+    parts: list[str] = []
     for text, fg in segments:
         if fg == "dim":
             parts.append(click.style(text, dim=True))
@@ -153,25 +138,51 @@ def _pretty_line(prefix: str, *segments: tuple[str, str | None]) -> str:
     return "".join(parts)
 
 
-def _read_file_pretty(args_json: str, result: str) -> str:
+@dataclass(frozen=True, slots=True)
-    """One-line summary for a ``read_file`` call: ``* Read 'path' L1-4 (done)``."""
+class PrettyLine:
+    """Split renderer for one pretty tool call.
+
+    ``prefix`` is written as soon as the call starts — a blocking tool call is
+    then visible while it runs — and ``detail`` is appended once its result
+    lands. Together they equal the single line the whole-line path prints.
+    """
+
+    prefix: Callable[[str], str]
+    detail: Callable[[str, str], str]
+
+
+def _read_file_prefix(args_json: str) -> str:
+    """``* Read 'path' ``: known before the result, so it can lead the line."""
     path = _json_str_arg(args_json, "path") or "?"
+    return _pretty_prefix(f"* Read '{path}' ")
+
+
+def _read_file_detail(_args_json: str, result: str) -> str:
+    """``read_file`` outcome: range/done, or which failure kind occurred."""
     if result.startswith("error: file not found"):
-        return _pretty_line(f"* Read '{path}' ", ("(file not found)", "red"))
+        return _pretty_segments(("(file not found)", "red"))
     if result.startswith("error: not a file"):
-        return _pretty_line(f"* Read '{path}' ", ("(not a file)", "red"))
+        return _pretty_segments(("(not a file)", "red"))
     if result.startswith("error:"):
-        return _pretty_line(f"* Read '{path}' ", ("(error)", "red"))
+        return _pretty_segments(("(error)", "red"))
-    first = result.splitlines()[0] if result.splitlines() else ""
+    lines = result.splitlines()
+    first = lines[0] if lines else ""
     if first.startswith("L") and "-" in first:
-        return _pretty_line(f"* Read '{path}' ", (f"{first} ", "dim"), ("(done)", "green"))
+        return _pretty_segments((f"{first} ", "dim"), ("(done)", "green"))
-    return _pretty_line(f"* Read '{path}' ", ("(done)", "green"))
+    return _pretty_segments(("(done)", "green"))
 
 
-def _todo_pretty(name: str, result: str) -> str:
+def _todo_line(name: str) -> PrettyLine:
-    """Summary for todo push/update: header + the rendered todo stack."""
+    """``todo_push`` / ``todo_update``: header first, rendered stack as the detail."""
     label = "Todo Push" if name == "todo_push" else "Todo Update"
-    return _pretty_line(f"* {label}:\n{result}")
+
+    def prefix(_args_json: str) -> str:
+        return click.style(f"* {label}:", fg="yellow")
+
+    def detail(_args_json: str, result: str) -> str:
+        return click.style(f"\n{result}", fg="yellow")
+
+    return PrettyLine(prefix=prefix, detail=detail)
 
 
 def _tree_line_stats(fmt: str, line: str) -> tuple[str, int] | None:
@@ -202,15 +213,19 @@ def _tree_line_stats(fmt: str, line: str) -> tuple[str, int] | None:
     return kind, depth
 
 
-def _tree_pretty(args_json: str, result: str) -> str:
+def _tree_prefix(args_json: str) -> str:
-    """One-line summary for a ``tree`` call: ``* Tree 'src' (3 dirs, 24 files, depth 3)``.
+    path = _json_str_arg(args_json, "path") or "."
+    return _pretty_prefix(f"* Tree '{path}' ")
+
+
+def _tree_detail(args_json: str, result: str) -> str:
+    """``tree`` outcome: dir/file counts, plus depth when the format has one.
 
     Flat trees have no depth, so the depth part is omitted for ``format=flat``.
     """
-    path = _json_str_arg(args_json, "path") or "."
-    fmt = _json_str_arg(args_json, "format") or "markdown"
     if result.startswith("error:"):
-        return f"* Tree '{path}' (error)"
+        return _pretty_segments(("(error)", "red"))
+    fmt = _json_str_arg(args_json, "format") or "markdown"
     dirs = 0
     files = 0
     depth = 0
@@ -225,17 +240,21 @@ def _tree_pretty(args_json: str, result: str) -> str:
             files += 1
         depth = max(depth, line_depth)
     if depth:
-        return _pretty_line(f"* Tree '{path}' ", (f"({dirs} dirs, {files} files, depth {depth})", None))
+        return _pretty_segments((f"({dirs} dirs, {files} files, depth {depth})", None))
-    return _pretty_line(f"* Tree '{path}' ", (f"({dirs} dirs, {files} files)", None))
+    return _pretty_segments((f"({dirs} dirs, {files} files)", None))
 
 
-def _listdir_pretty(args_json: str, result: str) -> str:
+def _listdir_prefix(args_json: str) -> str:
-    """One-line summary for a ``listdir`` call: ``* List 'path' (3 dirs, 5 files)``."""
     path = _json_str_arg(args_json, "path") or "."
+    return _pretty_prefix(f"* List '{path}' ")
+
+
+def _listdir_detail(_args_json: str, result: str) -> str:
+    """``listdir`` outcome: dir/file counts, or empty/error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* List '{path}' ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     if result == "(empty)":
-        return _pretty_line(f"* List '{path}' ", ("(empty)", "dim"))
+        return _pretty_segments(("(empty)", "dim"))
     dirs = 0
     files = 0
     for line in result.splitlines():
@@ -243,46 +262,55 @@ def _listdir_pretty(args_json: str, result: str) -> str:
             dirs += 1
         elif line.startswith("file\t"):
             files += 1
-    return _pretty_line(f"* List '{path}' ", (f"({dirs} dirs, {files} files)", None))
+    return _pretty_segments((f"({dirs} dirs, {files} files)", None))
 
 
-def _glob_pretty(args_json: str, result: str) -> str:
+def _glob_prefix(args_json: str) -> str:
-    """One-line summary for a ``glob_paths`` call: ``* Glob '**/*.py' in '.' (12 paths)``."""
     pattern = _json_str_arg(args_json, "pattern") or "?"
     path = _json_str_arg(args_json, "path") or "."
+    return _pretty_prefix(f"* Glob '{pattern}' in '{path}' ")
+
+
+def _glob_detail(_args_json: str, result: str) -> str:
+    """``glob_paths`` outcome: match count, or none/error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* Glob '{pattern}' in '{path}' ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     if result == "(no matches)":
-        return _pretty_line(f"* Glob '{pattern}' in '{path}' ", ("(no matches)", "dim"))
+        return _pretty_segments(("(no matches)", "dim"))
     count = sum(1 for line in result.splitlines() if line and not line.startswith("...[truncated"))
     unit = "path" if count == 1 else "paths"
-    return _pretty_line(f"* Glob '{pattern}' in '{path}' ", (f"({count} {unit})", None))
+    return _pretty_segments((f"({count} {unit})", None))
 
 
-def _grep_pretty(args_json: str, result: str) -> str:
+def _grep_prefix(args_json: str) -> str:
-    """One-line summary for a ``grep_files`` call: ``* Grep 'pat' in '.' (12 matches in 3 files)``."""
     pattern = _json_str_arg(args_json, "pattern") or "?"
     path = _json_str_arg(args_json, "path") or "."
+    return _pretty_prefix(f"* Grep '{pattern}' in '{path}' ")
+
+
+def _grep_detail(_args_json: str, result: str) -> str:
+    """``grep_files`` outcome: matches and file count, or none/error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* Grep '{pattern}' in '{path}' ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     if result == "(no matches)":
-        return _pretty_line(f"* Grep '{pattern}' in '{path}' ", ("(no matches)", "dim"))
+        return _pretty_segments(("(no matches)", "dim"))
     matches = [line for line in result.splitlines() if line and not line.startswith("...[truncated")]
     files = len({line.split(":", 1)[0] for line in matches})
     match_unit = "match" if len(matches) == 1 else "matches"
     file_unit = "file" if files == 1 else "files"
-    return _pretty_line(
+    return _pretty_segments((f"({len(matches)} {match_unit} in {files} {file_unit})", None))
-        f"* Grep '{pattern}' in '{path}' ",
-        (f"({len(matches)} {match_unit} in {files} {file_unit})", None),
-    )
 
 
-def _run_argv_pretty(args_json: str, result: str) -> str:
+def _run_argv_prefix(args_json: str) -> str:
-    """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 "?"
+    return _pretty_prefix(f"* Run $ {cmd} ")
+
+
+def _run_argv_detail(_args_json: str, result: str) -> str:
+    """``run_argv`` outcome: exit code, timeout, or error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* Run $ {cmd} ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     fields: dict[str, str] = {}
     for line in result.splitlines():
         if line.startswith("--- "):
@@ -291,25 +319,25 @@ def _run_argv_pretty(args_json: str, result: str) -> str:
             key, _, value = line.partition("=")
             fields[key] = value
     if fields.get("timed_out") == "true":
-        return _pretty_line(f"* Run $ {cmd} ", ("(timed out)", "red"))
+        return _pretty_segments(("(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))
+    return _pretty_segments((f"(exit code {code or 'killed'})", fg))
 
 
-def _run_argv_batch_pretty(_args_json: str, result: str) -> str:
+def _run_argv_batch_prefix(_args_json: str) -> str:
-    """One-line summary for a ``run_argv_batch`` call: ``* Run batch (done)``."""
+    return _pretty_prefix("* Run batch ")
+
+
+def _run_argv_batch_detail(_args_json: str, result: str) -> str:
+    """``run_argv_batch`` outcome: done, stopped early, or error."""
     if result.startswith("error:"):
-        return _pretty_line("* Run batch ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     lines = result.splitlines()
     head = lines[0] if lines else ""
-    stopped_early = False
+    if any(part == "stopped_early=true" for part in head.split()):
-    for part in head.split():
+        return _pretty_segments(("(stopped early)", "yellow"))
-        if part == "stopped_early=true":
+    return _pretty_segments(("(done)", "green"))
-            stopped_early = True
-    if stopped_early:
-        return _pretty_line("* Run batch ", ("(stopped early)", "yellow"))
-    return _pretty_line("* Run batch ", ("(done)", "green"))
 
 
 def _http_status_fg(status: str) -> str | None:
@@ -319,12 +347,16 @@ def _http_status_fg(status: str) -> str | None:
     return None
 
 
-def _fetch_pretty(args_json: str, result: str) -> str:
+def _fetch_prefix(args_json: str) -> str:
-    """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 "?"
+    return _pretty_prefix(f"* Fetch {method} {url} ")
+
+
+def _fetch_detail(_args_json: str, result: str) -> str:
+    """``fetch`` outcome: HTTP status, or error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* Fetch {method} {url} ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     status = ""
     for line in result.splitlines():
         if line.startswith("--- "):
@@ -332,7 +364,7 @@ def _fetch_pretty(args_json: str, result: str) -> str:
         if line.startswith("status="):
             status = line.partition("=")[2]
             break
-    return _pretty_line(f"* Fetch {method} {url} ", (f"({status or 'done'})", _http_status_fg(status)))
+    return _pretty_segments((f"({status or 'done'})", _http_status_fg(status)))
 
 
 _MUTATOR_VERBS: dict[str, str] = {
@@ -345,83 +377,107 @@ _MUTATOR_VERBS: dict[str, str] = {
 }
 
 
-def _mutator_pretty(name: str, args_json: str, result: str) -> str:
+def _mutator_prefix(name: str, args_json: 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"'{_json_str_arg(args_json, 'path') or '?'}'"
-        target = f"'{path}'"
+    return _pretty_prefix(f"* {_MUTATOR_VERBS[name]} {target} ")
+
+
+def _mutator_detail(name: str, result: str) -> str:
+    """File mutation outcome: written size, done, or error."""
     if result.startswith("error:"):
-        return _pretty_line(f"* {verb} {target} ", (f"({result})", "red"))
+        return _pretty_segments((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()
+            chars, _path = match.groups()
-            return _pretty_line(f"* Write '{path}' ", (f"({chars} chars)", None))
+            return _pretty_segments((f"({chars} chars)", None))
-    return _pretty_line(f"* {verb} {target} ", ("(done)", "green"))
+    return _pretty_segments(("(done)", "green"))
 
 
-def _vcs_status_pretty(_args_json: str, result: str) -> str:
+def _mutator_line(name: str) -> PrettyLine:
-    """One-line summary for ``vcs_status``: ``* VCS Status (done)``."""
+    return PrettyLine(
+        prefix=lambda args: _mutator_prefix(name, args),
+        detail=lambda _args, result: _mutator_detail(name, result),
+    )
+
+
+def _vcs_status_prefix(_args_json: str) -> str:
+    return _pretty_prefix("* VCS Status ")
+
+
+def _vcs_status_detail(_args_json: str, result: str) -> str:
+    """``vcs_status`` outcome: done, or error."""
     if result.startswith("error:"):
-        return _pretty_line("* VCS Status ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
-    return _pretty_line("* VCS Status ", ("(done)", "green"))
+    return _pretty_segments(("(done)", "green"))
 
 
-def _vcs_log_pretty(_args_json: str, result: str) -> str:
+def _vcs_log_prefix(_args_json: str) -> str:
-    """One-line summary for ``vcs_log``: ``* VCS Log (5 commits)``."""
+    return _pretty_prefix("* VCS Log ")
+
+
+def _vcs_log_detail(_args_json: str, result: str) -> str:
+    """``vcs_log`` outcome: commit count, none, or error."""
     if result.startswith("error:"):
-        return _pretty_line("* VCS Log ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     if result == "(no commits)":
-        return _pretty_line("* VCS Log ", ("(no commits)", "dim"))
+        return _pretty_segments(("(no commits)", "dim"))
     commits = [line for line in result.splitlines() if line.strip()]
     unit = "commit" if len(commits) == 1 else "commits"
-    return _pretty_line("* VCS Log ", (f"({len(commits)} {unit})", None))
+    return _pretty_segments((f"({len(commits)} {unit})", None))
 
 
-def _wait_pretty(_args_json: str, result: str) -> str:
+def _wait_prefix(_args_json: str) -> str:
-    """One-line summary for a ``wait`` call: ``* Wait (5s)`` / ``* Wait (disturbed)``."""
+    return _pretty_prefix("* Wait ")
+
+
+def _wait_detail(_args_json: str, result: str) -> str:
+    """``wait`` outcome: slept duration, disturbed, or error."""
     if result.startswith("error:"):
-        return _pretty_line("* Wait ", (f"({result})", "red"))
+        return _pretty_segments((f"({result})", "red"))
     if result.startswith("waited "):
-        return _pretty_line("* Wait ", (f"({result.removeprefix('waited ')})", "green"))
+        return _pretty_segments((f"({result.removeprefix('waited ')})", "green"))
-    return _pretty_line("* Wait ", ("(disturbed)", "yellow"))
+    return _pretty_segments(("(disturbed)", "yellow"))
 
 
-_PRETTY_BUILDERS: dict[str, Callable[[str, str], str]] = {
+# Pretty tools: ``prefix`` prints as the call starts, ``detail`` when it lands.
-    "read_file": _read_file_pretty,
+_PRETTY_LINES: dict[str, PrettyLine] = {
-    "tree": _tree_pretty,
+    "read_file": PrettyLine(prefix=_read_file_prefix, detail=_read_file_detail),
-    "todo_push": lambda _args, result: _todo_pretty("todo_push", result),
+    "tree": PrettyLine(prefix=_tree_prefix, detail=_tree_detail),
-    "todo_update": lambda _args, result: _todo_pretty("todo_update", result),
+    "todo_push": _todo_line("todo_push"),
-    "listdir": _listdir_pretty,
+    "todo_update": _todo_line("todo_update"),
-    "glob_paths": _glob_pretty,
+    "listdir": PrettyLine(prefix=_listdir_prefix, detail=_listdir_detail),
-    "grep_files": _grep_pretty,
+    "glob_paths": PrettyLine(prefix=_glob_prefix, detail=_glob_detail),
-    "run_argv": _run_argv_pretty,
+    "grep_files": PrettyLine(prefix=_grep_prefix, detail=_grep_detail),
-    "run_argv_batch": _run_argv_batch_pretty,
+    "run_argv": PrettyLine(prefix=_run_argv_prefix, detail=_run_argv_detail),
-    "fetch": _fetch_pretty,
+    "run_argv_batch": PrettyLine(prefix=_run_argv_batch_prefix, detail=_run_argv_batch_detail),
-    "edit_replace": lambda args, result: _mutator_pretty("edit_replace", args, result),
+    "fetch": PrettyLine(prefix=_fetch_prefix, detail=_fetch_detail),
-    "edit_lineno": lambda args, result: _mutator_pretty("edit_lineno", args, result),
+    "edit_replace": _mutator_line("edit_replace"),
-    "write_file": lambda args, result: _mutator_pretty("write_file", args, result),
+    "edit_lineno": _mutator_line("edit_lineno"),
-    "copy_path": lambda args, result: _mutator_pretty("copy_path", args, result),
+    "write_file": _mutator_line("write_file"),
-    "move_path": lambda args, result: _mutator_pretty("move_path", args, result),
+    "copy_path": _mutator_line("copy_path"),
-    "delete_path": lambda args, result: _mutator_pretty("delete_path", args, result),
+    "move_path": _mutator_line("move_path"),
-    "vcs_status": _vcs_status_pretty,
+    "delete_path": _mutator_line("delete_path"),
-    "vcs_log": _vcs_log_pretty,
+    "vcs_status": PrettyLine(prefix=_vcs_status_prefix, detail=_vcs_status_detail),
-    "wait": _wait_pretty,
+    "vcs_log": PrettyLine(prefix=_vcs_log_prefix, detail=_vcs_log_detail),
+    "wait": PrettyLine(prefix=_wait_prefix, detail=_wait_detail),
 }
 
+_PRETTY_TOOLS = frozenset(_PRETTY_LINES)
 
-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."""
+def _pretty_parts(name: str, args_json: str, result: str) -> tuple[str, str] | None:
-    builder = _PRETTY_BUILDERS.get(name)
+    """Prefix + detail for a known tool call; None keeps the ``[tool]`` style."""
-    if builder is None:
+    line = _PRETTY_LINES.get(name)
+    if line is None:
         return None
-    return builder(args_json, result)
+    return line.prefix(args_json), line.detail(args_json, result)
 
 
 def _echo_stream(text: str) -> None:
@@ -452,6 +508,63 @@ def _line_count_for_clear(label: str, body: str) -> int:
     return text.count("\n") + 1
 
 
+class _PrettyToolStream:
+    """Pretty tool lines: prefix at call time, detail appended when it lands.
+
+    Writing the prefix immediately keeps a blocking tool call visible; appending
+    the detail later reproduces the single whole-line write byte for byte, so
+    non-interactive output is unchanged. The calls of one batch are yielded
+    before their results, so a second call while a prefix is open (parallel
+    tools) cannot append its detail in order: that prefix is erased and the
+    batch falls back to whole lines.
+    """
+
+    _interactive: bool
+    _open: bool
+    _whole_lines: bool
+
+    def __init__(self, *, interactive: bool) -> None:
+        self._interactive = interactive
+        self._open = False
+        self._whole_lines = False
+
+    def start_call(self, prefix: str) -> None:
+        """Show *prefix* for a call that is about to run (cursor left mid-line)."""
+        if not (self._interactive and prefix) or self._whole_lines:
+            return
+        if self._open:
+            # Parallel batch: the open prefix could never receive its detail.
+            _clear_streamed_lines(1)
+            self._open = False
+            self._whole_lines = True
+            return
+        click.echo(f"\n{prefix}", nl=False)
+        with contextlib.suppress(OSError):
+            _ = sys.stdout.flush()
+        self._open = True
+
+    def finish_call(self, detail: str) -> bool:
+        """Append *detail* to the open prefix; False when a whole line is needed."""
+        if not (self._interactive and self._open):
+            return False
+        click.echo(f"{detail}\n", nl=False)
+        self._open = False
+        return True
+
+    def close_line(self) -> None:
+        """End a dangling prefix line before other output takes the next line."""
+        if not self._open:
+            return
+        click.echo()
+        self._open = False
+        self._whole_lines = True
+
+    def end_batch(self) -> None:
+        """Reset batch-scoped state once every call of the batch has a result."""
+        self.close_line()
+        self._whole_lines = False
+
+
 def print_markdown(text: str, *, label: str = "assistant:") -> None:
     """Render *text* as markdown via Rich; *label* on its own line when set."""
     from rich.console import Console
@@ -490,10 +603,15 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
     content source changes (reasoning ↔ assistant, or tools/errors), the
     assistant markdown buffer is flushed so streams do not mix and Rich can
     re-render completed assistant segments.
+
+    Pretty tool calls print their ``* Verb 'target' `` prefix as soon as the call
+    starts, so a blocking tool is visible while it runs; the outcome is appended
+    when the result lands (same bytes as the whole-line fallback).
     """
     show_full = get_verbose_tool_results() if verbose is None else verbose
     use_markdown = get_markdown_enabled() if markdown is None else markdown
     pretty = bool(use_markdown and markdown_render_available())
+    tool_lines = _PrettyToolStream(interactive=interactive_terminal() and not show_full)
 
     source: StreamSource | None = None
     assistant_buf: list[str] = []
@@ -521,6 +639,7 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
             return
         if source == "assistant":
             flush_assistant()
+        tool_lines.close_line()
         click.echo()
         click.secho("reasoning:", fg="bright_black")
         source = "reasoning"
@@ -533,6 +652,7 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
         if source == "reasoning":
             click.echo()  # end reasoning stream line
             source = None
+        tool_lines.close_line()
         click.echo()
         click.secho("assistant:", fg="cyan")
         source = "assistant"
@@ -554,26 +674,37 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
                 args = call.function.arguments
                 pretty_tool = name in _PRETTY_TOOLS
                 pending_tools.append((name, args, pretty_tool))
-                if not pretty_tool:
+                if pretty_tool:
+                    tool_lines.start_call(_PRETTY_LINES[name].prefix(args))
+                else:
+                    tool_lines.close_line()
                     preview = _preview(args, _TOOL_ARGS_PREVIEW)
                     click.secho(f"\n[tool] {name}({preview})", fg="yellow")
             else:
+                tool_lines.close_line()
                 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
             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
+            parts = _pretty_parts(name, args, content) if pretty_tool else None
-            if pretty_line is not None and not show_full:
+            if parts is not None and not show_full:
-                click.echo(f"\n{pretty_line}")
+                prefix, detail = parts
+                if not tool_lines.finish_call(detail):
+                    click.echo(f"\n{prefix}{detail}")
             elif show_full:
+                tool_lines.close_line()
                 click.secho(f"[tool ok]\n{content}", fg="magenta")
             else:
+                tool_lines.close_line()
                 preview = _preview_result(content, _TOOL_RESULT_PREVIEW)
                 click.secho(f"[tool ok] {preview}", fg="magenta")
+            if not pending_tools:
+                tool_lines.end_batch()
         elif isinstance(event, ErrorEvent):
             flush_assistant()
+            tool_lines.close_line()
             suffix = ""
             if event.source:
                 suffix += f" source={event.source}"
@@ -582,12 +713,14 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
             click.secho(f"\n[error]{suffix} {event.message}", fg="bright_red")
         elif isinstance(event, CancelledEvent):
             flush_assistant()
+            tool_lines.close_line()
             if event.reason:
                 click.secho(f"\n[cancelled] {event.reason}", fg="yellow")
             else:
                 click.secho("\n[cancelled]", fg="yellow")
         elif isinstance(event, MaxRoundsEvent):
             flush_assistant()
+            tool_lines.close_line()
             if event.continued:
                 click.secho(
                     f"\n[max rounds {event.rounds} reached — continuing with a higher allowance]",
@@ -602,6 +735,7 @@ async def render_events(  # noqa: C901, PLR0912, PLR0915
             _ = event
 
     # End-of-turn: flush any open assistant segment; close reasoning stream.
+    tool_lines.close_line()
     if assistant_buf or printed_assistant:
         flush_assistant()
     elif printed_reasoning:

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, ToolCallEvent, ToolResultEvent
+from plyngent.agent import ErrorEvent, ReasoningDeltaEvent, TextDeltaEvent, ToolCallEvent, ToolResultEvent
 from plyngent.cli.display import (
     get_markdown_enabled,
     markdown_render_available,
@@ -558,3 +558,94 @@ async def test_pretty_wait_error(capsys: pytest.CaptureFixture[str]) -> None:
     await render_events(_aiter([_pretty_call("wait", '{"duration": -1}'), _result(content)]))
     out = capsys.readouterr().out
     assert "* Wait (error: duration must not be negative)" in out
+
+
+async def test_pretty_prefix_shown_while_call_runs(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """An interactive terminal shows the prefix before the tool has a result."""
+    monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
+    while_running: list[str] = []
+
+    async def events() -> AsyncIterator[AgentEvent]:
+        yield _read_call('{"path": "a.txt"}')
+        while_running.append(capsys.readouterr().out)  # tool still running here
+        yield _result("L1-4\none\ntwo\n")
+
+    await render_events(events())
+    assert while_running == ["\n* Read 'a.txt' "]
+    assert capsys.readouterr().out == "L1-4 (done)\n\n"
+
+
+async def test_pretty_prefix_absent_when_not_a_tty(capsys: pytest.CaptureFixture[str]) -> None:
+    """Non-interactive output stays one whole-line write (pipes, logs, tests)."""
+    while_running: list[str] = []
+
+    async def events() -> AsyncIterator[AgentEvent]:
+        yield _read_call('{"path": "a.txt"}')
+        while_running.append(capsys.readouterr().out)
+        yield _result("L1-4\none\ntwo\n")
+
+    await render_events(events())
+    assert while_running == [""]
+    assert capsys.readouterr().out == "\n* Read 'a.txt' L1-4 (done)\n\n"
+
+
+async def test_pretty_verbose_keeps_full_result_output(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Verbose mode stays a full log: no prefix line, whole result."""
+    monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
+    set_verbose_tool_results(True)
+    try:
+        await render_events(_aiter([_read_call('{"path": "a.txt"}'), _result("L1-4\none\n")]))
+    finally:
+        set_verbose_tool_results(False)
+    out = capsys.readouterr().out
+    assert "[tool ok]\nL1-4\none\n" in out
+    assert "* Read 'a.txt' " not in out
+
+
+async def test_pretty_parallel_batch_erases_open_prefix(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """A second call in the batch erases the open prefix; results print whole."""
+    monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
+    await render_events(
+        _aiter(
+            [
+                _pretty_call("read_file", '{"path": "a.txt"}'),
+                _pretty_call("read_file", '{"path": "b.txt"}'),
+                _result("L1-4\none\n"),
+                _result("L1-9\ntwo\n"),
+            ]
+        )
+    )
+    out = capsys.readouterr().out
+    assert "\x1b[2K\x1b[1A" in out  # the in-flight prefix was erased
+    assert "* Read 'a.txt' L1-4 (done)\n" in out
+    assert "* Read 'b.txt' L1-9 (done)\n" in out
+
+
+async def test_pretty_prefix_closed_before_error_line(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Another event ends an open prefix line instead of appending to it."""
+    monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
+    await render_events(
+        _aiter(
+            [
+                _read_call('{"path": "a.txt"}'),
+                ErrorEvent(message="boom", source="tool"),
+                _result("error: tool 'read_file' failed: boom"),
+            ]
+        )
+    )
+    out = capsys.readouterr().out
+    assert "* Read 'a.txt' \n" in out
+    assert "[error] source=tool boom\n" in out
+    assert "* Read 'a.txt' (error)\n" in out