oak/benchmarks

"""devloop verdicts replayed on REAL identical-binary rows (QA P1-1, P1-2, detection power).

run_lane is replaced by a writer that serves slimmed real rows captured by

independent QA (tests/fixtures/devloop_real_rows), optionally transformed to

inject a known effect, so the whole verdict path (screen, confirmation,

guardrail, fail-closed) runs exactly as in production.

"""

from __future__ import annotations

import copy

import gzip

import importlib.util

import json

import sys

import tempfile

import unittest

from pathlib import Path

from unittest import mock

ROOT = Path(__file__).resolve().parents[1]

FIXTURES = ROOT / "tests" / "fixtures" / "devloop_real_rows"

sys.path.insert(0, str(ROOT / "scripts"))

def load_devloop():

    spec = importlib.util.spec_from_file_location("devloop_real_rows", ROOT / "scripts" / "devloop.py")

    module = importlib.util.module_from_spec(spec)

    spec.loader.exec_module(module)

    return module

def rows(name: str) -> list[dict]:

    with gzip.open(FIXTURES / f"{name}.jsonl.gz", "rt") as handle:

        return [json.loads(line) for line in handle if line.strip()]

def pooled(profile: str) -> list[dict]:

    """Two independent real A/A runs pooled as one 10-run confirmation sample."""

    first = rows(f"aa-{profile}")

    second = copy.deepcopy(rows(f"aa2-{profile}"))

    for row in second:

        row["run"] = row["run"] + 100

    return first + second

def shift(data: list[dict], pct: float, cell: tuple[str, str] | None = None, ms: float = 0.0) -> list[dict]:

    data = copy.deepcopy(data)

    for row in data:

        if row.get("subject") != "oak_local" or not isinstance(row.get("elapsed_ms"), (int, float)):

            continue

        if cell and (row["scenario"], row["operation"]) != cell:

            continue

        row["elapsed_ms"] = row["elapsed_ms"] * (1 + pct / 100.0) + ms

    return data

class DevloopRealRowsTest(unittest.TestCase):

    def run_devloop(self, *extra, stage1=None, confirm=None, runs_filter=None):

        devloop = load_devloop()

        stage1 = stage1 or {}

        confirm = confirm or {}

        calls = []

        def fake_lane(command, label):

            calls.append(label)

            results = Path(command[command.index("--results") + 1])

            profile = command[command.index("--profile") + 1]

            subjects = command[command.index("--subjects") + 1].split(",")

            data = (confirm if "--scenarios" in command else stage1)[profile]

            if "--scenarios" in command:

                wanted = set(command[command.index("--scenarios") + 1].split(","))

                data = [row for row in data if row["scenario"] in wanted]

            data = [row for row in data if row["subject"] in subjects]

            if runs_filter is not None:

                data = [row for row in data if row["run"] in runs_filter]

            results.mkdir(parents=True, exist_ok=True)

            (results / "latest.jsonl").write_text("".join(json.dumps(row) + "\n" for row in data))

            return 1.0

        with tempfile.TemporaryDirectory() as tmp:

            argv = ["devloop.py", "--skip-build", "--oak-baseline-bin", sys.executable, "--oak-local-bin", sys.executable,

                    "--lanes", "core", "--core-profiles", "micro,smoke", "--results", tmp, *extra]

            with mock.patch.object(sys, "argv", argv), mock.patch.object(devloop, "run_lane", side_effect=fake_lane), \

                 mock.patch("builtins.print"):

                code = devloop.main()

            out = next(Path(tmp).iterdir())

            verdict = (out / "verdict.md").read_text()

            cells = json.loads((out / "cells.json").read_text())

        return code, verdict, cells, calls

    def oak_cells(self, cells):

        return {(c["lane"], c["scenario"], c["operation"]): c for c in cells if c["comparison"] == "oak_local_vs_oak_installed"}

    def test_identical_binary_guardrail_crossing_is_not_a_new_breach(self):

        # aa2 is the reviewer's run that the pre-fix devloop called REGRESSED

        # ("new" few_large_binaries/branch.create Git breach, identical binaries).

        stage1 = {"micro": rows("aa2-micro"), "smoke": rows("aa2-smoke")}

        confirm = {"micro": pooled("micro"), "smoke": pooled("smoke")}

        code, verdict, cells, _ = self.run_devloop(stage1=stage1, confirm=confirm)

        self.assertEqual(code, 0, verdict)

        self.assertIn("**PASS**", verdict)

        self.assertIn("## New Git-guardrail breaches introduced by this changeset (0)", verdict)

        oak = self.oak_cells(cells)

        self.assertFalse([k for k, c in oak.items() if c["verdict"] in ("regression", "improvement")])

        for cell in oak.values():

            if cell["verdict"] == "no_change":

                self.assertIsNotNone(cell["mde_pct"], cell)  # never a bare no_change

        self.assertIn("at 80% power", verdict)

    def test_identical_binary_without_confirmation_never_claims(self):

        stage1 = {"micro": rows("aa-micro"), "smoke": rows("aa-smoke")}

        code, verdict, cells, calls = self.run_devloop("--no-confirm", stage1=stage1)

        # Without confirmation, n=5 screening candidates fail CLOSED (UNMEASURED),

        # never as a claim: zero Oak-vs-Oak regressions or improvements.

        self.assertIn(code, (0, 2), verdict)

        self.assertNotIn("**REGRESSED**", verdict)

        self.assertFalse([k for k, c in self.oak_cells(cells).items() if c["verdict"] in ("regression", "improvement")])

        self.assertFalse([label for label in calls if label.startswith("confirmation")])

    def test_missing_aa_control_fails_closed(self):

        # --skip-noise-floor with a +10 ms slowdown: every cell lacks a floor.

        stage1 = {"micro": shift(rows("aa-micro"), 0, ms=10), "smoke": shift(rows("aa-smoke"), 0, ms=10)}

        code, verdict, _, _ = self.run_devloop("--skip-noise-floor", stage1=stage1)

        self.assertEqual(code, 2, verdict)

        self.assertIn("**UNMEASURED**", verdict)

    def test_too_few_runs_fail_closed(self):

        stage1 = {"micro": shift(rows("aa-micro"), 0, ms=10), "smoke": shift(rows("aa-smoke"), 0, ms=10)}

        code, verdict, _, _ = self.run_devloop("--runs", "2", "--no-confirm", stage1=stage1, runs_filter={0, 1})

        self.assertEqual(code, 2, verdict)

        self.assertIn("**UNMEASURED**", verdict)

    def test_single_cell_twenty_percent_on_ten_ms_op_is_confirmed(self):

        # tiny_text/branch.create is a >= 10 ms Oak op (about 13-16 ms here).

        cell = ("tiny_text", "branch.create")

        stage1 = {"micro": shift(rows("aa2-micro"), 20, cell), "smoke": rows("aa2-smoke")}

        confirm = {"micro": shift(pooled("micro"), 20, cell), "smoke": pooled("smoke")}

        baseline = [r["elapsed_ms"] for r in stage1["micro"] if r["subject"] == "oak_installed" and (r["scenario"], r["operation"]) == cell]

        self.assertGreaterEqual(sorted(baseline)[len(baseline) // 2], 10.0)

        code, verdict, cells, calls = self.run_devloop(stage1=stage1, confirm=confirm)

        self.assertEqual(code, 1, verdict)

        oak = self.oak_cells(cells)

        target = oak[("core:micro", *cell)]

        self.assertEqual(target["verdict"], "regression", target)

        self.assertEqual(target["stage"], "confirmation")

        self.assertTrue([label for label in calls if label.startswith("confirmation (core:micro")])

        others = [k for k, c in oak.items() if c["verdict"] in ("regression", "improvement") and k[1:] != cell]

        self.assertEqual(others, [])

if __name__ == "__main__":

    unittest.main()