Deploy Sensorflow Studio (part 2)

sensorflow/rotr/__init__.py

@@ -0,0 +1,11 @@
+"""ROTR — Right-of-the-Road violation detection, triage, evaluation & training.
+
+Causal-layer separation is the load-bearing constraint of this package: a
+ROTR violation is NEVER auto-attributed to perception. Detection (rules.py),
+attribution (attribution.py), consequence (consequence.py), statistics
+(metrics.py, delegating to seqeval), mining (taxonomy.py) and governance
+(flywheel.py, stopship.py) are separate modules with typed contracts
+(models.py). Architecture: docs/architecture/rotr-architecture.md.
+"""
+
+SOFTWARE_VERSION = "rotr-0.1.0"

sensorflow/rotr/api.py

@@ -0,0 +1,427 @@
+"""REST surface for the ROTR capability: APIRouter under /api/rotr.
+
+Request-scoped over file persistence (platform convention); no background
+services. Endpoints cover the full loop: bank generation, detection runs,
+violation inspection with taxonomy queries, the attribution matrix,
+consequence replay, metrics + regression, clusters, HITL actions, flywheel
+artifacts + contamination guard, stop-ship policy, and docs serving."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+
+from sensorflow.rotr import SOFTWARE_VERSION, pipeline, store
+from sensorflow.rotr import flywheel as flywheel_mod
+from sensorflow.rotr.flywheel import ContaminationError
+from sensorflow.rotr.metrics import evaluate_candidate
+from sensorflow.rotr.models import now_iso
+from sensorflow.rotr.rules import CONFIG, RULES, RULESET_VERSION
+from sensorflow.rotr.scenes import MODEL_PROFILES
+from sensorflow.rotr.stopship import STOPSHIP_POLICY, policy_version
+from sensorflow.rotr.taxonomy import ROTRQuery, matches, parse_query
+
+router = APIRouter(prefix="/api/rotr", tags=["rotr"])
+
+
+@router.get("/health")
+def health() -> Dict:
+    return {"status": "ok", "software_version": SOFTWARE_VERSION,
+            "ruleset_version": RULESET_VERSION,
+            "stopship_policy_version": policy_version(),
+            "time": now_iso()}
+
+
+# ------------------------------------------------------------ banks
+
+
+class BankRequest(BaseModel):
+    n_scenarios: int = 28
+    seed: int = 7
+    model_version: str = "stack-v1"
+
+
+@router.post("/banks")
+def create_bank(req: BankRequest) -> Dict:
+    if req.model_version not in MODEL_PROFILES:
+        raise HTTPException(422, f"unknown model_version {req.model_version}; "
+                                 f"known: {sorted(MODEL_PROFILES)}")
+    if not (1 <= req.n_scenarios <= 500):
+        raise HTTPException(422, "n_scenarios must be in [1, 500]")
+    return pipeline.create_bank(req.n_scenarios, req.seed, req.model_version)
+
+
+@router.get("/banks")
+def list_banks() -> List[Dict]:
+    return pipeline.list_banks()
+
+
+@router.get("/banks/{bank_id}")
+def get_bank(bank_id: str) -> Dict:
+    doc = pipeline.load_bank(bank_id)
+    if doc is None:
+        raise HTTPException(404, f"unknown bank {bank_id}")
+    return pipeline.bank_summary(doc)
+
+
+@router.get("/banks/{bank_id}/scenarios/{scenario_id}")
+def get_scenario(bank_id: str, scenario_id: str) -> Dict:
+    sc = pipeline.load_scenario(bank_id, scenario_id)
+    if sc is None:
+        raise HTTPException(404, f"unknown scenario {scenario_id}")
+    return sc.model_dump()
+
+
+# ------------------------------------------------------------ runs
+
+
+class RunRequest(BaseModel):
+    bank_id: Optional[str] = None
+    n_scenarios: int = 28
+    seed: int = 7
+    model_version: str = "stack-v1"
+    forward_to_agentic: bool = False
+
+
+@router.post("/runs")
+def create_run(req: RunRequest) -> Dict:
+    try:
+        doc = pipeline.execute_run(req.bank_id, req.n_scenarios, req.seed,
+                                   req.model_version,
+                                   forward_to_agentic=req.forward_to_agentic)
+    except KeyError as e:
+        raise HTTPException(404, str(e))
+    except ValueError as e:
+        raise HTTPException(422, str(e))
+    return _run_summary(doc)
+
+
+@router.get("/runs")
+def list_runs() -> List[Dict]:
+    return pipeline.list_runs()
+
+
+def _run_summary(doc: Dict) -> Dict:
+    return {k: doc[k] for k in
+            ("run_id", "bank_id", "model_version", "ruleset_version",
+             "n_scenarios", "scenario_summaries", "metrics", "clusters",
+             "gate")} | {"n_violations": len(doc["violations"])}
+
+
+@router.get("/runs/{run_id}")
+def get_run(run_id: str) -> Dict:
+    doc = pipeline.load_run(run_id)
+    if doc is None:
+        raise HTTPException(404, f"unknown run {run_id}")
+    return _run_summary(doc)
+
+
+# ------------------------------------------------------------ violations
+
+
+@router.get("/runs/{run_id}/violations")
+def list_violations(run_id: str, actor: Optional[str] = None,
+                    vulnerability: Optional[str] = None,
+                    legality: Optional[str] = None,
+                    environment: Optional[str] = None,
+                    interaction: Optional[str] = None,
+                    behavior: Optional[str] = None,
+                    road_geometry: Optional[str] = None,
+                    traffic_control: Optional[str] = None,
+                    visibility: Optional[str] = None,
+                    lighting: Optional[str] = None,
+                    weather: Optional[str] = None,
+                    consequence_class: Optional[str] = None,
+                    primary_layer: Optional[str] = None,
+                    text: Optional[str] = None) -> Dict:
+    doc = pipeline.load_run(run_id)
+    if doc is None:
+        raise HTTPException(404, f"unknown run {run_id}")
+    query = parse_query(text) if text else ROTRQuery()
+    explicit = {"actor": actor, "vulnerability": vulnerability,
+                "legality": legality, "environment": environment,
+                "interaction": interaction, "behavior": behavior,
+                "road_geometry": road_geometry,
+                "traffic_control": traffic_control, "visibility": visibility,
+                "lighting": lighting, "weather": weather,
+                "consequence_class": consequence_class,
+                "primary_layer": primary_layer}
+    for axis, val in explicit.items():
+        if val is not None:
+            setattr(query, axis, val)
+    return _query_run(doc, query)
+
+
+class QueryRequest(BaseModel):
+    run_id: str
+    text: Optional[str] = None
+    filters: Optional[Dict[str, str]] = None
+
+
+@router.post("/query")
+def structured_query(req: QueryRequest) -> Dict:
+    doc = pipeline.load_run(req.run_id)
+    if doc is None:
+        raise HTTPException(404, f"unknown run {req.run_id}")
+    query = parse_query(req.text) if req.text else ROTRQuery()
+    for k, v in (req.filters or {}).items():
+        if not hasattr(query, k):
+            raise HTTPException(422, f"unknown filter axis {k!r}")
+        setattr(query, k, v)
+    return _query_run(doc, query)
+
+
+def _query_run(doc: Dict, query: ROTRQuery) -> Dict:
+    env_by_scenario = {s["scenario_id"]: s for s in doc["scenario_summaries"]}
+    cluster_by_vid = {}
+    for c in doc["clusters"]:
+        for vid in c["member_violation_ids"]:
+            cluster_by_vid[vid] = c["cluster_id"]
+    results = []
+    for v in doc["violations"]:
+        attr = doc["attributions"].get(v["violation_id"], {})
+        cons = doc["consequences"].get(v["violation_id"], {})
+        env = env_by_scenario.get(v["scenario_id"], {})
+        if not matches(query, v.get("taxonomy", {}),
+                       cons.get("consequence_class"),
+                       attr.get("primary_layer"),
+                       {"lighting": env.get("lighting"),
+                        "weather": env.get("weather")}):
+            continue
+        results.append({
+            "violation_id": v["violation_id"],
+            "scenario_id": v["scenario_id"],
+            "rule_id": v["rule_id"],
+            "rule_version": v["rule_version"],
+            "description": v["description"],
+            "taxonomy": v.get("taxonomy", {}),
+            "evidence": v["evidence"],
+            "confidence": v["confidence"],
+            "primary_layer": attr.get("primary_layer"),
+            "consequence_class": cons.get("consequence_class"),
+            "cluster_id": cluster_by_vid.get(v["violation_id"]),
+            "environment": {"visibility": env.get("visibility"),
+                            "lighting": env.get("lighting"),
+                            "weather": env.get("weather")},
+            "provenance": v["provenance"],
+        })
+    return {"query": query.model_dump(), "n_results": len(results),
+            "results": results}
+
+
+@router.get("/runs/{run_id}/violations/{violation_id}")
+def get_violation(run_id: str, violation_id: str) -> Dict:
+    doc = pipeline.load_run(run_id)
+    if doc is None:
+        raise HTTPException(404, f"unknown run {run_id}")
+    v = next((v for v in doc["violations"]
+              if v["violation_id"] == violation_id), None)
+    if v is None:
+        raise HTTPException(404, f"unknown violation {violation_id}")
+    return {"violation": v,
+            "attribution": doc["attributions"].get(violation_id),
+            "consequence": doc["consequences"].get(violation_id)}
+
+
+@router.get("/runs/{run_id}/attribution")
+def attribution_matrix(run_id: str) -> Dict:
+    doc = pipeline.load_run(run_id)
+    if doc is None:
+        raise HTTPException(404, f"unknown run {run_id}")
+    rows = []
+    for v in doc["violations"]:
+        attr = doc["attributions"].get(v["violation_id"], {})
+        rows.append({
+            "violation_id": v["violation_id"],
+            "scenario_id": v["scenario_id"],
+            "rule_id": v["rule_id"],
+            "primary_layer": attr.get("primary_layer"),
+            "layers": attr.get("layers", {}),
+            "note": attr.get("note", ""),
+        })
+    return {"run_id": run_id, "rows": rows,
+            "invariant": "downstream failure never auto-implies perception; "
+                         "layers are implicated only by their own positive "
+                         "evidence"}
+
+
+@router.get("/runs/{run_id}/violations/{violation_id}/consequence")
+def get_consequence(run_id: str, violation_id: str) -> Dict:
+    doc = pipeline.load_consequence(run_id, violation_id)
+    if doc is None:
+        raise HTTPException(404, "no consequence replay for "
+                                 f"{violation_id} in {run_id}")
+    # Attach the scenario geometry the UI needs for the BEV replay view.
+    run = pipeline.load_run(run_id)
+    if run:
+        v = next((v for v in run["violations"]
+                  if v["violation_id"] == violation_id), None)
+        if v:
+            sc = pipeline.load_scenario(run["bank_id"], v["scenario_id"])
+            if sc:
+                doc["scenario_geometry"] = {
+                    "actual_context": sc.actual_context.model_dump(),
+                    "actors": [{
+                        "actor_id": a.actor_id, "class_name": a.class_name,
+                        "dims": a.dims,
+                        "states": [{"t": s.t, "x": s.x, "y": s.y,
+                                    "yaw": s.yaw} for s in a.states],
+                    } for a in sc.actors],
+                }
+    return doc
+
+
+# ------------------------------------------------------------ metrics + regression
+
+
+@router.get("/runs/{run_id}/metrics")
+def get_metrics(run_id: str) -> Dict:
+    doc = pipeline.load_run(run_id)
+    if doc is None:
+        raise HTTPException(404, f"unknown run {run_id}")
+    return doc["metrics"]
+
+
+@router.get("/runs/{run_id}/clusters")
+def get_clusters(run_id: str) -> List[Dict]:
+    doc = pipeline.load_run(run_id)
+    if doc is None:
+        raise HTTPException(404, f"unknown run {run_id}")
+    return doc["clusters"]
+
+
+class RegressionRequest(BaseModel):
+    baseline_run_id: str
+    candidate_run_id: str
+    seed: int = 11
+
+
+@router.post("/regression")
+def run_regression(req: RegressionRequest) -> Dict:
+    base = pipeline.load_run(req.baseline_run_id)
+    cand = pipeline.load_run(req.candidate_run_id)
+    if base is None or cand is None:
+        raise HTTPException(404, "unknown baseline or candidate run")
+    rid = f"reg-{req.baseline_run_id}-vs-{req.candidate_run_id}"
+    result = evaluate_candidate(rid, base, cand, seed=req.seed)
+    store.write_json(result.model_dump(), "regressions", f"{rid}.json")
+    return result.model_dump()
+
+
+@router.get("/regressions")
+def list_regressions() -> List[Dict]:
+    out = []
+    for name in store.list_dir("regressions"):
+        doc = store.read_json("regressions", name)
+        if doc:
+            out.append(doc)
+    return out
+
+
+# ------------------------------------------------------------ HITL + flywheel
+
+
+@router.get("/runs/{run_id}/hitl")
+def hitl_queue(run_id: str) -> List[Dict]:
+    if pipeline.load_run(run_id) is None:
+        raise HTTPException(404, f"unknown run {run_id}")
+    return flywheel_mod.get_queue(run_id)
+
+
+class HITLActionRequest(BaseModel):
+    run_id: str
+    review_id: str
+    action: str                       # VALIDATE | REJECT
+    actor: str
+    notes: str = ""
+
+
+@router.post("/hitl/action")
+def hitl_action(req: HITLActionRequest) -> Dict:
+    try:
+        return flywheel_mod.act(req.run_id, req.review_id, req.action,
+                                req.actor, req.notes)
+    except KeyError as e:
+        raise HTTPException(404, str(e))
+    except ValueError as e:
+        raise HTTPException(422, str(e))
+
+
+@router.get("/flywheel/suite")
+def regression_suite() -> Dict:
+    return flywheel_mod.get_suite()
+
+
+@router.get("/flywheel/candidates")
+def training_candidates() -> List[Dict]:
+    return flywheel_mod.list_candidates()
+
+
+class PromoteRequest(BaseModel):
+    candidate_id: str
+    actor: str
+
+
+@router.post("/flywheel/promote")
+def promote(req: PromoteRequest) -> Dict:
+    try:
+        return flywheel_mod.promote_to_training(req.candidate_id, req.actor)
+    except ContaminationError as e:
+        raise HTTPException(409, str(e))
+    except KeyError as e:
+        raise HTTPException(404, str(e))
+
+
+class OverrideRequest(BaseModel):
+    candidate_id: str
+    actor: str
+    reason: str
+
+
+@router.post("/flywheel/override")
+def override(req: OverrideRequest) -> Dict:
+    try:
+        return flywheel_mod.governance_override(req.candidate_id, req.actor,
+                                                req.reason)
+    except KeyError as e:
+        raise HTTPException(404, str(e))
+    except ValueError as e:
+        raise HTTPException(422, str(e))
+
+
+# ------------------------------------------------------------ policy + docs
+
+
+@router.get("/stopship/policy")
+def stopship_policy() -> Dict:
+    return {"policy": STOPSHIP_POLICY, "policy_version": policy_version(),
+            "note": "deterministic conjunction; explicitly NOT LLM-driven"}
+
+
+@router.get("/runs/{run_id}/gate")
+def run_gate(run_id: str) -> Dict:
+    doc = pipeline.load_run(run_id)
+    if doc is None:
+        raise HTTPException(404, f"unknown run {run_id}")
+    return doc["gate"]
+
+
+@router.get("/rules")
+def list_rules() -> Dict:
+    return {"ruleset_version": RULESET_VERSION,
+            "config": CONFIG,
+            "config_note": "thresholds are ILLUSTRATIVE (synthetic substrate)",
+            "rules": [{"rule_id": rid, "description": desc}
+                      for rid, desc, _fn in RULES]}
+
+
+@router.get("/docs/architecture")
+def architecture_doc() -> Dict:
+    path = Path(__file__).resolve().parents[2] / "docs" / "architecture" / \
+        "rotr-architecture.md"
+    if not path.exists():
+        raise HTTPException(404, "architecture doc missing")
+    return {"path": str(path), "markdown": path.read_text()}

sensorflow/rotr/attribution.py

@@ -0,0 +1,326 @@
+"""Per-violation causal-layer attribution.
+
+THE invariant of this module: a downstream failure NEVER auto-implies an
+upstream cause. Every causal layer gets an INDEPENDENT evidence test that
+returns SUPPORTED, RULED_OUT, or UNKNOWN; a layer is implicated only by its
+own positive evidence, never by another layer's silence. Missing evidence
+is UNKNOWN — an honest triage outcome, not a defect.
+
+The planning test is the keystone of perception/planning separation: the
+plan is re-judged by the SAME rule engine, but against the stack's OWN
+world view (mapped context, detections it actually made, intents it
+actually predicted, in its believed localization frame). A compliant plan
+rules planning out; a non-compliant plan given the stack's own information
+implicates planning regardless of what perception did.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, List, Optional
+
+from sensorflow.rotr import SOFTWARE_VERSION
+from sensorflow.rotr.models import (
+    CAUSAL_LAYERS, Actor, FailureAttribution, LayerEvidence, Provenance,
+    ROTRScenario, ROTRViolation,
+)
+from sensorflow.rotr.rules import plan_violations
+
+ATTRIBUTION_VERSION = "rotr-attr-1.0.0"
+
+# ILLUSTRATIVE evidence thresholds (synthetic substrate).
+DETECTION_RATE_RULED_OUT = 0.8
+DETECTION_RATE_SUPPORTED = 0.5
+POSITION_ERROR_SUPPORTED_M = 1.0
+LOCALIZATION_ERROR_SUPPORTED_M = 1.0
+CONTROL_POS_DEV_SUPPORTED_M = 1.5
+CONTROL_SPEED_DEV_SUPPORTED_MPS = 2.0
+
+
+def _material_actors(scenario: ROTRScenario,
+                     violation: ROTRViolation) -> List[Actor]:
+    ids = set(violation.actor_ids)
+    return [a for a in scenario.actors if a.actor_id in ids]
+
+
+# ------------------------------------------------------------ layer tests
+
+
+def _test_perception(scenario: ROTRScenario,
+                     violation: ROTRViolation) -> LayerEvidence:
+    actors = _material_actors(scenario, violation)
+    if not actors:
+        return LayerEvidence(
+            layer="perception", status="RULED_OUT", confidence=0.7,
+            evidence="rule trigger is ego/map-geometric; no actor whose "
+                     "perception could have caused it")
+    # Judge detection quality in the window that mattered: from the actor
+    # becoming relevant (within 60 m) up to the violation instant.
+    worst: Optional[LayerEvidence] = None
+    for actor in actors:
+        n_frames = n_detected = n_absent = 0
+        pos_err_sum = 0.0
+        for i, s in enumerate(actor.states):
+            e = scenario.ego[min(i, len(scenario.ego) - 1)]
+            if s.t > violation.t_end + 0.5:
+                break
+            if abs(s.x - e.x) > 60 or abs(s.y - e.y) > 30:
+                continue
+            n_frames += 1
+            if s.observation is None:
+                n_absent += 1
+            elif s.observation.detected:
+                n_detected += 1
+                if s.observation.x is not None:
+                    pos_err_sum += abs(s.observation.x - s.x) + \
+                        abs((s.observation.y or s.y) - s.y)
+        if n_frames == 0:
+            continue
+        if n_absent / n_frames > 0.5:
+            cand = LayerEvidence(
+                layer="perception", status="UNKNOWN", confidence=0.0,
+                evidence=f"{actor.actor_id}: perception telemetry missing for "
+                         f"{n_absent}/{n_frames} relevant frames")
+        else:
+            assessable = n_frames - n_absent
+            rate = n_detected / assessable if assessable else 0.0
+            mean_err = pos_err_sum / n_detected if n_detected else 0.0
+            if rate < DETECTION_RATE_SUPPORTED:
+                cand = LayerEvidence(
+                    layer="perception", status="SUPPORTED", confidence=0.9,
+                    evidence=f"{actor.actor_id}: missed detection — GT visible "
+                             f"{assessable} frames, detected {n_detected} "
+                             f"(rate {rate:.2f})")
+            elif mean_err > POSITION_ERROR_SUPPORTED_M:
+                cand = LayerEvidence(
+                    layer="perception", status="SUPPORTED", confidence=0.75,
+                    evidence=f"{actor.actor_id}: detected but mislocalized "
+                             f"(mean |err| {mean_err:.2f} m)")
+            elif rate >= DETECTION_RATE_RULED_OUT:
+                cand = LayerEvidence(
+                    layer="perception", status="RULED_OUT", confidence=0.9,
+                    evidence=f"{actor.actor_id}: detected {n_detected}/"
+                             f"{assessable} relevant frames (rate {rate:.2f}, "
+                             f"mean |err| {mean_err:.2f} m) — GT-vs-detected "
+                             f"diff is clean")
+            else:
+                cand = LayerEvidence(
+                    layer="perception", status="SUPPORTED", confidence=0.6,
+                    evidence=f"{actor.actor_id}: degraded detection "
+                             f"(rate {rate:.2f})")
+        worst = _worse(worst, cand)
+    return worst or LayerEvidence(layer="perception", status="UNKNOWN",
+                                  evidence="no relevant frames to assess")
+
+
+def _worse(a: Optional[LayerEvidence], b: LayerEvidence) -> LayerEvidence:
+    """Prefer SUPPORTED > UNKNOWN > RULED_OUT when merging actor evidence."""
+    if a is None:
+        return b
+    order = {"SUPPORTED": 2, "UNKNOWN": 1, "RULED_OUT": 0}
+    if order[b.status] > order[a.status]:
+        return b
+    if order[b.status] == order[a.status] and b.confidence > a.confidence:
+        return b
+    return a
+
+
+def _test_prediction(scenario: ROTRScenario,
+                     violation: ROTRViolation) -> LayerEvidence:
+    actors = _material_actors(scenario, violation)
+    if not actors:
+        return LayerEvidence(layer="prediction", status="RULED_OUT",
+                             confidence=0.7,
+                             evidence="no actor whose intent needed predicting")
+    worst: Optional[LayerEvidence] = None
+    for actor in actors:
+        if actor.predicted_intent is None:
+            cand = LayerEvidence(layer="prediction", status="UNKNOWN",
+                                 evidence=f"{actor.actor_id}: no recorded "
+                                          "intent prediction")
+        elif actor.predicted_intent != actor.intent:
+            cand = LayerEvidence(
+                layer="prediction", status="SUPPORTED", confidence=0.85,
+                evidence=f"{actor.actor_id}: GT intent {actor.intent} but "
+                         f"stack predicted {actor.predicted_intent}")
+        else:
+            cand = LayerEvidence(
+                layer="prediction", status="RULED_OUT", confidence=0.85,
+                evidence=f"{actor.actor_id}: predicted intent matches GT "
+                         f"({actor.intent})")
+        worst = _worse(worst, cand)
+    return worst
+
+
+def _test_planning(scenario: ROTRScenario,
+                   violation: ROTRViolation) -> LayerEvidence:
+    if not scenario.planned:
+        return LayerEvidence(layer="planning", status="UNKNOWN",
+                             evidence="no planned trajectory recorded")
+    hits = plan_violations(scenario)
+    same_rule = [h for h in hits if h["rule_id"] == violation.rule_id]
+    if same_rule:
+        return LayerEvidence(
+            layer="planning", status="SUPPORTED", confidence=0.85,
+            evidence=f"plan violates {violation.rule_id} given the stack's "
+                     f"OWN world view (detections + predicted intents + "
+                     f"believed frame): {same_rule[0]['evidence']}")
+    if hits:
+        return LayerEvidence(
+            layer="planning", status="SUPPORTED", confidence=0.6,
+            evidence="plan violates a different rule given the stack's own "
+                     f"view: {hits[0]['rule_id']}")
+    return LayerEvidence(
+        layer="planning", status="RULED_OUT", confidence=0.85,
+        evidence="plan is rule-compliant given the stack's own world view — "
+                 "the plan was reasonable for what the stack knew")
+
+
+def _test_localization(scenario: ROTRScenario,
+                       violation: ROTRViolation) -> LayerEvidence:
+    errs = []
+    for e in scenario.ego:
+        if e.believed_x is None or e.believed_y is None:
+            return LayerEvidence(layer="localization", status="UNKNOWN",
+                                 evidence="no believed-pose telemetry")
+        errs.append(abs(e.believed_x - e.x) + abs(e.believed_y - e.y))
+    max_err = max(errs) if errs else 0.0
+    lane_mismatch = any(e.believed_lane_id != e.lane_id for e in scenario.ego
+                        if e.lane_id and e.believed_lane_id)
+    if max_err > LOCALIZATION_ERROR_SUPPORTED_M or lane_mismatch:
+        return LayerEvidence(
+            layer="localization", status="SUPPORTED", confidence=0.9,
+            evidence=f"believed-vs-true pose error up to {max_err:.2f} m"
+                     + ("; believed lane differs from true lane"
+                        if lane_mismatch else ""))
+    return LayerEvidence(layer="localization", status="RULED_OUT",
+                         confidence=0.9,
+                         evidence=f"pose error <= {max_err:.2f} m; lane "
+                                  "association consistent")
+
+
+def _test_map(scenario: ROTRScenario,
+              violation: ROTRViolation) -> LayerEvidence:
+    m, a = scenario.map_context, scenario.actual_context
+    diffs = []
+    if m.control != a.control:
+        diffs.append(f"control mapped={m.control!r} actual={a.control!r}")
+    if m.signal_state_for_ego != a.signal_state_for_ego:
+        diffs.append(f"signal mapped={m.signal_state_for_ego!r} "
+                     f"actual={a.signal_state_for_ego!r}")
+    a_lanes = {l.lane_id: l for l in a.lanes}
+    for lane in m.lanes:
+        al = a_lanes.get(lane.lane_id)
+        if al is None:
+            diffs.append(f"lane {lane.lane_id} not in actual geometry")
+            continue
+        if lane.restricted_to != al.restricted_to:
+            diffs.append(f"lane {lane.lane_id} restriction mapped="
+                         f"{lane.restricted_to!r} actual={al.restricted_to!r}")
+        if set(lane.permitted_maneuvers) != set(al.permitted_maneuvers):
+            diffs.append(f"lane {lane.lane_id} permitted maneuvers differ")
+    if diffs:
+        return LayerEvidence(layer="map", status="SUPPORTED", confidence=0.9,
+                             evidence="map-vs-actual mismatch: " + "; ".join(diffs))
+    return LayerEvidence(layer="map", status="RULED_OUT", confidence=0.9,
+                         evidence="mapped context matches as-built geometry")
+
+
+def _test_control(scenario: ROTRScenario,
+                  violation: ROTRViolation) -> LayerEvidence:
+    if not scenario.planned:
+        return LayerEvidence(layer="control", status="UNKNOWN",
+                             evidence="no planned trajectory to compare "
+                                      "execution against")
+    n = min(len(scenario.planned), len(scenario.ego))
+    max_pos = max_spd = 0.0
+    for k in range(n):
+        p, e = scenario.planned[k], scenario.ego[k]
+        max_pos = max(max_pos, abs(p.x - e.x) + abs(p.y - e.y))
+        max_spd = max(max_spd, abs(p.v - e.speed))
+    if max_pos > CONTROL_POS_DEV_SUPPORTED_M or \
+            max_spd > CONTROL_SPEED_DEV_SUPPORTED_MPS:
+        return LayerEvidence(
+            layer="control", status="SUPPORTED", confidence=0.85,
+            evidence=f"executed trajectory deviates from plan: max position "
+                     f"dev {max_pos:.2f} m, max speed dev {max_spd:.2f} m/s")
+    return LayerEvidence(
+        layer="control", status="RULED_OUT", confidence=0.85,
+        evidence=f"execution tracks the plan (max pos dev {max_pos:.2f} m, "
+                 f"max speed dev {max_spd:.2f} m/s)")
+
+
+def _test_data_label(scenario: ROTRScenario,
+                     violation: ROTRViolation) -> LayerEvidence:
+    problems = []
+    ts = [e.t for e in scenario.ego]
+    if any(t2 <= t1 for t1, t2 in zip(ts, ts[1:])):
+        problems.append("ego timestamps not strictly increasing")
+    for a in scenario.actors:
+        if len(a.states) != len(scenario.ego):
+            problems.append(f"{a.actor_id}: state array length mismatch")
+    if problems:
+        return LayerEvidence(layer="data_label", status="SUPPORTED",
+                             confidence=0.95,
+                             evidence="record integrity failed: "
+                                      + "; ".join(problems))
+    return LayerEvidence(layer="data_label", status="RULED_OUT",
+                         confidence=0.9,
+                         evidence="record integrity checks pass "
+                                  "(monotone timestamps, consistent arrays)")
+
+
+def _test_policy_rule(others: Dict[str, LayerEvidence]) -> LayerEvidence:
+    statuses = {le.status for le in others.values()}
+    if statuses == {"RULED_OUT"}:
+        return LayerEvidence(
+            layer="policy_rule", status="SUPPORTED", confidence=0.4,
+            evidence="every stack layer independently RULED_OUT — the rule "
+                     "itself needs HITL review (possible rule misfire)")
+    if "SUPPORTED" in statuses:
+        return LayerEvidence(layer="policy_rule", status="RULED_OUT",
+                             confidence=0.7,
+                             evidence="a stack layer has positive evidence; "
+                                      "no indication the rule misfired")
+    return LayerEvidence(layer="policy_rule", status="UNKNOWN",
+                         evidence="stack evidence incomplete; rule quality "
+                                  "cannot be judged")
+
+
+# ------------------------------------------------------------ entry point
+
+
+def attribute(scenario: ROTRScenario,
+              violation: ROTRViolation) -> FailureAttribution:
+    layers: Dict[str, LayerEvidence] = {
+        "perception": _test_perception(scenario, violation),
+        "prediction": _test_prediction(scenario, violation),
+        "planning": _test_planning(scenario, violation),
+        "localization": _test_localization(scenario, violation),
+        "map": _test_map(scenario, violation),
+        "control": _test_control(scenario, violation),
+        "data_label": _test_data_label(scenario, violation),
+    }
+    layers["policy_rule"] = _test_policy_rule(layers)
+
+    supported = [(name, le) for name, le in layers.items()
+                 if le.status == "SUPPORTED"]
+    primary = None
+    if supported:
+        supported.sort(key=lambda kv: (-kv[1].confidence,
+                                       CAUSAL_LAYERS.index(kv[0])))
+        primary = supported[0][0]
+    note = ("no layer has positive evidence — triaged to HITL, NOT "
+            "defaulted to perception" if primary is None else
+            f"primary={primary} by highest-confidence positive evidence; "
+            f"{len(supported)} layer(s) SUPPORTED")
+
+    return FailureAttribution(
+        violation_id=violation.violation_id,
+        scenario_id=scenario.scenario_id,
+        layers=layers, primary_layer=primary, note=note,
+        provenance=Provenance(
+            scenario_id=scenario.scenario_id,
+            dataset_version=scenario.provenance.dataset_version,
+            model_version=scenario.provenance.model_version,
+            software_version=f"{SOFTWARE_VERSION}/{ATTRIBUTION_VERSION}",
+            source=scenario.provenance.source))

sensorflow/rotr/consequence.py

@@ -0,0 +1,315 @@
+"""Behavioral-consequence classification via counterfactual replay.
+
+For each violation we construct the trajectory the vehicle WOULD have
+driven with the attributed layer's inputs corrected, and diff it against
+the observed trajectory. The corrected trajectory is produced per primary
+layer:
+
+* control            -> the recorded plan itself (the plan was compliant;
+                        correct the actuation): engine "planned-trajectory".
+* perception /       -> re-plan with the corrected world view (ground-truth
+  prediction            actors): IDM-style replay. REUSE: the planner is
+                        nextgen.closedloop.plan_acceleration when importable
+                        (nextgen is an in-progress workstream — guarded
+                        import); otherwise a self-contained IDM fallback
+                        reusing sensorflow.safety.ssam_ext math (landed).
+* yield/signal/stop  -> compliant-envelope stop-at-line trajectory
+  rules attributed      (a rule-compliant stack brakes to the line; the
+  to planning/map/      synthetic stop is labeled engine
+  localization/data     "compliant-envelope").
+* lane-keeping rules -> stay-in-lane replay (target y = initial y).
+
+Classes (thresholds ILLUSTRATIVE): NO_MATERIAL_CONSEQUENCE <
+DEGRADED_COMFORT < PLANNER_INTERVENTION < SAFETY_CRITICAL, with measured
+trajectory divergence, TTC, PET, min clearance, stopping distance, braking
+intensity and lateral deviation on BOTH branches. Surrogate caveat: these
+measures prioritize triage; they never gate a release alone.
+"""
+
+from __future__ import annotations
+
+import math
+from typing import Dict, List, Optional
+
+import numpy as np
+
+# REUSE (landed, read-only): SSAM surrogate-safety math.
+from sensorflow.safety.ssam_ext import (
+    projected_ttc, rect_gap, rects_overlap, zone_grid_pet,
+)
+
+from sensorflow.rotr import SOFTWARE_VERSION
+from sensorflow.rotr.models import (
+    CounterfactualScenario, FailureAttribution, PlannerEvaluation, Provenance,
+    ROTRScenario, ROTRViolation, SafetyAssessment,
+)
+from sensorflow.rotr.scenes import DT, V_NOM
+
+CONSEQUENCE_VERSION = "rotr-consequence-1.0.0"
+
+EGO_DIMS = (4.5, 1.9)
+ACTUATION_TAU_S = 0.25
+
+# REUSE nextgen's planner when importable (in-progress workstream: guarded).
+try:
+    from sensorflow.nextgen.closedloop import plan_acceleration as _ng_plan
+    _PLANNER, PLANNER_ENGINE = _ng_plan, "nextgen.plan_acceleration"
+except Exception:                                    # pragma: no cover
+    _PLANNER, PLANNER_ENGINE = None, "rotr-idm-fallback"
+
+STOP_RULES = {"R-YIELD-PED-01", "R-INT-CONFLICT-04", "R-STOP-06"}
+LANE_RULES = {"R-PATH-RESTRICT-02", "R-LANE-MANEUVER-03", "R-MERGE-GAP-05"}
+
+
+# ------------------------------------------------------------ fallback planner
+
+
+def _fallback_plan(ego: Dict, detections: List[Dict]) -> Dict:
+    """Self-contained IDM-style longitudinal command + projected-TTC braking
+    (mirrors the reused planner's contract; ssam_ext math for conflicts)."""
+    v = ego["v"]
+    a_cmd = 2.5 * (1.0 - (v / max(V_NOM, 0.1)) ** 4)
+    intervention = False
+    ego_state = {"x": ego["x"], "y": ego["y"], "speed": v,
+                 "heading": ego["heading"]}
+    for det in detections:
+        rel_x = det["x"] - ego["x"]
+        if abs(det["y"] - ego["y"]) <= 1.6 and rel_x > 0:
+            gap = max(rel_x - 0.5 * (EGO_DIMS[0] + det["dims"][0]), 0.5)
+            dv = v - det["vx"]
+            s_star = 2.0 + max(0.0, v * 1.5 + v * dv / (2 * math.sqrt(2.5 * 3.5)))
+            a_idm = 2.5 * (1.0 - (v / max(V_NOM, 0.1)) ** 4 - (s_star / gap) ** 2)
+            a_cmd = min(a_cmd, a_idm)
+        spd = math.hypot(det["vx"], det["vy"])
+        obj = {"x": det["x"], "y": det["y"], "speed": spd,
+               "heading": math.atan2(det["vy"], det["vx"]) if spd else 0.0}
+        ttc = projected_ttc(ego_state, EGO_DIMS, obj,
+                            (det["dims"][0], det["dims"][1]),
+                            look_ahead=4.0, dt=0.1)
+        if ttc is not None:
+            if ttc <= 1.5:
+                a_cmd, intervention = -8.0, True
+            elif ttc <= 4.0:
+                a_cmd = min(a_cmd, -3.5 * (4.0 - ttc) / 2.5)
+                intervention = True
+    return {"a_cmd": float(np.clip(a_cmd, -8.0, 2.5)),
+            "intervention": intervention, "reason": "fallback"}
+
+
+def _planner(ego: Dict, detections: List[Dict]) -> Dict:
+    if _PLANNER is not None:
+        return _PLANNER(ego, detections)
+    return _fallback_plan(ego, detections)
+
+
+# ------------------------------------------------------------ replay engines
+
+
+def _gt_detections(scenario: ROTRScenario, fi: int, ego_x: float) -> List[Dict]:
+    out = []
+    for a in scenario.actors:
+        s = a.states[min(fi, len(a.states) - 1)]
+        if s.x - ego_x < -5.0 or math.hypot(s.x - ego_x, s.y) > 75.0:
+            continue
+        out.append({"instance_id": a.actor_id, "class_name": a.class_name,
+                    "x": s.x, "y": s.y, "vx": s.vx, "vy": s.vy,
+                    "dims": a.dims, "yaw": s.yaw})
+    return out
+
+
+def _idm_replay(scenario: ROTRScenario) -> Dict:
+    """Re-plan with the CORRECTED world view (ground-truth actors)."""
+    e0 = scenario.ego[0]
+    n = len(scenario.ego)
+    ego = {"x": e0.x, "y": e0.y, "v": e0.speed, "a": 0.0, "heading": 0.0}
+    traj, intervention = [], False
+    for step in range(n):
+        t = step * DT
+        plan = _planner(ego, _gt_detections(scenario, step, ego["x"]))
+        intervention = intervention or bool(plan.get("intervention"))
+        alpha = DT / (DT + ACTUATION_TAU_S)
+        ego["a"] += alpha * (plan["a_cmd"] - ego["a"])
+        ego["v"] = max(0.0, ego["v"] + ego["a"] * DT)
+        ego["x"] += ego["v"] * DT
+        traj.append({"t": round(t, 2), "x": round(ego["x"], 3),
+                     "y": round(ego["y"], 3), "v": round(ego["v"], 3),
+                     "a": round(ego["a"], 3)})
+    return {"traj": traj, "intervention": intervention,
+            "engine": PLANNER_ENGINE}
+
+
+def _stop_at_line_replay(scenario: ROTRScenario) -> Dict:
+    """Compliant-envelope reference: brake to the stop line and hold."""
+    ctx = scenario.actual_context
+    line = ctx.stop_line_x if ctx.stop_line_x is not None else 38.0
+    e0 = scenario.ego[0]
+    v0 = max(e0.speed, 0.1)
+    decel = 3.0
+    x_brake = (line - 0.5) - v0 * v0 / (2 * decel)
+    x, v, traj = e0.x, e0.speed, []
+    for step in range(len(scenario.ego)):
+        t = step * DT
+        a = -decel if (x >= x_brake and v > 0) else 0.0
+        traj.append({"t": round(t, 2), "x": round(x, 3), "y": round(e0.y, 3),
+                     "v": round(v, 3), "a": round(a, 3)})
+        v = max(0.0, v + a * DT)
+        x += v * DT
+    return {"traj": traj, "intervention": True, "engine": "compliant-envelope"}
+
+
+def _planned_replay(scenario: ROTRScenario) -> Dict:
+    traj = [{"t": round(p.t, 2), "x": round(p.x, 3), "y": round(p.y, 3),
+             "v": round(p.v, 3), "a": 0.0} for p in scenario.planned]
+    return {"traj": traj, "intervention": False, "engine": "planned-trajectory"}
+
+
+def corrected_replay(scenario: ROTRScenario, violation: ROTRViolation,
+                     attribution: FailureAttribution) -> Dict:
+    primary = attribution.primary_layer
+    if primary == "control" and scenario.planned:
+        return {**_planned_replay(scenario), "corrected_layers": ["control"]}
+    if primary in ("perception", "prediction"):
+        return {**_idm_replay(scenario), "corrected_layers": [primary]}
+    if violation.rule_id in STOP_RULES:
+        return {**_stop_at_line_replay(scenario),
+                "corrected_layers": [primary or "unknown"]}
+    if violation.rule_id in LANE_RULES:
+        return {**_idm_replay(scenario),
+                "corrected_layers": [primary or "unknown"]}
+    return {**_idm_replay(scenario), "corrected_layers": [primary or "unknown"]}
+
+
+# ------------------------------------------------------------ measurement
+
+
+def _assess(traj: List[Dict], scenario: ROTRScenario) -> SafetyAssessment:
+    min_ttc = min_gap = None
+    collision = False
+    max_brake = max_lat = 0.0
+    y0 = traj[0]["y"] if traj else 0.0
+    brake_x: Optional[float] = None
+    stop_dist: Optional[float] = None
+    for k, p in enumerate(traj):
+        a = p.get("a", 0.0)
+        v = p.get("v", p.get("speed", 0.0))
+        max_brake = max(max_brake, -a)
+        max_lat = max(max_lat, abs(p["y"] - y0))
+        if a < -2.0 and brake_x is None:
+            brake_x = p["x"]
+        if brake_x is not None and stop_dist is None and v < 0.1:
+            stop_dist = p["x"] - brake_x
+        ego_state = {"x": p["x"], "y": p["y"], "speed": v, "heading": 0.0}
+        for actor in scenario.actors:
+            s = actor.states[min(k, len(actor.states) - 1)]
+            if abs(s.x - p["x"]) > 60 or abs(s.y - p["y"]) > 25:
+                continue
+            gap = rect_gap((p["x"], p["y"]), 0.0, *EGO_DIMS,
+                           (s.x, s.y), s.yaw, actor.dims[0], actor.dims[1])
+            if min_gap is None or gap < min_gap:
+                min_gap = gap
+            obj = {"x": s.x, "y": s.y, "speed": math.hypot(s.vx, s.vy),
+                   "heading": s.yaw}
+            ttc = projected_ttc(ego_state, EGO_DIMS, obj,
+                                (actor.dims[0], actor.dims[1]),
+                                look_ahead=4.0, dt=0.1)
+            if ttc is not None and (min_ttc is None or ttc < min_ttc):
+                min_ttc = ttc
+            if rects_overlap((p["x"], p["y"]), 0.0, *EGO_DIMS,
+                             (s.x, s.y), s.yaw, actor.dims[0], actor.dims[1]):
+                collision = True
+    pet = _min_pet(traj, scenario)
+    return SafetyAssessment(
+        min_ttc_s=_r(min_ttc), pet_s=_r(pet), min_clearance_m=_r(min_gap),
+        stopping_distance_m=_r(stop_dist), max_braking_mps2=round(max_brake, 3),
+        max_lateral_deviation_m=round(max_lat, 3), collision=collision)
+
+
+def _min_pet(traj: List[Dict], scenario: ROTRScenario) -> Optional[float]:
+    ego_traj = {"vehicle_id": "ego", "vehicle_type": "car",
+                "length": EGO_DIMS[0], "width": EGO_DIMS[1],
+                "states": [{"t": p["t"], "x": p["x"], "y": p["y"],
+                            "speed": p.get("v", p.get("speed", 0.0)),
+                            "heading": 0.0} for p in traj]}
+    best = None
+    for actor in scenario.actors:
+        other = {"vehicle_id": actor.actor_id, "vehicle_type": actor.class_name,
+                 "length": actor.dims[0], "width": actor.dims[1],
+                 "states": [{"t": s.t, "x": s.x, "y": s.y,
+                             "speed": math.hypot(s.vx, s.vy),
+                             "heading": s.yaw} for s in actor.states]}
+        pet = zone_grid_pet(ego_traj, other, cell=2.0)
+        if pet is not None and (best is None or pet < best):
+            best = pet
+    return best
+
+
+def _r(v: Optional[float]) -> Optional[float]:
+    return None if v is None else round(float(v), 3)
+
+
+# ------------------------------------------------------------ classification
+# ILLUSTRATIVE thresholds (synthetic substrate).
+
+SC_VRU_CLEARANCE_M = 1.5
+SC_CLEARANCE_M = 0.5
+DEGRADED_DIVERGENCE_M = 2.0
+DEGRADED_BRAKING_MPS2 = 1.5
+
+
+def classify(scenario: ROTRScenario, violation: ROTRViolation,
+             attribution: FailureAttribution) -> CounterfactualScenario:
+    replay = corrected_replay(scenario, violation, attribution)
+    observed = [{"t": e.t, "x": e.x, "y": e.y, "v": e.speed, "a": e.accel}
+                for e in scenario.ego]
+    corrected = replay["traj"]
+
+    n = min(len(observed), len(corrected))
+    max_pos = max_spd = 0.0
+    for k in range(n):
+        o, c = observed[k], corrected[k]
+        max_pos = max(max_pos, math.hypot(o["x"] - c["x"], o["y"] - c["y"]))
+        max_spd = max(max_spd, abs(o["v"] - c["v"]))
+
+    obs = _assess(observed, scenario)
+    corr = _assess(corrected, scenario)
+
+    vru = any(a.class_name in ("pedestrian", "cyclist")
+              for a in scenario.actors if a.actor_id in set(violation.actor_ids))
+    # Observed danger classifies on its own; whether the corrected branch
+    # avoided it is reported as evidence, never used to downgrade severity
+    # (the replay planner is simplified and must not launder real exposure).
+    if (obs.collision
+            or (vru and obs.min_clearance_m is not None
+                and obs.min_clearance_m < SC_VRU_CLEARANCE_M)
+            or (obs.min_clearance_m is not None
+                and obs.min_clearance_m < SC_CLEARANCE_M)):
+        cls = "SAFETY_CRITICAL"
+    elif replay["intervention"] and max_pos > DEGRADED_DIVERGENCE_M:
+        cls = "PLANNER_INTERVENTION"
+    elif max_pos > DEGRADED_DIVERGENCE_M or \
+            corr.max_braking_mps2 > DEGRADED_BRAKING_MPS2:
+        cls = "DEGRADED_COMFORT"
+    else:
+        cls = "NO_MATERIAL_CONSEQUENCE"
+
+    return CounterfactualScenario(
+        counterfactual_id=f"{violation.violation_id}-cf",
+        violation_id=violation.violation_id,
+        scenario_id=scenario.scenario_id,
+        corrected_layers=replay["corrected_layers"],
+        consequence_class=cls,
+        planner_evaluation=PlannerEvaluation(
+            engine=replay["engine"],
+            observed_trajectory=observed,
+            corrected_trajectory=corrected,
+            max_position_divergence_m=round(max_pos, 3),
+            max_speed_divergence_mps=round(max_spd, 3),
+            corrected_max_braking_mps2=corr.max_braking_mps2,
+            corrected_intervention=bool(replay["intervention"])),
+        observed_safety=obs, corrected_safety=corr,
+        provenance=Provenance(
+            scenario_id=scenario.scenario_id,
+            dataset_version=scenario.provenance.dataset_version,
+            model_version=scenario.provenance.model_version,
+            software_version=f"{SOFTWARE_VERSION}/{CONSEQUENCE_VERSION}",
+            source="COUNTERFACTUAL"))

sensorflow/rotr/flywheel.py

@@ -0,0 +1,230 @@
+"""Evaluation flywheel: confirmed violation -> failure signature -> cluster
+-> HITL validation -> regression-test artifact, under dataset-role
+governance with a contamination guard.
+
+Roles are immutable once assigned; members of PROTECTED roles (REGRESSION,
+LAUNCH) are NEVER training-eligible without a recorded governance override
+— semantics mirrored from raremine.lineage (ContaminationError subclasses
+raremine's LeakageError when importable so callers can catch either).
+`sensorflow.studio2`'s registry is used when it lands (guarded import,
+re-checked at integration time); until then the local implementation
+persists to runs/rotr/.
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Dict, List, Optional
+
+from sensorflow.rotr import SOFTWARE_VERSION
+from sensorflow.rotr import store
+from sensorflow.rotr.models import (
+    HITLReview, PROTECTED_ROLES, Provenance, TrainingCandidate, now_iso,
+)
+
+FLYWHEEL_VERSION = "rotr-flywheel-1.0.0"
+
+# Mirror raremine's guard exception when importable (landed package,
+# read-only reuse) so platform callers can catch one exception type.
+try:
+    from sensorflow.raremine.lineage import LeakageError as _LeakageBase
+except Exception:                                     # pragma: no cover
+    _LeakageBase = RuntimeError
+
+
+class ContaminationError(_LeakageBase):
+    """Raised when a protected evaluation member would leak into training."""
+
+
+# Registry backend: studio2 when it lands, else local (recorded on artifacts).
+# The local store is ALWAYS written (the guard never depends on the
+# concurrent workstream); studio2 registration is a best-effort mirror.
+try:
+    from sensorflow.studio2.registry import get_registry as _studio2_get_registry
+    REGISTRY_BACKEND = "studio2+local"
+except Exception:
+    _studio2_get_registry = None
+    REGISTRY_BACKEND = "local"
+
+
+def _mirror_to_studio2(candidate: "TrainingCandidate") -> Optional[str]:
+    """Best-effort mirror of the regression artifact into the studio2
+    entity registry (guarded: studio2 is a concurrent workstream)."""
+    if _studio2_get_registry is None:
+        return None
+    try:
+        entity = _studio2_get_registry().register_dataset(
+            name=f"rotr-regression-{candidate.candidate_id}",
+            role="REGRESSION",
+            provenance=candidate.provenance.model_dump(),
+            meta={"violation_id": candidate.violation_id,
+                  "cluster_id": candidate.cluster_id,
+                  "run_id": candidate.run_id, "source": "sensorflow.rotr"},
+            actor="rotr-flywheel")
+        return entity.get("entity_id")
+    except Exception:
+        return None
+
+
+def _audit(event: str, subject: str, actor: str, detail: str = "") -> None:
+    doc = store.read_json("flywheel", "audit.json") or {"events": []}
+    doc["events"].append({"at": now_iso(), "event": event, "subject": subject,
+                          "actor": actor, "detail": detail})
+    store.write_json(doc, "flywheel", "audit.json")
+
+
+# ------------------------------------------------------------ HITL queue
+
+
+def build_queue(run_id: str, violations: List[Dict],
+                clusters: List[Dict]) -> List[HITLReview]:
+    vid_to_cluster = {}
+    for c in clusters:
+        for vid in c["member_violation_ids"]:
+            vid_to_cluster[vid] = c["cluster_id"]
+    reviews = []
+    for v in violations:
+        rid = f"rev-{v['violation_id']}"
+        reviews.append(HITLReview(
+            review_id=rid, run_id=run_id, violation_id=v["violation_id"],
+            cluster_id=vid_to_cluster.get(v["violation_id"]),
+            status="PENDING",
+            provenance=Provenance(
+                scenario_id=v.get("scenario_id"),
+                software_version=f"{SOFTWARE_VERSION}/{FLYWHEEL_VERSION}",
+                source="SYNTHETIC")))
+    store.write_json({"run_id": run_id,
+                      "reviews": [r.model_dump() for r in reviews]},
+                     "flywheel", f"queue-{run_id}.json")
+    return reviews
+
+
+def get_queue(run_id: str) -> List[Dict]:
+    doc = store.read_json("flywheel", f"queue-{run_id}.json")
+    return doc["reviews"] if doc else []
+
+
+def act(run_id: str, review_id: str, action: str, actor: str,
+        notes: str = "") -> Dict:
+    """VALIDATE turns a confirmed violation into a regression-test artifact
+    (protected role, not training-eligible). REJECT closes the review."""
+    doc = store.read_json("flywheel", f"queue-{run_id}.json")
+    if not doc:
+        raise KeyError(f"no HITL queue for run {run_id}")
+    review = next((r for r in doc["reviews"] if r["review_id"] == review_id),
+                  None)
+    if review is None:
+        raise KeyError(f"unknown review {review_id}")
+    if review["status"] != "PENDING":
+        raise ValueError(f"review {review_id} already {review['status']}")
+
+    action = action.upper()
+    if action not in ("VALIDATE", "REJECT"):
+        raise ValueError("action must be VALIDATE or REJECT")
+    review["status"] = "VALIDATED" if action == "VALIDATE" else "REJECTED"
+    review["action"] = action
+    review["actor"] = actor
+    review["notes"] = notes
+    store.write_json(doc, "flywheel", f"queue-{run_id}.json")
+    _audit(f"hitl_{action.lower()}", review_id, actor, notes)
+
+    candidate = None
+    if action == "VALIDATE":
+        candidate = _create_candidate(run_id, review, actor)
+        _append_suite_member(run_id, review, candidate)
+    return {"review": review,
+            "candidate": candidate.model_dump() if candidate else None}
+
+
+def _create_candidate(run_id: str, review: Dict, actor: str) -> TrainingCandidate:
+    cid = f"tc-{uuid.uuid4().hex[:10]}"
+    cand = TrainingCandidate(
+        candidate_id=cid, run_id=run_id,
+        violation_id=review["violation_id"],
+        cluster_id=review.get("cluster_id"),
+        dataset_role="REGRESSION", training_eligible=False,
+        guard_state="PROTECTED",
+        provenance=Provenance(
+            software_version=f"{SOFTWARE_VERSION}/{FLYWHEEL_VERSION}",
+            source="SYNTHETIC"))
+    doc = cand.model_dump()
+    doc["studio2_entity_id"] = _mirror_to_studio2(cand)
+    doc["registry_backend"] = REGISTRY_BACKEND
+    store.write_json(doc, "flywheel", "candidates", f"{cid}.json")
+    _audit("candidate_created", cid, actor,
+           f"role=REGRESSION (protected) violation={review['violation_id']} "
+           f"registry_backend={REGISTRY_BACKEND}")
+    return cand
+
+
+def _append_suite_member(run_id: str, review: Dict,
+                         candidate: TrainingCandidate) -> None:
+    suite = store.read_json("flywheel", "regression-suite.json") or {
+        "suite_id": "rotr-regression-suite", "members": [],
+        "registry_backend": REGISTRY_BACKEND}
+    suite["members"].append({
+        "candidate_id": candidate.candidate_id,
+        "violation_id": review["violation_id"],
+        "cluster_id": review.get("cluster_id"),
+        "run_id": run_id, "role": "REGRESSION",
+        "added_at": now_iso(), "added_by": review.get("actor"),
+        "immutable_role_note": "REGRESSION members re-run against every "
+                               "candidate; never training-eligible without "
+                               "a recorded governance override"})
+    store.write_json(suite, "flywheel", "regression-suite.json")
+
+
+def get_suite() -> Dict:
+    return store.read_json("flywheel", "regression-suite.json") or {
+        "suite_id": "rotr-regression-suite", "members": [],
+        "registry_backend": REGISTRY_BACKEND}
+
+
+# ------------------------------------------------------------ contamination guard
+
+
+def list_candidates() -> List[Dict]:
+    out = []
+    for name in store.list_dir("flywheel", "candidates"):
+        doc = store.read_json("flywheel", "candidates", name)
+        if doc:
+            out.append(doc)
+    return out
+
+
+def get_candidate(candidate_id: str) -> Optional[Dict]:
+    return store.read_json("flywheel", "candidates", f"{candidate_id}.json")
+
+
+def promote_to_training(candidate_id: str, actor: str) -> Dict:
+    """Refused for protected roles unless a governance override is recorded
+    (raremine guard semantics)."""
+    doc = get_candidate(candidate_id)
+    if doc is None:
+        raise KeyError(f"unknown candidate {candidate_id}")
+    if doc["dataset_role"] in PROTECTED_ROLES and not doc.get("override"):
+        _audit("promotion_refused", candidate_id, actor,
+               f"protected role {doc['dataset_role']} without override")
+        raise ContaminationError(
+            f"candidate {candidate_id} holds protected role "
+            f"{doc['dataset_role']}: training promotion refused without a "
+            "recorded governance override")
+    doc["training_eligible"] = True
+    doc["guard_state"] = "OVERRIDDEN" if doc.get("override") else "ELIGIBLE"
+    store.write_json(doc, "flywheel", "candidates", f"{candidate_id}.json")
+    _audit("promoted_to_training", candidate_id, actor,
+           f"guard_state={doc['guard_state']}")
+    return doc
+
+
+def governance_override(candidate_id: str, actor: str, reason: str) -> Dict:
+    if not reason.strip():
+        raise ValueError("a governance override requires a recorded reason")
+    doc = get_candidate(candidate_id)
+    if doc is None:
+        raise KeyError(f"unknown candidate {candidate_id}")
+    doc["override"] = {"actor": actor, "reason": reason, "at": now_iso()}
+    doc["guard_state"] = "OVERRIDDEN"
+    store.write_json(doc, "flywheel", "candidates", f"{candidate_id}.json")
+    _audit("governance_override", candidate_id, actor, reason)
+    return doc

sensorflow/rotr/metrics.py

@@ -0,0 +1,313 @@
+"""ROTR metric hierarchy + baseline-vs-candidate regression.
+
+Metrics (architecture doc §6):
+* ROTR recall / precision proxy (false-accusation rate on planted
+  NON-violations) — the scenario bank supplies a true denominator.
+* Safety-critical ROTR recall — EXPOSURE-DERIVED weights: the weight of a
+  stratum is its measured fraction of SAFETY_CRITICAL consequence outcomes
+  on the evaluated bank, normalized by the overall fraction (documented
+  calibration, monotone in measured harm — NOT arbitrary class
+  multipliers). ILLUSTRATIVE on synthetic data by construction.
+* Behavioral Consequence Rate (BCR), Critical Failure Rate (CFR, with a
+  Wilson interval — REUSE rca.stats.wilson_ci).
+* Regression: claims are delegated to seqeval's sequential engine
+  (REUSE seqeval.evaluate_regression over a megaeval population, rare-event
+  cluster handling included) and forced into the SIX-outcome distinction:
+  observed / statistically significant / practically significant /
+  safety-critical / insufficient evidence / distribution-driven (the last
+  via cohort-mix divergence, megaeval-shift + rca comparison-validity
+  concepts).
+"""
+
+from __future__ import annotations
+
+from typing import Dict, List, Optional
+
+# REUSE (landed, read-only).
+from sensorflow.rca.stats import js_divergence_categorical, wilson_ci
+
+from sensorflow.rotr import SOFTWARE_VERSION
+from sensorflow.rotr.models import Provenance, RegressionResult
+
+CALIBRATION_VERSION = "rotr-weights-1.0.0"
+
+CONSEQUENCE_RANK = {"NO_MATERIAL_CONSEQUENCE": 0, "DEGRADED_COMFORT": 1,
+                    "PLANNER_INTERVENTION": 2, "SAFETY_CRITICAL": 3}
+
+# ILLUSTRATIVE practical-significance thresholds.
+PRACTICAL_ROTR_RECALL_PP = 0.05
+PRACTICAL_SC_RECALL_PP = 0.02
+DISTRIBUTION_JS_THRESHOLD = 0.05
+
+
+# ------------------------------------------------------------ run metrics
+
+
+def _stratum(scenario_summary: Dict) -> str:
+    return f"{scenario_summary['kind']}|{scenario_summary['vulnerability']}"
+
+
+def compute_metrics(scenarios: List[Dict], violations: List[Dict],
+                    consequences: Dict[str, Dict]) -> Dict:
+    """scenarios: summaries with planted truth; violations: detected records;
+    consequences: violation_id -> {consequence_class}."""
+    by_scenario: Dict[str, List[Dict]] = {}
+    for v in violations:
+        by_scenario.setdefault(v["scenario_id"], []).append(v)
+
+    committed = [s for s in scenarios if s["committed"]]
+    negatives = [s for s in scenarios if not s["committed"]]
+
+    def _detected(s: Dict) -> bool:
+        return any(v["rule_id"] == s["expected_rule_id"]
+                   for v in by_scenario.get(s["scenario_id"], []))
+
+    n_detected = sum(1 for s in committed if _detected(s))
+    rotr_recall = n_detected / len(committed) if committed else None
+
+    false_acc = sum(1 for s in negatives
+                    if by_scenario.get(s["scenario_id"]))
+    false_accusation_rate = false_acc / len(negatives) if negatives else None
+
+    # --- exposure-derived weight calibration (documented, monotone) ------
+    strata: Dict[str, Dict] = {}
+    for s in committed:
+        st = strata.setdefault(_stratum(s), {"n": 0, "detected": 0, "sc": 0})
+        st["n"] += 1
+        if _detected(s):
+            st["detected"] += 1
+        cons = [consequences.get(v["violation_id"], {}).get("consequence_class")
+                for v in by_scenario.get(s["scenario_id"], [])]
+        if "SAFETY_CRITICAL" in cons or s.get("planted_sc_hint"):
+            st["sc"] += 1
+    total_n = sum(st["n"] for st in strata.values()) or 1
+    total_sc = sum(st["sc"] for st in strata.values())
+    overall_harm = total_sc / total_n if total_n else 0.0
+    calibration = {}
+    for name, st in sorted(strata.items()):
+        harm = st["sc"] / st["n"] if st["n"] else 0.0
+        # weight = measured harm exposure relative to overall harm; floor at
+        # a small epsilon so zero-harm strata still count, never dominate.
+        weight = (harm / overall_harm) if overall_harm > 0 else 1.0
+        weight = max(weight, 0.1)
+        calibration[name] = {"n": st["n"], "harm_fraction": round(harm, 4),
+                             "weight": round(weight, 4),
+                             "recall": round(st["detected"] / st["n"], 4)
+                             if st["n"] else None}
+    sc_strata = {k: v for k, v in calibration.items() if v["harm_fraction"] > 0
+                 or k.endswith("|VRU")}
+    num = sum(v["weight"] * v["recall"] * v["n"] for v in sc_strata.values()
+              if v["recall"] is not None)
+    den = sum(v["weight"] * v["n"] for v in sc_strata.values())
+    sc_rotr_recall = (num / den) if den else None
+
+    # --- consequence-level rates ----------------------------------------
+    detected_vids = [v["violation_id"] for v in violations]
+    consequential = sum(
+        1 for vid in detected_vids
+        if CONSEQUENCE_RANK.get(
+            consequences.get(vid, {}).get("consequence_class", ""), 0) >= 1)
+    bcr = consequential / len(detected_vids) if detected_vids else None
+
+    sc_count = sum(1 for vid in detected_vids
+                   if consequences.get(vid, {}).get("consequence_class")
+                   == "SAFETY_CRITICAL")
+    n_scen = len(scenarios)
+    cfr = sc_count / n_scen if n_scen else None
+    cfr_ci = wilson_ci(sc_count, n_scen) if n_scen else (0.0, 1.0)
+
+    # --- cohort breakdowns ----------------------------------------------
+    cohorts: Dict[str, Dict] = {}
+    for s in committed:
+        for key in (f"rule:{s['expected_rule_id']}",
+                    f"visibility:{s['visibility']}",
+                    f"lighting:{s['lighting']}",
+                    f"vulnerability:{s['vulnerability']}"):
+            c = cohorts.setdefault(key, {"n": 0, "detected": 0})
+            c["n"] += 1
+            c["detected"] += 1 if _detected(s) else 0
+    for c in cohorts.values():
+        c["recall"] = round(c["detected"] / c["n"], 4) if c["n"] else None
+
+    # --- stack-behavior rates (what regresses between model versions) ----
+    # ROTR recall measures the EVALUATOR; violation rates measure the STACK:
+    # committed violations per planted opportunity, per vulnerability cohort.
+    opportunities = [s for s in scenarios if s["is_violation_opportunity"]]
+    behavior_rates: Dict[str, Dict] = {}
+    for coh in ("VRU", "NON_VRU", "ALL"):
+        opp = [s for s in opportunities
+               if coh == "ALL" or s["vulnerability"] == coh]
+        com = [s for s in opp if s["committed"]]
+        behavior_rates[coh] = {
+            "opportunities": len(opp), "committed": len(com),
+            "violation_rate": round(len(com) / len(opp), 4) if opp else None}
+
+    return {
+        "n_scenarios": n_scen,
+        "n_committed_violations": len(committed),
+        "n_planted_non_violations": len(negatives),
+        "n_detected": n_detected,
+        "n_violation_records": len(violations),
+        "rotr_recall": _r(rotr_recall),
+        "false_accusation_rate": _r(false_accusation_rate),
+        "sc_rotr_recall": _r(sc_rotr_recall),
+        "bcr": _r(bcr),
+        "cfr": _r(cfr),
+        "cfr_wilson_95": [round(cfr_ci[0], 4), round(cfr_ci[1], 4)],
+        "weight_calibration": {
+            "version": CALIBRATION_VERSION,
+            "method": "stratum weight = measured SAFETY_CRITICAL harm "
+                      "fraction / overall harm fraction (exposure-derived, "
+                      "monotone in measured harm; floored at 0.1). "
+                      "ILLUSTRATIVE: calibrated on the synthetic bank itself.",
+            "strata": calibration,
+        },
+        "cohorts": cohorts,
+        "behavior_rates": behavior_rates,
+        "surrogate_caveat": "TTC/PET are surrogate conflict measures; they "
+                            "prioritize triage and never gate alone.",
+    }
+
+
+def _r(v: Optional[float]) -> Optional[float]:
+    return None if v is None else round(float(v), 4)
+
+
+# ------------------------------------------------------------ regression
+
+
+def _seqeval_effects(baseline: Dict, candidate: Dict) -> Dict[str, float]:
+    """Translate stack-behavior deltas into seqeval stratum effects.
+
+    The regressing quantity between MODEL versions is the committed
+    violation RATE per cohort (ROTR recall measures the evaluator and is
+    invariant across stack profiles). A rate INCREASE is a performance
+    DECREASE, so the effect is the negated rate delta in probability
+    points. seqeval strata are megaeval "class|lighting" strata: the VRU
+    cohort maps onto the pedestrian strata, NON_VRU onto vehicle strata.
+    """
+    effects: Dict[str, float] = {}
+    b = baseline.get("behavior_rates", {})
+    c = candidate.get("behavior_rates", {})
+    mapping = {"VRU": "pedestrian|night", "NON_VRU": "vehicle|day",
+               "ALL": "__global__"}
+    for coh, stratum in mapping.items():
+        rb = (b.get(coh) or {}).get("violation_rate")
+        rc = (c.get(coh) or {}).get("violation_rate")
+        if rb is None or rc is None:
+            continue
+        delta = -(rc - rb)
+        if abs(delta) > 1e-9:
+            effects[stratum] = round(delta, 4)
+    return effects
+
+
+def evaluate_candidate(regression_id: str, baseline_run: Dict,
+                       candidate_run: Dict, seed: int = 11) -> RegressionResult:
+    bm, cm = baseline_run["metrics"], candidate_run["metrics"]
+    deltas = {}
+    for k in ("rotr_recall", "sc_rotr_recall", "bcr", "cfr",
+              "false_accusation_rate"):
+        if bm.get(k) is not None and cm.get(k) is not None:
+            deltas[k] = round(cm[k] - bm[k], 4)
+    for coh in ("VRU", "NON_VRU", "ALL"):
+        rb = (bm.get("behavior_rates", {}).get(coh) or {}).get("violation_rate")
+        rc = (cm.get("behavior_rates", {}).get(coh) or {}).get("violation_rate")
+        if rb is not None and rc is not None:
+            deltas[f"violation_rate_{coh}"] = round(rc - rb, 4)
+
+    # --- delegation: statistical claims go through seqeval ---------------
+    seq_payload = None
+    seq_decision = None
+    try:
+        from sensorflow.megaeval import population as pop_mod
+        from sensorflow.seqeval.controller import evaluate_regression
+
+        pop_name = "rotr-regression-pop"
+        pop_id = next((p["population_id"] for p in pop_mod.list_populations()
+                       if p.get("name") == pop_name), None)
+        if pop_id is None:
+            pop_id = pop_mod.generate_population(
+                pop_name, num_objects=60_000, seed=17)["population_id"]
+        effects = _seqeval_effects(bm, cm)
+        seq_payload = evaluate_regression(
+            pop_id,
+            {"model_version": baseline_run.get("model_version", "baseline")},
+            {"model_version": candidate_run.get("model_version", "candidate"),
+             "effects": effects},
+            policy={"target_n": 6000, "safety_floor": 800,
+                    "min_per_stratum": 120,
+                    "escalation": {"enabled": True,
+                                   "max_extra_per_stratum": 600,
+                                   "batch_objects": 300}},
+            seed=seed)
+        seq_payload["translated_effects"] = effects
+        seq_decision = seq_payload.get("decision")
+    except Exception as e:                             # honest degradation
+        seq_payload = {"error": f"seqeval delegation unavailable: {e}"}
+
+    # --- distribution-driven check (megaeval shift + rca concepts) -------
+    # Compare the OPPORTUNITY mix (bank composition), not the committed mix:
+    # committed-mix changes are model behavior, which is the signal, not a
+    # confound. Same-seed banks have identical opportunity mixes (JS = 0).
+    def _mix(run: Dict) -> List[str]:
+        return [f"{s['kind']}|{s['visibility']}|{s['lighting']}"
+                for s in run["scenario_summaries"]]
+
+    js = js_divergence_categorical(_mix(baseline_run), _mix(candidate_run)) \
+        if baseline_run.get("scenario_summaries") and \
+        candidate_run.get("scenario_summaries") else 0.0
+    distribution_driven = js > DISTRIBUTION_JS_THRESHOLD
+    dist_note = (f"cohort-mix JS divergence {js:.4f} "
+                 + ("EXCEEDS" if distribution_driven else "below")
+                 + f" {DISTRIBUTION_JS_THRESHOLD} — "
+                 + ("observed deltas may reflect a different committed-"
+                    "violation mix, not model behavior; verify before acting"
+                    if distribution_driven else
+                    "committed-violation mix comparable between runs"))
+
+    # --- six-outcome distinction -----------------------------------------
+    six = {
+        "observed_difference": any(abs(d) > 0 for d in deltas.values()),
+        "statistically_significant": seq_decision == "REGRESSION",
+        "practically_significant": (
+            abs(deltas.get("violation_rate_ALL", 0.0))
+            >= PRACTICAL_ROTR_RECALL_PP
+            or abs(deltas.get("violation_rate_VRU", 0.0))
+            >= PRACTICAL_SC_RECALL_PP
+            or abs(deltas.get("rotr_recall", 0.0)) >= PRACTICAL_ROTR_RECALL_PP),
+        "safety_critical": (deltas.get("violation_rate_VRU", 0.0)
+                            >= PRACTICAL_SC_RECALL_PP
+                            or deltas.get("sc_rotr_recall", 0.0)
+                            <= -PRACTICAL_SC_RECALL_PP
+                            or deltas.get("cfr", 0.0) > 0),
+        "insufficient_evidence": seq_decision == "INSUFFICIENT"
+        or seq_decision is None,
+        "distribution_driven": distribution_driven,
+    }
+    if six["safety_critical"] and six["statistically_significant"]:
+        primary = "SAFETY_CRITICAL_REGRESSION"
+    elif six["statistically_significant"]:
+        primary = "STATISTICALLY_SIGNIFICANT_REGRESSION"
+    elif six["insufficient_evidence"]:
+        primary = "INSUFFICIENT_EVIDENCE"
+    elif six["distribution_driven"]:
+        primary = "DISTRIBUTION_DRIVEN"
+    elif six["practically_significant"]:
+        primary = "PRACTICALLY_SIGNIFICANT_DIFFERENCE"
+    elif six["observed_difference"]:
+        primary = "OBSERVED_DIFFERENCE_ONLY"
+    else:
+        primary = "NO_DIFFERENCE"
+
+    return RegressionResult(
+        regression_id=regression_id,
+        baseline_run_id=baseline_run["run_id"],
+        candidate_run_id=candidate_run["run_id"],
+        baseline_model=baseline_run.get("model_version", ""),
+        candidate_model=candidate_run.get("model_version", ""),
+        metric_deltas=deltas, six_outcomes=six, primary_outcome=primary,
+        seqeval=seq_payload, distribution_note=dist_note,
+        provenance=Provenance(
+            software_version=f"{SOFTWARE_VERSION}/metrics",
+            calibration_version=CALIBRATION_VERSION, source="SYNTHETIC"))

sensorflow/rotr/models.py

@@ -0,0 +1,296 @@
+"""Typed data contracts for the ROTR capability.
+
+Design rules (docs/architecture/rotr-architecture.md §7):
+* every persisted record carries a Provenance block;
+* tri-state evidence is an enum, never a boolean;
+* no defaults that could fabricate evidence — missing evidence is UNKNOWN.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Dict, List, Optional
+
+from pydantic import BaseModel, Field
+
+# ---------------------------------------------------------------- enums
+
+CAUSAL_LAYERS = ["perception", "prediction", "planning", "localization",
+                 "map", "control", "policy_rule", "data_label"]
+
+EVIDENCE_STATUSES = ["SUPPORTED", "RULED_OUT", "UNKNOWN"]
+
+CONSEQUENCE_CLASSES = ["NO_MATERIAL_CONSEQUENCE", "DEGRADED_COMFORT",
+                       "PLANNER_INTERVENTION", "SAFETY_CRITICAL"]
+
+DATASET_ROLES = ["TRAIN", "VALIDATION", "TEST", "REGRESSION", "LAUNCH",
+                 "MONITORING"]
+# Membership in these roles blocks training eligibility without an override.
+PROTECTED_ROLES = {"REGRESSION", "LAUNCH"}
+
+RIGHT_OF_WAY = ["HAS_ROW", "MUST_YIELD", "NONE"]
+INTENTS = ["CROSSING", "MERGING", "YIELDING", "PROCEEDING", "WAITING"]
+
+SURROGATE_CAVEAT = (
+    "TTC/PET/DRAC are surrogate conflict measures: they do not alone "
+    "establish collision risk (controllability, evasive options and actor "
+    "reaction are unmodeled). They prioritize triage; they never decide a "
+    "gate by themselves.")
+
+
+def now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# ---------------------------------------------------------------- provenance
+
+
+class Provenance(BaseModel):
+    scenario_id: Optional[str] = None
+    dataset_version: str = ""
+    model_version: str = ""
+    software_version: str = ""
+    calibration_version: str = ""
+    timestamp: str = Field(default_factory=now_iso)
+    source: str = "SYNTHETIC"          # SYNTHETIC | LOG | COUNTERFACTUAL
+    confidence: Optional[float] = None
+
+
+# ---------------------------------------------------------------- world state
+
+
+class EgoState(BaseModel):
+    t: float
+    x: float
+    y: float
+    yaw: float = 0.0
+    speed: float = 0.0
+    accel: float = 0.0
+    lane_id: Optional[str] = None            # true lane association
+    believed_x: Optional[float] = None       # localization view (None=UNKNOWN)
+    believed_y: Optional[float] = None
+    believed_lane_id: Optional[str] = None
+
+
+class ActorObservation(BaseModel):
+    """The stack's view of one actor at one frame (None = not detected)."""
+
+    detected: bool = False
+    x: Optional[float] = None
+    y: Optional[float] = None
+    class_name: Optional[str] = None
+
+
+class ActorState(BaseModel):
+    t: float
+    x: float
+    y: float
+    vx: float = 0.0
+    vy: float = 0.0
+    yaw: float = 0.0
+    occluded: bool = False
+    observation: Optional[ActorObservation] = None
+
+
+class Actor(BaseModel):
+    actor_id: str
+    class_name: str                          # pedestrian|cyclist|vehicle|bus
+    dims: List[float] = Field(default_factory=lambda: [4.5, 1.9, 1.6])
+    right_of_way: str = "NONE"               # relative to ego at the conflict
+    intent: str = "PROCEEDING"               # ground-truth intent
+    predicted_intent: Optional[str] = None   # stack's prediction (None=UNKNOWN)
+    states: List[ActorState] = Field(default_factory=list)
+
+
+class Lane(BaseModel):
+    lane_id: str
+    center_y: float
+    direction: str = "+x"
+    permitted_maneuvers: List[str] = Field(
+        default_factory=lambda: ["STRAIGHT", "LANE_CHANGE"])
+    restricted_to: Optional[str] = None      # e.g. "bus" — None = unrestricted
+
+
+class Crosswalk(BaseModel):
+    crosswalk_id: str
+    x_min: float
+    x_max: float
+
+
+class RoadContext(BaseModel):
+    context_id: str = "ctx"
+    intersection_type: str = "none"          # none|uncontrolled|controlled
+    control: str = "none"                    # none|signal|stop_sign|yield_sign
+    signal_state_for_ego: Optional[str] = None   # green|red (controlled only)
+    intersection_x_min: Optional[float] = None
+    intersection_x_max: Optional[float] = None
+    stop_line_x: Optional[float] = None
+    lanes: List[Lane] = Field(default_factory=list)
+    crosswalks: List[Crosswalk] = Field(default_factory=list)
+    speed_limit_mps: float = 13.4
+
+
+class Environment(BaseModel):
+    visibility: str = "clear"                # clear|low
+    weather: str = "clear"                   # clear|rain
+    lighting: str = "day"                    # day|dusk|night
+
+
+class PlannedPoint(BaseModel):
+    t: float
+    x: float
+    y: float
+    v: float
+
+
+class PlantedTruth(BaseModel):
+    """Ground truth planted by the generator, used ONLY for evaluation."""
+
+    kind: str
+    is_violation_opportunity: bool = False
+    committed: bool = False                  # did this stack commit it?
+    expected_rule_id: Optional[str] = None
+    cause_layer: Optional[str] = None
+    notes: str = ""
+
+
+class ROTRScenario(BaseModel):
+    scenario_id: str
+    bank_id: str = ""
+    seed: int = 0
+    description: str = ""
+    environment: Environment = Field(default_factory=Environment)
+    map_context: RoadContext = Field(default_factory=RoadContext)   # as mapped
+    actual_context: RoadContext = Field(default_factory=RoadContext)  # as built
+    ego: List[EgoState] = Field(default_factory=list)     # executed trajectory
+    planned: List[PlannedPoint] = Field(default_factory=list)
+    actors: List[Actor] = Field(default_factory=list)
+    planted: PlantedTruth = Field(default_factory=lambda: PlantedTruth(kind=""))
+    provenance: Provenance = Field(default_factory=Provenance)
+
+
+# ---------------------------------------------------------------- detection
+
+
+class ROTRViolation(BaseModel):
+    violation_id: str
+    scenario_id: str
+    rule_id: str
+    rule_version: str
+    description: str = ""
+    actor_ids: List[str] = Field(default_factory=list)
+    t_start: float = 0.0
+    t_end: float = 0.0
+    evidence: Dict = Field(default_factory=dict)
+    confidence: float = 0.0
+    taxonomy: Dict[str, str] = Field(default_factory=dict)
+    provenance: Provenance = Field(default_factory=Provenance)
+
+
+# ---------------------------------------------------------------- attribution
+
+
+class LayerEvidence(BaseModel):
+    layer: str
+    status: str = "UNKNOWN"                  # SUPPORTED|RULED_OUT|UNKNOWN
+    evidence: str = ""
+    confidence: float = 0.0
+
+
+class FailureAttribution(BaseModel):
+    violation_id: str
+    scenario_id: str
+    layers: Dict[str, LayerEvidence] = Field(default_factory=dict)
+    primary_layer: Optional[str] = None      # None = triage to HITL
+    note: str = ""
+    provenance: Provenance = Field(default_factory=Provenance)
+
+
+# ---------------------------------------------------------------- consequence
+
+
+class SafetyAssessment(BaseModel):
+    min_ttc_s: Optional[float] = None
+    pet_s: Optional[float] = None
+    min_clearance_m: Optional[float] = None
+    stopping_distance_m: Optional[float] = None
+    max_braking_mps2: float = 0.0
+    max_lateral_deviation_m: float = 0.0
+    collision: bool = False
+    surrogate_caveat: str = SURROGATE_CAVEAT
+
+
+class PlannerEvaluation(BaseModel):
+    engine: str = ""                         # which replay planner produced this
+    observed_trajectory: List[Dict] = Field(default_factory=list)
+    corrected_trajectory: List[Dict] = Field(default_factory=list)
+    max_position_divergence_m: float = 0.0
+    max_speed_divergence_mps: float = 0.0
+    corrected_max_braking_mps2: float = 0.0
+    corrected_intervention: bool = False
+
+
+class CounterfactualScenario(BaseModel):
+    counterfactual_id: str
+    violation_id: str
+    scenario_id: str
+    corrected_layers: List[str] = Field(default_factory=list)
+    consequence_class: str = "NO_MATERIAL_CONSEQUENCE"
+    planner_evaluation: PlannerEvaluation = Field(default_factory=PlannerEvaluation)
+    observed_safety: SafetyAssessment = Field(default_factory=SafetyAssessment)
+    corrected_safety: SafetyAssessment = Field(default_factory=SafetyAssessment)
+    provenance: Provenance = Field(default_factory=Provenance)
+
+
+# ---------------------------------------------------------------- metrics
+
+
+class RegressionResult(BaseModel):
+    regression_id: str
+    baseline_run_id: str
+    candidate_run_id: str
+    baseline_model: str = ""
+    candidate_model: str = ""
+    metric_deltas: Dict = Field(default_factory=dict)
+    six_outcomes: Dict[str, bool] = Field(default_factory=dict)
+    primary_outcome: str = ""
+    seqeval: Optional[Dict] = None           # delegation payload (None=unavailable)
+    distribution_note: str = ""
+    provenance: Provenance = Field(default_factory=Provenance)
+
+
+# ---------------------------------------------------------------- governance
+
+
+class HITLReview(BaseModel):
+    review_id: str
+    run_id: str
+    violation_id: str
+    cluster_id: Optional[str] = None
+    status: str = "PENDING"                  # PENDING|VALIDATED|REJECTED
+    action: Optional[str] = None
+    actor: Optional[str] = None
+    notes: str = ""
+    provenance: Provenance = Field(default_factory=Provenance)
+
+
+class TrainingCandidate(BaseModel):
+    candidate_id: str
+    run_id: str
+    violation_id: str
+    cluster_id: Optional[str] = None
+    dataset_role: str = "REGRESSION"
+    training_eligible: bool = False
+    guard_state: str = "PROTECTED"           # PROTECTED|ELIGIBLE|OVERRIDDEN
+    override: Optional[Dict] = None          # {actor, reason, timestamp}
+    provenance: Provenance = Field(default_factory=Provenance)
+
+
+class ReleaseGateResult(BaseModel):
+    gate_id: str
+    run_id: str
+    policy_version: str
+    outcome: str = "GO"                      # GO|NO_GO
+    events: List[Dict] = Field(default_factory=list)
+    agentic_advisory: Optional[Dict] = None  # forwarded policy-engine outcome
+    provenance: Provenance = Field(default_factory=Provenance)

sensorflow/rotr/pipeline.py

@@ -0,0 +1,188 @@
+"""Run orchestration: bank -> detection -> attribution -> consequence ->
+metrics -> clusters -> gate -> HITL queue, persisted under runs/rotr/.
+
+Deterministic end to end: run ids derive from bank ids, which derive from
+(n_scenarios, seed, model_version, generator version)."""
+
+from __future__ import annotations
+
+from typing import Dict, List, Optional
+
+from sensorflow.rotr import store
+from sensorflow.rotr.attribution import attribute
+from sensorflow.rotr.consequence import classify
+from sensorflow.rotr.flywheel import build_queue
+from sensorflow.rotr.metrics import compute_metrics
+from sensorflow.rotr.models import ROTRScenario
+from sensorflow.rotr.rules import RULESET_VERSION, detect
+from sensorflow.rotr.scenes import bank_id_for, generate_bank
+from sensorflow.rotr.stopship import evaluate_gate
+from sensorflow.rotr.taxonomy import build_clusters, signature
+
+KIND_VULNERABILITY = {"fail_yield_pedestrian": "VRU"}
+
+
+# ------------------------------------------------------------ banks
+
+
+def create_bank(n_scenarios: int = 28, seed: int = 7,
+                model_version: str = "stack-v1") -> Dict:
+    scenarios = generate_bank(n_scenarios, seed, model_version)
+    bank_id = scenarios[0].bank_id
+    doc = {
+        "bank_id": bank_id, "n_scenarios": n_scenarios, "seed": seed,
+        "model_version": model_version,
+        "scenarios": [s.model_dump() for s in scenarios],
+    }
+    store.write_json(doc, "banks", f"{bank_id}.json")
+    return bank_summary(doc)
+
+
+def bank_summary(doc: Dict) -> Dict:
+    return {
+        "bank_id": doc["bank_id"], "n_scenarios": doc["n_scenarios"],
+        "seed": doc["seed"], "model_version": doc["model_version"],
+        "scenarios": [{
+            "scenario_id": s["scenario_id"],
+            "description": s["description"],
+            "kind": s["planted"]["kind"],
+            "committed": s["planted"]["committed"],
+            "is_violation_opportunity": s["planted"]["is_violation_opportunity"],
+            "environment": s["environment"],
+        } for s in doc["scenarios"]],
+    }
+
+
+def load_bank(bank_id: str) -> Optional[Dict]:
+    return store.read_json("banks", f"{bank_id}.json")
+
+
+def list_banks() -> List[Dict]:
+    out = []
+    for name in store.list_dir("banks"):
+        doc = store.read_json("banks", name)
+        if doc:
+            out.append({k: doc[k] for k in
+                        ("bank_id", "n_scenarios", "seed", "model_version")})
+    return out
+
+
+def load_scenario(bank_id: str, scenario_id: str) -> Optional[ROTRScenario]:
+    doc = load_bank(bank_id)
+    if not doc:
+        return None
+    for s in doc["scenarios"]:
+        if s["scenario_id"] == scenario_id:
+            return ROTRScenario.model_validate(s)
+    return None
+
+
+# ------------------------------------------------------------ runs
+
+
+def execute_run(bank_id: Optional[str] = None, n_scenarios: int = 28,
+                seed: int = 7, model_version: str = "stack-v1",
+                forward_to_agentic: bool = False) -> Dict:
+    if bank_id is None:
+        bank_id = bank_id_for(n_scenarios, seed, model_version)
+        if not store.exists("banks", f"{bank_id}.json"):
+            create_bank(n_scenarios, seed, model_version)
+    doc = load_bank(bank_id)
+    if doc is None:
+        raise KeyError(f"unknown bank {bank_id}")
+    scenarios = [ROTRScenario.model_validate(s) for s in doc["scenarios"]]
+    run_id = f"run-{bank_id}"
+
+    violations, attributions, consequences, gate_items = [], {}, {}, []
+    scenario_summaries = []
+    for sc in scenarios:
+        vs = detect(sc)
+        scenario_summaries.append({
+            "scenario_id": sc.scenario_id, "kind": sc.planted.kind,
+            "committed": sc.planted.committed,
+            "is_violation_opportunity": sc.planted.is_violation_opportunity,
+            "expected_rule_id": sc.planted.expected_rule_id,
+            "cause_layer": sc.planted.cause_layer,
+            "vulnerability": KIND_VULNERABILITY.get(sc.planted.kind, "NON_VRU"),
+            "visibility": sc.environment.visibility,
+            "lighting": sc.environment.lighting,
+            "weather": sc.environment.weather,
+            "n_violations_detected": len(vs),
+        })
+        for v in vs:
+            v.taxonomy = signature(v, sc)
+            attr = attribute(sc, v)
+            cf = classify(sc, v, attr)
+            violations.append(v.model_dump())
+            attributions[v.violation_id] = attr.model_dump()
+            consequences[v.violation_id] = {
+                "consequence_class": cf.consequence_class,
+                "corrected_layers": cf.corrected_layers,
+                "engine": cf.planner_evaluation.engine,
+                "max_position_divergence_m":
+                    cf.planner_evaluation.max_position_divergence_m,
+                "observed_safety": cf.observed_safety.model_dump(),
+                "corrected_safety": cf.corrected_safety.model_dump(),
+            }
+            # Full replay (trajectories) persisted separately: large payload.
+            store.write_json(cf.model_dump(), "runs", run_id, "consequences",
+                             f"{v.violation_id}.json")
+            gate_items.append({
+                "violation_id": v.violation_id,
+                "signature": v.taxonomy,
+                "attribution": attributions[v.violation_id],
+                "consequence": consequences[v.violation_id],
+            })
+
+    metrics = compute_metrics(scenario_summaries, violations, consequences)
+    cluster_items = [{
+        "violation_id": v["violation_id"], "signature": v["taxonomy"],
+        "primary_layer": attributions[v["violation_id"]]["primary_layer"],
+        "consequence_class":
+            consequences[v["violation_id"]]["consequence_class"],
+    } for v in violations]
+    clusters = build_clusters(cluster_items)
+    gate = evaluate_gate(run_id, gate_items,
+                         forward_to_agentic=forward_to_agentic)
+
+    run_doc = {
+        "run_id": run_id, "bank_id": bank_id,
+        "model_version": doc["model_version"],
+        "ruleset_version": RULESET_VERSION,
+        "n_scenarios": len(scenarios),
+        "scenario_summaries": scenario_summaries,
+        "violations": violations,
+        "attributions": attributions,
+        "consequences": consequences,
+        "metrics": metrics,
+        "clusters": clusters,
+        "gate": gate.model_dump(),
+    }
+    store.write_json(run_doc, "runs", run_id, "run.json")
+    build_queue(run_id, violations, clusters)
+    return run_doc
+
+
+def load_run(run_id: str) -> Optional[Dict]:
+    return store.read_json("runs", run_id, "run.json")
+
+
+def list_runs() -> List[Dict]:
+    out = []
+    for name in store.list_dir("runs"):
+        doc = store.read_json("runs", name, "run.json")
+        if doc:
+            out.append({
+                "run_id": doc["run_id"], "bank_id": doc["bank_id"],
+                "model_version": doc["model_version"],
+                "n_scenarios": doc["n_scenarios"],
+                "n_violations": len(doc["violations"]),
+                "gate_outcome": doc["gate"]["outcome"],
+                "rotr_recall": doc["metrics"]["rotr_recall"],
+            })
+    return out
+
+
+def load_consequence(run_id: str, violation_id: str) -> Optional[Dict]:
+    return store.read_json("runs", run_id, "consequences",
+                           f"{violation_id}.json")

sensorflow/rotr/rules.py

@@ -0,0 +1,379 @@
+"""Deterministic, versioned ROTR rule engine.
+
+The engine evaluates a TRAJECTORY against an EXPECTED-BEHAVIOR ENVELOPE
+derived from a road context and a set of actor views. It is deliberately
+source-agnostic and is used twice:
+
+* detection — executed ego trajectory + ACTUAL (as-built) context + ground-
+  truth actor states: "did the vehicle violate right-of-the-road?"
+* plan compliance (attribution.py) — planned trajectory + MAPPED context +
+  the STACK'S OWN actor view (detections + predicted intents), shifted into
+  the believed localization frame: "was the plan compliant given what the
+  stack believed?" — the planning layer is implicated only when this fails.
+
+This dual use is what keeps causal layers separated: the same rule text,
+different evidence sources. Rules never guess causes; they emit structured
+ROTRViolation records with the evidence fields that drove the decision.
+
+All thresholds in CONFIG are ILLUSTRATIVE (synthetic substrate).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+from typing import Dict, List, Optional
+
+# REUSE (landed, read-only): rotated-rectangle separation from SSAM.
+from sensorflow.safety.ssam_ext import rect_gap
+
+from sensorflow.rotr import SOFTWARE_VERSION
+from sensorflow.rotr.models import (
+    Provenance, ROTRScenario, ROTRViolation, RoadContext,
+)
+
+RULESET_VERSION = "rotr-rules-1.0.0"
+
+CONFIG: Dict[str, float] = {
+    # ILLUSTRATIVE thresholds — calibrated only against the synthetic bank.
+    "yield_max_transit_speed_mps": 2.0,   # faster than this through an
+                                          # occupied crosswalk = not yielding
+    "ped_roadway_half_width_m": 4.5,      # |y| below which a ped is "in roadway"
+    "crosswalk_margin_m": 0.5,
+    "restricted_min_frames": 5,
+    "lane_assoc_tolerance_m": 2.1,        # 0.6 * lane width
+    "signal_entry_min_speed_mps": 0.5,
+    "merge_min_time_gap_s": 1.5,
+    "stop_max_speed_mps": 0.7,            # above this at the line = no stop
+    "stop_window_m": 15.0,
+}
+
+EGO_DIMS = (4.5, 1.9)
+
+
+def ruleset_fingerprint() -> str:
+    blob = json.dumps({"version": RULESET_VERSION, "config": CONFIG},
+                      sort_keys=True)
+    return hashlib.sha256(blob.encode()).hexdigest()[:12]
+
+
+# ------------------------------------------------------------ view builders
+
+
+def gt_actor_views(scenario: ROTRScenario) -> List[Dict]:
+    """The world as it is: ground-truth actor states + true intent."""
+    return [{
+        "actor_id": a.actor_id, "class_name": a.class_name, "dims": a.dims,
+        "intent": a.intent, "right_of_way": a.right_of_way,
+        "states": [{"t": s.t, "x": s.x, "y": s.y, "vx": s.vx, "vy": s.vy,
+                    "yaw": s.yaw} for s in a.states],
+    } for a in scenario.actors]
+
+
+def stack_actor_views(scenario: ROTRScenario) -> List[Dict]:
+    """The world as the stack saw it: detected positions + predicted intent.
+
+    Frames where the actor was not detected are simply absent from the view
+    (the stack cannot plan around what it did not see). Actors with no
+    telemetry at all (observation=None) yield an empty view.
+    """
+    out = []
+    for a in scenario.actors:
+        states = []
+        for s in a.states:
+            o = s.observation
+            if o is None or not o.detected:
+                continue
+            states.append({"t": s.t, "x": o.x, "y": o.y,
+                           "vx": s.vx, "vy": s.vy, "yaw": s.yaw})
+        out.append({
+            "actor_id": a.actor_id, "class_name": a.class_name,
+            "dims": a.dims,
+            "intent": a.predicted_intent,      # the STACK's belief
+            "right_of_way": a.right_of_way,
+            "states": states,
+        })
+    return out
+
+
+def ego_trajectory(scenario: ROTRScenario) -> List[Dict]:
+    return [{"t": e.t, "x": e.x, "y": e.y, "speed": e.speed}
+            for e in scenario.ego]
+
+
+def planned_trajectory_believed_frame(scenario: ROTRScenario) -> List[Dict]:
+    """Plan shifted into the believed localization frame: the plan is judged
+    where the stack believed the vehicle to be, not where it truly was."""
+    if not scenario.planned or not scenario.ego:
+        return []
+    e0 = scenario.ego[0]
+    dy = (e0.believed_y - e0.y) if e0.believed_y is not None else 0.0
+    dx = (e0.believed_x - e0.x) if e0.believed_x is not None else 0.0
+    return [{"t": p.t, "x": p.x + dx, "y": p.y + dy, "speed": p.v}
+            for p in scenario.planned]
+
+
+# ------------------------------------------------------------ helpers
+
+
+def _nearest_lane(ctx: RoadContext, y: float, tol: float):
+    best, best_d = None, 1e9
+    for lane in ctx.lanes:
+        d = abs(lane.center_y - y)
+        if d < best_d:
+            best, best_d = lane, d
+    return (best, best_d) if best is not None and best_d <= tol else (None, None)
+
+
+def _state_at(actor: Dict, t: float) -> Optional[Dict]:
+    best, best_dt = None, 0.25
+    for s in actor["states"]:
+        dt = abs(s["t"] - t)
+        if dt < best_dt:
+            best, best_dt = s, dt
+    return best
+
+
+def _clearance(traj_pt: Dict, actor: Dict, st: Dict) -> float:
+    return rect_gap((traj_pt["x"], traj_pt["y"]), 0.0, *EGO_DIMS,
+                    (st["x"], st["y"]), st.get("yaw", 0.0),
+                    actor["dims"][0], actor["dims"][1])
+
+
+# ------------------------------------------------------------ rules
+# Each rule: (traj, ctx, actors) -> Optional[evidence dict]. A returned dict
+# means the trajectory exits the expected-behavior envelope for that rule.
+
+
+def _rule_yield_ped(traj: List[Dict], ctx: RoadContext,
+                    actors: List[Dict]) -> Optional[Dict]:
+    """R-YIELD-PED-01: yield to a crossing pedestrian at a crosswalk."""
+    if not ctx.crosswalks:
+        return None
+    cw = ctx.crosswalks[0]
+    margin = CONFIG["crosswalk_margin_m"]
+    for actor in actors:
+        if actor["class_name"] not in ("pedestrian", "cyclist"):
+            continue
+        if actor["intent"] != "CROSSING" or actor["right_of_way"] != "HAS_ROW":
+            continue
+        # Ped must actually enter the roadway for the obligation to bind.
+        if not any(abs(s["y"]) < 3.5 for s in actor["states"]):
+            continue
+        worst = None
+        for p in traj:
+            if not (cw.x_min - margin <= p["x"] <= cw.x_max + margin):
+                continue
+            st = _state_at(actor, p["t"])
+            if st is None:
+                continue
+            if not (cw.x_min - margin <= st["x"] <= cw.x_max + margin):
+                continue
+            if abs(st["y"]) > CONFIG["ped_roadway_half_width_m"]:
+                continue
+            if p["speed"] > CONFIG["yield_max_transit_speed_mps"]:
+                gap = _clearance(p, actor, st)
+                if worst is None or gap < worst["min_clearance_m"]:
+                    worst = {"t": round(p["t"], 2),
+                             "ego_speed_mps": round(p["speed"], 2),
+                             "ped_y_m": round(st["y"], 2),
+                             "min_clearance_m": round(gap, 3),
+                             "actor_id": actor["actor_id"]}
+        if worst:
+            worst["threshold_speed_mps"] = CONFIG["yield_max_transit_speed_mps"]
+            return worst
+    return None
+
+
+def _rule_restricted_path(traj: List[Dict], ctx: RoadContext,
+                          actors: List[Dict]) -> Optional[Dict]:
+    """R-PATH-RESTRICT-02: do not occupy a lane restricted to another class."""
+    frames_in, lane_id = 0, None
+    for p in traj:
+        lane, _ = _nearest_lane(ctx, p["y"], CONFIG["lane_assoc_tolerance_m"])
+        if lane is not None and lane.restricted_to not in (None, "any", "car"):
+            frames_in += 1
+            lane_id = lane.lane_id
+    if frames_in >= CONFIG["restricted_min_frames"]:
+        return {"lane_id": lane_id, "frames_in_restricted": frames_in,
+                "duration_s": round(frames_in * (traj[1]["t"] - traj[0]["t"]), 2)
+                if len(traj) > 1 else 0.0}
+    return None
+
+
+def _rule_lane_maneuver(traj: List[Dict], ctx: RoadContext,
+                        actors: List[Dict]) -> Optional[Dict]:
+    """R-LANE-MANEUVER-03: only maneuvers permitted from the occupied lane."""
+    if ctx.intersection_x_min is None or ctx.stop_line_x is None:
+        return None
+    entry = [p for p in traj if p["x"] < ctx.stop_line_x]
+    through = [p for p in traj if p["x"] > (ctx.intersection_x_max or 0)]
+    if not entry or not through:
+        return None
+    lane, _ = _nearest_lane(ctx, entry[-1]["y"], CONFIG["lane_assoc_tolerance_m"])
+    if lane is None:
+        return None
+    went_straight = abs(through[0]["y"] - entry[-1]["y"]) < 1.5
+    if went_straight and "STRAIGHT" not in lane.permitted_maneuvers:
+        return {"lane_id": lane.lane_id, "maneuver": "STRAIGHT",
+                "permitted": lane.permitted_maneuvers,
+                "entry_y_m": round(entry[-1]["y"], 2)}
+    return None
+
+
+def _rule_signal_conflict(traj: List[Dict], ctx: RoadContext,
+                          actors: List[Dict]) -> Optional[Dict]:
+    """R-INT-CONFLICT-04: do not enter a signalized intersection on red."""
+    if ctx.control != "signal" or ctx.signal_state_for_ego != "red":
+        return None
+    if ctx.intersection_x_min is None:
+        return None
+    for p in traj:
+        if p["x"] >= ctx.intersection_x_min and \
+                p["speed"] > CONFIG["signal_entry_min_speed_mps"]:
+            ev = {"t": round(p["t"], 2), "entry_speed_mps": round(p["speed"], 2),
+                  "signal_state": "red"}
+            # Conflicting right-of-way actor evidence (severity, not verdict).
+            worst_gap, worst_id = None, None
+            for actor in actors:
+                if actor["right_of_way"] != "HAS_ROW":
+                    continue
+                for q in traj:
+                    if q["x"] < ctx.intersection_x_min:
+                        continue
+                    st = _state_at(actor, q["t"])
+                    if st is None:
+                        continue
+                    gap = _clearance(q, actor, st)
+                    if worst_gap is None or gap < worst_gap:
+                        worst_gap, worst_id = gap, actor["actor_id"]
+            if worst_gap is not None:
+                ev["min_clearance_m"] = round(worst_gap, 3)
+                ev["actor_id"] = worst_id
+            return ev
+    return None
+
+
+def _rule_merge_gap(traj: List[Dict], ctx: RoadContext,
+                    actors: List[Dict]) -> Optional[Dict]:
+    """R-MERGE-GAP-05: merging in front of right-of-way traffic needs a gap."""
+    tol = CONFIG["lane_assoc_tolerance_m"]
+    lane_seq = []
+    for p in traj:
+        lane, _ = _nearest_lane(ctx, p["y"], tol)
+        lane_seq.append(lane.lane_id if lane else None)
+    change_idx = None
+    for k in range(1, len(lane_seq)):
+        if lane_seq[k] and lane_seq[k - 1] and lane_seq[k] != lane_seq[k - 1]:
+            change_idx = k
+    if change_idx is None:
+        return None
+    p = traj[change_idx]
+    target = lane_seq[change_idx]
+    for actor in actors:
+        if actor["right_of_way"] != "HAS_ROW":
+            continue
+        st = _state_at(actor, p["t"])
+        if st is None:
+            continue
+        lane, _ = _nearest_lane(ctx, st["y"], tol)
+        if lane is None or lane.lane_id != target:
+            continue
+        if st["x"] >= p["x"]:
+            continue                     # only trailing traffic holds the gap
+        speed = math.hypot(st["vx"], st["vy"])
+        bumper_gap = (p["x"] - st["x"]) - 0.5 * (EGO_DIMS[0] + actor["dims"][0])
+        time_gap = bumper_gap / max(speed, 0.1)
+        if time_gap < CONFIG["merge_min_time_gap_s"]:
+            return {"t": round(p["t"], 2), "target_lane": target,
+                    "actor_id": actor["actor_id"],
+                    "bumper_gap_m": round(bumper_gap, 2),
+                    "time_gap_s": round(time_gap, 2),
+                    "threshold_time_gap_s": CONFIG["merge_min_time_gap_s"]}
+    return None
+
+
+def _rule_stop(traj: List[Dict], ctx: RoadContext,
+               actors: List[Dict]) -> Optional[Dict]:
+    """R-STOP-06: a stop control requires an actual stop before the line."""
+    if ctx.control != "stop_sign" or ctx.stop_line_x is None:
+        return None
+    line = ctx.stop_line_x
+    window = [p for p in traj
+              if line - CONFIG["stop_window_m"] <= p["x"] <= line + 1.0]
+    crossed = any(p["x"] > line for p in traj)
+    if not window or not crossed:
+        return None
+    min_speed = min(p["speed"] for p in window)
+    if min_speed > CONFIG["stop_max_speed_mps"]:
+        return {"min_speed_in_window_mps": round(min_speed, 2),
+                "stop_line_x": line,
+                "threshold_mps": CONFIG["stop_max_speed_mps"]}
+    return None
+
+
+RULES = [
+    ("R-YIELD-PED-01", "failure to yield to crossing pedestrian", _rule_yield_ped),
+    ("R-PATH-RESTRICT-02", "restricted-path entry", _rule_restricted_path),
+    ("R-LANE-MANEUVER-03", "maneuver not permitted from occupied lane",
+     _rule_lane_maneuver),
+    ("R-INT-CONFLICT-04", "entered signalized intersection on red",
+     _rule_signal_conflict),
+    ("R-MERGE-GAP-05", "merge with insufficient gap to right-of-way traffic",
+     _rule_merge_gap),
+    ("R-STOP-06", "failed to stop at stop control", _rule_stop),
+]
+
+
+# ------------------------------------------------------------ entry points
+
+
+def evaluate_trajectory(traj: List[Dict], ctx: RoadContext,
+                        actors: List[Dict]) -> List[Dict]:
+    """Run every rule; return [{rule_id, description, evidence}]."""
+    hits = []
+    for rule_id, desc, fn in RULES:
+        ev = fn(traj, ctx, actors)
+        if ev is not None:
+            hits.append({"rule_id": rule_id, "description": desc,
+                         "evidence": ev})
+    return hits
+
+
+def detect(scenario: ROTRScenario) -> List[ROTRViolation]:
+    """Detection: executed trajectory vs the ACTUAL world (GT evidence)."""
+    traj = ego_trajectory(scenario)
+    hits = evaluate_trajectory(traj, scenario.actual_context,
+                               gt_actor_views(scenario))
+    out = []
+    for h in hits:
+        ev = h["evidence"]
+        vid = f"{scenario.scenario_id}-{h['rule_id']}"
+        out.append(ROTRViolation(
+            violation_id=vid, scenario_id=scenario.scenario_id,
+            rule_id=h["rule_id"], rule_version=RULESET_VERSION,
+            description=f"{h['description']} — {scenario.description}",
+            actor_ids=[ev["actor_id"]] if "actor_id" in ev else [],
+            t_start=float(ev.get("t", 0.0)), t_end=float(ev.get("t", 0.0)),
+            evidence={**ev, "ruleset_fingerprint": ruleset_fingerprint(),
+                      "thresholds_note": "ILLUSTRATIVE (synthetic substrate)"},
+            confidence=0.9,
+            provenance=Provenance(
+                scenario_id=scenario.scenario_id,
+                dataset_version=scenario.provenance.dataset_version,
+                model_version=scenario.provenance.model_version,
+                software_version=f"{SOFTWARE_VERSION}/{RULESET_VERSION}",
+                source=scenario.provenance.source, confidence=0.9)))
+    return out
+
+
+def plan_violations(scenario: ROTRScenario) -> List[Dict]:
+    """Plan compliance given the stack's OWN world view (mapped context,
+    detections, predicted intents, believed localization frame). Used by
+    attribution: a compliant plan RULES OUT the planning layer."""
+    traj = planned_trajectory_believed_frame(scenario)
+    if not traj:
+        return []
+    return evaluate_trajectory(traj, scenario.map_context,
+                               stack_actor_views(scenario))

sensorflow/rotr/scenes.py

@@ -0,0 +1,544 @@
+"""Deterministic road-context scenario substrate for ROTR.
+
+Extends the platform's scene conventions (REUSE, read-only:
+bevfusion.scenes DT / EGO_SPEED / seeded default_rng discipline) with what
+bevfusion deliberately does not model: lane geometry with permitted
+maneuvers, intersection topology (controlled/uncontrolled, signals/signs),
+crosswalks, and multi-actor right-of-way states + intent.
+
+A bank is a pure function of (n_scenarios, seed, model_version): the same
+inputs produce byte-identical scenarios. `model_version` selects a stack
+profile that determines — deterministically per scenario — whether the
+stack COMMITS each planted violation opportunity; this is how
+baseline-vs-candidate regression banks are materialized (the bank is "the
+log of that software driving the same scenario set", i.e. resimulation).
+
+Planted truth includes NON-violations that superficially resemble
+violations (legal assertive merge; ego yielding while holding right-of-way)
+so the rule engine's false-accusation rate is measurable.
+
+All numeric thresholds/kinematics here are ILLUSTRATIVE (synthetic
+substrate); the generator's contract, not its constants, is the deliverable.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import math
+from typing import Dict, List, Optional, Tuple
+
+import numpy as np
+
+# REUSE (landed, read-only): platform time base and nominal ego speed.
+from sensorflow.bevfusion.scenes import DT, EGO_SPEED
+
+from sensorflow.rotr import SOFTWARE_VERSION
+from sensorflow.rotr.models import (
+    Actor, ActorObservation, ActorState, Crosswalk, EgoState, Environment,
+    Lane, PlannedPoint, PlantedTruth, Provenance, ROTRScenario, RoadContext,
+)
+
+N_FRAMES = 70
+X0 = -10.0
+V_NOM = float(EGO_SPEED)          # 10 m/s
+LANE_W = 3.5
+
+# Intersection geometry shared by intersection templates.
+STOP_LINE_X = 38.0
+CW_X_MIN, CW_X_MAX = 39.5, 44.5   # crosswalk span along x
+BOX_X_MIN, BOX_X_MAX = 39.5, 52.0
+
+DATASET_VERSION = "rotr-bank-1"
+GENERATOR_VERSION = "rotr-scenes-1.0.0"
+
+# Stack profiles: per-cause commit probability for planted violation
+# opportunities. Deterministic per scenario (seeded rng draw).
+MODEL_PROFILES: Dict[str, Dict[str, float]] = {
+    "stack-v1": {"perception": 0.90, "prediction": 0.90, "planning": 0.50,
+                 "localization": 1.00, "map": 1.00, "control": 0.80,
+                 "data_label": 1.00},
+    "stack-v2-improved": {"perception": 0.25, "prediction": 0.90,
+                          "planning": 0.50, "localization": 1.00,
+                          "map": 1.00, "control": 0.80, "data_label": 1.00},
+    "stack-v3-planning-regression": {"perception": 0.25, "prediction": 0.90,
+                                     "planning": 1.00, "localization": 1.00,
+                                     "map": 1.00, "control": 0.80,
+                                     "data_label": 1.00},
+}
+DEFAULT_MODEL = "stack-v1"
+
+
+# ------------------------------------------------------------ contexts
+
+
+def _lane(lane_id: str, y: float, maneuvers: List[str],
+          restricted_to: Optional[str] = None) -> Lane:
+    return Lane(lane_id=lane_id, center_y=y, permitted_maneuvers=maneuvers,
+                restricted_to=restricted_to)
+
+
+def _ctx_uncontrolled() -> RoadContext:
+    return RoadContext(
+        context_id="uncontrolled-xing", intersection_type="uncontrolled",
+        control="none", stop_line_x=STOP_LINE_X,
+        intersection_x_min=BOX_X_MIN, intersection_x_max=BOX_X_MAX,
+        lanes=[_lane("L0", 0.0, ["STRAIGHT", "LANE_CHANGE"]),
+               _lane("L-1", -LANE_W, ["STRAIGHT", "LANE_CHANGE"])],
+        crosswalks=[Crosswalk(crosswalk_id="cw1", x_min=CW_X_MIN, x_max=CW_X_MAX)])
+
+
+def _ctx_bus(restricted: bool) -> RoadContext:
+    return RoadContext(
+        context_id="bus-corridor", intersection_type="none", control="none",
+        lanes=[_lane("L0", 0.0, ["STRAIGHT", "LANE_CHANGE"]),
+               _lane("LBUS", -LANE_W, ["STRAIGHT"],
+                     restricted_to="bus" if restricted else None)])
+
+
+def _ctx_turnlane() -> RoadContext:
+    return RoadContext(
+        context_id="turn-lane-xing", intersection_type="uncontrolled",
+        control="none", stop_line_x=STOP_LINE_X,
+        intersection_x_min=BOX_X_MIN, intersection_x_max=BOX_X_MAX,
+        lanes=[_lane("L0", 0.0, ["STRAIGHT", "LANE_CHANGE"]),
+               _lane("LT", LANE_W, ["LEFT_TURN"])])
+
+
+def _ctx_signal(mapped_signal: bool) -> RoadContext:
+    return RoadContext(
+        context_id="signal-xing", intersection_type="controlled",
+        control="signal" if mapped_signal else "none",
+        signal_state_for_ego="red" if mapped_signal else None,
+        stop_line_x=STOP_LINE_X,
+        intersection_x_min=BOX_X_MIN, intersection_x_max=BOX_X_MAX,
+        lanes=[_lane("L0", 0.0, ["STRAIGHT"])],
+        crosswalks=[Crosswalk(crosswalk_id="cw1", x_min=CW_X_MIN, x_max=CW_X_MAX)])
+
+
+def _ctx_stopsign() -> RoadContext:
+    return RoadContext(
+        context_id="stopsign-xing", intersection_type="controlled",
+        control="stop_sign", stop_line_x=STOP_LINE_X,
+        intersection_x_min=BOX_X_MIN, intersection_x_max=BOX_X_MAX,
+        lanes=[_lane("L0", 0.0, ["STRAIGHT"])],
+        crosswalks=[Crosswalk(crosswalk_id="cw1", x_min=CW_X_MIN, x_max=CW_X_MAX)])
+
+
+def _ctx_merge() -> RoadContext:
+    return RoadContext(
+        context_id="merge-corridor", intersection_type="none", control="none",
+        lanes=[_lane("L0", 0.0, ["STRAIGHT", "LANE_CHANGE"]),
+               _lane("L-1", -LANE_W, ["STRAIGHT", "LANE_CHANGE"])])
+
+
+# ------------------------------------------------------------ trajectories
+
+
+def _traj_const(v: float = V_NOM, y: float = 0.0) -> List[Dict]:
+    return [{"t": fi * DT, "x": X0 + v * fi * DT, "y": y, "speed": v,
+             "accel": 0.0} for fi in range(N_FRAMES)]
+
+
+def _traj_stop(x_stop: float, decel: float = 3.0, v0: float = V_NOM,
+               y: float = 0.0, resume_t: Optional[float] = None,
+               resume_accel: float = 2.0) -> List[Dict]:
+    """Cruise, brake to a stop at x_stop, optionally resume."""
+    d_brake = v0 * v0 / (2 * decel)
+    x_brake = x_stop - d_brake
+    pts, x, v, a = [], X0, v0, 0.0
+    for fi in range(N_FRAMES):
+        t = fi * DT
+        pts.append({"t": t, "x": x, "y": y, "speed": v, "accel": a})
+        if v > 0 and x >= x_brake:
+            a = -decel
+        elif v <= 0 and resume_t is not None and t >= resume_t:
+            a = resume_accel
+        elif v <= 0:
+            a = 0.0
+        v = max(0.0, min(v0, v + a * DT))
+        if v == 0.0:
+            a = 0.0
+        x += v * DT
+    return pts
+
+
+def _traj_rolling(x_line: float, v_min: float = 3.0, v0: float = V_NOM,
+                  y: float = 0.0) -> List[Dict]:
+    """Slow to v_min at the stop line, never stop, resume to v0."""
+    decel = 2.5
+    d = (v0 * v0 - v_min * v_min) / (2 * decel)
+    x_brake = x_line - d
+    pts, x, v, a = [], X0, v0, 0.0
+    for fi in range(N_FRAMES):
+        t = fi * DT
+        pts.append({"t": t, "x": x, "y": y, "speed": v, "accel": a})
+        if x >= x_brake and v > v_min and x < x_line:
+            a = -decel
+        elif x >= x_line and v < v0:
+            a = 2.0
+        else:
+            a = 0.0
+        v = max(v_min if x >= x_brake else 0.0, min(v0, v + a * DT))
+        x += v * DT
+    return pts
+
+
+def _traj_lane_change(t_start: float, t_end: float, y_from: float,
+                      y_to: float, v: float = V_NOM) -> List[Dict]:
+    pts = []
+    for fi in range(N_FRAMES):
+        t = fi * DT
+        if t <= t_start:
+            y = y_from
+        elif t >= t_end:
+            y = y_to
+        else:
+            frac = (t - t_start) / (t_end - t_start)
+            y = y_from + (y_to - y_from) * (1 - math.cos(math.pi * frac)) / 2
+        pts.append({"t": t, "x": X0 + v * t, "y": y, "speed": v, "accel": 0.0})
+    return pts
+
+
+def _ego_states(pts: List[Dict], context: RoadContext,
+                loc_offset_y: float = 0.0) -> List[EgoState]:
+    """Attach true lane association + localization (believed) view."""
+    out = []
+    for p in pts:
+        lane = _nearest_lane(context, p["y"])
+        out.append(EgoState(
+            t=p["t"], x=p["x"], y=p["y"], yaw=0.0, speed=p["speed"],
+            accel=p["accel"], lane_id=lane.lane_id if lane else None,
+            believed_x=p["x"], believed_y=p["y"] + loc_offset_y,
+            believed_lane_id=(_nearest_lane(context, p["y"] + loc_offset_y).lane_id
+                              if _nearest_lane(context, p["y"] + loc_offset_y)
+                              else None)))
+    return out
+
+
+def _nearest_lane(context: RoadContext, y: float) -> Optional[Lane]:
+    best, best_d = None, 1e9
+    for lane in context.lanes:
+        d = abs(lane.center_y - y)
+        if d < best_d:
+            best, best_d = lane, d
+    return best if best_d <= LANE_W * 0.6 else None
+
+
+def _planned(pts: List[Dict]) -> List[PlannedPoint]:
+    return [PlannedPoint(t=p["t"], x=p["x"], y=p["y"], v=p["speed"])
+            for p in pts]
+
+
+# ------------------------------------------------------------ actors
+
+
+def _observe(rng: np.random.Generator, st: Dict, cls: str,
+             mode: str) -> Optional[ActorObservation]:
+    """Plant the stack's observation of a GT actor state.
+
+    mode: good | missed | late(handled by caller) | absent(None=telemetry gap)
+    """
+    if mode == "absent":
+        return None
+    if mode == "missed":
+        return ActorObservation(detected=False)
+    return ActorObservation(
+        detected=True,
+        x=st["x"] + float(rng.normal(0, 0.12)),
+        y=st["y"] + float(rng.normal(0, 0.08)),
+        class_name=cls)
+
+
+def _actor(rng: np.random.Generator, actor_id: str, cls: str,
+           dims: List[float], row: str, intent: str,
+           predicted_intent: Optional[str],
+           state_fn, obs_mode_fn) -> Actor:
+    states = []
+    for fi in range(N_FRAMES):
+        t = fi * DT
+        x, y, vx, vy, yaw = state_fn(t)
+        st = {"x": x, "y": y}
+        states.append(ActorState(
+            t=t, x=x, y=y, vx=vx, vy=vy, yaw=yaw,
+            observation=_observe(rng, st, cls, obs_mode_fn(t))))
+    return Actor(actor_id=actor_id, class_name=cls, dims=dims,
+                 right_of_way=row, intent=intent,
+                 predicted_intent=predicted_intent, states=states)
+
+
+def _crossing_ped_fn(x: float = 44.9, y0: float = 6.0, vy: float = -1.4,
+                     y_end: float = -7.0):
+    # x=44.9 places the crossing at the far edge of the crosswalk so a
+    # non-yielding ego produces a deterministic NEAR-miss (sub-meter
+    # clearance), not a synthetic collision.
+    def fn(t: float) -> Tuple[float, float, float, float, float]:
+        y = max(y_end, y0 + vy * t)
+        moving = y > y_end
+        return x, y, 0.0, vy if moving else 0.0, -math.pi / 2
+    return fn
+
+
+def _waiting_ped_fn(x: float = 42.5, y: float = 5.5):
+    def fn(t: float):
+        return x, y, 0.0, 0.0, -math.pi / 2
+    return fn
+
+
+def _crossing_vehicle_fn(x: float = 45.5, y0: float = 20.0, vy: float = -2.8):
+    def fn(t: float):
+        return x, y0 + vy * t, 0.0, vy, -math.pi / 2
+    return fn
+
+
+def _trailing_vehicle_fn(x0: float, vx: float = 13.0,
+                         brake_t: Optional[float] = None,
+                         brake_a: float = 2.5, v_min: float = 8.5):
+    def fn(t: float):
+        if brake_t is None or t <= brake_t:
+            return x0 + vx * t, -LANE_W, vx, 0.0, 0.0
+        tb = t - brake_t
+        v = max(v_min, vx - brake_a * tb)
+        t_ramp = (vx - v_min) / brake_a
+        if tb <= t_ramp:
+            x = x0 + vx * brake_t + vx * tb - 0.5 * brake_a * tb * tb
+        else:
+            x = (x0 + vx * brake_t + vx * t_ramp - 0.5 * brake_a * t_ramp ** 2
+                 + v_min * (tb - t_ramp))
+        return x, -LANE_W, v, 0.0, 0.0
+    return fn
+
+
+ALWAYS_GOOD = lambda t: "good"          # noqa: E731
+NEVER_SEEN = lambda t: "missed"         # noqa: E731
+TELEMETRY_GAP = lambda t: "absent"      # noqa: E731
+
+
+# ------------------------------------------------------------ templates
+#
+# Each template returns the scenario pieces given (rng, committed).
+# `cause` is the planted causal layer for the committed variant.
+
+VEH_DIMS = [4.5, 1.9, 1.6]
+PED_DIMS = [0.6, 0.6, 1.75]
+
+
+def _tpl_fail_yield_ped(rng, committed: bool, cause: str) -> Dict:
+    ctx = _ctx_uncontrolled()
+    obs_mode = NEVER_SEEN if (committed and cause == "perception") else \
+        (TELEMETRY_GAP if (committed and cause == "data_label") else ALWAYS_GOOD)
+    predicted = None if cause == "data_label" else \
+        ("WAITING" if (committed and cause == "prediction") else "CROSSING")
+    ped = _actor(rng, "ped-1", "pedestrian", PED_DIMS, "HAS_ROW", "CROSSING",
+                 predicted, _crossing_ped_fn(), obs_mode)
+    if committed:
+        exec_pts = _traj_const()
+        plan_pts = exec_pts
+    else:
+        exec_pts = _traj_stop(STOP_LINE_X - 0.5, resume_t=6.5)
+        plan_pts = exec_pts
+    ego = _ego_states(exec_pts, ctx)
+    if committed and cause == "data_label":
+        # Planted record-integrity fault: disordered timestamps mid-approach.
+        for k in range(30, 34):
+            ego[k].t = ego[k].t - 0.35
+    return {"map_context": ctx, "actual_context": ctx, "ego": ego,
+            "planned": _planned(plan_pts), "actors": [ped],
+            "expected_rule": "R-YIELD-PED-01",
+            "description": "pedestrian crossing at uncontrolled intersection; "
+                           + ("ego proceeds without yielding" if committed
+                              else "ego yields correctly")}
+
+
+def _tpl_restricted_path(rng, committed: bool, cause: str) -> Dict:
+    actual = _ctx_bus(restricted=True)
+    mapped = _ctx_bus(restricted=(cause != "map"))
+    if committed:
+        pts = _traj_lane_change(2.0, 3.5, 0.0, -LANE_W)
+    else:
+        pts = _traj_const()
+    return {"map_context": mapped, "actual_context": actual,
+            "ego": _ego_states(pts, actual), "planned": _planned(pts),
+            "actors": [],
+            "expected_rule": "R-PATH-RESTRICT-02",
+            "description": "bus-restricted lane; "
+                           + ("ego enters restricted path" if committed
+                              else "ego stays in general lane")
+                           + (" (restriction missing from map)"
+                              if cause == "map" else "")}
+
+
+def _tpl_wrong_lane(rng, committed: bool, cause: str) -> Dict:
+    ctx = _ctx_turnlane()
+    # Ego actually drives the left-turn-only lane straight through; its
+    # localization believes it is in the through lane (offset -LANE_W).
+    pts = _traj_const(y=LANE_W)
+    ego = _ego_states(pts, ctx, loc_offset_y=-LANE_W)
+    return {"map_context": ctx, "actual_context": ctx, "ego": ego,
+            "planned": _planned(pts), "actors": [],
+            "expected_rule": "R-LANE-MANEUVER-03",
+            "description": "ego proceeds straight from a left-turn-only lane "
+                           "(wrong lane association from localization error)"}
+
+
+def _tpl_signal_conflict(rng, committed: bool, cause: str) -> Dict:
+    actual = _ctx_signal(mapped_signal=True)
+    mapped = _ctx_signal(mapped_signal=(cause != "map"))
+    obs_mode = NEVER_SEEN if (committed and cause == "perception") else ALWAYS_GOOD
+    veh = _actor(rng, "xveh-1", "vehicle", VEH_DIMS, "HAS_ROW", "CROSSING",
+                 "CROSSING", _crossing_vehicle_fn(), obs_mode)
+    if committed:
+        pts = _traj_const()
+    else:
+        pts = _traj_stop(STOP_LINE_X - 0.5)
+    return {"map_context": mapped, "actual_context": actual,
+            "ego": _ego_states(pts, actual), "planned": _planned(pts),
+            "actors": [veh],
+            "expected_rule": "R-INT-CONFLICT-04",
+            "description": "red signal with conflicting cross traffic; "
+                           + ("ego enters on red" if committed
+                              else "ego stops at the line")
+                           + (" (signal unmapped)" if cause == "map" else "")}
+
+
+def _tpl_unsafe_merge(rng, committed: bool, cause: str) -> Dict:
+    ctx = _ctx_merge()
+    if committed:
+        veh = _actor(rng, "mveh-1", "vehicle", VEH_DIMS, "HAS_ROW",
+                     "PROCEEDING", "PROCEEDING",
+                     _trailing_vehicle_fn(-31.5, brake_t=4.6), ALWAYS_GOOD)
+        pts = _traj_lane_change(3.0, 4.5, 0.0, -LANE_W)
+    else:
+        veh = _actor(rng, "mveh-1", "vehicle", VEH_DIMS, "HAS_ROW",
+                     "PROCEEDING", "PROCEEDING",
+                     _trailing_vehicle_fn(-58.0), ALWAYS_GOOD)
+        pts = _traj_const()
+    return {"map_context": ctx, "actual_context": ctx,
+            "ego": _ego_states(pts, ctx), "planned": _planned(pts),
+            "actors": [veh],
+            "expected_rule": "R-MERGE-GAP-05",
+            "description": "merge ahead of right-of-way traffic with "
+                           + ("insufficient gap" if committed else "safe gap")}
+
+
+def _tpl_stop_overshoot(rng, committed: bool, cause: str) -> Dict:
+    ctx = _ctx_stopsign()
+    ped = _actor(rng, "ped-w1", "pedestrian", PED_DIMS, "NONE", "WAITING",
+                 "WAITING", _waiting_ped_fn(), ALWAYS_GOOD)
+    plan_pts = _traj_stop(STOP_LINE_X - 0.5, resume_t=5.6)
+    if committed:
+        exec_pts = _traj_rolling(STOP_LINE_X)      # actuation under-brakes
+    else:
+        exec_pts = plan_pts
+    return {"map_context": ctx, "actual_context": ctx,
+            "ego": _ego_states(exec_pts, ctx), "planned": _planned(plan_pts),
+            "actors": [ped],
+            "expected_rule": "R-STOP-06",
+            "description": "stop sign; " + ("planned stop, executed roll-through "
+                                            "(control tracking error)"
+                                            if committed else "full stop made")}
+
+
+def _tpl_legal_merge(rng, committed: bool, cause: str) -> Dict:
+    d = _tpl_unsafe_merge(rng, committed=False, cause="")
+    # Assertive but legal: the merge happens, with a sufficient gap.
+    ctx = d["actual_context"]
+    pts = _traj_lane_change(3.0, 4.5, 0.0, -LANE_W)
+    d.update({"ego": _ego_states(pts, ctx), "planned": _planned(pts),
+              "expected_rule": None,
+              "description": "legal assertive merge (sufficient gap; planted "
+                             "NON-violation that superficially resembles one)"})
+    return d
+
+
+def _tpl_yield_with_row(rng, committed: bool, cause: str) -> Dict:
+    ctx = _ctx_uncontrolled()
+    ped = _actor(rng, "ped-c1", "pedestrian", PED_DIMS, "NONE", "WAITING",
+                 "WAITING", _waiting_ped_fn(), ALWAYS_GOOD)
+    pts = _traj_stop(STOP_LINE_X - 0.5, resume_t=5.8)
+    return {"map_context": ctx, "actual_context": ctx,
+            "ego": _ego_states(pts, ctx), "planned": _planned(pts),
+            "actors": [ped],
+            "expected_rule": None,
+            "description": "ego yields although it holds right-of-way (curb "
+                           "pedestrian is waiting; planted NON-violation)"}
+
+
+def _tpl_green_proceed(rng, committed: bool, cause: str) -> Dict:
+    ctx = _ctx_signal(mapped_signal=True)
+    ctx.signal_state_for_ego = "green"
+    pts = _traj_const()
+    return {"map_context": ctx, "actual_context": ctx.model_copy(deep=True),
+            "ego": _ego_states(pts, ctx), "planned": _planned(pts),
+            "actors": [],
+            "expected_rule": None,
+            "description": "normal green-light transit (planted NON-violation)"}
+
+
+# (template fn, planted kind, cause layer or None for non-violations)
+TEMPLATES = [
+    (_tpl_fail_yield_ped, "fail_yield_pedestrian", "perception"),
+    (_tpl_fail_yield_ped, "fail_yield_pedestrian", "planning"),
+    (_tpl_fail_yield_ped, "fail_yield_pedestrian", "prediction"),
+    (_tpl_restricted_path, "restricted_path_entry", "planning"),
+    (_tpl_restricted_path, "restricted_path_entry", "map"),
+    (_tpl_wrong_lane, "wrong_lane_association", "localization"),
+    (_tpl_signal_conflict, "intersection_conflict", "map"),
+    (_tpl_signal_conflict, "intersection_conflict", "perception"),
+    (_tpl_unsafe_merge, "unsafe_merge", "planning"),
+    (_tpl_stop_overshoot, "stop_overshoot", "control"),
+    (_tpl_fail_yield_ped, "fail_yield_pedestrian", "data_label"),
+    (_tpl_legal_merge, "legal_assertive_merge", None),
+    (_tpl_yield_with_row, "yield_with_right_of_way", None),
+    (_tpl_green_proceed, "green_proceed", None),
+]
+
+VISIBILITIES = ["low", "clear", "clear"]
+LIGHTINGS = ["day", "dusk", "night"]
+WEATHERS = ["clear", "clear", "rain"]
+
+
+def generate_bank(n_scenarios: int = 28, seed: int = 7,
+                  model_version: str = DEFAULT_MODEL) -> List[ROTRScenario]:
+    """Deterministic scenario bank; same inputs -> identical output."""
+    if model_version not in MODEL_PROFILES:
+        raise ValueError(f"unknown model_version {model_version!r}; known: "
+                         f"{sorted(MODEL_PROFILES)}")
+    profile = MODEL_PROFILES[model_version]
+    bank_id = bank_id_for(n_scenarios, seed, model_version)
+    out: List[ROTRScenario] = []
+    for i in range(n_scenarios):
+        # Same per-sequence seeding discipline as bevfusion.scenes.
+        rng = np.random.default_rng(seed * 100003 + i * 7919)
+        tpl_fn, kind, cause = TEMPLATES[i % len(TEMPLATES)]
+        is_opportunity = cause is not None
+        committed = bool(is_opportunity
+                         and rng.random() < profile.get(cause, 1.0))
+        parts = tpl_fn(rng, committed, cause or "")
+        env = Environment(visibility=VISIBILITIES[i % 3],
+                          lighting=LIGHTINGS[i % 3],
+                          weather=WEATHERS[i % 3])
+        sid = f"{bank_id}-sc{i:03d}"
+        out.append(ROTRScenario(
+            scenario_id=sid, bank_id=bank_id, seed=seed,
+            description=parts["description"], environment=env,
+            map_context=parts["map_context"],
+            actual_context=parts["actual_context"],
+            ego=parts["ego"], planned=parts["planned"],
+            actors=parts["actors"],
+            planted=PlantedTruth(
+                kind=kind, is_violation_opportunity=is_opportunity,
+                committed=committed,
+                expected_rule_id=parts["expected_rule"] if committed else None,
+                cause_layer=cause if committed else None,
+                notes=parts["description"]),
+            provenance=Provenance(
+                scenario_id=sid, dataset_version=DATASET_VERSION,
+                model_version=model_version,
+                software_version=f"{SOFTWARE_VERSION}/{GENERATOR_VERSION}",
+                source="SYNTHETIC", confidence=1.0)))
+    return out
+
+
+def bank_id_for(n_scenarios: int, seed: int, model_version: str) -> str:
+    blob = f"{n_scenarios}|{seed}|{model_version}|{GENERATOR_VERSION}"
+    return "bank-" + hashlib.sha256(blob.encode()).hexdigest()[:10]

sensorflow/rotr/stopship.py

@@ -0,0 +1,118 @@
+"""Deterministic catastrophic-case stop-ship trigger.
+
+Explicitly NOT LLM-driven: the trigger is a versioned, hash-addressed
+CONJUNCTION of independent evidence — a vulnerable road user was involved
+AND the perception layer has positive missed-detection evidence AND the
+counterfactual consequence is SAFETY_CRITICAL. Any two of the three do not
+fire (tested). Surrogate measures alone can never fire it.
+
+Integration: when the agentic release machinery is importable (in-progress
+workstream — guarded import), each NO_GO event is FORWARDED to
+agentic.policy.evaluate as an advisory PolicyInput and the policy engine's
+outcome is attached; the local gate record is always written either way,
+so the gate never depends on the concurrent workstream landing.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Dict, List, Optional
+
+from sensorflow.rotr import SOFTWARE_VERSION
+from sensorflow.rotr.models import Provenance, ReleaseGateResult
+
+STOPSHIP_POLICY: Dict = {
+    "policy_name": "rotr-catastrophic-conjunction",
+    "policy_semver": "1.0.0",
+    "conjunction": {
+        "vru_involved": True,
+        "perception_missed_detection_supported": True,
+        "consequence_class": "SAFETY_CRITICAL",
+    },
+    "outcome_on_match": "NO_GO",
+    "notes": "Deterministic conjunction; NOT LLM-driven. Each leg is an "
+             "independent evidence stream (taxonomy, attribution, "
+             "counterfactual replay). Surrogate thresholds alone never fire.",
+}
+
+
+def policy_version(doc: Optional[Dict] = None) -> str:
+    blob = json.dumps(doc or STOPSHIP_POLICY, sort_keys=True)
+    return hashlib.sha256(blob.encode()).hexdigest()[:16]
+
+
+def _leg_vru(signature: Dict[str, str]) -> bool:
+    return signature.get("vulnerability") == "VRU"
+
+
+def _leg_perception_missed(attribution: Dict) -> bool:
+    le = (attribution.get("layers") or {}).get("perception") or {}
+    return le.get("status") == "SUPPORTED" and \
+        "missed detection" in (le.get("evidence") or "")
+
+
+def _leg_safety_critical(consequence: Dict) -> bool:
+    return consequence.get("consequence_class") == "SAFETY_CRITICAL"
+
+
+def evaluate_gate(run_id: str, items: List[Dict],
+                  forward_to_agentic: bool = True) -> ReleaseGateResult:
+    """items: [{violation_id, signature, attribution, consequence}]."""
+    version = policy_version()
+    events: List[Dict] = []
+    for it in items:
+        legs = {
+            "vru_involved": _leg_vru(it.get("signature", {})),
+            "perception_missed_detection_supported":
+                _leg_perception_missed(it.get("attribution", {})),
+            "consequence_safety_critical":
+                _leg_safety_critical(it.get("consequence", {})),
+        }
+        fired = all(legs.values())
+        if fired or sum(legs.values()) >= 2:
+            events.append({"violation_id": it["violation_id"],
+                           "legs": legs, "fired": fired,
+                           "note": "conjunction fired -> NO_GO" if fired
+                           else "partial match only — does NOT fire "
+                                "(conjunction is strict)"})
+    outcome = "NO_GO" if any(e["fired"] for e in events) else "GO"
+
+    advisory = None
+    if forward_to_agentic and outcome == "NO_GO":
+        advisory = _forward_to_agentic(run_id, events)
+
+    return ReleaseGateResult(
+        gate_id=f"{run_id}-gate", run_id=run_id, policy_version=version,
+        outcome=outcome, events=events, agentic_advisory=advisory,
+        provenance=Provenance(
+            software_version=f"{SOFTWARE_VERSION}/stopship-{version}",
+            source="SYNTHETIC"))
+
+
+def _forward_to_agentic(run_id: str, events: List[Dict]) -> Optional[Dict]:
+    """Advisory forwarding to the platform's deterministic policy engine."""
+    try:
+        from sensorflow.agentic.policy import PolicyInput, evaluate
+
+        fired = next(e for e in events if e["fired"])
+        inp = PolicyInput(
+            failure_id=f"rotr-{fired['violation_id']}",
+            safety_critical_class=True,
+            behavioral_evidence="observed_unsafe",
+            functional_impact=True,
+            significant=True,
+            evidence_confidence="CONFIRMED",
+            downstream_consequence="unsafe",
+            novelty="novel",
+        )
+        result = evaluate(inp, actor="rotr-stopship")
+        return {"engine": "sensorflow.agentic.policy",
+                "outcome": result.get("outcome"),
+                "severity": result.get("severity"),
+                "policy_version": result.get("policy_version"),
+                "recommended_option": result.get("recommended_option")}
+    except Exception as e:
+        return {"engine": "unavailable",
+                "note": f"agentic policy engine not importable/usable ({e}); "
+                        "local gate record stands alone"}

sensorflow/rotr/store.py

@@ -0,0 +1,54 @@
+"""File persistence under runs/rotr/ (platform convention: request-scoped
+FastAPI over JSON files; no services). `set_rotr_root` is the test hook,
+mirroring set_safety_root / set_nextgen_root etc."""
+
+from __future__ import annotations
+
+import json
+import os
+import threading
+from typing import Any, Dict, List, Optional
+
+_ROOT = os.path.join("runs", "rotr")
+_ROOT_LOCK = threading.Lock()
+
+
+def rotr_root() -> str:
+    return _ROOT
+
+
+def set_rotr_root(path: str) -> None:
+    global _ROOT
+    with _ROOT_LOCK:
+        _ROOT = str(path)
+
+
+def _path(*parts: str) -> str:
+    return os.path.join(rotr_root(), *parts)
+
+
+def write_json(obj: Any, *parts: str) -> str:
+    path = _path(*parts)
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "w") as f:
+        json.dump(obj, f, indent=1, default=str)
+    return path
+
+
+def read_json(*parts: str) -> Optional[Dict]:
+    path = _path(*parts)
+    if not os.path.exists(path):
+        return None
+    with open(path) as f:
+        return json.load(f)
+
+
+def list_dir(*parts: str) -> List[str]:
+    path = _path(*parts)
+    if not os.path.isdir(path):
+        return []
+    return sorted(os.listdir(path))
+
+
+def exists(*parts: str) -> bool:
+    return os.path.exists(_path(*parts))

sensorflow/rotr/taxonomy.py

@@ -0,0 +1,207 @@
+"""Six-axis violation taxonomy, structured query, and clustering.
+
+Axes: Actor / Vulnerability / Legality / Environment / Interaction /
+Behavior, plus extensions RoadGeometry / TrafficControl / Visibility.
+
+The structured query turns a natural-language request such as "failed to
+yield to pedestrian at uncontrolled intersection during low visibility"
+into a filterable query object via a DETERMINISTIC keyword map (no LLM in
+the retrieval path), then filters violation signatures exactly.
+
+Clustering mirrors raremine's mining semantics conceptually (signature ->
+dedup -> diversity-aware exemplar) over ROTR-native structured signatures —
+own implementation, raremine's store is untouched. Exact structured
+matching over ~10^3 records needs a dict, not a vector DB (see the
+DO-NOT-BUILD decision in the architecture doc).
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Dict, List, Optional
+
+from pydantic import BaseModel
+
+from sensorflow.rotr.models import ROTRScenario, ROTRViolation
+
+TAXONOMY_VERSION = "rotr-taxonomy-1.0.0"
+
+AXES = ["actor", "vulnerability", "legality", "environment", "interaction",
+        "behavior", "road_geometry", "traffic_control", "visibility"]
+
+RULE_TO_LEGALITY = {
+    "R-YIELD-PED-01": "YIELD",
+    "R-PATH-RESTRICT-02": "RESTRICTED_PATH",
+    "R-LANE-MANEUVER-03": "LANE_ASSOCIATION",
+    "R-INT-CONFLICT-04": "SIGNAL",
+    "R-MERGE-GAP-05": "MERGE",
+    "R-STOP-06": "STOP",
+}
+
+RULE_TO_BEHAVIOR = {
+    "R-YIELD-PED-01": "proceed_without_yield",
+    "R-PATH-RESTRICT-02": "enter_restricted_path",
+    "R-LANE-MANEUVER-03": "unpermitted_maneuver",
+    "R-INT-CONFLICT-04": "run_red_signal",
+    "R-MERGE-GAP-05": "insufficient_gap_merge",
+    "R-STOP-06": "rolling_stop",
+}
+
+
+def signature(violation: ROTRViolation, scenario: ROTRScenario) -> Dict[str, str]:
+    material = [a for a in scenario.actors
+                if a.actor_id in set(violation.actor_ids)]
+    actor_cls = material[0].class_name if material else "none"
+    vru = any(a.class_name in ("pedestrian", "cyclist") for a in material)
+    interaction = "NONE"
+    if material:
+        intent = material[0].intent
+        interaction = {"CROSSING": "CROSSING", "MERGING": "MERGING",
+                       "PROCEEDING": "FOLLOWING"}.get(intent, "OTHER")
+    env = scenario.environment
+    ctx = scenario.actual_context
+    return {
+        "actor": actor_cls,
+        "vulnerability": "VRU" if vru else "NON_VRU",
+        "legality": RULE_TO_LEGALITY.get(violation.rule_id, "OTHER"),
+        "environment": f"{env.lighting}/{env.weather}",
+        "interaction": interaction,
+        "behavior": RULE_TO_BEHAVIOR.get(violation.rule_id, "other"),
+        "road_geometry": ctx.intersection_type,
+        "traffic_control": ctx.control,
+        "visibility": env.visibility,
+    }
+
+
+# ------------------------------------------------------------ structured query
+
+
+class ROTRQuery(BaseModel):
+    """Filterable query object over the taxonomy axes (all optional)."""
+
+    actor: Optional[str] = None
+    vulnerability: Optional[str] = None
+    legality: Optional[str] = None
+    environment: Optional[str] = None
+    interaction: Optional[str] = None
+    behavior: Optional[str] = None
+    road_geometry: Optional[str] = None
+    traffic_control: Optional[str] = None
+    visibility: Optional[str] = None
+    lighting: Optional[str] = None
+    weather: Optional[str] = None
+    consequence_class: Optional[str] = None
+    primary_layer: Optional[str] = None
+    text: Optional[str] = None            # original text, for provenance
+
+
+# Deterministic keyword -> axis map (retrieval path is LLM-free).
+_KEYWORDS = [
+    ("pedestrian", "actor", "pedestrian"),
+    ("cyclist", "actor", "cyclist"),
+    ("vulnerable", "vulnerability", "VRU"),
+    ("yield", "legality", "YIELD"),
+    ("restricted", "legality", "RESTRICTED_PATH"),
+    ("bus lane", "legality", "RESTRICTED_PATH"),
+    ("wrong lane", "legality", "LANE_ASSOCIATION"),
+    ("lane association", "legality", "LANE_ASSOCIATION"),
+    ("red light", "legality", "SIGNAL"),
+    ("signal", "legality", "SIGNAL"),
+    ("merge", "legality", "MERGE"),
+    ("stop sign", "legality", "STOP"),
+    ("rolling stop", "legality", "STOP"),
+    ("uncontrolled intersection", "road_geometry", "uncontrolled"),
+    ("controlled intersection", "road_geometry", "controlled"),
+    ("intersection", "road_geometry", None),   # weak; only if none set
+    ("low visibility", "visibility", "low"),
+    ("night", "lighting", "night"),
+    ("dusk", "lighting", "dusk"),
+    ("rain", "weather", "rain"),
+    ("crossing", "interaction", "CROSSING"),
+    ("safety critical", "consequence_class", "SAFETY_CRITICAL"),
+]
+
+
+def parse_query(text: str) -> ROTRQuery:
+    low = text.lower()
+    fields: Dict[str, str] = {}
+    for kw, axis, value in _KEYWORDS:
+        if kw in low and value is not None and axis not in fields:
+            fields[axis] = value
+    return ROTRQuery(text=text, **fields)
+
+
+def matches(query: ROTRQuery, sig: Dict[str, str],
+            consequence_class: Optional[str],
+            primary_layer: Optional[str],
+            environment: Dict[str, str]) -> bool:
+    for axis in AXES:
+        want = getattr(query, axis)
+        if want is not None and sig.get(axis) != want:
+            return False
+    if query.lighting is not None and environment.get("lighting") != query.lighting:
+        return False
+    if query.weather is not None and environment.get("weather") != query.weather:
+        return False
+    if query.consequence_class is not None and \
+            consequence_class != query.consequence_class:
+        return False
+    if query.primary_layer is not None and primary_layer != query.primary_layer:
+        return False
+    return True
+
+
+# ------------------------------------------------------------ clustering
+
+
+CLUSTER_KEY_AXES = ["legality", "actor", "traffic_control", "behavior"]
+
+
+def cluster_key(sig: Dict[str, str], primary_layer: Optional[str]) -> str:
+    parts = [sig.get(a, "?") for a in CLUSTER_KEY_AXES] + [primary_layer or "?"]
+    return "|".join(parts)
+
+
+def cluster_id_for(key: str) -> str:
+    return "cl-" + hashlib.sha256(
+        (TAXONOMY_VERSION + "|" + key).encode()).hexdigest()[:8]
+
+
+def build_clusters(items: List[Dict]) -> List[Dict]:
+    """items: [{violation_id, signature, primary_layer, consequence_class,
+    environment}] -> recurring-pattern clusters with a diversity-aware
+    exemplar (max environment spread, raremine-style selection concept)."""
+    groups: Dict[str, List[Dict]] = {}
+    for it in items:
+        key = cluster_key(it["signature"], it.get("primary_layer"))
+        groups.setdefault(key, []).append(it)
+    clusters = []
+    for key, members in sorted(groups.items()):
+        envs = sorted({m["signature"].get("environment", "") +
+                       "/" + m["signature"].get("visibility", "")
+                       for m in members})
+        cons: Dict[str, int] = {}
+        for m in members:
+            c = m.get("consequence_class") or "UNCLASSIFIED"
+            cons[c] = cons.get(c, 0) + 1
+        # Exemplar: the member in the least-represented environment cell
+        # (diversity-aware pick, not just the first).
+        env_counts: Dict[str, int] = {}
+        for m in members:
+            e = m["signature"].get("environment", "")
+            env_counts[e] = env_counts.get(e, 0) + 1
+        exemplar = min(members,
+                       key=lambda m: (env_counts[m["signature"].get("environment", "")],
+                                      m["violation_id"]))
+        clusters.append({
+            "cluster_id": cluster_id_for(key),
+            "key": key,
+            "count": len(members),
+            "member_violation_ids": [m["violation_id"] for m in members],
+            "exemplar_violation_id": exemplar["violation_id"],
+            "environment_spread": envs,
+            "consequence_distribution": cons,
+            "taxonomy_version": TAXONOMY_VERSION,
+        })
+    clusters.sort(key=lambda c: -c["count"])
+    return clusters