#!/usr/bin/env python3

"""Pre-publish gate: mechanical checks before any public table ships.

Run against the result files you intend to publish:

  python3 scripts/publish_gate.py results/latest.jsonl results/latest.mount.jsonl ...

Checks (fail = exit 1; manual items print as MANUAL and never fail the gate):

1. calibration     docs/token-calibration-factors.md exists and is committed —

                   published token deltas must cite calibrated factors, not

                   raw char/4 (README warns char/4 is biased for VCS output).

2. tuned-git       core-lane results include git_untracked_cache AND

                   git_fsmonitor subject rows. Stock git alone overstates oak

                   wins on wide trees.

3. transports      no row set mixes remote transports under one operation

                   name (remote.net.* vs remote.*); rows touching a remote

                   carry remote_transport.

4. levels          agent rows carry instruction_level and the file holds only

                   one level (never aggregate across levels).

5. nulls           null means unmeasured: summaries must not render null as 0.

                   (Spot-checked by scanning for '| 0 |' columns whose JSONL

                   value is null — heuristic, reported as MANUAL.)

6. skips           every returncode-77 row carries skip_reason (work items,

                   never silent absence); count is printed for the summary.

7. tails           rows with tail percentiles carry their sample counts

                   (tail_latency_summary embeds n; absence is a failure).

8. provenance      published Oak subject rows carry clean Oak source

                   provenance (hash present, status clean).

9. samples         public core rows carry enough repetitions per

                   subject/scenario/operation group (n>=30).

10. randomization  public core rows state whether subject order was randomized;

                   randomized rows carry the per-run subject order.

11. failures       public rows do not hide failed operations behind successful

                   averages.

12. tracks         each result file holds one benchmark_track.

13. env-isolation  public core rows carry one env isolation version so rows

                   measured before/after harness environment policy changes

                   cannot be compared silently.

14. conformance    inputs holding Oak rows also hold diagnostic conformance rows

                   (scripts/vcs_conformance.py --output) for every published Oak

                   binary; the Oak rows pass the conformance gate against

                   config/conformance_expected_failures.json (no regression, no

                   missing behaviour, no FIXED-but-still-listed entry, no

                   entry whose ticket is still pending:<slug>).

"""

from __future__ import annotations

import json

import sys

from pathlib import Path

from oakbench.integrity import content_integrity_public_trust_passed

from oakbench.public_output import output_evidence_failures, workflow_evidence_failures, workflow_qualification_failures

from oakbench.rows import row_returncode

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

CALIBRATION_DOC = ROOT / "docs" / "token-calibration-factors.md"

# Tuned subjects whose mode.setup.* rows must all succeed before any of their

# rows are publishable (a failed setup means stock git measured under a tuned

# label).

TUNED_GIT_SUBJECTS = {"git_untracked_cache", "git_fsmonitor", "git_lfs"}

# Tuned subjects every core publish must include. git_lfs is deliberately not

# required here: it only applies to the binary-fixture scenarios listed in

# config/git_modes.json and needs git-lfs installed, so its absence is an

# honest coverage gap (skip rows), while a present-but-failed lfs setup still

# fails the gate via TUNED_GIT_SUBJECTS above.

REQUIRED_TUNED_GIT_SUBJECTS = {"git_untracked_cache", "git_fsmonitor"}

CORE_PROFILES = {"smoke", "standard", "large", "micro"}

PUBLIC_CORE_PROFILES = {"standard", "large", "micro"}

PUBLIC_CORE_MIN_SAMPLES = 30

EXCLUDED_SAMPLE_OP_SUFFIXES = (".add", ".commit", ".determinism", ".inforecall")

EXCLUDED_SAMPLE_OP_PREFIXES = ("mode.setup",)

RANDOMIZATION_FIELDS = ("randomized_subject_order", "randomize_subject_order", "subject_order_randomized")

SUBJECT_ORDER_FIELDS = ("subject_order", "run_subject_order")

def load_rows(path: Path) -> list[dict]:

    rows = []

    for line in path.read_text().splitlines():

        line = line.strip()

        if line:

            rows.append(json.loads(line))

    return rows

def is_oak_subject(row: dict) -> bool:

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

    if subject.startswith("oak"):

        return True

    if row.get("subject_kind") == "oak":

        return True

    details = row.get("subject_details")

    if isinstance(details, dict):

        subject_detail = details.get(subject)

        if isinstance(subject_detail, dict):

            return subject_detail.get("kind") == "oak" or str(

                subject_detail.get("role", "")

            ).startswith("oak_")

    return False

def source_status_is_clean(status: object) -> bool:

    if not isinstance(status, str):

        return False

    normalized = " ".join(status.strip().lower().split())

    return normalized in {

        "",

        "clean",

        "no changes",

        "nothing to commit",

        "nothing to commit, working tree clean",

    }

def randomization_value(row: dict) -> object:

    for field in RANDOMIZATION_FIELDS:

        if field in row:

            return row[field]

    return None

def subject_order(row: dict) -> object:

    for field in SUBJECT_ORDER_FIELDS:

        if field in row:

            return row[field]

    return None

def is_skip(row: dict) -> bool:

    return row_returncode(row) == 77 or bool(row.get("skipped"))

def is_excluded_sample_operation(operation: str) -> bool:

    return operation.endswith(EXCLUDED_SAMPLE_OP_SUFFIXES) or operation.startswith(

        EXCLUDED_SAMPLE_OP_PREFIXES

    )

def sample_run_id(row: dict, fallback: int) -> str:

    for field in ("run", "run_index"):

        value = row.get(field)

        if value is not None:

            return str(value)

    return f"row:{fallback}"

def oak_binary_identity(row: dict) -> str | None:

    """SHA-256 of the Oak binary that produced a row, from any lane's identity field."""

    for key in ("binary_sha256", "subject_binary_sha256", "workflow_subject_binary_sha256"):

        value = row.get(key)

        if isinstance(value, str) and len(value) == 64:

            return value

    details = row.get("subject_details")

    if isinstance(details, dict):

        item = details.get(str(row.get("subject")))

        if isinstance(item, dict) and isinstance(item.get("binary_sha256"), str) and len(item["binary_sha256"]) == 64:

            return item["binary_sha256"]

    return None

def conformance_gate_lines(all_rows: list[tuple[Path, dict]]) -> tuple[list[str], list[str]]:

    """Check 14: every published Oak binary passes the conformance gate on ITS OWN rows."""

    from oakbench import conformance_gate

    from oakbench.conformance import SCENARIOS

    from oakbench.conformance_matrix import BEHAVIOURS

    conformance_rows = [row for _, row in all_rows if row.get("evidence_class") == "diagnostic_conformance"]

    oak_rows = [(path, row) for path, row in all_rows

                if is_oak_subject(row) and row.get("evidence_class") != "diagnostic_conformance"]

    if not oak_rows:

        return [], ["PASS   conformance: no Oak rows among inputs"]

    anonymous = [(str(path), row.get("subject"), row.get("operation")) for path, row in oak_rows

                 if oak_binary_identity(row) is None]

    failures = []

    if anonymous:

        failures.append("FAIL   conformance: Oak rows lack binary identity (binary_sha256 or subject_details), so "

                        f"their conformance evidence cannot be matched: {anonymous[:5]}")

    published = sorted({oak_binary_identity(row) for _, row in oak_rows} - {None})

    if not conformance_rows:

        failures.append("FAIL   conformance: Oak rows published without diagnostic conformance rows; run "

                        "scripts/vcs_conformance.py --output <file> with the published Oak binary and pass it here")

        return failures, []

    try:

        expected = conformance_gate.load_expected()

    except (OSError, ValueError) as exc:

        return failures + [f"FAIL   conformance: expected-failures list unreadable: {exc}"], []

    pending = conformance_gate.pending_tickets(expected)

    if pending:

        # A pending:<slug> ticket is a placeholder while the bug is being filed;

        # publication needs every expected failure to cite a filed fb-N

        # (especially intermittent ones, whose passes are never FIXED).

        failures.append("FAIL   conformance: expected-failures list has unfiled pending: tickets; file them and "

                        f"cite fb-N before publishing: {[f'{b} ({t}, {o})' for b, t, o in pending][:5]}")

    required = list(SCENARIOS) + [behaviour.id for behaviour in BEHAVIOURS]

    notes = []

    for sha in published:

        # Only this binary's subject rows (plus reference rows) are evaluated:

        # rows of another Oak build can neither satisfy nor fail this one.

        rows = [row for row in conformance_rows

                if row.get("subject") != expected["subject"] or row.get("binary_sha256") == sha]

        result = conformance_gate.evaluate(rows, expected, required)

        if not result.get("profile_recorded"):

            failures.append(f"FAIL   conformance[{sha[:12]}]: filesystem profile {result['profiles'] or 'missing'} "

                            "has no recorded coverage in config/conformance_expected_failures.json; publication "

                            "requires a reviewed coverage record for the host profile")

        if result["verdict"] != "PASS":

            failures.append(f"FAIL   conformance[{sha[:12]}]: {result['verdict']}: "

                            f"{(result['regressions'] + result['unmeasured'])[:5]}")

        if result["fixed"]:

            failures.append(f"FAIL   conformance[{sha[:12]}]: expected-failures list is stale (now passing): "

                            f"{result['fixed'][:5]}")

        if result["verdict"] == "PASS" and not result["fixed"] and result.get("profile_recorded"):

            notes.append(f"PASS   [{sha[:12]}] " + conformance_gate.summary_lines(result)[0])

    return failures, ([] if failures else notes)

def main(paths: list[str]) -> int:

    failures: list[str] = []

    notes: list[str] = []

    if CALIBRATION_DOC.exists():

        notes.append(f"PASS   calibration: {CALIBRATION_DOC.name} present")

    else:

        failures.append(

            "FAIL   calibration: docs/token-calibration-factors.md missing — run "

            "token_calibration.py per output style before publishing token deltas"

        )

    all_rows: list[tuple[Path, dict]] = []

    for raw in paths:

        path = Path(raw)

        if not path.exists():

            failures.append(f"FAIL   missing result file: {path}")

            continue

        for row in load_rows(path):

            all_rows.append((path, row))

    core_rows = [row for _, row in all_rows if row.get("profile") in CORE_PROFILES]

    if core_rows:

        subjects = {

            str(row.get("subject"))

            for row in core_rows

            if row_returncode(row) == 0

            and not is_excluded_sample_operation(str(row.get("operation") or ""))

        }

        missing = REQUIRED_TUNED_GIT_SUBJECTS - subjects

        if missing:

            failures.append(

                f"FAIL   tuned-git: core rows lack {sorted(missing)} — rerun with "

                "--git-modes untracked_cache,fsmonitor (verify fsmonitor--daemon health on darwin)"

            )

        else:

            notes.append("PASS   tuned-git: untracked_cache and fsmonitor rows present")

        bad_tuned_setup = [

            (row.get("subject"), row.get("operation"), row.get("returncode"))

            for row in core_rows

            if str(row.get("subject")) in TUNED_GIT_SUBJECTS

            and str(row.get("operation") or "").startswith(EXCLUDED_SAMPLE_OP_PREFIXES)

            and row_returncode(row) != 0

        ]

        if bad_tuned_setup:

            failures.append(

                "FAIL   tuned-git: tuned Git setup failed, so tuned rows are not publishable: "

                f"{bad_tuned_setup[:5]}"

            )

    else:

        notes.append("MANUAL tuned-git: no core-lane rows among inputs; check the core table separately")

    # Scale lane (suites/scale): claimable only when the calibration campaign's

    # own JSONL is among the inputs (matched by content sha256) and every

    # number is RECOMPUTED from it with the P1 cell machinery; the record's

    # stated values are never trusted (oakbench.scale_calibration).

    scale_rows = [(str(path), row) for path, row in all_rows if row.get("profile") == "scale"]

    if scale_rows:

        from oakbench import scale_calibration

        inputs: dict = {}

        input_sha: dict = {}

        for path, row in all_rows:

            inputs.setdefault(str(path), []).append(row)

        for path in inputs:

            input_sha[path] = scale_calibration.file_sha256(Path(path))

        try:

            registry = json.loads((ROOT / "config" / "fixtures.json").read_text()).get("fixtures", {})

        except (OSError, ValueError):

            registry = {}

        refused = scale_calibration.gate(scale_rows, inputs, input_sha, registry)

        if refused:

            failures.append(

                f"FAIL   scale-calibration: {len(refused)} scale row(s) are not verified quiet-host calibrated "

                f"evidence: {refused[:5]}"

            )

        else:

            notes.append("PASS   scale-calibration: every scale row verified against its recomputed calibration campaign")

    oak_provenance_failures = []

    for path, row in all_rows:

        if not is_oak_subject(row) or row.get("evidence_class") == "diagnostic_conformance":

            continue  # conformance rows carry binary identity, checked by the conformance gate

        source = row.get("source")

        source = source if isinstance(source, dict) else {}

        oak_hash = source.get("oak_hash")

        oak_status = source.get("oak_status")

        if not oak_hash or not source_status_is_clean(oak_status):

            oak_provenance_failures.append(

                (str(path), row.get("subject"), row.get("operation"), oak_hash, oak_status)

            )

    if oak_provenance_failures:

        failures.append(

            "FAIL   provenance: published Oak rows lack clean Oak source provenance: "

            f"{oak_provenance_failures[:5]}"

        )

    elif any(is_oak_subject(row) for _, row in all_rows):

        notes.append("PASS   provenance: Oak subject rows carry clean source hash/status")

    fake_provider_rows = [

        (str(path), row.get("platform"), row.get("driver"), row.get("scenario"), row.get("operation"))

        for path, row in all_rows

        if row.get("driver") == "fake-provider" or row.get("branch_triage_provider") == "fake"

    ]

    if fake_provider_rows:

        failures.append(

            "FAIL   fake-provider: simulated provider rows are not publishable: "

            f"{fake_provider_rows[:5]}"

        )

    public_core_rows = [

        (path, row)

        for path, row in all_rows

        if row.get("profile") in PUBLIC_CORE_PROFILES and not is_skip(row)

    ]

    public_failures = [

        (str(path), row.get("subject"), row.get("scenario"), row.get("operation"), row.get("returncode"))

        for path, row in all_rows

        if row.get("profile") in PUBLIC_CORE_PROFILES

        and not is_skip(row)

        and row_returncode(row) != 0

    ]

    if public_failures:

        failures.append(

            "FAIL   failures: public core rows contain failed operations: "

            f"{public_failures[:5]}"

        )

    workflow_evidence = workflow_evidence_failures([row for _, row in all_rows])

    workflow_evidence.extend(workflow_qualification_failures([row for _, row in all_rows], PUBLIC_CORE_MIN_SAMPLES))

    if workflow_evidence:

        failures.append("FAIL   workflow-evidence: " + "; ".join(workflow_evidence[:5]))

    unsupported_output = [

        (str(path), row.get("subject"), row.get("scenario"), row.get("operation"), reasons)

        for path, row in public_core_rows

        for reasons in [output_evidence_failures(row)]

        if reasons

    ]

    if unsupported_output:

        failures.append(

            "FAIL   output-evidence: public output usefulness claims lack exact evidence: "

            f"{unsupported_output[:5]}"

        )

    weak_content_trust = [

        (

            str(path),

            row.get("subject"),

            row.get("scenario"),

            row.get("operation"),

            row.get("content_integrity_source"),

            row.get("content_integrity_sources"),

        )

        for path, row in public_core_rows

        if row.get("content_integrity_check_passed") is True

        and not content_integrity_public_trust_passed(row)

    ]

    if weak_content_trust:

        failures.append(

            "FAIL   content-trust: public core rows claim passing content integrity "

            "from weak or mixed sources: "

            f"{weak_content_trust[:5]}"

        )

    elif any(row.get("content_integrity_check_passed") is True for _, row in public_core_rows):

        notes.append("PASS   content-trust: passing public content checks use strong sources")

    env_versions = {

        str(row.get("env_isolation_version"))

        for _, row in public_core_rows

        if isinstance(row.get("env_isolation_version"), str)

        and str(row.get("env_isolation_version")).strip()

    }

    missing_env_version = [

        (

            str(path),

            row.get("subject"),

            row.get("scenario"),

            row.get("operation"),

            row.get("run", row.get("run_index")),

        )

        for path, row in public_core_rows

        if not isinstance(row.get("env_isolation_version"), str)

        or not str(row.get("env_isolation_version")).strip()

    ]

    if missing_env_version:

        failures.append(

            "FAIL   env-isolation: public core rows missing env_isolation_version: "

            f"{missing_env_version[:5]}"

        )

    elif len(env_versions) > 1:

        failures.append(

            "FAIL   env-isolation: public core rows mix env_isolation_version values: "

            f"{sorted(env_versions)}"

        )

    elif env_versions:

        notes.append(f"PASS   env-isolation: public core rows use version {next(iter(env_versions))}")

    measured_groups: dict[tuple[str, str, str, str, str, str], set[str]] = {}

    sample_groups: dict[tuple[str, str, str, str, str, str], set[str]] = {}

    for index, (path, row) in enumerate(public_core_rows):

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

        if is_excluded_sample_operation(operation):

            continue

        key = (

            str(path),

            str(row.get("profile")),

            str(row.get("benchmark_track", "")),

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

            operation,

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

        )

        run_id = sample_run_id(row, index)

        measured_groups.setdefault(key, set()).add(run_id)

        if row_returncode(row) == 0:

            sample_groups.setdefault(key, set()).add(run_id)

    thin_groups = {

        "/".join(key[1:]): len(runs)

        for key, runs in measured_groups.items()

        if len(runs) < PUBLIC_CORE_MIN_SAMPLES

    }

    zero_success_groups = {

        "/".join(key[1:]): len(runs)

        for key, runs in measured_groups.items()

        if not sample_groups.get(key)

    }

    if thin_groups:

        failures.append(

            f"FAIL   samples: public core groups below n={PUBLIC_CORE_MIN_SAMPLES}: "

            f"{dict(list(thin_groups.items())[:5])}"

        )

    if zero_success_groups:

        failures.append(

            "FAIL   samples: public core groups with zero successful rows: "

            f"{dict(list(zero_success_groups.items())[:5])}"

        )

    if public_core_rows and not thin_groups and not zero_success_groups:

        notes.append(f"PASS   samples: public core groups have n>={PUBLIC_CORE_MIN_SAMPLES}")

    missing_randomization = [

        (

            str(path),

            row.get("subject"),

            row.get("scenario"),

            row.get("operation"),

            row.get("run", row.get("run_index")),

        )

        for path, row in public_core_rows

        if randomization_value(row) is None

    ]

    if missing_randomization:

        failures.append(

            "FAIL   randomization: public core rows missing randomization metadata: "

            f"{missing_randomization[:5]}"

        )

    else:

        not_randomized = [

            (

                str(path),

                row.get("subject"),

                row.get("scenario"),

                row.get("operation"),

                row.get("run", row.get("run_index")),

            )

            for path, row in public_core_rows

            if randomization_value(row) is not True

        ]

        randomized_without_order = []

        if not_randomized:

            failures.append(

                "FAIL   randomization: public core rows were not randomized: "

                f"{not_randomized[:5]}"

            )

        else:

            for path, row in public_core_rows:

                order = subject_order(row)

                if not isinstance(order, list) or not order or row.get("subject") not in order:

                    randomized_without_order.append(

                        (

                            str(path),

                            row.get("subject"),

                            row.get("scenario"),

                            row.get("operation"),

                            row.get("run", row.get("run_index")),

                        )

                    )

        if randomized_without_order:

            failures.append(

                "FAIL   randomization: randomized public core rows lack per-run subject order: "

                f"{randomized_without_order[:5]}"

            )

        elif public_core_rows:

            notes.append("PASS   randomization: public core rows state run-order metadata")

    tracks_by_file: dict[Path, set[str]] = {}

    for path, row in all_rows:

        track = row.get("benchmark_track")

        if track:

            tracks_by_file.setdefault(path, set()).add(str(track))

    mixed_tracks = {str(path): sorted(tracks) for path, tracks in tracks_by_file.items() if len(tracks) > 1}

    if mixed_tracks:

        failures.append(f"FAIL   tracks: files mix benchmark_track values: {mixed_tracks}")

    elif tracks_by_file:

        notes.append("PASS   tracks: each result file holds a single benchmark_track")

    # Transport hygiene: remote-touching rows must state their transport, and

    # one subject's operation must not mix transports. Side-by-side tables may

    # contain the same operation name for network and local_file subjects.

    missing_transport = [

        (str(path), row.get("subject"), row.get("operation"))

        for path, row in all_rows

        if (row.get("remote_repo") or row.get("remote_server"))

        and not (row.get("remote_transport") or row.get("workspace_transport"))

    ]

    if missing_transport:

        failures.append(f"FAIL   transports: remote rows without transport: {missing_transport[:5]}")

    op_transports: dict[tuple[str, str], set[str]] = {}

    for _, row in all_rows:

        transport = row.get("remote_transport") or row.get("workspace_transport")

        if transport:

            key = (str(row.get("subject")), str(row.get("operation")))

            op_transports.setdefault(key, set()).add(str(transport))

    mixed = {f"{subject} {op}": sorted(ts) for (subject, op), ts in op_transports.items() if len(ts) > 1}

    if mixed:

        failures.append(f"FAIL   transports: operations span multiple transports: {mixed}")

    elif op_transports:

        notes.append(f"PASS   transports: {len(op_transports)} remote-touching operations, one transport each")

    levels_by_file: dict[Path, set[str]] = {}

    for path, row in all_rows:

        level = row.get("instruction_level")

        if level is None and isinstance(row.get("agent"), dict):

            level = row["agent"].get("instruction_level")

        if level:

            levels_by_file.setdefault(path, set()).add(str(level))

    multi = {str(p): sorted(ls) for p, ls in levels_by_file.items() if len(ls) > 1}

    if multi:

        failures.append(f"FAIL   levels: files mix instruction levels (never aggregate): {multi}")

    elif levels_by_file:

        notes.append("PASS   levels: each agent result file holds a single instruction level")

    bad_skips = [

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

        for path, row in all_rows

        if row_returncode(row) == 77 and not row.get("skip_reason")

    ]

    skip_count = sum(1 for _, row in all_rows if row_returncode(row) == 77)

    if bad_skips:

        failures.append(f"FAIL   skips: returncode-77 rows without skip_reason: {bad_skips[:5]}")

    else:

        notes.append(f"PASS   skips: {skip_count} skip rows, all carrying skip_reason (account for each in prose)")

    for path, row in all_rows:

        tail = (row.get("parallel_metrics") or {}).get("commit_latency_tail")

        if isinstance(tail, dict) and "samples" not in tail and "n" not in tail and "sample_count" not in tail:

            failures.append(f"FAIL   tails: {path.name} {row.get('operation')} tail summary lacks sample count")

            break

    else:

        notes.append("PASS   tails: tail summaries carry sample counts (or none present)")

    notes.append("MANUAL nulls: confirm summaries render null as 'unmeasured', never 0")

    notes.append("MANUAL prose: state transport asymmetries (oak network vs git local) next to every side-by-side")

    conformance_failures, conformance_notes = conformance_gate_lines(all_rows)

    failures += conformance_failures

    notes += conformance_notes

    for line in notes:

        print(line)

    for line in failures:

        print(line)

    print(f"\n{'GATE FAIL' if failures else 'GATE PASS'} ({len(failures)} failure(s), inputs: {len(paths)})")

    return 1 if failures else 0

if __name__ == "__main__":

    raise SystemExit(main(sys.argv[1:]))