""" SnapKitty Parallel Swarm Engine Orchestrates 5 computational swarms from a single text input: 1. Resonance — tokenization to GF(2^64-2^32+1) field elements + lattice routing 2. SUBLEQ — attention via integer subtraction-and-branch on 256-cell memory 3. DAG — ICP governance graph: EVIDENCE → CLAIM → PROOF → DECISION → EXECUTION 4. Quantum — Fibonacci anyon fusion (classical simulation, explicitly labeled) 5. Algebra — Jordan fixed-point iteration: T(ρ) = φ⁻¹·U·ρ·U† + φ⁻²·ρ Each swarm produces events on a shared timeline. Cross-swarm connections are explicit. Every visual property traces back to a computational value. """ import math import hashlib from dataclasses import dataclass, field from typing import Optional import numpy as np from subleq_engine import ( attention_head, subleq_run, activations_to_triads, born_collapse, phi_weights, quantization_jacobian, SUBLEQStep, SUBLEQResult, ) from resonance_word import ( tokenize, lattice_route, rw_pack, rw_unpack, CLASS, CLASS_NAMES, P_GOLD, LATTICE_ORDER, ) PHI = 1.6180339887 PHI_INV = 1.0 / PHI SWARM_LAYER = { 'resonance': 0, 'subleq': 1, 'dag': 2, 'quantum': 3, 'algebra': 4, } SWARM_COLORS = { 'resonance': '#00aaff', 'subleq': '#00ff88', 'dag': '#ffaa00', 'quantum': '#cc66ff', 'algebra': '#ff6644', } SWARM_NAMES = { 'resonance': 'Resonance Words', 'subleq': 'SUBLEQ Attention', 'dag': 'ICP-DAG Governance', 'quantum': 'Fibonacci Anyon Fusion', 'algebra': 'Jordan Fixed-Point', } @dataclass class SwarmEvent: tick: int swarm: str node_id: str label: str data: dict x: float = 0.0 y: float = 0.0 z: float = 0.0 color: str = '#ffffff' size: float = 8.0 connections: list = field(default_factory=list) class SwarmEngine: """Run all 5 swarms from a single text input.""" def __init__(self, text: str, seed: int = 42): self.text = text or "sovereign" self.seed = seed self.events: list[SwarmEvent] = [] self.max_tick = 0 self._resonance_tokens = [] self._resonance_routes = [] self._subleq_result: Optional[SUBLEQResult] = None self._subleq_activations: list[float] = [] self._run_resonance() self._run_subleq() self._run_dag() self._run_quantum() self._run_algebra() self._link_cross_swarm() # ── Resonance Swarm ────────────────────────────────────────────────── def _run_resonance(self): tokens = tokenize(self.text[:64]) routes = [lattice_route(t) for t in tokens] self._resonance_tokens = tokens self._resonance_routes = routes for i, (tok, route) in enumerate(zip(tokens, routes)): ch = self.text[i] if i < len(self.text) else '?' self.events.append(SwarmEvent( tick=i, swarm='resonance', node_id=f'RW-{i:04d}', label=f'{ch} → 0x{tok.word:016x}', data={ 'char': ch, 'word_hex': f'0x{tok.word:016x}', 'class': tok.class_name, 'class_tag': f'0x{tok.cls:02x}', 'payload': tok.payload, 'payload_hex': f'0x{tok.payload:014x}', 'lattice_p': route.p, 'lattice_b': route.b, 'lattice_idx': route.idx, 'field': 'GF(2^64 - 2^32 + 1)', }, x=float(i), y=float(SWARM_LAYER['resonance']), z=float(route.idx) / LATTICE_ORDER, color=SWARM_COLORS['resonance'], size=7 + (tok.payload % 6), )) self.max_tick = max(self.max_tick, len(tokens)) # ── SUBLEQ Swarm ───────────────────────────────────────────────────── def _run_subleq(self): tokens = self._resonance_tokens if not tokens: return n = len(tokens) activations = [] for i, tok in enumerate(tokens): phase = math.sin(2 * math.pi * i / max(n, 1) + tok.payload * 0.01) mag = (tok.payload % 256) / 256.0 activations.append(abs(phase * mag)) while len(activations) < 12: activations.append(0.1 * (1 + len(activations) % 5)) self._subleq_activations = activations triads = activations_to_triads(activations) mem = [0] * 256 # Pre-seed entire memory with diverse signed values for i in range(256): mem[i] = int(200 * math.sin(i * 0.13 * PHI)) + int(80 * math.cos(i * 0.09)) # Overlay program region (triads + halt) prog = [x for t in triads for x in t] + [-1, -1, -1] for i, v in enumerate(prog[:128]): mem[i] = v # Overlay activation weights at 128+ for i, a in enumerate(activations[:64]): mem[128 + i] = int(500 * a) - 100 result = subleq_run(mem, maxsteps=200) self._subleq_result = result for i, step in enumerate(result.trace[:100]): tag = 'BRANCH' if step.branch_taken else 'FALL' self.events.append(SwarmEvent( tick=i, swarm='subleq', node_id=f'SQ-{i:04d}', label=f'PC={step.pc} M[{step.B}]-M[{step.A}]={step.result} {tag}→{step.next_pc}', data={ 'step': i, 'pc': step.pc, 'A_addr': step.A, 'B_addr': step.B, 'C_addr': step.C, 'mem_A': step.mem_a, 'mem_B_before': step.mem_b_before, 'result': step.result, 'branch_taken': step.branch_taken, 'next_pc': step.next_pc, 'mechanism': 'M[B] := M[B] - M[A]; if M[B] <= 0 goto C', }, x=float(i), y=float(SWARM_LAYER['subleq']), z=float(step.result) / max(abs(step.result), 1) * 0.5, color='#00ff88' if step.branch_taken else '#ff4444', size=5 + min(abs(step.result) / 50, 12), )) self.max_tick = max(self.max_tick, len(result.trace)) # ── DAG Swarm ──────────────────────────────────────────────────────── def _run_dag(self): # ICP-DAG governance flow from ICP-DAG.m / ICP-DAG.lp dag_spec = [ ('EVIDENCE', [], 'Observed data or measurement'), ('CLAIM', ['EVIDENCE'], 'Assertion derived from evidence'), ('PROOF', ['CLAIM'], 'Formal verification of claim'), ('DECISION', ['PROOF'], 'Authorized action based on proof'), ('EXECUTION', ['DECISION'], 'Sealed computation with WORM receipt'), ] tokens = self._resonance_tokens for i, (name, deps, desc) in enumerate(dag_spec): node_hash = hashlib.sha256(f'{self.text}:{name}'.encode()).hexdigest()[:16] tick = i * 3 payload_val = tokens[i % max(len(tokens), 1)].payload if tokens else 0 entropy_ok = (payload_val % 1000) / 1000.0 < 0.20 self.events.append(SwarmEvent( tick=tick, swarm='dag', node_id=f'DAG-{name}', label=f'{name}', data={ 'node_type': name, 'description': desc, 'dependencies': deps, 'state': 'COMPLETE', 'hash': node_hash, 'payload': payload_val, 'entropy_check': entropy_ok, 'governance': 'ICP-DAG (MUMPS + ASP)', 'invariant': 'Nothing executes without passing the graph', }, x=float(tick), y=float(SWARM_LAYER['dag']), z=float(i) / len(dag_spec), color=SWARM_COLORS['dag'], size=14, connections=[f'DAG-{d}' for d in deps], )) self.max_tick = max(self.max_tick, len(dag_spec) * 3) # ── Quantum Swarm ──────────────────────────────────────────────────── def _run_quantum(self): # Fibonacci anyon fusion: τ⊗τ = 1⊕τ # prob(→1) = 1/φ² ≈ 0.382, prob(→τ) = 1/φ ≈ 0.618 # Deterministic from input (seeded RNG) n_anyons = max(4, min(len(self.text), 16)) if n_anyons % 2 == 1: n_anyons -= 1 seed_int = int(hashlib.sha256(self.text.encode()).hexdigest()[:8], 16) qrng = np.random.default_rng(seed_int) anyons = ['τ'] * n_anyons tick = 0 # Initial anyon state for i in range(n_anyons): self.events.append(SwarmEvent( tick=0, swarm='quantum', node_id=f'Q-{0}-{i}', label=f'τ_{i}', data={ 'type': 'anyon', 'charge': 'τ', 'index': i, 'quantum_dim': f'φ = {PHI:.4f}', 'fusion_rule': 'τ⊗τ = 1⊕τ', 'note': 'CLASSICAL SIMULATION of Fibonacci anyon model', }, x=float(i) * 0.5, y=float(SWARM_LAYER['quantum']), z=0.0, color=SWARM_COLORS['quantum'], size=8, )) current = list(anyons) fusion_round = 0 while len(current) > 1: fusion_round += 1 next_gen = [] for j in range(0, len(current) - 1, 2): a, b = current[j], current[j + 1] tick += 1 if a == 'τ' and b == 'τ': p_trivial = PHI_INV ** 2 result = '1' if qrng.random() < p_trivial else 'τ' prob_str = f'P(1)={p_trivial:.3f}, P(τ)={1-p_trivial:.3f}' elif a == '1' and b == '1': result = '1' prob_str = 'P(1)=1.000' else: result = 'τ' prob_str = 'P(τ)=1.000' next_gen.append(result) self.events.append(SwarmEvent( tick=tick, swarm='quantum', node_id=f'Q-{fusion_round}-{j // 2}', label=f'{a}⊗{b} → {result}', data={ 'type': 'fusion', 'input_a': a, 'input_b': b, 'output': result, 'round': fusion_round, 'probability': prob_str, 'topological_charge': result, 'note': 'CLASSICAL SIMULATION — not physical quantum hardware', }, x=float(tick), y=float(SWARM_LAYER['quantum']), z=float(fusion_round) / 5, color='#cc66ff' if result == 'τ' else '#9944aa', size=8 + fusion_round * 3, connections=[f'Q-{fusion_round-1}-{j}', f'Q-{fusion_round-1}-{j+1}'] if fusion_round == 1 else [f'Q-{fusion_round-1}-{j//2}'], )) if len(current) % 2 == 1: next_gen.append(current[-1]) current = next_gen self.max_tick = max(self.max_tick, tick + 1) # ── Algebra Swarm ──────────────────────────────────────────────────── def _run_algebra(self): # Jordan fixed-point: T(ρ) = φ⁻¹·U·ρ·U† + φ⁻²·ρ # Converges to ρ* where [U, ρ*] = 0 (proved in Lean 4, 0 sorry) h = hashlib.sha256(self.text.encode()).digest() theta = (h[0] / 255.0) * 2 * math.pi U = np.array([ [math.cos(theta), -math.sin(theta)], [math.sin(theta), math.cos(theta)], ]) U_dag = U.T.conjugate() rho = np.array([[0.7, 0.2], [0.2, 0.3]]) n_iter = min(25, max(self.max_tick, 15)) for i in range(n_iter): rho_new = PHI_INV * (U @ rho @ U_dag) + (PHI_INV ** 2) * rho tr = np.trace(rho_new).real if abs(tr) > 1e-10: rho_new = rho_new / tr comm = U @ rho_new - rho_new @ U comm_norm = float(np.linalg.norm(comm)) evals = sorted(np.linalg.eigvalsh(rho_new).tolist()) self.events.append(SwarmEvent( tick=i, swarm='algebra', node_id=f'ALG-{i:04d}', label=f'T^{i}(ρ): ‖[U,ρ]‖={comm_norm:.4f}', data={ 'iteration': i, 'map': 'T(ρ) = φ⁻¹·U·ρ·U† + φ⁻²·ρ', 'eigenvalues': [round(e, 6) for e in evals], 'commutator_norm': round(comm_norm, 6), 'trace': round(float(np.trace(rho_new).real), 6), 'rho': [[round(rho_new[r, c].real, 6) for c in range(2)] for r in range(2)], 'converged': comm_norm < 0.001, 'theta_rad': round(theta, 4), 'proof': 'JordanMatrixProof.lean (0 sorry)', }, x=float(i), y=float(SWARM_LAYER['algebra']), z=min(comm_norm, 1.0), color='#44ff66' if comm_norm < 0.01 else '#ff6644', size=5 + min(comm_norm * 30, 15), )) rho = rho_new self.max_tick = max(self.max_tick, n_iter) # ── Cross-Swarm Links ──────────────────────────────────────────────── def _link_cross_swarm(self): res = [e for e in self.events if e.swarm == 'resonance'] sq = [e for e in self.events if e.swarm == 'subleq'] dag = [e for e in self.events if e.swarm == 'dag'] # Resonance feeds SUBLEQ (payloads become activations) if res and sq: res[-1].connections.append(sq[0].node_id) # SUBLEQ feeds DAG (output drives governance entry) if sq and dag: sq[-1].connections.append(dag[0].node_id) # DAG EXECUTION links to algebra convergence check alg = [e for e in self.events if e.swarm == 'algebra'] exec_node = next((e for e in dag if 'EXECUTION' in e.node_id), None) if exec_node and alg: exec_node.connections.append(alg[-1].node_id) # ── Query Methods ──────────────────────────────────────────────────── def events_up_to(self, tick: int) -> list[SwarmEvent]: return [e for e in self.events if e.tick <= tick] def events_for_swarm(self, swarm: str) -> list[SwarmEvent]: return [e for e in self.events if e.swarm == swarm] def get_node(self, node_id: str) -> Optional[SwarmEvent]: for e in self.events: if e.node_id == node_id: return e return None def summary(self) -> dict: counts = {s: len(self.events_for_swarm(s)) for s in SWARM_LAYER} branches = sum(1 for e in self.events_for_swarm('subleq') if e.data.get('branch_taken')) falls = sum(1 for e in self.events_for_swarm('subleq') if not e.data.get('branch_taken', True)) q_events = self.events_for_swarm('quantum') fusions = [e for e in q_events if e.data.get('type') == 'fusion'] tau_results = sum(1 for f in fusions if f.data.get('output') == 'τ') alg = self.events_for_swarm('algebra') final_comm = alg[-1].data['commutator_norm'] if alg else None return { 'input': self.text, 'total_events': len(self.events), 'max_tick': self.max_tick, 'swarm_counts': counts, 'subleq_branches': branches, 'subleq_fallthroughs': falls, 'quantum_fusions': len(fusions), 'quantum_tau_outcomes': tau_results, 'algebra_final_commutator': final_comm, } def inspect_node(self, node_id: str) -> str: """Format node as markdown for the inspector panel.""" node = self.get_node(node_id) if not node: return f"Node `{node_id}` not found." lines = [ f"## {SWARM_NAMES.get(node.swarm, node.swarm)}", "", f"**ID:** `{node.node_id}`", f"**Tick:** {node.tick}", f"**Label:** {node.label}", "", "| Field | Value |", "|-------|-------|", ] for k, v in node.data.items(): if isinstance(v, list): v_str = ', '.join(str(x) for x in v) elif isinstance(v, float): v_str = f'{v:.6f}' elif isinstance(v, bool): v_str = 'Yes' if v else 'No' else: v_str = str(v) lines.append(f"| {k} | {v_str} |") if node.connections: lines.append("") lines.append(f"**Connections:** {', '.join(f'`{c}`' for c in node.connections)}") return '\n'.join(lines)