#!/usr/bin/env python3 """μLattice single-page live demo server. Serves ONE self-contained HTML page (no CDN, no external assets) and a JSON API that runs the REAL trained model (checkpoints/capacity8k.pt, scale from its proto_cfg) through the REAL protocol module: POST /api/store {n_pairs, filler, overwrite_frac, seed} -> one causal forward over the generated stream; returns measured wall time, us/token, lattice cells grown, per-query answers vs gold, updated-vs-original correctness, and the learned 2D addresses of a sample of stored keys (for the live lattice map). POST /api/cost {sizes:[{n_pairs,filler}...]} -> forwards at increasing stream lengths; returns us/token and cells for each: the constant-cost proof, measured live. Everything the page shows is measured on this request. Nothing is canned. """ from __future__ import annotations import json import os import sys import time from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.join(HERE, "..") sys.path.insert(0, os.path.join(ROOT, "src")) import torch from mulattice.protocol import SynthCfg, build_model, force_fast_scan, key_addresses, synth_batch DEVICE = os.environ.get("DEVICE", "cuda" if torch.cuda.is_available() else ("mps" if torch.backends.mps.is_available() else "cpu")) CKPT = os.path.join(ROOT, "checkpoints", "capacity8k.pt") CFG = SynthCfg() print(f"loading {CKPT} on {DEVICE} ...", flush=True) ck = torch.load(CKPT, map_location="cpu", weights_only=False) cfg = SynthCfg() if ck.get("proto_cfg", {}).get("addr_scale"): cfg.addr_scale = float(ck["proto_cfg"]["addr_scale"]) MODEL = build_model(cfg) MODEL.load_state_dict(ck["model"]) MODEL = MODEL.to(DEVICE).eval() force_fast_scan(MODEL) PARAMS = sum(p.numel() for p in MODEL.parameters()) print(f"ready: scale {cfg.addr_scale:.0f}, {PARAMS:,} params", flush=True) @torch.no_grad() def run_store(n_pairs: int, filler: int, ow: float, seed: int) -> dict: from mulattice.protocol import logits_at g = torch.Generator().manual_seed(seed) Q = min(64, n_pairs) b = synth_batch(1, n_pairs, Q, cfg, filler=filler, overwrite_frac=ow, generator=g) t0 = time.time() pred = logits_at(MODEL, b["tokens"].to(DEVICE), b["ans_pos"].to(DEVICE)).argmax(-1).cpu() dt = time.time() - t0 ok = (pred == b["gold"]).squeeze(0) up = b["was_updated"].squeeze(0) # learned addresses of a sample of stored keys (for the live lattice map) n_sample = min(400, n_pairs) kp = b["key_pos"][0, :n_sample].unsqueeze(0) addr = key_addresses(MODEL, b["tokens"][:1].to(DEVICE), kp.to(DEVICE))[0].cpu() sc = cfg.addr_scale pts = [[round(float(a / sc), 4), round(float(c / sc), 4)] for a, c in zip(addr[:, 0], addr[:, 1])] keys_show = b["tokens"][0, b["key_pos"][0, :n_sample]].tolist() golds = {} rows = [] for q in range(Q): k = int(b["tokens"][0, b["ans_pos"][0][q] - 1]) rows.append({ "key": k, "gold": int(b["gold"][0, q]) - cfg.NK, "pred": int(pred[0, q]) - cfg.NK, "ok": bool(ok[q]), "updated": bool(up[q]), }) n_cells = MODEL.blocks[0].mixer.scan.last_n_cells return { "T": b["L"], "wall_s": round(dt, 3), "us_per_token": round(dt / b["L"] * 1e6, 1), "cells": int(n_cells), "acc": round(ok.float().mean().item(), 4), "acc_updated": round(ok[up].float().mean().item(), 4) if bool(up.any()) else None, "acc_untouched": round(ok[~up].float().mean().item(), 4) if bool((~up).any()) else None, "n_pairs": n_pairs, "n_upd": b["n_upd"], "Q": Q, "rows": rows, "addr_points": pts, "addr_keys": keys_show, "scale": sc, "params": PARAMS, "device": DEVICE, "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), } @torch.no_grad() def run_cost(sizes: list) -> list: from mulattice.protocol import logits_at out = [] for s in sizes: n_pairs, filler = int(s.get("n_pairs", 1024)), int(s.get("filler", 0)) g = torch.Generator().manual_seed(int(s.get("seed", 7))) Q = 16 b = synth_batch(1, n_pairs, Q, cfg, filler=filler, generator=g) t0 = time.time() logits_at(MODEL, b["tokens"].to(DEVICE), b["ans_pos"].to(DEVICE)) dt = time.time() - t0 out.append({ "n_pairs": n_pairs, "filler": filler, "T": b["L"], "wall_s": round(dt, 3), "us_per_token": round(dt / b["L"] * 1e6, 1), "cells": int(MODEL.blocks[0].mixer.scan.last_n_cells), "flops_attention_equiv": b["L"] * b["L"], # O(T^2) reference work }) return out @torch.no_grad() def run_custom(doc_text: str, pairs: list, seed: int) -> dict: """Stream the USER's own text with the USER's own named key->value pairs, some changed mid-stream, then query. Names map deterministically into the model's token space (shown in the response); the model only ever sees ids -- exactly the MQAR regime it was trained on.""" from mulattice.protocol import logits_at NK, NV = cfg.NK, cfg.NV g = torch.Generator().manual_seed(seed) def h(s, used, span, off): x = abs(hash((seed, s))) % span while (off + x) in used: x = (x + 1) % span used.add(off + x) return off + x used = set() kmap, vmap = {}, {} for p in pairs: kmap[p["name"]] = h("k" + p["name"], used, NK, 0) # filler ids: hash the user's own words into the key space (excl. used keys) words = [w for w in doc_text.split() if w][: 20000] or ["..."] fid = {} for w in set(words): fid[w] = h("f" + w, used, NK, 0) store_rows, upd_rows = [], [] for p in pairs: vid = h("v" + p["value"], used, NV, NK) vmap[p["value"]] = vid row = dict(p); row["kid"] = kmap[p["name"]]; row["vid"] = vid if p.get("new_value"): nid = h("v2" + p["new_value"], used, NV, NK) vmap[p["new_value"]] = nid row["nid"] = nid upd_rows.append(row) store_rows.append(row) # deterministic filler stream from the user's own words fill = [fid[w] for w in words] def take(n): out = [] while len(out) < n: i = int(torch.randint(0, len(fill), (1,), generator=g)) out.extend(fill[i:i + n - len(out)]) return out[:n] F = 6 # filler tokens between pairs block = F + 2 n_upd = len(upd_rows) Q = min(64, len(store_rows)) qrows = store_rows[:Q] L = (len(store_rows) + n_upd) * block + 2 * Q toks = torch.zeros(1, L, dtype=torch.long) gold = torch.zeros(1, Q, dtype=torch.long) latest = {r["kid"]: r["vid"] for r in store_rows} for i, r in enumerate(store_rows): s = i * block toks[0, s:s + F] = torch.tensor(take(F)) toks[0, s + F], toks[0, s + F + 1] = r["kid"], r["vid"] for j, r in enumerate(upd_rows): s = (len(store_rows) + j) * block toks[0, s:s + F] = torch.tensor(take(F)) toks[0, s + F], toks[0, s + F + 1] = r["kid"], r["nid"] latest[r["kid"]] = r["nid"] S2 = (len(store_rows) + n_upd) * block ans = [] for q, r in enumerate(qrows): toks[0, S2 + 2 * q], toks[0, S2 + 2 * q + 1] = r["kid"], cfg.SEP gold[0, q] = latest[r["kid"]] ans.append(S2 + 2 * q + 1) ans_pos = torch.tensor(ans).unsqueeze(0) t0 = time.time() pred = logits_at(MODEL, toks.to(DEVICE), ans_pos.to(DEVICE)).argmax(-1).cpu() dt = time.time() - t0 ok = (pred == gold).squeeze(0) inv_v = {v: k for k, v in vmap.items()} rows = [] for q, r in enumerate(qrows): rows.append({ "name": r["name"], "asked": inv_v.get(int(pred[0, q]), "id" + str(int(pred[0, q]) - NK)), "truth": r["new_value"] if r.get("new_value") else r["value"], "ok": bool(ok[q]), "updated": bool(r.get("new_value")), "old": r["value"] if r.get("new_value") else None, }) return { "T": L, "wall_s": round(dt, 3), "us_per_token": round(dt / L * 1e6, 1), "cells": int(MODEL.blocks[0].mixer.scan.last_n_cells), "acc": round(ok.float().mean().item(), 4), "n_pairs": len(store_rows), "n_upd": n_upd, "Q": Q, "rows": rows, "keymap": {r["name"]: r["kid"] for r in store_rows}, "filler_words": len(set(words)), "device": DEVICE, "scale": float(MODEL.blocks[0].mixer.cfg.addr_scale), "params": sum(p.numel() for p in MODEL.parameters()), "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), } @torch.no_grad() def run_cellview(name: str, value: str, new_value: str, seed: int) -> dict: """Cell's-eye view of one fact: where the key lives (learned address + 3x3 cells), and the delta rule replacing the stored value on overwrite.""" from mulattice.protocol import logits_at from mulattice.kernel import phi_2d g = torch.Generator().manual_seed(seed) used = set() def h(s, span, off): x = abs(hash((seed, s))) % span while (off + x) in used: x = (x + 1) % span used.add(off + x) return off + x kid = h("k" + name, cfg.NK, 0) vid = h("v" + value, cfg.NV, cfg.NK) nid = h("v2" + new_value, cfg.NV, cfg.NK) if new_value else vid def stream(with_upd): s = [kid, vid] if with_upd: s += [kid, nid] s += [kid, cfg.SEP] return torch.tensor([s]) answers = {} for tag, with_upd in (("before", False), ("after", True)): if tag == "after" and not new_value: answers[tag] = None continue t = stream(with_upd) pred = logits_at(MODEL, t.to(DEVICE), torch.tensor([[t.shape[1] - 1]]).to(DEVICE)).argmax(-1).cpu() answers[tag] = int(pred[0, 0]) - cfg.NK mx = MODEL.blocks[0].mixer x = MODEL.emb(torch.tensor([[kid]]).to(DEVICE)) xc = mx.conv(x) addr = (mx.cfg.addr_scale * torch.tanh(mx.Wa(xc)))[0, 0].cpu() w, z = phi_2d(addr.unsqueeze(0)) cells = [[{"z": [int(z[0, i, j, 0]), int(z[0, i, j, 1])], "phi": round(float(w[0, i, j]), 4)} for j in range(3)] for i in range(3)] return {"kid": kid, "vid": vid, "nid": nid, "addr": [round(float(addr[0]), 2), round(float(addr[1]), 2)], "scale": mx.cfg.addr_scale, "cells": cells, "answers": answers, "value": value, "new_value": new_value, "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} PAGE = r""" μLattice: Breaking the Quadratic Memory Barrier in AI
Tensorbend Research · October 2026 · live proof · real weights

Breaking the Quadratic Memory Barrier in AI

A sequence mixer with dynamically allocated memory. Each token reads and writes exactly 9 cells on a 2D lattice; memory grows with content while per-token cost stays constant. Measured: recall of 1.0000 at 8,192 pairs; 0.9980 on overwrite queries; 1.0000 on real-text retrieval at 259K tokens. This page runs three trained checkpoints. Every number is a live forward pass.

Tensorbend Research · Eric E. · October 2026 · Models loaded · content-addressed associative mixer (not a language model)
01

Stream your own document

Paste any text. Define facts as name → value pairs. Mark any to be changed mid-stream. The model reads the stream, stores the facts, and returns the latest value for each.

02

The store at scale ↗

The same mechanism at scale: store up to 8,192 pairs in one 133,120-token causal pass, overwrite half of them, then query. Measured recall: 1.0000.

2048
0
0%
Left: learned addresses of 400 stored keys from this run (grid scale shown). Right: 64 sampled queries; gold rows marked changed mid-stream were re-written later in the stream — the store must answer with the new value.
03

Constant cost while memory grows ↗

Each token touches exactly nine memory cells, so per-token work cannot grow with context. Measure it: the curve below runs the model at increasing stream lengths, live.

Blue: measured μs per token — flat is the guarantee. Green: lattice cells, growing with stored content. Dashed red: attention’s O(T²) work, analytic reference.
04

Inside one cell ↗

Where does a value physically live, and where does the new value go on overwrite? Pick one fact from section 01 (or any name/value). The model’s learned address for that key, the 3×3 cells it touches, and the before/after answers — same cells, value replaced in place by the delta rule.

05

Notes

This is a content-addressed associative mixer, not a language model. It stores key→value associations and retrieves them at constant per-token cost. Perfect scores are on the synthetic MQAR distribution (see PAPER.md).

Current demo is 2D (9-cell neighborhood per token). 3D (27 cells) is supported by the kernel math without architectural change — it reduces key density by 1000× and raises the estimated capacity from ~8K to ~495K keys (details).

""" class H(BaseHTTPRequestHandler): def _json(self, obj, code=200): body = json.dumps(obj).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): # Strip query parameters (HF Spaces appends ?__theme=... etc) path = self.path.split("?")[0] if path in ("/", "/index.html"): body = PAGE.encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) elif path == "/favicon.ico": self.send_response(204); self.end_headers() elif path == "/api/status": self._json({"status": "ok", "models": len(MODELS), "device": DEVICE, "uptime_s": round(time.time() - _T0, 0)}) else: self.send_error(404) def do_POST(self): n = int(self.headers.get("Content-Length", 0)) req = json.loads(self.rfile.read(n) or b"{}") try: if self.path == "/api/store": np_ = min(int(req.get("n_pairs", 2048)), 4096) # cap for CPU r = run_store(np_, int(req.get("filler", 14)), float(req.get("overwrite_frac", 0.5)), int(req.get("seed", 0))) self._json(r) elif self.path == "/api/cost": self._json(run_cost(req.get("sizes", []))) elif self.path == "/api/cellview": self._json(run_cellview(str(req.get("name", "key")), str(req.get("value", "v1")), str(req.get("new_value", "")), int(req.get("seed", 0)))) elif self.path == "/api/custom": text = str(req.get("text", ""))[:5000] # cap at 5K chars pairs = req.get("pairs", [])[:20] # max 20 facts for p in pairs: if isinstance(p, dict): p["name"] = str(p.get("name", ""))[:100] p["value"] = str(p.get("value", ""))[:200] p["new_value"] = str(p.get("new_value", ""))[:200] r = run_custom(text, pairs, int(req.get("seed", 0))) self._json(r) else: self.send_error(404) except Exception as e: # surface errors to the page log self._json({"error": str(e)}, 500) def log_message(self, fmt, *args): print(f"[{time.strftime('%H:%M:%S')}] {fmt % args}", flush=True) if __name__ == "__main__": port = int(os.environ.get("PORT", "7860")) # ThreadingHTTPServer: the single-threaded server hangs on browsers' # speculative idle connections (page "loads once, then never again"). srv = ThreadingHTTPServer(("0.0.0.0", port), H) srv.daemon_threads = True _T0 = time.time() print(f"demo on http://0.0.0.0:{port}/", flush=True) srv.serve_forever()