from __future__ import annotations import runpy from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[3] SMOKE = runpy.run_path(ROOT / "scripts" / "smoke-python-runtime.py") @pytest.mark.parametrize( ("prompt_name", "expected"), [ ("SNAPSHOT_DIRECT_CHILD_PROMPT", "DIRECT_CHILD_OK"), ("SNAPSHOT_WORKFLOW_CHILD_PROMPT", "WORKFLOW_CHILD_OK"), ], ) def test_child_prompt_precedes_runtime_context(prompt_name: str, expected: str) -> None: chunks = SMOKE["completion_chunks"]({ "messages": [ {"role": "user", "content": SMOKE[prompt_name]}, {"role": "user", "content": "Current runtime context"}, ], }) assert any( choice.get("delta", {}).get("content") == expected for chunk in chunks for choice in chunk.get("choices", []) ) def test_mcp_smoke_requests_the_discovered_tool() -> None: chunks = SMOKE["completion_chunks"]({ "messages": [{"role": "user", "content": SMOKE["MCP_PROMPT"]}], "tools": [{"type": "function", "function": {"name": "mcp__fixture__add"}}], }) calls = [ call for chunk in chunks for choice in chunk.get("choices", []) for call in choice.get("delta", {}).get("tool_calls", []) ] assert calls[0]["function"] == { "name": "mcp__fixture__add", "arguments": '{"a": 19, "b": 23}', } def test_mcp_smoke_accepts_the_external_server_result() -> None: chunks = SMOKE["completion_chunks"]({ "messages": [ {"role": "user", "content": SMOKE["MCP_PROMPT"]}, { "role": "assistant", "tool_calls": [{ "id": "mcp-add", "type": "function", "function": {"name": "mcp__fixture__add", "arguments": '{}'}, }], }, {"role": "tool", "tool_call_id": "mcp-add", "content": "42"}, ], }) assert any( choice.get("delta", {}).get("content") == SMOKE["MCP_TEXT"] for chunk in chunks for choice in chunk.get("choices", []) ) def test_snapshot_comparison_normalizes_only_session_generation_provenance() -> None: normalize = SMOKE["normalize_session_format_comparison"] expected = { "header": {"type": "session", "version": 0, "otherVersion": 7}, "accepted": { "type": "session-log-deepseek/delivery-accepted", "data": {"sessionId": "s", "throughSeq": 4}, }, "source": { "kind": "session-reference", "references": [{"sessionId": "other", "capturedThroughSeq": 8}], }, } actual = { "header": {"type": "session", "version": 1, "otherVersion": 7}, "accepted": { "type": "session-log-deepseek/delivery-accepted", "data": {"sessionId": "s", "sessionFormatVersion": 1, "throughSeq": 4}, }, "source": { "kind": "session-reference", "references": [{ "sessionId": "other", "capturedFormatVersion": 1, "capturedThroughSeq": 8, }], }, } assert normalize(expected) == normalize(actual) assert normalize(expected)["header"]["otherVersion"] == 7 def test_snapshot_value_normalizes_embedded_assistant_stream_timing() -> None: normalize = SMOKE["normalize_snapshot_value"] event = { "type": "assistant/message", "seq": 4, "time": 100, "data": { "stream": [ {"type": "chunk", "time": 101, "chunk": {"type": "finish"}}, {"type": "text-chunks", "time0": 102, "dt": [1, 2], "texts": ["a", "b", "c"]}, ], }, } normalized = normalize(event, []) assert normalized["time"] == 0 assert normalized["data"]["stream"] == [ {"type": "chunk", "time": 0, "chunk": {"type": "finish"}}, {"type": "text-chunks", "time0": 0, "dt": [0, 0], "texts": ["a", "b", "c"]}, ] def test_snapshot_comparison_expands_embedded_assistant_streams() -> None: normalize = SMOKE["normalize_session_format_comparison"] expected = [ { "type": "assistant/chunk", "seq": 4, "time": 0, "data": {"turn": 1, "step": 1, "chunk": { "type": "text-delta", "index": 0, "text": "done", }}, }, { "type": "assistant/message", "seq": 5, "time": 0, "data": {"turn": 1, "step": 1, "message": {"role": "assistant"}}, "sourceEventSeqs": [4], "surfaceOp": "append", }, ] actual = [{ "type": "assistant/message", "seq": 4, "time": 0, "data": { "turn": 1, "step": 1, "message": {"role": "assistant"}, "stream": [{ "type": "text-chunks", "time0": 0, "index": 0, "dt": [], "texts": ["done"], }], }, "surfaceOp": "append", }] assert normalize(actual, 2) == normalize(expected, 1) tool_result = { "type": "tool/result", "data": {"turn": 1, "step": 1}, "sourceEventSeqs": [4], } assert normalize(tool_result, 1)["sourceEventSeqs"] == [4] assert normalize(tool_result, 2)["sourceEventSeqs"] == [4] def test_snapshot_stream_expands_reasoning_and_tool_call_records() -> None: expand = SMOKE["expand_snapshot_stream_member"] assert expand({ "type": "reasoning-chunks", "time0": 0, "index": 1, "dt": [], "texts": ["think"], }) == [{"type": "reasoning-delta", "index": 1, "text": "think"}] assert expand({ "type": "tool-call-chunks", "time0": 0, "index": 2, "id": "call-1", "name": "read", "dt": [1], "args": ["{", "}"], }) == [ {"type": "tool-call-delta", "index": 2, "id": "call-1", "name": "read", "argumentsDelta": "{"}, {"type": "tool-call-delta", "index": 2, "id": "call-1", "name": "read", "argumentsDelta": "}"}, ] def test_snapshot_file_builder_order_is_checked_outside_update_mode(tmp_path: Path) -> None: compare = SMOKE["compare_snapshot_files"] with pytest.raises(AssertionError, match="snapshot builder produced"): compare({}, False, tmp_path, ("result.json",)) def test_snapshot_comparison_expands_sdk_wrapped_attempts() -> None: normalize = SMOKE["normalize_session_format_comparison"] actual = [{ "method": "session.event", "payload": { "sessionId": "s", "event": { "type": "assistant/attempt", "seq": 7, "time": 0, "data": { "turn": 1, "step": 1, "stream": [{"type": "chunk", "time": 0, "chunk": {"type": "finish"}}], }, }, }, }] assert normalize(actual) == [{ "method": "session.event", "payload": { "sessionId": "s", "event": { "type": "assistant/chunk", "data": {"turn": 1, "step": 1, "chunk": {"type": "finish"}}, }, }, }] def test_snapshot_generation_names_select_highest_role_without_double_counting( tmp_path: Path, ) -> None: render = SMOKE["snapshot_session_filename"] select = SMOKE["selected_snapshot_session_files"] assert render(0, 0) == "session.jsonl" assert render(0, 2) == "session.v2.jsonl" assert render(3, 0) == "session.3.jsonl" assert render(3, 2) == "session.3.v2.jsonl" (tmp_path / "session.jsonl").write_text( '{"type":"session","version":0}\n', encoding="utf-8", ) (tmp_path / "session.v1.jsonl").write_text( '{"type":"session","version":1}\n', encoding="utf-8", ) (tmp_path / "session.1.jsonl").write_text( '{"type":"session","version":0}\n', encoding="utf-8", ) assert {index: path.name for index, path in select(tmp_path).items()} == { 0: "session.v1.jsonl", 1: "session.1.jsonl", } def test_snapshot_generation_filename_must_match_header(tmp_path: Path) -> None: (tmp_path / "session.v1.jsonl").write_text( '{"type":"session","version":0}\n', encoding="utf-8", ) with pytest.raises(AssertionError, match="filename declares Session format v1"): SMOKE["selected_snapshot_session_files"](tmp_path)