hack-house/tests/test_sor_confirm_load_rq2.py

"""SS3 RQ2 loader — the RATIFIED per-circuit sender-posterior construction, on
SYNTHETIC specs/seeds ONLY.

These pin that the offline loader implements the ratified stage-05 rule
(observation-consistent anonymity set → uniform posterior → Miller–Madow H) and
produces the exact inputs confirm.rq2_p1_delta_h / rq2_p3_funnel consume — WITHOUT
reading any confirmatory record (prereg §2 blinding: ratification unblocks the
CODE, not the results). Specs are hand-built; the one reconstruction test drives
the deterministic assembler on chosen seeds (no confirmatory data dir).
"""

import math

from cmd_chat.sor.analysis import confirm, confirm_load_rq2 as rq2
from cmd_chat.sor.analysis.stats import miller_madow_entropy_bits
from cmd_chat.sor.assembler import CircuitSpec, HopSpec
from cmd_chat.sor.battery import derive_seed, enumerate_cells


def _spec(topology, entry, exit_house, bridge, *, matched_n=8):
    """A minimal 3-hop CircuitSpec: entry(houseA) → [bridge?] → exit(exit_house)."""
    hops = [HopSpec(0, "entry", "houseA", entry, False)]
    if bridge is not None:
        hops.append(HopSpec(1, "bridge", "bridge", bridge, True))
    else:
        hops.append(HopSpec(1, "relay", "houseA", f"{entry}-r", False))
    hops.append(HopSpec(2, "exit", exit_house, f"{exit_house}#x", False))
    return CircuitSpec(
        cell_id=f"RQ2/{topology}", rq="RQ2", seed=0, engine="docker",
        topology=topology, bridge_present=bridge is not None, padding_applied=False,
        matched_n=matched_n, houses=("houseA", exit_house), hops=tuple(hops),
    )


def test_single_house_arm_uses_full_matched_n_pool():
    # 1house-N: no federation narrows the pool → m_i = matched N for every circuit,
    # regardless of which entry each circuit realized (ratified doc: A_i = all N).
    specs = [_spec("1house-N", e, "houseA", None, matched_n=8) for e in ("a", "a", "b")]
    assert rq2.observation_consistent_sizes(specs) == [8, 8, 8]
    assert rq2.per_circuit_posteriors(specs) == [[1] * 8, [1] * 8, [1] * 8]
    exp = miller_madow_entropy_bits([1] * 8)
    assert all(abs(h - exp) < 1e-12 for h in rq2.per_circuit_entropy(specs))


def test_federated_anonymity_set_is_distinct_entries_per_exit_signature():
    # Two exit-signature groups. Group brZ→houseB carries entries {a,a,b,c}=3
    # distinct → m=3; group brY→houseB carries {d}=1 → m=1 (H=0).
    specs = [
        _spec("bridge-federated", "a", "houseB", "brZ"),
        _spec("bridge-federated", "a", "houseB", "brZ"),
        _spec("bridge-federated", "b", "houseB", "brZ"),
        _spec("bridge-federated", "c", "houseB", "brZ"),
        _spec("bridge-federated", "d", "houseB", "brY"),
    ]
    assert rq2.observation_consistent_sizes(specs) == [3, 3, 3, 3, 1]
    ent = rq2.per_circuit_entropy(specs)
    assert abs(ent[0] - miller_madow_entropy_bits([1, 1, 1])) < 1e-12
    assert ent[4] == 0.0  # singleton anonymity set → zero entropy


def test_bridge_concentration_is_per_circuit_load_share():
    specs = [
        _spec("bridge-federated", "a", "houseB", "brA"),
        _spec("bridge-federated", "b", "houseB", "brA"),
        _spec("bridge-federated", "c", "houseB", "brA"),
        _spec("bridge-federated", "d", "houseB", "brB"),
    ]
    assert rq2.bridge_concentration(specs) == [0.75, 0.75, 0.75, 0.25]
    # top-3 distinct-bridge share over the run (frozen §6 k=3): (3 + 1)/4 = 1.0.
    assert rq2.top_k_bridge_concentration(specs, k=3) == 1.0


def test_rq2_p3_pairs_drop_bridgeless_circuits():
    specs = [
        _spec("bridge-federated", "a", "houseB", "brA"),
        _spec("bridge-federated", "b", "houseB", "brA"),
        _spec("directory-federated", "c", "houseB", None),  # no bridge → dropped
    ]
    conc, ent = rq2.rq2_p3_pairs(specs)
    assert len(conc) == 2 and len(ent) == 2       # only the two bridged circuits
    assert conc == [1.0, 1.0]                      # both share brA (2/2)


def test_reconstruct_run_is_deterministic_from_seeds():
    # Drive the real assembler on chosen seeds (SYNTHETIC — no confirmatory dir).
    cell = next(c for c in enumerate_cells()
                if c.factors.get("topology") == "bridge-federated")
    seeds = [derive_seed(cell.cell_id, i) for i in range(4)]
    a = rq2.reconstruct_run(cell, seeds)
    b = rq2.reconstruct_run(cell, seeds)
    assert [s.fingerprint() for s in a] == [s.fingerprint() for s in b]
    # Each reconstructed federated circuit spans >=2 houses (split knowledge).
    assert all(s.span_houses() >= 2 for s in a)
    # Sizes are computable and positive for every circuit.
    assert all(m >= 1 for m in rq2.observation_consistent_sizes(a))


def test_feeds_confirm_rq2_p1_two_sided_shrink():
    # Federated (small sets) vs matched-N single-house (full pool): ΔH < 0 here, so
    # the frozen two-sided gate reports an honest shrink — exercises the wiring end
    # to end (no confirmatory data; hand-built vectors).
    federated = rq2.per_circuit_posteriors(
        [_spec("bridge-federated", e, "houseB", "brZ") for e in ("a", "b")]
    )  # each group has 2 distinct entries → m=2
    single = rq2.per_circuit_posteriors(
        [_spec("1house-N", e, "houseA", None, matched_n=8) for e in ("a", "b")]
    )  # m=8
    test = confirm.rq2_p1_delta_h(federated, single, seed=1, n_resamples=200)
    assert test.name == "RQ2-P1" and test.effect == "ΔH"
    assert test.ci.point < 0.0 and test.decision == "shrink"