Spaces:
Paused
Paused
| """Agents: seqeval delegation, small-sample flags, fusion-conflict escalation, | |
| hypothesis labeling, safety UNCERTAIN handling, advisory-only authority.""" | |
| from __future__ import annotations | |
| from sensorflow.agentic import data as data_mod | |
| from sensorflow.agentic import review as review_mod | |
| from sensorflow.agentic.agents import (SafetyImpactAgent, | |
| SensorFusionVerificationAgent, | |
| StatisticalRegressionAgent, | |
| VLMSceneAnalysisAgent) | |
| from sensorflow.agentic.agents import statistical as stat_mod | |
| from sensorflow.agentic.models import (DetectionBasis, FailureEvent, | |
| FailureInstance, new_id) | |
| def _instance(**overrides) -> FailureInstance: | |
| kwargs = dict( | |
| instance_id=new_id("inst"), sequence_id="bev-seq-101-0", | |
| frame_id="bev-seq-101-0-f0006", frame_index=6, | |
| object_instance_id="bev-seq-101-0-obj-1", | |
| gt_class="pedestrian", predicted_class="construction_cone", | |
| confidence=0.62, distance_m=22.0, construction_zone=True, | |
| time_of_day="day", weather="clear", geo_bucket="bay_area_urban", | |
| occluded=False, has_planner_trace=False) | |
| kwargs.update(overrides) | |
| return FailureInstance(**kwargs) | |
| def _failure(instances) -> FailureEvent: | |
| return FailureEvent( | |
| failure_id=new_id("fail"), kind="classification_flip", | |
| title="pedestrian misclassified as construction_cone", | |
| gt_class="pedestrian", predicted_class="construction_cone", | |
| detection_basis=DetectionBasis( | |
| method="test", candidate_events=25, baseline_events=5, | |
| denominator=240_000, candidate_rate=25 / 240_000, | |
| baseline_rate=5 / 240_000), | |
| instances=instances, dataset_fingerprint="fp-test", | |
| population_id="rate-pop-test") | |
| # ------------------------------------------------------------------ statistical | |
| def test_statistical_agent_delegates_to_seqeval_and_flags_small_samples(monkeypatch): | |
| constructed = [] | |
| real_cls = stat_mod.PairedSequentialTest | |
| class SpyTest(real_cls): | |
| def __init__(self, *a, **kw): | |
| constructed.append((a, kw)) | |
| super().__init__(*a, **kw) | |
| monkeypatch.setattr(stat_mod, "PairedSequentialTest", SpyTest) | |
| agent = StatisticalRegressionAgent() | |
| assessment = agent.assess(seed=data_mod.DEFAULT_SEED) | |
| # the anytime-valid machinery really is seqeval's, not a reimplementation | |
| assert constructed, "PairedSequentialTest (seqeval) was not instantiated" | |
| assert assessment.seqeval["delegated_to"] == \ | |
| "sensorflow.seqeval.sequential.PairedSequentialTest" | |
| assert assessment.seqeval["decision"] in ( | |
| "REGRESSION", "PASS", "INSUFFICIENT_EVIDENCE") | |
| assert assessment.power_mde["method"] == \ | |
| "sensorflow.seqeval.sequential.approx_mde" | |
| # rare-event handling: exact binomial drives the significance claim | |
| assert "binomial" in assessment.significance_method | |
| assert assessment.exact_binomial_p is not None | |
| assert assessment.significant is True | |
| # small-sample instability is flagged (few dozen events on 240k denom) | |
| assert assessment.small_sample_flags | |
| assert any("event count" in f for f in assessment.small_sample_flags) | |
| # Wilson CIs present and contain the point estimates | |
| lo, hi = assessment.candidate.wilson_ci | |
| assert lo <= assessment.candidate.rate <= hi | |
| def test_statistical_agent_reports_measurements_not_hypotheses(): | |
| result = StatisticalRegressionAgent().run("fail-x", seed=data_mod.DEFAULT_SEED) | |
| assert result.authority == "ADVISORY_ONLY" | |
| assert result.epistemic_status == "OBSERVED" | |
| assert result.status == "ok" | |
| # ------------------------------------------------------------------ fusion | |
| class _FakeDet: | |
| def __init__(self, inst_id, cls, dims, x, y): | |
| self.source_instance_id = inst_id | |
| self.class_name = cls | |
| self.confidence = 0.9 | |
| self.dims = list(dims) | |
| self.x, self.y = x, y | |
| def test_fusion_conflict_triggers_mandatory_human_review(ped_cone_failure, | |
| monkeypatch): | |
| """Force camera/LiDAR disagreement: camera says 'car' (non-VRU), LiDAR | |
| shape matches the GT pedestrian template -> modality_conflict, which must | |
| escalate to mandatory human review.""" | |
| from sensorflow.agentic.agents import fusion_verification as fv | |
| inst = ped_cone_failure.instances[0] | |
| gt_dims = fv.scenes_mod.CLASS_DIMS.get(inst.gt_class, (0.8, 0.8, 1.8)) | |
| monkeypatch.setattr( | |
| fv.sensors_mod, "simulate_camera", | |
| lambda frame, seq, rng: [_FakeDet(inst.object_instance_id, "car", | |
| gt_dims, 10.0, 5.0)]) | |
| monkeypatch.setattr( | |
| fv.sensors_mod, "simulate_lidar", | |
| lambda frame, seq, rng: [_FakeDet(inst.object_instance_id, | |
| inst.gt_class, gt_dims, 10.0, 5.0)]) | |
| result = SensorFusionVerificationAgent().run( | |
| ped_cone_failure.failure_id, failure=ped_cone_failure) | |
| assert result.output["overall_verdict"] == "modality_conflict" | |
| assert result.escalation.required | |
| assert "modality_disagreement" in result.escalation.human_review_triggers | |
| # and the governance layer fires the mandatory trigger on the verdict | |
| triggers = review_mod.mandatory_review_triggers( | |
| {"severity": "S1", "outcome": "CONTINUE_INVESTIGATION", | |
| "recommended_option": "EXPAND_EVALUATION"}, | |
| fusion_verdict="modality_conflict", gt_available=True, | |
| small_sample=False, behavioral_evidence="none") | |
| fired = {t["trigger"] for t in triggers if t["fired"]} | |
| assert "modality_disagreement" in fired | |
| def test_fusion_agent_runs_deterministically(ped_cone_failure): | |
| r1 = SensorFusionVerificationAgent().run( | |
| ped_cone_failure.failure_id, failure=ped_cone_failure) | |
| r2 = SensorFusionVerificationAgent().run( | |
| ped_cone_failure.failure_id, failure=ped_cone_failure) | |
| assert r1.output["overall_verdict"] == r2.output["overall_verdict"] | |
| assert r1.output["per_instance"] == r2.output["per_instance"] | |
| assert r1.output["overall_verdict"] != "verification_failed" | |
| # ------------------------------------------------------------------ VLM + safety | |
| def test_vlm_output_is_always_labeled_hypothesis(): | |
| failure = _failure([_instance(), _instance(time_of_day="night")]) | |
| result = VLMSceneAnalysisAgent().run(failure.failure_id, failure=failure) | |
| assert result.epistemic_status == "HYPOTHESIS" | |
| assert "HYPOTHESIS" in result.output["epistemic_label"] | |
| assert all(h["label"] == "HYPOTHESIS" for h in result.output["hypotheses"]) | |
| # confidence capped: hypotheses can never look like confirmed evidence | |
| assert result.confidence <= 0.6 | |
| def test_safety_agent_uncertain_without_replay_evidence(): | |
| failure = _failure([_instance(has_planner_trace=False)]) | |
| result = SafetyImpactAgent().run(failure.failure_id, failure=failure) | |
| assert result.output["behavioral_evidence"] == "none" | |
| assert result.output["assessment"] == \ | |
| "safety assessment UNCERTAIN — no behavioral evidence" | |
| chain = {c["link"]: c["status"] for c in result.output["chain"]} | |
| assert chain["planner"] == "UNAVAILABLE" | |
| assert chain["behavior"] == "UNAVAILABLE" | |
| assert result.escalation.required # uncertainty escalates, never passes | |
| def test_safety_agent_uses_replay_evidence_and_ssam_kinematics(): | |
| failure = _failure([_instance(has_planner_trace=True)]) | |
| result = SafetyImpactAgent().run(failure.failure_id, failure=failure) | |
| assert result.output["behavioral_evidence"] == "observed" | |
| assert result.output["replayed_instances"] == 1 | |
| kin = result.output["kinematics"] | |
| assert "ssam_ext" in kin["method"] | |
| # delayed reaction must cost stopping distance | |
| assert kin["stopping_distance_delayed_m"] > kin["stopping_distance_nominal_m"] | |
| chain = {c["link"]: c["status"] for c in result.output["chain"]} | |
| assert chain["planner"] == "OBSERVED" | |
| assert chain["behavior"] == "OBSERVED" | |
| def test_agent_failure_is_captured_not_raised(): | |
| result = SafetyImpactAgent().run("fail-x", failure=None) # assert trips | |
| assert result.status == "failed" | |
| assert result.escalation.required | |
| assert "agent_failure" in result.escalation.human_review_triggers | |