feat(cli,tui): issue #12 — presenter contract semantics amendment (v0.2.0)
Replaces the stateless _render_event / _render_event_to_log helpers with stateful per-turn presenters (CliPresenterState / TuiPresenterState). Coalesces thinking-event deltas into a single growing display per run; demotes telemetry events with editorial hierarchy; formats duration + usage for human reading. Headline behavior change: a 50-token thinking phase now renders as ONE coalesced growing line in CLI (or one closed RichLog entry + per-delta live Static widget in TUI), not 50 lines of [thinking] spam. Editorial promotion line (issue #12 INV-002): - Load-bearing (no demotion prefix): Text, Done, Error, Cancelled - Demoted telemetry (`. ` ASCII prefix in CLI; dim `· ` in TUI): WorkerPhase, Thinking, TextBoundary, ToolStart, ToolResult Stateful coalescing: - Thinking deltas accumulate into thinking_buffer; first non-thinking event closes the run with a single \n boundary in CLI / one closed dim RichLog entry in TUI. - TUI adds a dedicated Static(id="thinking-current") widget that shows the last ~200 chars of the active run, mirroring per-delta updates. Two-views-of-thinking decoupling per INV-004: chronological RichLog + always-visible widget. - CLI INV-005: when stdout text was streamed mid-line, text_written_since_newline triggers a stdout flush + \n before the next stderr terminal label — guarantees [done] / [error] / [cancelled] land on their own line in a TTY without breaking pipe-to-file scripted consumers. Formatting helpers (issue #12 INV-006 / INV-007): - _format_duration_ms — autoscale `347ms` / `5.5s` / `1.2m` - _format_usage — natural-language `6756 in -> 126 out (6882 total, 0 cached)` with arrow="->" CLI / "→" TUI Cross-frontier design pass (eitri-smithy-dev, althing 01KSBE52YZR5E3SPTKA672JE43) returned 16-of-16 confirmed decisions + 4 material divergences applied: - ASCII `. ` prefix in CLI (`·` is U+00B7, not ASCII) - RichLog one-closed-entry-per-run + Static per-delta updates (not inline-mirror as initially proposed) - presenter-state object instead of pure-function rendering - Framed as "contract semantics amendment", not "polish" Volva paraphrase round (5 prose-precision fixes applied to 12.contract.md): INV-001 "growing display" semantics; single hide mechanism for the Static widget (Textual reactive `display: bool`); [render_error] security clause (type-only, no exception message); text_written_since_newline `\n`-terminated text corner case; [create_session] integration path (bypasses state.render — not an SSE Event variant). Volva code-review round (5 findings applied): - F1 drift: render-exception fallback now writes BOTH a plain-label fallback line for the original event AND the `[render_error] <type>` line (was missing the fallback half). - F2 drift: dim Rich style applied to all demoted-telemetry RichLog writes via `rich.text.Text(..., style="dim")` (was plain str). - F3 drift: belt-and-braces widget clear+hide on EVERY terminal event (Done/Error/Cancelled), even when thinking_open was False. - F4 precision: _format_usage gains PRE-001 assertion on the four expected usage keys. - F5 precision: _run_turn signature amended in issue #3 contract to document the new `state: CliPresenterState | None = None` test- injection kwarg. [create_session] lifecycle line demoted to `. create_session:` (written directly by _amain; bypasses state.render since it's not a wire-level SSE Event variant). Pre-amendment _render_event / _render_event_to_log and their test classes removed under the no-backwards-compat rule. Issues #3 and #4 contracts amended in-place: #3 (CliPresenterState CLASS + FN block + helper FN blocks + _run_turn signature + _amain create_session demotion); #4 (TuiPresenterState CLASS + FN block + compose Static widget + _stream_turn_worker state construction). 209 tests GREEN; ruff clean. Bumps v0.1.0 → v0.2.0 (minor — output shape change breaks pre-amendment grep patterns like `[thinking] '`; no public API surface change beyond the rendering contract). Persistent-memory commit-along: captures the issue #12 decision, forward direction (require end_user_id for every access — declined worldtree-dev's requires_end_user_id offer because we'll send it universally), and the Heimdall scope-model foot-gun note (the "per-Tier-1-agent scope add" diagnosis was a phantom ask resolved by worldtree-dev's correction; agent.call:* baseline covers all Tier 1).
This commit is contained in:
+364
-113
@@ -16,7 +16,6 @@ from ratatoskr.cli import (
|
||||
_AuthError,
|
||||
_cancel_and_log,
|
||||
_parse_args,
|
||||
_render_event,
|
||||
_run_turn,
|
||||
main,
|
||||
)
|
||||
@@ -237,132 +236,378 @@ class TestParseArgs:
|
||||
SID = SseId(42, 5)
|
||||
|
||||
|
||||
class TestRenderEvent:
|
||||
def test_text_to_stdout_only(self) -> None:
|
||||
"""text_to_stdout_only [happy,tracer]: Text → stdout=="hello"; stderr empty; flushed."""
|
||||
stdout = _FlushCountingIO()
|
||||
stderr = io.StringIO()
|
||||
_render_event(Text(sse_id=SID, content="hello"), stdout=stdout, stderr=stderr)
|
||||
assert stdout.getvalue() == "hello"
|
||||
assert stderr.getvalue() == ""
|
||||
assert stdout.flush_count == 1 # INV-010: per-chunk flush
|
||||
# Issue #12 — presenter contract semantics amendment.
|
||||
# CliPresenterState replaces the stateless _render_event with a stateful per-turn
|
||||
# presenter that coalesces thinking runs and demotes telemetry events.
|
||||
# (Pre-amendment TestRenderEvent class and `_render_event` function have been
|
||||
# removed under the project's no-backwards-compatibility rule.)
|
||||
|
||||
def test_done_writes_newline_and_label(self) -> None:
|
||||
"""done_writes_newline_and_label: stdout=="\\n" (flushed); stderr "[done]" labels."""
|
||||
stdout = _FlushCountingIO()
|
||||
SID42 = SseId(42, 1)
|
||||
|
||||
|
||||
class TestCliPresenterState:
|
||||
"""Tests for the new CliPresenterState — per issue #12 contract."""
|
||||
|
||||
def test_thinking_coalesce_single_run(self) -> None:
|
||||
"""thinking_coalesce_single_run [happy,tracer]:
|
||||
Thinking("hello") + Thinking(" world") + Done →
|
||||
stderr has ". thinking: hello world\\n" followed by the [done] line.
|
||||
"""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
evt = Done(
|
||||
sse_id=SID,
|
||||
phase="completed",
|
||||
state = CliPresenterState()
|
||||
state.render(Thinking(sse_id=SID42, content="hello"), stdout=stdout, stderr=stderr)
|
||||
# After first delta: stderr has the open prefix + content, no \n yet.
|
||||
assert stderr.getvalue() == ". thinking: hello"
|
||||
state.render(Thinking(sse_id=SID42, content=" world"), stdout=stdout, stderr=stderr)
|
||||
# After second delta: still the same growing logical line, still no \n.
|
||||
assert stderr.getvalue() == ". thinking: hello world"
|
||||
# Now a Done event closes the thinking run with \n then writes the terminal label.
|
||||
done = Done(
|
||||
sse_id=SID42,
|
||||
phase="succeeded",
|
||||
response="hi",
|
||||
model="glm5-turbo",
|
||||
duration_ms=1234,
|
||||
usage={"prompt": 10, "completion": 5},
|
||||
model="m",
|
||||
duration_ms=1,
|
||||
usage={
|
||||
"prompt_tokens": 0,
|
||||
"completion_tokens": 0,
|
||||
"total_tokens": 0,
|
||||
"cached_input_tokens": 0,
|
||||
},
|
||||
)
|
||||
_render_event(evt, stdout=stdout, stderr=stderr)
|
||||
assert stdout.getvalue() == "\n"
|
||||
assert stdout.flush_count == 1 # INV-010: post-Done newline flushed
|
||||
out_err = stderr.getvalue()
|
||||
assert out_err.startswith("[done]")
|
||||
assert "turn_id=42" in out_err
|
||||
assert "model=glm5-turbo" in out_err
|
||||
assert "duration_ms=1234" in out_err
|
||||
|
||||
def test_error_to_stderr_only(self) -> None:
|
||||
"""error_to_stderr_only: Error → stderr "[error]" with code; stdout empty."""
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
evt = Error(sse_id=SID, phase="failed", message="boom", error_code="llm_output_invalid")
|
||||
_render_event(evt, stdout=stdout, stderr=stderr)
|
||||
state.render(done, stdout=stdout, stderr=stderr)
|
||||
captured = stderr.getvalue()
|
||||
# Thinking run closed with \n; terminal label landed; no demotion prefix on [done].
|
||||
assert captured.startswith(". thinking: hello world\n")
|
||||
assert "[done]" in captured
|
||||
# stdout untouched (no Text events were rendered)
|
||||
assert stdout.getvalue() == ""
|
||||
out_err = stderr.getvalue()
|
||||
assert out_err.startswith("[error]")
|
||||
assert "turn_id=42" in out_err
|
||||
assert "code=llm_output_invalid" in out_err
|
||||
|
||||
def test_cancelled_to_stderr_only(self) -> None:
|
||||
"""cancelled_to_stderr_only: Cancelled → stderr "[cancelled]" + reason + partial id."""
|
||||
stdout = io.StringIO()
|
||||
def test_thinking_closes_on_first_non_thinking_event(self) -> None:
|
||||
"""thinking_closes_on_first_non_thinking_event [happy]:
|
||||
Thinking → WorkerPhase → stderr has ". thinking: ...\\n" then ". worker_phase: ..."
|
||||
"""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
stderr = io.StringIO()
|
||||
evt = Cancelled(
|
||||
sse_id=SID, phase="cancelled", turn_id=42, reason="user", partial_message_id=7
|
||||
state = CliPresenterState()
|
||||
state.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=stderr)
|
||||
state.render(
|
||||
WorkerPhase(sse_id=SID42, phase="streaming", turn_id=42),
|
||||
stdout=io.StringIO(),
|
||||
stderr=stderr,
|
||||
)
|
||||
_render_event(evt, stdout=stdout, stderr=stderr)
|
||||
assert stdout.getvalue() == ""
|
||||
out_err = stderr.getvalue()
|
||||
assert out_err.startswith("[cancelled]")
|
||||
assert "reason='user'" in out_err
|
||||
assert "partial_message_id=7" in out_err
|
||||
out = stderr.getvalue()
|
||||
# Thinking run closed; worker_phase rendered with demotion prefix.
|
||||
assert ". thinking: x\n" in out
|
||||
assert ". worker_phase:" in out
|
||||
# `[worker_phase]` (bracketed, pre-amendment shape) MUST NOT appear.
|
||||
assert "[worker_phase]" not in out
|
||||
|
||||
def test_thinking_closes_on_error(self) -> None:
|
||||
"""thinking_closes_on_error [error]:
|
||||
Thinking → Error → thinking line closes with \\n, then [error] line rendered
|
||||
(partial thinking content is NOT discarded — observability requirement).
|
||||
"""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
stderr = io.StringIO()
|
||||
state = CliPresenterState()
|
||||
state.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=stderr)
|
||||
state.render(
|
||||
Error(sse_id=SID42, phase="failed", message="boom", error_code="bad"),
|
||||
stdout=io.StringIO(),
|
||||
stderr=stderr,
|
||||
)
|
||||
out = stderr.getvalue()
|
||||
# Partial thinking preserved with closing \n; error rendered without demotion prefix.
|
||||
assert ". thinking: x\n" in out
|
||||
assert "[error]" in out
|
||||
# Demotion prefix MUST NOT precede [error]: it's load-bearing.
|
||||
assert ". [error]" not in out
|
||||
|
||||
def test_cancelled_mid_thinking(self) -> None:
|
||||
"""cancelled_mid_thinking [scenario]: Thinking → Cancelled → thinking closes with \\n;
|
||||
then [cancelled] (no demotion prefix, partial thinking preserved).
|
||||
"""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
stderr = io.StringIO()
|
||||
state = CliPresenterState()
|
||||
state.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=stderr)
|
||||
state.render(
|
||||
Cancelled(
|
||||
sse_id=SID42, phase="cancelled", turn_id=42, reason="user", partial_message_id=None
|
||||
),
|
||||
stdout=io.StringIO(),
|
||||
stderr=stderr,
|
||||
)
|
||||
out = stderr.getvalue()
|
||||
assert ". thinking: x\n" in out
|
||||
assert "[cancelled]" in out
|
||||
assert ". [cancelled]" not in out
|
||||
|
||||
def test_text_then_done_newline_boundary(self) -> None:
|
||||
"""text_then_done_newline_boundary [trace]: Text("answer") → Done;
|
||||
stdout receives "answer\\n" (the \\n is the INV-005 boundary), stderr has "[done] ...".
|
||||
"""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
def test_worker_phase_to_stderr(self) -> None:
|
||||
"""worker_phase_to_stderr: WorkerPhase → stderr "[worker_phase]"; stdout empty."""
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
evt = WorkerPhase(sse_id=SID, phase="streaming", turn_id=42)
|
||||
_render_event(evt, stdout=stdout, stderr=stderr)
|
||||
assert stdout.getvalue() == ""
|
||||
assert stderr.getvalue().startswith("[worker_phase]")
|
||||
state = CliPresenterState()
|
||||
state.render(Text(sse_id=SID42, content="answer"), stdout=stdout, stderr=stderr)
|
||||
state.render(_make_done(), stdout=stdout, stderr=stderr)
|
||||
# INV-005: text without trailing \n → exactly one \n gets injected before terminal label
|
||||
assert stdout.getvalue() == "answer\n"
|
||||
assert "[done]" in stderr.getvalue()
|
||||
|
||||
def test_no_text_then_done_no_extra_newline(self) -> None:
|
||||
"""no_text_then_done_no_extra_newline [trace]: Done with no preceding Text →
|
||||
stdout untouched; stderr receives only "[done] ...".
|
||||
"""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
def test_thinking_truncated(self) -> None:
|
||||
"""thinking_truncated [trace]: …"""
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
_render_event(Thinking(sse_id=SID, content="a" * 500), stdout=stdout, stderr=stderr)
|
||||
out_err = stderr.getvalue()
|
||||
assert out_err.startswith("[thinking]")
|
||||
assert "a" * 500 not in out_err
|
||||
assert "a" * 200 in out_err
|
||||
state = CliPresenterState()
|
||||
state.render(_make_done(), stdout=stdout, stderr=stderr)
|
||||
# INV-005 boundary fires ONLY when text was written; no text → no \n injection.
|
||||
assert stdout.getvalue() == ""
|
||||
assert "[done]" in stderr.getvalue()
|
||||
|
||||
def test_newline_terminated_text_then_done(self) -> None:
|
||||
"""newline_terminated_text_then_done [trace]: Text("answer\\n") → Done;
|
||||
stdout receives "answer\\n" exactly ONCE (no double-newline before [done]).
|
||||
Tests the F4 Volva fix: text_written_since_newline tracks last-char-was-\\n.
|
||||
"""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
def test_tool_start_to_stderr(self) -> None:
|
||||
"""tool_start_to_stderr: ToolStart → stderr "[tool_start] name=... args=..."."""
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
evt = ToolStart(sse_id=SID, name="read_file", arguments={"path": "/x"})
|
||||
_render_event(evt, stdout=stdout, stderr=stderr)
|
||||
assert stdout.getvalue() == ""
|
||||
out_err = stderr.getvalue()
|
||||
assert out_err.startswith("[tool_start] name=read_file args=")
|
||||
state = CliPresenterState()
|
||||
state.render(Text(sse_id=SID42, content="answer\n"), stdout=stdout, stderr=stderr)
|
||||
state.render(_make_done(), stdout=stdout, stderr=stderr)
|
||||
# POST-003: content ends with \n → state.text_written_since_newline = False
|
||||
# → INV-005 does NOT inject an extra \n before [done].
|
||||
assert stdout.getvalue() == "answer\n"
|
||||
|
||||
def test_multiple_thinking_runs(self) -> None:
|
||||
"""multiple_thinking_runs [scenario]: Thinking → Text → Thinking → Done →
|
||||
TWO separate ". thinking: ..." runs in stderr; stdout has the text + INV-005 boundary.
|
||||
"""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
state = CliPresenterState()
|
||||
state.render(Thinking(sse_id=SID42, content="first"), stdout=stdout, stderr=stderr)
|
||||
state.render(Text(sse_id=SID42, content="answer"), stdout=stdout, stderr=stderr)
|
||||
state.render(Thinking(sse_id=SID42, content="second"), stdout=stdout, stderr=stderr)
|
||||
state.render(_make_done(), stdout=stdout, stderr=stderr)
|
||||
err = stderr.getvalue()
|
||||
# Each thinking RUN gets its own ". thinking: " prefix.
|
||||
assert err.count(". thinking: ") == 2
|
||||
assert ". thinking: first" in err
|
||||
assert ". thinking: second" in err
|
||||
assert stdout.getvalue() == "answer\n"
|
||||
assert "[done]" in err
|
||||
|
||||
|
||||
def test_tool_start_demoted(self) -> None:
|
||||
"""tool_start_demoted [trace]: ToolStart → stderr line starts with ". tool_start:" """
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
stderr = io.StringIO()
|
||||
state = CliPresenterState()
|
||||
state.render(
|
||||
ToolStart(sse_id=SID42, name="read_file", arguments={"path": "/x"}),
|
||||
stdout=io.StringIO(),
|
||||
stderr=stderr,
|
||||
)
|
||||
line = stderr.getvalue()
|
||||
assert line.startswith(". tool_start:")
|
||||
assert "[tool_start]" not in line
|
||||
|
||||
def test_tool_result_truncated(self) -> None:
|
||||
"""tool_result_truncated [trace]: ToolResult.result repr truncated to ≤200 chars."""
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
evt = ToolResult(sse_id=SID, name="x", result="b" * 500, duration_ms=42)
|
||||
_render_event(evt, stdout=stdout, stderr=stderr)
|
||||
out_err = stderr.getvalue()
|
||||
assert out_err.startswith("[tool_result]")
|
||||
# the contract uses `{event.result!r:.200}` — 200 chars max of repr output
|
||||
assert "b" * 500 not in out_err
|
||||
"""tool_result_truncated [trace]: long result repr truncates to ≤200 chars."""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
def test_text_boundary_to_stderr(self) -> None:
|
||||
"""text_boundary_to_stderr: TextBoundary → stderr "[text_boundary]"; stdout empty."""
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
evt = TextBoundary(sse_id=SID, kind="sentence", char_offset=128, ts="2026-05-21T00:00:00Z")
|
||||
_render_event(evt, stdout=stdout, stderr=stderr)
|
||||
assert stdout.getvalue() == ""
|
||||
out_err = stderr.getvalue()
|
||||
assert out_err.startswith("[text_boundary]")
|
||||
assert "kind=sentence" in out_err
|
||||
assert "char_offset=128" in out_err
|
||||
state = CliPresenterState()
|
||||
state.render(
|
||||
ToolResult(sse_id=SID42, name="x", result="b" * 500, duration_ms=42),
|
||||
stdout=io.StringIO(),
|
||||
stderr=stderr,
|
||||
)
|
||||
line = stderr.getvalue()
|
||||
assert line.startswith(". tool_result:")
|
||||
# Full 500-char result MUST NOT fit; truncation applied.
|
||||
assert "b" * 500 not in line
|
||||
|
||||
def test_invariant_inv003_stderr_only(self) -> None:
|
||||
"""invariant_inv003_stderr_only [scenario]: …"""
|
||||
for evt in [
|
||||
WorkerPhase(sse_id=SID, phase="x", turn_id=42),
|
||||
Thinking(sse_id=SID, content="x"),
|
||||
TextBoundary(sse_id=SID, kind="x", char_offset=0, ts="t"),
|
||||
ToolStart(sse_id=SID, name="x", arguments={}),
|
||||
ToolResult(sse_id=SID, name="x", result=None, duration_ms=0),
|
||||
Error(sse_id=SID, phase="failed", message="m", error_code="e"),
|
||||
Cancelled(
|
||||
sse_id=SID, phase="cancelled", turn_id=42, reason="r", partial_message_id=None
|
||||
def test_text_boundary_demoted(self) -> None:
|
||||
"""text_boundary_demoted [trace]: TextBoundary → stderr ". text_boundary:" prefix."""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
stderr = io.StringIO()
|
||||
state = CliPresenterState()
|
||||
state.render(
|
||||
TextBoundary(sse_id=SID42, kind="sentence", char_offset=128, ts="2026-05-21T00:00:00Z"),
|
||||
stdout=io.StringIO(),
|
||||
stderr=stderr,
|
||||
)
|
||||
line = stderr.getvalue()
|
||||
assert line.startswith(". text_boundary:")
|
||||
assert "[text_boundary]" not in line
|
||||
|
||||
def test_duration_format_seconds(self) -> None:
|
||||
"""duration_format_seconds [trace]: Done(duration_ms=5467) → "duration=5.5s"."""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
stderr = io.StringIO()
|
||||
state = CliPresenterState()
|
||||
state.render(_make_done(duration_ms=5467), stdout=io.StringIO(), stderr=stderr)
|
||||
assert "duration=5.5s" in stderr.getvalue()
|
||||
assert "duration_ms=5467" not in stderr.getvalue()
|
||||
|
||||
def test_duration_format_subsecond(self) -> None:
|
||||
"""duration_format_subsecond [trace]: Done(duration_ms=347) → "duration=347ms"."""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
stderr = io.StringIO()
|
||||
state = CliPresenterState()
|
||||
state.render(_make_done(duration_ms=347), stdout=io.StringIO(), stderr=stderr)
|
||||
assert "duration=347ms" in stderr.getvalue()
|
||||
|
||||
def test_duration_format_minutes(self) -> None:
|
||||
"""duration_format_minutes [trace]: Done(duration_ms=72000) → "duration=1.2m"."""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
stderr = io.StringIO()
|
||||
state = CliPresenterState()
|
||||
state.render(_make_done(duration_ms=72000), stdout=io.StringIO(), stderr=stderr)
|
||||
assert "duration=1.2m" in stderr.getvalue()
|
||||
|
||||
def test_usage_format_ascii_arrow(self) -> None:
|
||||
"""usage_format_ascii_arrow [trace]: stderr label contains the natural-language
|
||||
usage shape with ASCII arrow (-> not →) for CLI scriptability.
|
||||
"""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
stderr = io.StringIO()
|
||||
state = CliPresenterState()
|
||||
state.render(
|
||||
_make_done(
|
||||
usage={
|
||||
"prompt_tokens": 6756,
|
||||
"completion_tokens": 126,
|
||||
"total_tokens": 6882,
|
||||
"cached_input_tokens": 0,
|
||||
}
|
||||
),
|
||||
]:
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
_render_event(evt, stdout=stdout, stderr=stderr)
|
||||
assert stdout.getvalue() == "", f"INV-002 violated for {type(evt).__name__}"
|
||||
stdout=io.StringIO(),
|
||||
stderr=stderr,
|
||||
)
|
||||
out = stderr.getvalue()
|
||||
assert "usage 6756 in -> 126 out (6882 total, 0 cached)" in out
|
||||
# Raw dict shape MUST NOT leak through.
|
||||
assert "'prompt_tokens'" not in out
|
||||
|
||||
def test_state_reset_per_amain(self) -> None:
|
||||
"""state_reset_per_amain [trace]: fresh CliPresenterState() starts with no thinking open."""
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
# Simulate two _amain calls by constructing two independent states.
|
||||
s1 = CliPresenterState()
|
||||
s2 = CliPresenterState()
|
||||
# Run thinking into s1 — it should NOT bleed into s2.
|
||||
s1.render(Thinking(sse_id=SID42, content="x"), stdout=io.StringIO(), stderr=io.StringIO())
|
||||
assert s1.thinking_open is True
|
||||
assert s2.thinking_open is False
|
||||
# s2's first render produces its own ". thinking: " prefix.
|
||||
e2 = io.StringIO()
|
||||
s2.render(Thinking(sse_id=SID42, content="y"), stdout=io.StringIO(), stderr=e2)
|
||||
assert e2.getvalue() == ". thinking: y"
|
||||
|
||||
|
||||
class TestFormatDurationMs:
|
||||
"""Unit tests for _format_duration_ms per INV-006."""
|
||||
|
||||
def test_subsecond(self) -> None:
|
||||
from ratatoskr.cli import _format_duration_ms
|
||||
assert _format_duration_ms(347) == "347ms"
|
||||
|
||||
def test_exact_one_second(self) -> None:
|
||||
from ratatoskr.cli import _format_duration_ms
|
||||
assert _format_duration_ms(1000) == "1.0s"
|
||||
|
||||
def test_fractional_seconds(self) -> None:
|
||||
from ratatoskr.cli import _format_duration_ms
|
||||
assert _format_duration_ms(5467) == "5.5s"
|
||||
|
||||
def test_exact_one_minute(self) -> None:
|
||||
from ratatoskr.cli import _format_duration_ms
|
||||
assert _format_duration_ms(60000) == "1.0m"
|
||||
|
||||
def test_fractional_minutes(self) -> None:
|
||||
from ratatoskr.cli import _format_duration_ms
|
||||
assert _format_duration_ms(72000) == "1.2m"
|
||||
|
||||
def test_zero(self) -> None:
|
||||
from ratatoskr.cli import _format_duration_ms
|
||||
assert _format_duration_ms(0) == "0ms"
|
||||
|
||||
|
||||
class TestFormatUsage:
|
||||
"""Unit tests for _format_usage per INV-007."""
|
||||
|
||||
def test_ascii_arrow(self) -> None:
|
||||
from ratatoskr.cli import _format_usage
|
||||
usage = {
|
||||
"prompt_tokens": 6756,
|
||||
"completion_tokens": 126,
|
||||
"total_tokens": 6882,
|
||||
"cached_input_tokens": 0,
|
||||
}
|
||||
assert (
|
||||
_format_usage(usage, arrow="->")
|
||||
== "6756 in -> 126 out (6882 total, 0 cached)"
|
||||
)
|
||||
|
||||
def test_unicode_arrow(self) -> None:
|
||||
from ratatoskr.cli import _format_usage
|
||||
usage = {
|
||||
"prompt_tokens": 6756,
|
||||
"completion_tokens": 126,
|
||||
"total_tokens": 6882,
|
||||
"cached_input_tokens": 0,
|
||||
}
|
||||
assert (
|
||||
_format_usage(usage, arrow="→")
|
||||
== "6756 in → 126 out (6882 total, 0 cached)"
|
||||
)
|
||||
|
||||
|
||||
_USAGE_ZERO: dict[str, int] = {
|
||||
"prompt_tokens": 0,
|
||||
"completion_tokens": 0,
|
||||
"total_tokens": 0,
|
||||
"cached_input_tokens": 0,
|
||||
}
|
||||
|
||||
|
||||
def _make_done(*, duration_ms: int = 1, usage: dict[str, int] | None = None) -> Done:
|
||||
return Done(
|
||||
sse_id=SID42,
|
||||
phase="succeeded",
|
||||
response="r",
|
||||
model="m",
|
||||
duration_ms=duration_ms,
|
||||
usage=usage if usage is not None else _USAGE_ZERO,
|
||||
)
|
||||
|
||||
|
||||
class TestCancelAndLog:
|
||||
@@ -813,7 +1058,11 @@ class TestRunTurn:
|
||||
|
||||
@respx.mock
|
||||
async def test_render_called_once_per_event(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""render_called_once_per_event [trace]: spy → _render_event call_count == event count."""
|
||||
"""render_called_once_per_event [trace]: spy on render; call_count == event count.
|
||||
|
||||
Per issue #12: rendering went from stateless `_render_event` to
|
||||
`CliPresenterState.render`; the spy moves accordingly.
|
||||
"""
|
||||
chunks = [
|
||||
_sse_chunk("42:1", {"type": "worker_phase", "phase": "streaming", "turn_id": 42}),
|
||||
_sse_chunk("42:2", {"type": "text", "content": "hi"}),
|
||||
@@ -825,17 +1074,17 @@ class TestRunTurn:
|
||||
)
|
||||
)
|
||||
|
||||
from ratatoskr.cli import CliPresenterState
|
||||
|
||||
call_count = 0
|
||||
from ratatoskr import cli as cli_mod
|
||||
original = CliPresenterState.render
|
||||
|
||||
original = cli_mod._render_event
|
||||
|
||||
def spy(event, **kw): # type: ignore[no-untyped-def]
|
||||
def spy(self, event, **kw): # type: ignore[no-untyped-def]
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
return original(event, **kw)
|
||||
return original(self, event, **kw)
|
||||
|
||||
monkeypatch.setattr(cli_mod, "_render_event", spy)
|
||||
monkeypatch.setattr(CliPresenterState, "render", spy)
|
||||
|
||||
sigint = asyncio.Event()
|
||||
stdout, stderr = io.StringIO(), io.StringIO()
|
||||
@@ -921,9 +1170,11 @@ class TestAmain:
|
||||
assert exit_code == 0
|
||||
captured = capsys.readouterr()
|
||||
err = captured.err
|
||||
assert "[create_session]" in err
|
||||
# Per issue #12: [create_session] lifecycle line demoted to `. create_session:`.
|
||||
assert ". create_session:" in err
|
||||
assert "[create_session]" not in err # pre-amendment shape forbidden
|
||||
assert "[done]" in err
|
||||
assert err.index("[create_session]") < err.index("[done]")
|
||||
assert err.index(". create_session:") < err.index("[done]")
|
||||
|
||||
@respx.mock
|
||||
async def test_happy_existing_session(self, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
|
||||
Reference in New Issue
Block a user