#!/usr/bin/env python3 """Select temporal-view calls using a threshold or fixed uncertainty budget.""" from __future__ import annotations import argparse import json import math from pathlib import Path from typing import Any from experiments.analyze_uncertainty_temporal_escalation import choice_margin from experiments.evaluate_qwen3vl_recent4_full import atomic_write, sha256_file def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--recent", type=Path, required=True) policy = parser.add_mutually_exclusive_group(required=True) policy.add_argument("--threshold", type=Path) policy.add_argument( "--quantile", type=float, help="Acquire the lowest-margin fraction without reading cohort labels.", ) parser.add_argument("--samples", type=Path, required=True) parser.add_argument("--output-uids", type=Path, required=True) parser.add_argument("--output-samples", type=Path, required=True) parser.add_argument("--output-audit", type=Path, required=True) return parser.parse_args() def load_frozen_threshold(path: Path, model_manifest_sha256: str | None) -> float: value = json.loads(path.read_text(encoding="utf-8")) if ( value.get("status") != "source_disjoint_internal_threshold_calibration" or value.get("eligible_as_frozen_benchmark_controller_config") is not True or value.get("benchmark") != "internal_policy_holdout" or value.get("model_manifest_sha256") != model_manifest_sha256 ): raise ValueError("threshold is not an eligible source-disjoint calibration") threshold = float(value["policy"]["threshold"]) if not 0.0 <= threshold <= 1.0: raise ValueError("invalid calibrated threshold") return threshold def select_rows( recent: dict[str, Any], *, threshold: float ) -> tuple[list[str], list[dict[str, Any]]]: selected = [] audit_rows = [] for row in recent.get("rows") or []: uid = str(row.get("uid") or "") if not uid: raise ValueError("recent result has a row without uid") margin = choice_margin(row) escalated = margin <= threshold if escalated: selected.append(uid) audit_rows.append( { "uid": uid, "margin": margin, "threshold": threshold, "escalated": escalated, } ) if len(audit_rows) != int(recent.get("num_samples", -1)): raise ValueError("recent result sample count diverged") if int(recent.get("num_errors", -1)) != 0: raise ValueError("recent result contains inference errors") if len({row["uid"] for row in audit_rows}) != len(audit_rows): raise ValueError("recent result contains duplicate uids") return selected, audit_rows def select_rows_by_quantile( recent: dict[str, Any], *, quantile: float ) -> tuple[list[str], list[dict[str, Any]], float]: """Select an exact bottom-margin budget with deterministic UID tie-breaking.""" if not 0.0 < quantile <= 1.0: raise ValueError("quantile must be in (0, 1]") rows = list(recent.get("rows") or []) if len(rows) != int(recent.get("num_samples", -1)): raise ValueError("recent result sample count diverged") if int(recent.get("num_errors", -1)) != 0: raise ValueError("recent result contains inference errors") values = [] for row in rows: uid = str(row.get("uid") or "") if not uid: raise ValueError("recent result has a row without uid") values.append((choice_margin(row), uid)) if len({uid for _, uid in values}) != len(values): raise ValueError("recent result contains duplicate uids") budget = max(1, math.ceil(len(values) * quantile)) ranked = sorted(values, key=lambda item: (item[0], item[1])) selected_set = {uid for _, uid in ranked[:budget]} cutoff = float(ranked[budget - 1][0]) selected = [str(row["uid"]) for row in rows if str(row["uid"]) in selected_set] rank_by_uid = {uid: rank for rank, (_, uid) in enumerate(ranked, 1)} audit_rows = [ { "uid": str(row["uid"]), "margin": choice_margin(row), "uncertainty_rank": rank_by_uid[str(row["uid"])], "quantile": quantile, "escalated": str(row["uid"]) in selected_set, } for row in rows ] return selected, audit_rows, cutoff def main() -> None: args = parse_args() recent = json.loads(args.recent.read_text(encoding="utf-8")) threshold = None quantile = None cutoff = None if args.threshold is not None: threshold = load_frozen_threshold( args.threshold, recent.get("model_manifest_sha256") ) selected, audit_rows = select_rows(recent, threshold=threshold) else: quantile = float(args.quantile) selected, audit_rows, cutoff = select_rows_by_quantile( recent, quantile=quantile ) selected_set = set(selected) sample_rows = [ json.loads(line) for line in args.samples.read_text(encoding="utf-8").splitlines() if line.strip() ] ordered_uids = [str(row.get("uid") or "") for row in sample_rows] recent_uids = [str(row["uid"]) for row in recent["rows"]] if ordered_uids != recent_uids: raise ValueError("recent result does not follow the frozen sample order") selected_samples = [row for row in sample_rows if str(row["uid"]) in selected_set] if [str(row["uid"]) for row in selected_samples] != selected: raise ValueError("selected UID ordering diverged") args.output_uids.parent.mkdir(parents=True, exist_ok=True) args.output_uids.write_text("\n".join(selected) + "\n", encoding="utf-8") args.output_samples.write_text( "".join( json.dumps(row, sort_keys=True, ensure_ascii=False) + "\n" for row in selected_samples ), encoding="utf-8", ) atomic_write( args.output_audit, { "format": ( "streameva.internal_threshold_uid_selection.v0" if threshold is not None else "streameva.unlabeled_budget_uid_selection.v0" ), "status": "frozen_before_benchmark_temporal_inference", "benchmark": recent.get("benchmark"), "model_manifest_sha256": recent.get("model_manifest_sha256"), "threshold": threshold, "quantile": quantile, "empirical_margin_cutoff": cutoff, "threshold_audit": ( str(args.threshold.resolve()) if args.threshold is not None else None ), "threshold_audit_sha256": ( sha256_file(args.threshold) if args.threshold is not None else None ), "recent_result": str(args.recent.resolve()), "recent_result_sha256": sha256_file(args.recent), "samples_sha256": sha256_file(args.samples), "question_task_group_or_benchmark_router": False, "score_or_gold_fields_read_for_selection": False, "benchmark_identity_used_for_policy_parameters": False, "cohort_labels_read_for_selection": False, "total": len(audit_rows), "selected": len(selected), "selection_rate": len(selected) / len(audit_rows), "rows": audit_rows, }, ) print(json.dumps({"total": len(audit_rows), "selected": len(selected)})) if __name__ == "__main__": main()