"""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)