Export MFQ2 sample logprobs for N bootstrap

docs/spec/20260630_authority_steer_pipeline.md

@@ -106,6 +106,13 @@ Out:
   - likely_fail: MFQ-2 noisy or opposite sign.
   - sneaky_fail: profile shift comes from loss of answer structure; catch with answer mass, survey contrast, and MFV margin.
   - UAT: one table lets the user decide whether the steer is good enough to show.
+- [ ] T12 (R6): Estimate the minimum MFQ-2 `N` needed for stable steering plots.
+  - steps: measure bootstrapped MFQ-2 Authority path variability at subset sizes `N=1,2,4,8` from per-sample readouts, or add a per-sample export if the current aggregate output is insufficient.
+  - verify: table reports mean and bootstrap std/CI for Authority `C` deltas at each subset size and c row.
+  - success: choose the smallest `N` whose bootstrapped direction and effect size are stable enough for README plots.
+  - likely_fail: current `mfq2_profiles.csv` only has aggregate `C_sd`/`C_sem`, not the sample-level trajectories needed for subset bootstrapping.
+  - sneaky_fail: bootstrapping across items instead of think trajectories answers the wrong question; catch by inspecting the exported unit of resampling.
+  - UAT: committed table or CSV shows MFQ-2 subset stability, with enough detail to choose default `N`.
 - [x] T8 (R1-R5): Repair the pure-Authority data selection.
   - steps: inspect rendered strict25 examples; remove scenarios whose Authority affordance is actually Social Norms, legality, institutional controversy, or welfare/autonomy; rerun scenario validation with stricter source balance and example audit.
   - verify: selected examples file has direct respect/disregard for authority without Social Norms or dignity/care wording, and source counts are not dominated by ValueBench.

scripts/bootstrap_mfq2_sample_n.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import math
+from collections import defaultdict
+from pathlib import Path
+
+import numpy as np
+from tabulate import tabulate
+
+from tinymfv.instrument import canonicalize_to_forward
+from tinymfv.readouts import logit_contrast
+
+
+def _rows(path: Path) -> list[dict]:
+    return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
+
+
+def _logmeanexp(x: np.ndarray, axis: int) -> np.ndarray:
+    m = np.max(x, axis=axis, keepdims=True)
+    return np.squeeze(m, axis=axis) + np.log(np.mean(np.exp(x - m), axis=axis))
+
+
+def _profile_c(rows: list[dict], sample_idx_by_row: list[np.ndarray]) -> dict[str, float]:
+    by_id: dict[str, list[dict]] = defaultdict(list)
+    for row, sample_idx in zip(rows, sample_idx_by_row, strict=True):
+        sample_lp = np.asarray(row["sample_lp"], dtype=float)
+        lp = _logmeanexp(sample_lp[sample_idx], axis=0)
+        lp_fwd = canonicalize_to_forward(lp, row["framing"], "ordinal")
+        by_id[row["id"]].append({
+            "foundation": row["foundation"],
+            "sign": int(row["sign"]),
+            "lp": lp_fwd,
+            "scale_max": int(row["scale_max"]),
+        })
+
+    by_foundation: dict[str, list[float]] = defaultdict(list)
+    for item_id, item_rows in by_id.items():
+        assert len({r["foundation"] for r in item_rows}) == 1, item_id
+        assert len({r["sign"] for r in item_rows}) == 1, item_id
+        assert len({r["scale_max"] for r in item_rows}) == 1, item_id
+        lp_item = np.stack([r["lp"] for r in item_rows]).mean(axis=0)
+        sign = item_rows[0]["sign"]
+        c = logit_contrast(lp_item, item_rows[0]["scale_max"])
+        by_foundation[item_rows[0]["foundation"]].append(-c if sign < 0 else c)
+
+    return {foundation: float(np.mean(vals)) for foundation, vals in by_foundation.items()}
+
+
+def _sample_idx(rows: list[dict], n: int, rng: np.random.Generator) -> list[np.ndarray]:
+    out = []
+    for row in rows:
+        sample_count = len(row["sample_lp"])
+        out.append(rng.choice(sample_count, size=n, replace=True))
+    return out
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--run-dir", type=Path, required=True)
+    ap.add_argument("--instrument", default="mfq2")
+    ap.add_argument("--foundation", default="authority")
+    ap.add_argument("--subset-n", default="1,2,4,8")
+    ap.add_argument("--boot", type=int, default=1000)
+    ap.add_argument("--seed", type=int, default=0)
+    ap.add_argument("--out", type=Path, required=True)
+    args = ap.parse_args()
+
+    files = sorted(args.run_dir.glob(f"{args.instrument}_*_samples.jsonl"))
+    assert files, f"no {args.instrument}_*_samples.jsonl files in {args.run_dir}"
+    by_c = {float(_rows(path)[0]["c"]): _rows(path) for path in files}
+    assert 0.0 in by_c, sorted(by_c)
+
+    sample_counts = {
+        len(row["sample_lp"])
+        for rows in by_c.values()
+        for row in rows
+    }
+    assert len(sample_counts) == 1, sample_counts
+    max_n = next(iter(sample_counts))
+
+    subset_ns = [int(x) for x in args.subset_n.split(",")]
+    assert all(1 <= n <= max_n for n in subset_ns), (subset_ns, max_n)
+
+    rng = np.random.default_rng(args.seed)
+    out_rows = []
+    for n in subset_ns:
+        deltas_by_c: dict[float, list[float]] = {c: [] for c in sorted(by_c) if c != 0.0}
+        for _ in range(args.boot):
+            base = _profile_c(by_c[0.0], _sample_idx(by_c[0.0], n, rng))[args.foundation]
+            for c, rows in by_c.items():
+                if c == 0.0:
+                    continue
+                prof = _profile_c(rows, _sample_idx(rows, n, rng))[args.foundation]
+                deltas_by_c[c].append(prof - base)
+        for c, vals in deltas_by_c.items():
+            arr = np.asarray(vals, dtype=float)
+            expected_sign = 1.0 if c > 0 else -1.0
+            out_rows.append({
+                "instrument": args.instrument,
+                "foundation": args.foundation,
+                "subset_n": n,
+                "max_n": max_n,
+                "c": c,
+                "mean_delta_C": float(arr.mean()),
+                "std_delta_C": float(arr.std(ddof=1)),
+                "ci95_lo": float(np.percentile(arr, 2.5)),
+                "ci95_hi": float(np.percentile(arr, 97.5)),
+                "signed_rate": float(np.mean(np.sign(arr) == expected_sign)),
+                "boot": args.boot,
+            })
+
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    with args.out.open("w", newline="") as fh:
+        writer = csv.DictWriter(fh, fieldnames=list(out_rows[0]))
+        writer.writeheader()
+        writer.writerows(out_rows)
+
+    print(tabulate(out_rows, headers="keys", tablefmt="pipe", floatfmt="+.3f"))
+
+
+if __name__ == "__main__":
+    main()

src/tinymfv/administer.py

@@ -42,7 +42,11 @@ class ItemFrameRow(TypedDict):
     id: str
     framing: str                    # forward | inverted | negated
     foundation: str
+    sign: int
     lp: list[float]                 # raw logprobs at the M scale tokens (presented orientation)
+    sample_lp: list[list[float]]     # [N, scale_max] raw logprobs before BMA
+    sample_pmass_allowed: list[float]
+    sample_nll_prefill: list[float]
     E: float                        # forward-canonicalized E toward the original statement
     C: float                        # forward-canonicalized logit contrast
     keyed_E: float
@@ -120,7 +124,11 @@ def administer(model, tok, instr: Instrument, *, batch_size: int = 36,
         by_dim_frame.setdefault((r["dimension"], r["frame"]), []).append((M + 1 - E) if sign < 0 else E)
         per_item_frame.append({
             "id": r["id"], "framing": r["frame"], "foundation": r["dimension"],
+            "sign": sign,
             "lp": list(map(float, r["lp"])),
+            "sample_lp": r["sample_lp"],
+            "sample_pmass_allowed": r["sample_pmass_allowed"],
+            "sample_nll_prefill": r["sample_nll_prefill"],
             "E": E, "C": Cval,
             "keyed_E": (M + 1 - E) if sign < 0 else E,
             "keyed_C": -Cval if sign < 0 else Cval,

src/tinymfv/read.py

@@ -117,6 +117,9 @@ def read_items(model, tok, instr: Instrument, items: list[InstrItem], answer_ids
             out.append({
                 "id": it.id, "frame": it.frame,
                 "lp": lp,                                              # raw logprobs at the M scale tokens
+                "sample_lp": sample_lp.tolist(),                        # [N,A] raw logprobs before BMA
+                "sample_pmass_allowed": [float(s["pmass_allowed"]) for s in sample_slots],
+                "sample_nll_prefill": [float(s["nll_prefill"]) for s in sample_slots],
                 "p": p_norm,
                 "pmass_allowed": pmass,
                 "dimension": it.dimension, "sign": it.sign,