feat(sor): SS3 RQ1 confirmatory loader (pcap -> frozen pair set)

confirm_load.py reconstructs the RQ1 (entry,exit) circuit-pair set from
the REAL per-hop pcaps (bin_pcap_bytes -> score_matrix; diagonal=linked,
off-diagonal=unlinked) so confirm.rq1_p1_leak scores measured data — the
frozen §4 unit. classify_run reads persisted metrics.json;
collect_rq1_p1_pairs selects only bridge=on no-pad runs.

Plumbing unit-tested on SYNTHETIC pcaps only (prereg §2 blinding — no
inspecting intermediate confirmatory results before the battery
completes). RQ1-P2 nopad/pad pairing + all RQ2 held for the operator
ruling (NEEDS-OPERATOR: RQ2 per-circuit posterior underspecified).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

OVERSEER-STATUS.md

@@ -39,3 +39,6 @@ NEEDS-OPERATOR: How is the RQ2 per-circuit adversary sender-posterior constructe
 - 2026-07-20 OPERATOR GREEN-LIGHT + delegated overnight drive → paper. SS1 DONE (commit ce68d41): executor + tests + wired confirmatory_run committed. Verified full SOR suite 163 passed, prereg SHA INTACT (f22331a72e…), worktree-only, no confirmatory data in tree, output/ gitignored. NEXT: SS2 fire the confirmatory battery under token.
 - 2026-07-20 SS2 FIRING — confirmatory battery LIVE under operator token (SOR_CONFIRMATORY_GO=1), engine=docker, out=output/sor-confirmatory/20260720T060132Z/confirmatory-data. Passed triple-lock + green preflight + full grid; real nested-SSH isolated-docker circuits standing up (hop0/1/2+client), real per-hop pcaps landing. Frozen §4 R=30×C=50=9000 circuits HONORED (NOT trimmed). Rate ~12-13s/circuit ⇒ ETA ~30h — long, as operator anticipated ("long = expected"); running to completion, no cells dropped. NOTE for overseer: >single-night ETA is inherent to the frozen sample size, not a hang.
 - Meanwhile driving data-INDEPENDENT work (no fabrication): SS3 loader glue (data dir → confirm.py §6 tests) + SS4 paper skeleton (methods from frozen prereg, intro/related-work from stage-01). Results stay empty until real data lands.
+- 2026-07-20 SS3 (RQ1 half) DONE + committed: cmd_chat/sor/analysis/confirm_load.py reconstructs the RQ1 (entry,exit) circuit-pair set from the REAL per-hop pcaps (bin_pcap_bytes → score_matrix; diagonal=linked, off-diag=unlinked) so confirm.rq1_p1_leak scores measured data — the fully frozen-specified §4 unit. classify_run reads persisted metrics.json; collect_rq1_p1_pairs selects only bridge=on no-pad runs. Plumbing unit-tested on SYNTHETIC pcaps (test_sor_confirm_load.py, 3 passed) — no confirmatory data touched (prereg §2 blinding: no inspecting intermediate results before the battery completes). RQ1-P2 nopad/pad pairing + all RQ2 HELD for the ruling (top-of-file NEEDS-OPERATOR).
+- BATTERY PROGRESS: healthy + running; 1 full (cell,run)=50 circuits ≈ 14 min ⇒ full 180-run / 9000-circuit battery ETA ≈ 42h (frozen §4 R=30×C=50, NOT trimmed). This exceeds a single night — inherent to the frozen sample size; overseer/operator call whether to let it run to completion (~1.8 days) or intervene. No cells dropped; teardown clean each circuit.
+- DRIVE STATE: SS1 ✅, SS2 firing ✅, SS3-RQ1 ✅. GATED: SS3-RQ2 + RQ1-P2 pairing on the RQ2 ruling; SS4 paper + SS5 review on (a) battery completion (blinding) AND (b) the RQ2 ruling. Cannot proceed further without fabricating/guessing (forbidden) or breaking blinding (forbidden). Awaiting operator RQ2 ruling; loop to nudge.

cmd_chat/sor/analysis/confirm_load.py

@@ -0,0 +1,90 @@
+"""SS3 confirmatory loader — bind the immutable data dir to the frozen §6 gates.
+
+**RQ1 only** (the fully frozen-specified half). This reconstructs the RQ1
+``(entry-segment, exit-segment)`` circuit pairs from the **real per-hop pcaps**
+the executor wrote (``executor.bin_pcap_bytes`` → ``detectors.score_matrix``),
+never re-synthesising them, so :func:`confirm.rq1_p1_leak` scores measured data.
+The candidate true pairing is the diagonal (same circuit); off-diagonal pairs are
+unlinked — exactly the §4 "AUC over the (entry, exit) pair set, bootstrap over
+circuit pairs" unit.
+
+Deliberately NOT here (held for the operator ruling — see OVERSEER NEEDS-OPERATOR):
+  * **RQ2** — the frozen DV is a *per-circuit adversary sender posterior* (§3/§4),
+    whose construction is not formula-pinned in the prereg and not produced by the
+    instrument; guessing a posterior model post-freeze would be HARKing.
+  * **RQ1-P2** nopad/pad *pairing* convention for the paired bootstrap — the pairing
+    rule across the bridge-on and bridge-on+padding arms is a choice; not guessed.
+
+Blinding (prereg §2, binding): callers must **not** run this on a confirmatory
+data dir until the full battery has completed — no inspection of intermediate
+confirmatory results. Its plumbing is unit-tested on synthetic pcaps only.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Dict, List, Tuple
+
+from cmd_chat.sor.analysis.detectors import score_matrix
+from cmd_chat.sor.executor import DEFAULT_BINS, ExecutorError, bin_pcap_bytes
+
+Pair = Tuple[float, bool]
+
+
+def classify_run(run_dir: Path) -> Dict[str, object]:
+    """Read the run's ``metrics.json`` for its cell identity + arm flags (both
+    persisted by the executor). Raises if absent — a run is never guessed."""
+    mp = Path(run_dir) / "metrics.json"
+    if not mp.exists():
+        raise ExecutorError(f"no metrics.json in {run_dir} — cannot classify run")
+    d = json.loads(mp.read_text(encoding="utf-8"))
+    return {
+        "cell_id": d["cell_id"],
+        "rq": d["rq"],
+        "padding_applied": bool(d.get("padding_applied", False)),
+    }
+
+
+def _circuit_series(run_dir: Path, bins: int, hops: int) -> Tuple[List[List[int]], List[List[int]]]:
+    """Per-circuit (ingress=hop0, egress=last-hop) binned byte series, read from
+    the REAL pcaps under ``run_dir/circuits/<id>/pcap/``. Refuses a circuit that
+    is missing either endpoint pcap (never invents a flow)."""
+    cdir = Path(run_dir) / "circuits"
+    if not cdir.is_dir():
+        raise ExecutorError(f"no circuits/ under {run_dir}")
+    ingress: List[List[int]] = []
+    egress: List[List[int]] = []
+    for circuit in sorted(p for p in cdir.iterdir() if p.is_dir()):
+        pdir = circuit / "pcap"
+        ing, eg = pdir / "hop0.pcap", pdir / f"hop{hops - 1}.pcap"
+        if not ing.exists() or not eg.exists():
+            raise ExecutorError(f"missing ingress/egress pcap in {circuit}")
+        ingress.append(bin_pcap_bytes(ing, bins))
+        egress.append(bin_pcap_bytes(eg, bins))
+    return ingress, egress
+
+
+def reconstruct_run_pairs(run_dir: Path, *, bins: int = DEFAULT_BINS, hops: int = 3) -> List[Pair]:
+    """The RQ1 pair set for one run: ``S[i][j] = pearson(ingress_i, egress_j)``
+    over the run's circuits; the diagonal (``i == j``) is the linked (same-circuit)
+    pair, every off-diagonal cell an unlinked pair. Measured from real pcaps."""
+    ingress, egress = _circuit_series(Path(run_dir), bins, hops)
+    scores = score_matrix(ingress, egress)
+    n = len(scores)
+    return [(scores[i][j], i == j) for i in range(n) for j in range(n)]
+
+
+def collect_rq1_p1_pairs(data_dir: Path, *, bins: int = DEFAULT_BINS, hops: int = 3) -> List[Pair]:
+    """Aggregate the RQ1-P1 leak-arm pairs across every **bridge-on (no padding)**
+    run in the battery data dir — the pair set :func:`confirm.rq1_p1_leak` scores.
+    Frozen cell ids end in ``bridge=on`` for the leak arm (``bridge=on+padding`` and
+    ``bridge=off`` are excluded; ``padding_applied`` double-guards)."""
+    pairs: List[Pair] = []
+    for run_dir in sorted(p for p in Path(data_dir).iterdir() if p.is_dir()):
+        if not (run_dir / "metrics.json").exists():
+            continue
+        info = classify_run(run_dir)
+        if info["rq"] == "RQ1" and str(info["cell_id"]).endswith("bridge=on") and not info["padding_applied"]:
+            pairs.extend(reconstruct_run_pairs(run_dir, bins=bins, hops=hops))
+    return pairs

tests/test_sor_confirm_load.py

@@ -0,0 +1,80 @@
+"""SS3 RQ1 loader — pcap→pairs plumbing, on SYNTHETIC pcaps only.
+
+These pin that the loader reconstructs the RQ1 (entry,exit) pair set from real
+per-hop pcap bytes and classifies runs by their persisted metrics.json — WITHOUT
+touching any confirmatory data (prereg §2 blinding). Detector accuracy itself is
+covered by test_sor_analysis/test_sor_confirm; here we only prove the wiring.
+"""
+
+import json
+import struct
+
+from cmd_chat.sor.analysis import confirm_load
+
+
+def _write_series_pcap(path, series):
+    """Minimal LINKTYPE_RAW pcap placing one packet of size series[k] at t=k, so
+    bin_pcap_bytes(path, bins=len(series)) reproduces `series` bin-for-bin."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "wb") as f:
+        f.write(struct.pack("<IHHiIII", 0xA1B2C3D4, 2, 4, 0, 0, 65535, 101))
+        for k, nbytes in enumerate(series):
+            f.write(struct.pack("<IIII", k, 0, nbytes, nbytes))
+            f.write(b"x" * nbytes)
+
+
+def _write_circuit(run_dir, cid, ingress_series, egress_series):
+    pdir = run_dir / "circuits" / cid / "pcap"
+    _write_series_pcap(pdir / "hop0.pcap", ingress_series)      # ingress
+    _write_series_pcap(pdir / "hop2.pcap", egress_series)       # egress (hops=3)
+
+
+def _write_metrics(run_dir, cell_id, rq, padding):
+    run_dir.mkdir(parents=True, exist_ok=True)
+    (run_dir / "metrics.json").write_text(
+        json.dumps({"cell_id": cell_id, "rq": rq, "padding_applied": padding}),
+        encoding="utf-8",
+    )
+
+
+def test_classify_run_reads_persisted_metrics(tmp_path):
+    rd = tmp_path / "cell-x-r0"
+    _write_metrics(rd, "RQ1/topo=1house/selector=static/bridge=on", "RQ1", False)
+    info = confirm_load.classify_run(rd)
+    assert info["cell_id"].endswith("bridge=on")
+    assert info["rq"] == "RQ1" and info["padding_applied"] is False
+
+
+def test_reconstruct_run_pairs_shape_and_linkage(tmp_path):
+    rd = tmp_path / "cell-on-r0"
+    _write_metrics(rd, "RQ1/topo=1house/selector=static/bridge=on", "RQ1", False)
+    # Two circuits with anti-correlated temporal profiles; each hop0/hop2 share
+    # their own circuit's profile → linked (diagonal) pearson high, cross low.
+    _write_circuit(rd, "c0", [100, 10, 100, 10], [100, 10, 100, 10])
+    _write_circuit(rd, "c1", [10, 100, 10, 100], [10, 100, 10, 100])
+    pairs = confirm_load.reconstruct_run_pairs(rd, bins=4, hops=3)
+    assert len(pairs) == 4                                   # 2x2 score matrix
+    linked = [s for s, is_linked in pairs if is_linked]
+    unlinked = [s for s, is_linked in pairs if not is_linked]
+    assert len(linked) == 2 and len(unlinked) == 2
+    # Real measured signal carried through: same-circuit correlation beats cross.
+    assert min(linked) > max(unlinked)
+
+
+def test_collect_rq1_p1_selects_only_bridge_on_nopad(tmp_path):
+    # bridge=on (leak arm) — should be collected.
+    on = tmp_path / "cell-on-r0"
+    _write_metrics(on, "RQ1/topo=1house/selector=static/bridge=on", "RQ1", False)
+    _write_circuit(on, "c0", [100, 10, 100, 10], [100, 10, 100, 10])
+    # bridge=on+padding — excluded (padding arm).
+    pad = tmp_path / "cell-pad-r0"
+    _write_metrics(pad, "RQ1/topo=1house/selector=static/bridge=on+padding", "RQ1", True)
+    _write_circuit(pad, "c0", [50, 50, 50, 50], [50, 50, 50, 50])
+    # RQ2 cell — excluded (wrong RQ).
+    rq2 = tmp_path / "cell-rq2-r0"
+    _write_metrics(rq2, "RQ2/topo=directory-federated/selector=static/bridge=off", "RQ2", False)
+    _write_circuit(rq2, "c0", [1, 2, 3, 4], [1, 2, 3, 4])
+
+    pairs = confirm_load.collect_rq1_p1_pairs(tmp_path, bins=4, hops=3)
+    assert len(pairs) == 1                                   # only the 1-circuit bridge=on run
+    assert pairs[0][1] is True                               # its single diagonal pair