#!/usr/bin/env python3

"""Verify a third-party replay against a published repro bundle.

Third-party flow: regenerate fixtures byte-identically from the bundle's

seeds, re-run the suite, then run this verifier with the replay rows.

Machine-INDEPENDENT metrics (tokens, output bytes, tool calls) must match

the original rows EXACTLY (sorted value multisets per group). Machine-

DEPENDENT metrics (elapsed_ms) verify by ratio: the oak/git median-ratio

bootstrap CI on each side must overlap.

The bundle pins only hashes of the original raw rows, not the rows

themselves. Pass the original row files with --original to enable exact

machine-independent verification; without them those groups are

unverifiable and the best possible verdict is COMPARABLE.

Coverage is mandatory for full verification: EVERY raw_inputs entry pinned

by the bundle must be matched (by sha256) by a provided --original file.

Any pinned input not covered is reported as

"unverifiable:bundle_input_not_provided:<path>" and caps the verdict at

COMPARABLE -- a replay of a subset of the bundle's inputs can never reach

VERIFIED or VERIFIED-EXACT.

Verdicts and exit codes:

  VERIFIED-EXACT  all machine-independent values equal + ratio CIs overlap

                  + every pinned raw input covered by --original           -> 0

  VERIFIED        machine-independent values equal; ratio data absent;

                  every pinned raw input covered by --original             -> 0

  COMPARABLE      ratios overlap (or absent) but some machine-independent

                  values unverifiable or some pinned raw input uncovered   -> 3

  DIVERGENT       any exact mismatch or non-overlapping ratio CI           -> 1

  usage error                                                              -> 2

Null means unmeasured (ADR-0002): a metric absent on one side is reported

as unverifiable, never as a silent zero-mismatch.

Usage:

  python3 scripts/repro_verify.py replay-rows.jsonl ... --bundle bundle.json \

      [--original original-rows.jsonl ...] [--confidence 0.95]

"""

from __future__ import annotations

import argparse

import json

import math

import sys

from pathlib import Path

from typing import Any, Callable, Optional

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

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

from oakbench.baseline_book import sha256_file  # noqa: E402

from oakbench.rows import number_or_none, output_bytes, row_returncode, token_total, tool_calls  # noqa: E402

from oakbench.stats import ratio_ci  # noqa: E402

GroupKey = tuple[Optional[str], str, str, str]

MACHINE_INDEPENDENT_METRICS: tuple[tuple[str, Callable[[dict[str, Any]], Optional[float]]], ...] = (

    ("tokens", token_total),

    ("output_bytes", output_bytes),

    ("tool_calls", tool_calls),

)

VERDICT_EXIT_CODES = {"VERIFIED-EXACT": 0, "VERIFIED": 0, "DIVERGENT": 1, "COMPARABLE": 3}

DEFAULT_CONTENT_SOURCE_STRENGTH = {

    "git_head_payload_sha256": 3,

    "git_lfs_worktree_payload_sha256": 1,

    "worktree_payload_sha256": 1,

    "mixed_payload_sources": 1,

}

def load_jsonl(paths: list[Path]) -> list[dict[str, Any]]:

    rows: list[dict[str, Any]] = []

    for path in paths:

        with path.open() as fh:

            for raw_line in fh:

                line = raw_line.strip()

                if line:

                    rows.append(json.loads(line))

    return rows

def group_key(row: dict[str, Any]) -> GroupKey:

    profile = row.get("profile")

    return (

        None if profile is None else str(profile),

        str(row.get("scenario") or ""),

        str(row.get("operation") or ""),

        str(row.get("subject") or ""),

    )

def group_rows(rows: list[dict[str, Any]]) -> dict[GroupKey, list[dict[str, Any]]]:

    grouped: dict[GroupKey, list[dict[str, Any]]] = {}

    for row in rows:

        if row_returncode(row) != 0:

            continue

        grouped.setdefault(group_key(row), []).append(row)

    return grouped

def metric_values(rows: list[dict[str, Any]], accessor: Callable[[dict[str, Any]], Optional[float]]) -> list[float]:

    values = [value for row in rows if (value := accessor(row)) is not None]

    return sorted(values)

def elapsed_values(rows: list[dict[str, Any]]) -> list[float]:

    return [value for row in rows if (value := number_or_none(row.get("elapsed_ms"))) is not None]

def row_subject_kind(rows: list[dict[str, Any]], fallback_subject: str) -> str:

    first_kind = rows[0].get("subject_kind") if rows else None

    if isinstance(first_kind, str) and first_kind:

        return first_kind

    return "oak" if fallback_subject.startswith("oak") else fallback_subject

def subjects_for_kind(

    groups: dict[GroupKey, list[dict[str, Any]]],

    profile: Optional[str],

    scenario: str,

    operation: str,

    wanted_kind: str,

) -> set[str]:

    subjects: set[str] = set()

    for (row_profile, row_scenario, row_operation, subject), rows in groups.items():

        if (row_profile, row_scenario, row_operation) != (profile, scenario, operation):

            continue

        if row_subject_kind(rows, subject) == wanted_kind:

            subjects.add(subject)

    return subjects

def format_key(key: GroupKey) -> str:

    profile, scenario, operation, subject = key

    return f"{profile or '-'}/{scenario}/{operation}/{subject}"

def parse_formatted_key(value: str) -> GroupKey:

    profile, scenario, operation, subject = value.split("/", 3)

    return (None if profile == "-" else profile, scenario, operation, subject)

def content_integrity_summary(

    rows: list[dict[str, Any]],

    source_strength: dict[str, int],

) -> dict[GroupKey, dict[str, Any]]:

    grouped: dict[GroupKey, dict[str, Any]] = {}

    for row in rows:

        source = row.get("content_integrity_source")

        sources = row.get("content_integrity_sources")

        if not isinstance(source, str) and not isinstance(sources, list):

            continue

        key = group_key(row)

        entry = grouped.setdefault(

            key,

            {

                "sources": set(),

                "rows": 0,

                "passed": 0,

                "failed": 0,

                "weakest_source_strength": None,

            },

        )

        row_sources = [value for value in ([source] + sources if isinstance(sources, list) else [source]) if isinstance(value, str)]

        for value in row_sources:

            entry["sources"].add(value)

            strength = int(source_strength.get(value, 0))

            current = entry["weakest_source_strength"]

            entry["weakest_source_strength"] = strength if current is None else min(current, strength)

        entry["rows"] += 1

        if row.get("content_integrity_check_passed") is True:

            entry["passed"] += 1

        elif row.get("content_integrity_check_passed") is False:

            entry["failed"] += 1

    return grouped

def intervals_overlap(a: tuple[float, float], b: tuple[float, float]) -> bool:

    return a[0] <= b[1] and b[0] <= a[1]

def interval_usable(interval: tuple[float, float]) -> bool:

    return all(math.isfinite(value) for value in interval)

def verify(

    bundle: dict[str, Any],

    replay_rows: list[dict[str, Any]],

    original_rows: list[dict[str, Any]] | None,

    original_hash_failures: list[str],

    uncovered_pinned_inputs: tuple[str, ...] = (),

    *,

    confidence: float = 0.95,

) -> tuple[str, list[str]]:

    details: list[str] = []

    divergent = False

    unverifiable = False

    ratio_checked = False

    replay_groups = group_rows(replay_rows)

    content_integrity = bundle.get("content_integrity") if isinstance(bundle.get("content_integrity"), dict) else {}

    source_strength = dict(DEFAULT_CONTENT_SOURCE_STRENGTH)

    if isinstance(content_integrity.get("source_strength"), dict):

        for key, value in content_integrity["source_strength"].items():

            try:

                source_strength[str(key)] = int(value)

            except (TypeError, ValueError):

                continue

    bundled_content_groups = content_integrity.get("groups", {})

    replay_content_groups = content_integrity_summary(replay_rows, source_strength)

    if isinstance(bundled_content_groups, dict):

        for raw_key, expected in sorted(bundled_content_groups.items()):

            if not isinstance(raw_key, str) or not isinstance(expected, dict):

                continue

            try:

                key = parse_formatted_key(raw_key)

            except ValueError:

                unverifiable = True

                details.append(f"{raw_key}: unverifiable:content_integrity_group_key_invalid")

                continue

            replay = replay_content_groups.get(key)

            if replay is None:

                unverifiable = True

                details.append(f"{raw_key}: unverifiable:content_integrity_absent_in_replay")

                continue

            expected_failed = int(expected.get("failed") or 0)

            replay_failed = int(replay.get("failed") or 0)

            if expected_failed == 0 and replay_failed:

                divergent = True

                details.append(f"{raw_key}: divergent:content_integrity_failed_in_replay")

            expected_strength = expected.get("weakest_source_strength")

            replay_strength = replay.get("weakest_source_strength")

            if isinstance(expected_strength, int) and isinstance(replay_strength, int) and replay_strength < expected_strength:

                divergent = True

                details.append(

                    f"{raw_key}: divergent:content_integrity_source_downgrade "

                    f"original_strength={expected_strength} replay_strength={replay_strength}"

                )

    for failure in original_hash_failures:

        divergent = True

        details.append(f"original {failure}: divergent:original_hash_not_pinned_in_bundle")

    # Every raw input pinned by the bundle must be covered by a hash-matching

    # --original file; otherwise machine-independent verification is

    # incomplete and the verdict caps at COMPARABLE.

    for pinned_path in uncovered_pinned_inputs:

        unverifiable = True

        details.append(f"unverifiable:bundle_input_not_provided:{pinned_path}")

    if original_rows is None:

        unverifiable = True

        for key in sorted(replay_groups, key=format_key):

            details.append(f"{format_key(key)}: unverifiable:original_rows_absent")

        if not replay_groups:

            details.append("(no successful replay groups): unverifiable:original_rows_absent")

    else:

        original_groups = group_rows(original_rows)

        all_keys = sorted(set(original_groups) | set(replay_groups), key=format_key)

        for key in all_keys:

            originals = original_groups.get(key)

            replays = replay_groups.get(key)

            if originals is None:

                divergent = True

                details.append(f"{format_key(key)}: divergent:group_missing_in_original")

                continue

            if replays is None:

                divergent = True

                details.append(f"{format_key(key)}: divergent:group_missing_in_replay")

                continue

            for metric, accessor in MACHINE_INDEPENDENT_METRICS:

                original_values = metric_values(originals, accessor)

                replay_values = metric_values(replays, accessor)

                if not original_values and not replay_values:

                    continue

                if not original_values or not replay_values:

                    unverifiable = True

                    side = "original" if not original_values else "replay"

                    details.append(f"{format_key(key)}: unverifiable:{metric}_unmeasured_in_{side}")

                    continue

                if original_values != replay_values:

                    divergent = True

                    details.append(

                        f"{format_key(key)}: divergent:{metric}_mismatch "

                        f"original={original_values} replay={replay_values}"

                    )

                else:

                    details.append(f"{format_key(key)}: {metric} exact-match (n={len(original_values)})")

        # Machine-dependent: oak/git elapsed-ratio CI overlap per

        # (profile, scenario, operation).

        ratio_keys = sorted(

            {

                (profile, scenario, operation)

                for (profile, scenario, operation, _subject) in set(original_groups) & set(replay_groups)

            }

        , key=lambda key: (key[0] or "", key[1], key[2]))

        for profile, scenario, operation in ratio_keys:

            common_oak_subjects = (

                subjects_for_kind(original_groups, profile, scenario, operation, "oak")

                & subjects_for_kind(replay_groups, profile, scenario, operation, "oak")

            )

            common_git_subjects = (

                subjects_for_kind(original_groups, profile, scenario, operation, "git")

                & subjects_for_kind(replay_groups, profile, scenario, operation, "git")

            )

            if "git" in common_git_subjects:

                common_git_subjects = {"git"}

            checked_for_operation = False

            for oak_subject in sorted(common_oak_subjects):

                for git_subject in sorted(common_git_subjects):

                    sides: list[tuple[float, float]] = []

                    for groups in (original_groups, replay_groups):

                        numerator = elapsed_values(groups.get((profile, scenario, operation, oak_subject), []))

                        denominator = elapsed_values(groups.get((profile, scenario, operation, git_subject), []))

                        if not numerator or not denominator:

                            break

                        sides.append(ratio_ci(numerator, denominator, statistic="median", confidence=confidence))

                    label = f"{profile or '-'}/{scenario}/{operation}/{oak_subject}_vs_{git_subject}"

                    if len(sides) != 2 or not all(interval_usable(side) for side in sides):

                        details.append(f"{label}: elapsed_ms ratio unavailable (missing oak/git elapsed data)")

                        continue

                    checked_for_operation = True

                    original_ci, replay_ci = sides

                    if intervals_overlap(original_ci, replay_ci):

                        ratio_checked = True

                        details.append(

                            f"{label}: elapsed_ms oak/git ratio CIs overlap "

                            f"original=[{original_ci[0]:.4f}, {original_ci[1]:.4f}] "

                            f"replay=[{replay_ci[0]:.4f}, {replay_ci[1]:.4f}]"

                        )

                    else:

                        divergent = True

                        details.append(

                            f"{label}: divergent:elapsed_ratio_ci_no_overlap "

                            f"original=[{original_ci[0]:.4f}, {original_ci[1]:.4f}] "

                            f"replay=[{replay_ci[0]:.4f}, {replay_ci[1]:.4f}]"

                        )

            if not checked_for_operation:

                label = f"{profile or '-'}/{scenario}/{operation}"

                details.append(f"{label}: elapsed_ms ratio unavailable (missing oak/git elapsed data)")

    if divergent:

        return "DIVERGENT", details

    if unverifiable:

        return "COMPARABLE", details

    if ratio_checked:

        return "VERIFIED-EXACT", details

    return "VERIFIED", details

def parse_args(argv: list[str] | None = None) -> argparse.Namespace:

    parser = argparse.ArgumentParser(description="Verify a third-party replay against a repro bundle.")

    parser.add_argument("replay", nargs="+", type=Path, help="replay row JSONL paths (the third party's rerun)")

    parser.add_argument("--bundle", type=Path, required=True, help="repro bundle JSON produced by repro_bundle.py")

    parser.add_argument(

        "--original",

        action="append",

        default=None,

        type=Path,

        dest="originals",

        help="original raw row JSONL path pinned by the bundle (repeatable)",

    )

    parser.add_argument("--confidence", type=float, default=0.95, help="ratio bootstrap CI confidence")

    return parser.parse_args(argv)

def main(argv: list[str] | None = None) -> int:

    args = parse_args(argv)

    if not args.bundle.is_file():

        print(f"error: bundle not found: {args.bundle}", file=sys.stderr)

        return 2

    try:

        bundle = json.loads(args.bundle.read_text())

    except json.JSONDecodeError as exc:

        print(f"error: bundle is not valid JSON: {exc}", file=sys.stderr)

        return 2

    if bundle.get("schema_version") != 1:

        print(f"error: unsupported bundle schema_version: {bundle.get('schema_version')!r}", file=sys.stderr)

        return 2

    missing = [str(path) for path in args.replay if not path.is_file()]

    if args.originals:

        missing.extend(str(path) for path in args.originals if not path.is_file())

    if missing:

        print(f"error: row file(s) not found: {', '.join(missing)}", file=sys.stderr)

        return 2

    replay_rows = load_jsonl(args.replay)

    raw_inputs = bundle.get("raw_inputs", [])

    pinned_hashes = {entry.get("sha256") for entry in raw_inputs}

    provided_hashes: set[str] = set()

    original_rows: list[dict[str, Any]] | None = None

    original_hash_failures: list[str] = []

    if args.originals:

        for path in args.originals:

            digest = sha256_file(path)

            if digest in pinned_hashes:

                provided_hashes.add(digest)

            else:

                original_hash_failures.append(str(path))

        original_rows = load_jsonl(args.originals)

    uncovered_pinned_inputs = tuple(

        str(entry.get("path"))

        for entry in raw_inputs

        if entry.get("sha256") not in provided_hashes

    )

    verdict, details = verify(

        bundle,

        replay_rows,

        original_rows,

        original_hash_failures,

        uncovered_pinned_inputs,

        confidence=args.confidence,

    )

    for detail in details:

        print(detail)

    print(f"verdict: {verdict}")

    return VERDICT_EXIT_CODES[verdict]

if __name__ == "__main__":

    sys.exit(main())