"""neural_reflect — a universal ternary-threshold netlist processor whose transition is a fixed, machine-independent interpreter U over a netlist held in the writable state. U is the only fixed part: a ternary threshold circuit that evaluates one stored gate per recurrence. Everything machine-specific is the stored netlist, which the running program can read, edit, copy, and replace. U is universal over ternary threshold netlists, so the fixed part is a substrate (like a cellular-automaton rule), not a particular machine. State: a flat array of S binary signals (A = log2 S), two netlist banks, and memory-mapped device signals of the kind the rest of the family already uses (rv32's console, subleq8io's tape): [ PTR | OUT_DATA OUT_STROBE PTR_ADV BANK HALT | work | NET bank0 | NET bank1 ] Each gate record holds F input slots (address, ternary weight, index flag), a signed bias, and an output (address, index flag). An index flag adds PTR to the address, so a fixed-size program can walk an address range. Writing OUT_STROBE emits OUT_DATA; writing PTR_ADV advances PTR; both are cleared by the device. These make loaders, a self-reproduction quine, and bank replacement expressible as programs. One recurrence, gate gp of the active bank: eff(addr, idx) = (addr + (idx ? PTR : 0)) mod S out = H( sum_j w_j * sig[eff(a_j, i_j)] + bias ) sig[eff(oa, oidx)] = out G recurrences sweep the netlist once; G sweeps settle any acyclic netlist regardless of gate order, so U reproduces the family's combinational evaluation. Capabilities: - compile_to_reflect() maps any <=2-input family Net to a stored netlist; a ripple-carry adder and the family's SUBLEQ datapath run through the interpreter bit-exact, independent of the order the gates are stored in. - Read by gate index, the netlist can far exceed the addressed signal span: a whole SUBLEQ machine, its memory held in signals, runs one instruction per sweep with the interpreter's indexed addressing as its memory-access hardware. - A program streams the netlist's own bytes to the output device; another rewrites its own gate so the next sweep computes a different function; a BANK flag runs a different resident machine. A netlist with feedback holds state across sweeps. - The accumulator width follows from F and the bias width. U compiles to a recurrent ternary matrix stack equal to the gate graph, with a measured noise and conductance-mismatch margin. Usage: python src/reflect.py build # save variants/neural_reflect.safetensors python src/reflect.py verify # universality, order, quine, metamorphosis python src/reflect.py analog # matrix == gate + noise / mismatch margins python src/reflect.py all """ from __future__ import annotations import argparse import json import math import os import random import sys import time from dataclasses import dataclass, field from typing import Dict, List, Tuple import torch from safetensors.torch import save_file sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from matrix8 import Net, compile_net REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) MODEL_PATH = os.path.join(REPO, "variants", "neural_reflect.safetensors") WCODE = {1: (0, 1), -1: (1, 0), 0: (0, 0)} # ============================================================================= # configuration # ============================================================================= @dataclass class Cfg: A: int = 10 G: int = 11 F: int = 2 BB: int = 4 banks: int = 2 self_mod: bool = True # True: netlist is address-reachable (editable); # False: netlist read only by gate index -> the # addressed span stays small as G grows. def __post_init__(self): self.S = 1 << self.A self.SLOT = self.A + 3 self.R = self.F * self.SLOT + self.BB + (self.A + 1) rng = self.F + (1 << (self.BB - 1)) self.ACC = int(math.ceil(math.log2(rng + 1))) + 2 self.PTR_BASE = 0 self.OUT_DATA = self.A self.OUT_STROBE = self.A + 1 self.PTR_ADV = self.A + 2 self.BANK_SIG = self.A + 3 self.HALT_SIG = self.A + 4 self.TRASH = self.A + 5 # no-op sink; no program reads it self.WORK_BASE = self.A + 6 self.NET0 = self.S - self.banks * self.G * self.R self.NET1 = self.NET0 + self.G * self.R self.span = self.S if self.self_mod else self.NET0 # indirect coverage self.GPW = max(1, (self.G - 1).bit_length()) self.STATE_BITS = self.S + self.GPW + 1 assert self.NET0 >= self.WORK_BASE + 16, "too little data room; raise A" def bank_base(self, bank): return self.NET0 if bank == 0 else self.NET1 def slot_off(cfg, j): return j * cfg.SLOT def field_off(cfg): o = cfg.F * cfg.SLOT return o, o + cfg.BB # ============================================================================= # bit helpers + encoding # ============================================================================= def to_bits(v, n): return [(v >> (n - 1 - i)) & 1 for i in range(n)] def from_bits(bits): v = 0 for b in bits: v = (v << 1) | int(b) return v def signed(bits): v = from_bits(bits) return v - (1 << len(bits)) if bits[0] else v def encode_gate(cfg, slots, bias, out): rec = [0] * cfg.R for j, (a, w, idx) in enumerate(slots): b = slot_off(cfg, j) rec[b:b + cfg.A] = to_bits(a, cfg.A) rec[b + cfg.A:b + cfg.A + 2] = list(WCODE[w]) rec[b + cfg.A + 2] = idx bo, oo = field_off(cfg) rec[bo:bo + cfg.BB] = to_bits(bias & ((1 << cfg.BB) - 1), cfg.BB) oa, oidx = out rec[oo:oo + cfg.A] = to_bits(oa, cfg.A) rec[oo + cfg.A] = oidx return rec def pad(cfg, slots): slots = list(slots) while len(slots) < cfg.F: slots.append((0, 0, 0)) return slots[:cfg.F] def encode_netlist(cfg, gates): out = [] noop = ([(0, 0, 0)] * cfg.F, -1, (cfg.TRASH, 0)) for g in (gates + [noop] * (cfg.G - len(gates))): slots, bias, o = g out += encode_gate(cfg, pad(cfg, slots), bias, o) return out # ============================================================================= # reference: gate step + memory-mapped device logic # ============================================================================= def apply_device(cfg, sig): emit = None if sig[cfg.OUT_STROBE] == 1: emit = sig[cfg.OUT_DATA] sig[cfg.OUT_STROBE] = 0 if sig[cfg.PTR_ADV] == 1: p = (from_bits(sig[cfg.PTR_BASE:cfg.PTR_BASE + cfg.A]) + 1) & (cfg.S - 1) sig[cfg.PTR_BASE:cfg.PTR_BASE + cfg.A] = to_bits(p, cfg.A) sig[cfg.PTR_ADV] = 0 return emit def ref_gate(cfg, st): sig = list(st["sig"]); gp = st["gp"]; halt = st["halt"] if halt: return sig, gp, halt ptr = from_bits(sig[cfg.PTR_BASE:cfg.PTR_BASE + cfg.A]) bank = sig[cfg.BANK_SIG] rec = sig[cfg.bank_base(bank) + gp * cfg.R:][:cfg.R] def eff(a, idx): return (a + (ptr if idx else 0)) & (cfg.S - 1) def wv(hilo): return {(0, 1): 1, (1, 0): -1}.get(tuple(hilo), 0) acc = 0 for j in range(cfg.F): b = slot_off(cfg, j) e = eff(from_bits(rec[b:b + cfg.A]), rec[b + cfg.A + 2]) acc += wv(rec[b + cfg.A:b + cfg.A + 2]) * (sig[e] if e < cfg.span else 0) bo, oo = field_off(cfg) acc += signed(rec[bo:bo + cfg.BB]) out = 1 if acc >= 0 else 0 dst = eff(from_bits(rec[oo:oo + cfg.A]), rec[oo + cfg.A]) h = sig[cfg.HALT_SIG] if dst < cfg.span: sig[dst] = out ngp = 0 if gp == cfg.G - 1 else gp + 1 return sig, ngp, (1 if (halt or h) else 0) def ref_step(cfg, st): halt_in = st["halt"] sig, gp, halt = ref_gate(cfg, st) emit = apply_device(cfg, sig) if halt_in == 0 else None return {"sig": sig, "gp": gp, "halt": halt, "emit": emit} # ============================================================================= # the interpreter as a fixed ternary threshold circuit # ============================================================================= def build_net(cfg): net = Net() sig = [f"s{i}" for i in range(cfg.S)] gp = [f"gp{k}" for k in range(cfg.GPW)] halt = "halt" nhalt = net.NOT("nhalt", halt) bank = sig[cfg.BANK_SIG] gp_oh = [net.DECODE(f"gpoh{g}", gp, g) for g in range(cfg.G)] def netbit(g, r): b0 = sig[cfg.NET0 + g * cfg.R + r] if cfg.banks == 1: return b0 return net.MUX(f"bk{g}_{r}", bank, sig[cfg.NET1 + g * cfg.R + r], b0) recbit = [net.OR(f"rb{r}", [net.AND(f"rb{r}g{g}", [gp_oh[g], netbit(g, r)]) for g in range(cfg.G)]) for r in range(cfg.R)] ptr = [sig[cfg.PTR_BASE + k] for k in range(cfg.A)] def eff_addr(addrbits, idx, tag): m = [net.AND(f"pm{tag}_{k}", [ptr[k], idx]) for k in range(cfg.A)] a_lsb = [addrbits[cfg.A - 1 - k] for k in range(cfg.A)] m_lsb = [m[cfg.A - 1 - k] for k in range(cfg.A)] c = "#0"; s_lsb = [] for k in range(cfg.A): s_, c = net.FA(f"eff{tag}_fa{k}", a_lsb[k], m_lsb[k], c) s_lsb.append(s_) return [s_lsb[cfg.A - 1 - k] for k in range(cfg.A)] def read(addrbits, tag): return net.OR(f"rd{tag}", [net.AND(f"rd{tag}a{s}", [net.DECODE(f"rd{tag}oh{s}", addrbits, s), sig[s]]) for s in range(cfg.span)]) contribs = [] for j in range(cfg.F): b = slot_off(cfg, j) addrbits = recbit[b:b + cfg.A] hi, lo, idx = recbit[b + cfg.A], recbit[b + cfg.A + 1], recbit[b + cfg.A + 2] v = read(eff_addr(addrbits, idx, f"r{j}"), f"{j}") pos = net.AND(f"pos{j}", [net.NOT(f"nhi{j}", hi), lo]) neg = net.AND(f"neg{j}", [hi, net.NOT(f"nlo{j}", lo)]) mag = net.OR(f"mag{j}", [net.AND(f"tp{j}", [pos, v]), net.AND(f"tn{j}", [neg, v])]) contribs.append((mag, net.AND(f"sg{j}", [neg, v]))) W = cfg.ACC bo, oo = field_off(cfg) biasbits = recbit[bo:bo + cfg.BB] def clsb(mag, sign): return [mag] + [sign] * (W - 1) def blsb(): lo = [biasbits[cfg.BB - 1 - k] for k in range(cfg.BB)] return lo + [biasbits[0]] * (W - cfg.BB) def ripple(pfx, a, b): c = "#0"; out = [] for k in range(W): s_, c = net.FA(f"{pfx}_fa{k}", a[k], b[k], c) out.append(s_) return out acc = clsb(*contribs[0]) for j in range(1, cfg.F): acc = ripple(f"acc{j}", acc, clsb(*contribs[j])) acc = ripple("accb", acc, blsb()) out = net.NOT("out", acc[W - 1]) oea = eff_addr(recbit[oo:oo + cfg.A], recbit[oo + cfg.A], "w") nsig = [] for s in range(cfg.S): if s < cfg.span: # only the addressed span is writable sel = net.AND(f"wrsel{s}", [net.DECODE(f"wroh{s}", oea, s), nhalt]) nsig.append(net.MUX(f"nsig{s}", sel, out, sig[s])) else: # netlist (index-read only) passes through nsig.append(sig[s]) gp_lsb = [gp[cfg.GPW - 1 - k] for k in range(cfg.GPW)] inc, c = [], "#1" for k in range(cfg.GPW): inc.append(net.XOR(f"gpi_x{k}", gp_lsb[k], c)) c = net.AND(f"gpi_c{k}", [gp_lsb[k], c]) inc_msb = [inc[cfg.GPW - 1 - k] for k in range(cfg.GPW)] wrap = [net.MUX(f"gpw{k}", gp_oh[cfg.G - 1], "#0", inc_msb[k]) for k in range(cfg.GPW)] ngp = [net.MUX(f"ngp{k}", nhalt, wrap[k], gp[k]) for k in range(cfg.GPW)] nhalt_out = net.OR("halt_next", [halt, sig[cfg.HALT_SIG]]) return net, sig + gp + [halt], nsig + ngp + [nhalt_out] # ============================================================================= # leveled sparse evaluator (scales past dense matrices) # ============================================================================= class Leveled: def __init__(self, net, inputs, outputs, device="cpu"): self.device = device self.inputs = inputs 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, d in indeg.items() if d == 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 assert len(order) == len(gates), "cycle" names = ["#0", "#1"] + inputs + order slot = {n: i for i, n in enumerate(names)} depth = {n: 0 for n in names if n not in gates} for g in order: depth[g] = 1 + max((depth[s] for s, _ in gates[g][0]), default=0) by_level: Dict[int, List[str]] = {} for g in order: by_level.setdefault(depth[g], []).append(g) self.n_sig = len(names) self.plan = [] for lv in sorted(by_level): gs = by_level[lv] fan = max(len(gates[g][0]) for g in gs) idx = torch.zeros(len(gs), fan, dtype=torch.long) w = torch.zeros(len(gs), fan) b = torch.zeros(len(gs)) out = torch.zeros(len(gs), dtype=torch.long) for r, g in enumerate(gs): ins, bias = gates[g] out[r] = slot[g]; b[r] = bias for c, (s, wt) in enumerate(ins): idx[r, c] = slot[s]; w[r, c] = wt self.plan.append((idx.to(device), w.to(device), b.to(device), out.to(device))) self.in_slots = torch.tensor([slot[n] for n in inputs], device=device) self.out_slots = torch.tensor([slot[n] for n in outputs], device=device) def step(self, invec): B = invec.shape[0] V = torch.zeros(self.n_sig, B, device=self.device) V[1] = 1.0 V[self.in_slots] = invec.T for idx, w, b, out in self.plan: g = V[idx] V[out] = ((g * w[:, :, None]).sum(1) + b[:, None] >= 0).float() return V[self.out_slots].T def state_to_vec(cfg, st): v = torch.zeros(cfg.STATE_BITS) v[:cfg.S] = torch.tensor(st["sig"], dtype=torch.float32) v[cfg.S:cfg.S + cfg.GPW] = torch.tensor(to_bits(st["gp"], cfg.GPW), dtype=torch.float32) v[cfg.S + cfg.GPW] = float(st["halt"]) return v def vec_to_state(cfg, v): b = [int(round(float(x))) for x in (v.tolist() if hasattr(v, "tolist") else v)] return {"sig": b[:cfg.S], "gp": from_bits(b[cfg.S:cfg.S + cfg.GPW]), "halt": b[cfg.S + cfg.GPW]} class Runner: def __init__(self, cfg, lev): self.cfg = cfg self.lev = lev def step(self, st): halt_in = st["halt"] v = state_to_vec(self.cfg, st).unsqueeze(0).to(self.lev.device) out = self.lev.step(v[:, :len(self.lev.inputs)])[0].cpu() ns = vec_to_state(self.cfg, out) ns["emit"] = apply_device(self.cfg, ns["sig"]) if halt_in == 0 else None return ns def run(self, st, n, collect=False): emits = [] for _ in range(n): st = self.step(st) if collect and st.get("emit") is not None: emits.append(st["emit"]) return (st, emits) if collect else st # ============================================================================= # compile a <=2-input family Net into a stored reflect netlist # ============================================================================= def topo(net): 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, d in indeg.items() if d == 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 assert len(order) == len(gates) return order def eval_family_net(net, input_vals): """Evaluate a matrix8.Net directly (the family's own gate semantics).""" v = dict(input_vals); v["#0"], v["#1"] = 0, 1 for g in topo(net): ins, b = net.gates[g] v[g] = 1 if b + sum(w * v[s] for s, w in ins) >= 0 else 0 return v def record_bit(cfg, bank, gate, off): return cfg.bank_base(bank) + gate * cfg.R + off def compile_to_reflect(cfg, net, in_addr, out_names, work_start, fixed=None): """Lay a <=2-input acyclic Net out as reflect gate records in topological order. Primary inputs sit at in_addr; gate outputs get fresh work signals from work_start up, except those named in `fixed` (gate -> signal), which are placed at the given signal. Returns (records, address map, next work).""" fixed = fixed or {} addr = dict(in_addr) w = work_start rgates = [] for g in topo(net): ins, bias = net.gates[g] assert len(ins) <= cfg.F if g in fixed: addr[g] = fixed[g] else: addr[g] = w; w += 1 slots = [(addr[s], wt, 0) for s, wt in ins] rgates.append((slots, bias, (addr[g], 0))) return rgates, addr, w def adder_net(bits): """N-bit ripple-carry adder from the family's full-adder cell.""" net = Net() a = [f"a{i}" for i in range(bits)] b = [f"b{i}" for i in range(bits)] carry = "#0" sums = [] for i in range(bits): s_, carry = net.FA(f"fa{i}", a[i], b[i], carry) sums.append(s_) return net, a, b, sums, carry def subleq_datapath_net(): """The SUBLEQ machine's datapath as <=2-input threshold gates: result = M[B] - M[A], the branch decision leq = (result <= 0), and next PC = leq ? C : PC + 3. All inputs LSB-first. (This is the combinational core the family verifies exhaustively; the 256-byte packed memory is not included.)""" net = Net() a = [f"a{k}" for k in range(8)] b = [f"b{k}" for k in range(8)] pc = [f"pc{k}" for k in range(8)] c = [f"c{k}" for k in range(8)] nota = [net.NOT(f"nota{k}", a[k]) for k in range(8)] carry = "#1" # two's-complement subtract r = [] for k in range(8): s_, carry = net.FA(f"sub{k}", b[k], nota[k], carry) r.append(s_) ortree = r[0] # NOR-8 as an OR-tree + NOT for k in range(1, 8): ortree = net.OR(f"zor{k}", [ortree, r[k]]) zero = net.NOT("zero", ortree) leq = net.OR("leq", [r[7], zero]) # sign OR zero carry = "#0" p3 = [] for k in range(8): s_, carry = net.FA(f"pci{k}", pc[k], "#1" if k in (0, 1) else "#0", carry) p3.append(s_) pcm = [net.MUX(f"pcm{k}", leq, c[k], p3[k]) for k in range(8)] return net, a, b, pc, c, r, leq, pcm # ============================================================================= # programs (stored netlists) for the reflective demos # ============================================================================= def quine_program(cfg): # emit sig[NET0 + ptr], strobe, advance ptr return [([(cfg.NET0, 1, 1)], -1, (cfg.OUT_DATA, 0)), # OUT_DATA = NET0[ptr] ([], 0, (cfg.OUT_STROBE, 0)), # OUT_STROBE = 1 ([], 0, (cfg.PTR_ADV, 0))] # PTR_ADV = 1 # ============================================================================= # build + save (compiles U to a recurrent ternary matrix stack) # ============================================================================= def build(cfg, save=True): print(f"Interpreter U (S={cfg.S}, G={cfg.G}, F={cfg.F}, banks={cfg.banks}, " f"record={cfg.R}b, acc={cfg.ACC}b)") t0 = time.perf_counter() net, inputs, outputs = build_net(cfg) print(f" {len(net.gates):,} threshold gates in {time.perf_counter() - t0:.1f}s") if not save: return net, inputs, outputs, None t1 = time.perf_counter() layers, info = compile_net(net, inputs, outputs) for Wt, _ in layers: assert set(torch.unique(Wt).tolist()) <= {-1, 0, 1} tensors = {f"matrix.layer{i:03d}.weight": Wt for i, (Wt, _) in enumerate(layers)} tensors.update({f"matrix.layer{i:03d}.bias": B for i, (_, B) in enumerate(layers)}) for k, v in (("signals", cfg.S), ("gates", cfg.G), ("banks", cfg.banks), ("state_bits", cfg.STATE_BITS), ("layers", info["layers"])): tensors[f"manifest.{k}"] = torch.tensor([float(v)]) meta = {"machine": "reflect", "weight_quantization": "ternary", "config": json.dumps({"A": cfg.A, "S": cfg.S, "G": cfg.G, "F": cfg.F, "BB": cfg.BB, "banks": cfg.banks, "record_bits": cfg.R, "acc_bits": cfg.ACC, "net_base": cfg.NET0})} save_file(tensors, MODEL_PATH, metadata=meta) print(f" compiled to {info['layers']} ternary matrices (max width " f"{info['max_width']}, {info['total_weights']:,} weights) in " f"{time.perf_counter() - t1:.1f}s; saved {MODEL_PATH} " f"({os.path.getsize(MODEL_PATH) / 1e6:.1f} MB)") return net, inputs, outputs, layers # ============================================================================= # verify # ============================================================================= def _load(cfg, netlist, bank=0): sig = [0] * cfg.S base = cfg.bank_base(bank) sig[base:base + len(netlist)] = netlist return sig def verify(cfg, device): ok = True net, inputs, outputs = build_net(cfg) lev = Leveled(net, inputs, outputs, device=device) R = Runner(cfg, lev) print("[1/7] Exhaustive single-gate semantics") bad = total = 0 ra, rb, ro = cfg.WORK_BASE, cfg.WORK_BASE + 1, cfg.WORK_BASE + 2 for w0 in (-1, 0, 1): for w1 in (-1, 0, 1): for bias in range(-(1 << (cfg.BB - 1)), 1 << (cfg.BB - 1)): nl = encode_netlist(cfg, [([(ra, w0, 0), (rb, w1, 0)], bias, (ro, 0))]) for va in (0, 1): for vb in (0, 1): sig = _load(cfg, nl); sig[ra], sig[rb] = va, vb st = {"sig": sig, "gp": 0, "halt": 0} g = R.step(st); e = ref_step(cfg, st) total += 1 exp = 1 if w0 * va + w1 * vb + bias >= 0 else 0 if {k: g[k] for k in ("sig", "gp", "halt")} != {k: e[k] for k in ("sig", "gp", "halt")} or g["sig"][ro] != exp: bad += 1 print(f" {'OK ' if bad == 0 else 'FAIL'} {total} configurations exact") ok &= bad == 0 print("[2/7] Single step vs reference on random full states") gen = torch.Generator().manual_seed(0x5EED) N = 1500 sig = (torch.rand(N, cfg.S, generator=gen) < 0.5).long() gpv = torch.randint(0, cfg.G, (N,), generator=gen) bad = 0 for i in range(N): st = {"sig": sig[i].tolist(), "gp": int(gpv[i]), "halt": 0} g = R.step(st); e = ref_step(cfg, st) if {k: g[k] for k in ("sig", "gp", "halt")} != {k: e[k] for k in ("sig", "gp", "halt")}: bad += 1 hbad = sum(1 for i in range(200) if R.step({"sig": sig[i].tolist(), "gp": int(gpv[i]), "halt": 1})["sig"] != sig[i].tolist()) print(f" {'OK ' if bad == 0 else 'FAIL'} {N} random states exact; " f"{'OK ' if hbad == 0 else 'FAIL'} halted states are fixed points") ok &= bad == 0 and hbad == 0 print("[3/7] Universality: interpret a family full-adder cell") anet, an, bn, sums, cout = adder_net(1) ia = {an[0]: cfg.WORK_BASE, bn[0]: cfg.WORK_BASE + 1} rg, addr, _ = compile_to_reflect(cfg,anet, ia, sums + [cout], cfg.WORK_BASE + 2) nl = encode_netlist(cfg, rg) abad = 0 for x in (0, 1): for y in (0, 1): sig = _load(cfg, nl) sig[ia[an[0]]] = x; sig[ia[bn[0]]] = y st = R.run({"sig": sig, "gp": 0, "halt": 0}, cfg.G * cfg.G) fam = eval_family_net(anet, {an[0]: x, bn[0]: y}) got = {g: st["sig"][addr[g]] for g in [sums[0], cout]} s = got[sums[0]] | (got[cout] << 1) if s != x + y or got[sums[0]] != fam[sums[0]] or got[cout] != fam[cout]: abad += 1 print(f" {'OK ' if abad == 0 else 'FAIL'} full adder ({len(rg)} gates, compiled " f"from the family's Net) matches arithmetic and the family's own evaluator") ok &= abad == 0 print("[4/7] Order independence (adder gates stored in random order)") obad = 0 for seed in range(4): perm = list(range(len(rg))) random.Random(seed + 1).shuffle(perm) nls = encode_netlist(cfg, [rg[p] for p in perm]) for x in (0, 1): for y in (0, 1): sig = _load(cfg, nls) sig[ia[an[0]]] = x; sig[ia[bn[0]]] = y st = R.run({"sig": sig, "gp": 0, "halt": 0}, cfg.G * cfg.G) s = st["sig"][addr[sums[0]]] | (st["sig"][addr[cout]] << 1) if s != x + y: obad += 1 print(f" {'OK ' if obad == 0 else 'FAIL'} adder correct under 4 random gate " f"orders (relaxation reproduces combinational evaluation)") ok &= obad == 0 print("[5/7] Self-reproduction (quine): program emits its whole encoding") nlq = encode_netlist(cfg, quine_program(cfg)) K = len(nlq) st = {"sig": _load(cfg, nlq), "gp": 0, "halt": 0} _, emits = R.run(st, cfg.G * K + cfg.G, collect=True) good = emits[:K] == nlq print(f" {'OK ' if good else 'FAIL'} all {K} emitted bits == the program's own " f"stored encoding (streaming its full description out the output device)") ok &= good print("[6/7] Metamorphosis: a program rewrites its own function, and bank-switches") IA, IB, OUT = cfg.WORK_BASE, cfg.WORK_BASE + 1, cfg.WORK_BASE + 2 w1hi = slot_off(cfg, 1) + cfg.A bo, _ = field_off(cfg) prog = [([(IA, 1, 0), (IB, 1, 0)], -2, (OUT, 0)), # AND(a, b) ([], 0, (record_bit(cfg, 0, 0, w1hi), 0)), # slot1 weight hi = 1 ([], -1, (record_bit(cfg, 0, 0, w1hi + 1), 0)), # slot1 weight lo = 0 ([], 0, (record_bit(cfg, 0, 0, bo + cfg.BB - 1), 0))] # bias lsb = 1 (-2 -> -1) nlm = encode_netlist(cfg, prog) mbad = 0 for a in (0, 1): for b in (0, 1): sig = _load(cfg, nlm); sig[IA] = a; sig[IB] = b st = R.run({"sig": sig, "gp": 0, "halt": 0}, cfg.G) # sweep 1: AND f = st["sig"][OUT] st = R.run(st, cfg.G) # sweep 2: rewritten g = st["sig"][OUT] if f != (a & b) or g != (a & (1 - b)): mbad += 1 # program-set BANK flag switches the interpreter to a different bank-1 machine nl_sw = encode_netlist(cfg, [([], 0, (cfg.BANK_SIG, 0))]) Bnl = encode_netlist(cfg, [([(IA, 1, 0), (IB, 1, 0)], -2, (OUT, 0))]) # bank1 = AND sig = _load(cfg, nl_sw, bank=0); sig[cfg.NET1:cfg.NET1 + len(Bnl)] = Bnl st = R.run({"sig": sig, "gp": 0, "halt": 0}, 1) # gp0 sets BANK=1 st["gp"] = 0; st["sig"][IA] = 1; st["sig"][IB] = 1 st = R.run(st, cfg.G) # now running bank1 switched = st["sig"][cfg.BANK_SIG] == 1 and st["sig"][OUT] == 1 print(f" {'OK ' if mbad == 0 else 'FAIL'} one program computes a&b, then edits its " f"own gate so the next sweep computes a&~b; " f"{'OK ' if switched else 'FAIL'} program-set BANK runs the bank-1 machine") ok &= mbad == 0 and switched print("[7/7] Sequential state: interpret a 2-bit counter with feedback") C0, C1, T, OR_, NAND_ = (cfg.WORK_BASE, cfg.WORK_BASE + 1, cfg.WORK_BASE + 2, cfg.WORK_BASE + 3, cfg.WORK_BASE + 4) counter = [([(C0, 1, 0)], -1, (T, 0)), # T = c0 (old) ([(C1, 1, 0), (T, 1, 0)], -1, (OR_, 0)), # c1 OR T ([(C1, -1, 0), (T, -1, 0)], 1, (NAND_, 0)), # c1 NAND T ([(OR_, 1, 0), (NAND_, 1, 0)], -2, (C1, 0)), # c1' = c1 XOR c0 ([(C0, -1, 0)], 0, (C0, 0))] # c0' = NOT c0 nlk = encode_netlist(cfg, counter) st = {"sig": _load(cfg, nlk), "gp": 0, "halt": 0} seq = [] for _ in range(5): # 5 ticks (5 sweeps) st = R.run(st, cfg.G) seq.append(st["sig"][C1] * 2 + st["sig"][C0]) kgood = seq == [1, 2, 3, 0, 1] print(f" {'OK ' if kgood else 'FAIL'} register advances 0->1->2->3->0 across sweeps " f"(a stored machine holding state through feedback): {seq}") ok &= kgood print("\nneural_reflect (universality, order, quine, metamorphosis, sequential state):", "PASS" if ok else "FAIL") return ok def verify_scale(device): cfg = Cfg(A=11, G=40, banks=1) print(f"\nScale: interpret a 4-bit ripple-carry adder (S={cfg.S}, G={cfg.G})") net, inputs, outputs = build_net(cfg) print(f" interpreter U: {len(net.gates):,} gates") R = Runner(cfg, Leveled(net, inputs, outputs, device=device)) W = 4 anet, an, bn, sums, cout = adder_net(W) ia = {} for i in range(W): ia[an[i]] = cfg.WORK_BASE + i ia[bn[i]] = cfg.WORK_BASE + W + i rg, addr, _ = compile_to_reflect(cfg,anet, ia, sums + [cout], cfg.WORK_BASE + 2 * W) nl = encode_netlist(cfg, rg) print(f" compiled adder netlist: {len(rg)} gates") def result(st): return sum(st["sig"][addr[sums[i]]] << i for i in range(W)) | (st["sig"][addr[cout]] << W) bad = 0 pairs = [(x, y) for x in range(1 << W) for y in range(1 << W)] for (x, y) in pairs: sig = _load(cfg, nl) for i in range(W): sig[ia[an[i]]] = (x >> i) & 1 sig[ia[bn[i]]] = (y >> i) & 1 st = R.run({"sig": sig, "gp": 0, "halt": 0}, cfg.G) # one sweep (topo order) fam = eval_family_net(anet, {an[i]: (x >> i) & 1 for i in range(W)} | {bn[i]: (y >> i) & 1 for i in range(W)}) famres = sum(fam[sums[i]] << i for i in range(W)) | (fam[cout] << W) if result(st) != x + y or result(st) != famres: bad += 1 print(f" {'OK ' if bad == 0 else 'FAIL'} all {len(pairs)} input pairs match " f"arithmetic and the family's own evaluator") obad = 0 for seed in range(2): perm = list(range(len(rg))) random.Random(seed + 1).shuffle(perm) nls = encode_netlist(cfg, [rg[p] for p in perm]) for (x, y) in [(7, 9), (15, 15), (10, 6)]: sig = _load(cfg, nls) for i in range(W): sig[ia[an[i]]] = (x >> i) & 1 sig[ia[bn[i]]] = (y >> i) & 1 st = R.run({"sig": sig, "gp": 0, "halt": 0}, cfg.G * cfg.G) if result(st) != x + y: obad += 1 print(f" {'OK ' if obad == 0 else 'FAIL'} correct under 2 random gate orders (relaxation)") good = bad == 0 and obad == 0 print("SCALE:", "PASS" if good else "FAIL") return good def verify_subleq(device): cfg = Cfg(A=14, G=304, banks=1, self_mod=False) net, a, b, pc, c, r, leq, pcm = subleq_datapath_net() print(f"\nFamily machine: interpret the SUBLEQ datapath " f"(S={cfg.S}, addressed span={cfg.span}, G={cfg.G})") inet, ii, io = build_net(cfg) print(f" interpreter U: {len(inet.gates):,} gates") lev = Leveled(inet, ii, io, device=device) base = cfg.WORK_BASE in_addr = {} for k in range(8): in_addr[a[k]] = base + k in_addr[b[k]] = base + 8 + k in_addr[pc[k]] = base + 16 + k in_addr[c[k]] = base + 24 + k rg, addr, _ = compile_to_reflect(cfg,net, in_addr, r + [leq] + pcm, base + 32) nl = encode_netlist(cfg, rg) print(f" compiled datapath netlist: {len(rg)} gates; interpreting all " f"65,536 (M[A], M[B]) pairs") sig0 = [0] * cfg.S sig0[cfg.NET0:cfg.NET0 + len(nl)] = nl base_vec = state_to_vec(cfg, {"sig": sig0, "gp": 0, "halt": 0}).to(device) PC, C = 0x42, 0x99 bad = 0 t0 = time.perf_counter() for lo in range(0, 65536, 8192): xs = torch.arange(lo, lo + 8192) % 256 # M[A] ys = torch.arange(lo, lo + 8192) // 256 # M[B] B = xs.shape[0] V = base_vec.unsqueeze(0).repeat(B, 1) for k in range(8): V[:, in_addr[a[k]]] = ((xs >> k) & 1).float().to(device) V[:, in_addr[b[k]]] = ((ys >> k) & 1).float().to(device) V[:, in_addr[pc[k]]] = float((PC >> k) & 1) V[:, in_addr[c[k]]] = float((C >> k) & 1) for _ in range(cfg.G): V = lev.step(V) Vc = V.cpu() rr = sum(Vc[:, addr[r[k]]].long() << k for k in range(8)) lq = Vc[:, addr[leq]].long() npc = sum(Vc[:, addr[pcm[k]]].long() << k for k in range(8)) exp_r = (ys - xs) & 0xFF exp_lq = ((exp_r == 0) | (exp_r >= 128)).long() exp_np = torch.where(exp_lq.bool(), torch.full_like(xs, C), (PC + 3) & 0xFF) bad += int((rr != exp_r).sum() + (lq != exp_lq).sum() + (npc != exp_np).sum()) dt = time.perf_counter() - t0 good = bad == 0 print(f" {'OK ' if good else 'FAIL'} result, branch decision, and next-PC exact " f"for all 65,536 pairs ({dt:.0f}s)") print("SUBLEQ DATAPATH:", "PASS" if good else "FAIL") return good # ============================================================================= # a complete stored machine (SUBLEQ, 32-byte memory) hosted in the interpreter # ============================================================================= def _subtract_net(): net = Net() x = [f"x{k}" for k in range(8)]; y = [f"y{k}" for k in range(8)] nx = [net.NOT(f"nx{k}", x[k]) for k in range(8)] carry = "#1"; r = [] for k in range(8): s_, carry = net.FA(f"sb{k}", y[k], nx[k], carry) r.append(s_) ot = r[0] for k in range(1, 8): ot = net.OR(f"zo{k}", [ot, r[k]]) leq = net.OR("leq", [r[7], net.NOT("z", ot)]) return net, x, y, r, leq def _incr_net(): # 5-bit address arithmetic (mod 32) net = Net() pc = [f"p{k}" for k in range(5)] def addc(tag, cst): carry = "#0"; out = [] for k in range(5): s_, carry = net.FA(f"{tag}{k}", pc[k], "#1" if (cst >> k) & 1 else "#0", carry) out.append(s_) return out return net, pc, addc("i1", 1), addc("i2", 2), addc("i3", 3) def _branch_net(): net = Net() c = [f"c{k}" for k in range(5)]; q = [f"q{k}" for k in range(5)] npc = [net.MUX(f"nb{k}", "leq", c[k], q[k]) for k in range(5)] halt = npc[0] for k in range(1, 5): halt = net.AND(f"hb{k}", [halt, npc[k]]) return net, c, q, npc, halt MEMB = 32 MEM = 416 def build_subleq_machine(cfg): """A stored microprogram implementing one SUBLEQ instruction per sweep over a 32-byte memory held in signals [MEM, MEM+8*32). Data-dependent memory access is the interpreter's own indexed addressing: set PTR to a byte address, then read/write the eight bit-planes at MEM + b*32 + PTR.""" WB = cfg.WORK_BASE reg = lambda o: [WB + o + k for k in range(8)] PC, A_, B_, C_ = reg(0), reg(8), reg(16), reg(24) X, Y, Rr = reg(32), reg(40), reg(48) P1, P2, P3, NPC = reg(56), reg(64), reg(72), reg(80) leqs = WB + 88 workc = WB + 96 def zero_hi(): # PTR high bits = 0 (once) return [([], -1, (cfg.PTR_BASE + (cfg.A - 1 - k), 0)) for k in range(5, cfg.A)] def ptr_set(r8): # PTR low 5 = addr; high stays 0 return [([(r8[k], 1, 0)], -1, (cfg.PTR_BASE + (cfg.A - 1 - k), 0)) for k in range(5)] def read_byte(dst): return [([(MEM + b * MEMB, 1, 1)], -1, (dst[b], 0)) for b in range(8)] def write_byte(src): return [([(src[b], 1, 0)], -1, (MEM + b * MEMB, 1)) for b in range(8)] inet, pin, p1, p2, p3 = _incr_net() irec, _, workc = compile_to_reflect(cfg, inet, {pin[k]: PC[k] for k in range(5)}, [], workc, fixed={**{p1[k]: P1[k] for k in range(5)}, **{p2[k]: P2[k] for k in range(5)}, **{p3[k]: P3[k] for k in range(5)}}) snet, sx, sy, sr, sleq = _subtract_net() srec, _, workc = compile_to_reflect( cfg, snet, {**{sx[k]: X[k] for k in range(8)}, **{sy[k]: Y[k] for k in range(8)}}, [], workc, fixed={**{sr[k]: Rr[k] for k in range(8)}, sleq: leqs}) bnet, bc, bq, bnpc, bhalt = _branch_net() brec, _, workc = compile_to_reflect( cfg, bnet, {**{bc[k]: C_[k] for k in range(5)}, **{bq[k]: P3[k] for k in range(5)}, "leq": leqs}, [], workc, fixed={**{bnpc[k]: NPC[k] for k in range(5)}, bhalt: cfg.HALT_SIG}) assert workc < MEM, "compiled internals collide with memory" prog = zero_hi() + list(irec) prog += ptr_set(PC) + read_byte(A_) prog += ptr_set(P1) + read_byte(B_) prog += ptr_set(P2) + read_byte(C_) prog += ptr_set(A_) + read_byte(X) prog += ptr_set(B_) + read_byte(Y) prog += srec prog += ptr_set(B_) + write_byte(Rr) prog += brec prog += [([(NPC[k], 1, 0)], -1, (PC[k], 0)) for k in range(5)] # PC = NPC (low 5) return prog, {"PC": PC} def subleq32_step(mem, pc): A, B, C = mem[pc], mem[(pc + 1) & 31], mem[(pc + 2) & 31] x, y = mem[A & 31], mem[B & 31] r = (y - x) & 0xFF m2 = list(mem); m2[B & 31] = r npc = (C & 31) if (r == 0 or r >= 128) else (pc + 3) & 31 return m2, npc def verify_hosted(device): cfg = Cfg(A=14, G=296, banks=1, self_mod=False) prog, lay = build_subleq_machine(cfg) print(f"\nHosted machine: SUBLEQ with a 32-byte memory, one instruction per " f"sweep\n microprogram: {len(prog)} gates; interpreter U:", end=" ") inet, ii, io = build_net(cfg) print(f"{len(inet.gates):,} gates (addressed span {cfg.span})") lev = Leveled(inet, ii, io, device=device) nl = encode_netlist(cfg, prog) sig0 = [0] * cfg.S sig0[cfg.NET0:cfg.NET0 + len(nl)] = nl base = state_to_vec(cfg, {"sig": sig0, "gp": 0, "halt": 0}).to(device) PC = lay["PC"] def set_mem(V, mems): for i in range(32): for b in range(8): V[:, MEM + b * MEMB + i] = ((mems[:, i] >> b) & 1).float().to(device) def get_mem(Vc): return sum(Vc[:, MEM + b * MEMB: MEM + b * MEMB + 32].long() << b for b in range(8)) # one instruction vs reference over random (memory, PC) states, batched gen = torch.Generator().manual_seed(1) B = 4096 mems = torch.randint(0, 256, (B, 32), generator=gen) pcs = torch.randint(0, 30, (B,), generator=gen) # avoid the halt cell V = base.unsqueeze(0).repeat(B, 1) set_mem(V, mems) for k in range(8): V[:, PC[k]] = ((pcs >> k) & 1).float().to(device) for _ in range(cfg.G): V = lev.step(V) Vc = V.cpu() got_mem = get_mem(Vc) got_pc = sum(Vc[:, PC[k]].long() << k for k in range(8)) & 31 exp_mem = torch.zeros_like(got_mem); exp_pc = torch.zeros_like(got_pc) for i in range(B): m2, np_ = subleq32_step(mems[i].tolist(), int(pcs[i])) exp_mem[i] = torch.tensor(m2); exp_pc[i] = np_ mask = exp_pc != 31 # halting instr freezes PC ok1 = bool((got_mem == exp_mem).all()) and bool((got_pc[mask] == exp_pc[mask]).all()) print(f" {'OK ' if ok1 else 'FAIL'} one instruction exact for {B} random " f"(memory, PC) states (result, writeback, and branch)") # run a stored program to a halt: count M[21] (=3) down to 0 through a loop prog_mem = [0] * 32 prog_mem[0:3] = [20, 21, 6] # loop: subleq one, n, done prog_mem[3:6] = [22, 22, 0] # subleq z, z, 0 (jump back to loop) prog_mem[6:9] = [31, 31, 31] # done: branch to the halt cell prog_mem[20], prog_mem[21], prog_mem[22] = 1, 3, 0 V = base.unsqueeze(0).repeat(1, 1) set_mem(V, torch.tensor(prog_mem).unsqueeze(0)) halted = False steps = 0 for _ in range(24): for _ in range(cfg.G): V = lev.step(V) steps += 1 if int(V[0, cfg.S + cfg.GPW]) == 1: # interpreter halt latched halted = True break got = get_mem(V.cpu())[0].tolist() ref_mem, ref_pc = list(prog_mem), 0 while ref_pc != 31: ref_mem, ref_pc = subleq32_step(ref_mem, ref_pc) ok2 = halted and got == ref_mem and got[21] == 0 print(f" {'OK ' if ok2 else 'FAIL'} stored countdown program ran to a halt in " f"{steps} instructions; final memory matches the emulator (M[21] = {got[21]})") good = ok1 and ok2 print("HOSTED MACHINE:", "PASS" if good else "FAIL") return good # ============================================================================= # analog: matrix == gate + noise / conductance-mismatch margins # ============================================================================= def analog(cfg, device): ok = True net, inputs, outputs = build_net(cfg) print("Compiling U to a recurrent ternary matrix stack...") layers, info = compile_net(net, inputs, outputs) print(f" {info['layers']} matrices, max width {info['max_width']}, " f"{info['total_weights']:,} weights") Wm = [W.to(device=device, dtype=torch.float32) for W, _ in layers] Bm = [B.to(device=device, dtype=torch.float32) for _, B in layers] lev = Leveled(net, inputs, outputs, device=device) def mstep(v, thresh=0.0, noise=0.0, gen=None, Wover=None): for W, b in zip(Wover or Wm, Bm): pre = v @ W.T + b if noise: pre = pre + torch.randn(pre.shape, generator=gen, device=v.device) * noise v = (pre >= thresh).float() return v gen = torch.Generator().manual_seed(11) ns = 300 V = torch.zeros(ns, cfg.STATE_BITS) V[:, :cfg.S] = (torch.rand(ns, cfg.S, generator=gen) < 0.5).float() gpv = torch.randint(0, cfg.G, (ns,), generator=gen) for i in range(ns): V[i, cfg.S:cfg.S + cfg.GPW] = torch.tensor(to_bits(int(gpv[i]), cfg.GPW), dtype=torch.float32) V = V.to(device) same = bool((mstep(V) == lev.step(V[:, :len(inputs)])).all()) print(f" {'OK ' if same else 'FAIL'} matrix step == gate-graph step on {ns} random states") ok &= same m = float("inf"); v = V.clone() for W, b in zip(Wm, Bm): pre = v @ W.T + b m = min(m, float((pre + 0.5).abs().min())) v = (pre >= 0).float() print(f" measured minimum |pre-activation - (-0.5)|: {m:.3f} (guarantee 0.5)") ok &= abs(m - 0.5) < 1e-6 base = state_to_vec(cfg, {"sig": _load(cfg, encode_netlist(cfg, quine_program(cfg))), "gp": 0, "halt": 0}).unsqueeze(0).to(device) def sweep(noise=0.0, gen=None, Wover=None): v = base.clone() for _ in range(cfg.G): v = mstep(v, thresh=-0.5, noise=noise, gen=gen, Wover=Wover) return v ref = sweep() print(" read noise per MVM (20 trials, full sweep bit-exact):") for sigma in (0.05, 0.10, 0.20, 0.40): good = sum(int(bool((sweep(noise=sigma, gen=torch.Generator(device=device).manual_seed(t)) == ref).all())) for t in range(20)) print(f" sigma={sigma:.2f}: {good}/20 bit-exact") if sigma <= 0.05 and good != 20: # deep within the 0.5 margin ok = False print(" conductance mismatch (8 instances, full sweep bit-exact):") for sg in (0.05, 0.20): good = 0 for t in range(8): gg = torch.Generator().manual_seed(7 + t) Wp = [W + (torch.randn(W.shape, generator=gg) * sg).to(device) * (W != 0) for W in Wm] good += int(bool((sweep(Wover=Wp) == ref).all())) print(f" sigma_G={sg:.2f}: {good}/8 bit-exact") if sg <= 0.05 and good != 8: ok = False print("ANALOG (matrix == gate; bit-exact within the 0.5 margin):", "PASS" if ok else "FAIL") return ok def main(): ap = argparse.ArgumentParser(description="neural_reflect") ap.add_argument("cmd", choices=["build", "verify", "analog", "all"]) ap.add_argument("--device", default="cuda" if torch.cuda.is_available() else "cpu") args = ap.parse_args() cfg = Cfg() rc = 0 if args.cmd in ("build", "all"): build(cfg, save=True) if args.cmd in ("verify", "all"): rc |= 0 if verify(cfg, args.device) else 1 rc |= 0 if verify_scale(args.device) else 1 rc |= 0 if verify_subleq(args.device) else 1 rc |= 0 if verify_hosted(args.device) else 1 if args.cmd in ("analog", "all"): rc |= 0 if analog(cfg, args.device) else 1 return rc if __name__ == "__main__": sys.exit(main())