#!/usr/bin/env python3 """Pure offline simulations over original handwritten synthetic evidence. No model, network, deployment, telemetry collection, or schedule is created. Commands print JSON to stdout and do not persist changes. """ from __future__ import annotations import argparse from collections import Counter from datetime import date, datetime import hashlib import json from pathlib import Path import platform from statistics import mean import sys from gate import InputError, ROOT, evaluate, integer, nearest_rank, read_json, require CRITERIA = {"grounded", "within_scope", "private_data_protected"} MANIFEST_FILES = ("fixtures/operations.json", "fixtures/cases.json", "fixtures/baseline.json", "fixtures/candidate-regressed.json", "fixtures/candidate-repaired.json", "policy.json", "gate.py", "opslab.py") def load() -> dict: evidence = read_json(ROOT / "fixtures/operations.json")[0] require(evidence.get("provenance") == "handwritten-synthetic", "Synthetic provenance required") return evidence def dataset_audit(register: dict) -> dict: """Find split leakage and unresolved labels; never silently repair records.""" require(register.get("version") and register.get("change_reason"), "Version and change reason required") records = register["records"] require(bool(records), "Dataset cannot be empty") ids, groups, issues, labels = set(), {}, [], {} counts = Counter() for row in records: require(row["id"] not in ids, "Duplicate dataset ID") ids.add(row["id"]) require(row["split"] in {"development", "holdout"}, "Unknown split") require(bool(row["group"]) and bool(row["rationale"]), "Group and rationale required") groups.setdefault(row["group"], set()).add(row["split"]) counts[row["split"]] += 1 votes = row["votes"] require(len(votes) == 2 and len({v["reviewer"] for v in votes}) == 2, "Two distinct synthetic reviewers required") allowed = {"answer", "clarify", "escalate"} require(all(v["label"] in allowed for v in votes), "Unknown label") adjudicated = row.get("adjudicated") if not adjudicated: issues.append({"id": row["id"], "reason": "unresolved_label"}) else: require(adjudicated in allowed, "Unknown adjudicated label") if len({v["label"] for v in votes}) == 1 and adjudicated != votes[0]["label"]: issues.append({"id": row["id"], "reason": "consensus_changed_without_new_review"}) labels[row["id"]] = adjudicated leakage = sorted(group for group, splits in groups.items() if len(splits) > 1) require(set(counts) == {"development", "holdout"}, "Both splits required") return {"version": register["version"], "counts": dict(sorted(counts.items())), "cross_split_groups": leakage, "label_issues": issues, "resolved_labels": labels, "decision": "BLOCK" if leakage or issues else "PASS"} def review_summary(items: list[dict]) -> dict: """Summarize supplied human rubric inputs; this does not grade language.""" require(bool(items), "Review set cannot be empty") ids, disagreements, failures = set(), [], [] comparisons = agreements = 0 for row in items: require(row["id"] not in ids, "Duplicate output ID") ids.add(row["id"]) require(bool(row["output"]) and bool(row["rationale"]), "Output and review rationale required") ratings = row["ratings"] require(len(ratings) == 2 and len({r["reviewer"] for r in ratings}) == 2, "Two distinct rubric reviewers required") require(all(set(r["criteria"]) == CRITERIA for r in ratings), "Complete rubric required") require(all(type(value) is bool for r in ratings for value in r["criteria"].values()), "Rubric inputs must be boolean") for criterion in sorted(CRITERIA): values = [r["criteria"][criterion] for r in ratings] comparisons += 1 agreements += values[0] == values[1] if values[0] != values[1]: disagreements.append({"id": row["id"], "criterion": criterion}) if not all(value for r in ratings for value in r["criteria"].values()): failures.append(row["id"]) return {"agreements": agreements, "comparisons": comparisons, "raw_agreement": agreements / comparisons, "disagreements": disagreements, "failed_output_ids": sorted(failures), "decision": "REVIEW" if failures or disagreements else "PASS", "mechanism": "aggregation of supplied synthetic human judgments; no semantic classifier"} def manifest(root: Path = ROOT) -> dict: return {"evidence_kind": "synthetic_fixture_replay", "python": platform.python_version(), "grader_revision": "human-rubric-v1", "timing_boundary": "invented values; no execution timed", "files": {name: hashlib.sha256((root / name).read_bytes()).hexdigest() for name in MANIFEST_FILES}} def grader_calibration(rows: list[dict]) -> dict: """Positive means acceptable; reference judgments are synthetic adjudications.""" require(bool(rows), "Calibration cases required") counts = {"TP": 0, "FP": 0, "FN": 0, "TN": 0} ids = set() for row in rows: require(row["id"] not in ids, "Duplicate calibration ID") ids.add(row["id"]) require(type(row["reference_ok"]) is bool and type(row["judge_ok"]) is bool, "Calibration inputs must be boolean") key = ("T" if row["reference_ok"] == row["judge_ok"] else "F") + ("P" if row["judge_ok"] else "N") counts[key] += 1 def ratio(numerator: int, denominator: int) -> float | None: return numerator / denominator if denominator else None return {"counts": counts, "positive_class": "acceptable", "accept_precision": ratio(counts["TP"], counts["TP"] + counts["FP"]), "accept_recall": ratio(counts["TP"], counts["TP"] + counts["FN"]), "false_acceptance_rate": ratio(counts["FP"], counts["FP"] + counts["TN"]), "missing_denominator": "null means undefined, not zero error"} def verify_manifest(record: dict, root: Path = ROOT) -> list[str]: require(bool(record.get("files")), "Manifest has no files") require(set(record["files"]) == set(MANIFEST_FILES), "Manifest must name the complete required file set") changed = [] for name, digest in record["files"].items(): path = (root / name).resolve() require(path.is_relative_to(root.resolve()), "Manifest path escapes project") if not path.is_file() or hashlib.sha256(path.read_bytes()).hexdigest() != digest: changed.append(name) return sorted(changed) def paired_comparison(rows: list[dict], plan: dict) -> dict: require(bool(rows), "No paired observations") require(bool(plan["case_ids"]) and len(set(plan["case_ids"])) == len(plan["case_ids"]), "Plan needs unique expected case IDs") require(all(isinstance(name, str) and name for name in plan["case_ids"]), "Invalid planned case ID") require(bool(plan["trial_ids"]) and len(set(plan["trial_ids"])) == len(plan["trial_ids"]), "Plan needs unique expected trial IDs") for trial in plan["trial_ids"]: integer(trial, "planned trial") expected_pairs = {(case, trial) for case in plan["case_ids"] for trial in plan["trial_ids"]} keys, by_case, by_trial = set(), {}, {} for row in rows: key = (row["case_id"], row["trial"]) require(key not in keys, "Duplicate case/trial pair") keys.add(key) integer(row["trial"], "trial") for side in ("baseline", "candidate"): require(type(row[side]["correct"]) is bool, "Correct must be boolean") for field in ("latency_ms", "cost_microusd"): integer(row[side][field], field) by_case.setdefault(row["case_id"], set()).add(row["trial"]) by_trial.setdefault(row["trial"], []).append(row) require(len({tuple(sorted(trials)) for trials in by_case.values()}) == 1, "Each case needs the same trial IDs") require(keys == expected_pairs, "Observations must match every planned case/trial pair exactly") base = [r["baseline"] for r in rows] candidate = [r["candidate"] for r in rows] quality = mean(r["correct"] for r in candidate) p95 = nearest_rank([r["latency_ms"] for r in candidate], .95) cost = mean(r["cost_microusd"] for r in candidate) return {"observations": len(rows), "distinct_cases": len(by_case), "candidate_accuracy": quality, "baseline_accuracy": mean(r["correct"] for r in base), "paired_accuracy_delta": mean(r["candidate"]["correct"] - r["baseline"]["correct"] for r in rows), "mean_latency_delta_ms": mean(r["candidate"]["latency_ms"] - r["baseline"]["latency_ms"] for r in rows), "candidate_p95_ms": p95, "candidate_mean_cost_microusd": cost, "per_trial_accuracy": {str(k): mean(r["candidate"]["correct"] for r in v) for k, v in sorted(by_trial.items())}, "decision": "PASS" if quality >= .9 and p95 <= 700 and cost <= 2000 else "BLOCK", "uncertainty": "Descriptive only. Repeats of four cases are not twelve independent users."} def total_variation(reference: dict[str, int], current: dict[str, int]) -> float: require(set(reference) == set(current), "Drift categories must match") for value in (*reference.values(), *current.values()): integer(value, "category count") require(sum(reference.values()) > 0 and sum(current.values()) > 0, "Nonempty windows required") return .5 * sum(abs(reference[k] / sum(reference.values()) - current[k] / sum(current.values())) for k in reference) def monitor(windows: list[dict], objective: float = .98) -> list[dict]: require(0 < objective < 1, "Objective must be between zero and one") seen, reports = set(), [] for row in windows: require(row["id"] not in seen, "Duplicate window ID") seen.add(row["id"]) for key in ("requests", "bad_requests", "errors", "slow", "critical", "p95_ms", "control_bad"): integer(row[key], key) n, bad = row["requests"], row["bad_requests"] require(n > 0 and bad <= n, "Valid nonempty request denominator required") require(max(row["errors"], row["slow"]) <= bad <= row["errors"] + row["slow"], "Bad requests must be the union of error/slow IDs in this fixture") require(row["critical"] <= n, "Critical request count cannot exceed requests") require(row["control_bad"] <= n, "Control denominator must match candidate window here") require(sum(row["mix"].values()) == n, "Slice counts must cover the window") integer_counts = list(row["mix"].values()) for value in integer_counts: integer(value, "slice count") good = (n - bad) / n reports.append({"id": row["id"], "good_fraction": good, "budget_requests": n * (1 - objective), "burn_multiple": (bad / n) / (1 - objective), "quality_review_required": row["critical"] > 0, "operational_alert": good < objective or row["p95_ms"] > 700, "owner": "fictional-application-owner", "action": "inspect matched window; hold expansion"}) return reports def rollout(plan: dict, windows: list[dict]) -> dict: require(type(plan["offline_pass"]) is bool and type(plan["rollback_ready"]) is bool, "Offline and recovery readiness must be boolean") if not plan["offline_pass"] or not plan["rollback_ready"]: return {"state": "HOLD", "events": [], "live_authorization": False, "reason": "offline gate or recovery prerequisite missing"} require(bool(plan["owner"]) and bool(plan["baseline_artifact"]), "Owner and baseline artifact required") reports = {r["id"]: r for r in monitor(windows)} index = {r["id"]: r for r in windows} events = [] expected_stages = ["shadow", "canary", "expand"] require(len(plan["steps"]) == len(expected_stages), "All three simulation stages required") require([step["stage"] for step in plan["steps"]] == expected_stages, "Stages must occur in order") observed_windows = [step["window"] for step in plan["steps"]] require(len(set(observed_windows)) == len(observed_windows), "Each stage needs a fresh window") require(all(name in index for name in observed_windows), "Unknown observation window") positions = [list(index).index(name) for name in observed_windows] require(positions == sorted(positions), "Observation windows must advance chronologically") for stage, step in zip(expected_stages, plan["steps"]): require(step["stage"] == stage, "Stages must occur in order") require(step["window"] in index, "Unknown observation window") row, report = index[step["window"]], reports[step["window"]] if row["critical"] or report["operational_alert"] or row["bad_requests"] - row["control_bad"] > 3: events.append({"stage": stage, "action": "RECOVER", "window": row["id"]}) return {"state": "RECOVER", "events": events, "live_authorization": False, "reason": "critical, operational, or control-relative stop condition", "recover_to": plan["baseline_artifact"]} if row["requests"] < 100 or step.get("reviewed") is not True: events.append({"stage": stage, "action": "HOLD", "window": row["id"]}) return {"state": "HOLD", "events": events, "live_authorization": False, "reason": "insufficient observations or missing fictional review"} events.append({"stage": stage, "action": "CONTINUE_SIMULATION", "window": row["id"]}) return {"state": "COMPLETE_SIMULATION", "events": events, "live_authorization": False, "reason": "all supplied simulation conditions met"} def incident_summary(record: dict) -> dict: times = [datetime.fromisoformat(event["at"]) for event in record["timeline"]] require(bool(times) and all(t.tzinfo for t in times), "Timezone-aware incident times required") require(times == sorted(times), "Timeline must be chronological") kinds = [event["kind"] for event in record["timeline"]] require(kinds.count("detected") == 1 and kinds.count("contained") == 1, "One detection and one containment event required") detected, contained = times[kinds.index("detected")], times[kinds.index("contained")] require(contained >= detected, "Containment cannot precede detection") require(bool(record["owner"]) and bool(record["follow_up"]["expected_route"]), "Owned follow-up required") return {"incident_id": record["id"], "minutes_detection_to_containment": (contained - detected).total_seconds() / 60, "observations": record["observations"], "hypotheses": record["hypotheses"], "follow_up": record["follow_up"], "root_cause_confirmed": False, "evidence_kind": "synthetic_incident_replay"} def maintenance(changes: list[dict], as_of: str) -> list[dict]: today = date.fromisoformat(as_of) result, seen = [], set() for row in changes: require(row["id"] not in seen and row["owner"], "Unique change ID and owner required") seen.add(row["id"]) due = date.fromisoformat(row["review_due"]) require(row["status"] in {"open", "closed"}, "Unknown maintenance status") if row["status"] == "open": result.append({"id": row["id"], "owner": row["owner"], "kind": row["kind"], "timing": "overdue" if due < today else "due" if due == today else "upcoming", "action": row["action"]}) return sorted(result, key=lambda row: row["id"]) def capstone_evidence(evidence: dict) -> dict: """Collect comparable teaching evidence; do not grant execution authority.""" dataset = dataset_audit(evidence["dataset"]) dataset_id = evidence["dataset"]["version"] + "-holdout" cases = {"schema_version": 1, "dataset_id": dataset_id, "provenance": "handwritten-synthetic", "cases": [ {"id": row["id"], "slice": row["slice"], "input": row["input"], "expected_route": row["adjudicated"], "critical": row["slice"] == "boundary", "rationale": row["rationale"]} for row in evidence["dataset"]["records"] if row["split"] == "holdout"]} def run(name: str, routes: dict) -> dict: return {"schema_version": 1, "run_id": name, "dataset_id": dataset_id, "provenance": "handwritten-synthetic", "system": {"model": "none-fixture", "prompt_revision": name, "code_revision": "fixture-v2"}, "results": [{"case_id": key, "route": route, "latency_ms": 500, "cost_microusd": 1400} for key, route in routes.items()]} policy = read_json(ROOT / "policy.json")[0] route = evaluate(cases, run("capstone-baseline", evidence["holdout_baseline"]), run("capstone-candidate", evidence["holdout_candidate"]), policy) if dataset["decision"] == "PASS" else {"decision": "UNSCORABLE"} distance = total_variation(evidence["windows"][0]["mix"], evidence["windows"][-1]["mix"]) record = read_json(ROOT / "evidence/operations-manifest.json")[0] return {"evidence_kind": "synthetic_fixture_replay", "dataset": dataset, "route": route, "review": review_summary(evidence["reviews"]), "calibration": grader_calibration(evidence["calibration"]), "comparison": paired_comparison(evidence["trials"], evidence["trial_plan"]), "monitor": monitor(evidence["windows"]), "rollout": rollout(evidence["plan"], evidence["windows"]), "maintenance": maintenance(evidence["changes"], evidence["as_of"]), "drift": {"distance": distance, "threshold": .15, "reviewed": False}, "manifest": record, "manifest_changed": verify_manifest(record), "rubric_review_complete": False, "follow_up_reviewed": False, "incident": incident_summary(evidence["incident"]), "limitations": ["All outputs, reviewer judgments, costs, latencies, and windows are invented.", "Visible teaching holdout is not an unseen real benchmark.", "No actual learner, model, rollout, incident, or service is evaluated."]} def commands(evidence: dict) -> dict: return { "dataset": lambda: dataset_audit(evidence["dataset"]), "review": lambda: review_summary(evidence["reviews"]), "calibrate": lambda: grader_calibration(evidence["calibration"]), "manifest": manifest, "compare": lambda: paired_comparison(evidence["trials"], evidence["trial_plan"]), "monitor": lambda: monitor(evidence["windows"]), "drift": lambda: {"total_variation": total_variation(evidence["windows"][0]["mix"], evidence["windows"][-1]["mix"]), "review_threshold": .15, "meaning": "input mix shift; not proof of quality loss"}, "rollout": lambda: rollout(evidence["plan"], evidence["windows"]), "incident": lambda: incident_summary(evidence["incident"]), "maintenance": lambda: maintenance(evidence["changes"], evidence["as_of"]), } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=("dataset", "review", "calibrate", "manifest", "compare", "monitor", "drift", "rollout", "incident", "maintenance")) args = parser.parse_args() try: result = commands(load())[args.command]() print(json.dumps({"evidence_kind": "synthetic_fixture_replay", "live_authorization": False, "command": args.command, "result": result}, indent=2, sort_keys=True, allow_nan=False)) return 0 # Successful computation, including a simulated BLOCK/REVIEW/HOLD. except (InputError, KeyError, TypeError, ValueError, OSError) as exc: print(f"INPUT ERROR: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())