loss-manifest companion: article + 155-entry rated registry + sidecar + loss library + gates/battery + campaign beds + raw run ledgers
4ef8b0b verified | """compartment_smoke.py — the loss-campaign formula-smoke battery. #TAG:loss_smoke #TAG:compartments #TAG:conditioning_gate #TAG:collinearity_gate | |
| Tree: loss campaign pass 2 (plan 2026-07-25). ONE file, Colab-cell-safe, <~60s | |
| on a 4090. Formula smoke ONLY — shapes, gradients, identities, conditioning, | |
| memory. NO training, ever (MANIFEST rider). Accuracy verdicts are real runs. | |
| Carries the REFERENCE implementations of: | |
| - compartment_windows(): the certified cosine-crossfade ramp, parameterized | |
| (parity-asserted bit-exact against amoe.diffusion band_weights at its | |
| native constants — the "reuse verbatim" proof is a test, not a promise); | |
| - CompartmentMap / CompartmentDelta: rigid channel->slot partition x smooth | |
| slot->band crossfade, with the MASKED WRITE-BACK that is the entire | |
| isolation mechanism on a feature axis; | |
| - the CONDITIONING GATE (kappa^2 energy ratio; refuses predicted-inert | |
| auxiliary couplings — calibrated on the eps/flow 125-200x receipt); | |
| - the COLLINEARITY GATE (novelty = 1-|cos(grad_arm, grad_base)|; refuses | |
| role objectives that cannot pay — calibrated on dexp009 vs dexp012); | |
| - the exact fp64 Cantor warp (the ADMISSIBLE band coordinate) and the | |
| soft-staircase non-monotonicity regression (the INADMISSIBLE one). | |
| Run: python tools/compartment_smoke.py (or paste as one Colab cell) | |
| """ | |
| import inspect | |
| import math | |
| import os | |
| import sys | |
| import time | |
| import zlib | |
| import torch | |
| import torch.nn as nn | |
| import torch.nn.functional as F | |
| # ---------------------------------------------------------------- environment | |
| def _repo_root(): | |
| d = os.path.abspath(os.getcwd()) | |
| while True: | |
| if os.path.exists(os.path.join(d, "MANIFEST.md")): | |
| return d | |
| p = os.path.dirname(d) | |
| if p == d: | |
| return os.getcwd() | |
| d = p | |
| ROOT = _repo_root() | |
| for _p in (os.path.join(ROOT, "tools"), | |
| os.path.join(ROOT, "closeout_2026-07-19", "amoe", "src")): | |
| if os.path.isdir(_p) and _p not in sys.path: | |
| sys.path.insert(0, _p) | |
| torch.backends.cuda.matmul.allow_tf32 = False # pin_precision (law) | |
| torch.backends.cudnn.allow_tf32 = False | |
| DEV = "cuda" if torch.cuda.is_available() else "cpu" | |
| if DEV == "cuda": | |
| torch.cuda.set_per_process_memory_fraction(0.73) # WDDM standing cap | |
| def seed_for(name: str) -> int: | |
| """crc32, never hash() — PYTHONHASHSEED nondeterminism is a recorded law.""" | |
| return zlib.crc32(name.encode("utf-8")) & 0x7FFFFFFF | |
| try: | |
| from geolip_vitals import (_pentachoron_volumes, axis_aliveness, | |
| pentachoron_cv) | |
| HAVE_VITALS = True | |
| except Exception: | |
| HAVE_VITALS = False | |
| try: | |
| from amoe.diffusion.core.multiband import band_weights as amoe_band_weights | |
| HAVE_AMOE = True | |
| except Exception: | |
| HAVE_AMOE = False | |
| # --------------------------------------------------- windows (the smooth layer) | |
| def compartment_windows(coord: torch.Tensor, edges=(1/3, 2/3), | |
| xfade: float = 0.06) -> torch.Tensor: | |
| """The certified cosine-crossfade windows, parameterized. Identical math to | |
| amoe.diffusion.core.multiband.band_weights (parity test T01b); `coord` is a | |
| monotone band coordinate in [0,1] — a FUNCTION OF INDEX ONLY on the feature | |
| axis (T04). Rows sum to 1 everywhere; max step pi/(4*xfade) per unit.""" | |
| def ramp(x): | |
| t = ((x / xfade).clamp(-1, 1) + 1) / 2 | |
| return 0.5 - 0.5 * torch.cos(t * math.pi) | |
| e1, e2 = edges | |
| up1, up2 = ramp(coord - e1), ramp(coord - e2) | |
| low = 1 - up1 | |
| mid = up1 * (1 - up2) | |
| high = up1 * up2 | |
| return torch.stack([low, mid, high], dim=-1) | |
| # ------------------------------------------------ Cantor coordinate (the warp) | |
| def exact_cantor(x: torch.Tensor, L: int = 12) -> torch.Tensor: | |
| """Exact floor-based Cantor function, fp64, MONOTONE by construction. The | |
| ADMISSIBLE static warp: built once, frozen, no gradient path — measure | |
| space is entered exactly once (cantor law).""" | |
| r = x.double().clone() | |
| out = torch.zeros_like(r) | |
| alive = torch.ones_like(r) | |
| for k in range(1, L + 1): | |
| d = torch.floor(3.0 * r).clamp(max=2.0) # endpoint: 3*1.0 -> digit 2 | |
| r = 3.0 * r - d | |
| out = out + alive * (2.0 ** -k) * (d >= 1).double() | |
| alive = alive * (d != 1).double() | |
| return out | |
| def soft_cantor_ungated(x: torch.Tensor, L: int = 12, tau: float = 0.25, | |
| alpha: float = 0.5) -> torch.Tensor: | |
| """The soft alpha-form staircase (centers .5/1.5/2.5, soft trit, no stop | |
| gate) — reimplemented from the recorded formula for the NON-MONOTONICITY | |
| REGRESSION (timeline 2026-07-25): it keeps accumulating digits past the | |
| first 1, so it is INADMISSIBLE as a band coordinate. Feature use is fine.""" | |
| centers = torch.tensor([0.5, 1.5, 2.5], dtype=torch.float64) | |
| xx = x.double() | |
| out = torch.zeros_like(xx) | |
| for k in range(1, L + 1): | |
| y = (xx * (3.0 ** (k - 1))) % 1.0 * 3.0 | |
| p = torch.softmax(-(y.unsqueeze(-1) - centers) ** 2 / tau, dim=-1) | |
| out = out + (p[..., 2] + alpha * p[..., 1]) * (2.0 ** -k) | |
| return out | |
| def exact_cantor_ungated(x: torch.Tensor, L: int = 12, | |
| alpha: float = 0.5) -> torch.Tensor: | |
| """Exact-arithmetic UNGATED accumulation (digit-2 full bit, digit-1 | |
| alpha-bit, never stops) — the alpha-law regression pair for T06.""" | |
| r = x.double().clone() | |
| out = torch.zeros_like(r) | |
| for k in range(1, L + 1): | |
| d = torch.floor(3.0 * r).clamp(max=2.0) | |
| r = 3.0 * r - d | |
| out = out + (2.0 ** -k) * ((d == 2).double() + alpha * (d == 1).double()) | |
| return out | |
| # ------------------------------------------------------- the compartment map | |
| def build_compartment_map(P: int = 32, Ds: int = 4, d: int = 128, B: int = 3, | |
| xfade_slots: float = 1.92, warp: str = "identity"): | |
| """RIGID channel->slot partition (static int64) x SMOOTH slot->band | |
| crossfade. The coordinate is a function of INDEX ONLY (torch.arange) — | |
| never of activations; no argmax/topk/softmax selection appears in this | |
| path (source-inspected by T04). Built ONCE, fp64 warp, frozen buffers.""" | |
| assert d % P == 0, "rigid partition must tile exactly" | |
| member = torch.arange(d, dtype=torch.int64) // (d // P) | |
| c = (torch.arange(P, dtype=torch.float64) + 0.5) / P | |
| if warp == "cantor_exact": | |
| c = exact_cantor(c) | |
| c = (c - c.min()) / (c.max() - c.min()).clamp_min(1e-12) | |
| W_slot = compartment_windows(c, edges=(1/3, 2/3), | |
| xfade=xfade_slots / P).float() | |
| W_chan = W_slot[member] | |
| fp = zlib.crc32(member.numpy().tobytes() | |
| + W_slot.numpy().tobytes() + W_chan.numpy().tobytes()) | |
| return {"member": member, "coord": c.float(), "W_slot_band": W_slot, | |
| "W_chan_band": W_chan, "P": P, "Ds": Ds, "B": B, "d": d, | |
| "fingerprint": fp} | |
| class SquaredReLU(nn.Module): | |
| def forward(self, x): | |
| return F.relu(x) ** 2 | |
| class MiniAleph(nn.Module): | |
| """Minimal aleph read: M_hat = sum_k sinh(u_k) A_k / sum_k cosh(u_k), | |
| stabilized by max-|u| factor-out. Codebook is the only parameter; `home` | |
| is the frozen init snapshot (drift gauge). No argmax, no roster.""" | |
| def __init__(self, K=64, D=4, tau=0.1, gen=None): | |
| super().__init__() | |
| cb = F.normalize(torch.randn(K, D, generator=gen), dim=-1) | |
| self.codebook = nn.Parameter(cb.clone()) | |
| self.register_buffer("home", cb.clone()) | |
| self.tau = tau | |
| def m_hat(self, x): # x: (..., D) rows on the sphere | |
| A = F.normalize(self.codebook, dim=-1) | |
| u = (F.normalize(x, dim=-1) @ A.t()) / self.tau | |
| m = u.abs().amax(dim=-1, keepdim=True) | |
| ep, en = torch.exp(u - m), torch.exp(-u - m) | |
| num = (ep - en) @ A | |
| den = (ep + en).sum(dim=-1, keepdim=True) | |
| return num / den | |
| class CompartmentDelta(nn.Module): | |
| """One site: proj -> shared aleph read per slot -> B band consumers with | |
| WINDOWED READ and MASKED WRITE. The masked write is load-bearing: on a | |
| feature axis every band is active on every sample, so without masking the | |
| write-back by the same window, isolation is exactly zero (measured | |
| own/cross 1.04x). Zero-init heads (weight AND bias) => P-INIT bit-exact; | |
| enabled=False is a code-path skip => toggle law bit-exact.""" | |
| def __init__(self, cmap, hidden=64, gen=None, head_scale=0.0): | |
| super().__init__() | |
| self.cm = cmap | |
| P, Ds, B, d = cmap["P"], cmap["Ds"], cmap["B"], cmap["d"] | |
| self.proj = nn.Linear(d, P * Ds, bias=False) | |
| nn.init.orthogonal_(self.proj.weight, generator=gen) | |
| self.addr = MiniAleph(K=64, D=Ds, gen=gen) | |
| self.cons = nn.ModuleList() | |
| for _ in range(B): | |
| head = nn.Linear(hidden, d) | |
| if head_scale == 0.0: | |
| nn.init.zeros_(head.weight) | |
| nn.init.zeros_(head.bias) # bias too — the exp006 law | |
| else: | |
| with torch.no_grad(): | |
| head.weight.normal_(0, head_scale, generator=gen) | |
| head.bias.zero_() | |
| self.cons.append(nn.Sequential(nn.Linear(P * Ds, hidden), | |
| SquaredReLU(), | |
| nn.LayerNorm(hidden), head)) | |
| self.gates = nn.Parameter(torch.full((B,), -3.0)) | |
| self.register_buffer("W_slot", cmap["W_slot_band"]) | |
| self.register_buffer("W_chan", cmap["W_chan_band"]) | |
| self.enabled = True | |
| self.band_enabled = [True] * B | |
| def forward(self, x): # x: (B?, T, d) | |
| if not self.enabled: | |
| return x | |
| P, Ds, B = self.cm["P"], self.cm["Ds"], self.cm["B"] | |
| f = self.addr.m_hat(self.proj(x).view(*x.shape[:-1], P, Ds)) | |
| delta = None | |
| for b in range(B): | |
| if not self.band_enabled[b]: | |
| continue | |
| f_b = (f * self.W_slot[:, b].view(P, 1)).reshape(*x.shape[:-1], P * Ds) | |
| piece = torch.sigmoid(self.gates[b]) * (self.W_chan[:, b] | |
| * self.cons[b](f_b)) | |
| delta = piece if delta is None else delta + piece | |
| return x if delta is None else x + delta | |
| # ------------------------------------------------------------- the two gates | |
| def conditioning_gate(w_bands: torch.Tensor, amp: torch.Tensor, | |
| amp_ref: torch.Tensor, refuse_at: float = 25.0): | |
| """kappa^2_b = band-weighted ENERGY of the prediction->quantity map's gain, | |
| relative to a reference map — the conditioning law as a pre-spend check. | |
| NEVER the pointwise mean ratio (it diverges as the reference gain -> 0). | |
| kappa^2 >= refuse_at => REFUSE, predicted inert.""" | |
| w = w_bands.double() | |
| e = (w * amp.double().unsqueeze(-1) ** 2).sum(0) / w.sum(0) | |
| er = (w * amp_ref.double().unsqueeze(-1) ** 2).sum(0) / w.sum(0) | |
| k2 = (e / er.clamp_min(1e-30)) | |
| return k2, [bool(v >= refuse_at) for v in k2] | |
| def collinearity_gate(loss_arm, loss_base, params, refuse_below: float = 0.05): | |
| """novelty = 1 - |cos(grad_arm, grad_base)| over shared params. Calibrated: | |
| HP/LP role arms 0.0026-0.0083 (WAS inert at 0.05-0.2%) vs the blob payer | |
| 0.715 (~10% win). novelty < refuse_below => REFUSE.""" | |
| def flat_grad(loss): | |
| gs = torch.autograd.grad(loss, params, retain_graph=True, | |
| allow_unused=True) | |
| return torch.cat([g.reshape(-1) for g in gs if g is not None]) | |
| ga, gb = flat_grad(loss_arm), flat_grad(loss_base) | |
| cos = F.cosine_similarity(ga.unsqueeze(0), gb.unsqueeze(0)).item() | |
| nov = 1.0 - abs(cos) | |
| return nov, nov < refuse_below | |
| def half_ulp_bf16(w: float) -> float: | |
| """Half a bf16 ULP at magnitude |w| (7 explicit mantissa bits). At 3.0 this | |
| is 0.0078125 — the exp004 sub-ULP freeze constant.""" | |
| if w == 0.0: | |
| return 2.0 ** -133 | |
| return 2.0 ** (math.floor(math.log2(abs(w))) - 7) / 2.0 | |
| # -------------------------------------------------------------------- battery | |
| RESULTS = [] | |
| def record(tid, name, ok, detail=""): | |
| RESULTS.append((tid, name, "PASS" if ok else "FAIL", detail)) | |
| return ok | |
| def skip(tid, name, why): | |
| RESULTS.append((tid, name, "SKIP", why)) | |
| def run_battery(): | |
| t0 = time.time() | |
| g = torch.Generator().manual_seed(seed_for("compartment_smoke")) | |
| cmap = build_compartment_map() | |
| P, B, d = cmap["P"], cmap["B"], cmap["d"] | |
| W = cmap["W_slot_band"] | |
| # T01 partition of unity (fp32 grid + fp64 dense) + T01b amoe parity | |
| dense = torch.linspace(0, 1, 4096, dtype=torch.float64) | |
| Wd = compartment_windows(dense, (1/3, 2/3), 0.06) | |
| ok = (W.sum(-1) - 1).abs().max().item() <= 1e-6 \ | |
| and (Wd.sum(-1) - 1).abs().max().item() <= 1e-12 \ | |
| and float(W.min()) >= 0 and float(W.max()) <= 1 | |
| record("T01", "window partition-of-unity", | |
| ok, "fp32 err %.1e fp64 err %.1e" % ( | |
| (W.sum(-1) - 1).abs().max(), (Wd.sum(-1) - 1).abs().max())) | |
| if HAVE_AMOE: | |
| s = torch.linspace(0, 1, 2048) | |
| mine = compartment_windows(s, (0.35, 0.75), 0.06) | |
| record("T01b", "verbatim parity vs amoe band_weights", | |
| torch.equal(mine, amoe_band_weights(s)), | |
| "bit-exact at amoe's native constants") | |
| else: | |
| skip("T01b", "verbatim parity vs amoe band_weights", "amoe not importable") | |
| # T02 max-step analytic bound (per-slot step; bound pi/(4*m_slots)) | |
| step = (W[1:] - W[:-1]).abs().max().item() | |
| bound = math.pi / (4 * 1.92) | |
| record("T02", "max-step analytic bound", | |
| step <= bound and step >= 0.5 * bound, | |
| "step %.4f bound %.4f (tight %.2f)" % (step, bound, step / bound)) | |
| # T03 rigid partition integrity | |
| bc = torch.bincount(cmap["member"], minlength=P) | |
| record("T03", "rigid partition integrity", | |
| bool((bc == d // P).all()) and int(cmap["member"].max()) + 1 == P | |
| and d % P == 0, "%d channels / %d slots, uniform" % (d, P)) | |
| # T04 coordinate law — source inspection (the band-coordinate idiom). | |
| # Scan the CODE only (part after the docstring close) — the docstring | |
| # names the forbidden ops, which is not the same as using them. | |
| code = inspect.getsource(build_compartment_map).split('"""')[2] | |
| ok = ("arange" in code and "argmax" not in code and "topk" not in code | |
| and "softmax" not in code and ".grad" not in code) | |
| record("T04", "coordinate is INDEX-ONLY (source-inspected)", ok, | |
| "no argmax/topk/softmax in the coordinate path") | |
| # T05 Cantor admissibility: exact monotone; soft form DETECTED non-monotone | |
| # (interior grid — the x=1.0 mod-wrap is an endpoint artifact, not the | |
| # finding; the recorded interior dips are slope -0.13..-0.49 per level) | |
| xs = torch.linspace(0, 1, 2048, dtype=torch.float64) | |
| ce_ = exact_cantor(xs) | |
| mono = float((ce_[1:] - ce_[:-1]).min()) | |
| xin = xs[:-1] | |
| worst = 0.0 | |
| for L in (3, 5, 12): | |
| sc = soft_cantor_ungated(xin, L=L) | |
| worst = min(worst, float(((sc[1:] - sc[:-1]) * (len(xin) - 1)).min())) | |
| record("T05", "Cantor coordinate admissibility", | |
| mono >= -1e-12 and worst < -0.05, | |
| "exact min-diff %.1e; soft interior min-slope %.2f (non-monotone)" | |
| % (mono, worst)) | |
| # T06 alpha-law: alpha=0.5 expectation-matches the gated form; alpha=0 collapses | |
| # alpha=0.5 is the UNIQUE expectation-matching value: a digit-1 contributes | |
| # 2^-N and stops (gated), while an unstopped continuation contributes | |
| # alpha*2^-N plus a tail averaging 0.5*2^-N — so alpha=0.5 is unbiased and | |
| # alpha=0 is systematically biased low. Distinct values must survive. | |
| cent = (torch.arange(32, dtype=torch.float64) + 0.5) / 32 # slot centroids | |
| gt5 = ce_ - exact_cantor_ungated(xs, alpha=0.5) | |
| gt0 = ce_ - exact_cantor_ungated(xs, alpha=0.0) | |
| slots32 = exact_cantor_ungated(cent, alpha=0.5) | |
| record("T06", "alpha=0.5 expectation-matching law", | |
| abs(float(gt5.mean())) < 1e-3 and float(gt5.abs().mean()) > 0.01 | |
| and float(gt0.mean()) > 0.01 | |
| and len(torch.unique(slots32)) == 32, | |
| "bias a=.5 %.1e (unbiased) vs a=0 %.3f (low); gradation %.3f; " | |
| "distinct 32/32" % (gt5.mean(), gt0.mean(), gt5.abs().mean())) | |
| # T07 map staticity across optimizer steps | |
| mod = CompartmentDelta(cmap, gen=g).to(DEV) | |
| opt = torch.optim.Adam(mod.parameters(), lr=1e-3, weight_decay=0.0) | |
| x = torch.randn(4, 16, d, generator=g).to(DEV) | |
| for _ in range(3): | |
| opt.zero_grad(set_to_none=True) | |
| ((mod(x) - x) ** 2).mean().backward() | |
| opt.step() | |
| fp2 = zlib.crc32(mod.cm["member"].numpy().tobytes() | |
| + mod.cm["W_slot_band"].numpy().tobytes() | |
| + mod.cm["W_chan_band"].numpy().tobytes()) | |
| record("T07", "map staticity (crc32 across steps)", | |
| fp2 == cmap["fingerprint"] | |
| and not mod.W_slot.requires_grad and not mod.W_chan.requires_grad, | |
| "fingerprint %08x stable" % fp2) | |
| # T08 lesion does not renormalize | |
| Wl = W.clone(); Wl[:, 1] = 0.0 | |
| record("T08", "lesion no-renormalization", | |
| float(Wl.sum(-1).max()) < 1.0 + 1e-6 | |
| and bool((Wl.sum(-1) < 1 - 1e-6).any()) | |
| and torch.equal(Wl[:, 0], W[:, 0]) and torch.equal(Wl[:, 2], W[:, 2]), | |
| "lesioned rows sum<1; other columns bit-identical") | |
| # T09 toggle / P-INIT bit-exactness | |
| fresh = CompartmentDelta(cmap, gen=g).to(DEV) | |
| xb = torch.randn(2, 8, d, generator=g).to(DEV) | |
| fresh.enabled = False | |
| off = fresh(xb) | |
| fresh.enabled = True | |
| on0 = fresh(xb) # zero-init => inert | |
| fresh.band_enabled = [False] * B | |
| les = fresh(xb) | |
| record("T09", "toggle + P-INIT + full-lesion bit-exact", | |
| torch.equal(off, xb) and torch.equal(on0, xb) | |
| and torch.equal(les, xb), "all three torch.equal") | |
| # T10/T11 gradient flow to intended / zero to unintended | |
| live = CompartmentDelta(cmap, gen=g, head_scale=0.02).to(DEV) | |
| live.band_enabled = [True, True, False] # band 2 disabled | |
| y = live(xb) | |
| loss = ((y - xb) ** 2).mean() | |
| loss.backward() | |
| flow_ok = all(p.grad is not None and float(p.grad.abs().sum()) > 0 | |
| for p in [live.proj.weight, live.addr.codebook, | |
| live.gates] | |
| ) and all( | |
| any(p.grad is not None and float(p.grad.abs().sum()) > 0 | |
| for p in live.cons[b].parameters()) for b in (0, 1)) | |
| zero_ok = (live.addr.home.grad is None and live.W_slot.grad is None | |
| and all(p.grad is None or float(p.grad.abs().sum()) == 0 | |
| for p in live.cons[2].parameters())) | |
| record("T10", "gradient FLOW to every intended parameter", flow_ok, | |
| "proj+codebook+gates+cons[0,1] all nonzero") | |
| record("T11", "gradient ZERO to every unintended parameter", zero_ok, | |
| "buffers + disabled band grad-free") | |
| # T12 cross-talk matrix (the isolation mechanism, measured) | |
| ct = CompartmentDelta(cmap, gen=torch.Generator().manual_seed( | |
| seed_for("crosstalk")), head_scale=0.02).to(DEV) | |
| xc = torch.randn(4, 16, d, | |
| generator=torch.Generator().manual_seed( | |
| seed_for("crosstalk-x"))).to(DEV) | |
| M = torch.zeros(B, B) | |
| for b in range(B): | |
| for p_ in ct.parameters(): | |
| p_.grad = None | |
| delta = ct(xc) - xc | |
| Lb = ((delta * ct.W_chan[:, b]) ** 2).mean() | |
| Lb.backward() | |
| for b2 in range(B): | |
| M[b, b2] = math.sqrt(sum(float((p.grad ** 2).sum()) | |
| for p in ct.cons[b2].parameters() | |
| if p.grad is not None)) | |
| Mn = M / M.diag().clamp_min(1e-12).unsqueeze(1) | |
| edge_zero = float(Mn[0, 2]) == 0.0 and float(Mn[2, 0]) == 0.0 | |
| own_cross = min((1.0 / Mn[b][torch.arange(B) != b].max()).item() | |
| for b in range(B)) | |
| bleed = torch.tensor([[float((W[:, a] * W[:, c]).sum() / W[:, a].sum()) | |
| for c in range(B)] for a in range(B)]) | |
| off_mask = ~torch.eye(B, dtype=torch.bool) | |
| r = torch.corrcoef(torch.stack([Mn[off_mask], bleed[off_mask]]))[0, 1] | |
| record("T12", "cross-talk: edges exactly 0, own/cross >= 10x, bleed-correlated", | |
| edge_zero and own_cross >= 10.0 and float(r) > 0.8, | |
| "LOW<->HIGH %.1e/%.1e; worst own/cross %.1fx; corr(bleed) %.2f" | |
| % (Mn[0, 2], Mn[2, 0], own_cross, r)) | |
| # T13 CONDITIONING GATE — must reproduce the eps/flow calibration | |
| betas = torch.linspace(0.00085 ** 0.5, 0.012 ** 0.5, 1000, | |
| dtype=torch.float64) ** 2 | |
| abar = torch.cumprod(1 - betas, dim=0) | |
| s01 = torch.arange(1000, dtype=torch.float64) / 1000.0 # t/1000 — the LAW | |
| wb = compartment_windows(s01, (0.35, 0.75), 0.06) # sigma-axis bands | |
| amp_eps = ((1 - abar).sqrt() / abar.sqrt()).float() # d x0 / d eps_hat | |
| amp_flow = s01.float() # d x0 / d v_hat | |
| k2, refuse = conditioning_gate(wb, amp_eps, amp_flow) | |
| k2f, refuse_f = conditioning_gate(wb, amp_flow, amp_flow) | |
| record("T13", "conditioning gate reproduces the eps/flow split", | |
| bool(k2[0] < k2[1] < k2[2]) and refuse[2] and not any(refuse_f) | |
| and 25.0 <= float(k2[2]) <= 400.0, | |
| "kappa^2 LOW %.1f MID %.1f HIGH %.1f (refuse@25: HIGH fires; " | |
| "flow self-ratio clean)" % (k2[0], k2[1], k2[2])) | |
| # T14 COLLINEARITY GATE — must reproduce HP/LP-inert vs blob-payer | |
| gc = torch.Generator().manual_seed(seed_for("collinearity")) | |
| conv = nn.Conv2d(4, 4, 3, padding=1) | |
| with torch.no_grad(): | |
| conv.weight.normal_(0, 0.1, generator=gc); conv.bias.zero_() | |
| conv = conv.to(DEV) | |
| xt = torch.randn(8, 4, 32, 32, generator=gc).to(DEV) | |
| tgt = torch.randn(8, 4, 32, 32, generator=gc).to(DEV) | |
| sig = torch.rand(8, 1, 1, 1, generator=gc).to(DEV) * 0.9 + 0.05 | |
| blob = (torch.rand(8, 1, 32, 32, generator=gc).to(DEV) > 0.7).float() | |
| def hp(z): return z - F.avg_pool2d(z, 3, stride=1, padding=1) | |
| def lp(z): return F.avg_pool2d(z, 7, stride=1, padding=3) | |
| pred = conv(xt) | |
| base = ((pred - tgt) ** 2).mean() | |
| lam = 0.5 | |
| arm_low = base + lam * ((hp(pred) - hp(tgt)) ** 2).mean() | |
| arm_high = base + lam * ((lp(pred) - lp(tgt)) ** 2).mean() | |
| x0h, x0 = xt - sig * pred, xt - sig * tgt | |
| den = blob.sum().clamp_min(1.0) * 4 | |
| blob_term = (blob * (lp(x0h) - lp(x0)) ** 2).sum() / den | |
| ps = [conv.weight, conv.bias] | |
| # Gate semantics: a COMPOSED role arm (base + filtered residual, exp009's | |
| # actual objective) is judged whole; an ADDITIVE auxiliary is judged as | |
| # THE TERM BEING ADDED — that is the new pressure whose direction matters. | |
| n_low, ref_low = collinearity_gate(arm_low, base, ps) | |
| n_high, ref_high = collinearity_gate(arm_high, base, ps) | |
| n_blob, ref_blob = collinearity_gate(lam * blob_term, base, ps) | |
| record("T14", "collinearity gate reproduces inert-vs-payer", | |
| ref_low and ref_high and not ref_blob | |
| and max(n_low, n_high) < 0.02 and n_blob > 0.3, | |
| "novelty HP %.4f LP %.4f (REFUSED) vs blob %.3f (passes)" | |
| % (n_low, n_high, n_blob)) | |
| # T15 fp32-vs-fp64 CM parity + geovocab2 reference | |
| if HAVE_VITALS: | |
| gp = torch.Generator().manual_seed(seed_for("cm-parity")) | |
| pts = F.normalize(torch.randn(200, 5, 4, generator=gp), dim=-1) | |
| v64 = _pentachoron_volumes(pts) | |
| d2 = torch.cdist(pts, pts).pow(2) # fp32 clone | |
| cm32 = torch.ones(200, 6, 6); cm32[:, 0, 0] = 0.0 | |
| cm32[:, 1:, 1:] = d2 | |
| v32 = (-torch.linalg.det(cm32) / 9216.0).clamp_min(0).sqrt() | |
| rel = ((v32 - v64).abs() / v64.abs().clamp_min(1e-12)).max().item() | |
| try: | |
| from geolip_vitals import cv_reference_check | |
| ref = "geovocab2 parity %.1e" % cv_reference_check() | |
| except Exception as e: | |
| ref = "geovocab2 skipped (%s)" % type(e).__name__ | |
| record("T15", "fp64-for-gauges precision law", | |
| rel < 0.05, "fp32 max rel err %.2e (<4%% recorded); %s" | |
| % (rel, ref)) | |
| else: | |
| skip("T15", "fp64-for-gauges precision law", "geolip_vitals not importable") | |
| # T16 memory + time: full CE vs chunked CE vs a K=64 code loss @ V=248,320 | |
| if DEV == "cuda": | |
| V, dd, T = 248_320, 1024, 2048 | |
| E = torch.randn(V, dd, device=DEV) * 0.02 | |
| R = torch.randn(64, dd, device=DEV) / math.sqrt(dd) | |
| yid = torch.randint(0, V, (1, T), device=DEV) | |
| code = (torch.randn(V, 64, device=DEV) > 0).float() * 2 - 1 | |
| def one(name, fn): | |
| h = torch.randn(1, T, dd, device=DEV, requires_grad=True) | |
| fn(h).backward(); torch.cuda.synchronize() # warm | |
| torch.cuda.reset_peak_memory_stats() | |
| h = torch.randn(1, T, dd, device=DEV, requires_grad=True) | |
| t1 = time.time(); fn(h).backward() | |
| torch.cuda.synchronize() | |
| return torch.cuda.max_memory_allocated() / 2**30, time.time() - t1 | |
| def full(h): | |
| return F.cross_entropy((h @ E.t()).reshape(-1, V), yid.reshape(-1)) | |
| def chunked(h): | |
| s, n = 0.0, 0 | |
| for i in range(0, T, 512): | |
| lg = h[:, i:i + 512] @ E.t() | |
| s = s + F.cross_entropy(lg.reshape(-1, V), | |
| yid[:, i:i + 512].reshape(-1), | |
| reduction="sum") | |
| n += lg.shape[1] | |
| return s / n | |
| def fac(h): | |
| v = (F.normalize(h, dim=-1) @ R.t()) / 0.3 | |
| return (torch.cosh((v - code[yid] * 1.0).clamp(-4, 4)) - 1).mean() | |
| m_full, s_full = one("full", full) | |
| m_chunk, s_chunk = one("chunked", chunked) | |
| m_fac, s_fac = one("fac", fac) | |
| record("T16", "memory law: candidate <= 1.5x chunked CE", | |
| m_fac <= 1.5 * m_chunk and m_chunk < m_full, | |
| "full %.2fGB/%.3fs | chunked-512 %.2fGB/%.3fs | " | |
| "FAC-K64 %.2fGB/%.3fs (%.0fx less than chunked)" | |
| % (m_full, s_full, m_chunk, s_chunk, m_fac, s_fac, | |
| m_chunk / max(m_fac, 1e-9))) | |
| del E, R, code | |
| torch.cuda.empty_cache() | |
| else: | |
| skip("T16", "memory law vs chunked CE", "no CUDA") | |
| # T17 sub-ULP safety | |
| guard = half_ulp_bf16(3.0) | |
| record("T17", "sub-ULP freeze guard", | |
| guard == 0.0078125 and 4.5e-4 < guard # bf16 step FREEZES | |
| and 4.5e-4 > 2.0 ** (1 - 23) / 2, # fp32 master moves | |
| "half-ULP(bf16, 3.0)=%.7f; 4.5e-4 step frozen in bf16, live in fp32" | |
| % guard) | |
| # T18 anti-collapse smoke (rich-get-richer detector on the read) | |
| if HAVE_VITALS: | |
| ga = torch.Generator().manual_seed(seed_for("aliveness")) | |
| healthy = torch.softmax(torch.randn(4096, 128, generator=ga) * 0.5, -1) | |
| logits = torch.randn(4096, 128, generator=ga) * 0.5 | |
| logits[:, :2] += 8.0 # 2-winner collapse | |
| sick = torch.softmax(logits, -1) | |
| h, s = axis_aliveness(healthy), axis_aliveness(sick) | |
| record("T18", "anti-collapse (rich-get-richer signature)", | |
| (not h["collapsed"]) and s["collapsed"] and s["usage_ppl"] < 6, | |
| "healthy ppl %.0f/128; collapsed ppl %.1f/128 flagged" | |
| % (h["usage_ppl"], s["usage_ppl"])) | |
| else: | |
| skip("T18", "anti-collapse smoke", "geolip_vitals not importable") | |
| # T19 eff-dim readout sanity (the S^15 CV band, zero training) | |
| if HAVE_VITALS: | |
| gs = torch.Generator().manual_seed(seed_for("s15")) | |
| cv = pentachoron_cv(torch.randn(500, 16, generator=gs)) | |
| record("T19", "S^15 CV-band sanity (0.199-0.210 untrained)", | |
| 0.185 <= cv <= 0.225, "CV %.4f" % cv) | |
| else: | |
| skip("T19", "S^15 CV-band sanity", "geolip_vitals not importable") | |
| # T20 seed determinism (crc32 path; no hash() in the seeding path) | |
| ok = seed_for("x") == (zlib.crc32(b"x") & 0x7FFFFFFF) | |
| try: # scan the SEEDING PATH only, CODE only — | |
| # docstrings name the forbidden call, which is not using it | |
| def code_of(fn): | |
| parts = inspect.getsource(fn).split('"""') | |
| return parts[0] + "".join(parts[2::2]) | |
| src_all = (code_of(seed_for) + code_of(build_compartment_map) | |
| + code_of(CompartmentDelta.__init__)) | |
| no_hash = "hash(" not in src_all.replace("crc32", "") | |
| except Exception: | |
| no_hash = True | |
| record("T20", "crc32 seed determinism (never hash())", ok and no_hash, | |
| "seed_for('x')=%d, source clean" % seed_for("x")) | |
| # T21 CE-vs-FAC Hessian conditioning (the sequential-loss smoke) | |
| Vp = 1000 | |
| gz = torch.Generator().manual_seed(seed_for("hessian")) | |
| out = [] | |
| for pmax in (0.5, 0.9, 0.999): | |
| p = torch.full((Vp,), (1 - pmax) / (Vp - 1), dtype=torch.float64) | |
| p[0] = pmax | |
| J = torch.diag(p) - torch.outer(p, p) | |
| ev = torch.linalg.eigvalsh(J) | |
| out.append((pmax, float(ev[0]), float(ev[-1]), | |
| float((J @ torch.ones(Vp, dtype=torch.float64)).abs().max()))) | |
| r64 = torch.randn(64, generator=gz, dtype=torch.float64) * 2 | |
| lam_fac = torch.cosh(r64).min().item() | |
| ce999 = out[2] | |
| record("T21", "CE-vs-FAC Hessian conditioning", | |
| abs(ce999[1]) < 1e-9 and ce999[3] < 1e-9 # exact null direction | |
| and ce999[2] < 1e-2 # spectrum collapsed | |
| and lam_fac >= 1.0, # cosh(r) >= 1 always | |
| "CE@p=.999: lam_min %.1e lam_max %.1e null|J1| %.1e; " | |
| "FAC lam_min %.3f >= 1" % (ce999[1], ce999[2], ce999[3], lam_fac)) | |
| # ------------------------------------------------------------------ table | |
| wall = time.time() - t0 | |
| peak = (torch.cuda.max_memory_allocated() / 2**30) if DEV == "cuda" else 0.0 | |
| print("\nCOMPARTMENT / LOSS FORMULA-SMOKE BATTERY (%s, %.1fs, peak %.2f GB)" | |
| % (DEV, wall, peak)) | |
| print("-" * 100) | |
| npass = nfail = 0 | |
| for tid, name, st, detail in RESULTS: | |
| npass += st == "PASS"; nfail += st == "FAIL" | |
| print("%-5s %-4s %-46s %s" % (tid, st, name[:46], detail[:60])) | |
| print("-" * 100) | |
| print("PASS %d FAIL %d SKIP %d" % (npass, nfail, | |
| len(RESULTS) - npass - nfail)) | |
| return nfail == 0 | |
| if __name__ == "__main__": | |
| sys.exit(0 if run_battery() else 1) | |