"""Final investigation report: markdown + JSON assembly.""" from __future__ import annotations import time from typing import Dict, List, Optional from sensorflow.rca import diagnostics as dg, scoring from sensorflow.rca.models import (HYPOTHESIS_LABELS, Investigation, STAGES) from sensorflow.rca.scenario import ScenarioBundle _REMEDIATION: Dict[str, Dict[str, List[str]]] = { "TRUE_MODEL_REGRESSION": { "containment": ["Halt the rollout / keep B out of the serving path", "Pin A as the serving model; keep B in shadow only"], "short_term": ["Slice the regression by segment and error type to " "find what B lost", "Fine-tune B on recent-window data covering the " "regressing segments"], "long_term": ["Add a recent-window eval set to the offline gate", "Automate freshness checks on eval data before any " "launch decision"], }, "DISTRIBUTION_SHIFT": { "containment": ["Do not treat either aggregate as a launch signal; " "gate on segment-level deltas"], "short_term": ["Reweight offline eval to the production mix and " "re-decide", "Target B's night/urban weakness with " "focused training data"], "long_term": ["Continuously monitor production mix vs eval mix (PSI " "alerting)", "Stratified launch gates per segment"], }, "FEATURE_SKEW": { "containment": ["Freeze the rollout; the online model is consuming " "corrupted features"], "short_term": ["Fix the unit/normalization difference in the online " "feature pipeline; add a unit test on the skewed " "feature", "Re-run shadow after the fix"], "long_term": ["Continuous training-serving feature parity monitoring " "with per-feature SMD alerts", "Version-pin and log feature pipeline versions in both " "environments"], }, "SERVING_MISMATCH": { "containment": ["Align shadow serving config (threshold, precision) " "with the evaluated artifact before reading any " "shadow number"], "short_term": ["Re-run shadow with the offline config; if the gap " "closes, ship config alignment", "Calibrate B's confidence to the serving threshold"], "long_term": ["Config parity checks in CI between eval harness and " "serving stack", "One artifact, one config: promote the " "exact evaluated bundle"], }, "LABEL_LATENCY": { "containment": ["Suspend shadow-based verdicts until labels mature"], "short_term": ["Re-grade the frozen cohort post-maturity", "Exclude provisional labels from headline metrics"], "long_term": ["Separate provisional and mature metrics in dashboards", "Track label-age distribution as a first-class metric"], }, "SAMPLING_BIAS": { "containment": ["Stop reading the triage-sampled stream as an " "unbiased estimate"], "short_term": ["Add a parallel uniform sample for measurement", "Reweight existing sample by inclusion probability " "(if logged)"], "long_term": ["Log inclusion probabilities for every sampled unit", "Keep measurement sampling separate from triage " "sampling"], }, "STATISTICAL_NOISE": { "containment": ["Make no launch decision on current evidence; both " "+5% and -2% are within noise"], "short_term": ["Extend collection to the pre-computed effective-n " "target", "Use paired, cluster-aware tests only"], "long_term": ["Pre-register power calculations before shadow " "experiments", "Adopt anytime-valid sequential tests " "(seqeval) to allow early stopping without p-hacking"], }, "OFFLINE_CONTAMINATION": { "containment": ["Retract the +5% claim; do not use the current " "offline set for any decision"], "short_term": ["Rebuild the eval set with entity-level dedup against " "B's training data; re-run both models", "Pin versions and record environment locks"], "long_term": ["Automated leakage scans in the eval-set build " "pipeline", "Entity-level splitting enforced by " "tooling, not convention"], }, "NO_PRACTICAL_REGRESSION": { "containment": ["No containment needed"], "short_term": ["Proceed with a guarded rollout with segment-level " "monitoring"], "long_term": ["Tighten practical-margin definitions per surface"], }, } _STAGE_HEADLINE = { "comparison_validity": "Are the two numbers commensurable?", "offline_audit": "Is the +5% real and reproducible?", "population_validation": "Enough volume, comparable populations?", "distribution_shift": "Did the world move between the two evals?", "conditional_performance": "Where does the delta live?", "paired_comparison": "Which units actually flipped?", "statistical_significance": "Is -2% distinguishable from noise?", "feature_parity": "Same features online and offline?", "serving_parity": "Same config online and offline?", "shadow_traffic": "Is the scored sample fair?", "label_integrity": "Is shadow ground truth actually true?", } def build_report(bundle: ScenarioBundle, inv: Investigation, battery: Optional[Dict] = None) -> Dict: battery = battery or dg.run_all(bundle) scoreboard = scoring.build_scoreboard(bundle, inv, battery=battery) tree = scoring.evaluate_decision_tree(bundle, battery=battery) experiments = scoring.recommend_experiments(bundle, scoreboard, battery) top = scoreboard["rows"][0] conclusion = tree["conclusion"] remediation = _REMEDIATION.get(conclusion or top["hypothesis"], _REMEDIATION[top["hypothesis"]]) stage_summaries = [] for key, (data, findings) in battery.items(): mism = [f for f in findings if f.status == "MISMATCH"] unk = [f for f in findings if f.status == "UNKNOWN"] stage_summaries.append({ "stage": key, "headline": _STAGE_HEADLINE.get(key, key), "mismatches": [f.title for f in mism], "unknowns": [f.title for f in unk], "n_pass": sum(1 for f in findings if f.status == "PASS"), "verdict": ("mismatch" if mism else "unknown" if unk else "clean"), }) acked = [e for e in inv.events if e["kind"] == "unknowns_acknowledged"] payload = { "investigation_id": inv.id, "name": inv.name, "generated_at": time.time(), "models": {"baseline": inv.baseline_model, "candidate": inv.candidate_model}, "claims": inv.claims, "executive_finding": { "conclusion": conclusion, "conclusion_kind": tree["conclusion_kind"], "top_hypothesis": top["hypothesis"], "label": HYPOTHESIS_LABELS.get(top["hypothesis"], top["hypothesis"]), "confidence": top["auto_confidence"], "score": top["score"], "score_gap_to_runner_up": scoreboard["score_gap"], }, "stage_summaries": stage_summaries, "hypothesis_ranking": [ {"rank": r["rank"], "hypothesis": r["hypothesis"], "score": r["score"], "confidence": r["auto_confidence"], "human_confidence": r["human_confidence"], "evidence_for": len(r["evidence_for"]), "evidence_against": len(r["evidence_against"])} for r in scoreboard["rows"]], "decision_path": tree["path"], "minimum_additional_evidence": experiments["minimum_additional_evidence"], "recommended_experiments": experiments["experiments"], "remediation": remediation, "acknowledged_unknowns": acked, "human_findings": [f.to_dict() for f in inv.findings if f.source == "human"], } payload["markdown"] = _to_markdown(payload, inv) return payload def _to_markdown(p: Dict, inv: Investigation) -> str: ef = p["executive_finding"] lines = [ f"# RCA Report — {p['name']}", "", f"**Baseline:** `{p['models']['baseline']}`   " f"**Candidate:** `{p['models']['candidate']}`", "", f"**Claims under investigation:** offline " f"{p['claims'].get('offline_delta_pp', '+5')}pp vs shadow " f"{p['claims'].get('shadow_delta_pp', '-2')}pp " f"({p['claims'].get('metric', 'accuracy')})", "", "## Executive finding", "", f"**{ef['conclusion'] or ef['top_hypothesis']}** — " f"{ef['label']} (confidence: **{ef['confidence']}**, score " f"{ef['score']}, gap to runner-up {ef['score_gap_to_runner_up']}).", "", "## Hypothesis ranking", "", "| # | Hypothesis | Score | Confidence | For | Against |", "|---|-----------|-------|------------|-----|---------|", ] for r in p["hypothesis_ranking"]: hc = f" (human: {r['human_confidence']})" if r["human_confidence"] else "" lines.append(f"| {r['rank']} | {r['hypothesis']} | {r['score']} | " f"{r['confidence']}{hc} | {r['evidence_for']} | " f"{r['evidence_against']} |") lines += ["", "## Stage summaries", ""] for s in p["stage_summaries"]: badge = {"mismatch": "MISMATCH", "unknown": "UNKNOWN", "clean": "clean"}[s["verdict"]] lines.append(f"- **{s['stage']}** ({s['headline']}) — {badge}") for m in s["mismatches"]: lines.append(f" - ⚠ {m}") for u in s["unknowns"]: lines.append(f" - ? {u}") if p["acknowledged_unknowns"]: lines += ["", "## Acknowledged unknowns", ""] for e in p["acknowledged_unknowns"]: lines.append(f"- {e['message']} — note: " f"{e['data'].get('note', '')}") lines += ["", "## Minimum additional evidence", "", p["minimum_additional_evidence"], "", "## Recommended experiments", ""] for d in p["recommended_experiments"]: lines.append(f"{d['rank']}. **{d['design']}** (cost: {d['cost']}, " f"gain {d['information_gain']}) — {d['description']}") lines += ["", "## Remediation", "", "**Containment (now):**"] lines += [f"- {x}" for x in p["remediation"]["containment"]] lines += ["", "**Short term:**"] lines += [f"- {x}" for x in p["remediation"]["short_term"]] lines += ["", "**Long term:**"] lines += [f"- {x}" for x in p["remediation"]["long_term"]] return "\n".join(lines)