core/scripts/verify_lane_shas.py

"""Verify ADR-0092..0104 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.

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 os
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any


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

# Per-lane subprocess wall-clock budget. Overridable so CI can raise it under
# known cold/contended-runner conditions without a code change — same knob
# shape as CORE_SHOWCASE_HARD_BUDGET / CORE_SHOWCASE_SKIP_BUDGET elsewhere in
# this repo. The job-level workflow timeout (lane-shas.yml) is a separate,
# larger ceiling; this is what actually fires first when one lane hangs.
LANE_TIMEOUT_S = int(os.environ.get("CORE_LANE_VERIFY_TIMEOUT_S", "900"))


PINNED_SHAS: dict[str, str] = {
    "reviewer_registry": "681a2aab5aa4ffd58cd837ce5673c8b2a9545b570117aec3c02726a12f6876e6",
    "miner_loop_closure": "537094fe21d7e6cfbaf42bfc32b82d669fa9bb05a132d2bc93c72b3ceb7762a6",
    "curriculum_loop_closure": "cb94ca0042d78ec2624129ff6493d52e767b69feea32d2997b85d88f1c0883af",
    "domain_contract_validation": "98ace04e3f02bbc5a8ad655bb6593c3f1ee64cb67014f1122fe6c3c85f48d22f",
    "fabrication_control_summary": "01e1b6b711141f2b4a14551d7df3ea482d8d6dd7b364a25c509f4f8d08cda8a8",
    "demo_composition": "f0611a2ce41721dd40767fc6a83a08470d3c7fd7fc8f1ae8ba003abf8a25ec97",
    "public_demo": "da7fad654e77aac4573a6fcf6e9eaaf84540be8e135d2e033d9cfd15119df3fc",
    "math_teaching_corpus_v1": "eaf160d145da29f9050ede8d58bf111b0f651dd40aeae9201857d0b97e014dd4",
    "deductive_logic_v1": "97a230949016e38d5e3f37a69e4245b320575ee70e5af92ff7607f7b05f74b5f",
    # Re-pinned 2026-07-27: the report now carries the SERVED PROSE
    # (``surface_sha256`` + ``surfaces`` per split) as well as verdict counts.
    # Old: 0b461a5a49c8f8260ca87d0c9c9f9a17232bd1fdedd982e34649eedf9cca30b5
    # The move is the payload growing, not a behaviour change — verdicts are
    # identical (166/166 correct, wrong=0). Before this, a renderer emitting
    # "all SABOTAGE_dogs are SABOTAGE_animals" left all 11 pins byte-identical.
    "deduction_serve_v1": "c855d55cf316471fdfe092aa0d5c954e5ceb6f30c9a8db283e0b9a5d5e8b419a",
    "curriculum_serve_v1": "d9e7ba500f040b865870413a940ee9a49910ac22e1a89c9feec1a60bdd2513f1",
}


@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)
    # Run via ``python -m pkg.mod`` instead of ``python path/to/runner.py``.
    # Required when the runner's own directory holds a module that would shadow
    # an absolute import in script mode — e.g. the deductive lane's local
    # ``generate.py`` shadows the ``generate`` package when run as a script.
    run_as_module: bool = False

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

    @property
    def runner_dotted(self) -> str:
        return self.runner_module.removesuffix(".py").replace("/", ".")

    @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="curriculum_loop_closure",
        runner_module="evals/curriculum_loop_closure/runner.py",
        report_relative="evals/curriculum_loop_closure/results/v1_dev.json",
        accepts_report_flag=False,
    ),
    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(
        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",
    ),
    LaneSpec(
        lane_id="math_teaching_corpus_v1",
        runner_module="evals/math_teaching_corpus/v1/runner.py",
        report_relative="evals/math_teaching_corpus/v1/report.json",
        accepts_report_flag=False,
    ),
    LaneSpec(
        lane_id="deductive_logic_v1",
        runner_module="evals/deductive_logic/runner.py",
        report_relative="evals/deductive_logic/report.json",
        run_as_module=True,
    ),
    LaneSpec(
        lane_id="deduction_serve_v1",
        runner_module="evals/deduction_serve/runner.py",
        report_relative="evals/deduction_serve/report.json",
        run_as_module=True,
    ),
    LaneSpec(
        lane_id="curriculum_serve_v1",
        runner_module="evals/curriculum_serve/runner.py",
        report_relative="evals/curriculum_serve/report.json",
        run_as_module=True,
    ),
)


def _invoke_runner(spec: LaneSpec, *, target_path: Path | None = None) -> Path:
    env = {"PYTHONPATH": str(REPO_ROOT), **os.environ}
    # Force hermetic engine_state for every lane so local lived state and CI
    # workspaces cannot change demo / showcase report bytes.
    env.setdefault(
        "CORE_ENGINE_STATE_DIR",
        tempfile.mkdtemp(prefix=f"lane_{spec.lane_id}_engine_"),
    )
    if spec.run_as_module:
        args = [sys.executable, "-m", spec.runner_dotted]
    else:
        args = [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=LANE_TIMEOUT_S,
    )
    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 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
    timed_out: bool = False

    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,
            "timed_out": self.timed_out,
        }


def verify_all(*, ephemeral: bool = True, stream: bool = False) -> list[LaneVerification]:
    results: list[LaneVerification] = []
    for spec in LANE_SPECS:
        pinned = PINNED_SHAS.get(spec.lane_id, "")
        if stream:
            print(f"  -> {spec.lane_id} ...", end="", flush=True)
        started = time.monotonic()
        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 subprocess.TimeoutExpired as exc:
            elapsed = time.monotonic() - started
            if stream:
                print(f" TIMEOUT after {elapsed:.0f}s (budget {LANE_TIMEOUT_S}s)", flush=True)
            results.append(
                LaneVerification(
                    lane_id=spec.lane_id,
                    pinned_sha=pinned,
                    actual_sha="",
                    matched=False,
                    report_path=str(spec.canonical_report),
                    error=(
                        f"TimeoutExpired: lane exceeded {LANE_TIMEOUT_S}s "
                        f"(CORE_LANE_VERIFY_TIMEOUT_S) — likely runner "
                        f"contention/cold-start, not a content change: {exc}"
                    ),
                    timed_out=True,
                )
            )
            continue
        except Exception as exc:
            elapsed = time.monotonic() - started
            if stream:
                print(f" ERROR after {elapsed:.0f}s", flush=True)
            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
        if stream:
            elapsed = time.monotonic() - started
            print(f" done ({elapsed:.0f}s)", flush=True)
        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:
    text = Path(__file__).read_text(encoding="utf-8")
    start = text.index(_PIN_BLOCK_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")
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args(argv)

    if args.update:
        results = verify_all(ephemeral=False, stream=not args.json)
        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(stream=not args.json)
    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)
        timed_out = [r.lane_id for r in results if r.timed_out]
        print(f"\nlanes: {matched}/{total} match pinned SHAs")
        if timed_out:
            print(
                "\nremediation (timeout, not a content mismatch):\n"
                f"  lane(s) {', '.join(timed_out)} exceeded the "
                f"{LANE_TIMEOUT_S}s per-lane budget (CORE_LANE_VERIFY_TIMEOUT_S).\n"
                "  Do NOT re-pin on a timeout alone — that only masks runner\n"
                "  contention/cold-start under load. Re-run first; if it recurs, raise\n"
                "  CORE_LANE_VERIFY_TIMEOUT_S for this job or investigate Act runner load\n"
                "  (see docs/ci-optimization.md)."
            )
        if matched < total and len(timed_out) < (total - matched):
            print(
                "\nremediation (content drift):\n"
                "  if the drift is intentional (e.g. you touched core/cognition/result.py,\n"
                "  chat/runtime.py, generate/realizer.py, capability registries, or other\n"
                "  lane-affecting code), re-pin with:\n"
                "      python scripts/verify_lane_shas.py --update\n"
                "  then run `python scripts/generate_claims.py` and commit both changes.\n"
                "  if the drift is unintentional, investigate the upstream change before re-pinning."
            )

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


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