130 lines
4.0 KiB
Python
130 lines
4.0 KiB
Python
from beaver_gateway.backends.transcript import (
|
|
CLI_VERSION,
|
|
build_entries,
|
|
messages_from_entries,
|
|
)
|
|
|
|
HISTORY = [
|
|
{"role": "user", "content": "Write бобёр to notes.txt"},
|
|
{
|
|
"role": "assistant",
|
|
"content": [
|
|
{"type": "thinking", "thinking": "hm", "signature": "sig"},
|
|
{
|
|
"type": "tool_use",
|
|
"id": "toolu_1",
|
|
"name": "Write",
|
|
"input": {"file_path": "notes.txt"},
|
|
},
|
|
],
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "tool_result",
|
|
"tool_use_id": "toolu_1",
|
|
"content": "ok",
|
|
"is_error": False,
|
|
}
|
|
],
|
|
},
|
|
{"role": "assistant", "content": [{"type": "text", "text": "Done."}]},
|
|
{"role": "user", "content": [{"type": "text", "text": "thanks"}]},
|
|
{"role": "assistant", "content": "np"},
|
|
]
|
|
|
|
|
|
def test_round_trip() -> None:
|
|
entries = build_entries(HISTORY, session_id="sid", cwd="/vault", model="claude-x")
|
|
assert messages_from_entries(entries) == [
|
|
*HISTORY[:5],
|
|
{"role": "assistant", "content": [{"type": "text", "text": "np"}]},
|
|
]
|
|
|
|
|
|
def test_chain_and_shape() -> None:
|
|
entries = build_entries(HISTORY, session_id="sid", cwd="/vault", model="claude-x")
|
|
assert [e["type"] for e in entries] == [
|
|
"user",
|
|
"assistant",
|
|
"assistant",
|
|
"user",
|
|
"assistant",
|
|
"user",
|
|
"assistant",
|
|
]
|
|
assert entries[0]["parentUuid"] is None
|
|
for prev, cur in zip(entries, entries[1:], strict=False):
|
|
assert cur["parentUuid"] == prev["uuid"]
|
|
assert len({e["uuid"] for e in entries}) == len(entries)
|
|
|
|
user = entries[0]
|
|
assert user["promptSource"] == "sdk"
|
|
assert user["entrypoint"] == "sdk-py"
|
|
assert user["permissionMode"] == "bypassPermissions"
|
|
assert user["sessionId"] == "sid"
|
|
assert user["cwd"] == "/vault"
|
|
assert user["version"] == CLI_VERSION
|
|
|
|
thinking, tool_use = entries[1], entries[2]
|
|
assert thinking["message"]["id"] == tool_use["message"]["id"]
|
|
assert thinking["requestId"] == tool_use["requestId"]
|
|
assert tool_use["message"]["stop_reason"] == "tool_use"
|
|
assert tool_use["message"]["content"] == [
|
|
{
|
|
"type": "tool_use",
|
|
"id": "toolu_1",
|
|
"name": "Write",
|
|
"input": {"file_path": "notes.txt"},
|
|
}
|
|
]
|
|
|
|
result = entries[3]
|
|
assert result["sourceToolAssistantUUID"] == tool_use["uuid"]
|
|
assert result["promptId"] == user["promptId"]
|
|
assert result["toolUseResult"] == "ok"
|
|
assert result["message"]["content"][0]["is_error"] is False
|
|
|
|
final = entries[4]
|
|
assert final["message"]["stop_reason"] == "end_turn"
|
|
assert final["message"]["usage"]["input_tokens"] == 0
|
|
|
|
assert entries[5]["promptId"] != user["promptId"]
|
|
|
|
|
|
def test_tool_results_split_and_merged_back() -> None:
|
|
history = [
|
|
{"role": "user", "content": "go"},
|
|
{
|
|
"role": "assistant",
|
|
"content": [
|
|
{"type": "tool_use", "id": "a", "name": "Read", "input": {}},
|
|
{"type": "tool_use", "id": "b", "name": "Read", "input": {}},
|
|
],
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "tool_result", "tool_use_id": "a", "content": "1"},
|
|
{
|
|
"type": "tool_result",
|
|
"tool_use_id": "b",
|
|
"content": [{"type": "text", "text": "2"}],
|
|
},
|
|
],
|
|
},
|
|
]
|
|
entries = build_entries(history, session_id="s", cwd="/", model="m")
|
|
assert [e["type"] for e in entries] == [
|
|
"user",
|
|
"assistant",
|
|
"assistant",
|
|
"user",
|
|
"user",
|
|
]
|
|
assert entries[3]["parentUuid"] == entries[1]["uuid"]
|
|
assert entries[4]["parentUuid"] == entries[2]["uuid"]
|
|
assert entries[4]["toolUseResult"] == "2"
|
|
assert messages_from_entries(entries) == history
|