118 lines
4.5 KiB
Python
118 lines
4.5 KiB
Python
from anthropic.types import Message, TextBlock, ThinkingBlock, ToolUseBlock, Usage
|
|
|
|
from beaver_gateway.frontends.markdown import history, parser, renderer
|
|
from beaver_gateway.frontends.markdown.parser import TextSegment, ToolSegment
|
|
|
|
|
|
def _message(*blocks: object) -> Message:
|
|
return Message(
|
|
id="m",
|
|
type="message",
|
|
role="assistant",
|
|
model="x",
|
|
content=list(blocks), # type: ignore[arg-type]
|
|
stop_reason="end_turn",
|
|
stop_sequence=None,
|
|
usage=Usage(input_tokens=1, output_tokens=1),
|
|
)
|
|
|
|
|
|
def test_tool_calls_leave_no_blank_lines() -> None:
|
|
message = _message(
|
|
TextBlock(type="text", text="first"),
|
|
ToolUseBlock(type="tool_use", id="t1", name="Bash", input={"command": "ls"}),
|
|
ToolUseBlock(type="tool_use", id="t2", name="Read", input={"file_path": "x"}),
|
|
ToolUseBlock(type="tool_use", id="t3", name="Grep", input={"pattern": "y"}),
|
|
TextBlock(type="text", text="second"),
|
|
)
|
|
assert renderer.render_assistant_message(message) == (
|
|
"### Assistant:\n\nfirst\n\nsecond\n"
|
|
)
|
|
|
|
|
|
def _progress_turn() -> Message:
|
|
return _message(
|
|
TextBlock(type="text", text="Checking the containers."),
|
|
ThinkingBlock(type="thinking", thinking="", signature="s"),
|
|
ToolUseBlock(type="tool_use", id="t1", name="komodo", input={}),
|
|
ThinkingBlock(type="thinking", thinking="", signature="s"),
|
|
ThinkingBlock(type="thinking", thinking="All up, reading logs.", signature="s"),
|
|
ToolUseBlock(type="tool_use", id="t2", name="komodo", input={}),
|
|
ThinkingBlock(type="thinking", thinking="Found it.\nOne more.", signature="s"),
|
|
ToolUseBlock(type="tool_use", id="t3", name="komodo", input={}),
|
|
TextBlock(type="text", text="The stack is healthy."),
|
|
)
|
|
|
|
|
|
def test_progress_renders_as_text_in_place_and_empty_thinking_vanishes() -> None:
|
|
assert renderer.render_assistant_message(_progress_turn()) == (
|
|
"### Assistant:\n\nChecking the containers.\n\nAll up, reading logs.\n\n"
|
|
"Found it.\nOne more.\n\nThe stack is healthy.\n"
|
|
)
|
|
|
|
|
|
def test_progress_round_trips_through_the_parser_and_the_stored_history() -> None:
|
|
message = _progress_turn()
|
|
body = renderer.render_user_text("status?") + renderer.render_assistant_message(
|
|
message
|
|
)
|
|
turns = parser.parse(body + renderer.render_user_text("thanks")).turns
|
|
assert turns[1].structure == (
|
|
TextSegment(
|
|
text="Checking the containers.\n\nAll up, reading logs.\n\n"
|
|
"Found it.\nOne more.\n\nThe stack is healthy."
|
|
),
|
|
)
|
|
stored = [
|
|
{"role": "user", "content": "status?"},
|
|
{"role": "assistant", "content": [b.model_dump() for b in message.content]},
|
|
]
|
|
outcome = history.diff_and_fork(stored=stored, incoming=turns)
|
|
assert outcome.divergence_index is None
|
|
assert not outcome.edited
|
|
assert outcome.messages[1] == stored[1]
|
|
|
|
|
|
def test_an_edited_progress_turn_rebuilds_instead_of_rewriting_signed_blocks() -> None:
|
|
message = _progress_turn()
|
|
body = renderer.render_assistant_message(message).replace("Found it", "Fixed it")
|
|
turns = parser.parse(
|
|
renderer.render_user_text("status?") + body + renderer.render_user_text("ok")
|
|
).turns
|
|
stored = [
|
|
{"role": "user", "content": "status?"},
|
|
{"role": "assistant", "content": [b.model_dump() for b in message.content]},
|
|
]
|
|
outcome = history.diff_and_fork(stored=stored, incoming=turns)
|
|
assert outcome.edited
|
|
assert outcome.messages[1] == {
|
|
"role": "assistant",
|
|
"content": [{"type": "text", "text": turns[1].text}],
|
|
}
|
|
|
|
|
|
def test_a_file_with_collapsed_thinking_callouts_parses_as_before() -> None:
|
|
body = (
|
|
"### User:\n\nq\n\n---\n\n### Assistant:\n\n"
|
|
"> [!thinking]-\n> weighing it\n> twice\n\nanswer\n\n"
|
|
"> [!tool]- Read · notes.md\n> input\n\n"
|
|
"> [!thinking]-\n>\n\nmore\n\n---\n\n### User:\n\nnext\n"
|
|
)
|
|
parsed = parser.parse(body)
|
|
assert parsed.turns[1].structure == (
|
|
TextSegment(text="answer"),
|
|
ToolSegment(name="Read"),
|
|
TextSegment(text="more"),
|
|
)
|
|
assert parsed.messages == [
|
|
{"role": "user", "content": "q"},
|
|
{"role": "assistant", "content": "answer\n\nmore"},
|
|
{"role": "user", "content": "next"},
|
|
]
|
|
|
|
|
|
def test_user_scaffold_matches_rendered_user_turn_shape() -> None:
|
|
body = renderer.append_to_body("### Assistant:\n\nhi\n", renderer.USER_SCAFFOLD)
|
|
assert body.endswith("\n\n---\n\n### User:\n\n")
|
|
assert renderer.render_user_text("q").startswith("### User:\n\n")
|