feat(benchmark): Signal-to-noise analysis layer: SNR, scenario-health audit, contamination defense

decisiongrounding/Makefile

@@ -17,7 +17,10 @@ real-run:
 real-batch:
 	BATCH=1 ./scripts/run_real.sh
 # compare + the adherence-vs-N crossover over a real pool (set NS to taste; add
-# BATCH=1 for the Batch API, and SEEDS=0-4 for mean +/- CI error bars):
+# BATCH=1 for the Batch API, and SEEDS=0-4 for mean +/- CI error bars). The
+# default arms include rac + no_grounding, so both signal-to-noise contrasts
+# (rac-vs-naive_rag, rac-vs-no_grounding) populate; SEEDS>=2 is REQUIRED for the
+# signal-to-noise section (noise cannot be estimated from a single seed):
 real-crossover:
 	CROSSOVER=1 ./scripts/run_real.sh
 

decisiongrounding/runner/cli.py

@@ -46,6 +46,7 @@
 from scenarios.loader import Scenario, load_pool, load_scenarios  # noqa: E402
 from scoring import aggregate, score  # noqa: E402
 from scoring.crossover import (  # noqa: E402
+    DEFAULT_PAIRS,
     build_dataset,
     build_dataset_batched,
     build_dataset_multiseed,
@@ -727,15 +728,15 @@ def _on_cell(rec: dict) -> None:
                     pool=pool, pool_dir=pool_dir, scenarios_dir=args.scenarios,
                     batched=batch, poll=args.poll, progress=_on_cell)
                 dataset = merge_seed_datasets(existing, new_ps, list(arms), list(ns),
-                                              ("rac", "naive_rag"))
+                                              list(DEFAULT_PAIRS))
         elif seeds is not None:
             if len(seeds) > 1:
                 print(f"  (multi-seed: {seeds}; reporting mean +/- 95% CI)", file=sys.stderr)
             dataset = build_dataset_multiseed(
                 scenarios, arms=arms, ns=ns, seeds=seeds,
                 answering_model_name=args.answering, embedder_spec=args.embedder,
                 pool=pool, pool_dir=pool_dir, scenarios_dir=args.scenarios,
-                batched=batch, poll=args.poll, pair=("rac", "naive_rag"), progress=_on_cell,
+                batched=batch, poll=args.poll, pairs=list(DEFAULT_PAIRS), progress=_on_cell,
                 resume=resume_cells)
         elif batch:
             dataset = build_dataset_batched(

decisiongrounding/runner/dashboard.py

@@ -19,6 +19,8 @@
 
 from scoring.charts import grouped_bar_chart, line_chart
 from scoring.cost import cost_by_arm, dollars
+from scoring.health import scenario_health
+from scoring.snr import signal_to_noise
 
 _DEFAULT_TEMPLATE = Path(__file__).resolve().parent / "templates" / "dashboard.html"
 
@@ -267,6 +269,65 @@ def _scenarios_section(run):
     return matrix + "".join(drill)
 
 
+def _snr_txt(rec: dict) -> str:
+    """Plain-text SNR cell (number, or why it isn't one)."""
+    snr = rec.get("snr")
+    if snr is not None:
+        return f"{snr:.2f}" + (" (noise-dominated)" if rec.get("noise_dominated") else "")
+    return {"zero_noise_clean_separation": "clean (σ=0)",
+            "no_effect": "0 (tie)"}.get(rec.get("flag"), "n/a")
+
+
+def _signal_noise_section(dataset) -> str:
+    """The Signal / noise tab: SNR per contrast + the scenario-health audit."""
+    snr = signal_to_noise(dataset)
+    health = scenario_health(dataset)
+    if not snr["pairs"] and not health["total"]:
+        return '<p class=note>Needs a multi-seed crossover dataset.</p>'
+    out = []
+    if snr["pairs"]:
+        out.append("<h2>Signal-to-noise</h2>")
+        if snr["n_seeds"] < 2:
+            out.append('<p class=note>Only 1 seed — seed-to-seed noise is not '
+                       'estimable, so SNR cannot be computed. Re-run with '
+                       '<code>--seeds 0-4</code>.</p>')
+        else:
+            out.append('<p class=note>SNR = |between-arm adherence gap| ÷ '
+                       'across-seed σ of that gap. <b>SNR &lt; 1 is '
+                       'noise-dominated</b>.</p>')
+            for key, pair in snr["pairs"].items():
+                a, b = key.split("_vs_")
+                head = pair.get("headline") or {}
+                out.append(f'<p><b>{_esc(a)}</b> vs <b>{_esc(b)}</b> — headline SNR '
+                           f'at N={pair.get("headline_N")}: {_esc(_snr_txt(head))}</p>')
+                rows = "".join(
+                    f"<tr><td>{r['N']}</td><td>{_f(r.get('signal'))}</td>"
+                    f"<td>{_f(r.get('noise'))}</td><td>{_esc(_snr_txt(r))}</td></tr>"
+                    for r in pair["by_n"])
+                out.append("<table><thead><tr><th>N</th><th>signal</th>"
+                           f"<th>noise σ</th><th>SNR</th></tr></thead><tbody>{rows}"
+                           "</tbody></table>")
+    if health["total"]:
+        c = health["counts"]
+        out.append("<h2>Scenario health</h2>")
+        extra = f", {c['unknown']} unknown" if c["unknown"] else ""
+        out.append(f'<p>Of {health["total"]} scenarios: '
+                   f'<b>{c["discriminating"]} discriminating</b>, {c["broken"]} broken, '
+                   f'{c["contaminated"]} contaminated, {c["tie"]} tie{extra}.</p>')
+        flagged = [s for s in health["scenarios"] if s["class"] != "discriminating"]
+        if flagged:
+            rows = "".join(
+                f"<tr><td><code>{_esc(s['scenario_id'])}</code></td>"
+                f"<td>{_esc(s['class'])}</td><td>{_f(s['max_gap'])}</td></tr>"
+                for s in flagged)
+            out.append("<table><thead><tr><th>scenario</th><th>class</th>"
+                       f"<th>max arm gap</th></tr></thead><tbody>{rows}</tbody></table>")
+        out.append('<p class=note>broken = ceiling arm never adheres; '
+                   'contaminated = parametric-memory floor already adheres; '
+                   'tie = no arm separates.</p>')
+    return "".join(out)
+
+
 def _run_tab(paid: bool) -> str:
     if paid:
         real = (
@@ -287,7 +348,7 @@ def _run_tab(paid: bool) -> str:
                 "<code>DG_UI_ALLOW_PAID=1</code> (and <code>ANTHROPIC_API_KEY</code>) to enable; "
                 "you'll still have to estimate and tick a confirmation before any spend.</p>")
     return (
-        "<section class=tab id=s7><h2>Run the benchmark</h2>"
+        "<section class=tab id=s8><h2>Run the benchmark</h2>"
         "<p class=note>Triggers the CLI on the server; the page refreshes when the run finishes.</p>"
         "<button onclick=runOffline()>Run offline (free)</button> "
         "<button onclick=refreshMain()>↻ Refresh data</button> "
@@ -407,8 +468,10 @@ def render_main(run, dataset, cost_curve=None, *, live=False, paid_enabled=False
         secs.append('<section class=tab id=s4><p class=note>Needs both rac and naive_rag in a crossover dataset.</p></section>')
     # 5 Scenarios
     secs.append(f'<section class=tab id=s5><h2>Scenarios &amp; failures</h2>{_scenarios_section(run)}</section>')
-    # 6 Reproduce
-    secs.append('<section class=tab id=s6><h2>Reproduce</h2><pre>'
+    # 6 Signal / noise (SNR + scenario-health audit)
+    secs.append(f'<section class=tab id=s6>{_signal_noise_section(dataset) if dataset else "<p class=note>Needs a crossover dataset.</p>"}</section>')
+    # 7 Reproduce
+    secs.append('<section class=tab id=s7><h2>Reproduce</h2><pre>'
                 'make real-crossover   # headline + adherence-vs-N over the real pool\n'
                 'make real-batch       # via the Batch API (~50% cost)\n\n'
                 'make ui               # serve this dashboard live at 127.0.0.1:8099\n'
@@ -424,9 +487,10 @@ def render_main(run, dataset, cost_curve=None, *, live=False, paid_enabled=False
     return "".join(secs)
 
 
-# Tab labels, in section order (s0..s6, +s7 Run when live). Kept beside the
+# Tab labels, in section order (s0..s7, +s8 Run when live). Kept beside the
 # renderer so the nav and the sections stay in lockstep.
-_TABS = ["Overview", "Leaderboard", "Curves", "Cost", "rac vs RAG", "Scenarios", "Reproduce"]
+_TABS = ["Overview", "Leaderboard", "Curves", "Cost", "rac vs RAG", "Scenarios",
+         "Signal / noise", "Reproduce"]
 
 
 def build_dashboard(run, dataset, cost_curve=None, *, live=False, paid_enabled=False, template=None):

decisiongrounding/runner/templates/dashboard.html

@@ -8,7 +8,7 @@
     [CHIPS]  - the config chips (answering model, embedder, scenarios)
     [BODY]   - the tab radios + nav + main, with the data-bound sections
   Point the server / static generator at a custom copy with --template or the
-  DG_UI_TEMPLATE env var. Tabs are pure CSS (#t0..#t7); the JS only activates
+  DG_UI_TEMPLATE env var. Tabs are pure CSS (#t0..#t8); the JS only activates
   when a live Run tab (#run-status) is present, so static snapshots stay inert.
 -->
 <html lang=en>
@@ -43,6 +43,7 @@
 #t5:checked~nav label[for=t5]{color:var(--fg);border-bottom-color:var(--fg);font-weight:600}#t5:checked~main #s5{display:block}
 #t6:checked~nav label[for=t6]{color:var(--fg);border-bottom-color:var(--fg);font-weight:600}#t6:checked~main #s6{display:block}
 #t7:checked~nav label[for=t7]{color:var(--fg);border-bottom-color:var(--fg);font-weight:600}#t7:checked~main #s7{display:block}
+#t8:checked~nav label[for=t8]{color:var(--fg);border-bottom-color:var(--fg);font-weight:600}#t8:checked~main #s8{display:block}
 table{border-collapse:collapse;width:100%;margin:8px 0 18px;font-size:14px}
 th,td{border:1px solid var(--line);padding:7px 10px;text-align:right}
 th:first-child,td:first-child{text-align:left}

decisiongrounding/scoring/__init__.py

@@ -9,11 +9,15 @@
     summarize,
     t_critical_95,
 )
+from .health import scenario_health
 from .scorer import Score, score
+from .snr import signal_to_noise
 
 __all__ = [
     "Score",
     "score",
+    "signal_to_noise",
+    "scenario_health",
     "ArmMetrics",
     "aggregate",
     "adherence_variance",

decisiongrounding/scoring/crossover.py

@@ -617,26 +617,33 @@ def build_dataset_multiseed(
     scenarios_dir: str | None = None,
     batched: bool = False,
     poll: int = 20,
-    pair: tuple[str, str] = ("rac", "naive_rag"),
+    pair: "tuple[str, str] | None" = None,
+    pairs: "list[tuple[str, str]] | None" = None,
     progress: "Callable[[dict], None] | None" = None,
     resume: "dict[tuple[int, int, str, str], dict] | None" = None,
 ) -> dict:
     """Run the crossover over several seeds and aggregate per (arm, N) into
     mean +/- a t-based 95% CI. The plain fields stay the mean (backward
-    compatible); `<field>_ci` / `_std` / `_values`, `n_seeds`, `seeds`, and a
-    paired `pair[0]`-vs-`pair[1]` adherence difference (`paired`) are added.
+    compatible); `<field>_ci` / `_std` / `_values`, `n_seeds`, `seeds`, and
+    paired adherence differences (`paired`) are added.
+
+    `pairs` selects the arm contrasts for the paired diffs (default
+    `DEFAULT_PAIRS`: rac-vs-naive_rag and rac-vs-no_grounding); a single
+    `pair=(a, b)` tuple is still accepted for back-compat. A contrast whose
+    arms aren't both in the sweep is skipped.
 
     Calls the single-seed builders per seed, so batched + multiseed compose.
     Offline runs (deterministic stub + embedder) show little spread; the
     aggregation is exercised regardless.
     """
+    resolved = pairs if pairs is not None else ([pair] if pair is not None else DEFAULT_PAIRS)
     uniq = list(dict.fromkeys(int(s) for s in seeds)) or [0]
     per_seed = run_seeds(
         scenarios, arms, ns, uniq,
         answering_model_name=answering_model_name, embedder_spec=embedder_spec,
         pool=pool, pool_dir=pool_dir, scenarios_dir=scenarios_dir,
         batched=batched, poll=poll, progress=progress, resume=resume)
-    return _aggregate_seeds(per_seed, list(arms), list(ns), pair)
+    return _aggregate_seeds(per_seed, list(arms), list(ns), resolved)
 
 
 def run_seeds(
@@ -710,10 +717,28 @@ def _aggregate_arm_points(arms, ns, cols, n_seeds) -> dict:
     return points
 
 
-def _paired(per_seed, ns, pair) -> dict | None:
-    """Per-N paired adherence difference pair[0]-pair[1], differenced within each
-    seed (common random numbers), with its own CI."""
-    a, b = pair
+# The arm contrasts whose paired adherence difference (and thus signal-to-noise)
+# the multi-seed crossover reports by default: the pre-registered falsifier
+# (rac vs naive_rag, H1) and the parametric-memory floor (rac vs no_grounding,
+# H2). A pair whose arms aren't both in the sweep is silently skipped, so this
+# default is safe for any arm subset (offline runs without rac emit neither).
+DEFAULT_PAIRS = (("rac", "naive_rag"), ("rac", "no_grounding"))
+
+
+def _normalize_pairs(pairs) -> list[tuple[str, str]]:
+    """Accept a single ``(a, b)`` tuple or a list of them; return a list of
+    tuples. ``None``/empty -> ``[]``."""
+    if not pairs:
+        return []
+    if isinstance(pairs[0], str):  # a bare (a, b) tuple
+        return [(pairs[0], pairs[1])]
+    return [(p[0], p[1]) for p in pairs]
+
+
+def _paired_one(per_seed, ns, a, b) -> list | None:
+    """Per-N paired adherence difference a-b, differenced within each seed
+    (common random numbers), with its own CI. None when either arm is absent
+    from any seed's sweep."""
     if not per_seed or not all(a in ds["arms"] and b in ds["arms"] for _, ds in per_seed):
         return None
     out = []
@@ -727,7 +752,23 @@ def _paired(per_seed, ns, pair) -> dict | None:
             s = summarize(diffs)
             out.append({"N": n, "diff_mean": s["mean"], "diff_ci": s["ci"],
                         "diff_std": s["std"], "n": s["n"], "values": s["values"]})
-    return {f"{a}_vs_{b}": out} if out else None
+    return out or None
+
+
+def _paired(per_seed, ns, pairs) -> dict | None:
+    """Per-N within-seed paired adherence differences for one or more arm pairs.
+
+    `pairs` accepts a single ``(a, b)`` tuple (back-compat) or a list of them.
+    Each pair whose both arms appear in every seed's sweep contributes a
+    ``"{a}_vs_{b}"`` entry; pairs with a missing arm are skipped. The
+    across-seed std of each entry (`diff_std`) is the noise term the
+    signal-to-noise report divides into (see `scoring.snr`)."""
+    result = {}
+    for a, b in _normalize_pairs(pairs):
+        series = _paired_one(per_seed, ns, a, b)
+        if series:
+            result[f"{a}_vs_{b}"] = series
+    return result or None
 
 
 def _agg_per_scenario(per_seed, arms) -> dict:
@@ -755,7 +796,7 @@ def _agg_per_scenario(per_seed, arms) -> dict:
     return out
 
 
-def _aggregate_seeds(per_seed, arms, ns, pair) -> dict:
+def _aggregate_seeds(per_seed, arms, ns, pairs) -> dict:
     base = dict(per_seed[0][1])
     cols = _columns_from_datasets(per_seed, arms, ns)
     base["arms"] = _aggregate_arm_points(arms, ns, cols, len(per_seed))
@@ -769,7 +810,7 @@ def _aggregate_seeds(per_seed, arms, ns, pair) -> dict:
     cells = [c for _, ds in per_seed for c in (ds.get("cells") or [])]
     base["cells"] = cells
     base["stats"] = stats_by_n(cells) if cells else None
-    paired = _paired(per_seed, ns, pair)
+    paired = _paired(per_seed, ns, pairs)
     if paired:
         base["paired"] = paired
     return base
@@ -808,13 +849,19 @@ def merge_seed_datasets(existing: dict, new_per_seed, arms, ns, pair) -> dict:
     base = dict(existing)
     base["arms"] = _aggregate_arm_points(arms, ns, cols, len(all_seeds))
 
-    # Paired: prior per-seed diffs + the new seeds' diffs.
-    a, b = pair
-    old_paired = (existing.get("paired") or {}).get(f"{a}_vs_{b}")
-    new_paired = (_paired(add, ns, pair) or {}).get(f"{a}_vs_{b}")
-    if old_paired is not None or new_paired is not None:
-        old_by_n = {e["N"]: e for e in (old_paired or [])}
-        new_by_n = {e["N"]: e for e in (new_paired or [])}
+    # Paired: prior per-seed diffs + the new seeds' diffs, for every contrast —
+    # the requested pairs plus any already present in `existing` (so an
+    # under-specified merge never silently drops a contrast). Arm names contain
+    # underscores, but "_vs_" is a safe delimiter for the existing keys.
+    requested = _normalize_pairs(pair)
+    keys = {f"{a}_vs_{b}": (a, b) for a, b in requested}
+    for k in (existing.get("paired") or {}):
+        keys.setdefault(k, tuple(k.split("_vs_")))
+    new_paired = _paired(add, ns, list(keys.values())) or {}
+    merged_paired = {}
+    for key in keys:
+        old_by_n = {e["N"]: e for e in (existing.get("paired") or {}).get(key, [])}
+        new_by_n = {e["N"]: e for e in new_paired.get(key, [])}
         merged = []
         for n in ns:
             vals = list(old_by_n.get(n, {}).get("values", [])) + list(new_by_n.get(n, {}).get("values", []))
@@ -823,7 +870,9 @@ def merge_seed_datasets(existing: dict, new_per_seed, arms, ns, pair) -> dict:
                 merged.append({"N": n, "diff_mean": s["mean"], "diff_ci": s["ci"],
                                "diff_std": s["std"], "n": s["n"], "values": s["values"]})
         if merged:
-            base["paired"] = {f"{a}_vs_{b}": merged}
+            merged_paired[key] = merged
+    if merged_paired:
+        base["paired"] = merged_paired
 
     base["per_scenario"] = _merge_per_scenario(existing.get("per_scenario", {}), add, arms, n_old)
     base["errors"] = list(existing.get("errors", [])) + [e for _, ds in add for e in ds["errors"]]