Log in
benchmarksPublic

Viewing files at commit 2735d7f5558eCommit detailsView latest

Browse files
tests/test_devloop.py 569 lines · 26.3 KB · python
1
"""Focused regression tests for devloop verdict gates."""
2
3
from __future__ import annotations
4
5
import importlib.util
6
import sys
7
import tempfile
8
import unittest
9
from pathlib import Path
10
from unittest import mock
11
12
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
13
14
15
def load_devloop():
16
    path = Path(__file__).resolve().parents[1] / "scripts" / "devloop.py"
17
    spec = importlib.util.spec_from_file_location("devloop", path)
18
    module = importlib.util.module_from_spec(spec)
19
    sys.modules["devloop"] = module
20
    spec.loader.exec_module(module)
21
    return module
22
23
24
def row(subject: str, operation: str, *, returncode: int = 0, elapsed_ms: float = 10.0) -> dict:
25
    return {
26
        "subject": subject,
27
        "scenario": "tiny_text",
28
        "operation": operation,
29
        "run": 0,
30
        "returncode": returncode,
31
        "elapsed_ms": elapsed_ms,
32
        "vcs_tool_call_count": 1,
33
        "tool_call_count": 1,
34
        "raw_output_bytes": 10,
35
        "estimated_tokens_total": 3,
36
    }
37
38
39
class DevloopVerdictTest(unittest.TestCase):
40
    def setUp(self) -> None:
41
        self.devloop = load_devloop()
42
43
    def compare(self, rows: list[dict]) -> list[str]:
44
        regressions, _ = self.devloop.compare_subjects(
45
            rows,
46
            "oak_local",
47
            "oak_installed",
48
            "Oak baseline",
49
            latency_threshold_pct=10.0,
50
            min_delta_ms=5.0,
51
            efficiency_gates=True,
52
            output_bytes_threshold_pct=25.0,
53
        )
54
        return regressions
55
56
    def test_all_failing_target_is_regression_even_without_target_latency_median(self) -> None:
57
        rows = [
58
            row("oak_installed", "status.dirty", returncode=0, elapsed_ms=10.0),
59
            row("oak_installed", "status.dirty", returncode=0, elapsed_ms=11.0),
60
            row("oak_local", "status.dirty", returncode=1, elapsed_ms=100.0),
61
            row("oak_local", "status.dirty", returncode=1, elapsed_ms=101.0),
62
        ]
63
64
        regressions = self.compare(rows)
65
66
        self.assertTrue(any("NO SUCCESS" in item and "status.dirty" in item for item in regressions))
67
68
    def test_baseline_flake_does_not_hide_higher_target_failure_rate(self) -> None:
69
        rows = [
70
            row("oak_installed", "status.dirty", returncode=0, elapsed_ms=10.0),
71
            row("oak_installed", "status.dirty", returncode=0, elapsed_ms=11.0),
72
            row("oak_installed", "status.dirty", returncode=1, elapsed_ms=12.0),
73
            row("oak_local", "status.dirty", returncode=0, elapsed_ms=10.0),
74
            row("oak_local", "status.dirty", returncode=1, elapsed_ms=12.0),
75
            row("oak_local", "status.dirty", returncode=1, elapsed_ms=13.0),
76
        ]
77
78
        regressions = self.compare(rows)
79
80
        self.assertTrue(any("FAIL-RATE" in item and "2/3" in item for item in regressions))
81
82
    def test_embedded_null_excludes_probe_and_failed_or_skipped_groups(self) -> None:
83
        rows = [
84
            {**row("oak_installed", "read.file", elapsed_ms=100.0), "vcs_tool_call_count": 0},
85
            {**row("oak_local", "read.file", elapsed_ms=100.0), "vcs_tool_call_count": 0},
86
            {**row("oak_installed", "diff.dirty.inforecall", elapsed_ms=100.0), "vcs_tool_call_count": 0},
87
            {**row("oak_local", "diff.dirty.inforecall", elapsed_ms=900.0), "vcs_tool_call_count": 0},
88
            {**row("oak_installed", "test.step", elapsed_ms=100.0), "vcs_tool_call_count": 0},
89
            {**row("oak_local", "test.step", elapsed_ms=900.0), "vcs_tool_call_count": 0},
90
            {**row("oak_local", "test.step", returncode=1, elapsed_ms=901.0), "vcs_tool_call_count": 0},
91
            {**row("oak_installed", "skipped.step", elapsed_ms=100.0), "vcs_tool_call_count": 0},
92
            {**row("oak_local", "skipped.step", elapsed_ms=900.0), "vcs_tool_call_count": 0},
93
            {
94
                **row("oak_local", "skipped.step", returncode=self.devloop.SKIP_RETURNCODE, elapsed_ms=0.0),
95
                "vcs_tool_call_count": 0,
96
                "skipped": True,
97
            },
98
        ]
99
100
        null_pct, null_ops = self.devloop.embedded_null_p95(rows, "oak_local", "oak_installed")
101
102
        self.assertEqual(null_pct, 0.0)
103
        self.assertEqual(null_ops, 1)
104
105
    def test_noise_floor_is_invariant_to_identical_subject_label_order(self) -> None:
106
        for left, right in ((150.0, 12.0), (12.0, 150.0)):
107
            with mock.patch.object(self.devloop, "run_lane"), mock.patch.object(self.devloop, "load_jsonl", return_value=[
108
                row("oak_installed", "status.clean", elapsed_ms=left),
109
                row("oak_local", "status.clean", elapsed_ms=right),
110
            ]):
111
                spread, operations = self.devloop.measure_noise_floor(Path("/tmp/aa"), Path("/bin/oak"), 2)
112
            self.assertAlmostEqual(spread, 1150.0)
113
            self.assertEqual(operations, 1)
114
115
    def test_noise_floor_equal_medians_have_zero_spread(self) -> None:
116
        with mock.patch.object(self.devloop, "run_lane"), mock.patch.object(self.devloop, "NOISE_FLOOR_MIN_PCT", 0.0), \
117
             mock.patch.object(self.devloop, "load_jsonl", return_value=[
118
                 row("oak_installed", "status.clean"), row("oak_local", "status.clean")]):
119
            self.assertEqual(self.devloop.measure_noise_floor(Path("/tmp/aa"), Path("/bin/oak"), 2), (0.0, 1))
120
121
    def test_noise_floor_excludes_nonpositive_and_missing_medians(self) -> None:
122
        for left, right in ((0, 12), (12, 0), (-1, 12), (12, -1), (12, None), (None, 12)):
123
            with mock.patch.object(self.devloop, "run_lane"), mock.patch.object(self.devloop, "load_jsonl", return_value=[
124
                row("oak_installed", "status.clean", elapsed_ms=left),
125
                row("oak_local", "status.clean", elapsed_ms=right),
126
            ]):
127
                self.assertEqual(self.devloop.measure_noise_floor(Path("/tmp/aa"), Path("/bin/oak"), 2),
128
                                 (self.devloop.NOISE_FLOOR_MIN_PCT, 0))
129
130
    def test_noise_floor_command_skips_remote_operations(self) -> None:
131
        captured: list[list[str]] = []
132
133
        def fake_run_lane(command: list[str], label: str) -> None:
134
            captured.append(command)
135
136
        with (
137
            mock.patch.object(self.devloop, "run_lane", side_effect=fake_run_lane),
138
            mock.patch.object(self.devloop, "load_jsonl", return_value=[]),
139
        ):
140
            self.devloop.measure_noise_floor(Path("/tmp/aa"), Path("/bin/oak"), 3)
141
142
        self.assertIn("--skip-remote", captured[0])
143
144
    def test_contention_preflight_includes_workspace_and_clone_push_gate(self) -> None:
145
        requirements = self.devloop.remote_preflight_requirements(["contention"])
146
147
        self.assertEqual(len(requirements), 1)
148
        operations = set(requirements[0].operations)
149
        self.assertIn("contention_workspace_per_task_w2/parallel.total", operations)
150
        self.assertIn("contention_workspace_per_task_w8/parallel.total", operations)
151
        self.assertIn("contention_clone_push_storm_w2/parallel.total", operations)
152
        self.assertIn("contention_clone_push_storm_w8/parallel.total", operations)
153
154
    def test_contention_preflight_uses_custom_launch_worker_tiers(self) -> None:
155
        requirements = self.devloop.remote_preflight_requirements(
156
            ["contention"],
157
            contention_modes=["clone_push_storm"],
158
            contention_workers=["16", "128"],
159
        )
160
161
        operations = set(requirements[0].operations)
162
        self.assertEqual(
163
            operations,
164
            {
165
                "contention_clone_push_storm_w16/parallel.total",
166
                "contention_clone_push_storm_w128/parallel.total",
167
            },
168
        )
169
170
    def test_contention_command_runs_clone_push_storm_by_default(self) -> None:
171
        import tempfile
172
173
        captured: list[list[str]] = []
174
175
        def fake_run_lane(command: list[str], label: str) -> None:
176
            captured.append(command)
177
178
        with tempfile.TemporaryDirectory() as tmp:
179
            argv = [
180
                "devloop.py",
181
                "--lanes",
182
                "contention",
183
                "--skip-noise-floor",
184
                "--skip-build",
185
                "--oak-baseline-bin",
186
                "/bin/echo",
187
                "--oak-local-bin",
188
                "/bin/echo",
189
                "--results",
190
                tmp,
191
            ]
192
            with (
193
                mock.patch.object(sys, "argv", argv),
194
                mock.patch.object(self.devloop, "run_lane", side_effect=fake_run_lane),
195
                mock.patch.object(self.devloop, "load_jsonl", return_value=[]),
196
                mock.patch.object(self.devloop.oakbench_remotes, "print_remote_preflight_warnings"),
197
            ):
198
                rc = self.devloop.main()
199
200
        self.assertEqual(rc, 0)
201
        self.assertEqual(len(captured), 1)
202
        command = captured[0]
203
        self.assertIn("--modes", command)
204
        self.assertEqual(
205
            command[command.index("--modes") + 1],
206
            self.devloop.DEFAULT_CONTENTION_MODES,
207
        )
208
        self.assertIn("clone_push_storm", command[command.index("--modes") + 1])
209
210
    def test_diagnostic_probe_failure_does_not_trip_regression_gate(self) -> None:
211
        rows = [
212
            row("oak_installed", "status.dirty.determinism", returncode=0, elapsed_ms=10.0),
213
            row("oak_local", "status.dirty.determinism", returncode=1, elapsed_ms=11.0),
214
            row("oak_installed", "status.dirty", returncode=0, elapsed_ms=10.0),
215
            row("oak_local", "status.dirty", returncode=0, elapsed_ms=11.0),
216
        ]
217
218
        regressions = self.compare(rows)
219
220
        self.assertFalse(any("status.dirty.determinism" in item for item in regressions))
221
222
    def test_load_jsonl_reports_controlled_input_errors(self) -> None:
223
        with tempfile.TemporaryDirectory() as tmp:
224
            root = Path(tmp)
225
            with self.assertRaises(SystemExit) as missing:
226
                self.devloop.load_jsonl(root / "missing.jsonl")
227
            self.assertIn("result JSONL missing", str(missing.exception))
228
229
            empty = root / "empty.jsonl"
230
            empty.write_text("\n")
231
            with self.assertRaises(SystemExit) as empty_error:
232
                self.devloop.load_jsonl(empty)
233
            self.assertIn("contains no rows", str(empty_error.exception))
234
235
            malformed = root / "malformed.jsonl"
236
            malformed.write_text('{"ok": true}\nnot json\n')
237
            with self.assertRaises(SystemExit) as malformed_error:
238
                self.devloop.load_jsonl(malformed)
239
            self.assertIn("malformed result JSONL", str(malformed_error.exception))
240
            self.assertIn(":2:", str(malformed_error.exception))
241
242
            not_object = root / "not-object.jsonl"
243
            not_object.write_text("[1, 2, 3]\n")
244
            with self.assertRaises(SystemExit) as not_object_error:
245
                self.devloop.load_jsonl(not_object)
246
            self.assertIn("row is not an object", str(not_object_error.exception))
247
248
            valid = root / "valid.jsonl"
249
            valid.write_text('{"ok": true}\n\n')
250
            self.assertEqual(self.devloop.load_jsonl(valid), [{"ok": True}])
251
252
253
class DevloopScorecardTest(unittest.TestCase):
254
    def setUp(self) -> None:
255
        self.devloop = load_devloop()
256
257
    def test_scorecard_lines_explain_missing_book_and_targets(self) -> None:
258
        import json
259
        import tempfile
260
261
        with tempfile.TemporaryDirectory() as tmp:
262
            root = Path(tmp)
263
            self.assertIn("config/targets.json missing", self.devloop.scorecard_lines([], root)[0])
264
265
            (root / "config").mkdir()
266
            (root / "config" / "targets.json").write_text(json.dumps({"schema_version": 1, "targets": []}))
267
            self.assertIn("baselines/git/CURRENT missing", self.devloop.scorecard_lines([], root)[0])
268
269
    def test_scorecard_lines_render_verdicts_from_current_book(self) -> None:
270
        import json
271
        import tempfile
272
273
        with tempfile.TemporaryDirectory() as tmp:
274
            root = Path(tmp)
275
            (root / "config").mkdir()
276
            (root / "config" / "targets.json").write_text(
277
                json.dumps(
278
                    {
279
                        "schema_version": 1,
280
                        "targets": [
281
                            {
282
                                "id": "t",
283
                                "lane": "core",
284
                                "profile": "smoke",
285
                                "scenario": "tiny_text",
286
                                "operation": "status.clean",
287
                                "subject": "oak_local",
288
                                "comparator": "git",
289
                                "metric": "elapsed_ms",
290
                                "goal": {"type": "ratio_at_most", "value": 0.8},
291
                                "guards": {},
292
                            }
293
                        ],
294
                    }
295
                )
296
            )
297
            book_dir = root / "baselines" / "git" / "test-book"
298
            book_dir.mkdir(parents=True)
299
            (root / "baselines" / "git" / "CURRENT").write_text("test-book\n")
300
            (book_dir / "book.json").write_text(
301
                json.dumps(
302
                    {
303
                        "schema_version": 1,
304
                        "cells": [
305
                            {
306
                                "lane": "core",
307
                                "profile": "smoke",
308
                                "scenario": "tiny_text",
309
                                "operation": "status.clean",
310
                                "subject": "git",
311
                                "metric": "elapsed_ms",
312
                                "statistic": "median",
313
                                "value": 10.0,
314
                                "n": 50,
315
                            }
316
                        ],
317
                    }
318
                )
319
            )
320
            core_row = {
321
                "profile": "smoke",
322
                "scenario": "tiny_text",
323
                "operation": "status.clean",
324
                "subject": "oak_local",
325
                "subject_kind": "oak",
326
                "returncode": 0,
327
                "elapsed_ms": 5.0,
328
                "command": ["oak", "status"],
329
            }
330
            lines = self.devloop.scorecard_lines([core_row, core_row], root)
331
            self.assertEqual(len(lines), 1)
332
            self.assertIn("GREEN", lines[0])
333
334
    def test_scorecard_lines_applies_live_noise_floor_to_loaded_book(self) -> None:
335
        import json
336
        import tempfile
337
338
        with tempfile.TemporaryDirectory() as tmp:
339
            root = Path(tmp)
340
            (root / "config").mkdir()
341
            (root / "config" / "targets.json").write_text(
342
                json.dumps(
343
                    {
344
                        "schema_version": 1,
345
                        "targets": [
346
                            {
347
                                "id": "t",
348
                                "lane": "core",
349
                                "profile": "smoke",
350
                                "scenario": "tiny_text",
351
                                "operation": "status.clean",
352
                                "subject": "oak_local",
353
                                "comparator": "git",
354
                                "metric": "elapsed_ms",
355
                                "goal": {"type": "ratio_at_most", "value": 0.99},
356
                                "guards": {},
357
                            }
358
                        ],
359
                    }
360
                )
361
            )
362
            book_dir = root / "baselines" / "git" / "test-book"
363
            book_dir.mkdir(parents=True)
364
            (root / "baselines" / "git" / "CURRENT").write_text("test-book\n")
365
            (book_dir / "book.json").write_text(
366
                json.dumps(
367
                    {
368
                        "schema_version": 1,
369
                        "cells": [
370
                            {
371
                                "lane": "core",
372
                                "profile": "smoke",
373
                                "scenario": "tiny_text",
374
                                "operation": "status.clean",
375
                                "subject": "git",
376
                                "metric": "elapsed_ms",
377
                                "statistic": "median",
378
                                "value": 10.0,
379
                                "n": 50,
380
                                "noise_floor_pct": None,
381
                            }
382
                        ],
383
                    }
384
                )
385
            )
386
            core_row = {
387
                "profile": "smoke",
388
                "scenario": "tiny_text",
389
                "operation": "status.clean",
390
                "subject": "oak_local",
391
                "subject_kind": "oak",
392
                "returncode": 0,
393
                "elapsed_ms": 9.7,
394
                "command": ["oak", "status"],
395
            }
396
            no_noise = self.devloop.scorecard_lines([core_row, core_row], root, noise_floor_pct=0.0)
397
            self.assertIn("GREEN", no_noise[0])
398
            self.assertIn("`t`", no_noise[0])
399
            self.assertIn(
400
                "UNDECIDABLE",
401
                self.devloop.scorecard_lines([core_row, core_row], root, noise_floor_pct=5.0)[0],
402
            )
403
404
405
class OutputOracleGateTest(unittest.TestCase):
406
    def setUp(self):
407
        self.devloop = load_devloop()
408
409
    compare = DevloopVerdictTest.compare
410
411
    def pair(self):
412
        result = [row(subject, "diff.full.inforecall") for subject in ("oak_installed", "oak_local")]
413
        for item in result:
414
            item.update(output_oracle_version="exact-paths-v2", information_recall=1.0,
415
                        information_precision=1.0, patch_oracle_version="apply-tree-v1",
416
                        patch_apply_ok=True, patch_application_required=True, patch_input_sha256="a" * 64,
417
                        patch_expected_tree_sha256="b" * 64, patch_apply_returncode=0,
418
                        patch_instrument_sha256="d" * 64, patch_instrument_version="git version control")
419
        return result
420
421
    def test_mixed_recall_versions_are_not_compared(self):
422
        rows = self.pair()
423
        rows[0].pop("output_oracle_version")
424
        rows[1]["information_recall"] = 0.2
425
        failures = self.compare(rows)
426
        self.assertTrue(any("OUTPUT-ORACLE" in item for item in failures))
427
        self.assertFalse(any(item.startswith("RECALL") for item in failures))
428
429
    def test_exact_precision_regression_is_gated(self):
430
        rows = self.pair()
431
        rows[1]["information_precision"] = 0.5
432
        self.assertTrue(any("PRECISION" in item for item in self.compare(rows)))
433
434
    def test_patch_false_and_lost_measurement_cannot_pass(self):
435
        for value in (False,):
436
            rows = self.pair()
437
            rows[1]["patch_apply_ok"] = value
438
            self.assertTrue(any("PATCH-APPLY" in item for item in self.compare(rows)))
439
440
    def test_preexisting_patch_defect_and_unknown_instrument_are_separate(self):
441
        rows = self.pair()
442
        for item in rows:
443
            item.update(patch_apply_ok=False, patch_failure_reason="patch_result_tree_mismatch")
444
        diagnostics = []
445
        failures, _ = self.devloop.compare_subjects(rows, "oak_local", "oak_installed", "baseline", 100, 100,
446
            efficiency_gates=False, output_bytes_threshold_pct=100, diagnostics=diagnostics)
447
        self.assertFalse(failures)
448
        self.assertTrue(any("PREEXISTING" in text for text in diagnostics))
449
        rows[1].update(patch_apply_ok=None, patch_skip_reason="git_apply_instrument_timeout")
450
        diagnostics.clear()
451
        failures, _ = self.devloop.compare_subjects(rows, "oak_local", "oak_installed", "baseline", 100, 100,
452
            efficiency_gates=False, output_bytes_threshold_pct=100, diagnostics=diagnostics)
453
        self.assertFalse(failures)
454
        self.assertTrue(any("UNMEASURED" in text for text in diagnostics))
455
456
    def test_different_failed_patch_is_not_preexisting(self):
457
        for difference in ({"patch_input_sha256": "c" * 64}, {"patch_expected_tree_sha256": "c" * 64},
458
                           {"patch_input_sha256": None}, {"patch_apply_returncode": 2},
459
                           {"patch_instrument_sha256": "c" * 64}, {"patch_instrument_version": "changed"}):
460
            rows = self.pair()
461
            for item in rows:
462
                item.update(patch_apply_ok=False, patch_failure_reason="patch_application_failed", patch_apply_returncode=1)
463
            rows[1].update(difference)
464
            self.assertTrue(any("PATCH-APPLY" in text for text in self.compare(rows)))
465
466
    def test_skipped_instrument_is_not_generic_no_success_regression(self):
467
        rows = [row("oak_installed", "workflow.total"),
468
                {**row("oak_local", "workflow.total", returncode=77), "skip_reason": "git_unavailable"}]
469
        diagnostics = []
470
        failures, _ = self.devloop.compare_subjects(rows, "oak_local", "oak_installed", "baseline", 100, 100,
471
            efficiency_gates=False, output_bytes_threshold_pct=100, diagnostics=diagnostics)
472
        self.assertFalse(failures)
473
        self.assertTrue(any("UNMEASURED" in text for text in diagnostics))
474
        rows[1]["returncode"] = 1
475
        self.assertTrue(self.compare(rows))
476
477
    def test_optional_patch_unknown_is_informational(self):
478
        rows = self.pair()
479
        rows[1].update(patch_apply_ok=None, patch_application_required=False, patch_skip_reason="patch_output_exceeds_capture_limit")
480
        diagnostics = []
481
        failures, _ = self.devloop.compare_subjects(rows, "oak_local", "oak_installed", "baseline", 100, 100,
482
            efficiency_gates=False, output_bytes_threshold_pct=100, diagnostics=diagnostics)
483
        self.assertFalse(failures)
484
        self.assertTrue(any(text.startswith("INFO PATCH-APPLY") for text in diagnostics))
485
        self.assertFalse(any(text.startswith("UNMEASURED") for text in diagnostics))
486
487
    def test_unmeasured_baseline_needs_no_patch_instrument_identity(self):
488
        rows = self.pair()
489
        rows[0].update(patch_apply_ok=None, patch_skip_reason="subject_diff_command_failed")
490
        rows[0].pop("patch_instrument_sha256"); rows[0].pop("patch_instrument_version")
491
        diagnostics = []
492
        failures, _ = self.devloop.compare_subjects(rows, "oak_local", "oak_installed", "baseline", 100, 100,
493
            efficiency_gates=False, output_bytes_threshold_pct=100, diagnostics=diagnostics)
494
        self.assertFalse(failures)
495
        self.assertEqual(diagnostics, [])
496
497
    def test_absent_target_operation_remains_a_regression(self):
498
        self.assertTrue(any("NO SUCCESS" in text for text in self.compare([row("oak_installed", "workflow.total")])))
499
500
    def test_git_guardrail_different_patch_failure_is_not_prefix_suppressed(self):
501
        rows = self.pair()
502
        for item in rows:
503
            item.update(patch_apply_ok=False, patch_failure_reason="patch_application_failed", patch_apply_returncode=1)
504
        breach = f"PATCH-APPLY `{rows[0]['scenario']}/diff.full.inforecall`: exact patch application failed"
505
        self.assertTrue(self.devloop.preexisting_patch_breach(breach, rows))
506
        rows[1]["patch_input_sha256"] = "c" * 64
507
        self.assertFalse(self.devloop.preexisting_patch_breach(breach, rows))
508
509
    def test_repeated_failures_compare_matching_run_inputs(self):
510
        rows = []
511
        for run in (0, 1):
512
            for item in self.pair():
513
                item.update(run=run, patch_apply_ok=False, patch_failure_reason="patch_application_failed",
514
                            patch_apply_returncode=1, patch_input_sha256=str(run) * 64)
515
                rows.append(item)
516
        self.assertFalse(self.compare(rows))
517
        rows[-1]["patch_input_sha256"] = "f" * 64
518
        self.assertTrue(self.compare(rows))
519
520
    def test_cli_reports_preexisting_and_unmeasured_distinctly(self):
521
        import contextlib
522
        import io
523
        for value, expected_exit, expected_note in ((False, 0, "PREEXISTING"), (None, 2, "UNMEASURED")):
524
            rows = self.pair()
525
            for item in rows:
526
                item.update(patch_apply_ok=False, patch_failure_reason="patch_result_tree_mismatch", returncode=1)
527
            rows[1]["patch_apply_ok"] = value
528
            if value is None:
529
                rows[1].update(returncode=0, patch_skip_reason="git_apply_instrument_timeout")
530
            with tempfile.TemporaryDirectory() as tmp:
531
                argv = ["devloop.py", "--lanes", "core", "--skip-noise-floor", "--skip-build",
532
                        "--oak-baseline-bin", "/bin/echo", "--oak-local-bin", "/bin/echo", "--results", tmp]
533
                output = io.StringIO()
534
                with mock.patch.object(sys, "argv", argv), mock.patch.object(self.devloop, "run_lane"), mock.patch.object(self.devloop, "load_jsonl", return_value=rows), mock.patch.object(self.devloop.oakbench_remotes, "print_remote_preflight_warnings"), contextlib.redirect_stdout(output):
535
                    result = self.devloop.main()
536
                self.assertEqual(result, expected_exit, output.getvalue())
537
                self.assertIn(expected_note, output.getvalue())
538
                self.assertNotIn("**REGRESSED**", output.getvalue())
539
540
    def test_cli_new_bad_patch_is_a_new_git_guardrail_breach(self):
541
        import contextlib
542
        import io
543
        rows = self.pair()
544
        rows.append({**rows[0], "subject": "git"})
545
        for item in rows[:2]:
546
            item.update(patch_apply_ok=False, patch_failure_reason="patch_application_failed", patch_apply_returncode=1)
547
        rows[1]["patch_input_sha256"] = "c" * 64
548
        with tempfile.TemporaryDirectory() as tmp:
549
            argv = ["devloop.py", "--lanes", "core", "--skip-noise-floor", "--skip-build",
550
                    "--oak-baseline-bin", "/bin/echo", "--oak-local-bin", "/bin/echo", "--results", tmp]
551
            output = io.StringIO()
552
            with mock.patch.object(sys, "argv", argv), mock.patch.object(self.devloop, "run_lane"), mock.patch.object(self.devloop, "load_jsonl", return_value=rows), mock.patch.object(self.devloop.oakbench_remotes, "print_remote_preflight_warnings"), contextlib.redirect_stdout(output):
553
                self.assertEqual(self.devloop.main(), 1)
554
            self.assertIn("New Git-guardrail breaches introduced by this changeset (1)", output.getvalue())
555
            self.assertIn("Known Git gaps, pre-existing in the baseline (0)", output.getvalue())
556
557
    def test_patch_oracle_versions_must_match(self):
558
        rows = self.pair()
559
        rows[1]["patch_oracle_version"] = "different"
560
        self.assertTrue(any("PATCH-ORACLE" in item for item in self.compare(rows)))
561
562
    def test_omitting_target_patch_row_cannot_hide_lost_proof(self):
563
        rows = self.pair()[:1]
564
        rows.append(row("oak_local", "status.clean"))
565
        self.assertTrue(any("PATCH-APPLY" in item for item in self.compare(rows)))
566
567
568
if __name__ == "__main__":
569
    unittest.main()