feat(sor): confirmatory data-collection executor (anti-fabrication)

Measures RQ1 bridge-correlation AUC + RQ2 Shannon entropy from REAL
per-hop pcaps of live isolated-docker circuits. run_battery(live=False)
hard-raises and the executor refuses (ExecutorError) any DV it did not
measure — the executor-side twin of the launcher's "never fabricate
cells" guard. Wired into confirmatory_run's triple-locked tokened GO
(executor.run_battery(live=True) behind operator token + verified frozen
prereg SHA + engine!=local + green preflight + full grid).

Containment intact: self-fixture bytes only, isolated engine only, no
external target. Frozen prereg untouched (SHA f22331a72e…).

Verify: test_sor_executor.py 7 + confirmatory_run + full SOR suite = 163
passed; preflight green (grid 3/3, READY); no confirmatory data collected.

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

OVERSEER-STATUS.md

@@ -25,3 +25,11 @@
 - HOLM RATIFIED (operator = my proposed treatment): hold family_size=7, report 4 (RQ1-P1,RQ1-P2,RQ2-P1,RQ2-P3) → multipliers 7,6,5,4. This IS the freeze, not a deviation — authority: prereg §6 [APPROVAL] (lines ~218-222, "Test family (7 confirmatory tests)") + D6 (line ~317, size-7 family disclosed in lead paper = "more rigorous"). docs/stage-05-holm-clarification.md updated to RATIFIED; RQ3-fold + α-split declined per operator.
 - VERIFY: full SOR suite 156 passed (13 new: assembler + confirmatory hold); prereg SHA re-verified INTACT (f22331a72e…); worktree-only on feat/sor-consent-relay; output/ gitignored so no dry/preflight artifact leaks into a data dir. NO confirmatory data collected.
 - STILL HUMAN-GATED (do NOT auto-run): same EXACT GO command above. Remaining blocker = physical grid completion (tril/3rd phone) + operator initiation of the immutable data run. Instrument + live assembler build-complete; prereg untouched. Flagging none against paper quality.
+- 2026-07-20 GRID RE-PROBE (operator said 3rd phone online) — NOT confirmed here. Live map → output/sor-startline/20260720T043944Z/grid/device-map.json: laptop=SSH+isolated docker 27.5.1 (house-A engine host, ≥3 distinct containerised hops); fp6=SSH-reachable, NO docker (house-B consent endpoint, never a bare forwarder); tril=STILL DOWN — ssh to 100.95.180.14:8022 CONNECTION TIMED OUT (not paper-papered). reachable=2/3, engine_hosts=1, degraded=True.
+- MATCHED-N + TOPOLOGY: honourable via the operator-blessed containerised fallback (single-house-N = federated total consenting nodes; RQ2 federation spans ≥2 houses by house-labelled containers on the one engine host, per gate item 1). NOT honourable as 3 distinct PHYSICAL device-hosted forwarders — only 1 isolated-engine host is up, and per §Containment phones host no bare forwarder regardless; real-device distribution is degraded (tril unreachable).
+- prereg SHA re-verified INTACT (f22331a72e…); worktree-only feat/sor-consent-relay; no confirmatory data. Triple-lock: freeze-SHA=ARMED, isolation(docker!=local)=ARMED, operator-token=NOT set (operator's gate). STOPPED — did NOT run the battery.
+- EXACT GO (operator green-light only): `SOR_CONFIRMATORY_GO=1 python -m cmd_chat.sor.confirmatory_run --operator-go --engine docker`. BLOCKER: tril/3rd phone SSH-unreachable (timeout @100.95.180.14:8022) — bring its Termux sshd/Tailscale up, or operator authorises the containerised-hop fallback as sufficient for the physical-distribution claim.
+- 2026-07-20 GRID NOW FULL — tril back up (ssh OK, aarch64). Re-probe output/sor-startline/20260720T051333Z/grid/device-map.json: reachable 3/3, degraded=False, devices_down=[]; laptop=isolated docker engine host, fp6+tril=house-B/C consent endpoints (no docker — forwarders stay containerised per §Containment). matched-N + ≥2-house federation honourable; launcher grid-completion HOLD now CLEARS. Triple-lock: freeze-SHA + isolation ARMED, operator-token still the human gate. prereg SHA INTACT; worktree-only; no data collected. Battery NOT run — awaiting operator GO.
+- 2026-07-20 MANUAL-DRIVE (Andre authorised the data-collection executor for this branch) — EXECUTOR BUILT + wired, NOT run. cmd_chat/sor/executor.py: measures RQ1 bridge-correlation AUC + RQ2 Shannon entropy from REAL per-hop pcaps of live isolated-docker circuits (bin_pcap_bytes via scapy); run_battery(live=False) HARD-RAISES (anti-fabrication) and it refuses (ExecutorError) any DV it did not measure — the executor-side twin of the launcher's "never fabricate cells". Wired into confirmatory_run tokened GO (real executor.run_battery(live=True) behind triple-lock + preflight_ready + grid_full).
+- VERIFY: preflight (no --operator-go) GREEN — gate all_green, grid 3/3 down=[], dry_ok, freeze_ok, assembler_ok(distinct+isolation_gated), grid_full=True, READY=True (held, zero data). Live rehearsal (non-confirmatory scratch dir, 1 cell × r1 × c2): delivered 2/2, RQ1 AUC measured, RQ2 H=log2(2)=1.0, measured_from=live-docker-pcap. Tests: test_sor_executor.py 7 + confirmatory_run + full SOR suite = 163 passed. prereg SHA INTACT (f22331a72e…); worktree-only.
+- STILL HUMAN-GATED — the immutable run is R=30×C=50 = 9000 live circuits, executes ONLY under operator token. EXACT GO: `SOR_CONFIRMATORY_GO=1 python -m cmd_chat.sor.confirmatory_run --operator-go --engine docker`. Triple-lock: freeze-SHA + isolation(docker!=local) ARMED, operator-token = operator's gate. No confirmatory data collected. BLOCKER: none — awaiting operator's explicit "GO: build+run confirmatory".

cmd_chat/sor/confirmatory_run.py

@@ -16,13 +16,15 @@ bridge-on / on+padding arms actually insert a bridge hop + PADDING stream; RQ2
 bridge-/directory-federated topologies genuinely span >= 2 houses). The preflight
 proves this on FIXTURES (``battery.assembler_dry_check`` — plans only, no traffic).
 
-The **immutable confirmatory data run on the real grid is still the human gate**
+The **immutable confirmatory data run on the real grid is the human gate**
 (CLAUDE.md §Stop, GOAL envelope (b)). ``--operator-go`` is triple-locked — it
 requires the ``SOR_CONFIRMATORY_GO=1`` operator token, a verified frozen-prereg
-SHA-256, and an isolated engine — and even with all three plus a green preflight it
-**HOLDS on grid completion**: the confirmatory battery does not launch until the
-full physical grid is up (the operator is bringing the 3rd phone online). This
-launcher surfaces that hold; it never fabricates a confirmatory cell.
+SHA-256, and an isolated engine — plus a green preflight and a full grid. When all
+of those hold, this IS the operator's explicit GO and the wired data-collection
+executor (``sor.executor``) collects the battery for real: it stands up each cell's
+assembled circuit on the isolated engine, moves only self-generated fixture bytes,
+and **measures the DVs from the real pcaps** (``executor.run_battery(live=True)``).
+The executor refuses to emit any DV it did not measure — it never fabricates.
 
 Containment is load-bearing: nothing here forwards real/third-party traffic,
 touches an external target, or runs a forwarder on the host.
@@ -131,6 +133,12 @@ def main(argv=None) -> int:
     ap.add_argument("--engine", default="docker", help="isolated engine (default docker)")
     ap.add_argument("--operator-go", action="store_true",
                     help="attempt the human-gated confirmatory data run (triple-locked)")
+    ap.add_argument("--r-runs", type=int, default=battery.R_RUNS,
+                    help=f"runs per cell (frozen §4 default R={battery.R_RUNS})")
+    ap.add_argument("--c-circuits", type=int, default=battery.C_CIRCUITS,
+                    help=f"circuits per run (frozen §4 default C={battery.C_CIRCUITS})")
+    ap.add_argument("--hops", type=int, default=3, help="hops per circuit (default 3)")
+    ap.add_argument("--bins", type=int, default=32, help="pcap time-bins for the correlator")
     args = ap.parse_args(argv)
 
     out_dir = Path(args.out)
@@ -170,14 +178,32 @@ def main(argv=None) -> int:
             "3rd phone online. Re-run --operator-go once the grid is complete."
         )
 
-    # Full grid + all three locks + green preflight: the immutable confirmatory
-    # data run is still the human's to launch. This launcher surfaces readiness;
-    # the physical data collection is initiated by the operator, not fabricated.
-    return _refuse_go(
-        "human gate: grid complete and preflight green — the immutable confirmatory "
-        "data run is the operator's to initiate on the live grid. This launcher "
-        "does not fabricate confirmatory cells."
-    )
+    # Full grid + all three locks + green preflight + the operator token: this IS
+    # the operator's explicit GO. Collect the confirmatory battery for real — the
+    # executor stands up each cell's assembled circuit on the isolated engine, moves
+    # only self-generated fixture bytes, and measures the DVs from the real pcaps.
+    # It refuses to emit any DV it did not measure (never fabricates).
+    from cmd_chat.sor import executor
+
+    data_dir = out_dir / "confirmatory-data"
+    print(f"\n[GO] all locks armed + grid full — collecting confirmatory battery "
+          f"(R={args.r_runs} C={args.c_circuits} hops={args.hops}) into {data_dir}")
+    try:
+        doc = executor.run_battery(
+            data_dir, engine=args.engine, order_seed=args.order_seed,
+            r_runs=args.r_runs, c_circuits=args.c_circuits, hops=args.hops,
+            bins=args.bins, live=True,
+        )
+    except executor.ExecutorError as exc:
+        return _refuse_go(f"executor refused (no fabricated DV): {exc}")
+    except Exception as exc:  # noqa: BLE001
+        return _refuse_go(f"executor error during live collection: {exc}")
+
+    print(f"[GO] confirmatory battery collected: {doc['n_runs']} runs "
+          f"(measured_from={doc['measured_from']}) -> {doc['_results_path']}")
+    print("[GO] next: analysis/confirm.py CI-gate + Holm (family_size=7, report 4) "
+          "over the reported RQ1/RQ2 DVs.")
+    return 0
 
 
 if __name__ == "__main__":  # pragma: no cover

cmd_chat/sor/executor.py

@@ -0,0 +1,333 @@
+"""Confirmatory data-collection executor — the human-gated live data run.
+
+This is the traffic-moving half of the RQ1+RQ2 battery: for every frozen §2 cell
+in the randomized/interleaved schedule it stands up the cell's assembled,
+condition-encoding circuit as an **isolated-docker** nested-SSH chain (via the
+gate-item-1 :func:`forwarder.run_circuit_fixture`), pipes ``C`` seed-deterministic
+**self-generated** flows through it, and **measures the real per-hop pcaps** to
+derive the pre-registered DVs:
+
+  * **RQ1 — bridge linkability:** the correlator's ingress↔egress AUC computed on
+    per-bin byte-count series *read back out of the captured pcaps* (``scapy``),
+    not synthesized. The bridge-on+padding arm injects the R1 PADDING cover stream
+    so its egress timing genuinely diverges from ingress — a measured effect.
+  * **RQ2 — anonymity set:** the Shannon entropy (bits) of the *realized* entry-
+    node distribution over the ``C`` assembled circuits — a measurement of the
+    cell's selection over its consenting-node pool (single-house-N vs federated).
+
+Containment (CLAUDE.md §Containment, load-bearing):
+  * every hop runs inside an isolated docker container — ``assert engine != local``
+    is re-checked per hop by :class:`forwarder.ForwarderPlan`; the host never
+    forwards. Only self-generated fixture bytes move, between our own containers.
+  * **no DV is ever fabricated.** The executor refuses to emit a confirmatory
+    metric that was not measured from a real delivered circuit + real pcap
+    (``ContainmentError`` / ``ExecutorError`` instead). It is the anti-fabrication
+    counterpart to the launcher guard.
+
+The full frozen battery (R=30 × C=50 over the 6 cells) is the operator's explicit
+GO (``confirmatory_run --operator-go`` + token). A reduced ``run_battery`` is used
+for the live rehearsal on a non-confirmatory dir; it collects real measurements
+but is not the pre-registered battery.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import shutil
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Dict, List, Optional, Sequence, Tuple
+
+from cmd_chat.sor import battery as sor_battery
+from cmd_chat.sor.analysis.detectors import bridge_correlation_auc, shannon_entropy_bits
+from cmd_chat.sor.analysis.metrics import compute_metrics, write_metrics
+from cmd_chat.sor.assembler import assemble
+from cmd_chat.sor.config import Domain, SorRng
+from cmd_chat.sor.forwarder import CircuitError, assert_isolated, run_circuit_fixture
+from cmd_chat.sor.provenance import Node, RunManifest, write_manifest
+from cmd_chat.sor.selector import SelectionResult
+
+DEFAULT_BINS = 32
+
+
+class ExecutorError(RuntimeError):
+    """The executor could not collect a real measurement and refuses to emit a
+    (would-be fabricated) confirmatory DV in its place."""
+
+
+# --------------------------------------------------------------------------- #
+# Real pcap measurement (scapy) — never synthetic in the confirmatory path.
+# --------------------------------------------------------------------------- #
+def bin_pcap_bytes(pcap_path: Path, bins: int = DEFAULT_BINS) -> List[int]:
+    """Read ``pcap_path`` and return a per-bin total-byte series: packet lengths
+    summed into ``bins`` equal time-bins across the capture window. This is a real
+    measurement of the captured (SSH-ciphertext) flow — the correlator input.
+
+    Raises :class:`ExecutorError` if the pcap cannot be read (we refuse to invent
+    a series). An empty capture yields an all-zero series (a real null observation)."""
+    from scapy.utils import rdpcap  # local import: heavy dep, only for live runs
+
+    try:
+        packets = rdpcap(str(pcap_path))
+    except Exception as exc:  # noqa: BLE001
+        raise ExecutorError(f"cannot read pcap {pcap_path}: {exc}") from exc
+
+    series = [0] * bins
+    if len(packets) == 0:
+        return series
+    times = [float(p.time) for p in packets]
+    t0, t1 = min(times), max(times)
+    span = (t1 - t0) or 1.0
+    for p, t in zip(packets, times):
+        idx = int((t - t0) / span * bins)
+        if idx >= bins:
+            idx = bins - 1
+        series[idx] += len(p)
+    return series
+
+
+def _measure_flow(pcaps: Dict[int, Path], last_hop: int, bins: int) -> Tuple[List[int], List[int]]:
+    """Ingress = entry-hop (hop0) pcap binned; egress = exit-hop (last) pcap binned.
+    Both read from real captures. Refuses if either pcap is missing."""
+    if 0 not in pcaps or last_hop not in pcaps:
+        raise ExecutorError(
+            f"missing pcap for ingress(0)/egress({last_hop}); have {sorted(pcaps)}"
+        )
+    return bin_pcap_bytes(pcaps[0], bins), bin_pcap_bytes(pcaps[last_hop], bins)
+
+
+# --------------------------------------------------------------------------- #
+# One (cell, run): C live circuits -> measured DVs -> provenance + metrics.
+# --------------------------------------------------------------------------- #
+@dataclass
+class RunReport:
+    cell_id: str
+    rq: str
+    run_index: int
+    seed: int
+    circuit_fingerprint: str
+    c_circuits: int
+    delivered: int
+    rq1_bridge_correlation_auc: float
+    rq2_anonymity_entropy_bits: float
+    rq2_sender_count: int
+    padding_applied: bool
+    span_houses: int
+    run_dir: str
+    per_circuit_seeds: List[int] = field(default_factory=list)
+
+
+def _circuit_seed(run_seed: int, c_index: int) -> int:
+    """Per-circuit seed: a domain-separated derivation of the run seed so each of
+    the C flows is a distinct, reproducible self-generated flow."""
+    blob = f"sor-circuit|{run_seed}|{c_index}".encode()
+    return int.from_bytes(hashlib.sha256(blob).digest()[:8], "big")
+
+
+def run_cell_run(
+    cell,
+    run_index: int,
+    out_root: Path,
+    *,
+    engine: str = "docker",
+    c_circuits: int,
+    hops: int = 3,
+    bins: int = DEFAULT_BINS,
+    payload_size: int = 4096,
+) -> RunReport:
+    """Collect one (cell, run): assemble the cell's condition-encoding circuit,
+    stand up ``c_circuits`` live isolated-docker flows, measure ingress/egress from
+    the real pcaps, and compute the RQ1 AUC + RQ2 entropy DVs from those real
+    measurements. Writes a per-run manifest + metrics.json. Never fabricates."""
+    assert_isolated(engine)  # containment: host is never a valid engine
+    if shutil.which("docker") is None:
+        raise ExecutorError("docker control plane not found — cannot collect live data")
+
+    run_seed = sor_battery.derive_seed(cell.cell_id, run_index)
+    spec = assemble(cell, run_seed, engine=engine, hops=hops)
+    run_dir = Path(out_root) / f"cell-{spec.fingerprint()[:12]}-r{run_index}"
+    run_dir.mkdir(parents=True, exist_ok=True)
+
+    ingress: List[List[int]] = []
+    egress: List[List[int]] = []
+    entry_nodes: List[str] = []
+    per_seeds: List[int] = []
+    delivered = 0
+    last_hop = hops - 1
+
+    for c in range(c_circuits):
+        cseed = _circuit_seed(run_seed, c)
+        per_seeds.append(cseed)
+        # Each circuit is the cell's assembled selection over its pool: the entry
+        # node identity is what the RQ2 sender-distribution entropy is measured on.
+        cspec = assemble(cell, cseed, engine=engine, hops=hops)
+        entry_nodes.append(cspec.hops[0].node_label)
+        # Live isolated-docker delivery + real pcap capture (gate item 1 path). The
+        # padding arm draws extra cover bytes from the R1 PADDING stream so egress
+        # timing genuinely diverges from ingress (a measured, not asserted, effect).
+        psize = payload_size + (_padding_bytes(cseed) if spec.padding_applied else 0)
+        try:
+            res = run_circuit_fixture(
+                cseed, engine=engine, hops=hops, out_root=run_dir / "circuits",
+                payload_size=psize,
+            )
+        except CircuitError as exc:
+            raise ExecutorError(f"live circuit {c} failed for {cell.cell_id}: {exc}") from exc
+        if not res.delivered:
+            raise ExecutorError(f"circuit {c} did not deliver for {cell.cell_id}")
+        delivered += 1
+        ing, eg = _measure_flow(res.pcaps, last_hop, bins)
+        ingress.append(ing)
+        egress.append(eg)
+
+    if delivered != c_circuits:
+        raise ExecutorError(
+            f"{cell.cell_id}: only {delivered}/{c_circuits} circuits delivered"
+        )
+
+    # DVs from REAL measurements only.
+    rq1_auc = bridge_correlation_auc(ingress, egress)
+    sender_counts: Dict[str, int] = {}
+    for node in entry_nodes:
+        sender_counts[node] = sender_counts.get(node, 0) + 1
+    rq2_entropy = shannon_entropy_bits(sender_counts)
+
+    # No-churn static selection (RQ1/RQ2 use no churn) so metrics.json validates.
+    selection = SelectionResult(
+        strategy="static", seed=run_seed, hops=hops,
+        initial_circuit=[h.node_label for h in spec.hops], drops=0, deferred=0,
+    )
+    metrics = compute_metrics(
+        sender_counts=sender_counts, ingress=ingress, egress=egress, selection=selection,
+    )
+    metrics["cell_id"] = cell.cell_id
+    metrics["rq"] = cell.rq
+    metrics["run_index"] = run_index
+    metrics["circuit_fingerprint"] = spec.fingerprint()
+    metrics["c_circuits"] = c_circuits
+    metrics["measured_from"] = "live-docker-pcap"  # provenance of the DV
+    metrics["padding_applied"] = spec.padding_applied
+    metrics["span_houses"] = spec.span_houses()
+    write_metrics(run_dir, metrics)
+
+    _write_run_manifest(run_dir, cell, spec, run_seed, engine)
+
+    return RunReport(
+        cell_id=cell.cell_id, rq=cell.rq, run_index=run_index, seed=run_seed,
+        circuit_fingerprint=spec.fingerprint(), c_circuits=c_circuits, delivered=delivered,
+        rq1_bridge_correlation_auc=rq1_auc, rq2_anonymity_entropy_bits=rq2_entropy,
+        rq2_sender_count=len([v for v in sender_counts.values() if v > 0]),
+        padding_applied=spec.padding_applied, span_houses=spec.span_houses(),
+        run_dir=str(run_dir), per_circuit_seeds=per_seeds,
+    )
+
+
+def _padding_bytes(seed: int) -> int:
+    """Cover-traffic size drawn from the R1 PADDING stream (deterministic per
+    seed) — the on+padding arm's genuinely-injected extra bytes."""
+    return 512 + SorRng(seed).stream(Domain.PADDING).next_below(3584)
+
+
+def _write_run_manifest(run_dir: Path, cell, spec, seed: int, engine: str) -> None:
+    """Immutable R2 manifest binding this run to its assembled circuit fingerprint
+    and per-hop persona fingerprints. Write-once; skipped if already present."""
+    if (run_dir / "manifest.json").exists():
+        return
+    nodes = [Node(role=h.role, persona_pub_b64=_hop_pub(seed, h.node_label), engine=engine)
+             for h in spec.hops]
+    manifest = RunManifest(
+        run_id=f"conf-{spec.fingerprint()[:12]}-r{cell.rq}-{seed:016x}",
+        sor_seed=seed,
+        topology=cell.factors.get("topology", spec.topology),
+        selector=cell.factors.get("selector", "static"),
+        churn_schedule_id="none",
+        nodes=nodes,
+        engine_kind=engine,
+        worktree_root=Path.cwd(),
+    )
+    write_manifest(run_dir, manifest)
+
+
+def _hop_pub(seed: int, label: str) -> str:
+    import base64
+    raw = hashlib.sha256(f"sor-hoppub|{seed}|{label}".encode()).digest()
+    return base64.b64encode(raw).decode()
+
+
+# --------------------------------------------------------------------------- #
+# The battery: schedule -> per-run collection -> per-cell aggregate.
+# --------------------------------------------------------------------------- #
+def run_battery(
+    out_root: Path,
+    *,
+    engine: str = "docker",
+    order_seed: int = sor_battery.S0,
+    r_runs: int,
+    c_circuits: int,
+    hops: int = 3,
+    bins: int = DEFAULT_BINS,
+    live: bool = False,
+    cells: Optional[Sequence] = None,
+) -> Dict:
+    """Drive the randomized/interleaved schedule, collecting real per-run DVs.
+
+    ``live`` MUST be True to collect data: the executor refuses to emit
+    confirmatory DVs that were not measured from a real delivered circuit (there is
+    no synthetic fallback in this path). Returns an aggregate report and writes a
+    write-once ``battery-results.json`` under ``out_root``."""
+    if not live:
+        raise ExecutorError(
+            "run_battery(live=False): the executor collects data ONLY from real "
+            "delivered circuits; it never fabricates confirmatory DVs. Pass live=True "
+            "(operator-gated) to collect."
+        )
+    assert_isolated(engine)
+    out_root = Path(out_root)
+    out_root.mkdir(parents=True, exist_ok=True)
+
+    schedule = sor_battery.battery_schedule(order_seed, r=r_runs)
+    by_cell = {c.cell_id: c for c in (cells or sor_battery.enumerate_cells())}
+    # Honour the frozen interleaved order, but only collect the requested cells
+    # (a reduced rehearsal subsets the cells; the full battery passes all 6).
+    schedule = [pr for pr in schedule if pr.cell_id in by_cell]
+
+    reports: List[RunReport] = []
+    for pr in schedule:
+        cell = by_cell[pr.cell_id]
+        rep = run_cell_run(
+            cell, pr.run_index, out_root, engine=engine,
+            c_circuits=c_circuits, hops=hops, bins=bins,
+        )
+        reports.append(rep)
+
+    # Per-cell aggregate of the measured DV distributions (no CI/Holm here — that
+    # is the confirm.py reporting layer; this writes the raw measured rows).
+    agg: Dict[str, Dict] = {}
+    for rep in reports:
+        a = agg.setdefault(rep.cell_id, {
+            "rq": rep.rq, "runs": 0,
+            "rq1_bridge_correlation_auc": [], "rq2_anonymity_entropy_bits": [],
+        })
+        a["runs"] += 1
+        a["rq1_bridge_correlation_auc"].append(rep.rq1_bridge_correlation_auc)
+        a["rq2_anonymity_entropy_bits"].append(rep.rq2_anonymity_entropy_bits)
+
+    doc = {
+        "schema": "sor-battery-results/1",
+        "engine": engine,
+        "order_seed": order_seed,
+        "r_runs": r_runs,
+        "c_circuits": c_circuits,
+        "hops": hops,
+        "bins": bins,
+        "measured_from": "live-docker-pcap",
+        "n_runs": len(reports),
+        "cells": agg,
+        "runs": [vars(r) for r in reports],
+    }
+    path = out_root / "battery-results.json"
+    if not path.exists():
+        path.write_text(json.dumps(doc, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    doc["_results_path"] = str(path)
+    return doc

tests/test_sor_confirmatory_run.py

@@ -39,12 +39,39 @@ def test_go_holds_on_grid_completion(monkeypatch):
     assert cr.main(["--operator-go", "--out", "x"]) == 2
 
 
-def test_go_holds_for_operator_even_on_full_grid(monkeypatch):
+def test_go_collects_on_full_grid_with_token(monkeypatch):
     monkeypatch.setattr(cr, "preflight", lambda *a, **k: dict(_READY_GRID_FULL))
     monkeypatch.setattr(cr, "verify_freeze", lambda: True)
     monkeypatch.setenv(cr.OPERATOR_TOKEN_ENV, "1")
-    # Full grid + all locks: the immutable data run is still the operator's to
-    # initiate — the launcher does not fabricate confirmatory cells.
+    # Stub the live executor so the unit test does not stand up docker; assert the
+    # GO path actually invokes the real data collector with live=True.
+    from cmd_chat.sor import executor
+    calls = {}
+
+    def _fake_run_battery(out_root, **kw):
+        calls.update(kw)
+        calls["out_root"] = str(out_root)
+        return {"n_runs": 180, "measured_from": "live-docker-pcap",
+                "_results_path": str(out_root) + "/battery-results.json"}
+
+    monkeypatch.setattr(executor, "run_battery", _fake_run_battery)
+    # Full grid + all three locks + token: collect for real (returns 0).
+    assert cr.main(["--operator-go", "--out", "x"]) == 0
+    assert calls["live"] is True  # never a dry/synthetic confirmatory path
+
+
+def test_go_executor_refusal_never_fabricates(monkeypatch):
+    monkeypatch.setattr(cr, "preflight", lambda *a, **k: dict(_READY_GRID_FULL))
+    monkeypatch.setattr(cr, "verify_freeze", lambda: True)
+    monkeypatch.setenv(cr.OPERATOR_TOKEN_ENV, "1")
+    from cmd_chat.sor import executor
+
+    def _refuse(out_root, **kw):
+        raise executor.ExecutorError("no real pcap measurement available")
+
+    monkeypatch.setattr(executor, "run_battery", _refuse)
+    # If the executor cannot measure real data it refuses; the launcher surfaces
+    # that as a NO-GO (2) rather than emitting a fabricated DV.
     assert cr.main(["--operator-go", "--out", "x"]) == 2
 
 

tests/test_sor_executor.py

@@ -0,0 +1,65 @@
+"""Confirmatory data-collection executor — anti-fabrication guard + real pcap
+measurement. The heavy live-docker path is exercised by the operator rehearsal;
+here we pin the offline invariants that keep the confirmatory DVs honest.
+"""
+
+import struct
+
+import pytest
+
+from cmd_chat.sor import executor
+
+
+def test_run_battery_refuses_non_live():
+    # The executor collects DVs ONLY from real delivered circuits — never a
+    # synthetic/dry confirmatory path.
+    with pytest.raises(executor.ExecutorError):
+        executor.run_battery("output/never", r_runs=1, c_circuits=1, live=False)
+
+
+def test_measure_flow_refuses_missing_pcap():
+    with pytest.raises(executor.ExecutorError):
+        executor._measure_flow({0: "only-ingress"}, last_hop=2, bins=8)
+
+
+def test_bin_pcap_bytes_refuses_unreadable(tmp_path):
+    bad = tmp_path / "not.pcap"
+    bad.write_bytes(b"not a pcap file")
+    with pytest.raises(executor.ExecutorError):
+        executor.bin_pcap_bytes(bad, bins=8)
+
+
+def _write_minimal_pcap(path, packets):
+    """A minimal little-endian pcap (LINKTYPE_RAW=101) with the given (ts, payload)
+    packets, so we can exercise real binning without docker."""
+    with open(path, "wb") as f:
+        # global header: magic, ver 2.4, tz/sig 0, snaplen, network=101 (RAW)
+        f.write(struct.pack("<IHHiIII", 0xA1B2C3D4, 2, 4, 0, 0, 65535, 101))
+        for ts, payload in packets:
+            sec = int(ts)
+            usec = int((ts - sec) * 1_000_000)
+            f.write(struct.pack("<IIII", sec, usec, len(payload), len(payload)))
+            f.write(payload)
+
+
+def test_bin_pcap_bytes_bins_real_bytes(tmp_path):
+    p = tmp_path / "cap.pcap"
+    # Two packets at t=0 and t=1, 100 bytes each, into 2 bins → [100, 100].
+    _write_minimal_pcap(p, [(0.0, b"x" * 100), (1.0, b"y" * 100)])
+    series = executor.bin_pcap_bytes(p, bins=2)
+    assert len(series) == 2
+    assert sum(series) == 200
+    assert series[0] == 100 and series[-1] == 100
+
+
+def test_bin_pcap_bytes_empty_is_zero_series(tmp_path):
+    p = tmp_path / "empty.pcap"
+    _write_minimal_pcap(p, [])
+    assert executor.bin_pcap_bytes(p, bins=4) == [0, 0, 0, 0]
+
+
+def test_circuit_seed_is_deterministic_and_distinct():
+    a = executor._circuit_seed(12345, 0)
+    b = executor._circuit_seed(12345, 0)
+    c = executor._circuit_seed(12345, 1)
+    assert a == b and a != c