| |
| """Materialize one full prediction set from recent and triggered temporal calls.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from pathlib import Path |
| from statistics import fmean |
| from typing import Any |
|
|
| from experiments.evaluate_qwen3vl_recent4_full import atomic_write |
| from experiments.analyze_uncertainty_temporal_escalation import choice_margin |
|
|
|
|
| def parse_args() -> argparse.Namespace: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--recent", type=Path, required=True) |
| parser.add_argument("--temporal", type=Path, required=True) |
| parser.add_argument("--selection-audit", type=Path, required=True) |
| parser.add_argument("--threshold-audit", type=Path) |
| parser.add_argument( |
| "--temporal-model-identity-sha256", |
| help=( |
| "Previously sealed model identity to attach when the temporal result " |
| "was launched without embedding it; avoids rehashing the model." |
| ), |
| ) |
| parser.add_argument( |
| "--commit-threshold", |
| type=float, |
| help=( |
| "Optional fixed proposal-minus-current margin threshold. This is " |
| "required for an unlabeled fixed-budget selection and overrides " |
| "the commit threshold stored in an internal calibration." |
| ), |
| ) |
| parser.add_argument( |
| "--fallback-recent-on-temporal-error", |
| action="store_true", |
| help=( |
| "Fail closed to the aligned recent4 prediction when a preselected " |
| "temporal call has a recorded runtime error. The fallback is fixed " |
| "before scoring and never reads benchmark labels." |
| ), |
| ) |
| parser.add_argument("--output", type=Path, required=True) |
| return parser.parse_args() |
|
|
|
|
| def selected_view_name(policy: str, mode: str, *, escalated: bool) -> str: |
| if not escalated: |
| return "recent4" |
| if policy in {"causalbudget20", "causalbudget32r8"}: |
| suffix = "32_recent8" if policy == "causalbudget32r8" else "20" |
| mode_name = { |
| "native-video": "native_video", |
| "image-list": "image_list", |
| "hybrid-history-video-recent-images": ( |
| "hybrid_history_video_recent_images" |
| ), |
| "lowres-history-recent-images": "lowres_history_recent_images", |
| }[mode] |
| return f"causal_temporal{suffix}_{mode_name}" |
| if policy in {"causalhybrid32", "causalhybrid32r8"}: |
| recent_frames = 8 if policy == "causalhybrid32r8" else 4 |
| history_frames = 32 - recent_frames |
| if mode == "image-list": |
| return ( |
| f"causal_uniform_history{history_frames}_plus_recent" |
| f"{recent_frames}_image_list" |
| ) |
| return f"causal_hybrid_history{history_frames}_plus_recent{recent_frames}" |
| if policy == "prefix32": |
| return "causal_prefix32" |
| raise ValueError(f"unsupported second-view policy: {policy}") |
|
|
|
|
| def second_view_policy_name(policy: str, mode: str) -> str: |
| if policy in {"causalbudget20", "causalbudget32r8"}: |
| suffix = "32_recent8" if policy == "causalbudget32r8" else "20" |
| prefix = { |
| "native-video": "native_video", |
| "image-list": "image_list", |
| "hybrid-history-video-recent-images": ( |
| "hybrid_history_video_recent_images" |
| ), |
| "lowres-history-recent-images": "lowres_history_recent_images", |
| }[mode] |
| return f"{prefix}_causal_temporal{suffix}" |
| if policy in {"causalhybrid32", "causalhybrid32r8"}: |
| if mode == "hybrid-history-video-recent-images": |
| return ( |
| "hybrid_history_video_recent_images_causal_hybrid32r8" |
| if policy == "causalhybrid32r8" |
| else "hybrid_history_video_recent_images_causal_hybrid32" |
| ) |
| return ( |
| "image_list_uniform_history24_plus_recent8" |
| if policy == "causalhybrid32r8" |
| else "image_list_uniform_history28_plus_recent4" |
| ) |
| if policy == "prefix32": |
| return "image_list_causal_prefix32" |
| raise ValueError(f"unsupported second-view policy: {policy}") |
|
|
|
|
| def materialize( |
| recent: dict[str, Any], |
| temporal: dict[str, Any], |
| selection: dict[str, Any], |
| threshold: dict[str, Any] | None, |
| *, |
| commit_threshold_override: float | None = None, |
| fallback_recent_on_temporal_error: bool = False, |
| ) -> dict[str, Any]: |
| for field in ( |
| "benchmark", |
| "model_manifest_sha256", |
| "annotation_sha256", |
| "evaluation_protocol_sha256", |
| ): |
| if recent.get(field) != temporal.get(field): |
| raise ValueError(f"recent/temporal identity mismatch: {field}") |
| selection_format = selection.get("format") |
| fixed_quantile = None |
| frozen_threshold = None |
| if threshold is not None: |
| if ( |
| threshold.get("status") |
| != "source_disjoint_internal_threshold_calibration" |
| or threshold.get("eligible_as_frozen_benchmark_controller_config") |
| is not True |
| or threshold.get("model_manifest_sha256") |
| != recent.get("model_manifest_sha256") |
| ): |
| raise ValueError("threshold is not an eligible internal calibration") |
| frozen_threshold = float(threshold["policy"]["threshold"]) |
| commit_threshold = threshold["policy"].get("commit_threshold") |
| if selection_format not in (None, "streameva.internal_threshold_uid_selection.v0"): |
| raise ValueError("threshold calibration and selection mode diverge") |
| else: |
| if selection_format != "streameva.unlabeled_budget_uid_selection.v0": |
| raise ValueError("selection is not an unlabeled fixed-budget audit") |
| fixed_quantile = float(selection.get("quantile")) |
| if not 0.0 < fixed_quantile <= 1.0: |
| raise ValueError("fixed uncertainty quantile is invalid") |
| if selection.get("benchmark_identity_used_for_policy_parameters") is not False: |
| raise ValueError("fixed-budget acquisition used benchmark identity") |
| if selection.get("cohort_labels_read_for_selection") is not False: |
| raise ValueError("fixed-budget acquisition used cohort labels") |
| if commit_threshold_override is None: |
| raise ValueError("fixed-budget selection requires a commit threshold") |
| commit_threshold = commit_threshold_override |
| if commit_threshold_override is not None: |
| commit_threshold = commit_threshold_override |
| if commit_threshold is not None: |
| commit_threshold = float(commit_threshold) |
| if ( |
| selection.get("status") |
| != "frozen_before_benchmark_temporal_inference" |
| or selection.get("benchmark") != recent.get("benchmark") |
| or selection.get("model_manifest_sha256") |
| != recent.get("model_manifest_sha256") |
| or selection.get("score_or_gold_fields_read_for_selection") is not False |
| ): |
| raise ValueError("selection audit is not frozen or identity-aligned") |
| if threshold is not None and float(selection.get("threshold")) != frozen_threshold: |
| raise ValueError("selection threshold diverges from internal calibration") |
|
|
| recent_rows = {str(row["uid"]): row for row in recent.get("rows") or []} |
| temporal_rows = {str(row["uid"]): row for row in temporal.get("rows") or []} |
| decisions = {str(row["uid"]): row for row in selection.get("rows") or []} |
| if ( |
| len(recent_rows) != len(recent.get("rows") or []) |
| or len(temporal_rows) != len(temporal.get("rows") or []) |
| or len(decisions) != len(selection.get("rows") or []) |
| ): |
| raise ValueError("duplicate UID in input artifacts") |
| if set(decisions) != set(recent_rows): |
| raise ValueError("selection audit does not cover the recent result") |
| selected = {uid for uid, row in decisions.items() if bool(row["escalated"])} |
| temporal_input_count = len(temporal_rows) |
| if not selected.issubset(temporal_rows): |
| missing = sorted(selected - set(temporal_rows)) |
| raise ValueError( |
| f"temporal inference is missing {len(missing)} selected UIDs" |
| ) |
| temporal_rows = {uid: temporal_rows[uid] for uid in selected} |
| temporal_error_uids = { |
| uid for uid, row in temporal_rows.items() if bool(row.get("error")) |
| } |
| if temporal_error_uids and not fallback_recent_on_temporal_error: |
| raise ValueError( |
| f"{len(temporal_error_uids)} selected temporal rows contain errors" |
| ) |
| valid_temporal_rows = { |
| uid: row for uid, row in temporal_rows.items() if uid not in temporal_error_uids |
| } |
| if not valid_temporal_rows: |
| raise ValueError("no valid temporal row remains to establish the second-view policy") |
| second_view_policies = { |
| str((row.get("sampling") or {}).get("frame_policy") or "") |
| for row in valid_temporal_rows.values() |
| } |
| if second_view_policies not in ( |
| {"causalbudget20"}, |
| {"causalbudget32r8"}, |
| {"prefix32"}, |
| {"causalhybrid32"}, |
| {"causalhybrid32r8"}, |
| ): |
| raise ValueError( |
| "second-view rows must share causalbudget20, causalbudget32r8, prefix32, or " |
| "causalhybrid32*" |
| ) |
| second_view_policy = next(iter(second_view_policies)) |
| second_view_modes = { |
| str(row.get("visual_input_mode") or "") |
| for row in valid_temporal_rows.values() |
| } |
| if len(second_view_modes) != 1: |
| raise ValueError("second-view rows do not share one visual input mode") |
| second_view_mode = next(iter(second_view_modes)) |
| if second_view_mode not in { |
| "native-video", |
| "hybrid-history-video-recent-images", |
| "lowres-history-recent-images", |
| "image-list", |
| }: |
| raise ValueError("second-view visual input mode is missing or unsupported") |
|
|
| rows = [] |
| for recent_row in recent.get("rows") or []: |
| uid = str(recent_row["uid"]) |
| decision = decisions[uid] |
| escalated = bool(decision["escalated"]) |
| temporal_error_fallback = escalated and uid in temporal_error_uids |
| proposal = ( |
| recent_row |
| if temporal_error_fallback or not escalated |
| else temporal_rows[uid] |
| ) |
| sampling = proposal.get("sampling") or {} |
| if sampling.get("frame_timestamps") and max(sampling["frame_timestamps"]) > float( |
| proposal["query_time"] |
| ) + 1e-3: |
| raise ValueError(f"{uid}: selected inference crosses the causal frontier") |
| recent_frames = int(recent_row.get("frames_observed") or 4) |
| chosen_frames = int(proposal.get("frames_observed") or recent_frames) |
| probe_frames = 0 |
| extra_probe_frames = 0 |
| if escalated and not temporal_error_fallback: |
| if sampling.get("frame_policy") != second_view_policy: |
| raise ValueError(f"{uid}: second-view policy diverged") |
| if second_view_policy in {"causalbudget20", "causalbudget32r8"}: |
| candidate_indices = sampling.get("candidate_frame_indices") |
| if not isinstance(candidate_indices, list) or not candidate_indices: |
| raise ValueError(f"{uid}: temporal row lacks probe-frame audit") |
| probe_frames = len(candidate_indices) |
| extra_probe_frames = int(sampling.get("extra_frames_inspected", -1)) |
| if extra_probe_frames != max(0, probe_frames - chosen_frames): |
| raise ValueError(f"{uid}: temporal probe accounting diverged") |
| elif second_view_policy == "prefix32" and chosen_frames != 32: |
| raise ValueError(f"{uid}: prefix32 row does not expose 32 frames") |
| elif second_view_policy in { |
| "causalhybrid32", |
| "causalhybrid32r8", |
| } and not ( |
| 1 <= chosen_frames <= 32 |
| ): |
| raise ValueError(f"{uid}: causalhybrid32* frame budget is invalid") |
| recent_prediction = str(recent_row.get("prediction") or "") |
| proposal_prediction = str(proposal.get("prediction") or "") |
| confidence_gain = None |
| committed = escalated and not temporal_error_fallback |
| if escalated and not temporal_error_fallback and commit_threshold is not None: |
| confidence_gain = choice_margin(proposal) - choice_margin(recent_row) |
| committed = bool( |
| proposal_prediction != recent_prediction |
| and confidence_gain > commit_threshold |
| ) |
| final_prediction = proposal_prediction if committed else recent_prediction |
| rows.append( |
| { |
| **proposal, |
| "uid": uid, |
| "prediction": final_prediction, |
| "current_prediction": recent_prediction, |
| "proposal_prediction": proposal_prediction if escalated else None, |
| "agent_policy": { |
| "margin": float(decision["margin"]), |
| "threshold": frozen_threshold, |
| "uncertainty_quantile": fixed_quantile, |
| "uncertainty_rank": decision.get("uncertainty_rank"), |
| "escalated": escalated, |
| "temporal_error_fallback": temporal_error_fallback, |
| "temporal_error": ( |
| str(temporal_rows[uid].get("error")) |
| if temporal_error_fallback |
| else None |
| ), |
| "commit_threshold": commit_threshold, |
| "confidence_gain": confidence_gain, |
| "committed": committed, |
| "selected_view": ( |
| "recent4_temporal_error_fallback" |
| if temporal_error_fallback |
| else selected_view_name( |
| second_view_policy, second_view_mode, escalated=escalated |
| ) |
| ), |
| "question_task_group_or_benchmark_router": False, |
| }, |
| "unique_frames": chosen_frames, |
| "model_processed_frames": ( |
| recent_frames + chosen_frames |
| if escalated and not temporal_error_fallback |
| else recent_frames |
| ), |
| "low_resolution_probe_frames": probe_frames, |
| "extra_probe_frames_not_visible_to_vlm": extra_probe_frames, |
| "total_decoded_frame_instances": ( |
| recent_frames + probe_frames + chosen_frames |
| if escalated and not temporal_error_fallback |
| else recent_frames |
| ), |
| "media_calls": 2 if escalated else 1, |
| "num_turns": 2 if escalated else 1, |
| "latency_seconds": float(recent_row.get("latency_seconds") or 0.0) |
| + ( |
| float(temporal_rows[uid].get("latency_seconds") or 0.0) |
| if escalated |
| else 0.0 |
| ), |
| } |
| ) |
| if any(not row["prediction"] for row in rows): |
| raise ValueError("materialized agent contains an empty prediction") |
| escalated_count = len(selected) |
| return { |
| "format": "streameva.uncertainty_temporal_agent_eval.v0", |
| "status": ( |
| "internally_calibrated_router_free_agent" |
| if threshold is not None |
| else "fixed_budget_router_free_agent" |
| ), |
| "benchmark": recent.get("benchmark"), |
| "model": recent.get("model"), |
| "model_manifest_sha256": recent.get("model_manifest_sha256"), |
| "processor": recent.get("processor"), |
| "annotation": recent.get("annotation"), |
| "annotation_sha256": recent.get("annotation_sha256"), |
| "evaluation_protocol": recent.get("evaluation_protocol"), |
| "evaluation_protocol_sha256": recent.get("evaluation_protocol_sha256"), |
| "comparable_under_locked_recent4_protocol": False, |
| "policy": { |
| "first_view": "recent4", |
| "uncertainty": "first_step_full_vocabulary_top2_margin", |
| "threshold": frozen_threshold, |
| "uncertainty_quantile": fixed_quantile, |
| "commit_threshold": commit_threshold, |
| "second_view": second_view_policy_name( |
| second_view_policy, second_view_mode |
| ), |
| "commit_rule": ( |
| "changed_answer_and_confidence_gain_gt_global_threshold" |
| if commit_threshold is not None |
| else "legacy_always_accept_temporal_proposal" |
| ), |
| "question_task_group_or_benchmark_router": False, |
| "benchmark_identity_available_to_policy": False, |
| "gold_available_to_policy": False, |
| "temporal_error_policy": ( |
| "fail_closed_to_recent4" |
| if fallback_recent_on_temporal_error |
| else "reject_materialization" |
| ), |
| }, |
| "num_samples": len(rows), |
| "num_errors": 0, |
| "escalated": escalated_count, |
| "temporal_input_num_rows": temporal_input_count, |
| "temporal_unused_num_rows": temporal_input_count - escalated_count, |
| "temporal_error_fallback_count": len(temporal_error_uids), |
| "escalation_rate": escalated_count / len(rows), |
| "mean_model_calls": 1.0 + escalated_count / len(rows), |
| "mean_unique_frames": fmean(row["unique_frames"] for row in rows), |
| "mean_model_processed_frames": fmean( |
| row["model_processed_frames"] for row in rows |
| ), |
| "mean_low_resolution_probe_frames": fmean( |
| row["low_resolution_probe_frames"] for row in rows |
| ), |
| "mean_extra_probe_frames_not_visible_to_vlm": fmean( |
| row["extra_probe_frames_not_visible_to_vlm"] for row in rows |
| ), |
| "mean_total_decoded_frame_instances": fmean( |
| row["total_decoded_frame_instances"] for row in rows |
| ), |
| "rows": rows, |
| } |
|
|
|
|
| def main() -> None: |
| args = parse_args() |
| recent = json.loads(args.recent.read_text(encoding="utf-8")) |
| temporal = json.loads(args.temporal.read_text(encoding="utf-8")) |
| identity_override_applied = False |
| if args.temporal_model_identity_sha256 is not None: |
| identity = args.temporal_model_identity_sha256.strip().lower() |
| if len(identity) != 64 or any( |
| character not in "0123456789abcdef" for character in identity |
| ): |
| raise ValueError("temporal model identity must be a 64-character SHA-256") |
| embedded_identity = temporal.get("model_manifest_sha256") |
| if embedded_identity not in (None, "", identity): |
| raise ValueError("temporal model identity override conflicts with artifact") |
| if embedded_identity in (None, ""): |
| temporal["model_manifest_sha256"] = identity |
| identity_override_applied = True |
| selection = json.loads(args.selection_audit.read_text(encoding="utf-8")) |
| threshold = ( |
| json.loads(args.threshold_audit.read_text(encoding="utf-8")) |
| if args.threshold_audit is not None |
| else None |
| ) |
| output = materialize( |
| recent, |
| temporal, |
| selection, |
| threshold, |
| commit_threshold_override=args.commit_threshold, |
| fallback_recent_on_temporal_error=args.fallback_recent_on_temporal_error, |
| ) |
| output["artifacts"] = { |
| "recent": {"path": str(args.recent.resolve())}, |
| "temporal": {"path": str(args.temporal.resolve())}, |
| "selection_audit": {"path": str(args.selection_audit.resolve())}, |
| } |
| if args.threshold_audit is not None: |
| output["artifacts"]["threshold_audit"] = { |
| "path": str(args.threshold_audit.resolve()), |
| } |
| output["artifact_checksums_skipped"] = True |
| output["artifact_checksum_skip_reason"] = ( |
| "native artifacts succeeded and identities are embedded in their manifests" |
| ) |
| output["temporal_model_identity_override_applied"] = identity_override_applied |
| atomic_write(args.output, output) |
| print( |
| json.dumps( |
| { |
| "benchmark": output["benchmark"], |
| "num_samples": output["num_samples"], |
| "escalated": output["escalated"], |
| "mean_model_calls": output["mean_model_calls"], |
| }, |
| sort_keys=True, |
| ) |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
