Deploy Sensorflow Studio (part 2)

tests/test_rca/conftest.py

@@ -0,0 +1,44 @@
+"""Shared fixtures for the RCA workbench tests.
+
+The full diagnostic battery per planted cause is computed once per session
+(it is deterministic), and the API tests run against an isolated runs/rca
+root in a tmp dir.
+"""
+
+from __future__ import annotations
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from sensorflow.rca import diagnostics as dg, scoring, store
+from sensorflow.rca.models import ROOT_CAUSES
+from sensorflow.rca.scenario import generate_scenario
+
+SEED = 7
+
+
+@pytest.fixture(scope="session")
+def batteries():
+    """{cause: (bundle, battery, scoreboard, tree)} computed once."""
+    out = {}
+    for cause in ROOT_CAUSES:
+        bundle = generate_scenario(cause, seed=SEED)
+        battery = dg.run_all(bundle)
+        board = scoring.build_scoreboard(bundle, battery=battery)
+        tree = scoring.evaluate_decision_tree(bundle, battery=battery)
+        out[cause] = (bundle, battery, board, tree)
+    return out
+
+
+@pytest.fixture()
+def rca_client(tmp_path):
+    store.set_rca_root(str(tmp_path / "rca"))
+    from sensorflow.rca import api as rca_api
+    rca_api._battery_cache.clear()
+    app = FastAPI()
+    app.include_router(rca_api.router)
+    with TestClient(app) as client:
+        yield client
+    store.set_rca_root("runs/rca")
+    rca_api._battery_cache.clear()

tests/test_rca/test_api.py

@@ -0,0 +1,167 @@
+"""API lifecycle via TestClient: create -> diagnostics -> stage completion ->
+scoreboard -> decision tree -> experiments -> report -> reveal."""
+
+from __future__ import annotations
+
+
+def _create(client, **kw):
+    body = {"cause": "FEATURE_SKEW", "seed": 7}
+    body.update(kw)
+    res = client.post("/api/rca/investigations", json=body)
+    assert res.status_code == 200, res.text
+    return res.json()
+
+
+def test_meta_routes(rca_client):
+    causes = rca_client.get("/api/rca/causes").json()["causes"]
+    assert len(causes) == 8
+    stages = rca_client.get("/api/rca/stages").json()["stages"]
+    assert len(stages) == 13
+
+
+def test_create_and_list(rca_client):
+    inv = _create(rca_client)
+    assert inv["scenario_cause"] == "FEATURE_SKEW"
+    assert inv["claims"]["offline_delta_pp"] > 4.0
+    assert inv["claims"]["shadow_delta_pp"] < -1.0
+    listed = rca_client.get("/api/rca/investigations").json()["investigations"]
+    assert any(r["id"] == inv["id"] for r in listed)
+
+
+def test_training_mode_hides_cause_until_reveal(rca_client):
+    inv = _create(rca_client, training_mode=True, cause=None, seed=123)
+    assert "scenario_cause" not in inv
+    got = rca_client.get(f"/api/rca/investigations/{inv['id']}").json()
+    assert "scenario_cause" not in got
+    reveal = rca_client.post(f"/api/rca/investigations/{inv['id']}/reveal").json()
+    assert reveal["cause"] in {
+        "TRUE_MODEL_REGRESSION", "DISTRIBUTION_SHIFT", "FEATURE_SKEW",
+        "SERVING_MISMATCH", "LABEL_LATENCY", "SAMPLING_BIAS",
+        "STATISTICAL_NOISE", "OFFLINE_CONTAMINATION"}
+    assert reveal["explanation"]
+    got = rca_client.get(f"/api/rca/investigations/{inv['id']}").json()
+    assert got["scenario_cause"] == reveal["cause"]
+
+
+def test_all_diagnostic_endpoints(rca_client):
+    inv = _create(rca_client)
+    for key in ("comparison_validity", "offline_audit", "population_validation",
+                "distribution_shift", "conditional_performance",
+                "paired_comparison", "statistical_significance",
+                "feature_parity", "serving_parity", "shadow_traffic",
+                "label_integrity"):
+        res = rca_client.get(
+            f"/api/rca/investigations/{inv['id']}/diagnostics/{key}")
+        assert res.status_code == 200, f"{key}: {res.text}"
+        body = res.json()
+        assert body["stage"] == key
+        assert body["data"]
+        assert isinstance(body["findings"], list) and body["findings"]
+    res = rca_client.get(
+        f"/api/rca/investigations/{inv['id']}/diagnostics/nope")
+    assert res.status_code == 404
+
+
+def test_stage_ordering_enforced_via_api(rca_client):
+    inv = _create(rca_client)
+    res = rca_client.post(
+        f"/api/rca/investigations/{inv['id']}/stages/3/complete", json={})
+    assert res.status_code == 409
+    res = rca_client.post(
+        f"/api/rca/investigations/{inv['id']}/stages/0/complete", json={})
+    assert res.status_code == 200
+
+
+def test_unknown_ack_flow_via_api(rca_client):
+    inv = _create(rca_client)  # FEATURE_SKEW -> CV_UNKNOWN:feature_pipeline_version
+    iid = inv["id"]
+    rca_client.get(f"/api/rca/investigations/{iid}/diagnostics/comparison_validity")
+    res = rca_client.post(
+        f"/api/rca/investigations/{iid}/stages/0/complete", json={})
+    assert res.status_code == 409
+    assert "UNKNOWN" in res.json()["detail"]
+    res = rca_client.post(
+        f"/api/rca/investigations/{iid}/stages/0/complete",
+        json={"acknowledge_unknowns": True, "note": "pipeline team paged"})
+    assert res.status_code == 200
+    assert res.json()["stage"]["status"] == "complete_with_unknowns"
+
+
+def test_record_finding_and_scoreboard(rca_client):
+    inv = _create(rca_client)
+    iid = inv["id"]
+    res = rca_client.post(f"/api/rca/investigations/{iid}/findings", json={
+        "stage": "feature_parity", "title": "Confirmed with pipeline team",
+        "status": "MISMATCH", "severity": "CRITICAL",
+        "detail": "fp-2.5 deployed online only",
+        "code": "FP_FEATURE_SKEW:human_confirmed"})
+    assert res.status_code == 200
+    board = rca_client.get(f"/api/rca/investigations/{iid}/scoreboard").json()
+    assert board["rows"][0]["hypothesis"] == "FEATURE_SKEW"
+    # The human finding contributes to the score via the same rule table.
+    top = board["rows"][0]
+    assert any(e["finding_id"] and e["code"] == "FP_FEATURE_SKEW:human_confirmed"
+               for e in top["evidence_for"])
+
+    res = rca_client.post(f"/api/rca/investigations/{iid}/scoreboard/assess",
+                          json={"hypothesis": "FEATURE_SKEW",
+                                "confidence": "HIGH", "note": "confirmed"})
+    assert res.status_code == 200
+    row = next(r for r in res.json()["rows"] if r["hypothesis"] == "FEATURE_SKEW")
+    assert row["human_confidence"] == "HIGH"
+    assert row["auto_confidence"]  # both assessments are kept
+
+
+def test_tree_experiments_report(rca_client):
+    inv = _create(rca_client)
+    iid = inv["id"]
+    tree = rca_client.get(f"/api/rca/investigations/{iid}/decision-tree").json()
+    assert tree["conclusion"] == "FEATURE_SKEW"
+    assert tree["path"]
+    exp = rca_client.get(f"/api/rca/investigations/{iid}/experiments").json()
+    assert exp["minimum_additional_evidence"]
+    rep = rca_client.get(f"/api/rca/investigations/{iid}/report").json()
+    assert rep["executive_finding"]["conclusion"] == "FEATURE_SKEW"
+    assert rep["markdown"].startswith("# RCA Report")
+    assert len(rep["hypothesis_ranking"]) == 8
+    assert len(rep["stage_summaries"]) == 11
+    assert rep["remediation"]["containment"]
+
+
+def test_persistence_roundtrip(rca_client):
+    """Data survives the CSV round trip: fresh store caches cleared, then the
+    scoreboard still ranks correctly."""
+    from sensorflow.rca import store
+    from sensorflow.rca import api as rca_api
+    inv = _create(rca_client)
+    store._cache.clear()
+    store._data_cache.clear()
+    rca_api._battery_cache.clear()
+    board = rca_client.get(
+        f"/api/rca/investigations/{inv['id']}/scoreboard").json()
+    assert board["rows"][0]["hypothesis"] == "FEATURE_SKEW"
+
+
+def test_recorded_only_scoreboard_evolves(rca_client):
+    """The working-hypothesis banner must not leak unvisited stages: with no
+    diagnostics run, all 8 hypotheses stay in play; evidence narrows it."""
+    inv = _create(rca_client)
+    iid = inv["id"]
+    board = rca_client.get(
+        f"/api/rca/investigations/{iid}/scoreboard?recorded_only=true").json()
+    assert len(board["working_hypothesis_set"]) == 8
+    for key in ("feature_parity", "offline_audit", "shadow_traffic",
+                "label_integrity", "serving_parity"):
+        rca_client.get(f"/api/rca/investigations/{iid}/diagnostics/{key}")
+    board = rca_client.get(
+        f"/api/rca/investigations/{iid}/scoreboard?recorded_only=true").json()
+    ws = board["working_hypothesis_set"]
+    assert "FEATURE_SKEW" in ws
+    assert "OFFLINE_CONTAMINATION" not in ws  # ruled out by clean audit
+    assert len(ws) < 8
+
+
+def test_404s(rca_client):
+    assert rca_client.get("/api/rca/investigations/nope").status_code == 404
+    assert rca_client.get(
+        "/api/rca/investigations/nope/scoreboard").status_code == 404

tests/test_rca/test_scenarios.py

@@ -0,0 +1,206 @@
+"""For EACH planted root cause: run the full diagnostic battery and assert the
+scoring board ranks the true cause on top with the right signature findings,
+and that the decision tree walks to the right conclusion."""
+
+from __future__ import annotations
+
+import numpy as np
+import pytest
+
+from sensorflow.rca.models import ROOT_CAUSES
+from sensorflow.rca.scenario import generate_scenario
+
+SEED = 7
+
+
+def _codes(battery):
+    return {f.code for _, (_, fs) in battery.items() for f in fs}
+
+
+def _rank(board, cause):
+    return next(r["rank"] for r in board["rows"] if r["hypothesis"] == cause)
+
+
+# ------------------------------------------------------------- global shape
+
+
+@pytest.mark.parametrize("cause", ROOT_CAUSES)
+def test_claims_shape(batteries, cause):
+    bundle, _, _, _ = batteries[cause]
+    m = bundle.meta["measured"]
+    assert 0.040 <= m["offline_delta"] <= 0.060, "offline claim should be ~+5pp"
+    assert -0.035 <= m["shadow_delta"] <= -0.005, "shadow claim should be ~-2pp"
+
+
+@pytest.mark.parametrize("cause", ROOT_CAUSES)
+def test_determinism(cause):
+    a = generate_scenario(cause, seed=SEED)
+    b = generate_scenario(cause, seed=SEED)
+    assert a.offline["b_correct"].tolist() == b.offline["b_correct"].tolist()
+    assert a.shadow["a_correct"].tolist() == b.shadow["a_correct"].tolist()
+    assert a.meta["measured"] == b.meta["measured"]
+
+
+# ------------------------------------------------------- ranking + signatures
+
+
+@pytest.mark.parametrize("cause", ROOT_CAUSES)
+def test_true_cause_ranks_top(batteries, cause):
+    _, _, board, _ = batteries[cause]
+    assert _rank(board, cause) <= 2, (
+        f"{cause} ranked {_rank(board, cause)}: "
+        f"{[(r['hypothesis'], r['score']) for r in board['rows'][:3]]}")
+    top = board["rows"][0]
+    if top["hypothesis"] == cause:
+        assert top["auto_confidence"] in ("HIGH", "MEDIUM")
+
+
+@pytest.mark.parametrize("cause", ROOT_CAUSES)
+def test_decision_tree_conclusion(batteries, cause):
+    _, _, _, tree = batteries[cause]
+    assert tree["conclusion"] == cause
+    if cause == "STATISTICAL_NOISE":
+        assert tree["conclusion_kind"] == "insufficient_evidence"
+    assert tree["path"][-1] == tree["nodes"][
+        [n["id"] for n in tree["nodes"]].index(tree["path"][-1])]["id"]
+
+
+def test_feature_skew_signature(batteries):
+    bundle, battery, _, _ = batteries["FEATURE_SKEW"]
+    codes = _codes(battery)
+    assert any(c.startswith("FP_FEATURE_SKEW:obj_distance_m") for c in codes)
+    # The skewed feature must TOP the parity ranking.
+    rows = battery["feature_parity"][0]["rows"]
+    assert rows[0]["feature"] == "obj_distance_m"
+    assert rows[0]["skew_flag"]
+    # Median ratio reflects the meters->feet unit error.
+    assert 2.5 <= rows[0]["median_ratio"] <= 4.2
+    # Regression concentrated where the feature matters (highway).
+    assert any(c.startswith("CP_SEGMENT_CONCENTRATED:scene=highway")
+               for c in codes)
+
+
+def test_contamination_signature(batteries):
+    _, battery, _, _ = batteries["OFFLINE_CONTAMINATION"]
+    codes = _codes(battery)
+    assert "OA_LEAKAGE_DUPLICATES" in codes, "leakage scan must fire"
+    assert "OA_REPRO_FAIL" in codes, "reproducibility must fail"
+    assert "OA_SPLIT_ROW_LEVEL" in codes
+    leak = battery["offline_audit"][0]["leakage"]
+    assert leak["n_duplicates"] > 0.15 * leak["offline_n"]
+    # Leaked units carry the inflation; clean units do not.
+    assert leak["dup_delta_pp"] > leak["clean_delta_pp"] + 5.0
+
+
+def test_noise_signature(batteries):
+    _, battery, _, _ = batteries["STATISTICAL_NOISE"]
+    codes = _codes(battery)
+    assert "SS_INSUFFICIENT_EVIDENCE" in codes
+    assert "SS_LOW_ESS" in codes
+    assert "POP_VOLUME_LOW" in codes
+    sig = battery["statistical_significance"][0]
+    assert sig["outcome"] == "insufficient_evidence"
+    # CI must span the practical margin on both sides.
+    sp = sig["shadow_paired"]
+    assert sp["ci_low"] < -0.01 < 0.0 < sp["ci_high"] + 0.01
+    assert sp["effective_n"] < 800
+
+
+def test_true_regression_signature(batteries):
+    _, battery, _, _ = batteries["TRUE_MODEL_REGRESSION"]
+    codes = _codes(battery)
+    assert "SS_SIGNIFICANT_REGRESSION" in codes
+    assert "CP_UNIFORM_REGRESSION" in codes
+    # All artifact channels clean.
+    for clean in ("OA_LEAKAGE_CLEAN", "FP_CLEAN", "SP_CLEAN",
+                  "ST_SAMPLE_FAIR", "LI_CLEAN"):
+        assert clean in codes, f"{clean} expected for a genuine regression"
+
+
+def test_distribution_shift_signature(batteries):
+    _, battery, _, _ = batteries["DISTRIBUTION_SHIFT"]
+    codes = _codes(battery)
+    assert "DS_SHIFT_HIGH:time_of_day" in codes
+    assert "CP_SIMPSONS_DETECTED" in codes
+    # Simpson's: aggregate signs flip while within-segment deltas agree.
+    agg = battery["conditional_performance"][0]["aggregate"]
+    assert agg["offline_delta_pp"] > 0 > agg["shadow_delta_pp"]
+    assert agg["sign_consistency"] >= 0.55
+    # Night segments regress in BOTH environments (the per-segment truth).
+    rows = battery["conditional_performance"][0]["rows"]
+    night = [r for r in rows if r["time_of_day"] == "night"
+             and r["interpretation"] != "low_volume"]
+    assert night and all(r["offline_delta_pp"] < 0 and r["shadow_delta_pp"] < 0
+                         for r in night)
+
+
+def test_serving_mismatch_signature(batteries):
+    _, battery, _, _ = batteries["SERVING_MISMATCH"]
+    codes = _codes(battery)
+    assert "SP_CONFIG_DIFF:confidence_threshold" in codes
+    assert "SP_CONFIG_DIFF:quantization" in codes
+    assert "PC_CONF_BAND_CONCENTRATION" in codes
+    bands = battery["paired_comparison"][0]["by_band"]
+    focus = next(b for b in bands if b["segment"].startswith("[0.35"))
+    assert focus["lift"] >= 1.5
+
+
+def test_label_latency_signature(batteries):
+    _, battery, _, _ = batteries["LABEL_LATENCY"]
+    codes = _codes(battery)
+    assert "LI_MATURE_DIVERGES" in codes
+    assert "LI_PROVISIONAL_HIGH" in codes
+    li = battery["label_integrity"][0]
+    assert li["provisional_fraction"] > 0.3
+    # Verdict flips on mature labels: B is actually ahead.
+    assert li["mature_delta_pp"] > 0 > li["provisional_delta_pp"]
+
+
+def test_sampling_bias_signature(batteries):
+    _, battery, _, _ = batteries["SAMPLING_BIAS"]
+    codes = _codes(battery)
+    assert "ST_SELECTION_BIAS" in codes
+    sel = battery["shadow_traffic"][0]["selection"]
+    # Sampled stream regresses; the rest of the eligible stream does not.
+    assert sel["sampled_delta_pp"] < 0 < sel["unsampled_delta_pp"]
+    assert sel["difficulty_psi"] >= 0.05
+
+
+# ------------------------------------------------------ cross-cause exclusion
+
+
+@pytest.mark.parametrize("cause", ROOT_CAUSES)
+def test_no_foreign_critical_signatures(batteries, cause):
+    """Discriminating CRITICAL signatures must not fire for other causes."""
+    _, battery, _, _ = batteries[cause]
+    codes = _codes(battery)
+    if cause != "OFFLINE_CONTAMINATION":
+        assert "OA_LEAKAGE_DUPLICATES" not in codes
+        assert "OA_REPRO_FAIL" not in codes
+    if cause != "FEATURE_SKEW":
+        assert not any(c.startswith("FP_FEATURE_SKEW") for c in codes)
+    if cause != "SERVING_MISMATCH":
+        assert not any(c.startswith("SP_CONFIG_DIFF") for c in codes)
+    if cause != "LABEL_LATENCY":
+        assert "LI_MATURE_DIVERGES" not in codes
+    if cause != "SAMPLING_BIAS":
+        assert "ST_SELECTION_BIAS" not in codes
+    if cause != "DISTRIBUTION_SHIFT":
+        assert "CP_SIMPSONS_DETECTED" not in codes
+
+
+# ------------------------------------------------------------ experiments
+
+
+@pytest.mark.parametrize("cause", ROOT_CAUSES)
+def test_experiments_target_top_hypothesis(batteries, cause):
+    from sensorflow.rca import scoring
+    bundle, battery, board, _ = batteries[cause]
+    rec = scoring.recommend_experiments(bundle, board, battery)
+    assert rec["minimum_additional_evidence"]
+    assert len(rec["experiments"]) == 5
+    ranked = rec["experiments"]
+    assert ranked == sorted(ranked, key=lambda d: d["priority"], reverse=True)
+    # The top experiment must discriminate the leading hypothesis.
+    top_hyp = board["rows"][0]["hypothesis"]
+    assert any(top_hyp in d["discriminates"] for d in ranked[:2])

tests/test_rca/test_state_machine.py

@@ -0,0 +1,103 @@
+"""Stage state machine: enforced ordering + unknown acknowledgment."""
+
+from __future__ import annotations
+
+import pytest
+
+from sensorflow.rca.models import (Investigation, StageOrderError,
+                                   UnknownsNotAcknowledgedError, make_finding,
+                                   STAGES)
+
+
+def _inv() -> Investigation:
+    return Investigation.new("t", "a", "b", "STATISTICAL_NOISE", seed=1)
+
+
+def test_initial_state():
+    inv = _inv()
+    assert len(inv.stages) == 13
+    assert inv.stages[0].status == "in_progress"
+    assert all(s.status == "pending" for s in inv.stages[1:])
+
+
+def test_cannot_skip_ahead():
+    inv = _inv()
+    with pytest.raises(StageOrderError):
+        inv.complete_stage(5)
+    # Even the immediately-next stage is blocked until stage 0 completes.
+    with pytest.raises(StageOrderError):
+        inv.complete_stage(1)
+
+
+def test_ordered_completion_advances():
+    inv = _inv()
+    inv.complete_stage(0)
+    assert inv.stages[0].status == "complete"
+    assert inv.stages[1].status == "in_progress"
+    inv.complete_stage(1)
+    assert inv.stages[2].status == "in_progress"
+
+
+def test_critical_unknowns_block_without_ack():
+    inv = _inv()
+    inv.upsert_auto_findings("comparison_validity", [
+        make_finding("comparison_validity", "CV_UNKNOWN:x", "x unknown",
+                     "UNKNOWN", "CRITICAL")])
+    with pytest.raises(UnknownsNotAcknowledgedError):
+        inv.complete_stage(0)
+    stage = inv.complete_stage(0, acknowledge_unknowns=True,
+                               ack_note="shipping anyway, tracked in JIRA-123")
+    assert stage.status == "complete_with_unknowns"
+    assert stage.acknowledged_unknowns
+    assert stage.ack_note == "shipping anyway, tracked in JIRA-123"
+    # The acknowledgment is recorded in the event log ("proceeding with
+    # unknowns" is auditable).
+    kinds = [e["kind"] for e in inv.events]
+    assert "unknowns_acknowledged" in kinds
+
+
+def test_non_critical_unknowns_do_not_block():
+    inv = _inv()
+    inv.upsert_auto_findings("comparison_validity", [
+        make_finding("comparison_validity", "CV_X", "warn unknown",
+                     "UNKNOWN", "WARN")])
+    stage = inv.complete_stage(0)
+    assert stage.status == "complete"
+
+
+def test_reopen():
+    inv = _inv()
+    inv.complete_stage(0)
+    inv.reopen_stage(0)
+    assert inv.stages[0].status == "in_progress"
+    with pytest.raises(StageOrderError):
+        inv.complete_stage(1)
+
+
+def test_serialization_roundtrip():
+    inv = _inv()
+    inv.upsert_auto_findings("comparison_validity", [
+        make_finding("comparison_validity", "CV_UNKNOWN:x", "x", "UNKNOWN",
+                     "CRITICAL")])
+    inv.complete_stage(0, acknowledge_unknowns=True)
+    d = inv.to_json_dict()
+    inv2 = Investigation.from_json_dict(d)
+    assert inv2.stages[0].status == "complete_with_unknowns"
+    assert inv2.findings[0].code == "CV_UNKNOWN:x"
+    assert inv2.scenario_cause == "STATISTICAL_NOISE"
+
+
+def test_training_mode_hides_cause():
+    inv = Investigation.new("t", "a", "b", "FEATURE_SKEW", seed=1,
+                            training_mode=True)
+    assert "scenario_cause" not in inv.to_dict()
+    inv.revealed = True
+    assert inv.to_dict()["scenario_cause"] == "FEATURE_SKEW"
+
+
+def test_stage_keys_match_spec_order():
+    keys = [s["key"] for s in STAGES]
+    assert keys[0] == "comparison_validity"
+    assert keys[6] == "statistical_significance"
+    assert keys[11] == "root_cause_scoring"
+    assert keys[12] == "recommendations_report"