| 1 | """Golden-fixture tests for agent stream parsing.
|
| 2 |
|
| 3 | These fixtures are recorded stream shapes for each adapter. The claude
|
| 4 | fixture pins the turn-attribution bug class that shipped once already:
|
| 5 | assistant events carry tool_use blocks BEFORE their results arrive, so naive
|
| 6 | between-boundary counting credits tools to the wrong turn.
|
| 7 | """
|
| 8 |
|
| 9 | from __future__ import annotations
|
| 10 |
|
| 11 | import sys
|
| 12 | from pathlib import Path
|
| 13 |
|
| 14 | sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
| 15 |
|
| 16 | import json
|
| 17 | import unittest
|
| 18 | from pathlib import Path
|
| 19 |
|
| 20 | from oakbench.stream_adapters import StreamAnalysis, analyze_stream
|
| 21 |
|
| 22 | FIXTURES = Path(__file__).resolve().parent / "fixtures"
|
| 23 |
|
| 24 |
|
| 25 | def load_stream(name: str) -> list:
|
| 26 | return [json.loads(line) for line in (FIXTURES / name).read_text().splitlines() if line.strip()]
|
| 27 |
|
| 28 |
|
| 29 | class ClaudeStreamTest(unittest.TestCase):
|
| 30 | def analysis(self) -> StreamAnalysis:
|
| 31 | return analyze_stream(load_stream("claude_stream.jsonl"), "claude_print_stream_json")
|
| 32 |
|
| 33 | def test_turn_count_and_attribution(self) -> None:
|
| 34 | analysis = self.analysis()
|
| 35 | self.assertEqual(analysis.turn_source, "adapter_stream")
|
| 36 | self.assertEqual(analysis.assistant_turns_total, 3)
|
| 37 | self.assertEqual([t["tool_calls_observed"] for t in analysis.turns], [1, 2, 0])
|
| 38 | self.assertEqual(analysis.max_tool_calls_per_turn, 2)
|
| 39 | self.assertEqual(analysis.tool_calls_per_turn_avg, 1.0)
|
| 40 |
|
| 41 | def test_context_residency(self) -> None:
|
| 42 | analysis = self.analysis()
|
| 43 | # Cumulative input includes cache reads: (1000+0)+(1200+900)+(1300+1100).
|
| 44 | self.assertEqual(analysis.cumulative_input_tokens, 5500)
|
| 45 | self.assertEqual(analysis.peak_input_tokens, 2400)
|
| 46 |
|
| 47 | def test_compaction_is_measured(self) -> None:
|
| 48 | self.assertEqual(self.analysis().context_truncation_events, 1)
|
| 49 |
|
| 50 | def test_recovery_cost(self) -> None:
|
| 51 | analysis = self.analysis()
|
| 52 | self.assertEqual(analysis.failed_tool_calls_total, 1)
|
| 53 | self.assertEqual(analysis.turns_to_recovery, 1)
|
| 54 | # The reaction turn's output tokens.
|
| 55 | self.assertEqual(analysis.tokens_to_recovery, 80)
|
| 56 |
|
| 57 | def test_unknown_command_detected(self) -> None:
|
| 58 | self.assertEqual(self.analysis().unknown_command_failures_total, 1)
|
| 59 |
|
| 60 |
|
| 61 | class CodexStreamTest(unittest.TestCase):
|
| 62 | """Synthetic codex shapes; real-transcript semantics are pinned below."""
|
| 63 |
|
| 64 | def analysis(self) -> StreamAnalysis:
|
| 65 | return analyze_stream(load_stream("codex_stream.jsonl"), "codex_exec_json")
|
| 66 |
|
| 67 | def test_turns_are_agent_messages(self) -> None:
|
| 68 | analysis = self.analysis()
|
| 69 | # Codex turn.completed covers the whole session; model round-trips are
|
| 70 | # the agent_message items (verified against codex-cli 0.139).
|
| 71 | self.assertEqual(analysis.assistant_turns_total, 1)
|
| 72 | self.assertEqual([t["tool_calls_observed"] for t in analysis.turns], [3])
|
| 73 |
|
| 74 | def test_session_usage_no_cache_double_count(self) -> None:
|
| 75 | analysis = self.analysis()
|
| 76 | # OpenAI input_tokens already includes the cached subset; adding the
|
| 77 | # cached figure again would double count.
|
| 78 | self.assertEqual(analysis.cumulative_input_tokens, 900 + 1100)
|
| 79 | # Per-round-trip input is not adapter-visible for codex.
|
| 80 | self.assertIsNone(analysis.peak_input_tokens)
|
| 81 |
|
| 82 | def test_truncation_unmeasured_without_markers(self) -> None:
|
| 83 | # Codex streams emit no compaction markers; absence is unmeasured,
|
| 84 | # not zero (ADR-0002).
|
| 85 | self.assertIsNone(self.analysis().context_truncation_events)
|
| 86 |
|
| 87 | def test_recovery_and_unknown_command(self) -> None:
|
| 88 | analysis = self.analysis()
|
| 89 | self.assertEqual(analysis.failed_tool_calls_total, 1)
|
| 90 | # The next successful tool arrives with no turn boundary between:
|
| 91 | # recovered within the same round-trip.
|
| 92 | self.assertEqual(analysis.turns_to_recovery, 0)
|
| 93 | self.assertEqual(analysis.unknown_command_failures_total, 1)
|
| 94 |
|
| 95 |
|
| 96 | class RealClaudeTranscriptTest(unittest.TestCase):
|
| 97 | """Sanitized transcript from an actual `claude -p --output-format
|
| 98 | stream-json --verbose` bugfix run (claude 2.1.170, 2026-06-10). The
|
| 99 | headline numbers show why turn metrics exist: a task whose summed
|
| 100 | input+output usage reads ~2.3k tokens moved 151k input tokens through the
|
| 101 | model across 10 turns once cache reads are counted."""
|
| 102 |
|
| 103 | def analysis(self) -> StreamAnalysis:
|
| 104 | return analyze_stream(load_stream("claude_stream_real.jsonl"), "claude_print_stream_json")
|
| 105 |
|
| 106 | def test_real_run_shape(self) -> None:
|
| 107 | analysis = self.analysis()
|
| 108 | self.assertEqual(analysis.turn_source, "adapter_stream")
|
| 109 | self.assertEqual(analysis.assistant_turns_total, 6)
|
| 110 | self.assertEqual(
|
| 111 | [t["tool_calls_observed"] for t in analysis.turns],
|
| 112 | [1, 2, 1, 1, 1, 0],
|
| 113 | )
|
| 114 | self.assertEqual(analysis.max_tool_calls_per_turn, 2)
|
| 115 | self.assertEqual(analysis.tool_calls_per_turn_avg, 1.0)
|
| 116 | self.assertEqual(analysis.cumulative_input_tokens, 95693)
|
| 117 | self.assertEqual(analysis.peak_input_tokens, 19926)
|
| 118 | self.assertEqual(analysis.context_truncation_events, 0)
|
| 119 |
|
| 120 | def test_real_run_recovery(self) -> None:
|
| 121 | analysis = self.analysis()
|
| 122 | self.assertEqual(analysis.failed_tool_calls_total, 1)
|
| 123 | self.assertEqual(analysis.turns_to_recovery, 1)
|
| 124 |
|
| 125 |
|
| 126 | class RealCodexTranscriptTest(unittest.TestCase):
|
| 127 | """Sanitized transcript from an actual `codex exec --json` bugfix run
|
| 128 | (codex-cli 0.139.0, 2026-06-10)."""
|
| 129 |
|
| 130 | def analysis(self) -> StreamAnalysis:
|
| 131 | return analyze_stream(load_stream("codex_stream_real.jsonl"), "codex_exec_json")
|
| 132 |
|
| 133 | def test_real_run_shape(self) -> None:
|
| 134 | analysis = self.analysis()
|
| 135 | self.assertEqual(analysis.assistant_turns_total, 8)
|
| 136 | self.assertEqual(
|
| 137 | [t["tool_calls_observed"] for t in analysis.turns], [0, 3, 3, 0, 2, 0, 3, 1]
|
| 138 | )
|
| 139 | self.assertEqual(analysis.cumulative_input_tokens, 111415)
|
| 140 | self.assertIsNone(analysis.peak_input_tokens)
|
| 141 | self.assertIsNone(analysis.context_truncation_events)
|
| 142 |
|
| 143 | def test_real_run_recovery(self) -> None:
|
| 144 | analysis = self.analysis()
|
| 145 | self.assertEqual(analysis.failed_tool_calls_total, 2)
|
| 146 | self.assertEqual(analysis.turns_to_recovery, 1)
|
| 147 |
|
| 148 |
|
| 149 | class CodexGoldenTest(unittest.TestCase):
|
| 150 | """collect_usage over the real codex transcript (RealCodexTranscriptTest
|
| 151 | pins the stream-analysis side of the same fixture)."""
|
| 152 |
|
| 153 | def test_provider_usage_via_collect_usage(self) -> None:
|
| 154 | import agent_workflow
|
| 155 |
|
| 156 | usage = agent_workflow.collect_usage(load_stream("codex_stream_real.jsonl"))
|
| 157 | self.assertIsNotNone(usage["input_tokens_reported"])
|
| 158 | self.assertIsNotNone(usage["output_tokens_reported"])
|
| 159 |
|
| 160 |
|
| 161 | class CursorAgentGoldenTest(unittest.TestCase):
|
| 162 | """Recorded cursor-agent 2026.06.11 stream (scrubbed). Pins the camelCase
|
| 163 | usage fallback: this CLI reports {inputTokens, outputTokens,
|
| 164 | cacheReadTokens} only in its final result event."""
|
| 165 |
|
| 166 | def test_turns_extracted(self) -> None:
|
| 167 | events = load_stream("cursor_agent_stream.jsonl")
|
| 168 | analysis = analyze_stream(events, "cursor_agent_stream_json")
|
| 169 | self.assertEqual(analysis.turn_source, "adapter_stream")
|
| 170 | self.assertGreater(analysis.assistant_turns_total, 0)
|
| 171 | self.assertEqual(analysis.failed_tool_calls_total, 2)
|
| 172 | self.assertEqual(analysis.failed_test_commands_total, 1)
|
| 173 |
|
| 174 | def test_camelcase_usage_fallback(self) -> None:
|
| 175 | import agent_workflow
|
| 176 |
|
| 177 | usage = agent_workflow.collect_usage(load_stream("cursor_agent_stream.jsonl"))
|
| 178 | self.assertEqual(usage["input_tokens_reported"], 17208)
|
| 179 | self.assertEqual(usage["output_tokens_reported"], 1440)
|
| 180 | self.assertEqual(usage["cache_read_tokens_reported"], 95232)
|
| 181 |
|
| 182 | def test_tool_calls_classified_once(self) -> None:
|
| 183 | import agent_workflow
|
| 184 |
|
| 185 | counts, by_kind = agent_workflow.collect_tool_metrics(
|
| 186 | "",
|
| 187 | "",
|
| 188 | load_stream("cursor_agent_stream.jsonl"),
|
| 189 | )
|
| 190 | self.assertEqual(counts["tool_calls_total"], 9)
|
| 191 | self.assertEqual(counts["terminal_commands_total"], 4)
|
| 192 | self.assertEqual(counts["search_calls"], 2)
|
| 193 | self.assertEqual(counts["file_read_calls"], 2)
|
| 194 | self.assertEqual(counts["edit_calls"], 1)
|
| 195 | self.assertEqual(counts["file_write_calls"], 1)
|
| 196 | self.assertEqual(counts["test_commands_total"], 2)
|
| 197 | self.assertEqual(counts["vcs_commands_total"], 2)
|
| 198 | self.assertEqual(
|
| 199 | by_kind,
|
| 200 | {
|
| 201 | "edittoolcall": 1,
|
| 202 | "greptoolcall": 1,
|
| 203 | "globtoolcall": 1,
|
| 204 | "readtoolcall": 2,
|
| 205 | "shelltoolcall": 4,
|
| 206 | },
|
| 207 | )
|
| 208 |
|
| 209 | def test_claude_snake_case_never_double_counted(self) -> None:
|
| 210 | import agent_workflow
|
| 211 |
|
| 212 | usage = agent_workflow.collect_usage(load_stream("claude_stream_real.jsonl"))
|
| 213 | # The claude fixture carries snake_case usage; camelCase modelUsage
|
| 214 | # and nested iterations must not be summed on top of result.usage.
|
| 215 | self.assertEqual(usage["input_tokens_reported"], 225)
|
| 216 | self.assertEqual(usage["output_tokens_reported"], 1223)
|
| 217 | self.assertEqual(usage["cache_read_tokens_reported"], 95468)
|
| 218 | self.assertEqual(usage["cache_write_tokens_reported"], 20059)
|
| 219 |
|
| 220 |
|
| 221 | class EmptyAndUnknownStreamTest(unittest.TestCase):
|
| 222 | def test_empty_stream_is_unmeasured(self) -> None:
|
| 223 | analysis = analyze_stream([], "claude_print_stream_json")
|
| 224 | self.assertEqual(analysis.turn_source, "unmeasured")
|
| 225 | self.assertIsNone(analysis.assistant_turns_total)
|
| 226 | self.assertIsNone(analysis.failed_tool_calls_total)
|
| 227 | self.assertIsNone(analysis.context_truncation_events)
|
| 228 |
|
| 229 | def test_unknown_adapter_uses_generic_heuristics(self) -> None:
|
| 230 | events = load_stream("claude_stream.jsonl")
|
| 231 | analysis = analyze_stream(events, "some_future_cli")
|
| 232 | self.assertEqual(analysis.assistant_turns_total, 3)
|
| 233 | # Generic adapters never claim a measured-zero for compaction absence,
|
| 234 | # but this stream has an explicit marker.
|
| 235 | self.assertEqual(analysis.context_truncation_events, 1)
|
| 236 |
|
| 237 |
|
| 238 | if __name__ == "__main__":
|
| 239 | unittest.main()
|