"""neural_subleq8io + the universal constructor. The host is the family's one-instruction machine extended with three memory-mapped device cells (all device logic lives in the runtimes, exactly like neural_rv32's console at 0xFF00; the processor's gates are unchanged): 0xFD IN_DATA reads as the current tape byte (device-refreshed) 0xFC IN_STROBE any write advances the tape head (device clears to 0) 0xFE OUT any write emits the written byte (device clears to 0) A 21-instruction SUBLEQ program — the constructor — interprets a construction recipe streamed on the tape: tag 0 END tag 1..127 literal record: the next `tag` tape bytes, each stored negated mod 256, are emitted one per SUBLEQ instruction (M[0xFE] = 0 - M[0xFD] recovers the byte) tag 129..255 repeat record: the next tape byte (negated value) is emitted 256-tag times without strobing (tag 128 unused) `describe(file_bytes)` compiles any safetensors file in the family into a recipe; the constructor run on that recipe emits the file byte for byte, so the constructor is universal and the description selects what is built. Self-reproduction is the case where the description handed to the constructor is that of the running machine: the program's output stream is the byte-for-byte safetensors file of the host itself. The equality is machine-checked rather than observed on a single run: - the recipe codec round-trips every family file (byte equality + sha256); - the constructor program is executed on three proven-equal backends (pure-integer reference, the gate-graph SubleqThresholdCPU walking the shipped netlist, and the host compiled to recurrent ternary matrices with matrix8's compiler) with cycle-lockstep spot checks between them; - the full self-reproduction runs at threshold fidelity on the matrix backend, and the emitted stream must equal the host's file exactly; - the offspring file then boots as generation 2 and reproduces again. Usage: python constructor8.py build # emit variants/neural_subleq8io.safetensors python constructor8.py verify # host soundness + codec + constructor runs python constructor8.py self # full self-reproduction at threshold fidelity python constructor8.py all """ from __future__ import annotations import argparse import hashlib import json import os import sys import time from typing import Dict, List, Optional, Tuple import torch from safetensors import safe_open from safetensors.torch import save_file sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) 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") from matrix8 import Net, compile_net # machine-1 compiler, reused verbatim IN_DATA, IN_STROBE, OUT = 0xFD, 0xFC, 0xFE HALT_PC = 0xFF # ============================================================================= # Build the host file (family per-gate style, .inputs wiring, ternary strict) # ============================================================================= def build_host(save: bool = True) -> str: import build as B from quantize 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 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 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) # ============================================================================= Z, ONE, T1, T2 = 0xF0, 0xF1, 0xF2, 0xF3 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) # ============================================================================= 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]]: net = Net() pc = [f"pc{i}" for i in range(8)] halt = "halt" mem = [[f"m{j}b{i}" for i in range(8)] for j in range(256)] P = [net.DECODE(f"pd{j}", pc, j) for j in range(256)] byte = [] for o in range(3): # A, B, C operand bytes bo = [net.OR(f"f{o}b{bit}", [net.AND(f"f{o}a{j}b{bit}", [mem[j][bit], P[(j - o) % 256]]) for j in range(256)]) for bit in range(8)] byte.append(bo) A, Bb, C = byte AD = [net.DECODE(f"ad{j}", A, j) for j in range(256)] BD = [net.DECODE(f"bd{j}", Bb, j) for j in range(256)] x = [net.OR(f"x{bit}", [net.AND(f"x{bit}a{j}", [mem[j][bit], AD[j]]) for j in range(256)]) for bit in range(8)] y = [net.OR(f"y{bit}", [net.AND(f"y{bit}a{j}", [mem[j][bit], BD[j]]) for j in range(256)]) for bit in range(8)] nx = [net.NOT(f"nx{k}", x[7 - k]) for k in range(8)] # LSB-first c = "#1" r_l = [] for k in range(8): s_, c = net.FA(f"sub.fa{k}", y[7 - k], nx[k], c) r_l.append(s_) r = [r_l[7 - i] for i in range(8)] zero = net.NOR("rzero", r) leq = net.OR("leq", [r[0], zero]) pc_l = [pc[7 - k] for k in range(8)] c = "#0" p3_l = [] for k in range(8): addend = "#1" if k in (0, 1) else "#0" s_, c = net.FA(f"pc3.fa{k}", pc_l[k], addend, c) p3_l.append(s_) p3 = [p3_l[7 - i] for i in range(8)] pcm = [net.MUX(f"pcm{i}", leq, C[i], p3[i]) for i in range(8)] pcn = [net.MUX(f"pcn{i}", halt, pc[i], pcm[i]) for i in range(8)] hdec = net.DECODE("hdec", pcn, HALT_PC) halt_n = net.OR("haltn", [halt, hdec]) nh = net.NOT("nh", halt) mem_n = [] for j in range(256): ws = net.AND(f"ws{j}", [BD[j], nh]) nws = net.NOT(f"nws{j}", ws) row = [net.OR(f"mn{j}b{bit}", [net.AND(f"mn{j}b{bit}o", [mem[j][bit], nws]), net.AND(f"mn{j}b{bit}n", [r[bit], ws])]) for bit in range(8)] mem_n.append(row) outputs = pcn + [halt_n] + [s for row in mem_n for s in row] return net, subleq_state_names(), outputs class MatrixHost: """The subleq8io processor as a recurrent ternary linear-threshold network, with the device cells poked between recurrences.""" 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 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 = describe(data) good = 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 {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 = 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 {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 = 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 = describe(data) print(f"\nSelf-reproduction on the threshold matrices: target " f"{len(data):,} bytes (sha {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 {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 = 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() -> 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 if __name__ == "__main__": sys.exit(main())