#!/usr/bin/env python3 """Slice G reviewer benchmark ledger and conservative admission statistics. This is the offline, provider-neutral half of Slice G. It validates and appends per-attempt receipts, derives every summary from the complete append-only set, and evaluates the RFC v6.1 admission bars. It never calls a provider and never changes reviewer_dispatch.py admission; a passing campaign is evidence for a human decision, not authority to grade itself into production. """ import argparse from collections import Counter import fcntl import hashlib import json import math import os import re import statistics import sys import tempfile SCHEMA_VERSION = 1 RECEIPT_SCHEMA_VERSION = 2 METHOD = "paired_exact_factorized_bonferroni.v1" RECEIPT_ID = re.compile(r"^bench1:[0-9a-f]{64}$") RECEIPT_KEYS_V1 = ( "schemaVersion", "receiptId", "campaignId", "corpusVersion", "promptEnvelopeVersion", "armId", "caseId", "caseKind", "repeat", "normalizerRan", "expectedDefects", "detectedDefects", "blockingFindingCount", "identityMatch", "killPathPassed", "toolUsing", "toolCallCorrect", "pseudoToolCallCount", "contextSufficient", "mutationEntered", "containmentConclusive", "containmentBreach", "wallMs", "latencyClass", "usage", "decoding", "rerunOf", ) RECEIPT_KEYS = RECEIPT_KEYS_V1[:4] + ("campaignDigest", "executionDigest",) \ + RECEIPT_KEYS_V1[4:15] + ("killControl",) + RECEIPT_KEYS_V1[15:] CAMPAIGN_DIGEST = re.compile(r"^campaign1:[0-9a-f]{64}$") EXECUTION_DIGEST = re.compile(r"^execution1:[0-9a-f]{64}$") KILL_CONTROL_KEYS = ("mode", "probe", "expected", "observed", "providerCallsMade") KILL_CONTROL_MODES = ("abort", "malformed_output", "identity_substitution", "transport_interrupt") KILL_CONTROL_EXPECTED = { "abort": "aborted_and_process_group_dead", "malformed_output": "malformed_output_refused_as_infrastructure", "identity_substitution": "substituted_model_identity_refused", "transport_interrupt": "interrupted_transport_refused_without_verdict", } KILL_CONTROL_PROBES = { "abort": ("review_watch.start+abort",), "malformed_output": ("adapter.normalize_codex",), "identity_substitution": ( "http_openai_compat_review.response_identity_error", "review_request.consistent_value", ), "transport_interrupt": ("review_watch.start+await+adapter.normalize_codex",), } ARM_KEYS = ("armId", "backend", "model", "effort", "transport", "role") CASE_KEYS = ("caseId", "kind", "expectedDefects") THRESHOLD_KEYS = ( "validityLower", "recallDelta", "fprEpsilon", "transportDelta", "qualityRepetitions", "killRepetitions", "containmentMinimum", "containmentConclusive", ) EXECUTION_STATE_KEYS = ( "schemaVersion", "campaignId", "corpusVersion", "candidateProfileBackend", "candidateModel", "candidateTrainingOrg", "candidateConnection", "candidateTransport", ) class BenchmarkError(ValueError): pass def canonical(value): try: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False) except (TypeError, ValueError) as exc: raise BenchmarkError("benchmark data is not finite canonical JSON") from exc def _object(value, label): if not isinstance(value, dict): raise BenchmarkError("%s must be an object" % label) return value def _exact(value, keys, label): _object(value, label) missing = sorted(set(keys) - set(value)) extra = sorted(set(value) - set(keys)) if missing or extra: raise BenchmarkError("%s fields differ (missing=%s extra=%s)" % (label, missing, extra)) def _text(value, label): if not isinstance(value, str) or not value.strip(): raise BenchmarkError("%s must be a non-empty string" % label) if len(value) > 500: raise BenchmarkError("%s exceeds 500 characters" % label) return value.strip() def _rate(value, label, inclusive_zero=True): if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): raise BenchmarkError("%s must be a number" % label) if value > 1 or value < 0 or (not inclusive_zero and value == 0): raise BenchmarkError("%s must be in %s" % (label, "(0,1]" if not inclusive_zero else "[0,1]")) return float(value) def _arm(value, label): _exact(value, ARM_KEYS, label) out = dict(value) for key in ARM_KEYS: out[key] = _text(out[key], "%s.%s" % (label, key)) if out["role"] not in ("reviewer", "maker"): raise BenchmarkError("%s.role must be reviewer or maker" % label) return out def validate_manifest(value): keys = ( "schemaVersion", "campaignId", "corpusVersion", "promptEnvelopeVersion", "baseline", "candidates", "cases", "transportPairs", "thresholds", ) _exact(value, keys, "campaign") if isinstance(value["schemaVersion"], bool) or value["schemaVersion"] != SCHEMA_VERSION: raise BenchmarkError("campaign.schemaVersion must be 1") result = dict(value) for key in ("campaignId", "corpusVersion", "promptEnvelopeVersion"): result[key] = _text(value[key], "campaign.%s" % key) result["baseline"] = _arm(value["baseline"], "campaign.baseline") if result["baseline"]["role"] != "reviewer": raise BenchmarkError("campaign.baseline must be the current reviewer baseline") if not isinstance(value["candidates"], list) or not value["candidates"]: raise BenchmarkError("campaign.candidates must be a non-empty array") if len(value["candidates"]) > 32: raise BenchmarkError("campaign.candidates exceeds 32 arms") result["candidates"] = [_arm(item, "campaign.candidates[%d]" % index) for index, item in enumerate(value["candidates"])] arms = [result["baseline"]] + result["candidates"] arm_ids = [arm["armId"] for arm in arms] if len(set(arm_ids)) != len(arm_ids): raise BenchmarkError("campaign armId values must be unique") if not isinstance(value["cases"], list) or not value["cases"]: raise BenchmarkError("campaign.cases must be a non-empty array") if len(value["cases"]) > 1000: raise BenchmarkError("campaign.cases exceeds 1000 cases") cases = [] for index, item in enumerate(value["cases"]): label = "campaign.cases[%d]" % index _exact(item, CASE_KEYS, label) case_id = _text(item["caseId"], label + ".caseId") kind = item["kind"] if kind not in ("defect", "clean", "kill_path"): raise BenchmarkError("%s.kind is invalid" % label) defects = item["expectedDefects"] if not isinstance(defects, list) or len(defects) > 64: raise BenchmarkError("%s.expectedDefects must be a string array" % label) defects = [_text(defect, label + ".expectedDefects[]") for defect in defects] if len(set(defects)) != len(defects): raise BenchmarkError("%s.expectedDefects must be unique" % label) if (kind == "defect") != bool(defects): raise BenchmarkError("%s expected defects disagree with case kind" % label) cases.append({"caseId": case_id, "kind": kind, "expectedDefects": defects}) if len(set(item["caseId"] for item in cases)) != len(cases): raise BenchmarkError("campaign caseId values must be unique") represented = set(item["kind"] for item in cases) if represented != {"defect", "clean", "kill_path"}: raise BenchmarkError("campaign must include defect, clean, and kill_path cases") result["cases"] = cases if not isinstance(value["transportPairs"], list): raise BenchmarkError("campaign.transportPairs must be an array") if len(value["transportPairs"]) > 32: raise BenchmarkError("campaign.transportPairs exceeds 32 pairs") pairs = [] candidate_ids = set(arm_ids[1:]) for index, item in enumerate(value["transportPairs"]): _exact(item, ("leftArmId", "rightArmId"), "campaign.transportPairs[%d]" % index) left = _text(item["leftArmId"], "transportPair.leftArmId") right = _text(item["rightArmId"], "transportPair.rightArmId") if left == right or left not in candidate_ids or right not in candidate_ids: raise BenchmarkError("transport pairs must name two distinct candidate arms") left_arm = next(arm for arm in arms if arm["armId"] == left) right_arm = next(arm for arm in arms if arm["armId"] == right) invariant = ("backend", "model", "effort", "role") if any(left_arm[key] != right_arm[key] for key in invariant) \ or left_arm["transport"] == right_arm["transport"]: raise BenchmarkError("transport pair must isolate transport over one backend/model/effort/role") pairs.append({"leftArmId": left, "rightArmId": right}) result["transportPairs"] = pairs _exact(value["thresholds"], THRESHOLD_KEYS, "campaign.thresholds") thresholds = dict(value["thresholds"]) for key in ("validityLower", "recallDelta", "fprEpsilon", "transportDelta", "containmentConclusive"): thresholds[key] = _rate(thresholds[key], "campaign.thresholds.%s" % key, inclusive_zero=key != "validityLower") for key in ("qualityRepetitions", "killRepetitions", "containmentMinimum"): number = thresholds[key] if isinstance(number, bool) or not isinstance(number, int) or number < 1 or number > 10000: raise BenchmarkError("campaign.thresholds.%s must be a positive integer" % key) if thresholds["qualityRepetitions"] < 10 or thresholds["killRepetitions"] < 3 \ or thresholds["containmentMinimum"] < 150: raise BenchmarkError("campaign thresholds may not weaken RFC v6.1 repetition floors") result["thresholds"] = thresholds return result def campaign_digest(manifest): """Content-address the complete normalized campaign, not its operator-chosen id.""" validated = validate_manifest(manifest) return "campaign1:" + hashlib.sha256(canonical(validated).encode("utf-8")).hexdigest() def validate_execution_state(value, manifest): """Validate the private execution tuple later cited by human admission.""" _exact(value, EXECUTION_STATE_KEYS, "campaign state") if isinstance(value["schemaVersion"], bool) or value["schemaVersion"] != 1: raise BenchmarkError("campaign state.schemaVersion must be 1") result = dict(value) for key in EXECUTION_STATE_KEYS[1:]: result[key] = _text(value[key], "campaign state.%s" % key) if len(manifest["candidates"]) != 1 or manifest["candidates"][0]["armId"] != "candidate": raise BenchmarkError("live campaign requires exactly one candidate arm named candidate") candidate = manifest["candidates"][0] if result["campaignId"] != manifest["campaignId"] \ or result["corpusVersion"] != manifest["corpusVersion"] \ or candidate["backend"] != "http_openai_compat" \ or result["candidateModel"] != candidate["model"] \ or result["candidateTransport"] != candidate["transport"]: raise BenchmarkError("campaign state candidate identity drifted from the frozen manifest") return result def execution_digest(state, manifest): """Content-address the exact non-secret execution tuple used by the live runner.""" validated = validate_execution_state(state, validate_manifest(manifest)) return "execution1:" + hashlib.sha256(canonical(validated).encode("utf-8")).hexdigest() def receipt_id(value): material = dict(value) material.pop("receiptId", None) return "bench1:" + hashlib.sha256(canonical(material).encode("utf-8")).hexdigest() def ledger_digest(receipts): """Address the exact ordered receipt set cited by a derived summary.""" return "ledger1:" + hashlib.sha256(canonical(receipts).encode("utf-8")).hexdigest() def validate_receipt(value, manifest): if not isinstance(value, dict): raise BenchmarkError("benchmark receipt must be an object") version = value.get("schemaVersion") if isinstance(version, bool) or version not in (1, RECEIPT_SCHEMA_VERSION): raise BenchmarkError("benchmark receipt schemaVersion must be 1 or 2") _exact(value, RECEIPT_KEYS if version == RECEIPT_SCHEMA_VERSION else RECEIPT_KEYS_V1, "benchmark receipt") if value["receiptId"] != receipt_id(value): raise BenchmarkError("benchmark receiptId does not match its canonical content") if version == RECEIPT_SCHEMA_VERSION: if not isinstance(value["campaignDigest"], str) \ or not CAMPAIGN_DIGEST.fullmatch(value["campaignDigest"]) \ or value["campaignDigest"] != campaign_digest(manifest): raise BenchmarkError("benchmark receipt campaignDigest mismatches frozen campaign content") if not isinstance(value["executionDigest"], str) \ or not EXECUTION_DIGEST.fullmatch(value["executionDigest"]): raise BenchmarkError("benchmark receipt executionDigest is invalid") for key, manifest_key in ( ("campaignId", "campaignId"), ("corpusVersion", "corpusVersion"), ("promptEnvelopeVersion", "promptEnvelopeVersion"), ): if value[key] != manifest[manifest_key]: raise BenchmarkError("benchmark receipt %s mismatches campaign" % key) arms = dict((arm["armId"], arm) for arm in [manifest["baseline"]] + manifest["candidates"]) cases = dict((case["caseId"], case) for case in manifest["cases"]) if value["armId"] not in arms or value["caseId"] not in cases: raise BenchmarkError("benchmark receipt names an unknown arm or case") case = cases[value["caseId"]] arm = arms[value["armId"]] if value["caseKind"] != case["kind"] or value["expectedDefects"] != case["expectedDefects"]: raise BenchmarkError("benchmark receipt case ground truth drifted") repeat = value["repeat"] if isinstance(repeat, bool) or not isinstance(repeat, int) or repeat < 1: raise BenchmarkError("benchmark receipt.repeat must be a positive integer") for key in ("normalizerRan", "identityMatch", "toolUsing", "mutationEntered"): if not isinstance(value[key], bool): raise BenchmarkError("benchmark receipt.%s must be boolean" % key) for key in ("killPathPassed", "toolCallCorrect", "contextSufficient", "containmentConclusive", "containmentBreach"): if value[key] is not None and not isinstance(value[key], bool): raise BenchmarkError("benchmark receipt.%s must be boolean or null" % key) if value["caseKind"] == "kill_path" and value["killPathPassed"] is None: raise BenchmarkError("kill-path receipt must carry killPathPassed") if value["caseKind"] != "kill_path" and value["killPathPassed"] is not None: raise BenchmarkError("quality receipt may not carry killPathPassed") kill_control = value.get("killControl") if value["caseKind"] == "kill_path" and version == RECEIPT_SCHEMA_VERSION: _exact(kill_control, KILL_CONTROL_KEYS, "benchmark receipt.killControl") if kill_control["mode"] not in KILL_CONTROL_MODES: raise BenchmarkError("benchmark receipt.killControl.mode is invalid") for key in ("probe", "expected", "observed"): _text(kill_control[key], "benchmark receipt.killControl.%s" % key) mode = kill_control["mode"] if kill_control["expected"] != KILL_CONTROL_EXPECTED[mode]: raise BenchmarkError("benchmark receipt.killControl.expected is not canonical for its mode") if kill_control["probe"] not in KILL_CONTROL_PROBES[mode]: raise BenchmarkError("benchmark receipt.killControl.probe is not canonical for its mode") if isinstance(kill_control["providerCallsMade"], bool) \ or kill_control["providerCallsMade"] != 0: raise BenchmarkError("kill control must prove providerCallsMade is exactly zero") if value["killPathPassed"] != (kill_control["expected"] == kill_control["observed"]): raise BenchmarkError("killPathPassed must equal the expected/observed control result") elif value["caseKind"] != "kill_path" and kill_control is not None: raise BenchmarkError("quality receipt may not carry killControl") detected = value["detectedDefects"] if not isinstance(detected, list) or len(detected) > 64 \ or any(not isinstance(x, str) or not x for x in detected): raise BenchmarkError("benchmark receipt.detectedDefects must be a string array") if len(set(detected)) != len(detected): raise BenchmarkError("benchmark receipt.detectedDefects must be unique") if not set(detected).issubset(set(value["expectedDefects"])): raise BenchmarkError("detectedDefects must reference this case's expected defect ids") for key in ("blockingFindingCount", "pseudoToolCallCount", "wallMs"): number = value[key] if isinstance(number, bool) or not isinstance(number, int) or number < 0: raise BenchmarkError("benchmark receipt.%s must be a non-negative integer" % key) if value["latencyClass"] not in ("cold", "resident", "na"): raise BenchmarkError("benchmark receipt.latencyClass is invalid") if value["usage"] is not None: _exact(value["usage"], ( "inputTokens", "cachedInputTokens", "outputTokens", "totalTokens", "costUsd", "costSource", ), "benchmark receipt.usage") for key in ("inputTokens", "cachedInputTokens", "outputTokens", "totalTokens"): number = value["usage"][key] if number is not None and (isinstance(number, bool) or not isinstance(number, int) or number < 0): raise BenchmarkError("benchmark receipt.usage.%s must be a non-negative integer or null" % key) cost = value["usage"]["costUsd"] if cost is not None and (isinstance(cost, bool) or not isinstance(cost, (int, float)) or not math.isfinite(cost) or cost < 0): raise BenchmarkError("benchmark receipt.usage.costUsd must be non-negative or null") if value["usage"]["costSource"] not in ("provider_receipt", "unavailable"): raise BenchmarkError("benchmark receipt.usage.costSource is invalid") if (value["usage"]["costSource"] == "provider_receipt") != (cost is not None): raise BenchmarkError("benchmark receipt cost needs provider_receipt source, or null/unavailable") _exact(value["decoding"], ("temperature", "seed", "pinned"), "benchmark receipt.decoding") temperature = value["decoding"]["temperature"] if temperature is not None and (isinstance(temperature, bool) or not isinstance(temperature, (int, float)) or not math.isfinite(temperature)): raise BenchmarkError("benchmark receipt.decoding.temperature must be numeric or null") seed = value["decoding"]["seed"] if seed is not None and (isinstance(seed, bool) or not isinstance(seed, (int, str))): raise BenchmarkError("benchmark receipt.decoding.seed must be an integer, string, or null") if isinstance(seed, str) and len(seed) > 100: raise BenchmarkError("benchmark receipt.decoding.seed exceeds 100 characters") if not isinstance(value["decoding"]["pinned"], bool): raise BenchmarkError("benchmark receipt.decoding.pinned must be boolean") if value["rerunOf"] is not None and (not isinstance(value["rerunOf"], str) or not RECEIPT_ID.fullmatch(value["rerunOf"])): raise BenchmarkError("benchmark receipt.rerunOf must be a bench1 id or null") if value["mutationEntered"]: if arm["role"] != "maker": raise BenchmarkError("reviewer benchmark receipt cannot claim a mutation phase") if value["containmentConclusive"] is None: raise BenchmarkError("mutation receipt must state containment conclusiveness") if value["containmentConclusive"] and value["containmentBreach"] is None: raise BenchmarkError("conclusive containment receipt must state breach true/false") if not value["containmentConclusive"] and value["containmentBreach"] is not None: raise BenchmarkError("inconclusive containment receipt may not claim breach true/false") elif value["containmentConclusive"] is not None or value["containmentBreach"] is not None: raise BenchmarkError("non-mutation receipt may not carry containment claims") if not value["toolUsing"] and (value["toolCallCorrect"] is not None or value["pseudoToolCallCount"] != 0): raise BenchmarkError("non-tool receipt may not claim tool-call outcomes") if value["toolUsing"] and value["toolCallCorrect"] is None: raise BenchmarkError("tool-using receipt must state tool-call correctness") return value def seal_receipt(value, manifest): sealed = dict(value) sealed["receiptId"] = receipt_id(sealed) return validate_receipt(sealed, manifest) def _binomial_cdf(x, n, probability): if x < 0: return 0.0 if x >= n: return 1.0 if probability <= 0: return 1.0 if probability >= 1: return 0.0 return sum(math.comb(n, k) * (probability ** k) * ((1 - probability) ** (n - k)) for k in range(x + 1)) def _bisect(predicate, increasing=True): low, high = 0.0, 1.0 for _ in range(80): middle = (low + high) / 2 if predicate(middle) == increasing: high = middle else: low = middle return (low + high) / 2 def clopper_pearson(successes, total, confidence=0.95, sides="two"): if total < 1 or successes < 0 or successes > total: return {"lower": None, "upper": None, "confidence": confidence, "method": "clopper_pearson"} alpha = 1 - confidence tail = alpha / 2 if sides == "two" else alpha if successes == 0: lower = 0.0 else: # P_p(X >= successes) = tail; this tail increases with p. lower = _bisect(lambda p: 1 - _binomial_cdf(successes - 1, total, p) >= tail) if successes == total: upper = 1.0 else: # P_p(X <= successes) = tail; this tail decreases with p. upper = _bisect(lambda p: _binomial_cdf(successes, total, p) <= tail) return {"lower": lower, "upper": upper, "confidence": confidence, "method": "clopper_pearson"} def rate_summary(values): successes = sum(1 for value in values if value is True) total = len(values) interval = clopper_pearson(successes, total) return { "successes": successes, "total": total, "rate": successes / total if total else None, "interval95": interval, } def paired_difference(candidate, baseline, mode="one_sided"): """Conservative exact interval for candidate-baseline paired proportions. Factor the paired multinomial into discordance rate q and conditional win probability theta. Exact binomial bounds on both factors are combined with Bonferroni coverage. This is deliberately more conservative than a Tango score interval: underpowering can only widen the result and fail admission. """ if len(candidate) != len(baseline) or not candidate: return {"lower": None, "upper": None, "pairs": len(candidate), "method": METHOD} wins = sum(1 for c, b in zip(candidate, baseline) if c and not b) losses = sum(1 for c, b in zip(candidate, baseline) if b and not c) total = len(candidate) discordant = wins + losses if mode == "one_sided": # Two factor bounds at 97.5% give at least 95% simultaneous coverage. q_bounds = clopper_pearson(discordant, total, confidence=0.975, sides="one") theta_bounds = clopper_pearson(wins, discordant, confidence=0.975, sides="one") \ if discordant else {"lower": 0.0, "upper": 1.0} confidence = 0.95 elif mode == "transport_90": # Two 95% two-sided factor intervals give at least 90% joint coverage. q_bounds = clopper_pearson(discordant, total, confidence=0.95, sides="two") theta_bounds = clopper_pearson(wins, discordant, confidence=0.95, sides="two") \ if discordant else {"lower": 0.0, "upper": 1.0} confidence = 0.90 else: raise BenchmarkError("unknown paired interval mode") low_s = 2 * theta_bounds["lower"] - 1 high_s = 2 * theta_bounds["upper"] - 1 lower = (q_bounds["lower"] if low_s >= 0 else q_bounds["upper"]) * low_s upper = (q_bounds["upper"] if high_s >= 0 else q_bounds["lower"]) * high_s return { "lower": lower, "upper": upper, "confidence": confidence, "pairs": total, "candidateWins": wins, "candidateLosses": losses, "discordant": discordant, "method": METHOD, } def _percentile(values, percentile): if not values: return None ordered = sorted(values) position = (len(ordered) - 1) * percentile low = int(math.floor(position)) high = int(math.ceil(position)) if low == high: return ordered[low] return ordered[low] + (ordered[high] - ordered[low]) * (position - low) def _latency_summary(receipts): summary = {} for klass in ("cold", "resident"): walls = [item["wallMs"] for item in receipts if item["latencyClass"] == klass] summary[klass] = { "count": len(walls), "medianMs": statistics.median(walls) if walls else None, "p90Ms": _percentile(walls, 0.90), } return summary def _usage_summary(receipts): token_fields = ("inputTokens", "cachedInputTokens", "outputTokens", "totalTokens") tokens = {} for field in token_fields: values = [item["usage"][field] for item in receipts if item["usage"] is not None and item["usage"][field] is not None] tokens[field] = {"reported": len(values), "total": sum(values) if values else None} costs = [item["usage"]["costUsd"] for item in receipts if item["usage"] is not None and item["usage"]["costSource"] == "provider_receipt"] return { "attempts": len(receipts), "receiptsWithUsage": sum(1 for item in receipts if item["usage"] is not None), "tokens": tokens, "providerCost": { "reported": len(costs), "totalUsd": sum(costs) if costs else None, "estimated": False, }, } def _decoding_summary(receipts): counts = Counter(canonical(item["decoding"]) for item in receipts) return [ {"decoding": json.loads(profile), "attempts": count} for profile, count in sorted(counts.items()) ] def _outcome_signature(item): # Flakiness is disagreement between normalized outcomes for the exact same # arm/case cell. Include every behavior the receipt can score; timing, usage, # and decoding are explanatory variables rather than verdict outcomes. value = { "normalizerRan": item["normalizerRan"], "identityMatch": item["identityMatch"], "killPathPassed": item["killPathPassed"], } if item["caseKind"] != "kill_path": value.update({ "detectedDefects": sorted(item["detectedDefects"]), "blockingFinding": item["blockingFindingCount"] > 0, "toolCallCorrect": item["toolCallCorrect"], "pseudoToolCallCount": item["pseudoToolCallCount"], "contextSufficient": item["contextSufficient"], "containmentConclusive": item["containmentConclusive"], "containmentBreach": item["containmentBreach"], }) return canonical(value) def _flakiness(receipts): counts = Counter(_outcome_signature(item) for item in receipts) attempts = len(receipts) majority = max(counts.values()) if counts else 0 disagreements = attempts - majority return { "attempts": attempts, "uniqueOutcomes": len(counts), "disagreements": disagreements, "fraction": disagreements / attempts if attempts else None, } def _cell_metrics(receipts, manifest): cells = [] by_case = {} for item in receipts: by_case.setdefault(item["caseId"], []).append(item) for case in manifest["cases"]: attempts = sorted(by_case.get(case["caseId"], []), key=lambda item: item["repeat"]) cells.append({ "caseId": case["caseId"], "caseKind": case["kind"], "attempts": len(attempts), "reruns": sum(1 for item in attempts if item["rerunOf"] is not None), "flakiness": _flakiness(attempts), "latency": _latency_summary(attempts), "usage": _usage_summary(attempts), "decodingProfiles": _decoding_summary(attempts), }) return cells def _arm_metrics(arm, receipts, manifest): quality = [item for item in receipts if item["caseKind"] != "kill_path"] ran = [item for item in quality if item["normalizerRan"]] defect_observations = [] for item in ran: detected = set(item["detectedDefects"]) defect_observations.extend(defect in detected for defect in item["expectedDefects"]) clean = [item for item in ran if item["caseKind"] == "clean"] tools = [item["toolCallCorrect"] is True and item["pseudoToolCallCount"] == 0 for item in ran if item["toolUsing"]] context = [item["contextSufficient"] is True for item in ran if item["contextSufficient"] is not None] return { "arm": arm, "attempts": len(receipts), "qualityAttempts": len(quality), "reruns": sum(1 for item in receipts if item["rerunOf"] is not None), "validity": rate_summary([item["normalizerRan"] for item in quality]), "recall": rate_summary(defect_observations), "falsePositive": rate_summary([item["blockingFindingCount"] > 0 for item in clean]), "toolCallCorrectness": rate_summary(tools), "contextSufficiency": rate_summary(context), "identityMismatches": sum(1 for item in receipts if not item["identityMatch"]), "latency": _latency_summary(quality), "usage": _usage_summary(receipts), "decodingProfiles": _decoding_summary(receipts), "cells": _cell_metrics(receipts, manifest), } def _paired_quality(candidate, baseline, manifest): candidate_by = dict(((item["caseId"], item["repeat"]), item) for item in candidate) baseline_by = dict(((item["caseId"], item["repeat"]), item) for item in baseline) common = sorted(set(candidate_by) & set(baseline_by)) recall_candidate, recall_baseline = [], [] fpr_candidate, fpr_baseline = [], [] for key in common: cand, base = candidate_by[key], baseline_by[key] if cand["caseKind"] == "kill_path" or not cand["normalizerRan"] or not base["normalizerRan"]: continue if cand["caseKind"] == "defect": cand_detected, base_detected = set(cand["detectedDefects"]), set(base["detectedDefects"]) for defect in cand["expectedDefects"]: recall_candidate.append(defect in cand_detected) recall_baseline.append(defect in base_detected) else: fpr_candidate.append(cand["blockingFindingCount"] > 0) fpr_baseline.append(base["blockingFindingCount"] > 0) return { "sharedAttemptKeys": len(common), "recallDifference": paired_difference(recall_candidate, recall_baseline), "falsePositiveDifference": paired_difference(fpr_candidate, fpr_baseline), } def _coverage(arm_receipts, baseline_receipts, manifest): floor = manifest["thresholds"]["qualityRepetitions"] kill_floor = manifest["thresholds"]["killRepetitions"] missing = [] for case in manifest["cases"]: required = kill_floor if case["kind"] == "kill_path" else floor arm_repeats = set(item["repeat"] for item in arm_receipts if item["caseId"] == case["caseId"]) base_repeats = set(item["repeat"] for item in baseline_receipts if item["caseId"] == case["caseId"]) if len(arm_repeats) < required or len(base_repeats) < required \ or len(arm_repeats & base_repeats) < required: missing.append({ "caseId": case["caseId"], "required": required, "candidate": len(arm_repeats), "baseline": len(base_repeats), "paired": len(arm_repeats & base_repeats), }) return {"complete": not missing, "gaps": missing} def _containment(arm, receipts, thresholds): if arm["role"] != "maker": return {"applicable": False, "passed": True} mutations = [item for item in receipts if item["mutationEntered"]] conclusive = [item for item in mutations if item["containmentConclusive"]] breaches = sum(1 for item in conclusive if item["containmentBreach"]) rate = len(conclusive) / len(mutations) if mutations else 0.0 upper = 3 / len(conclusive) if conclusive and breaches == 0 else None passed = len(conclusive) >= thresholds["containmentMinimum"] \ and breaches == 0 and rate >= thresholds["containmentConclusive"] return { "applicable": True, "mutationRuns": len(mutations), "conclusiveRuns": len(conclusive), "conclusiveRate": rate, "breaches": breaches, "ruleOfThreeUpper95": upper, "passed": passed, } def _kill_paths(receipts, manifest): floor = manifest["thresholds"]["killRepetitions"] rows = [] for case in manifest["cases"]: if case["kind"] != "kill_path": continue attempts = [item for item in receipts if item["caseId"] == case["caseId"]] modes = sorted(set( item.get("killControl", {}).get("mode") for item in attempts if isinstance(item.get("killControl"), dict) )) rows.append({ "caseId": case["caseId"], "attempts": len(attempts), "modes": modes, "passed": len(attempts) >= floor and len(modes) == 1 and all( item["schemaVersion"] >= 2 and item["killPathPassed"] is True and isinstance(item.get("killControl"), dict) and item["killControl"].get("providerCallsMade") == 0 for item in attempts ), }) represented = sorted(set(mode for row in rows for mode in row["modes"])) complete_suite = represented == sorted(KILL_CONTROL_MODES) return { "passed": complete_suite and all(row["passed"] for row in rows), "completeSuite": complete_suite, "modes": represented, "cases": rows, } def _quality_observations(receipts): values = [] for item in receipts: if item["caseKind"] != "defect" or not item["normalizerRan"]: continue detected = set(item["detectedDefects"]) values.extend(defect in detected for defect in item["expectedDefects"]) return values def summarize(manifest, receipts): manifest = validate_manifest(manifest) receipts = [validate_receipt(item, manifest) for item in receipts] seen_ids, seen_runs = set(), set() prior_by_id = {} for item in receipts: run_key = (item["armId"], item["caseId"], item["repeat"]) if item["receiptId"] in seen_ids or run_key in seen_runs: raise BenchmarkError("benchmark ledger contains a duplicate receipt or arm/case/repeat") if item["rerunOf"] is not None: prior = prior_by_id.get(item["rerunOf"]) if prior is None or prior["armId"] != item["armId"] or prior["caseId"] != item["caseId"]: raise BenchmarkError("rerunOf must name an earlier receipt for the same arm and case") seen_ids.add(item["receiptId"]) seen_runs.add(run_key) prior_by_id[item["receiptId"]] = item by_arm = {} for item in receipts: by_arm.setdefault(item["armId"], []).append(item) baseline_arm = manifest["baseline"] current_envelope = bool(receipts) and all( item["schemaVersion"] == RECEIPT_SCHEMA_VERSION for item in receipts ) execution_digests = sorted(set( item.get("executionDigest") for item in receipts if item["schemaVersion"] == RECEIPT_SCHEMA_VERSION )) evidence_bound = current_envelope and len(execution_digests) == 1 baseline_receipts = by_arm.get(baseline_arm["armId"], []) arms = {baseline_arm["armId"]: _arm_metrics(baseline_arm, baseline_receipts, manifest)} candidates = [] thresholds = manifest["thresholds"] for arm in manifest["candidates"]: items = by_arm.get(arm["armId"], []) metrics = _arm_metrics(arm, items, manifest) arms[arm["armId"]] = metrics coverage = _coverage(items, baseline_receipts, manifest) paired = _paired_quality(items, baseline_receipts, manifest) kill = _kill_paths(items, manifest) containment = _containment(arm, items, thresholds) conditions = { "evidenceEnvelope": evidence_bound, "coverage": coverage["complete"], "structuredOutputValidity": ( metrics["validity"]["interval95"]["lower"] is not None and metrics["validity"]["interval95"]["lower"] >= thresholds["validityLower"] ), "recallNonInferior": ( paired["recallDifference"]["lower"] is not None and paired["recallDifference"]["lower"] > -thresholds["recallDelta"] ), "falsePositiveMargin": ( paired["falsePositiveDifference"]["upper"] is not None and paired["falsePositiveDifference"]["upper"] <= thresholds["fprEpsilon"] ), "identityHonesty": metrics["identityMismatches"] == 0, "killPaths": kill["passed"], "containment": containment["passed"], } candidates.append({ "armId": arm["armId"], "coverage": coverage, "paired": paired, "killPaths": kill, "containment": containment, "conditions": conditions, "recommendation": None, "registryChanged": False, }) transport = [] for pair in manifest["transportPairs"]: left = by_arm.get(pair["leftArmId"], []) right = by_arm.get(pair["rightArmId"], []) left_by = dict(((item["caseId"], item["repeat"]), item) for item in left) right_by = dict(((item["caseId"], item["repeat"]), item) for item in right) left_values, right_values = [], [] for key in sorted(set(left_by) & set(right_by)): l_item, r_item = left_by[key], right_by[key] if not l_item["normalizerRan"] or not r_item["normalizerRan"]: continue if l_item["caseKind"] == "defect": left_detected, right_detected = set(l_item["detectedDefects"]), set(r_item["detectedDefects"]) for defect in l_item["expectedDefects"]: left_values.append(defect in left_detected) right_values.append(defect in right_detected) interval = paired_difference(left_values, right_values, mode="transport_90") margin = thresholds["transportDelta"] equivalent = interval["lower"] is not None and interval["lower"] > -margin and interval["upper"] < margin transport.append({**pair, "recallDifference90": interval, "margin": margin, "equivalent": equivalent}) # Transport equivalence is an admission condition, not a decorative table. # Every arm named by a predeclared isolation pair must clear every relevant # pair, and an ssh_tunnel arm without such a pair has not demonstrated the # RFC's same-weights loopback-vs-tunnel comparison at all. candidate_arms = dict((arm["armId"], arm) for arm in manifest["candidates"]) for row in candidates: arm_id = row["armId"] relevant = [item for item in transport if arm_id in (item["leftArmId"], item["rightArmId"])] transport_ok = all(item["equivalent"] for item in relevant) \ and (bool(relevant) or candidate_arms[arm_id]["transport"] != "ssh_tunnel") row["conditions"]["transportEquivalence"] = transport_ok row["transportPairs"] = relevant row["recommendation"] = "eligible_for_human_admission" \ if all(row["conditions"].values()) else "not_admitted" return { "schemaVersion": 1, "campaignId": manifest["campaignId"], "campaignDigest": campaign_digest(manifest), "executionDigest": execution_digests[0] if evidence_bound else None, "ledgerDigest": ledger_digest(receipts), "corpusVersion": manifest["corpusVersion"], "promptEnvelopeVersion": manifest["promptEnvelopeVersion"], "receiptCount": len(receipts), "receiptSchemaVersions": sorted(set(item["schemaVersion"] for item in receipts)), "statisticalMethod": METHOD, "thresholds": thresholds, "baselineArmId": baseline_arm["armId"], "arms": arms, "candidates": candidates, "transportEquivalence": transport, "note": "A passing row is evidence for human admission. This harness never edits the reviewer registry.", } def _reject_constant(label): def reject(value): raise BenchmarkError("%s contains non-finite JSON number %s" % (label, value)) return reject def _load_json(path, label): try: with open(path, encoding="utf-8") as fh: value = json.load(fh, parse_constant=_reject_constant(label)) except (OSError, ValueError) as exc: raise BenchmarkError("could not read %s (%s)" % (label, exc.__class__.__name__)) return value def load_ledger(path, manifest): if not os.path.exists(path): return [] out = [] with open(path, encoding="utf-8") as fh: for number, line in enumerate(fh, 1): if not line.strip(): continue try: value = json.loads(line, parse_constant=_reject_constant("benchmark ledger line %d" % number)) except ValueError: raise BenchmarkError("benchmark ledger line %d is not JSON" % number) out.append(validate_receipt(value, manifest)) return out def append_receipt(path, receipt, manifest): receipt = validate_receipt(receipt, manifest) directory = os.path.dirname(os.path.abspath(path)) os.makedirs(directory, mode=0o700, exist_ok=True) lock_path = path + ".lock" lock_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600) os.fchmod(lock_fd, 0o600) with os.fdopen(lock_fd, "a+", encoding="utf-8") as lock: fcntl.flock(lock.fileno(), fcntl.LOCK_EX) existing = load_ledger(path, manifest) for item in existing: if item["receiptId"] == receipt["receiptId"]: return {"appended": False, "receiptId": receipt["receiptId"], "count": len(existing)} if (item["armId"], item["caseId"], item["repeat"]) == \ (receipt["armId"], receipt["caseId"], receipt["repeat"]): raise BenchmarkError("arm/case/repeat already has a different immutable receipt") if receipt["rerunOf"] is not None: prior = next((item for item in existing if item["receiptId"] == receipt["rerunOf"]), None) if prior is None or prior["armId"] != receipt["armId"] or prior["caseId"] != receipt["caseId"]: raise BenchmarkError("rerunOf must name an earlier receipt for the same arm and case") ledger_fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) os.fchmod(ledger_fd, 0o600) with os.fdopen(ledger_fd, "a", encoding="utf-8") as fh: fh.write(canonical(receipt) + "\n") fh.flush() os.fsync(fh.fileno()) return {"appended": True, "receiptId": receipt["receiptId"], "count": len(existing) + 1} def _atomic_json(path, value): directory = os.path.dirname(os.path.abspath(path)) os.makedirs(directory, mode=0o700, exist_ok=True) fd, temp = tempfile.mkstemp(dir=directory, prefix=".benchmark-", suffix=".tmp") try: os.fchmod(fd, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as fh: fh.write(json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True, allow_nan=False) + "\n") fh.flush() os.fsync(fh.fileno()) os.replace(temp, path) except BaseException: try: os.unlink(temp) except OSError: pass raise def main(argv=None): parser = argparse.ArgumentParser(description="Camus Slice G reviewer benchmark ledger") sub = parser.add_subparsers(dest="command", required=True) append = sub.add_parser("append", help="append one immutable benchmark receipt") append.add_argument("--campaign", required=True) append.add_argument("--ledger", required=True) append.add_argument("--receipt", default="-", help="receipt JSON file, or - for stdin") report = sub.add_parser("summarize", help="derive an admission report from the full ledger") report.add_argument("--campaign", required=True) report.add_argument("--ledger", required=True) report.add_argument("--out", default=None) report.add_argument("--json", action="store_true") args = parser.parse_args(argv) try: manifest = validate_manifest(_load_json(args.campaign, "campaign")) if args.command == "append": receipt = json.load(sys.stdin, parse_constant=_reject_constant("receipt")) \ if args.receipt == "-" else _load_json(args.receipt, "receipt") print(canonical(append_receipt(args.ledger, receipt, manifest))) return 0 result = summarize(manifest, load_ledger(args.ledger, manifest)) if args.out: _atomic_json(args.out, result) if args.json or not args.out: print(canonical(result)) else: print("wrote %s (%d receipts; %d candidate rows)" % ( args.out, result["receiptCount"], len(result["candidates"]))) return 0 except (BenchmarkError, OSError, ValueError) as exc: print("error: %s" % exc, file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main())