sensorflow/rca/report.py

"""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)