Spaces:
Paused
Paused
File size: 9,183 Bytes
b5df635 | 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 191 192 193 194 195 196 | """Agent pipeline tests: canonical fixtures, fact/inference separation,
stripped-telemetry -> UNKNOWN, scorecard completeness and tier tagging."""
from __future__ import annotations
import json
import pytest
from sensorflow.retro import store as retro_store
from sensorflow.retro.agent.orchestrator import analyze, analyze_fixture
from sensorflow.retro.scorecard import (EvidenceTier, LaunchRecommendation,
RetrospectiveScorecard, Severity)
@pytest.fixture()
def phantom(retro_root) -> RetrospectiveScorecard:
return analyze_fixture("phantom_brake_plastic_bag", backend="mock")
@pytest.fixture()
def missed_ped(retro_root) -> RetrospectiveScorecard:
return analyze_fixture("missed_pedestrian_rain", backend="mock")
# ----------------------------------------------------- canonical fixture A
def test_phantom_brake_fixture_analysis(phantom):
sc = phantom
assert sc.failure_type == "FALSE_POSITIVE"
assert sc.severity == Severity.CRITICAL # hard brake + tailgater
assert sc.launch_recommendation == LaunchRecommendation.FAIL
assert sc.object_class == "plastic_bag"
# causal hypotheses reference the actual evidence
texts = " ".join(h.hypothesis.lower() for h in sc.root_cause_hypotheses)
assert "deformable" in texts or "training-distribution" in texts
assert any("velocity latch" in h.hypothesis.lower()
for h in sc.root_cause_hypotheses)
# phantom braking requirement was retrieved
assert any(s.doc_id == "SFS-SAFE-001" for s in sc.retrieved_standards)
# ----------------------------------------------------- canonical fixture B
def test_missed_pedestrian_fixture_analysis(missed_ped):
sc = missed_ped
assert sc.failure_type == "FALSE_NEGATIVE"
assert sc.severity == Severity.CRITICAL
assert sc.launch_recommendation == LaunchRecommendation.FAIL
assert sc.human_review_required
# kinematics: pedestrian inside stopping distance
assert sc.stopping_distance_m is not None
assert sc.distance_to_object_m <= sc.stopping_distance_m * 1.05
assert sc.ttc_s is not None and sc.ttc_s < 2.5
# rain-conditioned hypothesis grounded in sensor evidence
assert any("fusion" in h.hypothesis.lower() for h in sc.root_cause_hypotheses)
# SCR regression computed and significant -> reflected in the gate
assert sc.safety_critical_recall_impact == pytest.approx(-3 / 1240, abs=1e-6)
assert sc.statistical_significance.significant is True
assert any(s.doc_id in ("SFS-SAFE-001", "SFS-PERC-002")
for s in sc.retrieved_standards)
# --------------------------------------------------- fact/inference separation
def test_fact_inference_separation(missed_ped):
sc = missed_ped
tiers = {e.tier for e in sc.evidence}
assert {EvidenceTier.OBSERVED, EvidenceTier.DERIVED,
EvidenceTier.RETRIEVED, EvidenceTier.AI_HYPOTHESIS} <= tiers
# hypotheses never appear as observed/derived facts
for e in sc.evidence:
if e.tier == EvidenceTier.AI_HYPOTHESIS:
assert "inference" in e.provenance
for h in sc.root_cause_hypotheses:
assert h.tier == EvidenceTier.AI_HYPOTHESIS
assert h.supporting_evidence_keys, "hypotheses must cite evidence"
# observed items carry artifact provenance
observed = [e for e in sc.evidence if e.tier == EvidenceTier.OBSERVED]
assert all("artifact" in e.provenance or "eval log field" in e.provenance
for e in observed)
# rendered scorecard keeps the five-way distinction
md = sc.render_markdown()
for heading in ("FACT", "DERIVED FACT", "RETRIEVED REQUIREMENT",
"AI HYPOTHES", "SAFETY DETERMINATION"):
assert heading in md
# ------------------------------------------------- stripped telemetry -> UNKNOWN
def test_stripped_telemetry_produces_unknown_not_guesses(retro_root):
sc = analyze_fixture("missing_telemetry_variant", backend="mock")
assert sc.failure_type == "FALSE_NEGATIVE"
# stripped fields must be UNKNOWN, never guessed
assert sc.ego_speed_mps is None
assert sc.stopping_distance_m is None
assert sc.ttc_s is None
assert "ego.speed_mps" in sc.uncertainty.missing_fields
assert "counterfactual_planner_response" in sc.uncertainty.missing_fields
assert "stopping_distance" in sc.uncertainty.unknown_metrics
assert "ttc" in sc.uncertainty.unknown_metrics
# UNKNOWN markers surface in the evidence chain
unknown_ev = [e for e in sc.evidence if e.value == "UNKNOWN"]
assert unknown_ev
# and the gate refuses to make a decision
assert sc.launch_recommendation == LaunchRecommendation.INSUFFICIENT_EVIDENCE
assert sc.human_review_required
def test_dynamically_stripped_field_becomes_unknown(retro_root):
"""Strip a field from fixture B at runtime -> the output marks it UNKNOWN."""
from sensorflow.retro.tools.builtin import FIXTURES_DIR
log = json.loads((FIXTURES_DIR / "missed_pedestrian_rain.json").read_text())
del log["ego"]["speed_mps"]
log["evaluation_id"] = "EVAL-STRIPPED-01"
upload = retro_store.uploads_dir() / "stripped.json"
upload.write_text(json.dumps(log))
sc = analyze(path=str(upload), backend="mock")
assert sc.ego_speed_mps is None
assert "ego.speed_mps" in sc.uncertainty.missing_fields
assert sc.stopping_distance_m is None # not computable without speed
assert sc.launch_recommendation == LaunchRecommendation.INSUFFICIENT_EVIDENCE
# ------------------------------------------------- severity divergence flow
def test_llm_severity_divergence_flagged(retro_root):
"""On the missing-telemetry variant the scripted 'LLM' proposes CRITICAL
(near-miss heuristic) while policy computes DISRUPTIVE from the sparse
evidence -> divergence recorded, policy wins, human review required."""
sc = analyze_fixture("missing_telemetry_variant", backend="mock")
assert sc.ai_proposed_severity == Severity.CRITICAL
assert sc.severity == Severity.DISRUPTIVE
assert sc.severity_divergence
assert any("diverged" in r for r in sc.human_review_reasons)
assert any("severity adjudication" in n for n in sc.uncertainty.notes)
# ------------------------------------------------------ variants + persistence
def test_benign_variants_pass(retro_root):
fp = analyze_fixture("benign_fp_static_sign", backend="mock")
fn = analyze_fixture("noncritical_fn_parked_vehicle", backend="mock")
assert fp.severity == Severity.BENIGN
assert fp.launch_recommendation == LaunchRecommendation.PASS
assert fn.severity == Severity.BENIGN
assert fn.launch_recommendation == LaunchRecommendation.PASS
assert not fn.human_review_required
def test_analysis_persisted_with_audit(retro_root):
sc = analyze_fixture("phantom_brake_plastic_bag", backend="mock")
stored = retro_store.load_analysis(sc.evaluation_id)
assert stored and stored["scorecard"]["failure_type"] == "FALSE_POSITIVE"
assert "# Retrospective Scorecard" in stored["markdown"]
audit = retro_store.read_audit(stored["audit_analysis_id"])
tools_used = {r["tool"] for r in audit}
assert {"log_reader", "metric_calculator", "safety_standard_rag",
"historical_failure_search", "distribution_analysis"} <= tools_used
# no write-tool call happened during analysis
assert "create_evaluation_case" not in tools_used
def test_scorecard_completeness_spec_fields(missed_ped):
"""Every spec field exists and the strictly-typed model round-trips."""
sc = missed_ped
dumped = json.loads(sc.model_dump_json())
for field in ["evaluation_id", "failure_type", "severity",
"safety_critical_recall_impact", "behavioral_consequence",
"launch_recommendation", "baseline_model", "candidate_model",
"scenario", "object_class", "ground_truth", "prediction",
"confidence", "ego_speed_mps", "distance_to_object_m",
"stopping_distance_m", "ttc_s", "planner_response",
"disengagement_probability", "metric_delta",
"statistical_significance", "distribution_shift",
"root_cause_hypotheses", "evidence", "retrieved_standards",
"uncertainty", "human_review_required", "policy_version",
"agent_version"]:
assert field in dumped, f"missing spec field {field}"
assert dumped["policy_version"].startswith("retro-policy/")
assert dumped["agent_version"].startswith("retro-agent/")
restored = RetrospectiveScorecard(**dumped)
assert restored.severity == sc.severity
def test_historical_search_finds_prior_analyses(retro_root):
analyze_fixture("missed_pedestrian_rain", backend="mock")
from sensorflow.retro.tools.builtin import build_registry
reg = build_registry(analysis_id="hist-test")
res = reg.call("historical_failure_search",
{"query_text": "pedestrian rain night missed detection", "k": 3})
assert res.result["corpus_size"] >= 1
assert any(m["evaluation_id"] == "EVAL-2026-0802-MP01"
for m in res.result["matches"])
|