loss-manifest / code /compartment_smoke.py
AbstractPhil's picture
loss-manifest companion: article + 155-entry rated registry + sidecar + loss library + gates/battery + campaign beds + raw run ledgers
4ef8b0b verified
Raw
History Blame Contribute Delete
30.7 kB
"""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)