"""FailureDetectionAgent — layer 1. Scans candidate-vs-ground-truth predictions on the synthetic populations (bevfusion scene campaign + vectorized rate population) for: * classification flips (pedestrian->construction_cone, pedestrian->vehicle, bicycle->vehicle, ..., anything->background), * localization regressions (mean center error delta), * detection regressions (per-class recall delta), * confidence-calibration regressions (mean confidence on correct predictions), * temporal regressions (per-track prediction churn). Every emitted FailureEvent carries its deterministic detection basis (counts, denominators, metric deltas) — the agent flags, it does not judge. """ from __future__ import annotations from collections import defaultdict from typing import Any, Dict, List, Tuple import numpy as np from sensorflow.agentic import data as data_mod from sensorflow.agentic.agents.base import BaseAgent, compact, no_escalation from sensorflow.agentic.models import (AgentEscalation, DetectionBasis, FailureEvent, FailureInstance, new_id) SAFETY_CRITICAL = {"pedestrian", "cyclist", "motorcycle"} LOCALIZATION_DELTA_THRESHOLD_M = 0.02 CONFIDENCE_DELTA_THRESHOLD = 0.03 CHURN_DELTA_THRESHOLD = 0.02 class FailureDetectionAgent(BaseAgent): name = "failure_detection" version = "1.0.0" epistemic_status = "DERIVED" def detect(self, seed: int = data_mod.DEFAULT_SEED) -> Tuple[List[FailureEvent], Dict]: """Full campaign scan. Returns (events, scan_summary).""" campaign = data_mod.get_campaign(seed) rates = data_mod.get_rate_arrays(seed) events: List[FailureEvent] = [] # ---- classification flips (candidate wrong, baseline right) ------ flips: Dict[Tuple[str, str], List] = defaultdict(list) for obs in campaign.observations: if (obs.candidate.predicted_class != obs.gt_class and obs.baseline.predicted_class == obs.gt_class): flips[(obs.gt_class, obs.candidate.predicted_class)].append(obs) for (gt_cls, pred_cls), members in sorted(flips.items()): safety = gt_cls in SAFETY_CRITICAL if not safety and len(members) < 3: continue # noise floor for non-safety patterns if gt_cls == "pedestrian" and pred_cls == "construction_cone": basis = DetectionBasis( method="paired candidate-vs-GT scan + rate-population " "telemetry denominator", candidate_events=int(rates.candidate_flip.sum()), baseline_events=int(rates.baseline_flip.sum()), denominator=rates.n, candidate_rate=float(rates.candidate_flip.mean()), baseline_rate=float(rates.baseline_flip.mean()), notes=[f"{len(members)} concrete scene instances captured " "in the campaign for snippet/replay evidence", "rate telemetry provides the population denominator"]) else: n_cls = sum(1 for o in campaign.observations if o.gt_class == gt_cls) b_ev = sum(1 for o in campaign.observations if o.gt_class == gt_cls and o.baseline.predicted_class == pred_cls) basis = DetectionBasis( method="paired candidate-vs-GT scan (campaign only)", candidate_events=len(members), baseline_events=b_ev, denominator=n_cls, candidate_rate=len(members) / max(n_cls, 1), baseline_rate=b_ev / max(n_cls, 1)) instances = [FailureInstance( instance_id=new_id("inst"), sequence_id=o.sequence_id, frame_id=o.frame_id, frame_index=o.frame_index, object_instance_id=o.object_instance_id, gt_class=o.gt_class, predicted_class=o.candidate.predicted_class, confidence=o.candidate.confidence, distance_m=o.distance_m, construction_zone=o.construction_zone, time_of_day=o.time_of_day, weather=o.weather, geo_bucket=o.geo_bucket, occluded=o.occluded, has_planner_trace=o.has_planner_trace, ) for o in members] events.append(FailureEvent( failure_id=new_id("fail"), kind="classification_flip", title=f"{gt_cls} misclassified as {pred_cls}", description=(f"Candidate {campaign.candidate_model} predicts " f"{pred_cls} where GT and baseline say {gt_cls}."), gt_class=gt_cls, predicted_class=pred_cls, detection_basis=basis, instances=instances, baseline_model=campaign.baseline_model, candidate_model=campaign.candidate_model, dataset_fingerprint=data_mod.campaign_fingerprint(campaign), population_id=f"rate-pop-{seed}-n{rates.n}", )) # ---- metric regressions ------------------------------------------ summary = self._metric_scan(campaign) for kind, title, delta, threshold, extra in summary["checks"]: if abs(delta) <= threshold: continue events.append(FailureEvent( failure_id=new_id("fail"), kind=kind, title=title, description=f"metric delta {delta:+.4f} exceeded threshold " f"{threshold} on the paired campaign scan", detection_basis=DetectionBasis( method="paired campaign metric scan", candidate_events=0, baseline_events=0, denominator=len(campaign.observations), candidate_rate=0.0, baseline_rate=0.0, metric_deltas={title: round(delta, 5), **extra}), baseline_model=campaign.baseline_model, candidate_model=campaign.candidate_model, dataset_fingerprint=data_mod.campaign_fingerprint(campaign), population_id=f"campaign-{seed}", )) return events, summary def _metric_scan(self, campaign: data_mod.Campaign) -> Dict: obs = campaign.observations def center_err(pred, gt): return float(np.hypot(pred[0] - gt[0], pred[1] - gt[1])) cand_loc = float(np.mean([center_err(o.candidate.bbox_3d, o.gt_bbox_3d) for o in obs])) base_loc = float(np.mean([center_err(o.baseline.bbox_3d, o.gt_bbox_3d) for o in obs])) cand_conf = float(np.mean([o.candidate.confidence for o in obs if o.candidate.predicted_class == o.gt_class])) base_conf = float(np.mean([o.baseline.confidence for o in obs if o.baseline.predicted_class == o.gt_class])) cand_acc = float(np.mean([o.candidate.predicted_class == o.gt_class for o in obs])) base_acc = float(np.mean([o.baseline.predicted_class == o.gt_class for o in obs])) def churn(model: str) -> float: by_track: Dict[str, List[str]] = defaultdict(list) for o in sorted(obs, key=lambda x: x.frame_index): pred = o.candidate if model == "candidate" else o.baseline by_track[f"{o.sequence_id}:{o.object_instance_id}"].append( pred.predicted_class) switches = total = 0 for classes in by_track.values(): for a, b in zip(classes, classes[1:]): total += 1 switches += a != b return switches / max(total, 1) cand_churn, base_churn = churn("candidate"), churn("baseline") return { "observations": len(obs), "candidate_accuracy": round(cand_acc, 5), "baseline_accuracy": round(base_acc, 5), "checks": [ ("localization_regression", "mean_center_error_delta_m", cand_loc - base_loc, LOCALIZATION_DELTA_THRESHOLD_M, {"candidate_m": round(cand_loc, 4), "baseline_m": round(base_loc, 4)}), ("confidence_calibration_regression", "mean_correct_confidence_delta", cand_conf - base_conf, CONFIDENCE_DELTA_THRESHOLD, {"candidate": round(cand_conf, 4), "baseline": round(base_conf, 4)}), ("detection_regression", "classification_accuracy_delta", base_acc - cand_acc, 0.01, {"candidate": round(cand_acc, 4), "baseline": round(base_acc, 4)}), ("temporal_regression", "track_class_churn_delta", cand_churn - base_churn, CHURN_DELTA_THRESHOLD, {"candidate": round(cand_churn, 4), "baseline": round(base_churn, 4)}), ], } # BaseAgent plumbing (detect() is the natural API; _analyze wraps it). def _analyze(self, failure_id: str, **inputs): events, summary = self.detect(inputs.get("seed", data_mod.DEFAULT_SEED)) output = { "events_detected": len(events), "titles": [e.title for e in events], "scan_summary": {k: v for k, v in summary.items() if k != "checks"}, } conf = 0.9 if events else 0.6 return (output, conf, "deterministic paired scan; confidence reflects scan coverage", no_escalation()) def _llm_prompt(self, failure_id, output, **inputs): return ("Summarize this perception failure detection scan for an " "engineering audience: " + compact(output))