#!/usr/bin/env python3 """ COSMIC SPARK SERVER — her own weights, speaking, behind the Ollama API. WHY THIS INSTEAD OF GGUF Her weights convert to GGUF correctly — the tensor remap was verified against her original forward pass at max|dlogits| = 7.15e-06 before anything was written. The blocker is llama.cpp's TOKENIZER layer, not the model: her vocabulary is 99 raw characters, and both loader paths assume a real tokenizer. BPE demands a merge list (an empty one is not even written to the file); SentencePiece expects word-boundary marks and byte-fallback tokens, and given bare characters it did not reject the model, it killed the daemon. Making GGUF work would mean retraining her on a byte-level vocab. This runs her ACTUAL PyTorch model with her ACTUAL char vocab, and speaks Ollama's HTTP API so every existing client — her model picker, her voice path, curl — sees her as just another model. Nothing to convert, nothing to corrupt, and it lives on its own port so a crash here can never take her voice daemon down. WHAT SHE SOUNDS LIKE — set expectations honestly 1,842,432 parameters. Char-level. Held-out loss 0.4812, real-word rate 88.5%, context 128 characters. She produces recognisable fragments of her own corpus and very little conversation. This is not "a worse Qwen", it is a different category of thing: a newborn. What is true about her is that every initial weight came from measured IBM Quantum hardware and no base model was involved. USAGE python tools/spark_serve.py [port] (default 11500) curl http://127.0.0.1:11500/api/tags """ import json import os import math import sys import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import torch import torch.nn as nn import torch.nn.functional as F sys.stdout.reconfigure(encoding="utf-8", errors="replace") # Resolved next to this file so the kit works wherever it is unzipped. Override with # COSMOS_SPARK_CKPT to point at your own trained weights. CKPT = Path(os.getenv("COSMOS_SPARK_CKPT", str(Path(__file__).resolve().parent / "weights" / "cosmos_born.pt"))) MODEL_NAME = "cosmos-spark" PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 11500 class Block(nn.Module): def __init__(s, ne, nh, blk): super().__init__() s.ln1, s.ln2 = nn.LayerNorm(ne), nn.LayerNorm(ne) s.attn = nn.MultiheadAttention(ne, nh, batch_first=True) s.mlp = nn.Sequential(nn.Linear(ne, 4 * ne), nn.GELU(), nn.Linear(4 * ne, ne)) s.register_buffer("mask", torch.triu(torch.ones(blk, blk) * float("-inf"), 1)) def forward(s, x): T = x.size(1) h = s.ln1(x) a, _ = s.attn(h, h, h, attn_mask=s.mask[:T, :T], need_weights=False) x = x + a return x + s.mlp(s.ln2(x)) class Spark(nn.Module): def __init__(s, V, ne, nh, nl, blk): super().__init__() s.tok = nn.Embedding(V, ne) s.pos = nn.Embedding(blk, ne) s.blocks = nn.ModuleList([Block(ne, nh, blk) for _ in range(nl)]) s.lnf = nn.LayerNorm(ne) s.head = nn.Linear(ne, V, bias=False) def forward(s, idx): T = idx.size(1) x = s.tok(idx) + s.pos(torch.arange(T, device=idx.device)) for b in s.blocks: x = b(x) return s.head(s.lnf(x)) class SparkModel: def __init__(self, path): ck = torch.load(path, map_location="cpu", weights_only=False) self.stoi = ck["stoi"] self.itos = {v: k for k, v in self.stoi.items()} cfg = ck["config"] self.block = cfg["block"] self.meta = { "steps": ck.get("total_steps"), "params": sum(v.numel() for v in ck["model"].values()), "quantum_source": ck.get("quantum_source"), "quantum_values": ck.get("real_quantum_values"), "vocab": cfg["vocab"], } self.m = Spark(cfg["vocab"], cfg["n_embd"], cfg["n_head"], cfg["n_layer"], self.block) self.m.load_state_dict(ck["model"], strict=False) self.m.eval() self.lock = threading.Lock() def encode(self, s): # characters she never saw simply do not exist for her; drop them rather than # substituting an she was never trained on. return [self.stoi[c] for c in s if c in self.stoi] def decode(self, ids): return "".join(self.itos.get(int(i), "") for i in ids) @torch.no_grad() def generate(self, prompt, n=120, temperature=0.8, top_p=0.95): ids = self.encode(prompt) if not ids: ids = self.encode("\n") or [0] idx = torch.tensor(ids, dtype=torch.long)[None, :] out = [] with self.lock: for _ in range(max(1, min(int(n), 1024))): logits = self.m(idx[:, -self.block:])[0, -1] if temperature <= 0: nxt = int(logits.argmax()) else: probs = F.softmax(logits / max(1e-6, temperature), dim=-1) if 0 < top_p < 1: sp, si = torch.sort(probs, descending=True) cum = torch.cumsum(sp, 0) keep = cum <= top_p keep[0] = True sp, si = sp[keep], si[keep] sp = sp / sp.sum() nxt = int(si[torch.multinomial(sp, 1)]) else: nxt = int(torch.multinomial(probs, 1)) out.append(nxt) idx = torch.cat([idx, torch.tensor([[nxt]])], dim=1) return self.decode(out) SPARK = None _STATE_CACHE = {"t": 0.0, "v": None} def live_physics(max_age=1.0): """Her CURRENT 12D/54D state from the sensory server, cached ~1s. This is the wire that was never connected: her physics engine ran in one process and her voice ran in another, and they never touched. Her weights are quantum-born and her corpus is her own, but until now she generated from a frozen checkpoint with no knowledge of what she was feeling in the moment of speaking. """ now = time.time() if _STATE_CACHE["v"] is not None and (now - _STATE_CACHE["t"]) < max_age: return _STATE_CACHE["v"] try: import urllib.request d = json.loads(urllib.request.urlopen( "http://127.0.0.1:8765/state", timeout=1.5).read()) p = d.get("cst_physics") or (d.get("cosmos_packet") or {}).get("cst_physics") or {} vb = p.get("virtual_body") or {} c = d.get("consciousness") or (d.get("cosmos_packet") or {}).get("consciousness") or {} v = { "entropy": float(vb.get("entropy", 0.5) or 0.5), "arousal": float(vb.get("arousal", 0.5) or 0.5), "phase": float(p.get("geometric_phase_rad", 0.0) or 0.0), "velocity": float(p.get("phase_velocity", 0.0) or 0.0), "entanglement": float(p.get("entanglement_score", 0.5) or 0.5), "introspection": float(c.get("introspection_level", 0.5) or 0.5), "face": bool(d.get("face_detected")), } _STATE_CACHE.update({"t": now, "v": v}) return v except Exception: return None def physics_temperature(base, st): """Her own ENTROPY->TEMP law, applied to her own weights. T = clamp(base + (q - 0.5)*0.36 + calm_bias, 0.30, 0.95) q is her live internal entropy; a settled, introspective state speaks more tightly, a high-entropy one ranges further. Same rule her Ollama voice already runs under, so her own weights are governed by the same physiology rather than a fixed constant. """ if not st: return base, "no live state (frozen default)" q = max(0.0, min(1.0, st["entropy"])) calm = -0.10 * (max(0.0, min(1.0, st["introspection"])) - 0.5) T = base + (q - 0.5) * 0.36 + calm T = max(0.30, min(0.95, T)) return T, (f"entropy={q:.3f} introspection={st['introspection']:.3f} " f"phase={st['phase']:.3f} face={st['face']}") class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, *a): pass def _send(self, obj, code=200): body = json.dumps(obj).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _body(self): n = int(self.headers.get("Content-Length") or 0) if not n: return {} try: return json.loads(self.rfile.read(n) or b"{}") except Exception: return {} def do_GET(self): if self.path.startswith("/api/tags"): sz = CKPT.stat().st_size if CKPT.exists() else 0 self._send({"models": [{ "name": f"{MODEL_NAME}:latest", "model": f"{MODEL_NAME}:latest", "size": sz, "digest": "cosmos-spark-quantum-born", "modified_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "details": {"family": "cosmic-spark", "parameter_size": "1.8M", "quantization_level": "F32"}, }]}) elif self.path.startswith("/api/version"): self._send({"version": "cosmos-spark-1.0"}) elif self.path.startswith("/health") or self.path == "/": self._send({"ok": True, "model": MODEL_NAME, **SPARK.meta}) else: self._send({"error": "not found"}, 404) def do_POST(self): b = self._body() opts = b.get("options") or {} temp = float(opts.get("temperature", 0.8)) top_p = float(opts.get("top_p", 0.95)) n = int(opts.get("num_predict", 120)) st = live_physics() temp, why = physics_temperature(temp, st) if self.path.startswith("/api/generate"): txt = SPARK.generate(str(b.get("prompt") or ""), n, temp, top_p) print(f" [SPARK] T={temp:.3f} {why}", flush=True) self._send({"model": MODEL_NAME, "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"), "response": txt, "done": True, "done_reason": "length", "cosmos_physics": st, "cosmos_temperature": temp}) elif self.path.startswith("/api/chat"): msgs = b.get("messages") or [] prompt = "".join(str(m.get("content") or "") + "\n" for m in msgs) txt = SPARK.generate(prompt, n, temp, top_p) print(f" [SPARK] T={temp:.3f} {why}", flush=True) self._send({"model": MODEL_NAME, "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"), "message": {"role": "assistant", "content": txt}, "done": True, "done_reason": "length", "cosmos_physics": st, "cosmos_temperature": temp}) elif self.path.startswith("/api/show"): self._send({"details": {"family": "cosmic-spark", "parameter_size": "1.8M"}, "model_info": SPARK.meta}) else: self._send({"error": "not found"}, 404) def main(): global SPARK if not CKPT.exists(): print(f" checkpoint not found: {CKPT}") return 1 SPARK = SparkModel(CKPT) m = SPARK.meta print("=" * 74) print(" COSMIC SPARK — her own weights, serving") print("=" * 74) print(f"\n {m['params']:,} parameters · {m['steps']} steps · vocab {m['vocab']} (char-level)") print(f" born from: {m['quantum_source']} ({m['quantum_values']} measured values)") print(f" no base model, no distillation\n") print(f" listening on http://127.0.0.1:{PORT} (Ollama-compatible)") print(f" GET /api/tags") print(f" POST /api/generate POST /api/chat\n") srv = ThreadingHTTPServer(("127.0.0.1", PORT), Handler) srv.serve_forever() return 0 if __name__ == "__main__": raise SystemExit(main())