Spaces:
Paused
Paused
| """Five-layer pipeline orchestrator. | |
| FAILURE DETECTION -> EVIDENCE AGGREGATION -> FAILURE ANALYSIS -> | |
| LAUNCH DECISION -> LEARNING FLYWHEEL | |
| Explicit staged state (pydantic PipelineState), orchestrated by this | |
| deterministic module — never one autonomous agent controlling the chain. | |
| Deterministic components own metrics, thresholds, statistical tests, lineage, | |
| policy, gates and the audit log; agents contribute typed advisory results. | |
| Every stage transition, agent output and policy evaluation is appended to the | |
| immutable audit trail (store.audit). | |
| """ | |
| from __future__ import annotations | |
| from typing import Dict, List, Optional | |
| from sensorflow.agentic import concentration as conc_mod | |
| from sensorflow.agentic import data as data_mod | |
| from sensorflow.agentic import evidence as evidence_mod | |
| from sensorflow.agentic import flywheel as flywheel_mod | |
| from sensorflow.agentic import policy as policy_mod | |
| from sensorflow.agentic import review as review_mod | |
| from sensorflow.agentic import scorecard as scorecard_mod | |
| from sensorflow.agentic import snippets as snippets_mod | |
| from sensorflow.agentic import store as store_mod | |
| from sensorflow.agentic.agents import (EvalFlywheelAgent, FailureDetectionAgent, | |
| LaunchDecisionAgent, SafetyImpactAgent, | |
| ScenarioMiningAgent, | |
| SensorFusionVerificationAgent, | |
| StatisticalRegressionAgent, | |
| VLMSceneAnalysisAgent) | |
| from sensorflow.agentic.models import (STAGES, EvidenceGraph, FailureEvent, | |
| PipelineState, StatisticalAssessment, | |
| now_iso) | |
| SAFETY_CRITICAL = {"pedestrian", "cyclist", "motorcycle"} | |
| MAX_SNIPPETS = 6 | |
| # ------------------------------------------------------------------ persistence | |
| def save_failure(failure: FailureEvent) -> None: | |
| store_mod.write_json(failure, "failures", f"{failure.failure_id}.json") | |
| def get_failure(failure_id: str) -> Optional[FailureEvent]: | |
| doc = store_mod.read_json("failures", f"{failure_id}.json") | |
| return FailureEvent.model_validate(doc) if doc else None | |
| def list_failures() -> List[Dict]: | |
| out = [] | |
| for name in store_mod.list_dir("failures"): | |
| doc = store_mod.read_json("failures", name) | |
| if doc: | |
| doc["instances"] = doc.get("instances", [])[:3] # queue view: trim | |
| out.append(doc) | |
| return sorted(out, key=lambda d: d.get("created_at", "")) | |
| def save_state(state: PipelineState) -> None: | |
| state.updated_at = now_iso() | |
| store_mod.write_json(state, "pipeline", f"{state.failure_id}.json") | |
| def get_state(failure_id: str) -> Optional[PipelineState]: | |
| doc = store_mod.read_json("pipeline", f"{failure_id}.json") | |
| return PipelineState.model_validate(doc) if doc else None | |
| def get_evidence_graph(failure_id: str) -> Optional[EvidenceGraph]: | |
| doc = store_mod.read_json("evidence", f"{failure_id}.json") | |
| return EvidenceGraph.model_validate(doc) if doc else None | |
| def get_snippets(failure_id: str) -> List[Dict]: | |
| return [store_mod.read_json("snippets", failure_id, n) | |
| for n in store_mod.list_dir("snippets", failure_id)] | |
| def _transition(state: PipelineState, stage: str, status: str, | |
| detail: str = "") -> None: | |
| rec = state.stage_record(stage) | |
| if status == "running": | |
| rec.started_at = now_iso() | |
| else: | |
| rec.finished_at = now_iso() | |
| rec.status = status | |
| rec.detail = detail | |
| save_state(state) | |
| store_mod.audit("stage_transition", state.failure_id, "pipeline", | |
| detail=f"{stage} -> {status}" + (f": {detail}" if detail else "")) | |
| def _record_agent(state: PipelineState, result) -> None: | |
| state.agent_results[result.agent] = result | |
| store_mod.audit("agent_output", state.failure_id, result.agent, | |
| detail=f"status={result.status} confidence={result.confidence}", | |
| payload={"agent_version": result.agent_version, | |
| "escalation": result.escalation.model_dump(), | |
| "epistemic_status": result.epistemic_status}) | |
| # ------------------------------------------------------------------ layer 1 | |
| def detect_failures(seed: int = data_mod.DEFAULT_SEED, | |
| use_llm: bool = False) -> List[FailureEvent]: | |
| agent = FailureDetectionAgent(use_llm=use_llm) | |
| events, summary = agent.detect(seed) | |
| scan_result = agent.run("_detection_scan", seed=seed) | |
| for failure in events: | |
| save_failure(failure) | |
| state = PipelineState(failure_id=failure.failure_id) | |
| _transition(state, "FAILURE_DETECTION", "running") | |
| scan_copy = scan_result.model_copy(update={"failure_id": failure.failure_id}) | |
| _record_agent(state, scan_copy) | |
| _transition(state, "FAILURE_DETECTION", "complete", | |
| detail=f"{failure.title} " | |
| f"(candidate {failure.detection_basis.candidate_events} " | |
| f"vs baseline {failure.detection_basis.baseline_events} " | |
| f"over n={failure.detection_basis.denominator})") | |
| store_mod.audit("detection_scan", None, "failure_detection", | |
| detail=f"{len(events)} failure patterns emitted", | |
| payload={"scan": {k: v for k, v in summary.items() | |
| if k != "checks"}}) | |
| return events | |
| # ------------------------------------------------------------------ layer 2 | |
| def run_evidence_aggregation(failure: FailureEvent, state: PipelineState, | |
| seed: int) -> None: | |
| _transition(state, "EVIDENCE_AGGREGATION", "running") | |
| campaign = data_mod.get_campaign(seed) | |
| graph = evidence_mod.build_graph(failure, campaign) | |
| store_mod.write_json(graph, "evidence", f"{failure.failure_id}.json") | |
| n_snips = 0 | |
| for inst in failure.instances[:MAX_SNIPPETS]: | |
| snip = snippets_mod.build_snippet(failure, inst, campaign) | |
| store_mod.write_json(snip, "snippets", failure.failure_id, | |
| f"{snip.snippet_id}.json") | |
| n_snips += 1 | |
| _transition(state, "EVIDENCE_AGGREGATION", "complete", | |
| detail=f"evidence graph ({len(graph.nodes)} nodes) + " | |
| f"{n_snips} snippets") | |
| # ------------------------------------------------------------------ layer 3 | |
| def _is_rate_population_failure(failure: FailureEvent) -> bool: | |
| return (failure.kind == "classification_flip" | |
| and failure.gt_class == "pedestrian" | |
| and failure.predicted_class == "construction_cone") | |
| def run_failure_analysis(failure: FailureEvent, state: PipelineState, | |
| seed: int, use_llm: bool = False) -> None: | |
| _transition(state, "FAILURE_ANALYSIS", "running") | |
| campaign = data_mod.get_campaign(seed) | |
| vlm = VLMSceneAnalysisAgent(use_llm=use_llm).run( | |
| failure.failure_id, failure=failure) | |
| _record_agent(state, vlm) | |
| fusion = SensorFusionVerificationAgent(use_llm=use_llm).run( | |
| failure.failure_id, failure=failure, seed=seed) | |
| _record_agent(state, fusion) | |
| mining = ScenarioMiningAgent(use_llm=use_llm).run( | |
| failure.failure_id, failure=failure) | |
| _record_agent(state, mining) | |
| stat_agent = StatisticalRegressionAgent(use_llm=use_llm) | |
| if _is_rate_population_failure(failure): | |
| stat = stat_agent.run(failure.failure_id, seed=seed) | |
| state.statistical = StatisticalAssessment.model_validate(stat.output) | |
| state.concentration = conc_mod.analyze_concentration(failure.failure_id, seed) | |
| else: | |
| stat = stat_agent.run(failure.failure_id, seed=seed) | |
| stat.output["scope_note"] = ( | |
| "rate-population telemetry instruments the pedestrian->" | |
| "construction_cone channel; for this failure the campaign counts " | |
| "in detection_basis are the denominator of record") | |
| state.statistical = None | |
| state.concentration = None | |
| _record_agent(state, stat) | |
| safety = SafetyImpactAgent(use_llm=use_llm).run( | |
| failure.failure_id, failure=failure) | |
| _record_agent(state, safety) | |
| # ---- deterministic evidence-graph updates ------------------------- | |
| graph = get_evidence_graph(failure.failure_id) | |
| if graph is not None: | |
| if fusion.status != "failed": | |
| evidence_mod.set_node( | |
| graph, "Sensors", | |
| "OBSERVED" if fusion.output.get("overall_verdict") | |
| != "verification_failed" else "UNAVAILABLE", | |
| f"fusion verdict: {fusion.output.get('overall_verdict')}", | |
| {"verdict": fusion.output.get("overall_verdict"), | |
| "verdict_counts": fusion.output.get("verdict_counts")}, | |
| "sensor fusion verification agent (deterministic re-simulation)") | |
| if mining.status == "ok": | |
| evidence_mod.set_node( | |
| graph, "HistoricalSimilarity", "DERIVED", | |
| f"novelty: {mining.output.get('novelty')} " | |
| f"({mining.output.get('exact_precedent_count')} precedents)", | |
| {"novelty": mining.output.get("novelty"), | |
| "top_similar": mining.output.get("similar_historical", [])[:3], | |
| "clusters": mining.output.get("clusters", [])}, | |
| "scenario mining agent (deterministic retrieval)") | |
| if state.statistical is not None: | |
| s = state.statistical | |
| evidence_mod.set_node( | |
| graph, "Frequency", "OBSERVED", | |
| f"candidate {s.candidate.rate:.6f} " | |
| f"(CI {s.candidate.wilson_ci}) vs baseline " | |
| f"{s.baseline.rate:.6f}; significant={s.significant}", | |
| {"statistical": s.model_dump()}, | |
| "statistical regression agent (deterministic; stats via " | |
| "seqeval + exact binomial)") | |
| if safety.status != "failed": | |
| observed = safety.output.get("behavioral_evidence") == "observed" | |
| evidence_mod.set_node( | |
| graph, "SafetyConsequence", | |
| "OBSERVED" if observed else "UNAVAILABLE", | |
| safety.output.get("assessment", ""), | |
| {"behavioral_evidence": safety.output.get("behavioral_evidence"), | |
| "classification": safety.output.get("behavioral_classification"), | |
| "worst_case": safety.output.get("worst_case"), | |
| "chain": safety.output.get("chain")}, | |
| "safety impact agent (replay evidence via safety.ssam_ext)") | |
| if observed: | |
| evidence_mod.set_node( | |
| graph, "Planner", "OBSERVED", | |
| f"replayed {safety.output.get('replayed_instances')} " | |
| "planner-response trace(s)", | |
| {"replays": safety.output.get("replays", [])}, | |
| "safety impact agent (safety.ssam_ext replay)") | |
| # VLM hypotheses attach to Environment as clearly-labeled hypotheses | |
| env = graph.node("Environment") | |
| if env is not None and vlm.status == "ok": | |
| env.fields["scene_hypotheses"] = vlm.output.get("hypotheses", []) | |
| env.caveats.append("scene_hypotheses are HYPOTHESIS-status agent " | |
| "conjectures, not observations") | |
| store_mod.write_json(graph, "evidence", f"{failure.failure_id}.json") | |
| failure.status = "investigating" | |
| save_failure(failure) | |
| _transition(state, "FAILURE_ANALYSIS", "complete", | |
| detail=f"5 analysis agents ran; fusion=" | |
| f"{fusion.output.get('overall_verdict')}") | |
| # ------------------------------------------------------------------ layer 4 | |
| def evidence_confidence(state: PipelineState) -> str: | |
| s = state.statistical | |
| if s is None: | |
| return "INSUFFICIENT_EVIDENCE" | |
| if s.significant and not s.small_sample_flags: | |
| return "CONFIRMED" | |
| if s.significant: | |
| return "LIKELY" | |
| if s.candidate.events > 0: | |
| return "POSSIBLE" | |
| return "INSUFFICIENT_EVIDENCE" | |
| def build_policy_input(failure: FailureEvent, state: PipelineState, | |
| overrides: Optional[Dict] = None) -> policy_mod.PolicyInput: | |
| """Deterministic mapping from pipeline artifacts to the policy input.""" | |
| safety = state.agent_results.get("safety_impact") | |
| fusion = state.agent_results.get("sensor_fusion_verification") | |
| mining = state.agent_results.get("scenario_mining") | |
| behavioral = "none" | |
| contained = False | |
| downstream = "uncertain" | |
| if safety is not None and safety.status != "failed": | |
| ev = safety.output.get("behavioral_evidence") | |
| if ev == "observed": | |
| cls = safety.output.get("behavioral_classification") | |
| behavioral = cls or "observed_contained" | |
| contained = cls == "observed_contained" | |
| downstream = ("observed_critical" if cls == "observed_unsafe" | |
| else "observed_degraded") | |
| else: | |
| behavioral, downstream = "none", "uncertain" | |
| stat = state.statistical | |
| conc = state.concentration | |
| concentrated_dim = (conc.concentrated_dimensions[0] | |
| if conc and conc.concentrated_dimensions else None) | |
| resid_outside = None | |
| excl_share = None | |
| if conc is not None and concentrated_dim == "construction_zone": | |
| row_out = next((s for s in conc.strata | |
| if s.stratum == "non_construction"), None) | |
| row_in = next((s for s in conc.strata | |
| if s.stratum == "construction"), None) | |
| if row_out is not None: | |
| resid_outside = row_out.stratum_rate | |
| if row_in is not None: | |
| excl_share = row_in.exposure_share | |
| agent_failures = [r.agent for r in state.agent_results.values() | |
| if r.status == "failed"] | |
| kwargs = dict( | |
| failure_id=failure.failure_id, | |
| safety_critical_class=(failure.gt_class in SAFETY_CRITICAL | |
| if failure.gt_class else False), | |
| collision_observed=False, | |
| behavioral_evidence=behavioral, | |
| downstream_contained=contained, | |
| functional_impact=True, | |
| rate=stat.candidate.rate if stat else None, | |
| rate_ci=stat.candidate.wilson_ci if stat else None, | |
| denominator=stat.candidate.denominator if stat | |
| else failure.detection_basis.denominator, | |
| significant=stat.significant if stat else False, | |
| small_sample=bool(stat.small_sample_flags) if stat else True, | |
| exposure_share=excl_share, | |
| novelty=(mining.output.get("novelty", "unknown") | |
| if mining is not None and mining.status == "ok" else "unknown"), | |
| evidence_confidence=evidence_confidence(state), | |
| downstream_consequence=downstream, | |
| gt_available=True, | |
| lineage_complete=bool(failure.dataset_fingerprint | |
| and failure.candidate_model | |
| and failure.baseline_model), | |
| telemetry_available=failure.detection_basis.denominator > 0, | |
| fusion_verdict=(fusion.output.get("overall_verdict", | |
| "verification_failed") | |
| if fusion is not None and fusion.status != "failed" | |
| else "verification_failed"), | |
| agent_conflict=bool(agent_failures), | |
| conflict_details=[f"agent {a} failed" for a in agent_failures], | |
| gate_violated=False, | |
| mitigation_validated=False, | |
| concentration_dimension=concentrated_dim, | |
| concentrated=bool(conc and conc.determination == "concentrated"), | |
| odd_detector_recall=None, # REQUIRED-EVIDENCE: no zone detector measured | |
| odd_exclusion_share=excl_share, | |
| residual_rate_outside_odd=resid_outside, | |
| ) | |
| kwargs.update(overrides or {}) | |
| return policy_mod.PolicyInput(**kwargs) | |
| def run_launch_decision(failure: FailureEvent, state: PipelineState, | |
| seed: int, use_llm: bool = False, | |
| policy_input_overrides: Optional[Dict] = None) -> Dict: | |
| _transition(state, "LAUNCH_DECISION", "running") | |
| pol_input = build_policy_input(failure, state, policy_input_overrides) | |
| evaluation = policy_mod.evaluate(pol_input) | |
| state.policy_evaluation = evaluation | |
| narrative = LaunchDecisionAgent(use_llm=use_llm).run( | |
| failure.failure_id, failure=failure, policy_evaluation=evaluation) | |
| _record_agent(state, narrative) | |
| card = scorecard_mod.build_scorecard( | |
| failure, state.statistical, state.concentration, evaluation, | |
| state.agent_results) | |
| state.scorecard_id = card.scorecard_id | |
| triggers = review_mod.mandatory_review_triggers( | |
| evaluation, pol_input.fusion_verdict, pol_input.gt_available, | |
| pol_input.small_sample, pol_input.behavioral_evidence) | |
| evaluation["mandatory_review_triggers"] = triggers | |
| evaluation["human_review_required"] = any(t["fired"] for t in triggers) | |
| failure.severity = evaluation.get("severity") | |
| failure.policy_outcome = evaluation.get("outcome") | |
| failure.status = "decided" | |
| save_failure(failure) | |
| _transition(state, "LAUNCH_DECISION", "complete", | |
| detail=f"{evaluation['outcome']} -> " | |
| f"{evaluation['recommended_option']} " | |
| f"(policy {evaluation['policy_version']})") | |
| return evaluation | |
| # ------------------------------------------------------------------ layer 5 | |
| def run_learning_flywheel(failure: FailureEvent, state: PipelineState, | |
| use_llm: bool = False) -> Dict: | |
| _transition(state, "LEARNING_FLYWHEEL", "running") | |
| agent = EvalFlywheelAgent(use_llm=use_llm) | |
| result = agent.run(failure.failure_id, failure=failure, | |
| concentration=state.concentration) | |
| _record_agent(state, result) | |
| if result.output.get("proposal") is None: | |
| _transition(state, "LEARNING_FLYWHEEL", "blocked", | |
| detail="failure not human-validated; flywheel refused") | |
| return {"created": False, "reason": result.output.get("reason")} | |
| suite = flywheel_mod.create_or_update_suite(failure, result.output["proposal"]) | |
| state.suite_ids.append(suite.suite_id) | |
| _transition(state, "LEARNING_FLYWHEEL", "complete", | |
| detail=f"suite {suite.name} v{suite.version} " | |
| f"({len(suite.members)} members)") | |
| return {"created": True, "suite_id": suite.suite_id, | |
| "suite_name": suite.name, "version": suite.version, | |
| "members": len(suite.members)} | |
| # ------------------------------------------------------------------ dispatcher | |
| STAGE_ORDER = {s: i for i, s in enumerate(STAGES)} | |
| def run_stage(failure_id: str, stage: Optional[str] = None, | |
| seed: int = data_mod.DEFAULT_SEED, use_llm: bool = False, | |
| policy_input_overrides: Optional[Dict] = None) -> PipelineState: | |
| """Run one stage (the next pending one when stage is None).""" | |
| failure = get_failure(failure_id) | |
| state = get_state(failure_id) | |
| if failure is None or state is None: | |
| raise KeyError(f"Unknown failure {failure_id}") | |
| if stage is None: | |
| pending = [s.stage for s in state.stages if s.status == "pending"] | |
| if not pending: | |
| return state | |
| stage = pending[0] | |
| if stage not in STAGE_ORDER: | |
| raise ValueError(f"Unknown stage {stage}") | |
| for prior in state.stages: | |
| if STAGE_ORDER[prior.stage] < STAGE_ORDER[stage] \ | |
| and prior.status not in ("complete", "blocked"): | |
| raise ValueError(f"stage {stage} requires {prior.stage} to be " | |
| f"complete (currently {prior.status})") | |
| if stage == "EVIDENCE_AGGREGATION": | |
| run_evidence_aggregation(failure, state, seed) | |
| elif stage == "FAILURE_ANALYSIS": | |
| run_failure_analysis(failure, state, seed, use_llm) | |
| elif stage == "LAUNCH_DECISION": | |
| run_launch_decision(failure, state, seed, use_llm, | |
| policy_input_overrides) | |
| elif stage == "LEARNING_FLYWHEEL": | |
| run_learning_flywheel(failure, state, use_llm) | |
| elif stage == "FAILURE_DETECTION": | |
| pass # detection happens at detect_failures(); already complete | |
| return get_state(failure_id) or state | |