2 Commits

4 changed files with 90 additions and 12 deletions
+4 -1
View File
@@ -154,7 +154,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' (done)``."""
"""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 _pretty_line(f"* Read '{path}' ", ("(file not found)", "red"))
@@ -162,6 +162,9 @@ 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"))
+29 -10
View File
@@ -1,20 +1,44 @@
from __future__ import annotations
import asyncio
import re
import sys
from click import style
from plyngent.agent import ToolTag, tool
from plyngent.prompting import get_prompt_backend, read_line_with_timeout, run_prompt_async
WAIT_PROMPT = "Waiting for {duration}s. Press Enter to disturb. Feel free to ship any reason before pressing Enter. "
_ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;]*m")
def _wait_prompt(duration: int, *, reason: str | None) -> str:
"""Two-line wait prompt: status line, then the optional-reason input prompt.
``read_line_with_timeout`` writes the prompt verbatim to stdout (no click
echo ANSI stripping), so styling is applied here and stripped when stdout
is not a TTY. On a terminal: cyan status, dimmed detail, yellow input
prompt; plain text otherwise (tests, redirected output).
"""
status = style(f"Waiting for {duration}s", fg="cyan")
if reason:
status += style(f" ({reason})", fg="bright_black")
status += style(". Press Enter to disturb.", fg="bright_black")
input_prompt = style("Reason (optional): ", fg="yellow")
rendered = f"{status}\n{input_prompt}"
if not sys.stdout.isatty():
return _ANSI_ESCAPE_RE.sub("", rendered)
return rendered
@tool(name="wait", tags=ToolTag.LOCAL | ToolTag.READ_ONLY)
async def wait(duration: int, *, reason: str | None = None) -> str:
"""Wait ``duration`` seconds before continuing.
Interactive sessions show a line prompt: pressing Enter (optionally after
typing a reason) "disturbs" the wait so the turn continues immediately.
Non-interactive runs simply sleep the full duration.
Interactive sessions show a two-line prompt: a status line, then an
optional-reason input — pressing Enter (optionally after typing a reason)
"disturbs" the wait so the turn continues immediately. Non-interactive runs
simply sleep the full duration.
"""
if duration < 0:
return "error: duration must not be negative"
@@ -24,12 +48,7 @@ async def wait(duration: int, *, reason: str | None = None) -> str:
if not backend.is_interactive():
await asyncio.sleep(duration)
return f"waited {duration}s"
prompt = WAIT_PROMPT.format(duration=duration)
if reason:
prompt = (
f"Waiting for {duration}s (reason: {reason}). Press Enter to disturb. "
"Feel free to ship any reason before pressing Enter. "
)
prompt = _wait_prompt(duration, reason=reason)
line = await run_prompt_async(read_line_with_timeout, prompt, float(duration))
if line is None:
return f"waited {duration}s"
+9 -1
View File
@@ -98,11 +98,19 @@ 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' (done)" in 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_numbered_no_range(capsys: pytest.CaptureFixture[str]) -> None:
"""Numbered reads carry no L header, so the summary falls back to ``(done)``."""
content = " 1|one\n 2|two\n"
await render_events(_aiter([_read_call('{"path": "a.txt", "with_lineno": true}'), _result(content)]))
out = capsys.readouterr().out
assert "* Read 'a.txt' (done)" in out
async def test_pretty_read_file_statuses(capsys: pytest.CaptureFixture[str]) -> None:
cases = [
("error: file not found: a.txt", "(file not found)"),
+48
View File
@@ -144,3 +144,51 @@ async def test_wait_tool_times_out(monkeypatch) -> None:
registry = ToolRegistry([wait])
out = await registry.execute("wait", '{"duration": 3}')
assert out == "waited 3s"
async def test_wait_tool_prompt_two_lines_with_reason(monkeypatch) -> None:
prompts: list[str] = []
monkeypatch.setattr(
wait_module,
"read_line_with_timeout",
lambda prompt, timeout: prompts.append(prompt) or None,
)
with temporary_backend(ScriptedBackend([])):
registry = ToolRegistry([wait])
out = await registry.execute("wait", '{"duration": 5, "reason": "tests are slow"}')
assert out == "waited 5s"
assert prompts == ["Waiting for 5s (tests are slow). Press Enter to disturb.\nReason (optional): "]
async def test_wait_tool_prompt_two_lines_no_reason(monkeypatch) -> None:
prompts: list[str] = []
monkeypatch.setattr(
wait_module,
"read_line_with_timeout",
lambda prompt, timeout: prompts.append(prompt) or None,
)
with temporary_backend(ScriptedBackend([])):
registry = ToolRegistry([wait])
out = await registry.execute("wait", '{"duration": 5}')
assert out == "waited 5s"
assert prompts == ["Waiting for 5s. Press Enter to disturb.\nReason (optional): "]
def test_wait_prompt_plain_when_not_a_tty() -> None:
"""Prompt styling is stripped when stdout is not a TTY (pytest capture)."""
prompt = wait_module._wait_prompt(5, reason="tests")
assert "\x1b[" not in prompt
assert prompt == "Waiting for 5s (tests). Press Enter to disturb.\nReason (optional): "
def test_wait_prompt_colored_on_tty(monkeypatch) -> None:
"""On a TTY the prompt is styled: cyan status, yellow input prompt."""
class _FakeStdout:
def isatty(self) -> bool:
return True
monkeypatch.setattr(wait_module.sys, "stdout", _FakeStdout())
prompt = wait_module._wait_prompt(5, reason="tests")
assert "\x1b[36m" in prompt # cyan status
assert "\x1b[33m" in prompt # yellow input prompt