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>

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