| |
| """Materialize a task-blind confidence-gated third-view temporal agent. |
| |
| The third view is acquired on the same answer-disagreement cohort as the direct |
| third-view agent. Its proposal is committed only when its first-token margin |
| improves over the margin of the view that supplied the current answer by more |
| than one source-disjoint, globally frozen threshold. Runtime policy inputs do |
| not include task, benchmark identity, or gold labels. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import copy |
| import json |
| from pathlib import Path |
| from statistics import fmean |
| from typing import Any |
|
|
| from experiments.analyze_uncertainty_temporal_escalation import choice_margin |
| from experiments.materialize_third_view_disagreement_agent import ( |
| IDENTITY_FIELDS, |
| _frame_indices, |
| _index, |
| ) |
|
|
|
|
| def materialize( |
| recent: dict[str, Any], |
| event: dict[str, Any], |
| base_agent: dict[str, Any], |
| third: dict[str, Any], |
| threshold: dict[str, Any], |
| ) -> dict[str, Any]: |
| for field in IDENTITY_FIELDS: |
| values = {value.get(field) for value in (recent, event, base_agent, third)} |
| if len(values) != 1: |
| raise ValueError(f"view identity mismatch: {field}") |
| 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 identity-aligned internal calibration") |
| commit_threshold = float(threshold["policy"]["commit_threshold"]) |
|
|
| recent_rows, event_rows, base_rows, third_rows = map( |
| _index, (recent, event, base_agent, third) |
| ) |
| if set(base_rows) != set(recent_rows) or not set(event_rows) <= set(recent_rows): |
| raise ValueError("base/recent/event UID coverage diverged") |
| disagreements = { |
| uid |
| for uid, row in event_rows.items() |
| if str(row.get("prediction") or "") |
| != str(recent_rows[uid].get("prediction") or "") |
| } |
| if not disagreements <= set(third_rows): |
| raise ValueError("third-view coverage must include the event disagreement cohort") |
|
|
| output_rows: list[dict[str, Any]] = [] |
| committed_count = 0 |
| for recent_row in recent.get("rows") or []: |
| uid = str(recent_row["uid"]) |
| base = copy.deepcopy(base_rows[uid]) |
| if uid not in disagreements: |
| base["third_view_policy"] = { |
| "triggered": False, |
| "reason": "no_acquired_event_disagreement", |
| "question_task_group_or_benchmark_router": False, |
| } |
| output_rows.append(base) |
| continue |
|
|
| event_row = event_rows[uid] |
| third_row = third_rows[uid] |
| if event_row.get("error") or third_row.get("error"): |
| raise ValueError(f"{uid}: acquired view contains an error") |
| if not bool((base.get("agent_policy") or {}).get("escalated")): |
| raise ValueError(f"{uid}: third view was triggered outside acquired cohort") |
| timestamps = (third_row.get("sampling") or {}).get("frame_timestamps") or [] |
| if timestamps and max(map(float, timestamps)) > float(third_row["query_time"]) + 1e-3: |
| raise ValueError(f"{uid}: third view crosses the causal frontier") |
|
|
| recent_prediction = str(recent_row.get("prediction") or "") |
| event_prediction = str(event_row.get("prediction") or "") |
| current_prediction = str(base.get("prediction") or "") |
| third_prediction = str(third_row.get("prediction") or "") |
| if not all((recent_prediction, event_prediction, current_prediction, third_prediction)): |
| raise ValueError(f"{uid}: an agent proposal is empty") |
| if recent_prediction == event_prediction: |
| raise ValueError(f"{uid}: third view lacks a disagreement trigger") |
| if current_prediction == recent_prediction: |
| current_source = recent_row |
| current_view = "recent4" |
| elif current_prediction == event_prediction: |
| current_source = event_row |
| current_view = "causalbudget20" |
| else: |
| raise ValueError(f"{uid}: current answer is not supplied by recent or event view") |
|
|
| confidence_gain = choice_margin(third_row) - choice_margin(current_source) |
| committed = bool( |
| third_prediction != current_prediction |
| and confidence_gain > commit_threshold |
| ) |
| if committed: |
| committed_count += 1 |
| base["prediction"] = third_prediction |
| base["raw_response"] = third_row.get("raw_response") |
| unique_indices = _frame_indices(event_row) | _frame_indices(third_row) |
| base.update( |
| { |
| "current_prediction": current_prediction, |
| "third_view_prediction": third_prediction, |
| "third_view_policy": { |
| "triggered": True, |
| "trigger": "acquired_event_prediction_differs_from_recent_prediction", |
| "commit": "global_source_disjoint_confidence_gain", |
| "current_view": current_view, |
| "current_margin": choice_margin(current_source), |
| "third_margin": choice_margin(third_row), |
| "confidence_gain": confidence_gain, |
| "commit_threshold": commit_threshold, |
| "committed": committed, |
| "question_task_group_or_benchmark_router": False, |
| "gold_available_to_policy": False, |
| }, |
| "media_calls": int(base.get("media_calls") or 2) + 1, |
| "num_turns": int(base.get("num_turns") or 2) + 1, |
| "model_processed_frames": int(base.get("model_processed_frames") or 24) |
| + int(third_row.get("frames_observed") or 32), |
| "unique_frames": len(unique_indices) if unique_indices else 32, |
| "latency_seconds": float(base.get("latency_seconds") or 0.0) |
| + float(third_row.get("latency_seconds") or 0.0), |
| "visual_tokens": int(recent_row.get("visual_tokens") or 0) |
| + int(event_row.get("visual_tokens") or 0) |
| + int(third_row.get("visual_tokens") or 0), |
| } |
| ) |
| output_rows.append(base) |
|
|
| if any(not str(row.get("prediction") or "") for row in output_rows): |
| raise ValueError("materialized agent contains an empty prediction") |
| triggered = len(disagreements) |
| return { |
| **{ |
| key: base_agent.get(key) |
| for key in ( |
| "annotation", |
| "annotation_sha256", |
| "benchmark", |
| "evaluation_protocol", |
| "evaluation_protocol_sha256", |
| "model", |
| "model_manifest_sha256", |
| "processor", |
| ) |
| }, |
| "format": "streameva.confidence_gated_third_view_agent.v0", |
| "status": "benchmark_selected_router_free_development_agent", |
| "classification": "benchmark_selected_development_not_confirmatory", |
| "comparable_under_locked_recent4_protocol": False, |
| "policy": { |
| "first_view": "recent4", |
| "second_view": "frozen_uncertainty_triggered_causalbudget20", |
| "third_view": "causalbudget32r8_on_acquired_answer_disagreement", |
| "third_view_commit": "global_source_disjoint_confidence_gain", |
| "commit_threshold": commit_threshold, |
| "question_task_group_or_benchmark_router": False, |
| "benchmark_identity_available_to_policy": False, |
| "gold_available_to_policy": False, |
| }, |
| "num_samples": len(output_rows), |
| "num_errors": 0, |
| "third_view_triggered": triggered, |
| "third_view_input_rows": len(third_rows), |
| "third_view_unused_precomputed_rows": len(set(third_rows) - disagreements), |
| "third_view_committed": committed_count, |
| "third_view_trigger_rate": triggered / len(output_rows), |
| "third_view_commit_rate": committed_count / triggered if triggered else 0.0, |
| "mean_model_calls": fmean(int(row.get("media_calls") or 1) for row in output_rows), |
| "mean_unique_frames": fmean(int(row.get("unique_frames") or 4) for row in output_rows), |
| "mean_model_processed_frames": fmean( |
| int(row.get("model_processed_frames") or 4) for row in output_rows |
| ), |
| "rows": output_rows, |
| } |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--recent", type=Path, required=True) |
| parser.add_argument("--event", type=Path, required=True) |
| parser.add_argument("--base-agent", type=Path, required=True) |
| parser.add_argument("--third-view", type=Path, required=True) |
| parser.add_argument("--threshold-audit", type=Path, required=True) |
| parser.add_argument("--third-view-model-identity-sha256") |
| parser.add_argument("--output", type=Path, required=True) |
| args = parser.parse_args() |
| if args.output.exists(): |
| raise FileExistsError(args.output) |
| values = [ |
| json.loads(path.read_text(encoding="utf-8")) |
| for path in ( |
| args.recent, |
| args.event, |
| args.base_agent, |
| args.third_view, |
| args.threshold_audit, |
| ) |
| ] |
| identity_override = None |
| if args.third_view_model_identity_sha256: |
| if len(args.third_view_model_identity_sha256) != 64 or any( |
| value not in "0123456789abcdef" |
| for value in args.third_view_model_identity_sha256 |
| ): |
| raise ValueError("third-view model identity is malformed") |
| third = values[3] |
| if third.get("model_manifest_sha256") not in { |
| None, |
| args.third_view_model_identity_sha256, |
| }: |
| raise ValueError("third-view model identity override diverges") |
| if third.get("model_manifest_sha256") is None: |
| third["model_manifest_sha256"] = args.third_view_model_identity_sha256 |
| identity_override = { |
| "field": "model_manifest_sha256", |
| "source": "previously_sealed_checkpoint_identity", |
| "value": args.third_view_model_identity_sha256, |
| } |
| result = materialize(*values) |
| result["third_view_identity_override"] = identity_override |
| result["artifacts"] = { |
| "recent": str(args.recent.resolve()), |
| "event": str(args.event.resolve()), |
| "base_agent": str(args.base_agent.resolve()), |
| "third_view": str(args.third_view.resolve()), |
| "threshold_audit": str(args.threshold_audit.resolve()), |
| } |
| args.output.parent.mkdir(parents=True, exist_ok=True) |
| args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n") |
| print( |
| json.dumps( |
| { |
| "output": str(args.output), |
| "num_samples": result["num_samples"], |
| "triggered": result["third_view_triggered"], |
| "committed": result["third_view_committed"], |
| }, |
| sort_keys=True, |
| ) |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
