tests/test_repro_verify.py · oak/benchmarks

Viewing files at commit 88e86b3faccdCommit details View latest

Browse files

tests/test_repro_verify.py 325 lines · 13.5 KB · python

"""Tests for third-party replay verification against a repro bundle."""

from __future__ import annotations

import contextlib
import io
import json
import sys
import tempfile
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))

import repro_bundle
import repro_verify

OAK_ELAPSED = [10.0, 11.0, 12.0, 10.5, 11.5]
GIT_ELAPSED = [20.0, 21.0, 22.0, 20.5, 21.5]


def make_row(subject: str, run: int, elapsed: float, tokens: float = 120.0) -> dict:
    return {
        "bench_id": "verify-test",
        "profile": "standard",
        "scenario": "core-small",
        "operation": "status",
        "run": run,
        "subject": subject,
        "subject_kind": "oak" if subject.startswith("oak") else "git",
        "elapsed_ms": elapsed,
        "returncode": 0,
        "command": [subject, "status"],
        "estimated_tokens_total": tokens,
        "raw_output_bytes": 400,
        "tool_call_count": 3,
        "binary_sha256": ("a" if subject == "oak" else "b") * 64,
    }


def baseline_rows() -> list[dict]:
    rows = [make_row("oak", run, elapsed) for run, elapsed in enumerate(OAK_ELAPSED)]
    rows += [make_row("git", run, elapsed) for run, elapsed in enumerate(GIT_ELAPSED)]
    return rows


def configured_subject_rows() -> list[dict]:
    rows = [make_row("oak_local", run, elapsed) for run, elapsed in enumerate(OAK_ELAPSED)]
    rows += [make_row("git", run, elapsed) for run, elapsed in enumerate(GIT_ELAPSED)]
    return rows


def multi_oak_subject_rows() -> list[dict]:
    rows = [make_row("oak_installed", run, elapsed) for run, elapsed in enumerate(OAK_ELAPSED)]
    rows += [make_row("oak_local", run, elapsed) for run, elapsed in enumerate([30.0, 31.0, 32.0, 30.5, 31.5])]
    rows += [make_row("git", run, elapsed) for run, elapsed in enumerate(GIT_ELAPSED)]
    return rows


def write_rows(path: Path, rows: list[dict]) -> Path:
    path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")
    return path


def run_verify(argv: list[str]) -> tuple[int, str]:
    stdout = io.StringIO()
    with contextlib.redirect_stdout(stdout):
        code = repro_verify.main(argv)
    return code, stdout.getvalue()


class ReproVerifyTest(unittest.TestCase):
    def make_bundle(self, tmp: Path, original_rows: list[dict]) -> tuple[Path, Path]:
        original_path = write_rows(tmp / "original.jsonl", original_rows)
        bundle_path = tmp / "bundle.json"
        self.assertEqual(
            repro_bundle.main(
                [str(original_path), "--out", str(bundle_path), "--suite-revision", "rev-1"]
            ),
            0,
        )
        return bundle_path, original_path

    def test_identical_replay_with_originals_is_verified_exact(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            originals = baseline_rows()
            bundle_path, original_path = self.make_bundle(tmp, originals)
            replay_path = write_rows(tmp / "replay.jsonl", originals)

            code, out = run_verify(
                [str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]
            )
            self.assertEqual(code, 0, out)
            self.assertIn("verdict: VERIFIED-EXACT", out)
            self.assertIn("ratio CIs overlap", out)
            self.assertIn("tokens exact-match", out)

    def test_tampered_token_count_is_divergent(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            originals = baseline_rows()
            bundle_path, original_path = self.make_bundle(tmp, originals)

            tampered = [dict(row) for row in originals]
            tampered[0]["estimated_tokens_total"] = 999
            replay_path = write_rows(tmp / "replay.jsonl", tampered)

            code, out = run_verify(
                [str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]
            )
            self.assertEqual(code, 1, out)
            self.assertIn("verdict: DIVERGENT", out)
            self.assertIn("divergent:tokens_mismatch", out)

    def test_replay_without_originals_is_comparable(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            originals = baseline_rows()
            bundle_path, _original_path = self.make_bundle(tmp, originals)
            replay_path = write_rows(tmp / "replay.jsonl", originals)

            code, out = run_verify([str(replay_path), "--bundle", str(bundle_path)])
            self.assertEqual(code, 3, out)
            self.assertIn("verdict: COMPARABLE", out)
            self.assertIn("unverifiable:original_rows_absent", out)

    def test_non_overlapping_latency_ratio_is_divergent(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            originals = baseline_rows()
            bundle_path, original_path = self.make_bundle(tmp, originals)

            # Same machine-independent values, wildly different oak/git ratio.
            slow_oak = [dict(row) for row in originals]
            for row in slow_oak:
                if row["subject"] == "oak":
                    row["elapsed_ms"] = row["elapsed_ms"] * 50.0
            replay_path = write_rows(tmp / "replay.jsonl", slow_oak)

            code, out = run_verify(
                [str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]
            )
            self.assertEqual(code, 1, out)
            self.assertIn("verdict: DIVERGENT", out)
            self.assertIn("divergent:elapsed_ratio_ci_no_overlap", out)

    def test_elapsed_ratio_uses_subject_kind_for_configured_oak_subject_names(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            originals = configured_subject_rows()
            bundle_path, original_path = self.make_bundle(tmp, originals)

            slow_oak = [dict(row) for row in originals]
            for row in slow_oak:
                if row["subject_kind"] == "oak":
                    row["elapsed_ms"] = row["elapsed_ms"] * 50.0
            replay_path = write_rows(tmp / "replay.jsonl", slow_oak)

            code, out = run_verify(
                [str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]
            )
            self.assertEqual(code, 1, out)
            self.assertIn("verdict: DIVERGENT", out)
            self.assertIn("divergent:elapsed_ratio_ci_no_overlap", out)

    def test_elapsed_ratio_reports_each_oak_subject_separately(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            originals = multi_oak_subject_rows()
            bundle_path, original_path = self.make_bundle(tmp, originals)
            replay_path = write_rows(tmp / "replay.jsonl", originals)

            code, out = run_verify(
                [str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]
            )
            self.assertEqual(code, 0, out)
            self.assertIn("core-small/status/oak_installed_vs_git", out)
            self.assertIn("core-small/status/oak_local_vs_git", out)

    def test_content_integrity_source_downgrade_is_divergent(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            originals = [
                dict(
                    row,
                    content_integrity_check_passed=True,
                    content_integrity_source="git_head_payload_sha256",
                )
                for row in baseline_rows()
            ]
            bundle_path, original_path = self.make_bundle(tmp, originals)

            replay = [
                dict(
                    row,
                    content_integrity_check_passed=True,
                    content_integrity_source="mixed_payload_sources",
                    content_integrity_sources=["git_head_payload_sha256", "git_lfs_worktree_payload_sha256"],
                )
                for row in originals
            ]
            replay_path = write_rows(tmp / "replay.jsonl", replay)

            code, out = run_verify(
                [str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]
            )
            self.assertEqual(code, 1, out)
            self.assertIn("verdict: DIVERGENT", out)
            self.assertIn("divergent:content_integrity_source_downgrade", out)

    def test_tampered_original_hash_is_divergent(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            originals = baseline_rows()
            bundle_path, original_path = self.make_bundle(tmp, originals)
            replay_path = write_rows(tmp / "replay.jsonl", originals)

            # Rewrite the original file after bundling: its sha256 no longer
            # matches the pinned raw_inputs entry.
            tampered = [dict(row) for row in originals]
            tampered[0]["estimated_tokens_total"] = 1
            write_rows(original_path, tampered)

            code, out = run_verify(
                [str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]
            )
            self.assertEqual(code, 1, out)
            self.assertIn("verdict: DIVERGENT", out)
            self.assertIn("original_hash_not_pinned_in_bundle", out)

    def test_unmeasured_tokens_on_original_side_caps_at_comparable(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            # Null means unmeasured: originals without token fields cannot be
            # exact-verified for tokens, but matching latency keeps the run
            # COMPARABLE rather than DIVERGENT.
            originals = baseline_rows()
            for row in originals:
                row["estimated_tokens_total"] = None
            bundle_path, original_path = self.make_bundle(tmp, originals)

            replay = [dict(row) for row in originals]
            for row in replay:
                row["estimated_tokens_total"] = 120
            replay_path = write_rows(tmp / "replay.jsonl", replay)

            code, out = run_verify(
                [str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]
            )
            self.assertEqual(code, 3, out)
            self.assertIn("verdict: COMPARABLE", out)
            self.assertIn("unverifiable:tokens_unmeasured_in_original", out)

    def test_partial_original_coverage_caps_at_comparable(self) -> None:
        # A bundle pinning two raw inputs must not certify from one alone.
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            core_rows = baseline_rows()
            workflow_rows = [dict(row, scenario="workflow-medium") for row in baseline_rows()]
            core_path = write_rows(tmp / "core.jsonl", core_rows)
            workflow_path = write_rows(tmp / "workflow.jsonl", workflow_rows)
            bundle_path = tmp / "bundle.json"
            self.assertEqual(
                repro_bundle.main(
                    [
                        str(core_path),
                        str(workflow_path),
                        "--out",
                        str(bundle_path),
                        "--suite-revision",
                        "rev-1",
                    ]
                ),
                0,
            )

            # Replay + --original covering only core.jsonl: everything that is
            # checked matches exactly, but the uncovered pinned input caps the
            # verdict at COMPARABLE.
            replay_path = write_rows(tmp / "replay.jsonl", core_rows)
            code, out = run_verify(
                [str(replay_path), "--bundle", str(bundle_path), "--original", str(core_path)]
            )
            self.assertEqual(code, 3, out)
            self.assertIn("verdict: COMPARABLE", out)
            self.assertIn(f"unverifiable:bundle_input_not_provided:{workflow_path}", out)
            self.assertNotIn(f"unverifiable:bundle_input_not_provided:{core_path}", out)

            # Covering both pinned inputs restores the previous behavior.
            full_replay_path = write_rows(tmp / "replay_full.jsonl", core_rows + workflow_rows)
            code, out = run_verify(
                [
                    str(full_replay_path),
                    "--bundle",
                    str(bundle_path),
                    "--original",
                    str(core_path),
                    "--original",
                    str(workflow_path),
                ]
            )
            self.assertEqual(code, 0, out)
            self.assertIn("verdict: VERIFIED-EXACT", out)
            self.assertNotIn("bundle_input_not_provided", out)

    def test_missing_bundle_is_usage_error(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            replay_path = write_rows(tmp / "replay.jsonl", baseline_rows())
            code, _out = run_verify([str(replay_path), "--bundle", str(tmp / "missing.json")])
            self.assertEqual(code, 2)

    def test_bad_bundle_schema_is_usage_error(self) -> None:
        with tempfile.TemporaryDirectory() as tmp_str:
            tmp = Path(tmp_str)
            bundle_path = tmp / "bundle.json"
            bundle_path.write_text(json.dumps({"schema_version": 99}))
            replay_path = write_rows(tmp / "replay.jsonl", baseline_rows())
            code, _out = run_verify([str(replay_path), "--bundle", str(bundle_path)])
            self.assertEqual(code, 2)


if __name__ == "__main__":
    unittest.main()

About

Reversible clean-root candidate rebuilt from oak/benchmarks main@009dead264ca; original repository retained.

Download

RunnersDetails

No runs on main yet.

Open branchesHistory

mrmrs-516758agent_loop_hosted_v1: hosted agent-loop gap map lane (Oak vs GitHub vs ideal)checking…6 behind1mo ago

mrmrs-b26e28Add stale_branch_land_v1: baseline for landing branches that are behind mainchecking…6 behind1mo ago

mrmrs-fad256task_loop: shallow_worker_branch_refresh_v1 (fresh shallow worker -> task branch -> first parent refresh, Oak vs Git, HTTP request counts)checking…6 behind1mo ago

oakmark-throughput-benchmark-transplantAdd oakmark: per-repo push/clone throughput benchmark for oak (ForgeMark analogue)checking…11 behind3mo ago

All branches →

Activity12 commits to main

May 2026now

1	`"""Tests for third-party replay verification against a repro bundle."""`
2
3	`from __future__ import annotations`
4
5	`import contextlib`
6	`import io`
7	`import json`
8	`import sys`
9	`import tempfile`
10	`import unittest`
11	`from pathlib import Path`
12
13	`sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))`
14
15	`import repro_bundle`
16	`import repro_verify`
17
18	`OAK_ELAPSED = [10.0, 11.0, 12.0, 10.5, 11.5]`
19	`GIT_ELAPSED = [20.0, 21.0, 22.0, 20.5, 21.5]`
20
21
22	`def make_row(subject: str, run: int, elapsed: float, tokens: float = 120.0) -> dict:`
23	`return {`
24	`"bench_id": "verify-test",`
25	`"profile": "standard",`
26	`"scenario": "core-small",`
27	`"operation": "status",`
28	`"run": run,`
29	`"subject": subject,`
30	`"subject_kind": "oak" if subject.startswith("oak") else "git",`
31	`"elapsed_ms": elapsed,`
32	`"returncode": 0,`
33	`"command": [subject, "status"],`
34	`"estimated_tokens_total": tokens,`
35	`"raw_output_bytes": 400,`
36	`"tool_call_count": 3,`
37	`"binary_sha256": ("a" if subject == "oak" else "b") * 64,`
38	`}`
39
40
41	`def baseline_rows() -> list[dict]:`
42	`rows = [make_row("oak", run, elapsed) for run, elapsed in enumerate(OAK_ELAPSED)]`
43	`rows += [make_row("git", run, elapsed) for run, elapsed in enumerate(GIT_ELAPSED)]`
44	`return rows`
45
46
47	`def configured_subject_rows() -> list[dict]:`
48	`rows = [make_row("oak_local", run, elapsed) for run, elapsed in enumerate(OAK_ELAPSED)]`
49	`rows += [make_row("git", run, elapsed) for run, elapsed in enumerate(GIT_ELAPSED)]`
50	`return rows`
51
52
53	`def multi_oak_subject_rows() -> list[dict]:`
54	`rows = [make_row("oak_installed", run, elapsed) for run, elapsed in enumerate(OAK_ELAPSED)]`
55	`rows += [make_row("oak_local", run, elapsed) for run, elapsed in enumerate([30.0, 31.0, 32.0, 30.5, 31.5])]`
56	`rows += [make_row("git", run, elapsed) for run, elapsed in enumerate(GIT_ELAPSED)]`
57	`return rows`
58
59
60	`def write_rows(path: Path, rows: list[dict]) -> Path:`
61	`path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")`
62	`return path`
63
64
65	`def run_verify(argv: list[str]) -> tuple[int, str]:`
66	`stdout = io.StringIO()`
67	`with contextlib.redirect_stdout(stdout):`
68	`code = repro_verify.main(argv)`
69	`return code, stdout.getvalue()`
70
71
72	`class ReproVerifyTest(unittest.TestCase):`
73	`def make_bundle(self, tmp: Path, original_rows: list[dict]) -> tuple[Path, Path]:`
74	`original_path = write_rows(tmp / "original.jsonl", original_rows)`
75	`bundle_path = tmp / "bundle.json"`
76	`self.assertEqual(`
77	`repro_bundle.main(`
78	`[str(original_path), "--out", str(bundle_path), "--suite-revision", "rev-1"]`
79	`),`
80	`0,`
81	`)`
82	`return bundle_path, original_path`
83
84	`def test_identical_replay_with_originals_is_verified_exact(self) -> None:`
85	`with tempfile.TemporaryDirectory() as tmp_str:`
86	`tmp = Path(tmp_str)`
87	`originals = baseline_rows()`
88	`bundle_path, original_path = self.make_bundle(tmp, originals)`
89	`replay_path = write_rows(tmp / "replay.jsonl", originals)`
90
91	`code, out = run_verify(`
92	`[str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]`
93	`)`
94	`self.assertEqual(code, 0, out)`
95	`self.assertIn("verdict: VERIFIED-EXACT", out)`
96	`self.assertIn("ratio CIs overlap", out)`
97	`self.assertIn("tokens exact-match", out)`
98
99	`def test_tampered_token_count_is_divergent(self) -> None:`
100	`with tempfile.TemporaryDirectory() as tmp_str:`
101	`tmp = Path(tmp_str)`
102	`originals = baseline_rows()`
103	`bundle_path, original_path = self.make_bundle(tmp, originals)`
104
105	`tampered = [dict(row) for row in originals]`
106	`tampered[0]["estimated_tokens_total"] = 999`
107	`replay_path = write_rows(tmp / "replay.jsonl", tampered)`
108
109	`code, out = run_verify(`
110	`[str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]`
111	`)`
112	`self.assertEqual(code, 1, out)`
113	`self.assertIn("verdict: DIVERGENT", out)`
114	`self.assertIn("divergent:tokens_mismatch", out)`
115
116	`def test_replay_without_originals_is_comparable(self) -> None:`
117	`with tempfile.TemporaryDirectory() as tmp_str:`
118	`tmp = Path(tmp_str)`
119	`originals = baseline_rows()`
120	`bundle_path, _original_path = self.make_bundle(tmp, originals)`
121	`replay_path = write_rows(tmp / "replay.jsonl", originals)`
122
123	`code, out = run_verify([str(replay_path), "--bundle", str(bundle_path)])`
124	`self.assertEqual(code, 3, out)`
125	`self.assertIn("verdict: COMPARABLE", out)`
126	`self.assertIn("unverifiable:original_rows_absent", out)`
127
128	`def test_non_overlapping_latency_ratio_is_divergent(self) -> None:`
129	`with tempfile.TemporaryDirectory() as tmp_str:`
130	`tmp = Path(tmp_str)`
131	`originals = baseline_rows()`
132	`bundle_path, original_path = self.make_bundle(tmp, originals)`
133
134	`# Same machine-independent values, wildly different oak/git ratio.`
135	`slow_oak = [dict(row) for row in originals]`
136	`for row in slow_oak:`
137	`if row["subject"] == "oak":`
138	`row["elapsed_ms"] = row["elapsed_ms"] * 50.0`
139	`replay_path = write_rows(tmp / "replay.jsonl", slow_oak)`
140
141	`code, out = run_verify(`
142	`[str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]`
143	`)`
144	`self.assertEqual(code, 1, out)`
145	`self.assertIn("verdict: DIVERGENT", out)`
146	`self.assertIn("divergent:elapsed_ratio_ci_no_overlap", out)`
147
148	`def test_elapsed_ratio_uses_subject_kind_for_configured_oak_subject_names(self) -> None:`
149	`with tempfile.TemporaryDirectory() as tmp_str:`
150	`tmp = Path(tmp_str)`
151	`originals = configured_subject_rows()`
152	`bundle_path, original_path = self.make_bundle(tmp, originals)`
153
154	`slow_oak = [dict(row) for row in originals]`
155	`for row in slow_oak:`
156	`if row["subject_kind"] == "oak":`
157	`row["elapsed_ms"] = row["elapsed_ms"] * 50.0`
158	`replay_path = write_rows(tmp / "replay.jsonl", slow_oak)`
159
160	`code, out = run_verify(`
161	`[str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]`
162	`)`
163	`self.assertEqual(code, 1, out)`
164	`self.assertIn("verdict: DIVERGENT", out)`
165	`self.assertIn("divergent:elapsed_ratio_ci_no_overlap", out)`
166
167	`def test_elapsed_ratio_reports_each_oak_subject_separately(self) -> None:`
168	`with tempfile.TemporaryDirectory() as tmp_str:`
169	`tmp = Path(tmp_str)`
170	`originals = multi_oak_subject_rows()`
171	`bundle_path, original_path = self.make_bundle(tmp, originals)`
172	`replay_path = write_rows(tmp / "replay.jsonl", originals)`
173
174	`code, out = run_verify(`
175	`[str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]`
176	`)`
177	`self.assertEqual(code, 0, out)`
178	`self.assertIn("core-small/status/oak_installed_vs_git", out)`
179	`self.assertIn("core-small/status/oak_local_vs_git", out)`
180
181	`def test_content_integrity_source_downgrade_is_divergent(self) -> None:`
182	`with tempfile.TemporaryDirectory() as tmp_str:`
183	`tmp = Path(tmp_str)`
184	`originals = [`
185	`dict(`
186	`row,`
187	`content_integrity_check_passed=True,`
188	`content_integrity_source="git_head_payload_sha256",`
189	`)`
190	`for row in baseline_rows()`
191	`]`
192	`bundle_path, original_path = self.make_bundle(tmp, originals)`
193
194	`replay = [`
195	`dict(`
196	`row,`
197	`content_integrity_check_passed=True,`
198	`content_integrity_source="mixed_payload_sources",`
199	`content_integrity_sources=["git_head_payload_sha256", "git_lfs_worktree_payload_sha256"],`
200	`)`
201	`for row in originals`
202	`]`
203	`replay_path = write_rows(tmp / "replay.jsonl", replay)`
204
205	`code, out = run_verify(`
206	`[str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]`
207	`)`
208	`self.assertEqual(code, 1, out)`
209	`self.assertIn("verdict: DIVERGENT", out)`
210	`self.assertIn("divergent:content_integrity_source_downgrade", out)`
211
212	`def test_tampered_original_hash_is_divergent(self) -> None:`
213	`with tempfile.TemporaryDirectory() as tmp_str:`
214	`tmp = Path(tmp_str)`
215	`originals = baseline_rows()`
216	`bundle_path, original_path = self.make_bundle(tmp, originals)`
217	`replay_path = write_rows(tmp / "replay.jsonl", originals)`
218
219	`# Rewrite the original file after bundling: its sha256 no longer`
220	`# matches the pinned raw_inputs entry.`
221	`tampered = [dict(row) for row in originals]`
222	`tampered[0]["estimated_tokens_total"] = 1`
223	`write_rows(original_path, tampered)`
224
225	`code, out = run_verify(`
226	`[str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]`
227	`)`
228	`self.assertEqual(code, 1, out)`
229	`self.assertIn("verdict: DIVERGENT", out)`
230	`self.assertIn("original_hash_not_pinned_in_bundle", out)`
231
232	`def test_unmeasured_tokens_on_original_side_caps_at_comparable(self) -> None:`
233	`with tempfile.TemporaryDirectory() as tmp_str:`
234	`tmp = Path(tmp_str)`
235	`# Null means unmeasured: originals without token fields cannot be`
236	`# exact-verified for tokens, but matching latency keeps the run`
237	`# COMPARABLE rather than DIVERGENT.`
238	`originals = baseline_rows()`
239	`for row in originals:`
240	`row["estimated_tokens_total"] = None`
241	`bundle_path, original_path = self.make_bundle(tmp, originals)`
242
243	`replay = [dict(row) for row in originals]`
244	`for row in replay:`
245	`row["estimated_tokens_total"] = 120`
246	`replay_path = write_rows(tmp / "replay.jsonl", replay)`
247
248	`code, out = run_verify(`
249	`[str(replay_path), "--bundle", str(bundle_path), "--original", str(original_path)]`
250	`)`
251	`self.assertEqual(code, 3, out)`
252	`self.assertIn("verdict: COMPARABLE", out)`
253	`self.assertIn("unverifiable:tokens_unmeasured_in_original", out)`
254
255	`def test_partial_original_coverage_caps_at_comparable(self) -> None:`
256	`# A bundle pinning two raw inputs must not certify from one alone.`
257	`with tempfile.TemporaryDirectory() as tmp_str:`
258	`tmp = Path(tmp_str)`
259	`core_rows = baseline_rows()`
260	`workflow_rows = [dict(row, scenario="workflow-medium") for row in baseline_rows()]`
261	`core_path = write_rows(tmp / "core.jsonl", core_rows)`
262	`workflow_path = write_rows(tmp / "workflow.jsonl", workflow_rows)`
263	`bundle_path = tmp / "bundle.json"`
264	`self.assertEqual(`
265	`repro_bundle.main(`
266	`[`
267	`str(core_path),`
268	`str(workflow_path),`
269	`"--out",`
270	`str(bundle_path),`
271	`"--suite-revision",`
272	`"rev-1",`
273	`]`
274	`),`
275	`0,`
276	`)`
277
278	`# Replay + --original covering only core.jsonl: everything that is`
279	`# checked matches exactly, but the uncovered pinned input caps the`
280	`# verdict at COMPARABLE.`
281	`replay_path = write_rows(tmp / "replay.jsonl", core_rows)`
282	`code, out = run_verify(`
283	`[str(replay_path), "--bundle", str(bundle_path), "--original", str(core_path)]`
284	`)`
285	`self.assertEqual(code, 3, out)`
286	`self.assertIn("verdict: COMPARABLE", out)`
287	`self.assertIn(f"unverifiable:bundle_input_not_provided:{workflow_path}", out)`
288	`self.assertNotIn(f"unverifiable:bundle_input_not_provided:{core_path}", out)`
289
290	`# Covering both pinned inputs restores the previous behavior.`
291	`full_replay_path = write_rows(tmp / "replay_full.jsonl", core_rows + workflow_rows)`
292	`code, out = run_verify(`
293	`[`
294	`str(full_replay_path),`
295	`"--bundle",`
296	`str(bundle_path),`
297	`"--original",`
298	`str(core_path),`
299	`"--original",`
300	`str(workflow_path),`
301	`]`
302	`)`
303	`self.assertEqual(code, 0, out)`
304	`self.assertIn("verdict: VERIFIED-EXACT", out)`
305	`self.assertNotIn("bundle_input_not_provided", out)`
306
307	`def test_missing_bundle_is_usage_error(self) -> None:`
308	`with tempfile.TemporaryDirectory() as tmp_str:`
309	`tmp = Path(tmp_str)`
310	`replay_path = write_rows(tmp / "replay.jsonl", baseline_rows())`
311	`code, _out = run_verify([str(replay_path), "--bundle", str(tmp / "missing.json")])`
312	`self.assertEqual(code, 2)`
313
314	`def test_bad_bundle_schema_is_usage_error(self) -> None:`
315	`with tempfile.TemporaryDirectory() as tmp_str:`
316	`tmp = Path(tmp_str)`
317	`bundle_path = tmp / "bundle.json"`
318	`bundle_path.write_text(json.dumps({"schema_version": 99}))`
319	`replay_path = write_rows(tmp / "replay.jsonl", baseline_rows())`
320	`code, _out = run_verify([str(replay_path), "--bundle", str(bundle_path)])`
321	`self.assertEqual(code, 2)`
322
323
324	`if __name__ == "__main__":`
325	`unittest.main()`