core/scripts/verify_lane_shas.py

"""Verify ADR-0092..0099 lane SHA-256 pins.

Each ADR lane writes a deterministic JSON report. This script runs
every pinned lane and asserts the SHA-256 of the report bytes matches
the value pinned below. Pinned SHAs come from the commits that landed
each ADR:

- ADR-0092 reviewer_registry          afdd2ee
- ADR-0095 miner_loop_closure         7dc7e9d
- ADR-0093 domain_contract_validation 7784c39
- ADR-0096 fabrication_control        d7713b0
- ADR-0098 demo_composition           4f640af
- ADR-0099 public_demo                bfb54fb

Update the pins with ``--update`` when an ADR-tracked change to the
lane is intentional. The diff between the in-tree pin and the freshly
computed SHA is the audit trail.

Usage:

    python scripts/verify_lane_shas.py            # verify, exit non-zero on mismatch
    python scripts/verify_lane_shas.py --update   # rewrite the pin block in this file
    python scripts/verify_lane_shas.py --json     # machine-readable report
"""

from __future__ import annotations

import argparse
import hashlib
import json
import subprocess
import sys
import tempfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any


REPO_ROOT = Path(__file__).resolve().parent.parent


# ---------------------------------------------------------------------------
# Pinned SHA-256 of each lane's canonical report bytes.
# Generated by running the lane's runner module; update with --update.
# ---------------------------------------------------------------------------
PINNED_SHAS: dict[str, str] = {
    "reviewer_registry": "681a2aab5aa4ffd58cd837ce5673c8b2a9545b570117aec3c02726a12f6876e6",
    "miner_loop_closure": "9f071733abe7dcacf759f928548ce738fb639af3fd6e4c621a651b306d7e77ce",
    "domain_contract_validation": "f9c06cdeea8fb36a0d3c320007618c3afc92d67702ef31bd36ebd9ae9ced473f",
    "fabrication_control_summary": "01e1b6b711141f2b4a14551d7df3ea482d8d6dd7b364a25c509f4f8d08cda8a8",
    "demo_composition": "27d838241bf3ed9e15d0e918ec6d89a823494d7e17c2dab9777825af7188f20f",
    "public_demo": "4be6f47509435a24984713acfcebd88e61f4e1278096fa5dc88a09e8af2f87ba",
}


@dataclass(frozen=True, slots=True)
class LaneSpec:
    lane_id: str
    runner_module: str
    report_relative: str
    accepts_report_flag: bool = True
    extra_args: tuple[str, ...] = field(default_factory=tuple)

    @property
    def runner_path(self) -> Path:
        return REPO_ROOT / self.runner_module

    @property
    def canonical_report(self) -> Path:
        return REPO_ROOT / self.report_relative


LANE_SPECS: tuple[LaneSpec, ...] = (
    LaneSpec(
        lane_id="reviewer_registry",
        runner_module="evals/reviewer_registry/runner.py",
        report_relative="evals/reviewer_registry/results/v1_dev.json",
    ),
    LaneSpec(
        lane_id="miner_loop_closure",
        runner_module="evals/miner_loop_closure/runner.py",
        report_relative="evals/miner_loop_closure/results/v1_dev.json",
    ),
    LaneSpec(
        lane_id="domain_contract_validation",
        runner_module="evals/domain_contract_validation/runner.py",
        report_relative="evals/domain_contract_validation/results/v1_dev.json",
    ),
    LaneSpec(
        # fabrication_control runner emits per-split + summary reports
        # into its own results/ directory; ``--lane-dir`` is the
        # accepted redirect, not ``--report``. We inspect the
        # ``v1_summary.json`` written by that runner.
        lane_id="fabrication_control_summary",
        runner_module="evals/fabrication_control/runner.py",
        report_relative="evals/fabrication_control/results/v1_summary.json",
        accepts_report_flag=False,
    ),
    LaneSpec(
        lane_id="demo_composition",
        runner_module="evals/demo_composition/runner.py",
        report_relative="evals/demo_composition/results/v1_dev.json",
    ),
    LaneSpec(
        lane_id="public_demo",
        runner_module="evals/public_demo/runner.py",
        report_relative="evals/public_demo/results/v1_dev.json",
    ),
)


def _invoke_runner(spec: LaneSpec, *, target_path: Path | None = None) -> Path:
    """Execute a lane runner as a subprocess and return the report path.

    Subprocess isolation guarantees each lane runs in a clean Python
    state — relevant for adapters that mutate global runtime config or
    cache pack loads at module import time.
    """
    import os

    env = {"PYTHONPATH": str(REPO_ROOT), **os.environ}
    args: list[str] = [sys.executable, str(spec.runner_path)]
    if target_path is not None and spec.accepts_report_flag:
        args.extend(["--report", str(target_path)])
    args.extend(spec.extra_args)
    result = subprocess.run(
        args,
        cwd=REPO_ROOT,
        env=env,
        capture_output=True,
        text=True,
        timeout=300,
    )
    if result.returncode != 0:
        raise RuntimeError(
            f"lane runner {spec.lane_id} exited non-zero "
            f"(code={result.returncode})\nSTDOUT:\n{result.stdout}\n"
            f"STDERR:\n{result.stderr}"
        )
    # If the runner accepts --report, the report lives at target_path;
    # otherwise the runner wrote to its canonical in-tree location.
    if target_path is not None and spec.accepts_report_flag:
        report_path = target_path
    else:
        report_path = spec.canonical_report
    if not report_path.exists():
        raise RuntimeError(
            f"lane {spec.lane_id} runner returned 0 but report not found at {report_path}"
        )
    return report_path


def _sha_of(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


@dataclass(frozen=True, slots=True)
class LaneVerification:
    lane_id: str
    pinned_sha: str
    actual_sha: str
    matched: bool
    report_path: str
    error: str | None = None

    def as_dict(self) -> dict[str, Any]:
        return {
            "lane_id": self.lane_id,
            "pinned_sha": self.pinned_sha,
            "actual_sha": self.actual_sha,
            "matched": self.matched,
            "report_path": self.report_path,
            "error": self.error,
        }


def verify_all(*, ephemeral: bool = True) -> list[LaneVerification]:
    """Run every pinned lane and verify its SHA matches the pin.

    ``ephemeral`` writes the report to a temp directory so the in-tree
    artifacts are not perturbed during verification. Disable to rewrite
    the canonical results path (used by ``--update``).
    """
    results: list[LaneVerification] = []
    for spec in LANE_SPECS:
        pinned = PINNED_SHAS.get(spec.lane_id, "")
        try:
            if ephemeral:
                with tempfile.TemporaryDirectory(prefix=f"lane_{spec.lane_id}_") as d:
                    target = Path(d) / "report.json"
                    report_path = _invoke_runner(spec, target_path=target)
                    actual = _sha_of(report_path)
            else:
                report_path = _invoke_runner(spec)
                actual = _sha_of(report_path)
        except Exception as exc:  # pylint: disable=broad-except
            results.append(
                LaneVerification(
                    lane_id=spec.lane_id,
                    pinned_sha=pinned,
                    actual_sha="",
                    matched=False,
                    report_path=str(spec.canonical_report),
                    error=f"{type(exc).__name__}: {exc}",
                )
            )
            continue
        results.append(
            LaneVerification(
                lane_id=spec.lane_id,
                pinned_sha=pinned,
                actual_sha=actual,
                matched=(actual == pinned),
                report_path=str(report_path),
            )
        )
    return results


_PIN_BLOCK_START = "PINNED_SHAS: dict[str, str] = {"
_PIN_BLOCK_END = "}"


def _rewrite_pins(new_pins: dict[str, str]) -> None:
    """Rewrite the PINNED_SHAS block in this file with new values."""
    text = Path(__file__).read_text(encoding="utf-8")
    start = text.index(_PIN_BLOCK_START)
    # Find the *next* closing brace after start.
    rel_end = text[start:].index(_PIN_BLOCK_END)
    end = start + rel_end + 1
    new_block_lines = [_PIN_BLOCK_START]
    for lane_id, sha in new_pins.items():
        new_block_lines.append(f'    "{lane_id}": "{sha}",')
    new_block_lines.append("}")
    new_block = "\n".join(new_block_lines)
    Path(__file__).write_text(text[:start] + new_block + text[end:], encoding="utf-8")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="verify ADR lane SHAs")
    parser.add_argument(
        "--update",
        action="store_true",
        help="rewrite PINNED_SHAS in this file with freshly-computed values",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="emit machine-readable JSON report",
    )
    args = parser.parse_args(argv)

    if args.update:
        results = verify_all(ephemeral=False)
        new_pins = {r.lane_id: r.actual_sha for r in results if not r.error}
        _rewrite_pins(new_pins)
        if args.json:
            print(json.dumps({"updated": new_pins}, indent=2, sort_keys=True))
        else:
            print("Updated PINNED_SHAS:")
            for lane_id, sha in new_pins.items():
                print(f"  {lane_id:>32}: {sha}")
        return 0

    results = verify_all()
    if args.json:
        payload = {
            "total": len(results),
            "matched": sum(1 for r in results if r.matched),
            "mismatched": [r.as_dict() for r in results if not r.matched],
            "results": [r.as_dict() for r in results],
        }
        print(json.dumps(payload, indent=2, sort_keys=True))
    else:
        width = max(len(r.lane_id) for r in results)
        for r in results:
            mark = "✓" if r.matched else "✗"
            print(f"  {mark} {r.lane_id:<{width}}  {r.actual_sha[:16]}..", end="")
            if not r.matched:
                if r.error:
                    print(f"  ERROR: {r.error}")
                else:
                    print(f"  expected {r.pinned_sha[:16]}..")
            else:
                print()
        total = len(results)
        matched = sum(1 for r in results if r.matched)
        print(f"\nlanes: {matched}/{total} match pinned SHAs")

    return 0 if all(r.matched for r in results) else 1


if __name__ == "__main__":
    sys.exit(main())