mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
1339 lines
53 KiB
Python
1339 lines
53 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from plyngent.agent import (
|
|
ErrorEvent,
|
|
ReasoningDeltaEvent,
|
|
TextDeltaEvent,
|
|
ToolCallEvent,
|
|
ToolResultEvent,
|
|
TruncatedEvent,
|
|
)
|
|
from plyngent.cli.display import (
|
|
_RUN_BATCH_CMD_LIMIT,
|
|
_clear_streamed_lines,
|
|
_line_count_for_clear,
|
|
_pretty_line_for,
|
|
close_open_pretty_line,
|
|
get_markdown_enabled,
|
|
markdown_render_available,
|
|
print_markdown,
|
|
render_events,
|
|
set_markdown_enabled,
|
|
set_verbose_tool_results,
|
|
)
|
|
from plyngent.lmproto.openai_compatible.model import (
|
|
AssistantFunctionTool,
|
|
AssistantFunctionToolCall,
|
|
ToolChatMessage,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import AsyncIterator
|
|
|
|
import pytest
|
|
|
|
from plyngent.agent import AgentEvent
|
|
|
|
|
|
async def _aiter(events: list[AgentEvent]) -> AsyncIterator[AgentEvent]:
|
|
for event in events:
|
|
yield event
|
|
|
|
|
|
async def test_render_reasoning_and_text(capsys: pytest.CaptureFixture[str]) -> None:
|
|
set_markdown_enabled(False)
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
ReasoningDeltaEvent(content="think"),
|
|
TextDeltaEvent(content="hello"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "reasoning:" in out
|
|
assert "think" in out
|
|
assert "assistant:" in out
|
|
assert "hello" in out
|
|
# Labels on their own lines (content begins after newline).
|
|
assert "assistant:\nhello" in out or "assistant:\r\nhello" in out
|
|
set_markdown_enabled(True)
|
|
|
|
|
|
async def test_flush_markdown_on_source_change(
|
|
capsys: pytest.CaptureFixture[str],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Assistant segment before tools is flushed so later text is a new segment."""
|
|
monkeypatch.setattr("plyngent.cli.display.markdown_render_available", lambda: True)
|
|
set_markdown_enabled(True)
|
|
from plyngent.agent import ToolCallEvent
|
|
from plyngent.lmproto.openai_compatible.model import (
|
|
AssistantFunctionTool,
|
|
AssistantFunctionToolCall,
|
|
)
|
|
|
|
call = AssistantFunctionToolCall(
|
|
id="1",
|
|
function=AssistantFunctionTool(name="acme_ping", arguments='{"host": "example.com"}'),
|
|
)
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
TextDeltaEvent(content="before **tool**"),
|
|
ToolCallEvent(tool_call=call),
|
|
TextDeltaEvent(content="after"),
|
|
]
|
|
),
|
|
markdown=True,
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "[tool]" in out
|
|
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_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)"),
|
|
("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}"
|
|
|
|
|
|
def _tree_call(args_json: str) -> ToolCallEvent:
|
|
return ToolCallEvent(
|
|
tool_call=AssistantFunctionToolCall(
|
|
id="1",
|
|
function=AssistantFunctionTool(name="tree", arguments=args_json),
|
|
)
|
|
)
|
|
|
|
|
|
async def test_pretty_tree_markdown(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "- src/\n - main.py\n - nested/\n - deep.txt\n- a.txt\n"
|
|
await render_events(_aiter([_tree_call('{"path": "."}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Tree '.' (2 dirs, 3 files, depth 3)" in out
|
|
assert "[tool]" not in out
|
|
|
|
|
|
async def test_pretty_tree_flat_omits_depth(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "src/\nsrc/main.py\na.txt\n… (5 more paths not shown)"
|
|
await render_events(_aiter([_tree_call('{"path": ".", "format": "flat"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Tree '.' (1 dirs, 2 files)" in out
|
|
assert "depth" not in out
|
|
|
|
|
|
async def test_pretty_tree_decorated(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "src/\n├── main.py\n└── nested/\n └── deep.txt\n"
|
|
await render_events(_aiter([_tree_call('{"path": "src", "format": "decorated"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Tree 'src' (1 dirs, 2 files, depth 2)" in out
|
|
|
|
|
|
async def test_pretty_tree_error(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_tree_call('{"path": "nope"}'), _result("error: not a directory: nope")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Tree 'nope' (error)" in out
|
|
|
|
|
|
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=[T1]\n"
|
|
"(LIFO of groups: depth=1; TOP group = current breakdown level)\n"
|
|
"group d=0 TOP:\n"
|
|
" [ ] 1. T1"
|
|
)
|
|
await render_events(_aiter([call, _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Todo Push:" in out
|
|
assert " [ ] 1. T1" in out
|
|
assert "[tool]" not in out
|
|
|
|
|
|
async def test_pretty_todo_batch_merges_to_one_stack(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""A batch of todo calls: one summary line, then the last result's stack."""
|
|
stack = "(LIFO of groups: depth=1; TOP group = current breakdown level)\ngroup d=0 TOP:\n [ ] 1. T1"
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("todo_push", '{"titles": ["T1"]}'),
|
|
_pretty_call("todo_update", '{"number": 1, "status": "done"}'),
|
|
_result(f"pushed group (depth=1) items=[T1]\n{stack}"),
|
|
_result(f"updated 1 → done: T1\n{stack}"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Todo x2: Push, Update" in out
|
|
assert "* Todo Push:" not in out
|
|
assert "* Todo Update:" not in out
|
|
# Only the last result is printed, so the stack renders once.
|
|
assert out.count("group d=0 TOP:") == 1
|
|
assert "pushed group" not in out
|
|
assert "updated 1 → done: T1" in out
|
|
|
|
|
|
async def test_pretty_todo_batch_mixed_with_other_tool_stays_per_call(
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""A batch holding a non-todo call keeps one line (and stack) per todo call."""
|
|
stack = "group d=0 TOP:\n [ ] 1. T1"
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("todo_push", '{"titles": ["T1"]}'),
|
|
_pretty_call("listdir", '{"path": "src"}'),
|
|
_result(f"pushed group (depth=1) items=[a1]\n{stack}"),
|
|
_result("dir\tsrc\nfile\tREADME.md\n"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Todo Push:" in out
|
|
assert "group d=0 TOP:" in out
|
|
assert "src" in out
|
|
assert "* Todo x" not in out
|
|
|
|
|
|
async def test_non_pretty_tool_keeps_old_style(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Tools with no pretty line (plugins, unknown names) keep the ``[tool]`` style."""
|
|
call = ToolCallEvent(
|
|
tool_call=AssistantFunctionToolCall(
|
|
id="3",
|
|
function=AssistantFunctionTool(name="acme_ping", arguments='{"host": "example.com"}'),
|
|
)
|
|
)
|
|
await render_events(_aiter([call, _result("pong")]))
|
|
out = capsys.readouterr().out
|
|
assert "[tool] acme_ping(" 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")
|
|
set_verbose_tool_results(False)
|
|
set_markdown_enabled(False)
|
|
await render_events(_aiter([ToolResultEvent(message=msg)]))
|
|
out = capsys.readouterr().out
|
|
assert "…" in out
|
|
assert long not in out
|
|
|
|
set_verbose_tool_results(True)
|
|
await render_events(_aiter([ToolResultEvent(message=msg)]), verbose=True)
|
|
out2 = capsys.readouterr().out
|
|
assert long in out2
|
|
set_verbose_tool_results(False)
|
|
set_markdown_enabled(True)
|
|
|
|
|
|
async def test_markdown_off_keeps_plain_stream(
|
|
capsys: pytest.CaptureFixture[str],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr("plyngent.cli.display.markdown_render_available", lambda: True)
|
|
set_markdown_enabled(False)
|
|
await render_events(_aiter([TextDeltaEvent(content="**bold**")]))
|
|
out = capsys.readouterr().out
|
|
assert "**bold**" in out
|
|
set_markdown_enabled(True)
|
|
|
|
|
|
async def test_markdown_on_replaces_with_rich(
|
|
capsys: pytest.CaptureFixture[str],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr("plyngent.cli.display.markdown_render_available", lambda: True)
|
|
set_markdown_enabled(True)
|
|
await render_events(_aiter([TextDeltaEvent(content="hello **world**")]), markdown=True)
|
|
out = capsys.readouterr().out
|
|
# Rich markdown renders emphasis; raw ** markers should not remain as the sole form.
|
|
assert "assistant:" in out or "assistant: " in out
|
|
assert "world" in out
|
|
|
|
|
|
def test_print_markdown_renders(capsys: pytest.CaptureFixture[str]) -> None:
|
|
print_markdown("# Title\n\n`code`", label="assistant: ")
|
|
out = capsys.readouterr().out
|
|
assert "Title" in out
|
|
assert "code" in out
|
|
|
|
|
|
def test_markdown_flags_roundtrip() -> None:
|
|
set_markdown_enabled(False)
|
|
assert get_markdown_enabled() is False
|
|
set_markdown_enabled(True)
|
|
assert get_markdown_enabled() is True
|
|
|
|
|
|
def test_markdown_render_available_respects_plain_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("PLYNGENT_PLAIN", "1")
|
|
assert markdown_render_available() is False
|
|
monkeypatch.delenv("PLYNGENT_PLAIN", raising=False)
|
|
|
|
|
|
async def test_pretty_verbose_shows_full_result(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Verbose mode prints the full tool result even for pretty tools."""
|
|
await render_events(
|
|
_aiter([_read_call('{"path": "a.txt"}'), _result("L1-4\none\ntwo\n")]),
|
|
verbose=True,
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "[tool ok]" in out
|
|
assert "L1-4\none\ntwo" in out
|
|
assert "* Read" not in out
|
|
|
|
|
|
async def test_pretty_plain_output_has_no_ansi(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Pretty summaries are plain text when stdout is not a TTY."""
|
|
await render_events(_aiter([_read_call('{"path": "a.txt"}'), _result("L1-4\none\n")]))
|
|
out = capsys.readouterr().out
|
|
assert "\x1b[" not in out
|
|
|
|
|
|
async def test_tool_result_preview_first_line_and_count(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Non-verbose preview shows the first line plus a line-count tail."""
|
|
set_markdown_enabled(False)
|
|
content = "first line\nsecond\nthird"
|
|
await render_events(_aiter([ToolResultEvent(message=ToolChatMessage(content=content, tool_call_id="1"))]))
|
|
out = capsys.readouterr().out
|
|
assert "[tool ok] first line (…2 more lines)" in out
|
|
|
|
await render_events(_aiter([ToolResultEvent(message=ToolChatMessage(content="only line", tool_call_id="2"))]))
|
|
out = capsys.readouterr().out
|
|
assert "[tool ok] only line" in out
|
|
assert "more" not in out
|
|
|
|
await render_events(_aiter([ToolResultEvent(message=ToolChatMessage(content="", tool_call_id="3"))]))
|
|
out = capsys.readouterr().out
|
|
assert "[tool ok] (no output)" in out
|
|
set_markdown_enabled(True)
|
|
|
|
|
|
def _pretty_call(name: str, args_json: str) -> ToolCallEvent:
|
|
return ToolCallEvent(
|
|
tool_call=AssistantFunctionToolCall(
|
|
id="1",
|
|
function=AssistantFunctionTool(name=name, arguments=args_json),
|
|
)
|
|
)
|
|
|
|
|
|
async def test_pretty_listdir(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "dir\tsrc\nfile\tREADME.md\nfile\tpyproject.toml\n"
|
|
await render_events(_aiter([_pretty_call("listdir", '{"path": "src"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* List 'src' (1 dirs, 2 files)" in out
|
|
assert "[tool]" not in out
|
|
assert "[tool ok]" not in out
|
|
|
|
|
|
async def test_pretty_listdir_empty(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("listdir", '{"path": "src"}'), _result("(empty)")]))
|
|
out = capsys.readouterr().out
|
|
assert "* List 'src' (empty)" in out
|
|
|
|
|
|
async def test_pretty_listdir_error(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("listdir", '{"path": "nope"}'), _result("error: not a directory: nope")]))
|
|
out = capsys.readouterr().out
|
|
assert "* List 'nope' (error: not a directory: nope)" in out
|
|
|
|
|
|
async def test_pretty_glob(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "src/a.py\nsrc/b.py\n"
|
|
await render_events(_aiter([_pretty_call("glob_paths", '{"pattern": "**/*.py"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Glob '**/*.py' in '.' (2 paths)" in out
|
|
|
|
|
|
async def test_pretty_glob_no_matches(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("glob_paths", '{"pattern": "**/*.rs"}'), _result("(no matches)")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Glob '**/*.rs' in '.' (no matches)" in out
|
|
|
|
|
|
async def test_pretty_glob_truncated(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "a\nb\n...[truncated at 200 matches]"
|
|
await render_events(_aiter([_pretty_call("glob_paths", '{"pattern": "*"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Glob '*' in '.' (2 paths)" in out
|
|
|
|
|
|
async def test_pretty_regex_files(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "src/a.py:3: x = 1\nsrc/a.py:9: x = 2\nsrc/b.py:1: y = 3\n"
|
|
await render_events(_aiter([_pretty_call("regex_files", '{"pattern": "x"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Search 'x' in '.' (3 matches in 2 files)" in out
|
|
|
|
|
|
async def test_pretty_regex_files_no_matches(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("regex_files", '{"pattern": "zzz"}'), _result("(no matches)")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Search 'zzz' in '.' (no matches)" in out
|
|
|
|
|
|
async def test_pretty_run_argv_success(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = (
|
|
"exit_code=0\ntimed_out=false\ncwd=.\ncmd=git status --short\n--- stdout ---\n M src/foo.py\n--- stderr ---\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("run_argv", '{"argv": ["git", "status", "--short"]}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Run $ git status --short (exit code 0)" in out
|
|
assert "[tool]" not in out
|
|
assert "[tool ok]" not in out
|
|
|
|
|
|
async def test_pretty_run_argv_failure(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = (
|
|
"exit_code=1\ntimed_out=false\ncwd=.\ncmd=ls /nope\n--- stdout ---\n--- stderr ---\nls: cannot access '/nope'\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("run_argv", '{"argv": ["ls", "/nope"]}'), _result(result)]))
|
|
out = capsys.readouterr().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)" in out
|
|
|
|
|
|
async def test_pretty_run_argv_error(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(
|
|
_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
|
|
|
|
|
|
async def test_pretty_run_argv_batch_stopped(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = (
|
|
"steps=3 ran=2 stop_on_error=true stopped_early=true\n"
|
|
"--- step 1 ---\nexit_code=1\ntimed_out=false\ncwd=.\ncmd=git status\n"
|
|
"pipe=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n"
|
|
"--- step 2 ---\nexit_code=0\ntimed_out=false\ncwd=.\ncmd=git diff\n"
|
|
"pipe=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n"
|
|
"--- summary ---\nlast_exit=1\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("run_argv_batch", '{"steps": []}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Run batch (stopped early)" in out
|
|
|
|
|
|
async def test_pretty_run_argv_batch_complete(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = (
|
|
"steps=2 ran=2 stop_on_error=true stopped_early=false\n"
|
|
"--- step 1 ---\nexit_code=0\n--- stdout ---\n--- stderr ---\n"
|
|
"--- step 2 ---\nexit_code=0\n--- stdout ---\n--- stderr ---\n"
|
|
"--- summary ---\nlast_exit=0\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("run_argv_batch", '{"steps": []}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Run batch (done)" in out
|
|
|
|
|
|
_BATCH_DONE_RESULT = (
|
|
"steps=2 ran=2 stop_on_error=true stopped_early=false\n"
|
|
"--- step 1 ---\nexit_code=0\n--- stdout ---\n--- stderr ---\n"
|
|
"--- step 2 ---\nexit_code=0\n--- stdout ---\n--- stderr ---\n"
|
|
"--- summary ---\nlast_exit=0\n"
|
|
)
|
|
|
|
|
|
async def test_pretty_run_argv_batch_lists_commands(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""The batch row leads with one ``$ command`` line per step."""
|
|
args = '{"steps": [{"argv": ["git", "status"]}, {"argv": ["git", "diff", "--stat"]}]}'
|
|
await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)]))
|
|
out = capsys.readouterr().out
|
|
assert "$ git status\n$ git diff --stat\n* Run batch (done)" in out
|
|
|
|
|
|
async def test_pretty_run_argv_batch_joins_pipe_chain(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""A pipe chain shares one line; a node it does not feed keeps its own."""
|
|
args = (
|
|
'{"steps": [{"argv": ["grep", "-n", "foo", "a.py"], "pipe": {"argv": ["wc", "-l"]}},'
|
|
' {"argv": ["git", "status"]}]}'
|
|
)
|
|
await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)]))
|
|
out = capsys.readouterr().out
|
|
assert "$ grep -n foo a.py | wc -l\n$ git status\n* Run batch (done)" in out
|
|
|
|
|
|
async def test_pretty_run_argv_batch_indents_branches(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""A ``then`` / ``else`` body sits below the node it follows."""
|
|
args = (
|
|
'{"steps": [{"argv": ["pytest", "-q"], "then": {"argv": ["ruff", "format", "."]},'
|
|
' "else": [{"argv": ["git", "status"]}]}]}'
|
|
)
|
|
await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)]))
|
|
out = capsys.readouterr().out
|
|
assert "$ pytest -q\n then ruff format .\n else git status\n* Run batch (done)" in out
|
|
|
|
|
|
async def test_pretty_run_argv_batch_elides_a_long_command(capsys: pytest.CaptureFixture[str]) -> None:
|
|
long_arg = "x" * 200
|
|
args = f'{{"steps": [{{"argv": ["echo", "{long_arg}"]}}]}}'
|
|
await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)]))
|
|
out = capsys.readouterr().out
|
|
command = f"echo {long_arg}"
|
|
line = next(text for text in out.splitlines() if text.startswith("$ "))
|
|
assert line == f"$ {command[:_RUN_BATCH_CMD_LIMIT]}…"
|
|
|
|
|
|
async def test_pretty_run_argv_batch_steps_as_json_string(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""``steps`` may arrive as a JSON string holding the array."""
|
|
args = '{"steps": "[{\\"argv\\": [\\"git\\", \\"status\\"]}]"}'
|
|
await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)]))
|
|
out = capsys.readouterr().out
|
|
assert "$ git status\n* Run batch (done)" in out
|
|
|
|
|
|
async def test_pretty_run_argv_batch_unreadable_steps_hide_commands(
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("run_argv_batch", '{"steps": "nope"}'),
|
|
_result("error: steps must be a JSON array: Expecting value"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "$ " not in out
|
|
assert "* Run batch (error: steps must be a JSON array" in out
|
|
|
|
|
|
async def test_pretty_fetch(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = (
|
|
"status=200\nmethod=GET\nfinal_url=https://example.com\ncontent_type=application/json\n"
|
|
"body_kind=json\nbytes=1229\ntruncated=false\nredirects=0\nsecurity=public\n--- body ---\n{...}\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("fetch_local", '{"url": "https://example.com"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Fetch GET https://example.com (200)" in out
|
|
assert "[tool]" not in out
|
|
|
|
|
|
async def test_pretty_fetch_error_status(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = (
|
|
"status=404\nmethod=GET\nfinal_url=https://example.com/nope\ncontent_type=text/html\n"
|
|
"body_kind=html\nbytes=45\ntruncated=false\nredirects=0\nsecurity=public\n--- body ---\nNot found\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("fetch_local", '{"url": "https://example.com/nope"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Fetch GET https://example.com/nope (404)" in out
|
|
|
|
|
|
async def test_pretty_fetch_truncated(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = (
|
|
"status=200\nmethod=GET\nfinal_url=https://example.com\ncontent_type=text/plain\n"
|
|
"body_kind=text\nbytes=50000\ntruncated=true\nredirects=0\nsecurity=public\n--- body ---\n..."
|
|
)
|
|
await render_events(_aiter([_pretty_call("fetch_local", '{"url": "https://example.com"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Fetch GET https://example.com (200)" in out
|
|
|
|
|
|
async def test_pretty_fetch_error(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(
|
|
_aiter(
|
|
[_pretty_call("fetch_local", '{"url": "https://example.com"}'), _result("error: fetch_local failed: boom")]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Fetch GET https://example.com (error: fetch_local failed: boom)" in out
|
|
|
|
|
|
async def test_pretty_remote_fetch(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "title=Example article\nurl=https://example.com\nchars=1234\ntruncated=false\n--- content ---\n..."
|
|
await render_events(_aiter([_pretty_call("fetch", '{"url": "https://example.com"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Fetch https://example.com (1234 chars)" in out
|
|
|
|
|
|
async def test_pretty_fetch_reads_a_legacy_local_row(capsys: pytest.CaptureFixture[str]) -> None:
|
|
# Rows written before the local tool was renamed carry the local status block.
|
|
result = (
|
|
"status=200\nmethod=GET\nfinal_url=https://example.com\ncontent_type=text/plain\n"
|
|
"body_kind=text\nbytes=12\ntruncated=false\nredirects=0\nsecurity=public\n--- body ---\nhi\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("fetch", '{"url": "https://example.com"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Fetch https://example.com (200)" in out
|
|
|
|
|
|
async def test_pretty_web_search(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = (
|
|
"- CPython free-threading — https://example.com/ft\n"
|
|
" PEP 703 landed in 3.13\n"
|
|
"- Another page — https://example.com/other\n"
|
|
)
|
|
call = _pretty_call("web_search", '{"query": "python 3.14 free-threading"}')
|
|
await render_events(_aiter([call, _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Web Search 'python 3.14 free-threading' (2 hits)" in out
|
|
assert "[tool]" not in out
|
|
assert "[tool ok]" not in out
|
|
|
|
|
|
async def test_pretty_web_search_truncates_the_query(capsys: pytest.CaptureFixture[str]) -> None:
|
|
query = "a" * 80
|
|
await render_events(
|
|
_aiter([_pretty_call("web_search", f'{{"query": "{query}"}}'), _result("- One — https://example.com\n")])
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert f"* Web Search '{'a' * 60}…' (1 hit)" in out
|
|
|
|
|
|
async def test_pretty_web_search_no_results(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = "error: the search found nothing (DeepseekSearchBackend: no results; BingSearchBackend: no results)"
|
|
await render_events(_aiter([_pretty_call("web_search", '{"query": "zzz"}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Web Search 'zzz' (no results)" in out
|
|
|
|
|
|
async def test_pretty_web_search_failing_source(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("web_search", '{"query": "x"}'), _result("error: search HTTP 401")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Web Search 'x' (error: search HTTP 401)" in out
|
|
|
|
|
|
async def test_pretty_mutators(capsys: pytest.CaptureFixture[str]) -> None:
|
|
cases = [
|
|
("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')",
|
|
"* 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",
|
|
"* Edit 'src/a.py' (done)",
|
|
),
|
|
(
|
|
"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)"),
|
|
(
|
|
"delete_path",
|
|
'{"path": "tmp/x.txt"}',
|
|
"deleted file tmp/x.txt (moved to system trash)",
|
|
"* Delete 'tmp/x.txt' (to trash)",
|
|
),
|
|
]
|
|
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", '{"src": "x", "dst": "y"}'), _result("error: source does not exist: x")])
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Copy 'x' → 'y' (error: source does not exist: x)" in out
|
|
|
|
|
|
async def test_pretty_vcs_status_done(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("vcs_status", "{}"), _result("(clean)")]))
|
|
out = capsys.readouterr().out
|
|
assert "* VCS Status (done)" in out
|
|
assert "[tool]" not in out
|
|
|
|
|
|
async def test_pretty_vcs_status_error(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("vcs_status", "{}"),
|
|
_result("error: no supported VCS detected under workspace (currently: git)"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* VCS Status (error: no supported VCS detected under workspace (currently: git))" in out
|
|
|
|
|
|
async def test_pretty_vcs_log_count(capsys: pytest.CaptureFixture[str]) -> None:
|
|
result = (
|
|
"5436457 2026-08-15 worldmozara core/tools: allow edit_lineno append\nabc1234 2026-08-14 worldmozara cli: fix\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("vcs_log", '{"limit": 2}'), _result(result)]))
|
|
out = capsys.readouterr().out
|
|
assert "* VCS Log (2 commits)" in out
|
|
|
|
|
|
async def test_pretty_vcs_log_single(capsys: pytest.CaptureFixture[str]) -> None:
|
|
content = "abc1234 2026-08-14 worldmozara x\n"
|
|
await render_events(_aiter([_pretty_call("vcs_log", '{"limit": 1}'), _result(content)]))
|
|
out = capsys.readouterr().out
|
|
assert "* VCS Log (1 commit)" in out
|
|
|
|
|
|
async def test_pretty_vcs_log_no_commits(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("vcs_log", "{}"), _result("(no commits)")]))
|
|
out = capsys.readouterr().out
|
|
assert "* VCS Log (no commits)" in out
|
|
|
|
|
|
async def test_pretty_wait_waited(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("wait", '{"duration": 5}'), _result("waited 5s")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Wait (5s)" in out
|
|
assert "[tool]" not in out
|
|
|
|
|
|
async def test_pretty_wait_disturbed(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("wait", '{"duration": 30}'), _result("disturbed by user: ship it")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Wait (disturbed)" in out
|
|
|
|
|
|
async def test_pretty_wait_cancelled(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Ctrl+C during the wait prompt reads as cancelled, not disturbed."""
|
|
await render_events(_aiter([_pretty_call("wait", '{"duration": 30}'), _result("cancelled by user")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Wait (cancelled)" in out
|
|
|
|
|
|
async def test_pretty_wait_error(capsys: pytest.CaptureFixture[str]) -> None:
|
|
content = "error: duration must not be negative"
|
|
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_pty_open_and_read(capsys: pytest.CaptureFixture[str]) -> None:
|
|
opened = "session_id=3\nalive=true\nexit_code=\ncmd=ls -la"
|
|
await render_events(_aiter([_pretty_call("open_pty", '{"command": ["ls", "-la"]}'), _result(opened)]))
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Open $ ls -la (session 3)" in out
|
|
|
|
read = (
|
|
"session_id=3\nalive=true\nexit_code=\nmatched=false\ntruncated=false\n"
|
|
"budget_exhausted=false\n--- data ---\nhello\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("read_pty", '{"session_id": 3}'), _result(read)]))
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Read 3 (alive, 1 line)" in out
|
|
|
|
|
|
async def test_pretty_pty_read_exited_and_matched(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""An exited session reports the code; ``until`` reports that it matched."""
|
|
exited = (
|
|
"session_id=3\nalive=false\nexit_code=0\nmatched=false\ntruncated=false\n"
|
|
"budget_exhausted=false\n--- data ---\nbye\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("read_pty", '{"session_id": 3}'), _result(exited)]))
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Read 3 (exited 0)" in out
|
|
|
|
matched = (
|
|
"session_id=3\nalive=true\nexit_code=\nmatched=true\ntruncated=false\n"
|
|
"budget_exhausted=false\n--- data ---\nprompt\n$ \n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("read_pty", '{"session_id": 3, "until": "$ "}'), _result(matched)]))
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Read 3 (matched, 2 lines)" in out
|
|
|
|
|
|
async def test_pretty_pty_read_empty_and_error(capsys: pytest.CaptureFixture[str]) -> None:
|
|
empty = (
|
|
"session_id=3\nalive=true\nexit_code=\nmatched=false\ntruncated=false\nbudget_exhausted=false\n--- data ---\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("read_pty", '{"session_id": 3}'), _result(empty)]))
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Read 3 (alive, no output)" in out
|
|
|
|
await render_events(
|
|
_aiter([_pretty_call("read_pty", '{"session_id": 9}'), _result("error: unknown PTY session 9")])
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Read 9 (error: unknown PTY session 9)" in out
|
|
|
|
|
|
async def test_pretty_pty_write_and_keys(capsys: pytest.CaptureFixture[str]) -> None:
|
|
status = "session_id=3\nalive=true\nexit_code=\nwrote=5"
|
|
await render_events(_aiter([_pretty_call("write_pty", '{"session_id": 3, "data": "ls -la"}'), _result(status)]))
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Write 3 'ls -la' (5 chars)" in out
|
|
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("write_pty_keys", '{"session_id": 3, "data": "<c-c><enter>"}'),
|
|
_result("session_id=3\nalive=true\nexit_code=\nwrote=2"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Keys 3 '<c-c><enter>' (2 chars)" in out
|
|
|
|
|
|
async def test_pretty_pty_write_escapes_and_shortens_the_payload(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""A newline or an ESC must not break the line, and a long payload is cut."""
|
|
status = "session_id=3\nalive=true\nexit_code=\nwrote=4"
|
|
await render_events(_aiter([_pretty_call("write_pty", '{"session_id": 3, "data": "a\\nb"}'), _result(status)]))
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Write 3 'a\\nb' (4 chars)" in out
|
|
|
|
long = "x" * 60
|
|
call = _pretty_call("write_pty", f'{{"session_id": 3, "data": "{long}"}}')
|
|
await render_events(_aiter([call, _result(status)]))
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Write 3 '" + "x" * 40 + "…'" in out
|
|
assert "x" * 41 not in out
|
|
|
|
|
|
async def test_pretty_ask_into_pty_never_shows_the_answer(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""The result only says the answer was written; the payload stays local."""
|
|
status = "session_id=3\nalive=true\nexit_code=\nwrote=true\nsecret=true\nsubmit=true\nsource=human"
|
|
await render_events(
|
|
_aiter([_pretty_call("ask_into_pty", '{"session_id": 3, "message": "sudo password"}'), _result(status)])
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Ask 3 (answer sent, secret)" in out
|
|
assert "wrote=true" not in out
|
|
|
|
|
|
async def test_pretty_close_pty(capsys: pytest.CaptureFixture[str]) -> None:
|
|
closed = "session_id=3\nclosed=true\nalive=false\nexit_code=0\nmessage=closed"
|
|
await render_events(_aiter([_pretty_call("close_pty", '{"session_id": 3}'), _result(closed)]))
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Close 3 (closed)" in out
|
|
|
|
already = "session_id=3\nclosed=false\nalive=false\nexit_code=0\nmessage=already closed"
|
|
await render_events(_aiter([_pretty_call("close_pty", '{"session_id": 3}'), _result(already)]))
|
|
out = capsys.readouterr().out
|
|
assert "* PTY Close 3 (already closed)" in out
|
|
|
|
|
|
async def test_pretty_ask_user_line_and_choice(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("ask_user_line", '{"question": "Which port?"}'), _result("8080")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Ask 'Which port?' (answered)" in out
|
|
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("ask_user_choice", '{"question": "Pick one", "options": ["alpha", "beta"]}'),
|
|
_result("beta"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Ask Choose 'Pick one' (beta)" in out
|
|
|
|
|
|
async def test_pretty_ask_user_form_counts_answers(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("ask_user_form", '{"title": "Setup", "fields": [{"name": "port", "prompt": "Port?"}]}'),
|
|
_result('{"port": 8080, "host": "localhost"}'),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Ask Form 'Setup' (2 answers)" in out
|
|
|
|
|
|
async def test_pretty_ask_user_error(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("ask_user_line", '{"question": "Port?"}'),
|
|
_result("error: interactive prompts are unavailable (stdin is not a TTY)"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Ask 'Port?' (error: interactive prompts are unavailable (stdin is not a TTY))" in out
|
|
|
|
|
|
async def test_pretty_request_directory_access(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("request_directory_access", '{"path": "/data", "mode": "write"}'),
|
|
_result("access granted: /data (write, session)\nnote: the path denylist still applies"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Access '/data' (write) (granted)" in out
|
|
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("request_directory_access", '{"path": "/data"}'),
|
|
_result("already accessible: /data is inside the workspace"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Access '/data' (read) (already accessible)" in out
|
|
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("request_directory_access", '{"path": "/root"}'),
|
|
_result("error: access to /root denied (declined or timed out after 30s)"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "(error: access to /root denied (declined or timed out after 30s))" in out
|
|
|
|
|
|
async def test_pretty_new_temporary_workspace(capsys: pytest.CaptureFixture[str]) -> None:
|
|
content = "temporary_workspace=/tmp/plyngent-ws-abc\nnote: project workspace unchanged"
|
|
await render_events(_aiter([_pretty_call("new_temporary_workspace", '{"prefix": "ws"}'), _result(content)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Temp Dir (/tmp/plyngent-ws-abc)" in out
|
|
|
|
|
|
async def test_pretty_get_truncated(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("get_truncated", '{"token": "abc"}'), _result("L81-160\nmore\n")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Resume Truncated L81-160 (done)" in out
|
|
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("get_truncated", '{"token": "abc"}'),
|
|
_result("error: truncate token expired (no longer in memory); re-run the tool"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "(error: truncate token expired (no longer in memory); re-run the tool)" in out
|
|
|
|
|
|
async def test_pretty_vcs_kind_branch_and_diff(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(_aiter([_pretty_call("vcs_kind", "{}"), _result("git")]))
|
|
out = capsys.readouterr().out
|
|
assert "* VCS Kind (git)" in out
|
|
|
|
await render_events(_aiter([_pretty_call("vcs_branch", "{}"), _result("main")]))
|
|
out = capsys.readouterr().out
|
|
assert "* VCS Branch (main)" in out
|
|
|
|
diff = (
|
|
"diff --git a/x.py b/x.py\nindex 111..222 100644\n--- a/x.py\n+++ b/x.py\n@@ -1 +1,2 @@\n-old\n+new\n+extra\n"
|
|
)
|
|
await render_events(_aiter([_pretty_call("vcs_diff", '{"path": ""}'), _result(diff)]))
|
|
out = capsys.readouterr().out
|
|
assert "* VCS Diff (1 file, +2/-1)" in out
|
|
|
|
await render_events(_aiter([_pretty_call("vcs_diff", '{"staged": true}'), _result("(no diff)")]))
|
|
out = capsys.readouterr().out
|
|
assert "* VCS Diff (staged) (no diff)" in out
|
|
|
|
|
|
async def test_pretty_todo_list_pop_and_clear(capsys: pytest.CaptureFixture[str]) -> None:
|
|
stack = " [ ] 1. T1\n [x] 2. T2"
|
|
await render_events(_aiter([_pretty_call("todo_list", "{}"), _result(stack)]))
|
|
out = capsys.readouterr().out
|
|
assert "* Todo List:" in out
|
|
assert " [ ] 1. T1" in out
|
|
|
|
await render_events(_aiter([_pretty_call("todo_pop", "{}"), _result(f"popped TOP group (T1)\n{stack}")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Todo Pop:" in out
|
|
|
|
await render_events(_aiter([_pretty_call("todo_clear", "{}"), _result("cleared 3 item(s)")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Todo Clear:" in out
|
|
assert "cleared 3 item(s)" in out
|
|
|
|
|
|
async def test_pretty_mcp_tools_use_one_generic_line(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Namespaced MCP tools render as ``* MCP <server>:<tool> ``."""
|
|
await render_events(_aiter([_pretty_call("mcp__docs__search", '{"query": "x"}'), _result("found 1 page")]))
|
|
out = capsys.readouterr().out
|
|
assert "* MCP docs:search (done)" in out
|
|
assert "[tool]" not in out
|
|
|
|
await render_events(_aiter([_pretty_call("mcp__docs__search", '{"query": "x"}'), _result("error: server died")]))
|
|
out = capsys.readouterr().out
|
|
assert "* MCP docs:search (error: server died)" in out
|
|
|
|
|
|
def test_every_builtin_tool_has_a_pretty_line() -> None:
|
|
"""A new builtin tool must ship a renderer, or it silently loses the syntax."""
|
|
from plyngent.tools.catalog import default_tool_definitions
|
|
from plyngent.tools.net.search import WEB_SEARCH_TOOL_NAME
|
|
|
|
names = [d.name for d in default_tool_definitions()]
|
|
# The CLI host composes ``web_search`` over the builtins, so a catalog walk
|
|
# on its own would let a renderer-less tool ship.
|
|
names.append(WEB_SEARCH_TOOL_NAME)
|
|
missing = [name for name in names if _pretty_line_for(name) is None]
|
|
assert missing == []
|
|
|
|
|
|
def test_a_confirmed_result_keeps_its_pretty_detail() -> None:
|
|
"""The approval line leads the model's text; a pretty detail reads past it."""
|
|
from plyngent.cli.display import _pretty_line_for, _pretty_parts
|
|
|
|
line = _pretty_line_for("write_file")
|
|
assert line is not None
|
|
approved = "[approval] user (once)\nwrote 120 characters to a.txt"
|
|
_prefix, detail = _pretty_parts(line, '{"path": "a.txt", "overwrite": true}', approved)
|
|
assert "(120 chars)" in detail
|
|
|
|
|
|
async def test_pretty_skill_tools(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(
|
|
_aiter(
|
|
[_pretty_call("skill_list", "{}"), _result("skills: 3\npdf-processing [plyngent-user] /x — Extract PDFs")]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Skills (3 skills)" in out
|
|
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("skill_read", '{"name": "pdf-processing"}'),
|
|
_result("L1-40\n# PDF processing\n"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Skill 'pdf-processing' L1-40 (done)" in out
|
|
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("skill_read", '{"name": "pdf-processing", "file": "scripts/run.sh"}'),
|
|
_result("L1-3\n#!/bin/sh\n"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Skill 'pdf-processing' scripts/run.sh L1-3 (done)" in out
|
|
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("skill_search", '{"pattern": "pdftotext"}'),
|
|
_result("alpha/SKILL.md:3: Use pdftotext.\nbeta/notes.md:1: pdftotext again"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Skill Search 'pdftotext' (2 matches in 2 skills)" in out
|
|
|
|
await render_events(_aiter([_pretty_call("skill_search", '{"pattern": "nope"}'), _result("(no matches)")]))
|
|
out = capsys.readouterr().out
|
|
assert "* Skill Search 'nope' (no matches)" in out
|
|
|
|
|
|
async def test_pretty_skill_create_and_edit(capsys: pytest.CaptureFixture[str]) -> None:
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("skill_create", '{"name": "acme", "description": "d", "body": "b"}'),
|
|
_result("created skill 'acme' at /skills/acme\nnotes: SKILL.md has no frontmatter description"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Skill Create 'acme' (created) — SKILL.md has no frontmatter description" in out
|
|
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("skill_edit", '{"name": "acme"}'),
|
|
_result("updated skill 'acme' (SKILL.md: replaced 1 of 1 occurrence(s))"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Skill Edit 'acme' (SKILL.md: replaced 1 of 1 occurrence(s))" in out
|
|
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("skill_edit", '{"name": "acme", "file": "scripts/run.sh"}'),
|
|
_result("error: skill write to /skills/acme denied (declined or timed out after 30s)"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert "* Skill Edit 'acme' scripts/run.sh (error: skill write to /skills/acme denied" 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_gives_every_call_its_own_row(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""A second call in the batch holds its prefix until the first row is done."""
|
|
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 out == ("\n* Read 'a.txt' L1-4 (done)\n\n* Read 'b.txt' L1-9 (done)\n\n")
|
|
assert "\x1b[" not in out # the cursor is never moved
|
|
|
|
|
|
async def test_pretty_edit_batch_prints_one_row_per_edit(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""Two edits in one round: one finished row each, no abandoned prefix row."""
|
|
monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
_pretty_call("edit_replace", '{"path": "a.py"}'),
|
|
_pretty_call("edit_replace", '{"path": "b.py"}'),
|
|
_result("replaced 1 occurrence in a.py ('x' → 'y')"),
|
|
_result("replaced 1 occurrence in b.py ('x' → 'y')"),
|
|
]
|
|
)
|
|
)
|
|
out = capsys.readouterr().out
|
|
assert out == "\n* Edit 'a.py' (done)\n\n* Edit 'b.py' (done)\n\n"
|
|
|
|
|
|
async def test_pretty_parallel_batch_keeps_the_line_above_the_prefix(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""Rendering a batch must not touch the line above the prefix.
|
|
|
|
Reasoning streams without a trailing newline, so the prefix line sits right
|
|
below the reasoning text; moving the cursor up used to wipe that reasoning
|
|
line out of the screen.
|
|
"""
|
|
monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
ReasoningDeltaEvent(content="think"),
|
|
_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 out == ("\nreasoning:\nthink\n* Read 'a.txt' L1-4 (done)\n\n* Read 'b.txt' L1-9 (done)\n\n\n")
|
|
assert "\x1b[" not in out
|
|
|
|
|
|
async def test_pretty_line_closed_by_out_of_band_output(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""A prompt/confirm printed while the call runs ends the open prefix line.
|
|
|
|
The prompt backend runs :func:`close_open_pretty_line` before writing, so its
|
|
box starts on a fresh line and the result then prints whole.
|
|
"""
|
|
monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
|
|
|
|
async def _events():
|
|
yield _pretty_call("read_file", '{"path": "a.txt"}')
|
|
close_open_pretty_line() # what a confirm box does mid-call
|
|
yield _result("L1-4\none\n")
|
|
|
|
await render_events(_events())
|
|
out = capsys.readouterr().out
|
|
assert out == ("\n* Read 'a.txt' \n\n* Read 'a.txt' L1-4 (done)\n\n")
|
|
assert "\x1b[" not in out
|
|
|
|
|
|
def test_clear_streamed_lines_clears_exactly_the_given_rows(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""Two rows means two clears; the cursor lands on the topmost cleared row."""
|
|
_clear_streamed_lines(2)
|
|
assert capsys.readouterr().out == "\r\x1b[2K\x1b[1A\r\x1b[2K"
|
|
|
|
|
|
def test_clear_streamed_lines_ignores_non_positive_counts(capsys: pytest.CaptureFixture[str]) -> None:
|
|
_clear_streamed_lines(0)
|
|
_clear_streamed_lines(-1)
|
|
assert capsys.readouterr().out == ""
|
|
|
|
|
|
def test_line_count_for_clear_counts_wrapped_rows(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A streamed line wider than the terminal owns more than one row."""
|
|
monkeypatch.setenv("COLUMNS", "20")
|
|
assert _line_count_for_clear("assistant:", "x" * 20) == 2 # label + full row
|
|
assert _line_count_for_clear("assistant:", "x" * 21) == 3 # label + two rows
|
|
assert _line_count_for_clear("assistant:", "a\nb") == 3
|
|
assert _line_count_for_clear("", "") == 0
|
|
|
|
|
|
async def test_flush_markdown_erases_wrapped_assistant_rows(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""A wrapped assistant segment is fully erased before the markdown re-render."""
|
|
monkeypatch.setattr("plyngent.cli.display.interactive_terminal", lambda: True)
|
|
monkeypatch.setattr("plyngent.cli.display.markdown_render_available", lambda: True)
|
|
monkeypatch.setenv("COLUMNS", "20")
|
|
set_markdown_enabled(True)
|
|
try:
|
|
await render_events(
|
|
_aiter(
|
|
[
|
|
TextDeltaEvent(content="x" * 21),
|
|
_pretty_call("read_file", '{"path": "a.txt"}'),
|
|
_result("L1-4\none\n"),
|
|
]
|
|
)
|
|
)
|
|
finally:
|
|
set_markdown_enabled(True)
|
|
out = capsys.readouterr().out
|
|
# label + two wrapped body rows + the blank separator above the label.
|
|
assert out.count("\x1b[2K") == 4
|
|
assert out.count("\x1b[1A") == 3
|
|
|
|
|
|
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
|
|
|
|
|
|
async def test_render_truncated_round(capsys: pytest.CaptureFixture[str]) -> None:
|
|
"""A round cut off at the output limit says so, and how the turn goes on."""
|
|
await render_events(_aiter([TruncatedEvent(kept_chars=42, continued=True)]))
|
|
assert "[limit] output truncated (42 chars kept); continuing\n" in capsys.readouterr().out
|
|
|
|
await render_events(_aiter([TruncatedEvent(kept_chars=42, continued=False)]))
|
|
assert "[limit] output truncated (42 chars kept); /retry continues from here\n" in capsys.readouterr().out
|