hack-house/cmd_chat/sor/analysis/rq2p3_calibration.py

"""RQ2-P3 mechanism-instrument calibration gate (`rq2p3-mechanism-prereg.md` §7).

DRY, SYNTHETIC-ONLY calibration of the new ``bridge-federated-pool`` topology: it
assembles circuits from harness-generated per-circuit seeds (NOT a confirmatory data
dir), then reports the four §7 gate items. **No confirmatory record is read** — this
is the same blind-safe discipline as the lead-paper loaders; the confirmatory battery
stays HARD-HELD until the prereg is frozen.

Gate items (§7):
  1. Reproduce the lead degeneracy — ``B=50, alpha=0`` → ``c_i`` ≈ 1/C, near-constant,
     Spearman ρ inconclusive.
  2. Extreme ``B=1`` — all circuits share one bridge → ``c=1.0`` (a degenerate
     concentration extreme). NOTE: under the RATIFIED posterior a fully-shared bridge
     *maximizes* the observation-consistent anonymity set, so realized H comes out
     HIGH, not low — the mechanical "mix" the study exists to test. This is surfaced,
     not silently reconciled with the naive "low H" gloss.
  3. Monotonicity — mean top-3 concentration decreases in B and increases in alpha.
  4. Entropy calibration (inherited) — plug-in entropy of N equiprobable senders is
     exactly log2(N); Miller–Madow adds only the documented finite-N bias term.
"""

from __future__ import annotations

import hashlib
import json
import math
import statistics
import sys
from typing import Dict, List

from cmd_chat.sor.analysis.confirm_load_rq2 import (
    bridge_concentration,
    per_circuit_entropy,
    top_k_bridge_concentration,
)
from cmd_chat.sor.analysis.stats import miller_madow_entropy_bits, spearman
from cmd_chat.sor.assembler import assemble
from cmd_chat.sor.battery import Cell, derive_seed, enumerate_rq2p3_cells

R_DRY = 30   # runs/cell for the dry calibration (matches the prereg §6 R)
C_DRY = 50   # circuits/run (matches C)


def _run_circuit_seeds(cell_id: str, run_index: int, c: int = C_DRY) -> List[int]:
    """Harness-side per-circuit seeds for the DRY pass (the confirmatory executor
    persists real per_circuit_seeds; this is calibration only). Deterministic from
    the frozen per-run seed rule so the calibration is reproducible."""
    run_seed = derive_seed(cell_id, run_index)
    return [int.from_bytes(hashlib.sha256(f"{run_seed}|circ|{j}".encode()).digest()[:8], "big")
            for j in range(c)]


def _assemble_run(cell: Cell, run_index: int, c: int = C_DRY):
    return [assemble(cell, s) for s in _run_circuit_seeds(cell.cell_id, run_index, c)]


def _pool_cell(b: int, alpha: float) -> Cell:
    return Cell("RQ2P3", f"RQ2P3/dry/B={b}/alpha={alpha}",
                {"bridge": "off", "topology": "bridge-federated-pool",
                 "selector": "static", "pool_B": str(b), "pool_alpha": str(alpha)}, False)


def cell_report(cell: Cell, r: int = R_DRY, c: int = C_DRY) -> Dict:
    """Per-cell dry report: mean top-3 concentration over runs, pooled per-circuit
    (c_i, H_i) Spearman ρ, and the concentration spread."""
    per_run_top3: List[float] = []
    per_run_mean_h: List[float] = []
    conc_all: List[float] = []
    h_all: List[float] = []
    for ri in range(r):
        specs = _assemble_run(cell, ri, c)
        per_run_top3.append(top_k_bridge_concentration(specs, k=3))
        ent = per_circuit_entropy(specs)
        per_run_mean_h.append(statistics.fmean(ent))
        for ci, hi in zip(bridge_concentration(specs), ent):
            if ci is not None:
                conc_all.append(ci)
                h_all.append(hi)
    b = int(cell.factors["pool_B"])
    alpha = float(cell.factors["pool_alpha"])
    return {
        "B": b,
        "alpha": alpha,
        "mean_top3_concentration": statistics.fmean(per_run_top3),
        "concentration_stdev": statistics.pstdev(conc_all) if conc_all else 0.0,
        "mean_entropy_bits": statistics.fmean(per_run_mean_h),
        "spearman_rho_conc_vs_H": spearman(conc_all, h_all),
        "n_bridged_circuits": len(conc_all),
    }


def calibration_gate(r: int = R_DRY, c: int = C_DRY) -> Dict:
    """Run all four §7 gate items on the DRY pass and return a report dict."""
    sweep = [cr for cr in (cell_report(cell, r, c) for cell in enumerate_rq2p3_cells())]
    grid = {(cr["B"], cr["alpha"]): cr for cr in sweep}

    # Item 1 — anchor B=50, alpha=0 reproduces the lead degeneracy.
    anchor = grid[(50, 0.0)]
    item1_pass = anchor["concentration_stdev"] < 1e-3 and abs(anchor["spearman_rho_conc_vs_H"]) < 0.05

    # Item 2 — extreme B=1: all circuits share one bridge (degenerate concentration).
    ext = cell_report(_pool_cell(1, 0.0), r, c)
    lead_like = cell_report(_pool_cell(1_000_000, 0.0), r, c)  # huge pool ≈ fresh bridge/circuit
    item2_conc_ok = abs(ext["mean_top3_concentration"] - 1.0) < 1e-9
    # The naive gate text says "low H"; the ratified posterior yields HIGH H. Surface it.
    item2_entropy_direction = "HIGH (mechanical mix)" if ext["mean_entropy_bits"] > lead_like["mean_entropy_bits"] else "low"
    item2_matches_naive_low_H = ext["mean_entropy_bits"] < lead_like["mean_entropy_bits"]

    # Item 3 — monotonicity: mean top-3 concentration decreasing in B, increasing in alpha.
    dec_in_B = all(
        grid[(2, a)]["mean_top3_concentration"] >= grid[(4, a)]["mean_top3_concentration"] >= grid[(8, a)]["mean_top3_concentration"]
        for a in (0.0, 1.0, 2.0)
    )
    inc_in_alpha = all(
        grid[(b, 0.0)]["mean_top3_concentration"] <= grid[(b, 1.0)]["mean_top3_concentration"] <= grid[(b, 2.0)]["mean_top3_concentration"]
        for b in (2, 4, 8)
    )
    item3_pass = dec_in_B and inc_in_alpha

    # Item 4 — entropy calibration (inherited): plug-in H of N equiprobable = log2(N) exactly.
    entropy_checks = []
    item4_pass = True
    for n in (2, 4, 8, 16, 50):
        mm = miller_madow_entropy_bits([1] * n)
        bias = (n - 1) / (2.0 * n * math.log(2.0))
        plugin = mm - bias
        ok = abs(plugin - math.log2(n)) < 1e-9
        item4_pass = item4_pass and ok
        entropy_checks.append({"N": n, "plugin_bits": plugin, "log2N": math.log2(n),
                               "miller_madow_bits": mm, "plugin_equals_log2N": ok})

    return {
        "schema": "sor-rq2p3-calibration/1",
        "dry_only": True,
        "no_confirmatory_data_read": True,
        "R": r, "C": c,
        "sweep": sweep,
        "gate": {
            "item1_reproduce_lead_degeneracy": {
                "anchor_B50_alpha0": anchor,
                "pass": item1_pass,
            },
            "item2_extreme_B1": {
                "mean_top3_concentration": ext["mean_top3_concentration"],
                "mean_entropy_bits": ext["mean_entropy_bits"],
                "fresh_bridge_reference_mean_entropy_bits": lead_like["mean_entropy_bits"],
                "concentration_degenerate_c_eq_1": item2_conc_ok,
                "entropy_direction_vs_naive": item2_entropy_direction,
                "matches_naive_low_H_expectation": item2_matches_naive_low_H,
                "NEEDS_OPERATOR": not item2_matches_naive_low_H,
                "note": (
                    "Concentration extreme is as specified (c=1.0). But the ratified "
                    "posterior makes a fully-shared bridge MAXIMIZE the anonymity set, so "
                    "realized H is HIGH, contradicting the §7 gate-item-2 'low H' gloss. "
                    "This is exactly the mechanical 'mix' (ρ>0) the study exists to test "
                    "(see note-unique-bridge-artifact.md), surfaced early at calibration. "
                    "Not silently reconciled: flagged for the operator's freeze decision."
                ),
            },
            "item3_monotonicity": {
                "decreasing_in_B": dec_in_B,
                "increasing_in_alpha": inc_in_alpha,
                "pass": item3_pass,
            },
            "item4_entropy_calibration": {
                "checks": entropy_checks,
                "pass": item4_pass,
            },
        },
    }


if __name__ == "__main__":
    print(json.dumps(calibration_gate(), indent=2, sort_keys=True))
    sys.exit(0)