"""The SUBLEQ host: its family, serialization, device, constructor, self-reproduction and evaluators.""" from __future__ import annotations import argparse import hashlib import json import os import struct import sys import time from typing import Dict, List, Tuple, Optional import numpy as np import torch from safetensors import safe_open from safetensors.torch import save_file from machines import Net, compile_net _DT = {torch.float32:"F32", torch.float64:"F64", torch.int64:"I64", torch.int32:"I32", torch.int16:"I16", torch.int8:"I8", torch.uint8:"U8", torch.bool:"BOOL"} def canonical_bytes(tensors, metadata_pairs): """Canonical safetensors bytes from tensors and metadata pairs in canonical order.""" header = {} if metadata_pairs is not None: header["__metadata__"] = {k: v for k, v in metadata_pairs} blobs, off = [], 0 for name, t in tensors: t = t.contiguous() b = t.numpy().tobytes() header[name] = {"dtype": _DT[t.dtype], "shape": list(t.shape), "data_offsets": [off, off + len(b)]} blobs.append(b); off += len(b) s = json.dumps(header, separators=(",", ":"), ensure_ascii=False).encode("utf-8") pad = (-(8 + len(s))) % 8 s = s + b" " * pad return struct.pack(" Dict[str, int]: S = 1 << A return {"c_wr": S - 7, "c_eot": S - 6, "c_rw": S - 5, "c_rd": S - 4, "r_in": S - 3, "r_out": S - 2, "halt_pc": S - 1} def layout(A: int) -> Dict[str, List[int]]: return {"pc": [0, A], "halt": [A, 1], "mem": [A + 1, 1 << A, A]} def state_names(A: int) -> List[str]: S = 1 << A return ([f"pc{i}" for i in range(A)] + ["halt"] + [f"m{j}b{t}" for j in range(S) for t in range(A)]) def build(A: int, fold: bool = False) -> Tuple[Net, List[str], List[str]]: """The step netlist of H_A, units in serialization order.""" S = 1 << A net = Net() g = net.g pc = [f"pc{i}" for i in range(A)] halt = "halt" mu = lambda j, t: f"m{j}b{t}" def lits(bits: List[str], value: int): """The literals asserting that the MSB-first `bits` spell `value`.""" return [(bits[i], 1 if (value >> (A - 1 - i)) & 1 else -1) for i in range(A)] # B1 decode lines of the counter for j in range(S): g(f"P{j}", lits(pc, j), -bin(j).count("1")) # B2 the counter held by the halt bit for i in range(A): g(f"HP{i}", [(pc[i], 1), (halt, 1)], -2) # B3 the fetch conjunctions: operand o, bit t, counter value j for o in range(3): for t in range(A): for j in range(S): if fold: g(f"FA{o}_{t}_{j}", [(mu((j + o) % S, t), 1)] + lits(pc, j), -1 - bin(j).count("1")) else: g(f"FA{o}_{t}_{j}", [(mu((j + o) % S, t), 1), (f"P{j}", 1)], -2) # B4 the bits of p + 3, read off the decode lines for i in range(A): g(f"Q{i}", [(f"P{j}", 1) for j in range(S) if (((j + 3) % S) >> (A - 1 - i)) & 1], -1) # B5 the three operand words A, B, C for o in range(3): for t in range(A): g(f"F{o}_{t}", [(f"FA{o}_{t}_{j}", 1) for j in range(S)], -1) Aop = [f"F0_{t}" for t in range(A)] Bop = [f"F1_{t}" for t in range(A)] Cop = [f"F2_{t}" for t in range(A)] # B6 decode lines of A and B if not fold: for j in range(S): g(f"PA{j}", lits(Aop, j), -bin(j).count("1")) for j in range(S): g(f"PB{j}", lits(Bop, j), -bin(j).count("1")) # B7 C is the halt address g("AC", [(c, 1) for c in Cop], -A) # B8 write enables for j in range(S): if fold: g(f"W{j}", [(halt, -1)] + lits(Bop, j), -bin(j).count("1")) else: g(f"W{j}", [(halt, -1), (f"PB{j}", 1)], -1) # B9 the read conjunctions of M[A] and M[B] for t in range(A): for j in range(S): if fold: g(f"RX{t}_{j}", [(mu(j, t), 1)] + lits(Aop, j), -1 - bin(j).count("1")) else: g(f"RX{t}_{j}", [(mu(j, t), 1), (f"PA{j}", 1)], -2) for t in range(A): for j in range(S): if fold: g(f"RY{t}_{j}", [(mu(j, t), 1)] + lits(Bop, j), -1 - bin(j).count("1")) else: g(f"RY{t}_{j}", [(mu(j, t), 1), (f"PB{j}", 1)], -2) # B10 the cells kept for j in range(S): for t in range(A): g(f"KP{j}_{t}", [(mu(j, t), 1), (f"W{j}", -1)], -1) # B11 X = M[A], Y = M[B] for t in range(A): g(f"X{t}", [(f"RX{t}_{j}", 1) for j in range(S)], -1) for t in range(A): g(f"Y{t}", [(f"RY{t}_{j}", 1) for j in range(S)], -1) xv = lambda i: f"X{A - 1 - i}" # value bit i, least significant first yv = lambda i: f"Y{A - 1 - i}" # B12 propagate signals of Y + not X for i in range(A - 1): g(f"PR{i}", [(xv(i), -1), (yv(i), 1)], 0) # B13 carry terms into bit k: the incoming carry through all lower # positions, and a generate at j propagated through j+1..k-1 for k in range(1, A): g(f"T{k}_n", [(f"PR{i}", 1) for i in range(k)], -k) for j in range(k): g(f"T{k}_{j}", [(yv(j), 1), (xv(j), -1)] + [(f"PR{i}", 1) for i in range(j + 1, k)], -(1 + (k - 1 - j))) # B14 carries for k in range(1, A): g(f"C{k}", [(f"T{k}_n", 1)] + [(f"T{k}_{j}", 1) for j in range(k)], -1) # B15 X = Y, bit by bit for i in range(A): g(f"AX{i}", [(xv(i), 1), (yv(i), 1)], -2) for i in range(A): g(f"NX{i}", [(xv(i), -1), (yv(i), -1)], 0) for i in range(A): g(f"XN{i}", [(f"AX{i}", 1), (f"NX{i}", 1)], -1) # B16 g("EQ", [(f"XN{i}", 1) for i in range(A)], -A) # B17 the sum bits: bit 0 is X xor Y, bit k the parity of y_k + not x_k + c_k g("O0", [(yv(0), 1), (xv(0), 1)], -1) g("N0", [(yv(0), -1), (xv(0), -1)], 1) g("R0", [("O0", 1), ("N0", 1)], -2) for k in range(1, A): for m, b in ((1, 0), (2, -1), (3, -2)): g(f"H{m}_{k}", [(yv(k), 1), (xv(k), -1), (f"C{k}", 1)], b) for k in range(1, A): g(f"R{k}", [(f"H1_{k}", 1), (f"H2_{k}", -1), (f"H3_{k}", 1)], -1) top = A - 1 g("LEQ", [(f"H1_{top}", 1), (f"H2_{top}", -1), (f"H3_{top}", 1), ("EQ", 1)], -1) # B18 the next counter for i in range(A): g(f"TA{i}", [(halt, -1), ("LEQ", 1), (Cop[i], 1)], -2) for i in range(A): g(f"TB{i}", [(halt, -1), ("LEQ", -1), (f"Q{i}", 1)], -1) for i in range(A): g(f"NP{i}", [(f"HP{i}", 1), (f"TA{i}", 1), (f"TB{i}", 1)], -1) # B19 the halt latch: the next counter is 2^A - 1 g("HA", [(halt, -1), ("LEQ", 1), ("AC", 1)], -2) g("HB", [(halt, -1), ("LEQ", -1), (f"P{S - 4}", 1)], -1) g("NH", [(halt, 1), ("HA", 1), ("HB", 1)], -1) # B20 the write: bit t of the word is value bit A-1-t of the difference for j in range(S): for t in range(A): g(f"WR{j}_{t}", [(f"R{A - 1 - t}", 1), (f"W{j}", 1)], -2) for j in range(S): for t in range(A): g(f"NM{j}_{t}", [(f"KP{j}_{t}", 1), (f"WR{j}_{t}", 1)], -1) outputs = ([f"NP{i}" for i in range(A)] + ["NH"] + [f"NM{j}_{t}" for j in range(S) for t in range(A)]) return net, state_names(A), outputs # --------------------------------------------------------------------------- # serialization # --------------------------------------------------------------------------- _DTORD_FAMILY = {"F64": 0, "F32": 1, "F16": 2, "I64": 3, "I32": 4, "I16": 5, "I8": 6} _NP = {"I8": " str: return "I16" if maxabs <= 32767 else "I32" def encode(A: int, bias, fanin, pred, nxt, machine: str, lay, io) -> bytes: """The serialization of an ordered netlist from its four vectors.""" n, v = len(nxt), len(bias) t = {"net.bias": ("I8", np.asarray(bias, dtype=np.int64)), "net.fanin": (_itype(int(np.max(fanin))), np.asarray(fanin, dtype=np.int64)), "net.next": (_itype(int(np.max(np.abs(nxt)))), np.asarray(nxt, dtype=np.int64)), "net.pred": (_itype(int(np.max(np.abs(pred)))), np.asarray(pred, dtype=np.int64))} assert int(t["net.bias"][1].min()) >= -128 and int(t["net.bias"][1].max()) <= 127 for k, val in (("addr_bits", A), ("data_bits", A), ("memory_words", 1 << A), ("pc_width", A), ("state_bits", n), ("units", v), ("version", VERSION)): t[f"manifest.{k}"] = ("I32", np.asarray([val], dtype=np.int64)) order = sorted(t, key=lambda k: (_DTORD_FAMILY[t[k][0]], k)) header = {"__metadata__": { "machine": machine, "weight_quantization": "ternary", "signal_registry": json.dumps(lay, separators=(",", ":"), sort_keys=True), "io": json.dumps(io, separators=(",", ":"), sort_keys=True)}} blobs, off = [], 0 for k in order: dt, arr = t[k] b = arr.astype(_NP[dt]).tobytes() header[k] = {"dtype": dt, "shape": [int(arr.size)], "data_offsets": [off, off + len(b)]} blobs.append(b) off += len(b) s = json.dumps(header, separators=(",", ":"), ensure_ascii=False).encode("utf-8") s += b" " * ((-(8 + len(s))) % 8) return struct.pack(" bytes: """The serialization of a Net in the order its units were built.""" slot = {s: i for i, s in enumerate(inputs)} for gname in net.gates: slot[gname] = len(slot) bias, fanin, pred = [], [], [] for gname, (ins, b) in net.gates.items(): bias.append(b) fanin.append(len(ins)) for s, w in ins: assert w in (-1, 1) pred.append(w * (slot[s] + 1)) nxt = [slot[s] for s in outputs] return encode(A, bias, fanin, pred, nxt, machine or f"subleq{A}", lay or layout(A), io or io_cells(A)) def sigma(A: int) -> bytes: """The serialization of H_A, written block by block with numpy.""" S = 1 << A n = A + 1 + A * S bias, fanin, pred = [], [], [] base = {} count = [n] def block(name: str, size: int): base[name] = count[0] count[0] += size J = np.arange(S, dtype=np.int64) popc = np.array([bin(j).count("1") for j in range(S)], dtype=np.int64) bits = ((J[:, None] >> (A - 1 - np.arange(A))[None, :]) & 1) # [j, i] MSB first lsign = np.where(bits == 1, 1, -1) MU = lambda j, t: A + 1 + A * j + t # state index HALT, PC = A, np.arange(A, dtype=np.int64) def emit(b, f, p): bias.append(np.asarray(b, dtype=np.int64).ravel()) fanin.append(np.asarray(f, dtype=np.int64).ravel()) pred.append(np.asarray(p, dtype=np.int64).ravel()) # B1 block("P", S) emit(-popc, np.full(S, A), (lsign * (PC[None, :] + 1)).ravel()) # B2 block("HP", A) emit(np.full(A, -2), np.full(A, 2), np.stack([PC + 1, np.full(A, HALT + 1)], 1)) # B3 block("FA", 3 * A * S) o, t, j = np.meshgrid(np.arange(3), np.arange(A), J, indexing="ij") emit(np.full(3 * A * S, -2), np.full(3 * A * S, 2), np.stack([MU((j + o) % S, t) + 1, base["P"] + j + 1], -1)) # B4 block("Q", A) qf, qp = [], [] for i in range(A): sel = J[((((J + 3) % S) >> (A - 1 - i)) & 1) == 1] qf.append(len(sel)) qp.append(base["P"] + sel + 1) emit(np.full(A, -1), qf, np.concatenate(qp)) # B5 block("F", 3 * A) o, t, j = np.meshgrid(np.arange(3), np.arange(A), J, indexing="ij") emit(np.full(3 * A, -1), np.full(3 * A, S), base["FA"] + (o * A + t) * S + j + 1) F = lambda o, t: base["F"] + o * A + t # B6 block("PA", S) emit(-popc, np.full(S, A), (lsign * (F(0, np.arange(A))[None, :] + 1)).ravel()) block("PB", S) emit(-popc, np.full(S, A), (lsign * (F(1, np.arange(A))[None, :] + 1)).ravel()) # B7 block("AC", 1) emit([-A], [A], F(2, np.arange(A)) + 1) # B8 block("W", S) emit(np.full(S, -1), np.full(S, 2), np.stack([np.full(S, -(HALT + 1)), base["PB"] + J + 1], -1)) # B9 t, j = np.meshgrid(np.arange(A), J, indexing="ij") block("RX", A * S) emit(np.full(A * S, -2), np.full(A * S, 2), np.stack([MU(j, t) + 1, base["PA"] + j + 1], -1)) block("RY", A * S) emit(np.full(A * S, -2), np.full(A * S, 2), np.stack([MU(j, t) + 1, base["PB"] + j + 1], -1)) # B10 j, t = np.meshgrid(J, np.arange(A), indexing="ij") block("KP", S * A) emit(np.full(S * A, -1), np.full(S * A, 2), np.stack([MU(j, t) + 1, -(base["W"] + j + 1)], -1)) # B11 t, j = np.meshgrid(np.arange(A), J, indexing="ij") block("X", A) emit(np.full(A, -1), np.full(A, S), base["RX"] + t * S + j + 1) block("Y", A) emit(np.full(A, -1), np.full(A, S), base["RY"] + t * S + j + 1) X = lambda i: base["X"] + (A - 1 - i) Y = lambda i: base["Y"] + (A - 1 - i) # B12 block("PR", A - 1) ii = np.arange(A - 1) emit(np.zeros(A - 1), np.full(A - 1, 2), np.stack([-(X(ii) + 1), Y(ii) + 1], -1)) PR = lambda i: base["PR"] + i # B13 tb, tf, tp = [], [], [] tidx = {} block("T", sum(k + 1 for k in range(1, A))) cur = base["T"] for k in range(1, A): tidx[(k, "n")] = cur; cur += 1 tb.append(-k); tf.append(k); tp += [PR(i) + 1 for i in range(k)] for jj in range(k): tidx[(k, jj)] = cur; cur += 1 tb.append(-(1 + (k - 1 - jj))); tf.append(2 + (k - 1 - jj)) tp += [Y(jj) + 1, -(X(jj) + 1)] + [PR(i) + 1 for i in range(jj + 1, k)] emit(tb, tf, tp) # B14 block("C", A - 1) cp = [] for k in range(1, A): cp += [tidx[(k, "n")] + 1] + [tidx[(k, jj)] + 1 for jj in range(k)] emit(np.full(A - 1, -1), [k + 1 for k in range(1, A)], cp) C = lambda k: base["C"] + k - 1 # B15 ii = np.arange(A) block("AX", A) emit(np.full(A, -2), np.full(A, 2), np.stack([X(ii) + 1, Y(ii) + 1], -1)) block("NX", A) emit(np.zeros(A), np.full(A, 2), np.stack([-(X(ii) + 1), -(Y(ii) + 1)], -1)) block("XN", A) emit(np.full(A, -1), np.full(A, 2), np.stack([base["AX"] + ii + 1, base["NX"] + ii + 1], -1)) # B16 block("EQ", 1) emit([-A], [A], base["XN"] + ii + 1) # B17 block("R0", 3) emit([-1, 1, -2], [2, 2, 2], [Y(0) + 1, X(0) + 1, -(Y(0) + 1), -(X(0) + 1), base["R0"] + 1, base["R0"] + 2]) block("H", 3 * (A - 1)) hb, hp = [], [] for k in range(1, A): for b in (0, -1, -2): hb.append(b) hp += [Y(k) + 1, -(X(k) + 1), C(k) + 1] emit(hb, np.full(3 * (A - 1), 3), hp) H = lambda m, k: base["H"] + 3 * (k - 1) + (m - 1) block("R", A - 1) rp = [] for k in range(1, A): rp += [H(1, k) + 1, -(H(2, k) + 1), H(3, k) + 1] emit(np.full(A - 1, -1), np.full(A - 1, 3), rp) Rv = lambda k: base["R0"] + 2 if k == 0 else base["R"] + k - 1 block("LEQ", 1) top = A - 1 emit([-1], [4], [H(1, top) + 1, -(H(2, top) + 1), H(3, top) + 1, base["EQ"] + 1]) LEQ = base["LEQ"] # B18 ii = np.arange(A) block("TA", A) emit(np.full(A, -2), np.full(A, 3), np.stack([np.full(A, -(HALT + 1)), np.full(A, LEQ + 1), F(2, ii) + 1], -1)) block("TB", A) emit(np.full(A, -1), np.full(A, 3), np.stack([np.full(A, -(HALT + 1)), np.full(A, -(LEQ + 1)), base["Q"] + ii + 1], -1)) block("NP", A) emit(np.full(A, -1), np.full(A, 3), np.stack([base["HP"] + ii + 1, base["TA"] + ii + 1, base["TB"] + ii + 1], -1)) # B19 block("HA", 3) emit([-2, -1, -1], [3, 3, 3], [-(HALT + 1), LEQ + 1, base["AC"] + 1, -(HALT + 1), -(LEQ + 1), base["P"] + (S - 4) + 1, HALT + 1, base["HA"] + 1, base["HA"] + 2]) NH = base["HA"] + 2 # B20 j, t = np.meshgrid(J, np.arange(A), indexing="ij") rdata = np.array([Rv(A - 1 - tt) for tt in range(A)], dtype=np.int64) block("WR", S * A) emit(np.full(S * A, -2), np.full(S * A, 2), np.stack([rdata[t] + 1, base["W"] + j + 1], -1)) block("NM", S * A) emit(np.full(S * A, -1), np.full(S * A, 2), np.stack([base["KP"] + j * A + t + 1, base["WR"] + j * A + t + 1], -1)) nxt = np.concatenate([base["NP"] + np.arange(A), [NH], base["NM"] + np.arange(S * A)]) b = np.concatenate(bias); f = np.concatenate(fanin); p = np.concatenate(pred) assert len(b) == count[0] - n and int(f.sum()) == len(p) return encode(A, b, f, p, nxt, f"subleq{A}", layout(A), io_cells(A)) # constant signals, which Net folds into biases and which therefore appear only # as a next-state assignment CONST0, CONST1 = -1, -2 _DTORD = {"F64": 0, "F32": 1, "F16": 2, "I64": 3, "I32": 4, "I16": 5, "I8": 6, "U8": 7, "BOOL": 8} _TAG = {torch.float32: "F32", torch.float64: "F64", torch.int64: "I64", torch.int32: "I32", torch.int16: "I16", torch.int8: "I8"} def depths(net: Net) -> Dict[str, int]: """Depth of every unit: one more than the greatest depth of a predecessor.""" d: Dict[str, int] = {} order = topological(net) for g in order: ins, _ = net.gates[g] d[g] = 1 + max((d[s] for s, _ in ins if s in net.gates), default=0) return d def topological(net: Net) -> List[str]: """The units in some order consistent with the predecessor relation.""" gates = net.gates indeg = {g: 0 for g in gates} cons: Dict[str, List[str]] = {} for g, (ins, _) in gates.items(): for s, _w in ins: if s in gates: indeg[g] += 1 cons.setdefault(s, []).append(g) order = [g for g, k in indeg.items() if k == 0] i = 0 while i < len(order): for c in cons.get(order[i], []): if c in gates: indeg[c] -= 1 if indeg[c] == 0: order.append(c) i += 1 if len(order) != len(gates): raise ValueError("the predecessor relation has a cycle") return order def unit_order(net: Net, inputs: List[str]) -> Tuple[List[str], Dict[str, int]]: """The canonical order of the units, and the signal index of every signal.""" d = depths(net) by_level: Dict[int, List[str]] = {} for g in net.gates: by_level.setdefault(d[g], []).append(g) slot = {s: i for i, s in enumerate(inputs)} order: List[str] = [] for lv in sorted(by_level): keyed = [] for g in by_level[lv]: ins, b = net.gates[g] keyed.append(((b, tuple(sorted((slot[s], w) for s, w in ins))), g)) keyed.sort(key=lambda kv: kv[0]) seen: Dict[tuple, int] = {} for key, g in keyed: if key not in seen: seen[key] = len(inputs) + len(order) order.append(g) slot[g] = seen[key] return order, slot def tensors_of_net(net: Net, inputs: List[str], outputs: List[str]): """The five tensors of the serialization, in canonical unit order.""" order, slot = unit_order(net, inputs) bias, fanin, pred = [], [], [] for g in order: ins, b = net.gates[g] bias.append(b) fanin.append(len(ins)) for k, w in sorted((slot[s], w) for s, w in ins): assert w in (-1, 1), f"weight {w} is not an edge" pred.append(w * (k + 1)) nxt = [] for s in outputs: if s == "#0": nxt.append(CONST0) elif s == "#1": nxt.append(CONST1) else: nxt.append(slot[s]) i8, i16 = torch.int8, torch.int16 return { "net.bias": torch.tensor(bias, dtype=i8), "net.fanin": torch.tensor(fanin, dtype=i16), "net.next": torch.tensor(nxt, dtype=i16), "net.pred": torch.tensor(pred, dtype=i16), } def sigma_of_net(net: Net, inputs: List[str], outputs: List[str], machine: str, layout: Dict, io: Dict) -> bytes: """The serialization of an ordered clocked netlist.""" slot = {s: i for i, s in enumerate(inputs)} for g in net.gates: slot[g] = len(slot) bias, fanin, pred = [], [], [] for g, (ins, b) in net.gates.items(): bias.append(b) fanin.append(len(ins)) for s, w in ins: assert w in (-1, 1), f"weight {w} is not an edge" assert slot[s] < slot[g], "a predecessor is built after its unit" pred.append(w * (slot[s] + 1)) nxt = [CONST0 if s == "#0" else CONST1 if s == "#1" else slot[s] for s in outputs] return encode(layout["pc"][1], bias, fanin, pred, nxt, machine, layout, io) def parse(data: bytes) -> Tuple[Dict[str, torch.Tensor], Dict[str, str]]: """Header and tensors of a safetensors encoding, without a library call.""" n = struct.unpack(" Tuple[Net, List[str], List[str], Dict[str, str]]: """The netlist a serialization denotes, with canonical signal names.""" t, meta = parse(data) n = int(t["manifest.state_bits"][0]) v = int(t["manifest.units"][0]) bias = t["net.bias"].tolist() fanin = t["net.fanin"].tolist() pred = t["net.pred"].tolist() nxt = t["net.next"].tolist() if len(bias) != v or len(fanin) != v: raise ValueError("unit count disagrees with the manifest") if len(pred) != sum(fanin): raise ValueError("predecessor count disagrees with the fan-in vector") if len(nxt) != n: raise ValueError("next-state vector disagrees with the manifest") inputs = [f"s{i}" for i in range(n)] name = lambda k: inputs[k] if k < n else f"u{k - n}" net = Net() p = 0 for j in range(v): ins = [] for _ in range(fanin[j]): e = pred[p] if e == 0: raise ValueError("predecessor entry carries no weight") k, w = abs(e) - 1, (1 if e > 0 else -1) if k >= n + v: raise ValueError("predecessor index out of range") if k >= n + j: raise ValueError("predecessor is not earlier in the order") ins.append((name(k), w)) p += 1 net.g(f"u{j}", ins, bias[j]) outputs = [] for k in nxt: if k == CONST0: outputs.append("#0") elif k == CONST1: outputs.append("#1") elif 0 <= k < n + v: outputs.append(name(k)) else: raise ValueError("next-state index out of range") return net, inputs, outputs, meta HALT_PC = 0xFF def subleq_state_names() -> List[str]: names = [f"pc{i}" for i in range(8)] + ["halt"] for j in range(256): names += [f"m{j}b{i}" for i in range(8)] return names def build_subleq_step_net() -> Tuple[Net, List[str], List[str]]: """The step netlist of the host: the member H_8 of the constant-depth family.""" return build(8) REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) HOST_PATH = os.path.join(REPO, "variants", "neural_subleq8io_netlist.safetensors") # device cells C_WR, C_EOT, C_RW, C_RD, R_IN, R_OUT = 0xF9, 0xFA, 0xFB, 0xFC, 0xFD, 0xFE # program variables Z, ONE, T1, T2, NEG1, EOK = 0xF0, 0xF1, 0xF2, 0xF3, 0xF4, 0xF5 # ============================================================================= # Recipe language (grammar and decoder), reused from constructor8 # ============================================================================= def describe(data: bytes) -> bytes: """Compile bytes into a recipe.""" tape = bytearray() i, n = 0, len(data) while i < n: j = i while j < n and data[j] == data[i] and j - i < 127: j += 1 if j - i >= 4: tape.append(256 - (j - i)) tape.append((256 - data[i]) % 256) i = j continue k = i while k < n and k - i < 127: m = k while m < n and data[m] == data[k] and m - k < 4: m += 1 if m - k >= 4: break k += 1 k = max(k, i + 1) tape.append(k - i) tape.extend((256 - x) % 256 for x in data[i:k]) i = k tape.append(0) return bytes(tape) def describe_literal(data: bytes) -> bytes: """The all-literal encoding used in the proof of the length bound.""" tape = bytearray() for i in range(0, len(data), 127): block = data[i:i + 127] tape.append(len(block)) tape.extend((256 - x) % 256 for x in block) tape.append(0) return bytes(tape) def decode(tape: bytes) -> bytes: """delta: the decoding map on well-formed recipes.""" out = bytearray() i = 0 while True: t = tape[i] i += 1 if t == 0: return bytes(out) if t <= 127: for _ in range(t): out.append((256 - tape[i]) % 256) i += 1 elif t == 128: raise ValueError("tag 128 is reserved") else: out.extend([(256 - tape[i]) % 256] * (256 - t)) i += 1 # ============================================================================= # Framing: ser and inst # ============================================================================= def lam(k: int) -> bytes: """Eight-byte little-endian length field.""" return struct.pack(" bytes: return lam(len(u)) + u def ser(sigma: bytes, m: bytes, tau: bytes) -> bytes: assert len(m) == 256 return field(sigma) + field(m) + tau def inst(s: bytes) -> Tuple[bytes, bytes, bytes]: """Partial inverse of ser: recover (sigma, m, tau).""" if len(s) < 8: raise ValueError("truncated") a = struct.unpack(" Dict[int, int]: mem = {} for idx, (a, b, c) in enumerate(prog): mem[idx * 3] = a mem[idx * 3 + 1] = b mem[idx * 3 + 2] = c return mem # The decoding loop. Addresses are 3k; every branch target below is written as # an instruction index and resolved to 3*index by _emit. # # k0 T2 <- 0 # k1 T1 <- 0 # k2 request read of the tag # k3 T1 <- -T # k4 T2 <- T # k5 branch to k7 when T = 0 or T >= 128 # k6 goto LITERAL # k7 branch to END when 256-T <= 0, that is T in {0,128} # k8 goto REPEAT with T1 = 256-T the run length # k9 LITERAL: request read of the next byte # k10 R_OUT <- b # k11 emit # k12 T2 <- T2-1; branch to k14 when the count is exhausted # k13 goto k9 # k14 goto k0 # k15 REPEAT: request read of the value byte # k16 R_OUT <- b # k17 emit # k18 T1 <- T1-1; branch to k14 when the count is exhausted # k19 goto k16 # k20 END DECODE_LOOP = [ (T2, T2, 3 * 1), # k0 (T1, T1, 3 * 2), # k1 (NEG1, C_RD, 3 * 3), # k2 (R_IN, T1, 3 * 4), # k3 (T1, T2, 3 * 5), # k4 (Z, T2, 3 * 7), # k5 (Z, Z, 3 * 9), # k6 (Z, T1, 3 * 20), # k7 (Z, Z, 3 * 15), # k8 (NEG1, C_RD, 3 * 10), # k9 (R_IN, R_OUT, 3 * 11), # k10 (NEG1, C_WR, 3 * 12), # k11 (ONE, T2, 3 * 14), # k12 (Z, Z, 3 * 9), # k13 (Z, Z, 3 * 0), # k14 (NEG1, C_RD, 3 * 16), # k15 (R_IN, R_OUT, 3 * 17), # k16 (NEG1, C_WR, 3 * 18), # k17 (ONE, T1, 3 * 14), # k18 (Z, Z, 3 * 16), # k19 ] # P: the constructor. The end token halts. P = DECODE_LOOP + [(Z, Z, HALT_PC)] # P*: the end token enters the rewind phase, then the copy phase. # k20 R: request rewind # k21 B0: request read # k22 B1: EOK <- 1-EOT; halt when the end of tape is reached # k23 B2: Z <- -b # k24 B3: R_OUT <- b # k25 B4: emit # k26 B5: Z <- 0 and go to B0 P_STAR = DECODE_LOOP + [ (NEG1, C_RW, 3 * 21), # k20 (NEG1, C_RD, 3 * 22), # k21 (C_EOT, EOK, HALT_PC), # k22 (R_IN, Z, 3 * 24), # k23 (Z, R_OUT, 3 * 25), # k24 (NEG1, C_WR, 3 * 26), # k25 (Z, Z, 3 * 21), # k26 ] # P_e: P with an end-of-tape guard after the tag read, so that the machine # halts on every tape, recipe or not. Instruction j3 assigns # M[EOK] <- M[EOK] - M[C_EOT]; a successful tag read has cleared C_EOT and # leaves EOK at 1, while a read at the end of the tape sets it and the result # 0 transfers control to the halt cell. P_TOTAL = [ (T2, T2, 3 * 1), # j0 (T1, T1, 3 * 2), # j1 (NEG1, C_RD, 3 * 3), # j2 read the tag (C_EOT, EOK, HALT_PC), # j3 halt if that read was past the end of the tape (R_IN, T1, 3 * 5), # j4 (T1, T2, 3 * 6), # j5 (Z, T2, 3 * 8), # j6 (Z, Z, 3 * 10), # j7 (Z, T1, 3 * 21), # j8 (Z, Z, 3 * 16), # j9 (NEG1, C_RD, 3 * 11), # j10 LITERAL (R_IN, R_OUT, 3 * 12), # j11 (NEG1, C_WR, 3 * 13), # j12 (ONE, T2, 3 * 15), # j13 (Z, Z, 3 * 10), # j14 (Z, Z, 3 * 0), # j15 (NEG1, C_RD, 3 * 17), # j16 REPEAT (R_IN, R_OUT, 3 * 18), # j17 (NEG1, C_WR, 3 * 19), # j18 (ONE, T1, 3 * 15), # j19 (Z, Z, 3 * 17), # j20 (Z, Z, HALT_PC), # j21 END ] def decode_any(tape: bytes, guard: bool) -> Tuple[bytes, bool]: """The output of the decoding loop on an arbitrary tape.""" out = bytearray() h, rin = 0, 0 while True: if h < len(tape): rin = tape[h] h += 1 eot = 0 else: eot = 1 if guard and eot: return bytes(out), True t = rin if not guard and eot: return bytes(out), t in (0, 128) if t == 0 or t == 128: return bytes(out), True reps = 1 if t > 128 else t count = (256 - t) if t > 128 else 1 for _ in range(reps): if h < len(tape): rin = tape[h] h += 1 out.extend([(256 - rin) % 256] * count) def memory_image(prog: List[Tuple[int, int, int]]) -> List[int]: """The 256-byte initial memory image holding a program and its constants.""" mem = [0] * 256 for addr, val in _emit(prog).items(): mem[addr] = val & 0xFF mem[Z] = 0 mem[ONE] = 1 mem[T1] = 0 mem[T2] = 0 mem[NEG1] = 0xFF mem[EOK] = 1 return mem M_P = memory_image(P) M_STAR = memory_image(P_STAR) def program_a(prog: List[Tuple[int, int, int]], A: int) -> List[Tuple[int, int, int]]: """A program moved to 2^A words: the cells from 0xF0 up name the top sixteen words.""" top = (1 << A) - 256 return [tuple(x + top if x >= 0xF0 else x for x in ins) for ins in prog] def memory_image_a(prog: List[Tuple[int, int, int]], A: int) -> List[int]: """The initial memory of 2^A words holding a program moved there and its constants.""" S = 1 << A top = S - 256 mem = [0] * S for addr, val in _emit(program_a(prog, A)).items(): mem[addr] = val for cell, val in ((Z, 0), (ONE, 1), (T1, 0), (T2, 0), (NEG1, S - 1), (EOK, 1)): mem[cell + top] = val return mem # ============================================================================= # Device # ============================================================================= class Tape: """Environment state (tau, h, omega) with the operations of the definition.""" def __init__(self, tau: bytes): self.tau = tau self.h = 0 self.out = bytearray() def apply(self, mem: List[int]) -> None: """One device application to the post-instruction memory image.""" if mem[C_WR] == 1: self.out.append(mem[R_OUT]) mem[R_OUT] = 0 if mem[C_RD] == 1: if self.h < len(self.tau): mem[R_IN] = self.tau[self.h] mem[C_EOT] = 0 self.h += 1 else: mem[C_EOT] = 1 if mem[C_RW] == 1: self.h = 0 mem[C_WR] = 0 mem[C_RD] = 0 mem[C_RW] = 0 # ============================================================================= # Evaluator 1: integer reference # ============================================================================= def run_reference(mem0: List[int], tau: bytes, max_steps: int = 1 << 34, expect: Optional[bytes] = None) -> Tuple[bytes, int]: mem = list(mem0) dev = Tape(tau) pc = 0 steps = 0 while pc != HALT_PC and steps < max_steps: A = mem[pc] B = mem[(pc + 1) & 0xFF] C = mem[(pc + 2) & 0xFF] r = (mem[B] - mem[A]) & 0xFF mem[B] = r pc = C if (r == 0 or r >= 0x80) else (pc + 3) & 0xFF n_before = len(dev.out) dev.apply(mem) if expect is not None and len(dev.out) > n_before: k = len(dev.out) - 1 if k >= len(expect) or dev.out[k] != expect[k]: raise AssertionError(f"stream diverged at byte {k}") steps += 1 return bytes(dev.out), steps def run_reference_a(mem0: List[int], tau: bytes, A: int, max_steps: int = 1 << 40) -> Tuple[bytes, int]: """SUBLEQ_A with the device on its top words, a read extending the byte by its sign.""" S, half = 1 << A, 1 << (A - 1) io = io_cells(A) wr, eot, rw, rd, rin, rout, halt = (io[k] for k in ('c_wr', 'c_eot', 'c_rw', 'c_rd', 'r_in', 'r_out', 'halt_pc')) mem, out = list(mem0), bytearray() pc = h = steps = 0 while pc != halt and steps < max_steps: a, b, c = mem[pc], mem[(pc + 1) % S], mem[(pc + 2) % S] r = (mem[b] - mem[a]) % S mem[b] = r pc = c if r == 0 or r >= half else (pc + 3) % S if mem[wr] == 1: out.append(mem[rout] & 0xFF) mem[rout] = 0 if mem[rd] == 1: if h < len(tau): v = tau[h] mem[rin] = v if v < 128 else v + S - 256 mem[eot] = 0 h += 1 else: mem[eot] = 1 if mem[rw] == 1: h = 0 mem[wr] = mem[rd] = mem[rw] = 0 steps += 1 return bytes(out), steps # ============================================================================= # The host netlist and its canonical serialization # ============================================================================= STATE_LAYOUT = {"pc": [0, 8], "halt": [8, 1], "mem": [9, 256, 8]} IO_CELLS = {"c_wr": C_WR, "c_eot": C_EOT, "c_rw": C_RW, "c_rd": C_RD, "r_in": R_IN, "r_out": R_OUT, "halt_pc": HALT_PC} STATE_BITS = 8 + 1 + 2048 def host_netlist(): """The clocked netlist of the host, assembled from its source description.""" return build_subleq_step_net() def sigma_host() -> bytes: """sigma(N_host): the canonical serialization of that netlist.""" net, inputs, outputs = host_netlist() return sigma_of_net(net, inputs, outputs, "subleq8", STATE_LAYOUT, IO_CELLS) def read_host() -> bytes: """The distributed serialization of the host.""" return open(HOST_PATH, "rb").read() def sha(b: bytes) -> str: return hashlib.sha256(b).hexdigest() def tau_star(sigma: bytes, m: List[int]) -> bytes: return describe(field(sigma) + field(bytes(m))) # ============================================================================= # Evaluators of the step map, each built from sigma alone # ============================================================================= class _Transducer: """State marshalling shared by the threshold evaluators.""" N = STATE_BITS _DEVCELLS = (C_WR, C_EOT, C_RW, C_RD, R_IN, R_OUT) def _vec(self, pc: int, mem: List[int]): torch = self.torch v = torch.zeros(self.N) for k in range(8): v[k] = (pc >> (7 - k)) & 1 for j in range(256): for k in range(8): v[9 + j * 8 + k] = (mem[j] >> (7 - k)) & 1 return v @staticmethod def _byte(vc, j: int) -> int: x = 0 for k in range(8): x = (x << 1) | int(vc[9 + j * 8 + k]) return x def _set_byte(self, v, j: int, val: int) -> None: for k in range(8): v[0, 9 + j * 8 + k] = float((val >> (7 - k)) & 1) def capture(self): """Capture one application of the map as a CUDA graph.""" torch = self.torch assert self.device.startswith("cuda") self.gin = torch.zeros(1, self.N, device=self.device) # every buffer the body writes must stay alive for the life of the # graph, or the allocator will hand its memory to something else and # the replay will overwrite that instead self.gbuf = self._buffers() for _ in range(5): self._body(self.gin) torch.cuda.synchronize() self.graph = torch.cuda.CUDAGraph() with torch.cuda.graph(self.graph): self.gout = self._body(self.gin) torch.cuda.synchronize() gen = torch.Generator(device=self.device).manual_seed(3) probe = (torch.rand(1, self.N, generator=gen, device=self.device) < 0.5).float() self.gin.copy_(probe) self.graph.replay() torch.cuda.synchronize() assert bool((self.gout == self._eager(probe)).all()), \ "the captured graph differs from the eager step" return self def _replay(self, v): # the state lives in gin, which is ordinary memory; gout belongs to the # graph's private pool and is only ever read as a whole if v.data_ptr() != self.gin.data_ptr(): self.gin.copy_(v) self.graph.replay() self.gin.copy_(self.gout) return self.gin def run(self, mem0: List[int], tau: bytes, max_steps: int, expect: Optional[bytes] = None, progress: int = 0, margin: bool = False) -> Tuple[bytes, int]: """Iterate the map with the device applied after each step.""" import time torch = self.torch v = self._vec(0, mem0).unsqueeze(0).to(self.device) dev = Tape(tau) cells = list(self._DEVCELLS) cell_t = torch.tensor([9 + j * 8 + k for j in cells for k in range(8)], device=self.device) pow2_t = torch.tensor([1 << (7 - k) for k in range(8)], device=self.device, dtype=torch.float32) shift_t = torch.tensor([7 - k for k in range(8)]) shadow = [0] * 256 self.min_margin = float("inf") n = 0 t0 = time.perf_counter() while n < max_steps: if margin: self._accumulate_margin(v) v = self.step(v) n += 1 vals = torch.cat([v[0, 8:9], (v[0, cell_t].reshape(len(cells), 8) * pow2_t) .sum(-1)]).to("cpu").to(torch.int64).tolist() for c, j in enumerate(cells): shadow[j] = vals[1 + c] before = len(dev.out) dev.apply(shadow) new = torch.tensor([shadow[j] for j in cells], dtype=torch.int64) v[0, cell_t] = (((new.unsqueeze(-1) >> shift_t) & 1) .reshape(-1).float().to(self.device)) if expect is not None and len(dev.out) > before: k = len(dev.out) - 1 if k >= len(expect) or dev.out[k] != expect[k]: raise AssertionError(f"stream diverged at byte {k}") if vals[0] >= 1: break if progress and n % progress == 0: rate = n / (time.perf_counter() - t0) print(f" {self.tag} {n:,} steps, {len(dev.out):,} bytes " f"({rate:,.0f} steps/s)", flush=True) self.seconds = time.perf_counter() - t0 return bytes(dev.out), n class NetEvaluator(_Transducer): """The netlist of sigma, evaluated unit by unit.""" tag = "net" def __init__(self, sigma: bytes, device: str = "cpu", graph: bool = False): import time import torch from reflect import Leveled t0 = time.perf_counter() self.torch = torch self.device = device net, inputs, outputs, meta = net_of_sigma(sigma) self.net, self.inputs, self.outputs, self.meta = net, inputs, outputs, meta assert len(inputs) == self.N, "state width disagrees with the layout" self.lev = Leveled(net, inputs, outputs, device=device) self.info = {"levels": len(self.lev.plan), "units": len(net.gates), "entries": sum(len(i) for i, _ in net.gates.values())} self.graph = None if graph: self.capture() self.build_seconds = time.perf_counter() - t0 def _buffers(self): return self.torch.zeros(self.lev.n_sig, 1, device=self.device) def _body(self, inp): V = self.gbuf lev = self.lev V.zero_() V[1] = 1.0 V[lev.in_slots] = inp.T for idx, w, b, out in lev.plan: g = V[idx] V[out] = ((g * w[:, :, None]).sum(1) + b[:, None] >= 0).float() return V[lev.out_slots].T def _eager(self, v): return self.lev.step(v) def step(self, v): if self.graph is not None: return self._replay(v) return self.lev.step(v) class LevEvaluator(_Transducer): """Lev(N) for the netlist N of sigma.""" tag = "lev" def __init__(self, sigma: bytes, device: str = "cpu", dense: bool = True, graph: bool = False): import time import torch from machines import compile_net t0 = time.perf_counter() self.torch = torch self.device = device self.dense = dense net, inputs, outputs, meta = net_of_sigma(sigma) self.net, self.inputs, self.outputs, self.meta = net, inputs, outputs, meta assert len(inputs) == self.N, "state width disagrees with the layout" layers, info = compile_net(net, inputs, outputs) for W, _ in layers: assert set(torch.unique(W).tolist()) <= {-1, 0, 1} self.info = dict(info) self.info["size"] = sum(int(W.shape[0]) for W, _ in layers) self.info["nonzero"] = sum(int((W != 0).sum()) for W, _ in layers) if dense: self.W = [W.to(device=device, dtype=torch.float32) for W, _ in layers] self.B = [b.to(device=device, dtype=torch.float32) for _, b in layers] else: pad = device.startswith("cuda") self.plan = [self._sparsify(W, b, device, pad) for W, b in layers] self.widths = [int(W.shape[0]) for W, _ in layers] self.graph = None if graph: self.capture() self.build_seconds = time.perf_counter() - t0 @staticmethod def _sparsify(W, b, device, pad): """One layer as the nonzero entries of its rows.""" import torch nz = (W != 0) counts = nz.sum(1) n = int(W.shape[0]) sizes = [int(counts.max())] if pad else sorted(set(counts.tolist())) groups = [] for k in sizes: rows = (torch.arange(n) if pad else torch.nonzero(counts == k, as_tuple=False).flatten()) idx = torch.zeros(len(rows), max(k, 1), dtype=torch.long) w = torch.zeros(len(rows), max(k, 1)) for r, row in enumerate(rows.tolist()): cols = torch.nonzero(nz[row], as_tuple=False).flatten() idx[r, :len(cols)] = cols w[r, :len(cols)] = W[row, cols] groups.append((rows.to(device), idx.to(device), w.to(device=device, dtype=torch.float32), b[rows].to(device=device, dtype=torch.float32))) return groups def _buffers(self): import torch return [torch.zeros(1, n, device=self.device) for n in self.widths] def _sparse_step(self, v, buf): x = v for groups, y in zip(self.plan, buf): for rows, idx, w, b in groups: y[:, rows] = ((x[:, idx] * w).sum(-1) + b >= 0).float() x = y return x def _body(self, inp): return self._sparse_step(inp, self.gbuf) def _eager(self, v): return self._sparse_step(v, [self.torch.zeros(v.shape[0], n, device=self.device) for n in self.widths]) def step(self, v): if self.dense: for W, b in zip(self.W, self.B): v = ((v @ W.T + b) >= 0).float() return v if self.graph is not None: return self._replay(v) return self._eager(v) def _accumulate_margin(self, v): y = v for W, b in zip(self.W, self.B): pre = y @ W.T + b m = float((pre + 0.5).abs().min()) if m < self.min_margin: self.min_margin = m y = (pre >= 0).float() def step_noisy(self, v, sigma: float, gen): """One step with additive Gaussian read noise per pre-activation and the comparator at -1/2.""" torch = self.torch if self.dense: for W, b in zip(self.W, self.B): pre = v @ W.T + b pre = pre + torch.randn(pre.shape, generator=gen, device=pre.device) * sigma v = (pre >= -0.5).float() return v x = v for groups, n in zip(self.plan, self.widths): y = torch.empty(x.shape[0], n, device=x.device) for rows, idx, w, b in groups: pre = (x[:, idx] * w).sum(-1) + b pre = pre + torch.randn(pre.shape, generator=gen, device=pre.device) * sigma y[:, rows] = (pre >= -0.5).float() x = y return x REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # repo root; this module lives in src/ MODEL_PATH = os.path.join(REPO, "variants", "neural_subleq8io.safetensors") IN_DATA, IN_STROBE, OUT = 0xFD, 0xFC, 0xFE # ============================================================================= # Build the host file (family per-gate style, .inputs wiring, ternary strict) # ============================================================================= def build_host(save: bool = True) -> str: import build as B from build import quantize_tensors tensors: Dict[str, torch.Tensor] = {} B.add_decoder(tensors, 8, 256) B.add_memory_read_mux(tensors, 256) B.add_memory_write_cells(tensors, 256) B.add_subleq_core(tensors) tensors["manifest.data_bits"] = torch.tensor([8.0]) tensors["manifest.addr_bits"] = torch.tensor([8.0]) tensors["manifest.memory_bytes"] = torch.tensor([256.0]) tensors["manifest.pc_width"] = torch.tensor([8.0]) tensors["manifest.version"] = torch.tensor([4.0]) tensors, reg, stats = B.build_inputs(tensors) tensors, counts, _, _ = quantize_tensors(tensors, ternary=False) for k, t in tensors.items(): if k.endswith(".weight") and not k.startswith("manifest."): tf = t.float() if t.dtype.is_floating_point else t.to(torch.float32) assert (tf.abs() <= 1).all(), f"non-ternary weight {k}" meta = { "machine": "subleq8io", "weight_quantization": "ternary", "signal_registry": reg.to_metadata(), "io": json.dumps({"in_data": IN_DATA, "in_strobe": IN_STROBE, "out": OUT, "halt_pc": HALT_PC}), } if save: save_file(tensors, MODEL_PATH, metadata=meta) sz = os.path.getsize(MODEL_PATH) print(f" built {MODEL_PATH}: {stats['added']} wired gates, " f"{len(tensors)} tensors, {sz:,} bytes") return MODEL_PATH # ============================================================================= # Recipe codec (the construction-description language) # ============================================================================= def c8_describe(data: bytes) -> bytes: """Compile raw bytes into a construction recipe for the constructor.""" tape = bytearray() i, n = 0, len(data) while i < n: j = i while j < n and data[j] == data[i] and j - i < 127: j += 1 if j - i >= 4: tape.append(256 - (j - i)) # repeat tag 129..252 tape.append((256 - data[i]) % 256) # value, negated i = j continue k = i while k < n and k - i < 127: # literal until a run starts m = k while m < n and data[m] == data[k] and m - k < 4: m += 1 if m - k >= 4: break k += 1 k = max(k, i + 1) tape.append(k - i) # literal tag 1..127 tape.extend((256 - x) % 256 for x in data[i:k]) i = k tape.append(0) # END return bytes(tape) def c8_decode(tape: bytes) -> bytes: """Pure-Python reference semantics of the recipe language.""" out = bytearray() i = 0 while True: t = tape[i] i += 1 if t == 0: return bytes(out) if t <= 127: for _ in range(t): out.append((256 - tape[i]) % 256) i += 1 else: out.extend([(256 - tape[i]) % 256] * (256 - t)) i += 1 # ============================================================================= # The constructor program (21 SUBLEQ instructions + 4 variable cells) # ============================================================================= CONSTRUCTOR = [ # main loop: read tag T; T1 = -T (== run length for repeat tags), T2 = T (T2, T2, 0x03), # 00 T2 = 0 (T1, T1, 0x06), # 03 T1 = 0 (IN_DATA, T1, 0x09), # 06 T1 = -T (Z, IN_STROBE, 0x0C), # 09 consume tag byte (T1, T2, 0x0F), # 0C T2 = T (Z, T2, 0x15), # 0F if T == 0 or T >= 128 -> 15, else literal (Z, Z, 0x24), # 12 goto LITERAL (Z, T1, 0x1B), # 15 T1 = -T: leq only when T == 0 -> END (Z, Z, 0x30), # 18 goto REPEAT (T1 holds 256-T = run length) (Z, Z, HALT_PC), # 1B END: halt (Z, Z, 0x21), # 1E (pad) (Z, Z, 0x24), # 21 (pad) (IN_DATA, OUT, 0x27), # 24 LITERAL: emit 0 - tape_byte (Z, IN_STROBE, 0x2A), # 27 consume it (ONE, T2, 0x00), # 2A T2 -= 1; done -> main loop (Z, Z, 0x24), # 2D next literal byte (IN_DATA, OUT, 0x33), # 30 REPEAT: emit 0 - value (no strobe) (ONE, T1, 0x39), # 33 T1 -= 1; done -> 39 (Z, Z, 0x30), # 36 emit again (Z, IN_STROBE, 0x3C), # 39 consume the value byte (Z, Z, 0x00), # 3C goto main loop ] def constructor_image() -> List[int]: mem = [0] * 256 for idx, (a, b, c) in enumerate(CONSTRUCTOR): mem[idx * 3] = a mem[idx * 3 + 1] = b mem[idx * 3 + 2] = c mem[ONE] = 1 return mem # ============================================================================= # Backend 1: fast pure-integer reference with devices # ============================================================================= def ref_construct(tape: bytes, max_steps: int = 1 << 34, ) -> Tuple[bytes, int]: mem = constructor_image() mem[IN_DATA] = tape[0] if tape else 0 out = bytearray() head = 0 pc = 0 steps = 0 tn = len(tape) while pc != HALT_PC and steps < max_steps: A = mem[pc] Bc = mem[(pc + 1) & 0xFF] C = mem[(pc + 2) & 0xFF] r = (mem[Bc] - mem[A]) & 0xFF mem[Bc] = r pc = C if (r == 0 or r >= 0x80) else (pc + 3) & 0xFF if Bc == OUT: out.append(r) mem[OUT] = 0 elif Bc == IN_STROBE: head += 1 mem[IN_STROBE] = 0 mem[IN_DATA] = tape[head] if head < tn else 0 steps += 1 return bytes(out), steps # ============================================================================= # Backend 2: the gate-graph processor (shipped netlist) with devices # ============================================================================= class GateHost: def __init__(self, path: str = MODEL_PATH): from machines import SubleqThresholdCPU, load_tensors from eval import load_metadata T = load_tensors(path) reg = load_metadata(path)["signal_registry"] self.cpu = SubleqThresholdCPU(T, reg) def run(self, tape: bytes, max_steps: int) -> Tuple[bytes, int, dict]: s = {"pc": 0, "mem": constructor_image(), "halted": False} s["mem"][IN_DATA] = tape[0] if tape else 0 out = bytearray() head = 0 n = 0 while not s["halted"] and n < max_steps: Bc = s["mem"][(s["pc"] + 1) & 0xFF] s = self.cpu.step(s) if Bc == OUT: out.append(s["mem"][OUT]) s["mem"][OUT] = 0 elif Bc == IN_STROBE: head += 1 s["mem"][IN_STROBE] = 0 s["mem"][IN_DATA] = tape[head] if head < len(tape) else 0 n += 1 return bytes(out), n, s # ============================================================================= # Backend 3: the host compiled to recurrent ternary matrices (machine 1 tech) # ============================================================================= class MatrixHost: """The subleq8io processor as a recurrent ternary threshold network, the device applied between steps.""" N = 8 + 1 + 2048 def __init__(self, device: str = "cpu"): t0 = time.perf_counter() net, inputs, outputs = build_subleq_step_net() layers, info = compile_net(net, inputs, outputs) for W, Bt in layers: vals = set(torch.unique(W).tolist()) assert vals <= {-1, 0, 1} self.W = [W.to(device=device, dtype=torch.float32) for W, _ in layers] self.B = [Bt.to(device=device, dtype=torch.float32) for _, Bt in layers] self.device = device self.info = info print(f" host compiled to {info['layers']} ternary matrices " f"(max width {info['max_width']}, " f"{info['total_weights']:,} weights) in " f"{time.perf_counter() - t0:.1f}s on {device}") def _vec(self, pc: int, mem: List[int], halt: bool = False) -> torch.Tensor: v = torch.zeros(self.N) for k in range(8): v[k] = (pc >> (7 - k)) & 1 v[8] = 1.0 if halt else 0.0 for j in range(256): for k in range(8): v[9 + j * 8 + k] = (mem[j] >> (7 - k)) & 1 return v def step(self, v: torch.Tensor) -> torch.Tensor: for W, b in zip(self.W, self.B): v = ((v @ W.T + b) >= 0).float() return v @staticmethod def _byte(vcpu: torch.Tensor, j: int) -> int: x = 0 for k in range(8): x = (x << 1) | int(vcpu[9 + j * 8 + k]) return x def run(self, tape: bytes, max_steps: int, expect: Optional[bytes] = None, report_every: int = 100000) -> Tuple[bytes, int]: mem = constructor_image() mem[IN_DATA] = tape[0] if tape else 0 v = self._vec(0, mem).unsqueeze(0).to(self.device) out = bytearray() head = 0 n = 0 idx_pch = torch.arange(0, 9, device=self.device) t0 = time.perf_counter() while n < max_steps: front = v[0, idx_pch].tolist() if front[8] >= 0.5: break pcv = 0 for k in range(8): pcv = (pcv << 1) | int(front[k]) bidx = (pcv + 1) & 0xFF bbits = v[0, 9 + bidx * 8: 9 + bidx * 8 + 8].tolist() Bc = 0 for k in range(8): Bc = (Bc << 1) | int(bbits[k]) v = self.step(v) n += 1 if Bc == OUT: ob = v[0, 9 + OUT * 8: 9 + OUT * 8 + 8].tolist() val = 0 for k in range(8): val = (val << 1) | int(ob[k]) out.append(val) if expect is not None and out[-1] != expect[len(out) - 1]: raise AssertionError( f"stream diverged at byte {len(out) - 1}: " f"got {out[-1]:#04x} want {expect[len(out) - 1]:#04x}") v[0, 9 + OUT * 8: 9 + OUT * 8 + 8] = 0.0 elif Bc == IN_STROBE: head += 1 v[0, 9 + IN_STROBE * 8: 9 + IN_STROBE * 8 + 8] = 0.0 nxt = tape[head] if head < len(tape) else 0 for k in range(8): v[0, 9 + IN_DATA * 8 + k] = float((nxt >> (7 - k)) & 1) if report_every and n % report_every == 0: rate = n / (time.perf_counter() - t0) print(f" ... {n:,} recurrences, {len(out):,} bytes emitted " f"({rate:,.0f} instr/s)") return bytes(out), n def exhaustive_datapath(self) -> bool: """All 65,536 (x, y) operand pairs through the matrices in one batch.""" xs = torch.arange(65536) % 256 ys = torch.arange(65536) // 256 V = torch.zeros(65536, self.N) mem = [0] * 256 mem[0], mem[1], mem[2] = 0x90, 0x91, 0x30 v0 = self._vec(0, mem) V[:] = v0 for k in range(8): V[:, 9 + 0x90 * 8 + k] = ((xs >> (7 - k)) & 1).float() V[:, 9 + 0x91 * 8 + k] = ((ys >> (7 - k)) & 1).float() V = V.to(self.device) for W, b in zip(self.W, self.B): V = ((V @ W.T + b) >= 0).float() V = V.cpu() r = torch.zeros(65536, dtype=torch.long) for k in range(8): r = (r << 1) | V[:, 9 + 0x91 * 8 + k].long() pcv = torch.zeros(65536, dtype=torch.long) for k in range(8): pcv = (pcv << 1) | V[:, k].long() exp_r = (ys - xs) & 0xFF exp_leq = (exp_r == 0) | (exp_r >= 0x80) exp_pc = torch.where(exp_leq, torch.full_like(pcv, 0x30), torch.full_like(pcv, 3)) ok = bool((r == exp_r).all()) and bool((pcv == exp_pc).all()) print(f" {'OK ' if ok else 'FAIL'} matrix datapath exhaustive: " f"65,536/65,536 subtract results and branch decisions exact") return ok # ============================================================================= # Verification # ============================================================================= def c8_sha(b: bytes) -> str: return hashlib.sha256(b).hexdigest()[:16] def family_files() -> List[str]: files = [os.path.join(REPO, "neural_computer.safetensors")] vdir = os.path.join(REPO, "variants") files += sorted(os.path.join(vdir, f) for f in os.listdir(vdir) if f.endswith(".safetensors")) return files def verify(device: str) -> bool: ok = True print("\n[1/5] Host machine soundness (shipped netlist, exhaustive datapath)") from eval import NetlistEvaluator, load_metadata from machines import load_tensors T = load_tensors(MODEL_PATH) reg = load_metadata(MODEL_PATH)["signal_registry"] ne = NetlistEvaluator(T, reg, "subleq") a_v = torch.arange(65536) % 256 b_v = torch.arange(65536) // 256 ext = {} for k in range(8): ext[f"$a[{k}]"] = ((a_v >> k) & 1).float() ext[f"$b[{k}]"] = ((b_v >> k) & 1).float() ext[f"$pc[{k}]"] = torch.zeros(65536) ext[f"$c[{k}]"] = torch.zeros(65536) o = ne.run(ext) rr = sum(o[f"subleq.sub.fa{k}.ha2.sum.layer2"][:, 0].long() << k for k in range(8)) lq = o["subleq.leq"][:, 0].long() exp_r = (b_v - a_v) & 0xFF exp_l = ((exp_r == 0) | (exp_r >= 128)).long() good = bool((rr == exp_r).all()) and bool((lq == exp_l).all()) print(f" {'OK ' if good else 'FAIL'} gate netlist exhaustive: 65,536/65,536 exact") ok &= good print("\n[2/5] Recipe codec round-trip over the whole family") total = 0 for p in family_files(): data = open(p, "rb").read() tape = c8_describe(data) good = c8_decode(tape) == data ok &= good total += len(data) print(f" {'OK ' if good else 'FAIL'} {os.path.basename(p):<44} " f"{len(data):>10,} B -> tape {len(tape):>10,} B sha {c8_sha(data)}") print(f" ({total / 1e6:.0f} MB described and decoded back, byte-identical)") print("\n[3/5] Constructor program on the integer reference " "(output stream == target file, byte for byte)") targets = [MODEL_PATH, os.path.join(REPO, "variants", "neural_subleq8.safetensors"), os.path.join(REPO, "variants", "neural_matrix8.safetensors")] for p in targets: data = open(p, "rb").read() tape = c8_describe(data) t0 = time.perf_counter() out, steps = ref_construct(tape) dt = time.perf_counter() - t0 good = out == data ok &= good tag = "SELF-REPRODUCTION" if p == MODEL_PATH else "universal build" print(f" {'OK ' if good else 'FAIL'} {os.path.basename(p):<40} " f"{len(out):>10,} B in {steps:>12,} instructions ({dt:.1f}s) " f"[{tag}] sha {c8_sha(out)}") print("\n[4/5] Matrix backend (machine-1 compile of the host)") mh = MatrixHost(device=device) ok &= mh.exhaustive_datapath() print("\n[5/5] Backend lockstep on a shared prefix " "(gate graph vs matrices vs reference, first 600 cycles)") data = open(MODEL_PATH, "rb").read() tape = c8_describe(data) gh = GateHost() s = {"pc": 0, "mem": constructor_image(), "halted": False} s["mem"][IN_DATA] = tape[0] mem2 = constructor_image() mem2[IN_DATA] = tape[0] v = mh._vec(0, mem2).unsqueeze(0).to(device) headg = headm = 0 outg, outm = bytearray(), bytearray() good = True for n in range(600): Bg = s["mem"][(s["pc"] + 1) & 0xFF] s = gh.cpu.step(s) if Bg == OUT: outg.append(s["mem"][OUT]) s["mem"][OUT] = 0 elif Bg == IN_STROBE: headg += 1 s["mem"][IN_STROBE] = 0 s["mem"][IN_DATA] = tape[headg] if headg < len(tape) else 0 vc = v[0].cpu() pcm = 0 for k in range(8): pcm = (pcm << 1) | int(vc[k]) Bm = MatrixHost._byte(vc, (pcm + 1) & 0xFF) v = mh.step(v) if Bm == OUT: outm.append(MatrixHost._byte(v[0].cpu(), OUT)) v[0, 9 + OUT * 8: 9 + OUT * 8 + 8] = 0.0 elif Bm == IN_STROBE: headm += 1 v[0, 9 + IN_STROBE * 8: 9 + IN_STROBE * 8 + 8] = 0.0 nxt = tape[headm] if headm < len(tape) else 0 for k in range(8): v[0, 9 + IN_DATA * 8 + k] = float((nxt >> (7 - k)) & 1) vc = v[0].cpu() pcm = 0 for k in range(8): pcm = (pcm << 1) | int(vc[k]) memm = [MatrixHost._byte(vc, j) for j in range(256)] if pcm != s["pc"] or memm != s["mem"] or outg != outm: print(f" FAIL lockstep diverged at cycle {n}") good = False break print(f" {'OK ' if good else 'FAIL'} gate graph and matrix form agree " f"cycle-for-cycle for 600 cycles ({len(outg)} bytes emitted identically)") ok &= good print("\nCONSTRUCTOR VERIFICATION:", "PASS" if ok else "FAIL") return ok def self_reproduce(device: str, gen2: bool = True) -> bool: """Full self-reproduction at threshold fidelity on the matrix backend.""" data = open(MODEL_PATH, "rb").read() tape = c8_describe(data) print(f"\nSelf-reproduction on the threshold matrices: target " f"{len(data):,} bytes (sha {c8_sha(data)}), tape {len(tape):,} bytes") ref_out, ref_steps = ref_construct(tape) assert ref_out == data, "reference construction failed" print(f" reference: {ref_steps:,} instructions") mh = MatrixHost(device=device) t0 = time.perf_counter() out, n = mh.run(tape, max_steps=ref_steps + 64, expect=data) dt = time.perf_counter() - t0 ok = out == data and n == ref_steps print(f" GEN 1: emitted {len(out):,} bytes in {n:,} recurrences " f"({dt / 60:.1f} min, {n / dt:,.0f} instr/s) sha {c8_sha(out)}") print(f" {'OK ' if ok else 'FAIL'} output stream == host safetensors file, " f"byte for byte; instruction count matches reference exactly") if ok and gen2: off_path = os.path.join(REPO, "variants", "_offspring_subleq8io.safetensors") with open(off_path, "wb") as f: f.write(out) gh = GateHost(off_path) # boot the offspring's own gates probe_tape = c8_describe(out[:2048]) want = out[:2048] got, steps, _ = gh.run(probe_tape, max_steps=len(probe_tape) * 8 + 64) good2 = got == want print(f" GEN 2: offspring file booted as a machine (gate graph); " f"constructed the first {len(want):,} bytes of itself in " f"{steps:,} instructions: {'exact' if good2 else 'MISMATCH'}") os.remove(off_path) ok &= good2 print("SELF-REPRODUCTION:", "PASS" if ok else "FAIL") return ok def main_constructor8() -> int: ap = argparse.ArgumentParser() ap.add_argument("cmd", choices=["build", "verify", "self", "all"]) ap.add_argument("--device", default="cuda" if torch.cuda.is_available() else "cpu") args = ap.parse_args() rc = 0 if args.cmd in ("build", "all"): build_host(save=True) if args.cmd in ("verify", "all"): rc |= 0 if verify(args.device) else 1 if args.cmd in ("self", "all"): rc |= 0 if self_reproduce(args.device) else 1 return rc def cmd_family(): import hashlib import time for A in [int(a) for a in sys.argv[1:]] or [3, 4, 5, 8]: t0 = time.perf_counter() s = sigma(A) t1 = time.perf_counter() line = f"A={A}: sigma {len(s):,} bytes in {t1 - t0:.2f}s, sha256 {hashlib.sha256(s).hexdigest()[:16]}" if A <= 12: net, inp, out = build(A) same = family_sigma_of_net(net, inp, out, A) == s depth = {} for gname, (ins, _) in net.gates.items(): depth[gname] = 1 + max((depth.get(x, 0) for x, _ in ins), default=0) line += (f"; Net {len(net.gates):,} units, depth {max(depth.values())}, " f"edges {sum(len(i) for i, _ in net.gates.values()):,}; " f"equal to the Net's serialization: {same}") N2, i2, o2, _ = net_of_sigma(s) line += f"; parses to {len(N2.gates):,} units" print(line, flush=True) COMMANDS = {'family': cmd_family, 'constructor8': main_constructor8} if __name__ == "__main__": if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS: sys.exit('commands: ' + ', '.join(COMMANDS)) cmd = sys.argv.pop(1) sys.argv[0] += ' ' + cmd sys.exit(COMMANDS[cmd]())