leetcrypt
72a6c7b519
Enumerates the frozen §2 confirmatory cells (3 RQ1 bridge levels + 3 RQ2 topology
levels at their controls; bridge-off+padding recorded as a declared N/A, not run),
derives each run's seed by the frozen §4 rule (SHA256('<S0>|<cell_id>|<run_index>')
big-endian first 8 bytes -> u64, S0=20260719), and lays out the §2 schedule with
run order randomized within each cell and each RQ's control interleaved before and
after its treatments.
write_cell_plan emits the auditable cell x run plan artifact (R=30, C=50 -> 180
runs / 9000 circuits, seed rule, matched-N rule, full schedule). dry_pass exercises
the R2/R3 provenance pipeline on FIXTURES only (deterministic replay_and_seal, no
engine, no traffic) to prove schema-valid + checksummed provenance and that a seed
reproduces its circuit-build sequence. Collects no confirmatory data — that remains
the human gate.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
127 lines
4.9 KiB
Python
127 lines
4.9 KiB
Python
"""Confirmatory-battery orchestration — plan integrity + DRY provenance pass.
|
|
|
|
Validates the RQ1+RQ2 start-line layer without collecting any confirmatory data:
|
|
the frozen cell enumeration, the §4 seed rule (independently recomputed), the §2
|
|
randomized/interleaved schedule, and a fixture-only DRY pass that proves the
|
|
R2/R3 provenance pipeline emits schema-valid, checksummed artifacts and that a
|
|
seed reproduces its circuit-build sequence.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
|
|
from cmd_chat.sor.battery import (
|
|
C_CIRCUITS,
|
|
R_RUNS,
|
|
S0,
|
|
battery_schedule,
|
|
declared_na_cells,
|
|
derive_seed,
|
|
dry_pass,
|
|
enumerate_cells,
|
|
plan_runs,
|
|
write_cell_plan,
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Seed rule (frozen §4).
|
|
# --------------------------------------------------------------------------- #
|
|
def test_derive_seed_matches_independent_recomputation():
|
|
cell_id = "RQ1/topo=1house/selector=static/bridge=on"
|
|
got = derive_seed(cell_id, 7)
|
|
want = int.from_bytes(
|
|
hashlib.sha256(f"{S0}|{cell_id}|7".encode()).digest()[:8], "big"
|
|
)
|
|
assert got == want
|
|
assert 0 <= got <= (1 << 64) - 1
|
|
|
|
|
|
def test_derive_seed_is_deterministic_and_per_run_distinct():
|
|
cid = "RQ2/bridge=off/selector=static/topo=bridge-federated"
|
|
assert derive_seed(cid, 3) == derive_seed(cid, 3)
|
|
seeds = {derive_seed(cid, i) for i in range(R_RUNS)}
|
|
assert len(seeds) == R_RUNS # no collisions across a cell's runs
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Cell enumeration.
|
|
# --------------------------------------------------------------------------- #
|
|
def test_enumerate_cells_is_three_rq1_and_three_rq2():
|
|
cells = enumerate_cells()
|
|
assert len(cells) == 6
|
|
assert sum(c.rq == "RQ1" for c in cells) == 3
|
|
assert sum(c.rq == "RQ2" for c in cells) == 3
|
|
# exactly one control per RQ
|
|
assert sum(c.is_control for c in cells if c.rq == "RQ1") == 1
|
|
assert sum(c.is_control for c in cells if c.rq == "RQ2") == 1
|
|
assert all(not c.na for c in cells)
|
|
|
|
|
|
def test_declared_na_cell_is_bridge_off_padding_and_not_run():
|
|
na = declared_na_cells()
|
|
assert len(na) == 1 and na[0].na
|
|
assert na[0].factors["bridge"] == "off+padding"
|
|
# N/A cells never appear in the run plan.
|
|
ids = {pr.cell_id for pr in plan_runs()}
|
|
assert na[0].cell_id not in ids
|
|
|
|
|
|
def test_plan_runs_is_six_cells_by_R():
|
|
plan = plan_runs()
|
|
assert len(plan) == 6 * R_RUNS
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Schedule — within-cell randomization + control interleave (§2).
|
|
# --------------------------------------------------------------------------- #
|
|
def test_schedule_brackets_each_rq_control_around_its_treatments():
|
|
sched = battery_schedule(order_seed=S0)
|
|
# Every planned run appears exactly once.
|
|
assert len(sched) == 6 * R_RUNS
|
|
for rq in ("RQ1", "RQ2"):
|
|
idxs = [i for i, pr in enumerate(sched) if pr.rq == rq]
|
|
controls = [i for i in idxs if sched[i].is_control]
|
|
treatments = [i for i in idxs if not sched[i].is_control]
|
|
# control runs exist before the first and after the last treatment.
|
|
assert min(controls) < min(treatments)
|
|
assert max(controls) > max(treatments)
|
|
|
|
|
|
def test_schedule_is_deterministic_from_order_seed():
|
|
a = [(pr.cell_id, pr.run_index) for pr in battery_schedule(order_seed=123)]
|
|
b = [(pr.cell_id, pr.run_index) for pr in battery_schedule(order_seed=123)]
|
|
assert a == b
|
|
c = [(pr.cell_id, pr.run_index) for pr in battery_schedule(order_seed=999)]
|
|
assert a != c # a different ordering seed reshuffles
|
|
|
|
|
|
def test_write_cell_plan_artifact(tmp_path):
|
|
path = write_cell_plan(tmp_path)
|
|
doc = json.loads(path.read_text())
|
|
assert doc["base_seed_S0"] == S0
|
|
assert doc["R_runs_per_cell"] == R_RUNS
|
|
assert doc["C_circuits_per_run"] == C_CIRCUITS
|
|
assert doc["n_run_cells"] == 6
|
|
assert doc["total_runs"] == 6 * R_RUNS
|
|
assert doc["total_circuits"] == 6 * R_RUNS * C_CIRCUITS
|
|
assert len(doc["schedule"]) == 6 * R_RUNS
|
|
assert len(doc["declared_na_cells"]) == 1
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# DRY provenance pass (fixtures only).
|
|
# --------------------------------------------------------------------------- #
|
|
def test_dry_pass_emits_valid_checksummed_provenance(tmp_path):
|
|
summary = dry_pass(tmp_path, runs=2)
|
|
assert summary["all_sha_match"] is True
|
|
assert summary["all_seed_reproduces"] is True
|
|
assert summary["distinct_seeds"] is True
|
|
# Each run wrote a manifest + events log whose sha is sealed.
|
|
for rep in summary["reports"]:
|
|
run_dir = tmp_path / rep["run_id"]
|
|
manifest = json.loads((run_dir / "manifest.json").read_text())
|
|
assert manifest["events"]["sha256"] == rep["events_sha256"]
|
|
assert manifest["sor_seed"] == rep["seed"]
|
|
assert (run_dir / "events.jsonl").exists()
|