"""Layered release gating + Safety Evidence Package. Industry concept: staged release gates for AV perception stacks, producing an auditable evidence trail structured after three families of standards: - ISO 26262 (functional safety): deterministic, repeatable verification — geometric plausibility checks, calibration validation, regression testing against a pinned baseline with pinned thresholds. - ISO 21448 (SOTIF): performance-limitation risk — ODD coverage of the scenario space, under-covered cells, surrogate-safety (SSAM-style) analysis of near-miss severity. - UL 4600 (safety case): the Release Readiness Gate compiles all gate evidence, methodology and lineage into a claim/evidence-structured Safety Evidence Package (JSON + rendered markdown). IMPORTANT WORDING: the package SUPPORTS a safety case by organizing verifiable evidence. It does NOT certify or demonstrate compliance with any standard, and several inputs are simulated (marked per-record). Gates (evaluated in order, block-on-fail; every gate always produces an evidence record even when an earlier gate blocked): 1. scenario_quality — geometric plausibility (evaluation.validation) + multi-sensor calibration validation (safety.calibration) 2. coverage — ODD combinatorial coverage thresholds (safety.odd) 3. regression — model-vs-model compare (megaeval.analysis.compare_runs, not duplicated here) 4. safety — surrogate-safety: aggregate CSI must not increase vs baseline beyond tolerance (safety.ssam_ext) 5. release_readiness — all of the above + compiles the Safety Evidence Package """ from __future__ import annotations from datetime import datetime, timezone from typing import Dict, List, Optional from sensorflow.safety import calibration as calib_mod from sensorflow.safety import odd as odd_mod from sensorflow.safety import ssam_ext from sensorflow.safety.store import read_json, write_json DEFAULT_POLICY: Dict = { "scenario_quality": { "enabled": True, "min_geometric_pass_rate": 0.70, "geometric_sample_size": 24, "block_on_miscalibration": True, }, "coverage": { "enabled": True, "dims": ["weather", "lighting", "road_type"], "min_samples": 50, "max_ci_width": 0.25, "min_coverage_rate": 0.60, "min_production_weighted_coverage": 0.80, }, "regression": { "enabled": True, # thresholds delegated to megaeval.analysis.DEFAULT_PROMOTION_POLICY; # entries here override it (e.g. max_recall_drop) "overrides": {}, }, "safety": { "enabled": True, "max_csi_increase_ratio": 0.15, # candidate CSI <= baseline * (1 + ratio) "max_csi_absolute": None, # optional hard ceiling }, } _POLICY_FILE = ("gate_policy.json",) def _now() -> str: return datetime.now(timezone.utc).isoformat() def _merge(base: Dict, override: Dict) -> Dict: out = dict(base) for k, v in (override or {}).items(): if isinstance(v, dict) and isinstance(out.get(k), dict): out[k] = _merge(out[k], v) else: out[k] = v return out def get_policy() -> Dict: stored = read_json(*_POLICY_FILE) or {} return _merge(DEFAULT_POLICY, stored) def set_policy(overrides: Dict) -> Dict: stored = _merge(read_json(*_POLICY_FILE) or {}, overrides or {}) write_json(stored, *_POLICY_FILE) return get_policy() def _check(name: str, actual, threshold, passed: bool, direction: str = "") -> Dict: return {"check": name, "actual": actual, "threshold": threshold, "direction": direction, "passed": bool(passed)} def _gate_record(gate: str, name: str, standard_refs: List[str], checks: List[Dict], evidence: Dict, skipped: bool = False, notes: str = "") -> Dict: if skipped: status = "SKIPPED" else: status = "PASS" if all(c["passed"] for c in checks) else "BLOCK" return {"gate": gate, "name": name, "status": status, "standard_refs": standard_refs, "checks": checks, "evidence": evidence, "notes": notes, "evaluated_at": _now()} # ------------------------------------------------------------------ gate 1 def scenario_quality_gate(policy: Dict, eval_store=None) -> Dict: """Geometric plausibility sample + calibration validation. Geometric part: validates a deterministic sample of auto-label annotations from the newest labeleval dataset (generating a small probe dataset if the store is empty) through the existing strict validation engine. Calibration part: latest calibration validation status, refreshed with a clean run if none exists. """ p = policy["scenario_quality"] if not p.get("enabled", True): return _gate_record("scenario_quality", "Scenario Quality Gate", ["ISO 26262-style deterministic verification"], [], {}, skipped=True, notes="disabled by policy") from sensorflow.evaluation import synthetic from sensorflow.evaluation.records import get_store from sensorflow.evaluation.validation import validate_annotation store = eval_store or get_store() datasets = store.all("datasets") probe_generated = False if datasets: ds = max(datasets, key=lambda d: d.created_at) else: ds = synthetic.generate_dataset(store, name="gate-probe", num_sequences=2, frames_per_sequence=6, seed=101) synthetic.generate_labels(store, ds, seed_offset=1) probe_generated = True anns = sorted(store.where("annotations", dataset_id=ds.dataset_id), key=lambda a: a.annotation_id) sample = anns[:max(1, int(p["geometric_sample_size"]))] frames = {f.frame_id: f for f in store.where("frames", dataset_id=ds.dataset_id)} points_cache: Dict[str, object] = {} n_pass = 0 fail_hist: Dict[str, int] = {} for ann in sample: frame = frames.get(ann.frame_id) if frame is None: continue if ann.frame_id not in points_cache: points_cache[ann.frame_id] = synthetic.frame_points(store, frame) res = validate_annotation(store, ann, frame, points=points_cache[ann.frame_id]) if res.passed: n_pass += 1 else: for c in res.checks: if c.applicable and not c.passed: fail_hist[c.gate] = fail_hist.get(c.gate, 0) + 1 pass_rate = n_pass / max(len(sample), 1) calib = calib_mod.latest_status() if calib is None: calib = calib_mod.run_validation(mode="clean", persist=True) calib_ok = (calib["status"] != "MISCALIBRATED" if p["block_on_miscalibration"] else True) checks = [ _check("geometric_pass_rate", round(pass_rate, 4), p["min_geometric_pass_rate"], pass_rate >= p["min_geometric_pass_rate"], ">="), _check("calibration_status", calib["status"], "not MISCALIBRATED", calib_ok, "=="), ] return _gate_record( "scenario_quality", "Scenario Quality Gate", ["ISO 26262-style deterministic verification (geometric plausibility, " "calibration validation)"], checks, evidence={ "dataset_id": ds.dataset_id, "probe_dataset_generated": probe_generated, "annotations_sampled": len(sample), "annotations_passed": n_pass, "failing_check_histogram": fail_hist, "calibration": {"status": calib["status"], "metrics": calib.get("metrics"), "mode": calib.get("mode"), "diagnosis": calib.get("diagnosis")}, }, notes="geometric checks run on the strict validation engine; calibration " "scenes are simulated (see safety.calibration)") # ------------------------------------------------------------------ gate 2 def coverage_gate(mega_store, candidate, policy: Dict) -> Dict: p = policy["coverage"] if not p.get("enabled", True): return _gate_record("coverage", "Coverage Gate", ["ISO 21448/SOTIF-style ODD coverage"], [], {}, skipped=True, notes="disabled by policy") cov = odd_mod.coverage_for_run( mega_store, candidate, dims=p["dims"], min_samples=p["min_samples"], max_ci_width=p["max_ci_width"], max_gaps=10) s = cov["summary"] checks = [ _check("coverage_rate", s["coverage_rate"], p["min_coverage_rate"], s["coverage_rate"] >= p["min_coverage_rate"], ">="), _check("production_weighted_coverage", s["production_weighted_coverage"], p["min_production_weighted_coverage"], s["production_weighted_coverage"] >= p["min_production_weighted_coverage"], ">="), ] return _gate_record( "coverage", "Coverage Gate", ["ISO 21448/SOTIF-style performance-limitation risk (ODD coverage per " "ISO 34503-inspired taxonomy)"], checks, evidence={"dims": cov["dims"], "thresholds": cov["thresholds"], "summary": s, "top_gaps": cov["gaps"][:5], "fill_requests_available": len(cov["fill_requests"])}, notes="coverage over exact metric-cube counts; production shares use a " "marginal-independence assumption") # ------------------------------------------------------------------ gate 3 def regression_gate(mega_store, candidate, baseline, policy: Dict) -> Dict: p = policy["regression"] if not p.get("enabled", True): return _gate_record("regression", "Regression Gate", ["ISO 26262-style regression verification"], [], {}, skipped=True, notes="disabled by policy") from sensorflow.megaeval.analysis import compare_runs cmp = compare_runs(mega_store, candidate, baseline, policy=p.get("overrides") or None) ok = cmp["recommendation"] == "PROMOTE" checks = [_check("compare_recommendation", cmp["recommendation"], "PROMOTE", ok, "==")] return _gate_record( "regression", "Regression Gate", ["ISO 26262-style deterministic regression testing vs pinned baseline"], checks, evidence={"headline_deltas": cmp["headline_deltas"], "regressions": cmp["regressions"], "blockers": cmp["blockers"], "policy": cmp["policy"], "worst_cohorts": cmp["worst_cohorts"][:5]}, notes="delegates to megaeval model-vs-model compare (single source of truth)") # ------------------------------------------------------------------ gate 4 def safety_gate(candidate, baseline, policy: Dict) -> Dict: p = policy["safety"] if not p.get("enabled", True): return _gate_record("safety", "Safety Gate", ["ISO 21448/SOTIF-style surrogate safety"], [], {}, skipped=True, notes="disabled by policy") cand = ssam_ext.csi_for_run(candidate) base = ssam_ext.csi_for_run(baseline) cand_csi, base_csi = cand["aggregate_csi"], base["aggregate_csi"] limit = base_csi * (1.0 + p["max_csi_increase_ratio"]) increase_ratio = ((cand_csi - base_csi) / base_csi) if base_csi > 1e-9 else ( 0.0 if cand_csi <= 1e-9 else float("inf")) checks = [ _check("aggregate_csi_vs_baseline", round(cand_csi, 4), round(limit, 4), cand_csi <= limit + 1e-9, "<="), ] if p.get("max_csi_absolute") is not None: checks.append(_check("aggregate_csi_absolute", round(cand_csi, 4), p["max_csi_absolute"], cand_csi <= p["max_csi_absolute"], "<=")) return _gate_record( "safety", "Safety Gate", ["ISO 21448/SOTIF-style surrogate safety (FHWA SSAM: TTC/PET/DRAC/CSI)"], checks, evidence={ "candidate_csi": cand_csi, "baseline_csi": base_csi, "csi_increase_ratio": (round(increase_ratio, 4) if increase_ratio != float("inf") else "inf"), "candidate_reaction_model": cand["reaction_model"], "baseline_reaction_model": base["reaction_model"], "candidate_min_ttc_s": cand["min_ttc_s"], "candidate_scenarios": cand["scenarios"], }, notes="CSI from a deterministic simulated scenario suite conditioned on the " "model's perception profile (marked simulated); the SSAM formulas " "themselves are real math") # ------------------------------------------------------------------ orchestration GATE_ORDER = ["scenario_quality", "coverage", "regression", "safety"] def evaluate_gates(mega_store, candidate, baseline, policy_overrides: Optional[Dict] = None, eval_store=None) -> Dict: """Run all gates for candidate vs baseline; compile + persist the Safety Evidence Package. Layered semantics: overall decision blocks on the first failing gate, but every gate is still evaluated for evidence.""" policy = _merge(get_policy(), policy_overrides or {}) gates = [ scenario_quality_gate(policy, eval_store=eval_store), coverage_gate(mega_store, candidate, policy), regression_gate(mega_store, candidate, baseline, policy), safety_gate(candidate, baseline, policy), ] blocking = [g["gate"] for g in gates if g["status"] == "BLOCK"] release_ready = not blocking readiness = _gate_record( "release_readiness", "Release Readiness Gate", ["UL 4600-style safety-case evidence compilation"], [_check("all_gates_pass", f"{len(gates) - len(blocking)}/{len(gates)}", f"{len(gates)}/{len(gates)}", release_ready, "==")], evidence={"blocking_gates": blocking, "first_blocking_gate": blocking[0] if blocking else None}, notes="compiles the Safety Evidence Package below") gates.append(readiness) package = build_evidence_package(mega_store, candidate, baseline, policy, gates, release_ready, blocking) write_json(package, "evidence", f"{candidate.run_id}.json") from sensorflow.safety.store import safety_path md = render_markdown(package) with open(safety_path("evidence", f"{candidate.run_id}.md"), "w") as f: f.write(md) result = { "candidate_run_id": candidate.run_id, "baseline_run_id": baseline.run_id, "decision": "RELEASE_READY" if release_ready else "BLOCKED", "blocking_gates": blocking, "gates": gates, "policy": policy, "evidence_package_id": package["package_id"], "evaluated_at": _now(), } write_json(result, "gates", f"{candidate.run_id}.json") return result def latest_gate_result(run_id: str) -> Optional[Dict]: return read_json("gates", f"{run_id}.json") def load_evidence(run_id: str) -> Optional[Dict]: return read_json("evidence", f"{run_id}.json") # ------------------------------------------------------------------ evidence package DISCLAIMER = ( "This package organizes verifiable evidence to SUPPORT a safety case. It does " "not certify, demonstrate or imply compliance with ISO 26262, ISO 21448, " "UL 4600 or any other standard. Several evidence sources are simulated and " "are marked as such on the individual records.") STANDARD_MAPPINGS = [ {"standard": "ISO 26262 (functional safety)", "style": "deterministic, repeatable verification with pinned thresholds", "evidence": ["gates.scenario_quality (geometric plausibility, calibration)", "gates.regression (model-vs-model compare vs pinned baseline)", "lineage (pinned dataset/model/label/metric/code versions + seed)"]}, {"standard": "ISO 21448 / SOTIF (safety of the intended functionality)", "style": "performance-limitation risk in the operational domain", "evidence": ["gates.coverage + odd_coverage_summary (under-covered ODD cells, " "risk-ranked gaps)", "gates.safety + surrogate_safety_analysis (SSAM-style near-miss " "severity, CSI vs baseline)"]}, {"standard": "UL 4600 (safety case for autonomous products)", "style": "claim -> argument -> evidence structure", "evidence": ["decision + gates[] (claims with thresholds, actuals and " "pass/block outcomes)", "sampling_methodology (statistical argument: stratified " "risk-weighted review, Wilson CIs)", "the package itself as the evidence index"]}, ] def build_evidence_package(mega_store, candidate, baseline, policy: Dict, gates: List[Dict], release_ready: bool, blocking: List[str]) -> Dict: cov_gate = next(g for g in gates if g["gate"] == "coverage") reg_gate = next(g for g in gates if g["gate"] == "regression") saf_gate = next(g for g in gates if g["gate"] == "safety") review = None try: review = mega_store.artifacts(candidate.run_id).get("review") except Exception: pass supplementary: Dict = {} try: from sensorflow.safety.scenario_db import get_db supplementary["scenario_db"] = get_db().counts() except Exception: pass try: cal = calib_mod.latest_status() if cal: supplementary["calibration_status"] = { "status": cal["status"], "metrics": cal.get("metrics")} except Exception: pass return { "package_id": f"sep-{candidate.run_id}", "package_format": "sensorflow-safety-evidence/v1", "generated_at": _now(), "disclaimer": DISCLAIMER, "candidate": {"run_id": candidate.run_id, "model_version": candidate.model_version, "population_id": candidate.population_id, "headline": candidate.headline}, "baseline": {"run_id": baseline.run_id, "model_version": baseline.model_version, "headline": baseline.headline}, "decision": {"release_ready": release_ready, "blocking_gates": blocking}, "gates": gates, "gate_policy": policy, "odd_coverage_summary": cov_gate["evidence"], "regression_analysis": reg_gate["evidence"], "surrogate_safety_analysis": saf_gate["evidence"], "sampling_methodology": { "config": candidate.lineage.get("sampling_config"), "confidence_intervals": "Wilson score intervals on per-cell and " "review-sample proportions", "review_artifact": ({"target_n": review.get("target_n"), "achieved_n": review.get("achieved_n")} if isinstance(review, dict) else None), }, "lineage": {"candidate": candidate.lineage, "baseline": baseline.lineage}, "standard_mappings": STANDARD_MAPPINGS, "supplementary": supplementary, } def render_markdown(pkg: Dict) -> str: d = pkg["decision"] lines = [ f"# Safety Evidence Package — {pkg['candidate']['model_version']}", "", f"Package `{pkg['package_id']}` generated {pkg['generated_at']}", "", f"> {pkg['disclaimer']}", "", f"**Decision: {'RELEASE READY' if d['release_ready'] else 'BLOCKED'}**" + (f" — blocking gates: {', '.join(d['blocking_gates'])}" if d["blocking_gates"] else ""), "", f"Candidate `{pkg['candidate']['run_id']}` ({pkg['candidate']['model_version']}) " f"vs baseline `{pkg['baseline']['run_id']}` ({pkg['baseline']['model_version']})", "", "## Gate results", "", "| Gate | Status | Checks (actual vs threshold) |", "|---|---|---|", ] for g in pkg["gates"]: checks = "; ".join( f"{c['check']}: {c['actual']} {c['direction'] or 'vs'} {c['threshold']} " f"({'ok' if c['passed'] else 'FAIL'})" for c in g["checks"]) or "—" lines.append(f"| {g['name']} | {g['status']} | {checks} |") s = pkg["odd_coverage_summary"].get("summary", {}) lines += [ "", "## ODD coverage (SOTIF-style)", "", f"- dims: {', '.join(pkg['odd_coverage_summary'].get('dims', []))}", f"- adequate cells: {s.get('adequate_cells')}/{s.get('total_cells')} " f"(coverage rate {s.get('coverage_rate')})", f"- production-weighted coverage: {s.get('production_weighted_coverage')}", f"- gap cells: {s.get('gap_cells')}", "", "## Surrogate safety (SSAM-style)", "", f"- candidate aggregate CSI: {pkg['surrogate_safety_analysis'].get('candidate_csi')}", f"- baseline aggregate CSI: {pkg['surrogate_safety_analysis'].get('baseline_csi')}", f"- increase ratio: {pkg['surrogate_safety_analysis'].get('csi_increase_ratio')}", f"- candidate min TTC: {pkg['surrogate_safety_analysis'].get('candidate_min_ttc_s')} s", "", "## Regression analysis", "", ] for h in pkg["regression_analysis"].get("headline_deltas", []): lines.append(f"- {h['metric']}: {h['baseline']} -> {h['candidate']} (Δ {h['delta']:+})") blockers = pkg["regression_analysis"].get("blockers", []) if blockers: lines += ["", "Blockers:"] + [f"- {b}" for b in blockers] lines += [ "", "## Methodology", "", f"- sampling: {pkg['sampling_methodology'].get('config')}", f"- CIs: {pkg['sampling_methodology'].get('confidence_intervals')}", "", "## Standard mappings (evidence organization, not compliance)", "", ] for m in pkg["standard_mappings"]: lines.append(f"### {m['standard']}") lines.append(f"_{m['style']}_") lines += [f"- {e}" for e in m["evidence"]] lines.append("") lines += [ "## Lineage", "", "```json", __import__("json").dumps(pkg["lineage"], indent=2), "```", "", ] return "\n".join(lines)