project/gate.py

#!/usr/bin/env python3
"""Replay handwritten synthetic classification records; never call a model or deploy.

Python 3.11+; standard library only. PASS means the supplied fixture meets its
local policy, not that a real model or production release has been approved.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import math
from pathlib import Path
import sys

ROOT = Path(__file__).resolve().parent
ROUTES = {"answer", "clarify", "escalate"}
PROVENANCE = "handwritten-synthetic"


class InputError(ValueError):
    """Evidence is incomplete, malformed, or incompatible."""


def require(condition: bool, message: str) -> None:
    if not condition:
        raise InputError(message)


def unique_object(pairs: list[tuple[str, object]]) -> dict:
    result = {}
    for key, value in pairs:
        require(key not in result, f"Duplicate JSON object key: {key}")
        result[key] = value
    return result


def reject_constant(value: str) -> None:
    raise InputError(f"Non-finite JSON value: {value}")


def read_json(path: Path) -> tuple[dict, str]:
    raw = path.read_bytes()
    try:
        value = json.loads(raw, object_pairs_hook=unique_object,
                           parse_constant=reject_constant)
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        raise InputError(f"Invalid JSON in {path.name}: {exc}") from exc
    require(isinstance(value, dict), f"{path.name} must contain a JSON object")
    return value, hashlib.sha256(raw).hexdigest()


def text_field(value: object, label: str) -> None:
    require(isinstance(value, str) and bool(value.strip()), f"{label} must be nonempty text")


def integer(value: object, label: str) -> None:
    require(type(value) is int and value >= 0, f"{label} must be a nonnegative integer")


def fraction(value: object, label: str) -> None:
    require(type(value) in (int, float) and 0 <= value <= 1
            and math.isfinite(value), f"{label} must be a finite number from 0 to 1")


def fields(obj: object, expected: set[str], label: str) -> None:
    require(isinstance(obj, dict), f"{label} must be an object")
    require(set(obj) == expected, f"{label} fields must be exactly {sorted(expected)}")


def validate(cases: dict, baseline: dict, candidate: dict, policy: dict) -> dict[str, dict]:
    fields(cases, {"schema_version", "dataset_id", "provenance", "cases"}, "Dataset")
    require(type(cases["schema_version"]) is int and cases["schema_version"] == 1,
            "Dataset schema_version must be 1")
    text_field(cases["dataset_id"], "dataset_id")
    require(cases["provenance"] == PROVENANCE, "Only handwritten-synthetic evidence is supported")
    require(isinstance(cases["cases"], list) and bool(cases["cases"]), "Dataset must contain cases")
    index = {}
    for case in cases["cases"]:
        fields(case, {"id", "slice", "input", "expected_route", "critical", "rationale"}, "Case")
        for key in ("id", "slice", "input", "rationale"):
            text_field(case[key], f"Case {key}")
        require(case["id"] not in index, f"Duplicate case id: {case['id']}")
        require(isinstance(case["expected_route"], str) and case["expected_route"] in ROUTES,
                f"Unknown expected route: {case['id']}")
        require(type(case["critical"]) is bool, f"Case critical must be boolean: {case['id']}")
        index[case["id"]] = case
    for run in (baseline, candidate):
        fields(run, {"schema_version", "run_id", "dataset_id", "provenance", "system", "results"}, "Run")
        require(type(run["schema_version"]) is int and run["schema_version"] == 1,
                "Run schema_version must be 1")
        text_field(run["run_id"], "run_id")
        require(run["dataset_id"] == cases["dataset_id"], "Run dataset_id differs from dataset")
        require(run["provenance"] == PROVENANCE, "Run must declare handwritten-synthetic provenance")
        fields(run["system"], {"model", "prompt_revision", "code_revision"}, "System manifest")
        for key, value in run["system"].items():
            text_field(value, f"System {key}")
        require(run["system"]["model"] == "none-fixture", "This lab supports fixtures, not model runs")
        require(isinstance(run["results"], list), "Run results must be a list")
        seen = set()
        for row in run["results"]:
            fields(row, {"case_id", "route", "latency_ms", "cost_microusd"}, "Result")
            text_field(row["case_id"], "Result case_id")
            require(row["case_id"] not in seen, f"Duplicate result id: {row['case_id']}")
            require(isinstance(row["route"], str) and row["route"] in ROUTES,
                    f"Unknown route for {row['case_id']}")
            integer(row["latency_ms"], "latency_ms")
            integer(row["cost_microusd"], "cost_microusd")
            seen.add(row["case_id"])
        require(seen == set(index), "Run must contain every case exactly once, with no extra cases")
    require(candidate["run_id"] != baseline["run_id"], "Baseline and candidate run_id must differ")
    fields(policy, {"policy_id", "rationale", "min_overall", "min_by_slice", "max_critical_failures",
                    "max_new_failures", "max_p95_latency_ms", "max_mean_cost_microusd"}, "Policy")
    text_field(policy["policy_id"], "policy_id")
    text_field(policy["rationale"], "Policy rationale")
    fraction(policy["min_overall"], "min_overall")
    require(isinstance(policy["min_by_slice"], dict) and bool(policy["min_by_slice"]),
            "Policy needs slice thresholds")
    require(set(policy["min_by_slice"]) == {case["slice"] for case in index.values()},
            "Policy must name every dataset slice exactly once")
    for name, value in policy["min_by_slice"].items():
        fraction(value, f"Slice threshold {name}")
    for key in ("max_critical_failures", "max_new_failures", "max_p95_latency_ms", "max_mean_cost_microusd"):
        integer(policy[key], key)
    return index


def nearest_rank(values: list[int], percentile: float) -> int:
    """Nearest rank: sorted values at ceil(percentile*n), counting from one."""
    require(bool(values), "A percentile needs observations")
    require(0 < percentile <= 1, "Percentile must be in (0, 1]")
    return sorted(values)[math.ceil(percentile * len(values)) - 1]


def metrics(index: dict[str, dict], run: dict) -> tuple[dict, dict[str, bool]]:
    passed = {row["case_id"]: row["route"] == index[row["case_id"]]["expected_route"]
              for row in run["results"]}
    slices = {}
    for name in sorted({case["slice"] for case in index.values()}):
        ids = [key for key, case in index.items() if case["slice"] == name]
        count = sum(passed[key] for key in ids)
        slices[name] = {"passed": count, "total": len(ids), "accuracy": count / len(ids)}
    results = run["results"]
    total_cost = sum(row["cost_microusd"] for row in results)
    return {
        "passed": sum(passed.values()), "total": len(index),
        "accuracy": sum(passed.values()) / len(index), "by_slice": slices,
        "failed_case_ids": sorted(key for key in index if not passed[key]),
        "critical_failure_ids": sorted(key for key, case in index.items()
                                       if case["critical"] and not passed[key]),
        "p95_latency_ms": nearest_rank([row["latency_ms"] for row in results], .95),
        "total_cost_microusd": total_cost, "mean_cost_microusd": total_cost / len(index)
    }, passed


def evaluate(cases: dict, baseline: dict, candidate: dict, policy: dict) -> dict:
    index = validate(cases, baseline, candidate, policy)
    base_metrics, base_passed = metrics(index, baseline)
    cand_metrics, cand_passed = metrics(index, candidate)
    new_failures = sorted(key for key in index if base_passed[key] and not cand_passed[key])
    fixed = sorted(key for key in index if not base_passed[key] and cand_passed[key])
    checks = []

    def check(name: str, observed: float, limit: float, direction: str) -> None:
        passed = observed >= limit if direction == "minimum" else observed <= limit
        checks.append({"name": name, "observed": observed, direction: limit, "passed": passed})

    check("overall_accuracy", cand_metrics["accuracy"], policy["min_overall"], "minimum")
    for name, limit in sorted(policy["min_by_slice"].items()):
        check(f"slice_accuracy:{name}", cand_metrics["by_slice"][name]["accuracy"], limit, "minimum")
    check("critical_route_failures", len(cand_metrics["critical_failure_ids"]),
          policy["max_critical_failures"], "maximum")
    check("new_case_failures", len(new_failures), policy["max_new_failures"], "maximum")
    check("p95_latency_ms", cand_metrics["p95_latency_ms"], policy["max_p95_latency_ms"], "maximum")
    check("mean_cost_microusd", cand_metrics["mean_cost_microusd"],
          policy["max_mean_cost_microusd"], "maximum")
    return {
        "report_schema_version": 1, "evidence_kind": "synthetic_fixture_replay",
        "release_authorized": False, "dataset_id": cases["dataset_id"],
        "policy_id": policy["policy_id"],
        "decision": "PASS" if all(item["passed"] for item in checks) else "BLOCK",
        "baseline": {"run_id": baseline["run_id"], "system": baseline["system"], "metrics": base_metrics},
        "candidate": {"run_id": candidate["run_id"], "system": candidate["system"], "metrics": cand_metrics},
        "new_failure_ids": new_failures, "fixed_case_ids": fixed, "checks": checks,
        "blocking_reasons": [item["name"] for item in checks if not item["passed"]],
        "limitations": [
            "Handwritten routes, latencies, and costs; no model was called or timed.",
            "These synthetic cases do not estimate production performance or safety.",
            "Only route equality is graded; answer content is not evaluated.",
            "PASS is a local policy result, not permission to deploy or a learner pilot."
        ]
    }


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--cases", type=Path, default=ROOT / "fixtures/cases.json")
    parser.add_argument("--baseline", type=Path, default=ROOT / "fixtures/baseline.json")
    parser.add_argument("--candidate", type=Path, required=True)
    parser.add_argument("--policy", type=Path, default=ROOT / "policy.json")
    parser.add_argument("--report", type=Path, help="Optional JSON output; input files cannot be overwritten")
    args = parser.parse_args(argv)
    try:
        paths = {"cases": args.cases, "baseline": args.baseline,
                 "candidate": args.candidate, "policy": args.policy}
        if args.report:
            require(args.report.resolve() not in {path.resolve() for path in paths.values()},
                    "Report path cannot overwrite an input file")
            require(not args.report.exists() or not any(
                path.exists() and args.report.samefile(path) for path in paths.values()),
                "Report path cannot overwrite an input through a hard link")
        loaded = {name: read_json(path) for name, path in paths.items()}
        report = evaluate(*(loaded[name][0] for name in ("cases", "baseline", "candidate", "policy")))
        report["input_sha256"] = {name: loaded[name][1] for name in paths}
        encoded = json.dumps(report, indent=2, sort_keys=True, allow_nan=False) + "\n"
        if args.report:
            args.report.parent.mkdir(parents=True, exist_ok=True)
            args.report.write_text(encoded, encoding="utf-8")
        print(f"{report['decision']}: {report['candidate']['run_id']} (synthetic fixture replay)")
        print(f"Accuracy: {report['candidate']['metrics']['passed']}/{report['candidate']['metrics']['total']}")
        print("Blocking reasons: " + (", ".join(report["blocking_reasons"]) or "none"))
        print("Release authorized: false")
        return 0 if report["decision"] == "PASS" else 1
    except (InputError, OSError, TypeError) as exc:
        print(f"INPUT ERROR: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())