"""Evaluate a frozen candidate; fit probability and abstention on disjoint calibration groups.""" import argparse import ast import hashlib import json import math import shutil import time from datetime import datetime, timezone from pathlib import Path import numpy as np import torch from train_foundation_head import write from veyra.calibrate import fit_temperature from veyra.candidates import candidates_for from veyra.constants import QUESTION_TYPES from veyra.data import read_records from veyra.decision_metrics import observation, report, summarize from veyra.features import FeatureSample from veyra.option_model import OptionModel from veyra.probability import Calibration, typed_answer def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def structural_annotations(): """Reporting metadata only; never supplied to the model or used to select an answer.""" snapshot = Path("reports/workflow-v12/source-snapshots/build_workflow_v12.py.txt") expected = json.loads(Path("data/workflow-v12/protocol.json").read_text(encoding="utf-8"))[ "generator_sha256" ] if digest(snapshot) != expected: raise ValueError("original rule structure snapshot changed") tree = ast.parse(snapshot.read_text(encoding="utf-8")) families = next( ast.literal_eval(node.value) for node in tree.body if isinstance(node, ast.Assign) and any(isinstance(target, ast.Name) and target.id == "FAMILIES" for target in node.targets) ) def depth(expression): if isinstance(expression, int): return 0 terms = expression[2:] if expression[0] == "count" else expression[1:] return 1 + max(depth(term) for term in terms) return { name: {"rule_count": len(rules), "boolean_operator_depth": max(depth(r[0]) for r in rules)} for name, _, _, rules, _ in families }, snapshot def expected_error(record, name, winner): if "teacher_distribution" in record.tags: gold = next((t[9:] for t in record.tags if t.startswith("gold_key:")), None) if gold is not None: return float(winner != gold) return 1 - record.targets[name][winner] def augment(row, record, name): critical = next((t[13:] for t in record.tags if t.startswith("critical_key:")), "") row["condition"] = next((t[10:] for t in record.tags if t.startswith("condition:")), "none") row["expected_error"] = expected_error(record, name, row["prediction"]) row["expected_cost"] = row["expected_error"] if critical and row["prediction"] != critical: row["expected_cost"] += 4 * record.targets[name].get(critical, 0.0) row["critical_key"] = critical return row def risk_report(rows): ordered = sorted( rows, key=lambda r: (-r["confidence"], hashlib.sha256(r["id"].encode()).hexdigest()) ) points = {} for coverage in (0.5, 0.8, 0.9, 1.0): accepted = ordered[: max(1, math.floor(len(ordered) * coverage))] points[str(coverage)] = { "accepted": len(accepted), "coverage": len(accepted) / len(rows), "expected_error": float(np.mean([r["expected_error"] for r in accepted])), "expected_cost": float(np.mean([r["expected_cost"] for r in accepted])), } accepted = [r for r in rows if not r["abstained"]] expected_ece = 0.0 for bin_index in range(15): selected = [r for r in rows if min(14, int(r["confidence"] * 15)) == bin_index] if selected: expected_ece += ( len(selected) / len(rows) * abs( np.mean([r["confidence"] for r in selected]) - np.mean([1 - r["expected_error"] for r in selected]) ) ) return { "conditional_expected_ece_15_bins": float(expected_ece), "matched_coverage": points, "actual_policy": { "accepted": len(accepted), "coverage": len(accepted) / len(rows), "abstention_rate": 1 - len(accepted) / len(rows), "expected_error": float(np.mean([r["expected_error"] for r in accepted])) if accepted else None, "expected_cost": float(np.mean([r["expected_cost"] for r in accepted])) if accepted else None, }, } def fit_policy(indices, logits, descriptors, temperature, max_error, minimum_coverage, device): observations = [] for i in indices: record, name, question = descriptors[i] # Match the runtime's float32 device kernel at the exact threshold boundary. probabilities = (logits[i].to(device).float() / temperature).softmax(-1) candidates = candidates_for(question) winner = candidates[int(probabilities.argmax())].key observations.append((float(probabilities.max()), expected_error(record, name, winner))) minimum = max(20, math.ceil(len(observations) * minimum_coverage)) for threshold in sorted({p for p, _ in observations}): accepted = [e for p, e in observations if p >= threshold] if len(accepted) >= minimum and sum(accepted) / len(accepted) <= max_error: return { "threshold": threshold, "accepted": len(accepted), "questions": len(observations), "coverage": len(accepted) / len(observations), "expected_error": sum(accepted) / len(accepted), "always_abstain": False, "passed": True, } return { "threshold": 1.0, "accepted": 0, "questions": len(observations), "coverage": 0.0, "expected_error": None, "always_abstain": True, "passed": False, } @torch.inference_mode() def main(): parser = argparse.ArgumentParser() parser.add_argument("--checkpoint", type=Path, required=True) parser.add_argument("--records", type=Path, default=Path("data/workflow-v12/records.jsonl")) parser.add_argument("--split", choices=["dev", "calibration", "test"], required=True) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--calibrated-output", type=Path) parser.add_argument("--policy-design", type=Path) parser.add_argument("--selection", type=Path) parser.add_argument("--final-freeze", type=Path) parser.add_argument("--baseline", action="store_true") args = parser.parse_args() if args.output.exists(): raise FileExistsError("evaluation output is immutable") if args.calibrated_output and args.split != "calibration": raise ValueError("only calibration data may fit probabilities or abstention") if args.calibrated_output and args.calibrated_output.exists(): raise FileExistsError("calibrated checkpoint is immutable") config_path = Path("configs/workflow-release-v12.json") config = json.loads(config_path.read_text(encoding="utf-8")) policy_design = None if args.policy_design: if not args.calibrated_output or not args.selection: raise ValueError("a policy design requires calibrated output and frozen selection") policy_design = json.loads(args.policy_design.read_text(encoding="utf-8")) selected_policy = json.loads(args.selection.read_text(encoding="utf-8")) fit_design = Path(policy_design["fitting_data_design"]) fit_data = json.loads(fit_design.read_text(encoding="utf-8")) method = policy_design["method"] if not ( policy_design["protocol"] == "workflow-depth-conservative-policy-v12" and policy_design["release_protocol_sha256"] == digest(config_path) and selected_policy["study_protocol_sha256"] == digest(args.policy_design) and policy_design["fitting_data_design_sha256"] == digest(fit_design) and args.records == Path(fit_data["output"]) / "records.jsonl" and method["name"] == "conservative_empirical_policy_with_independent_validation" and 0 < method["policy_fit_expected_error_max"] < config["gates"]["uncertainty"]["calibration_expected_error_max"] and method["policy_fit_minimum_coverage"] == config["gates"]["uncertainty"]["calibration_minimum_coverage"] ): raise ValueError("undeclared conservative policy-fitting configuration") weights_hash = digest(args.checkpoint / "head.safetensors") manifest_hash = digest(args.checkpoint / "manifest.json") if args.baseline: if ( weights_hash != config["baseline_weights_sha256"] or manifest_hash != config["baseline_manifest_sha256"] ): raise ValueError("baseline must be the unmodified frozen release") elif args.split in {"calibration", "test"}: if not args.selection: raise ValueError("a frozen development selection is required") selection = json.loads(args.selection.read_text(encoding="utf-8")) if ( args.calibrated_output and "workflow_depth_policy_v12" in selection.get("release_extensions", {}) and args.policy_design is None ): raise ValueError("conservative policy fitting requires its declared policy design") if selection.get("eligible") is not True: raise ValueError("selected candidate did not pass development requirements") if selection["weights_sha256"] != weights_hash or selection[ "release_protocol_sha256" ] != digest(config_path): raise ValueError("candidate does not match the frozen selection") if args.split == "test": if not args.final_freeze: raise ValueError("final evaluation needs a prewritten final-freeze record") freeze = json.loads(args.final_freeze.read_text(encoding="utf-8")) if freeze["records_sha256"] != digest(args.records) or freeze[ "release_protocol_sha256" ] != digest(config_path): raise ValueError("final data or release protocol mismatch") if not args.baseline and ( freeze["weights_sha256"] != weights_hash or freeze["manifest_sha256"] != manifest_hash ): raise ValueError("final model changed after freezing") records = [r for r in read_records(args.records) if r.split == args.split] if not records or any(len(r.request.questions) != 1 for r in records): raise ValueError("require single-question observation views") annotations, structure_source = structural_annotations() protocol = { "started_at_utc": datetime.now(timezone.utc).isoformat(), "arguments": vars(args), "weights_sha256": weights_hash, "manifest_sha256": manifest_hash, "records_sha256": digest(args.records), "release_protocol_sha256": digest(config_path), "source_sha256": digest(Path(__file__)), "rule_structure_source_sha256": digest(structure_source), "boolean_operator_depth_definition": ( "Maximum nested Boolean operators in any supplied rule; atoms have depth zero. " "Rule count is separate. Legacy rule_depth tags were rule counts and are not used." ), "token_length_bins": [256, 512, 1024, 2048], "selection_sha256": digest(args.selection) if args.selection else None, "policy_design_sha256": digest(args.policy_design) if args.policy_design else None, "final_freeze_sha256": digest(args.final_freeze) if args.final_freeze else None, "dependency_source_sha256": { str(path): digest(path) for path in ( Path("src/veyra/decision_metrics.py"), Path("src/veyra/probability.py"), Path("src/veyra/option_model.py"), Path("src/veyra/calibrate.py"), ) }, "merged_bf16_deployment": True, "probability_serialization": "typed_answer float32 softmax on the model device", "subset_sha256": hashlib.sha256( "".join(r.model_dump_json() + "\n" for r in records).encode() ).hexdigest(), "teacher_risk_interpretation": ( "Teacher consensus hard labels for policy errors; " "exact conditional distributions for constructed uncertain cases" ), } write(args.output / "protocol.json", protocol) model = OptionModel.load(args.checkpoint, local_files_only=True, merge=True) torch.set_num_threads(4) samples, logits, descriptors, token_counts = [], [], [], [] started = time.perf_counter() for index, record in enumerate(records, start=1): outputs, tokens = model(record.request, args.records.parent) for name, values in outputs.items(): question = record.request.questions[name] candidates = candidates_for(question) samples.append( FeatureSample( features=torch.zeros(len(candidates), 1), context=torch.zeros(1), levels=torch.tensor([c.level for c in candidates]), targets=torch.tensor([record.targets[name][c.key] for c in candidates]), type_id=QUESTION_TYPES.index(question.type), record_id=record.id, group_id=record.group_id, split=record.split, family=record.family, language=record.language, question_id=name, tags=record.tags, ) ) logits.append(values.cpu().float()) descriptors.append((record, name, question)) token_counts.append(tokens) if index % 200 == 0: progress = { "evaluated": index, "total": len(records), "split": args.split, "elapsed_seconds": time.perf_counter() - started, } write(args.output / "progress.json", progress) print(json.dumps(progress), flush=True) calibration, fitting = model.calibration, None if args.calibrated_output: temperatures, thresholds, always, fitting = [], [], [], {} gate = config["gates"]["uncertainty"] fit_error_max = ( policy_design["method"]["policy_fit_expected_error_max"] if policy_design else gate["calibration_expected_error_max"] ) for type_id, kind in enumerate(QUESTION_TYPES): fitting_indices, policy_indices = [], [] for i, sample in enumerate(samples): if sample.type_id == type_id: bucket = int(hashlib.sha256(sample.group_id.encode()).hexdigest()[:8], 16) % 2 (fitting_indices if bucket == 0 else policy_indices).append(i) if not fitting_indices or not policy_indices: raise ValueError("insufficient independent calibration groups") temperature = max( 1.0, fit_temperature( [samples[i] for i in fitting_indices], [logits[i] for i in fitting_indices] ), ) policy = fit_policy( policy_indices, logits, descriptors, temperature, fit_error_max, gate["calibration_minimum_coverage"], model.encoder.device, ) temperatures.append(temperature) thresholds.append(policy["threshold"]) if policy["always_abstain"]: always.append(kind) fitting[kind] = { **policy, "policy_fit_expected_error_max": fit_error_max, "temperature": temperature, "temperature_questions": len(fitting_indices), "temperature_group_ids": sorted({samples[i].group_id for i in fitting_indices}), "policy_group_ids": sorted({samples[i].group_id for i in policy_indices}), } calibration = Calibration( tuple(temperatures), tuple(thresholds), tuple(QUESTION_TYPES), tuple(QUESTION_TYPES), tuple(always), ) rows = [] for (record, name, question), values, tokens in zip( descriptors, logits, token_counts, strict=True ): answer = typed_answer(question, values.to(model.encoder.device), calibration) row = augment( observation(record, name, answer["probabilities"], abstained=answer["abstained"]), record, name, ) row["confidence"] = answer["confidence"] row["input_tokens"] = tokens row["token_length_bin"] = next( str(limit) for limit in (256, 512, 1024, 2048) if tokens <= limit ) row.update( annotations.get( record.family, { "rule_count": "unannotated", "boolean_operator_depth": "unannotated", }, ) ) rows.append(row) serialized_policy_check = None if args.calibrated_output: serialized_policy_check = {} for kind, fit in fitting.items(): policy_groups = set(fit["policy_group_ids"]) selected = [r for r in rows if r["type"] == kind and r["group"] in policy_groups] accepted = [r for r in selected if not r["abstained"]] error = sum(r["expected_error"] for r in accepted) / len(accepted) if accepted else None matches = ( len(selected) == fit["questions"] and len(accepted) == fit["accepted"] and ( error == fit["expected_error"] or ( error is not None and fit["expected_error"] is not None and abs(error - fit["expected_error"]) <= 1e-12 ) ) ) serialized_policy_check[kind] = { "matches_fitted_policy": matches, "questions": len(selected), "accepted": len(accepted), "expected_error": error, } if not matches: raise ValueError( "serialized runtime policy differs from the fitted policy: " + kind ) args.calibrated_output.mkdir(parents=True) shutil.copyfile( args.checkpoint / "head.safetensors", args.calibrated_output / "head.safetensors" ) manifest = json.loads((args.checkpoint / "manifest.json").read_text(encoding="utf-8")) manifest["calibration"] = calibration.to_dict() manifest["training"]["intermediate"] = not all(x["passed"] for x in fitting.values()) if args.selection and selection.get("release_extensions"): manifest["training"]["release_extensions"] = selection["release_extensions"] manifest["training"]["calibration"] = { "split": "calibration", "group_disjoint": True, "protocol": protocol, "fits": fitting, "serialized_policy_check": serialized_policy_check, "all_policy_requirements_passed": all(x["passed"] for x in fitting.values()), } write(args.calibrated_output / "manifest.json", manifest) path = args.output / "predictions.jsonl" path.write_text( "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows), encoding="utf-8", newline="\n", ) result = { "protocol": protocol, **report(rows, bootstrap=args.split == "test"), "risk": risk_report(rows), "calibration": calibration.to_dict(), "fitting": fitting, "serialized_policy_check": serialized_policy_check, "calibrated_manifest_sha256": digest(args.calibrated_output / "manifest.json") if args.calibrated_output else None, "predictions_sha256": digest(path), "elapsed_seconds": time.perf_counter() - started, } for field in ("domain", "family", "condition", "token_length_bin", "boolean_operator_depth"): result["detailed_by_" + field] = { value: { **summarize([r for r in rows if r[field] == value], bootstrap=args.split == "test"), **risk_report([r for r in rows if r[field] == value]), } for value in sorted({r[field] for r in rows}, key=str) } write(args.output / "evaluation.json", result) print(json.dumps({"overall": result["overall"], "risk": result["risk"]}), flush=True) if __name__ == "__main__": main()