Spaces:
Paused
Paused
File size: 8,390 Bytes
963e554 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 | """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
|