from beaver_gateway.core.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