wehub-resource-sync
562 lines
21 KiB
Python
562 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from opensquilla.cli.chat.turn import UsageSummary
|
|
from opensquilla.cli.tui.backend.render_summary import summarize_args, summarize_result
|
|
from opensquilla.cli.tui.opentui.renderer import OpenTuiStreamRenderer, _format_tokens
|
|
from opensquilla.engine.usage import SessionTotalsSnapshot
|
|
|
|
|
|
def test_web_search_args_render_query_summary() -> None:
|
|
assert summarize_args("web_search", {"query": "OpenSquilla canonical search"}) == (
|
|
"OpenSquilla canonical search"
|
|
)
|
|
assert summarize_args("web_discover", {"query": "OpenSquilla discover links"}) == (
|
|
"OpenSquilla discover links"
|
|
)
|
|
|
|
|
|
class _RecordingHandle:
|
|
def __init__(self) -> None:
|
|
self.sent: list[tuple[str, dict]] = []
|
|
|
|
async def send_message(self, message_type: str, payload: dict) -> None:
|
|
self.sent.append((message_type, payload))
|
|
|
|
|
|
class _ToolbarRecordingHandle(_RecordingHandle):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.toolbar: dict[str, object] = {}
|
|
self.toolbar_updates: list[tuple[str, object | None]] = []
|
|
self.invalidated = 0
|
|
|
|
def set_toolbar(self, key: str, value: object | None) -> None:
|
|
self.toolbar_updates.append((key, value))
|
|
if value is None:
|
|
self.toolbar.pop(key, None)
|
|
return
|
|
self.toolbar[key] = value
|
|
|
|
def invalidate(self) -> None:
|
|
self.invalidated += 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_intermediate_text_is_thinking_final_text_is_answer_card() -> None:
|
|
"""Intermediate narration before a tool (presentation="intermediate") opens
|
|
a purple thinking block; the final answer (presentation="answer") opens a
|
|
cyan answer card. Each is the right kind from its first delta — no retype."""
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("Let me check", presentation="intermediate")
|
|
await r.atool_start("web_search", {"query": "x"}, "c1")
|
|
await r.atool_finished("c1", success=True, result="result line")
|
|
await r.aappend_text("Final answer", presentation="answer")
|
|
await r.afinalize(None)
|
|
|
|
assert not [t for t, _ in handle.sent if t == "block.retype"]
|
|
begins = [
|
|
(p["kind"], p["id"])
|
|
for t, p in handle.sent
|
|
if t == "block.begin" and p.get("kind") in {"thinking", "answer"}
|
|
]
|
|
# intermediate -> thinking block, final -> answer card, in that order
|
|
assert [kind for kind, _id in begins] == ["thinking", "answer"]
|
|
tool_begins = [p for t, p in handle.sent if t == "block.begin" and p.get("kind") == "tool"]
|
|
assert tool_begins and tool_begins[0]["meta"]["name"] == "web_search"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_final_answer_is_a_card_from_the_first_delta() -> None:
|
|
"""A pure-answer turn opens an answer card on the very first delta and
|
|
streams into it — never a thinking block, never a retype."""
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("The ", presentation="answer")
|
|
await r.aappend_text("answer.", presentation="answer")
|
|
await r.afinalize(None)
|
|
|
|
assert not [t for t, _ in handle.sent if t == "block.retype"]
|
|
answer_begins = [p for t, p in handle.sent if t == "block.begin" and p.get("kind") == "answer"]
|
|
assert len(answer_begins) == 1
|
|
assert not [
|
|
p for t, p in handle.sent if t == "block.begin" and p.get("kind") == "thinking"
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reasoning_streams_into_its_own_block_and_closes_before_text() -> None:
|
|
"""Reasoning (the model's extended-thinking process) streams live into a
|
|
dedicated 'reasoning' block — the host shows a dim rolling peek and
|
|
collapses it to 'Thought for Ns' on block.end — and the block closes
|
|
before the answer text opens so the two never interleave."""
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_reasoning("let me think ")
|
|
await r.aappend_reasoning("step by step about the internals")
|
|
await r.aappend_text("the answer")
|
|
await r.afinalize(None)
|
|
|
|
# the reasoning block is its own kind, distinct from intermediate "thinking"
|
|
# text and from the "answer" card
|
|
reasoning_begins = [
|
|
p for t, p in handle.sent if t == "block.begin" and p.get("kind") == "reasoning"
|
|
]
|
|
assert len(reasoning_begins) == 1
|
|
reasoning_id = reasoning_begins[0]["id"]
|
|
# every reasoning delta streams into that block (the host renders the peek)
|
|
reasoning_appends = [
|
|
p["delta"] for t, p in handle.sent if t == "block.append" and p["id"] == reasoning_id
|
|
]
|
|
assert reasoning_appends == ["let me think ", "step by step about the internals"]
|
|
# the reasoning block closes before the answer block opens
|
|
events = [
|
|
(t, p.get("id"), p.get("kind"))
|
|
for t, p in handle.sent
|
|
if t in ("block.begin", "block.end")
|
|
]
|
|
end_reasoning = events.index(("block.end", reasoning_id, None))
|
|
begin_answer = next(
|
|
i for i, (t, _i, kind) in enumerate(events) if t == "block.begin" and kind == "answer"
|
|
)
|
|
assert end_reasoning < begin_answer
|
|
assert not [t for t, _ in handle.sent if t == "block.retype"]
|
|
# the reasoning marker is closed before the answer block opens
|
|
ends = [p["id"] for t, p in handle.sent if t == "block.end"]
|
|
assert reasoning_id in ends
|
|
answer_begins = [
|
|
p for t, p in handle.sent if t == "block.begin" and p.get("kind") == "answer"
|
|
]
|
|
assert len(answer_begins) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_answer_only_turn_has_no_retype() -> None:
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("Direct answer")
|
|
await r.afinalize(None)
|
|
assert not [t for t, _ in handle.sent if t == "block.retype"]
|
|
answer_begins = [p for t, p in handle.sent if t == "block.begin" and p.get("kind") == "answer"]
|
|
assert len(answer_begins) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_renderer_marks_tool_error_and_cancel() -> None:
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.atool_start("grep", {"pattern": "x"}, "c2")
|
|
await r.atool_finished("c2", success=False, error="boom")
|
|
await r.aerror("turn-level failure")
|
|
await r.afinalize(None, cancelled=True)
|
|
updates = [p for t, p in handle.sent if t == "block.update"]
|
|
assert any(p["patch"].get("status") == "error" for p in updates)
|
|
error_begins = [p for t, p in handle.sent if t == "block.begin" and p.get("kind") == "error"]
|
|
assert error_begins and error_begins[0]["meta"]["text"] == "turn-level failure"
|
|
end = [p for t, p in handle.sent if t == "turn.end"][0]
|
|
assert end["cancelled"] is True
|
|
# the failed tool's detail line was appended into the tool block
|
|
appends = [p for t, p in handle.sent if t == "block.append" and p["id"] == "c2"]
|
|
assert any("boom" in p["delta"] for p in appends)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_midtool_closes_open_tool_block() -> None:
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.atool_start("grep", {"pattern": "x"}, "c9")
|
|
# NO atool_finished — simulate cancellation
|
|
await r.afinalize(None, cancelled=True)
|
|
# the open tool must be force-closed: an error update + an end for its id
|
|
updates = [p for t, p in handle.sent if t == "block.update" and p["id"] == "c9"]
|
|
ends = [p for t, p in handle.sent if t == "block.end" and p["id"] == "c9"]
|
|
assert updates and updates[-1]["patch"].get("status") == "error"
|
|
assert ends, "cancelled in-flight tool block was never closed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_aclose_without_finalize_tears_down_errored_turn() -> None:
|
|
"""Error paths end the turn without afinalize; the guaranteed aclose must
|
|
still end the turn, idle the pill, and re-enable the composer so the UI
|
|
never stays busy and the next turn never merges into the errored card."""
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("partial answer")
|
|
await r.aerror("provider exploded")
|
|
await r.aclose()
|
|
|
|
types = [t for t, _ in handle.sent]
|
|
assert "turn.end" in types
|
|
statuses = [p for t, p in handle.sent if t == "turn.status"]
|
|
assert statuses[-1]["phase"] == "idle"
|
|
assert statuses[-1]["active"] is False
|
|
composer_sets = [p for t, p in handle.sent if t == "composer.set"]
|
|
assert composer_sets[-1] == {"disabled": False}
|
|
# the open answer block was force-closed
|
|
begins = {p["id"] for t, p in handle.sent if t == "block.begin" and p.get("kind") == "answer"}
|
|
ends = {p["id"] for t, p in handle.sent if t == "block.end"}
|
|
assert begins <= ends
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_aclose_force_closes_inflight_tool_block() -> None:
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.atool_start("grep", {"pattern": "x"}, "c7")
|
|
# NO atool_finished, NO afinalize — e.g. a provider timeout mid-tool
|
|
await r.aclose()
|
|
|
|
updates = [p for t, p in handle.sent if t == "block.update" and p["id"] == "c7"]
|
|
ends = [p for t, p in handle.sent if t == "block.end" and p["id"] == "c7"]
|
|
assert updates and updates[-1]["patch"].get("status") == "error"
|
|
assert ends
|
|
assert [t for t, _ in handle.sent if t == "turn.end"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_aclose_after_afinalize_emits_no_second_teardown() -> None:
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("done")
|
|
await r.afinalize(None)
|
|
await r.aclose()
|
|
|
|
assert len([t for t, _ in handle.sent if t == "turn.end"]) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_aclose_before_any_output_is_a_noop() -> None:
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aclose()
|
|
assert handle.sent == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_aclose_tolerates_dead_output_handle() -> None:
|
|
class _DeadHandle:
|
|
async def send_message(self, message_type: str, payload: dict) -> None:
|
|
raise RuntimeError("OpenTUI bridge is not started")
|
|
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("partial")
|
|
# the bridge dies before teardown; aclose must not raise from its emits
|
|
r.output_handle = _DeadHandle()
|
|
await r.aclose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_pill_returns_to_output_when_text_resumes_after_tool() -> None:
|
|
"""In the narrate-then-act flow the pill must not stay stuck on the
|
|
finished tool's name while the final answer streams."""
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("Let me look", presentation="intermediate")
|
|
await r.atool_start("grep", {"pattern": "x"}, "c1")
|
|
await r.atool_finished("c1", success=True, result="hit")
|
|
await r.aappend_text("Final answer", presentation="answer")
|
|
await r.afinalize(None)
|
|
|
|
phases = [p["phase"] for t, p in handle.sent if t == "turn.status"]
|
|
assert phases == ["thinking", "output", "tool", "output", "idle"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_astatus_updates_pill_and_renders_dim_status_line() -> None:
|
|
"""Status messages (artifact saved, task-group progress) must be visible:
|
|
a transient pill label plus a dim in-card status line."""
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("working")
|
|
await r.astatus("artifact written: report.md")
|
|
await r.aappend_text(" more")
|
|
await r.afinalize(None)
|
|
|
|
status_blocks = [
|
|
p for t, p in handle.sent if t == "block.begin" and p.get("kind") == "status"
|
|
]
|
|
assert status_blocks and status_blocks[0]["meta"]["text"] == "artifact written: report.md"
|
|
assert status_blocks[0]["meta"]["style"] == "dim"
|
|
ends = {p["id"] for t, p in handle.sent if t == "block.end"}
|
|
assert status_blocks[0]["id"] in ends
|
|
|
|
labels = [(p["phase"], p["label"]) for t, p in handle.sent if t == "turn.status"]
|
|
assert ("output", "artifact written: report.md") in labels
|
|
# the pill label is transient: the next text delta restores the phase label
|
|
status_index = labels.index(("output", "artifact written: report.md"))
|
|
assert ("output", "output") in labels[status_index + 1 :]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_astatus_ignores_blank_messages() -> None:
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.astatus(" ")
|
|
assert not [p for t, p in handle.sent if t == "block.begin"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_usage_block_emitted_before_turn_end() -> None:
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("done")
|
|
await r.afinalize(None)
|
|
types = [t for t, _ in handle.sent]
|
|
usage_begin = next(
|
|
i
|
|
for i, (t, p) in enumerate(handle.sent)
|
|
if t == "block.begin" and p.get("kind") == "usage"
|
|
)
|
|
turn_end = types.index("turn.end")
|
|
assert usage_begin < turn_end, "usage block must render in the active turn, before turn.end"
|
|
# answer card still closes (its block.end) before usage
|
|
answer_end = next(
|
|
i
|
|
for i, (t, p) in enumerate(handle.sent)
|
|
if t == "block.end" and p["id"].endswith("-b1")
|
|
)
|
|
assert answer_end < usage_begin
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anonymous_tools_each_close_independently() -> None:
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.atool_start("a", {}, None)
|
|
await r.atool_finished(None, success=True, result="ra")
|
|
await r.atool_start("b", {}, None)
|
|
await r.atool_finished(None, success=True, result="rb")
|
|
begins = [p for t, p in handle.sent if t == "block.begin" and p.get("kind") == "tool"]
|
|
ends = [p for t, p in handle.sent if t == "block.end"]
|
|
assert len(begins) == 2
|
|
# each tool block gets its own end (distinct ids), so neither overwrites
|
|
# the other and no dangling block is left without a close.
|
|
begin_ids = {p["id"] for p in begins}
|
|
end_ids = {p["id"] for p in ends}
|
|
assert len(begin_ids) == 2
|
|
assert begin_ids <= end_ids
|
|
|
|
|
|
def test_format_tokens_abbreviates_thousands() -> None:
|
|
assert _format_tokens(856) == "856"
|
|
assert _format_tokens(1234) == "1.2k"
|
|
assert _format_tokens(0) == "0"
|
|
assert _format_tokens(None) == "0"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_afinalize_writes_usage_to_toolbar_and_invalidates() -> None:
|
|
handle = _ToolbarRecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("done")
|
|
await r.afinalize(UsageSummary(input_tokens=1234, output_tokens=856))
|
|
assert handle.toolbar.get("router_usage") == "1.2k/856"
|
|
assert handle.invalidated == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_afinalize_writes_session_input_to_toolbar() -> None:
|
|
handle = _ToolbarRecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("done")
|
|
await r.afinalize(
|
|
UsageSummary(
|
|
input_tokens=1,
|
|
output_tokens=2,
|
|
session_totals=SessionTotalsSnapshot(input_tokens=84_000),
|
|
)
|
|
)
|
|
|
|
assert handle.toolbar.get("router_usage") == "1/2"
|
|
assert handle.toolbar.get("router_session_input") == 84_000
|
|
assert handle.invalidated == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_afinalize_clears_stale_session_input_when_snapshot_missing() -> None:
|
|
handle = _ToolbarRecordingHandle()
|
|
handle.toolbar["router_session_input"] = 84_000
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("done")
|
|
await r.afinalize(UsageSummary(input_tokens=1, output_tokens=2))
|
|
|
|
assert handle.toolbar.get("router_usage") == "1/2"
|
|
assert "router_session_input" not in handle.toolbar
|
|
assert handle.invalidated == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_turn_begin_clears_stale_router_usage_before_finalize_writes_current_usage() -> None:
|
|
handle = _ToolbarRecordingHandle()
|
|
handle.toolbar["router_usage"] = "999/888"
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.astatus("thinking")
|
|
|
|
assert ("router_usage", None) in handle.toolbar_updates
|
|
assert "router_usage" not in handle.toolbar
|
|
assert handle.invalidated == 1
|
|
|
|
await r.afinalize(UsageSummary(input_tokens=5, output_tokens=7))
|
|
|
|
assert handle.toolbar.get("router_usage") == "5/7"
|
|
assert handle.toolbar_updates[-1] == ("router_usage", "5/7")
|
|
assert handle.invalidated == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_usage_turn_keeps_router_usage_cleared() -> None:
|
|
handle = _ToolbarRecordingHandle()
|
|
handle.toolbar["router_usage"] = "999/888"
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("done")
|
|
await r.afinalize(None)
|
|
assert "router_usage" not in handle.toolbar
|
|
assert handle.invalidated == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_afinalize_tolerates_handle_without_set_toolbar() -> None:
|
|
# The plain recording handle has no set_toolbar/invalidate — afinalize must
|
|
# not crash when wiring usage into the router toolbar.
|
|
handle = _RecordingHandle()
|
|
r = OpenTuiStreamRenderer(output_handle=handle)
|
|
r.__enter__()
|
|
await r.aappend_text("done")
|
|
await r.afinalize(UsageSummary(input_tokens=5, output_tokens=7))
|
|
assert [t for t, _ in handle.sent if t == "turn.end"]
|
|
|
|
|
|
def test_tool_result_summary_keeps_meaningful_lines_without_banners() -> None:
|
|
summary = summarize_result(
|
|
"exit_code=0\n"
|
|
".\n"
|
|
"·\n"
|
|
"...\n"
|
|
"═══ 一级模块 ═══\n"
|
|
"agents\n"
|
|
"────────\n"
|
|
"exit_code=1\n"
|
|
"================\n"
|
|
"src/opensquilla/main.py\n"
|
|
)
|
|
|
|
assert summary == "agents\nexit_code=1\nsrc/opensquilla/main.py"
|
|
assert "exit_code=0" not in summary
|
|
assert " / " not in summary
|
|
assert "═══" not in summary
|
|
|
|
|
|
def test_tool_result_summary_stringifies_single_structured_msg_payload() -> None:
|
|
summary = summarize_result(
|
|
{
|
|
"type": "msg",
|
|
"msg": [
|
|
{"kind": "text", "text": "first"},
|
|
{"kind": "data", "value": {"rows": [1, 2]}},
|
|
],
|
|
}
|
|
)
|
|
|
|
assert summary.startswith("[")
|
|
assert '"type": "msg"' not in summary
|
|
assert '"rows": [1, 2]' in summary
|
|
|
|
|
|
def test_tool_result_summary_stringifies_structured_msg_payloads() -> None:
|
|
summary = summarize_result(
|
|
[
|
|
{"type": "msg", "msg": {"files": ["main.py"], "count": 1}},
|
|
{"type": "msg", "msg": ["ok", {"status": "done"}]},
|
|
]
|
|
)
|
|
|
|
assert summary == (
|
|
'{"count": 1, "files": ["main.py"]}\n'
|
|
'["ok", {"status": "done"}]'
|
|
)
|
|
|
|
|
|
async def test_aturn_started_announces_thinking_before_any_provider_event() -> None:
|
|
"""The stream loop announces the turn as soon as it starts, so the pill
|
|
pulses "thinking" through the silent model-thinking window instead of the
|
|
UI sitting on "ready" until the first token."""
|
|
handle = _RecordingHandle()
|
|
renderer = OpenTuiStreamRenderer(output_handle=handle)
|
|
|
|
await renderer.aturn_started()
|
|
|
|
types = [message_type for message_type, _payload in handle.sent]
|
|
assert "turn.begin" in types
|
|
status = next(p for t, p in handle.sent if t == "turn.status")
|
|
assert status["phase"] == "thinking"
|
|
assert status["active"] is True
|
|
# Idempotent: the first real event must not begin a second turn.
|
|
await renderer.aturn_started()
|
|
assert types.count("turn.begin") == 1
|
|
|
|
|
|
async def test_answer_stream_strips_routing_directive_tags() -> None:
|
|
handle = _RecordingHandle()
|
|
renderer = OpenTuiStreamRenderer(output_handle=handle)
|
|
|
|
# Tag split across deltas, then the real answer.
|
|
await renderer.aappend_text("[[reply_to", presentation="answer")
|
|
await renderer.aappend_text("_current]]\n", presentation="answer")
|
|
await renderer.aappend_text("My name is OpenSquilla.", presentation="answer")
|
|
await renderer.afinalize()
|
|
|
|
appends = [p["delta"] for t, p in handle.sent if t == "block.append"]
|
|
joined = "".join(appends)
|
|
assert "reply_to_current" not in joined
|
|
assert joined == "My name is OpenSquilla."
|
|
# The raw logical buffer (TurnResult text) keeps the model's exact output.
|
|
assert "[[reply_to_current]]" in renderer.buffer
|
|
|
|
|
|
async def test_tag_only_delta_opens_no_answer_block() -> None:
|
|
handle = _RecordingHandle()
|
|
renderer = OpenTuiStreamRenderer(output_handle=handle)
|
|
|
|
await renderer.aappend_text("[[reply_to_current]]", presentation="answer")
|
|
await renderer.afinalize()
|
|
|
|
kinds = [p["kind"] for t, p in handle.sent if t == "block.begin"]
|
|
assert "answer" not in kinds
|
|
|
|
|
|
async def test_held_bracket_prefix_flushes_into_the_block_on_close() -> None:
|
|
handle = _RecordingHandle()
|
|
renderer = OpenTuiStreamRenderer(output_handle=handle)
|
|
|
|
await renderer.aappend_text("see ", presentation="answer")
|
|
# A bracket run that never completes into a directive is ordinary text.
|
|
await renderer.aappend_text("[[re", presentation="answer")
|
|
await renderer.afinalize()
|
|
|
|
appends = [p["delta"] for t, p in handle.sent if t == "block.append"]
|
|
assert "".join(appends) == "see [[re"
|