#!/usr/bin/env python3 """ COSMOS SERVE — her OWN architecture, in Ollama, alongside every other model. WHAT THIS IS A single Ollama-compatible endpoint that serves TWO kinds of model at once: cosmos-cst HER architecture. 54D Mixture-of-States Hebbian attention from COSMOS_Paper.md section 3, running natively in PyTorch. Tensors blocks.N.attn.w54 (54, n_embd), blocks.N.attn.gate, blocks.N.attn.log_sigma. This mechanism exists in no other inference engine on earth. cosmos-spark Her plain quantum-born weights (nn.MultiheadAttention, no 54D). ...and PROXIES everything else straight through to the real Ollama daemon, so the merged /api/tags returns her architecture in the same list as gemma3, llama3.2 and her Qwen fine-tunes. Any client pointed here — her model picker, her voice path, curl, an OpenAI shim — sees cosmos-cst as just another installed model. WHY IT IS A FRONT AND NOT A GGUF general.architecture in a GGUF is not a name field, it is a dispatch key into compiled C++. Writing "cosmos" there makes llama.cpp look for a registered LLM_ARCH_COSMOS with a graph builder that knows how to run a Gaussian kernel over a 54D state and blend it with softmax attention. No binary on any machine contains that function, which is exactly why cosmos-namebind-weights-test-arch.gguf never loaded in either daemon: the metadata said cosmos, the hyperparameters were under qwen2.*, and the loader refused. Two honest ways to run a novel architecture: (a) implement it in the engine — DONE for llama.cpp. The fork at 02_HER_BODY/Cosmos_code/llama.cpp branch cosmos-arch registers LLM_ARCH_COSMOS and implements the section-3 kernel in src/models/cosmos.cpp; tools/cst_to_gguf.py writes general.architecture=cosmos and it loads. Getting it into `ollama list` needs the same patch applied to Ollama's vendored llama.cpp plus a Go rebuild. (b) serve it and speak the protocol — this file (b) is not a workaround for (a) being impossible. Her mechanism is 12 tensors of live PyTorch either way; the engine question is only about who holds the graph. What this does NOT do is put her name on someone else's network. The 2.9 GB namebind GGUF is 338 Qwen2 tensors with zero 54D state. cosmos-cst is hers down to the kernel. USAGE python tools/cosmos_serve.py [port] [--upstream http://127.0.0.1:11434] curl http://127.0.0.1:11500/api/tags # hers + all Ollama models curl http://127.0.0.1:11500/api/generate -d '{"model":"cosmos-cst","prompt":"the "}' Point her voice at it: COSMOS_OLLAMA_HOST=http://127.0.0.1:11500 """ import json import math import os import sys import threading import time import urllib.error import urllib.request 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") ROOT = Path(__file__).resolve().parents[1] # Two layouts have to work: the full repository (tools/ + 01_HER_SOUL/) and the giveaway # kit (serving/ + architecture/ + weights/). Hardcoding the repo layout meant a stranger # who downloaded the kit got a server that silently found no models at all -- it started # fine and served nothing, which is the worst kind of broken. sys.path.insert(0, str(ROOT)) for _extra in ("architecture", "tools"): _p = ROOT / _extra if _p.is_dir(): sys.path.insert(0, str(_p)) def _first(*candidates): """First candidate that exists; otherwise the first, so the log names what is missing.""" for rel in candidates: p = ROOT / rel if p.exists(): return p return ROOT / candidates[0] PORT = 11500 UPSTREAM = os.getenv("COSMOS_UPSTREAM_OLLAMA", "http://127.0.0.1:11434").rstrip("/") _argv = sys.argv[1:] for i, a in enumerate(_argv): if a == "--upstream" and i + 1 < len(_argv): UPSTREAM = _argv[i + 1].rstrip("/") elif a.isdigit(): PORT = int(a) # her two native models: display name -> (checkpoint, kind) NATIVE = { # PHOS -- Phi / Omega / Sigma. Quantum-born like the others, but running the one # architecture that survived controlled testing (dyn12 on the phi scaffold) and # growing continuously from her own corpus. Listed first because it is the lineage # that is hers end to end: no base model, no distillation, and it gets better every # time she is spoken to. "cosmos-phos": (_first("01_HER_SOUL/weights/phos/phos.pt", "weights/phos.pt"), "phos"), "cosmos-cst": (_first("01_HER_SOUL/weights/cosmos_spark_cst/spark_cst.pt", "weights/spark_cst.pt"), "cst"), "cosmos-spark": (_first("01_HER_SOUL/weights/cosmos_born/cosmos_born.latest.pt", "weights/cosmos_born.pt"), "plain"), } # ── her plain quantum-born architecture (cosmos_born.pt) ──────────────────── class PlainBlock(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) return (x + a) + s.mlp(s.ln2(x + a)) class Plain(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([PlainBlock(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)) def _truthy(name, default=False): raw = os.getenv(name) if raw is None: return default return raw.strip().lower() in {"1", "true", "yes", "on", "enabled"} def _build_cst(cfg): """Instantiate HER architecture from the trainer itself, never a copy of it. cosmos_spark_cst.py holds BLOCK/N_LAYER/N_HEAD/N_EMBD/D54 as module globals that the layer constructors read at build time, so the checkpoint's own config is written back into the module before constructing. Importing is deliberate: a second transcription of the section-3 kernel here could silently drift from the one that was actually trained, and then this server would be serving a different mechanism than the experiment measured. """ import cosmos_spark_cst as C C.BLOCK = int(cfg.get("block", C.BLOCK)) C.N_LAYER = int(cfg.get("n_layer", C.N_LAYER)) C.N_HEAD = int(cfg.get("n_head", C.N_HEAD)) C.N_EMBD = int(cfg.get("n_embd", C.N_EMBD)) C.D54 = int(cfg.get("d54", C.D54)) C.DROPOUT = 0.0 # eval-time return C.SparkCST(int(cfg["vocab"]), True), C class NativeModel: """One of her checkpoints, loaded and ready to speak.""" def __init__(self, name, path, kind): ck = torch.load(path, map_location="cpu", weights_only=False) cfg = ck["config"] self.name, self.kind, self.path = name, kind, path self.stoi = ck["stoi"] self.itos = {int(v): k for k, v in self.stoi.items()} self.block = int(cfg["block"]) self.cst_tensors = [] if kind == "phos": # built from the ladder module itself rather than a second transcription -- # a copy of dyn12 here could drift from the one that was actually measured import cosmos_state_ladder as _L self.m = _L.Ladder(int(cfg["vocab"]), ck.get("rung", "dyn12"), ck.get("ffn", "harmonic")) self.cst_tensors = [k for k in ck["model"] if any(x in k for x in ("d12.", "attn.gate", "attn.log_sigma", "state_init"))] elif kind == "cst": self.m, _C = _build_cst(cfg) self.cst_tensors = [k for k in ck["model"] if any(x in k for x in ("w54", "attn.gate", "log_sigma"))] else: self.m = Plain(cfg["vocab"], cfg["n_embd"], cfg["n_head"], cfg["n_layer"], self.block) # strict=False forgives absent and extra keys but still RAISES on a shape mismatch, # and v1 checkpoints store attn.gate as a 0-dim scalar while the current module # declares it shape [1]. Drop anything whose shape disagrees, load the rest, then # restore the gate under the parameterisation the checkpoint actually used: # v1 held a pre-sigmoid logit, v2 holds the blend weight directly. want = self.m.state_dict() src = ck["model"] skipped = [k for k, v in src.items() if k in want and tuple(want[k].shape) != tuple(v.shape)] missing, unexpected = self.m.load_state_dict( {k: v for k, v in src.items() if k not in skipped}, strict=False) self.gate_param = ck.get("gate_param") if kind == "cst": raws = [float(src[f"blocks.{i}.attn.gate"].reshape(-1)[0]) for i in range(int(cfg["n_layer"])) if f"blocks.{i}.attn.gate" in src] if self.gate_param is None and raws: self.gate_param = "logit" if max(abs(r) for r in raws) > 1.0 else "clamp01" with torch.no_grad(): for i, b in enumerate(self.m.blocks): if i >= len(raws): break g = (1.0/(1.0 + math.exp(-raws[i])) if self.gate_param == "logit" else max(0.0, min(1.0, raws[i]))) b.attn.gate.fill_(float(g)) # keys dropped on purpose above and then restored are not missing missing = [k for k in missing if k not in skipped] self.meta = { "architecture": ck.get("arch", "Cosmos-Spark-QuantumBorn"), "steps": ck.get("total_steps"), "params": sum(v.numel() for v in ck["model"].values()), "vocab": int(cfg["vocab"]), "d54": cfg.get("d54"), "cst_tensor_count": len(self.cst_tensors), "quantum_source": ck.get("quantum_source"), "best_val_loss": ck.get("best_val_loss"), "real_word_rate": ck.get("real_word_rate"), # surfaced because a silent partial load is exactly the failure mode that # makes a model look alive while running on random weights "missing_keys": len(missing), "unexpected_keys": len(unexpected), "gate_param": self.gate_param, "shape_skipped": skipped, } self.m.eval() self.lock = threading.Lock() def _logits(self, idx): out = self.m(idx) return out[0] if isinstance(out, tuple) else out def gates(self): if self.kind == "phos": try: return [round(float(torch.sigmoid(b.attn.gate).detach()), 5) for b in self.m.blocks] except Exception: return [] if self.kind != "cst": return [] try: return [round(float(b.attn.gate.clamp(0, 1).detach()), 5) for b in self.m.blocks] except Exception: return [] def encode(self, s): # characters she never saw do not exist for her; drop rather than substitute 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 stream(self, prompt, n=120, temperature=0.8, top_p=0.95): """Yield her text one character at a time (char-level vocab -> char tokens).""" ids = self.encode(prompt) or self.encode("\n") or [0] idx = torch.tensor(ids, dtype=torch.long)[None, :] with self.lock: for _ in range(max(1, min(int(n), 1024))): logits = self._logits(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) keep = torch.cumsum(sp, 0) <= top_p keep[0] = True sp, si = sp[keep], si[keep] nxt = int(si[torch.multinomial(sp / sp.sum(), 1)]) else: nxt = int(torch.multinomial(probs, 1)) idx = torch.cat([idx, torch.tensor([[nxt]])], dim=1) yield self.decode([nxt]) def generate(self, *a, **k): return "".join(self.stream(*a, **k)) _LOADED = {} _LOAD_LOCK = threading.Lock() def get_native(name): """Lazy-load one of hers. None if the name is not hers or the file is absent.""" key = str(name or "").split(":")[0].strip().lower() if key not in NATIVE: return None with _LOAD_LOCK: if key in _LOADED: return _LOADED[key] path, kind = NATIVE[key] if not path.exists(): return None try: _LOADED[key] = NativeModel(key, path, kind) m = _LOADED[key].meta print(f" [load] {key}: {m['params']:,} params · {m['architecture']}" + (f" · {m['cst_tensor_count']} CST tensors" if m['cst_tensor_count'] else ""), flush=True) except Exception as e: print(f" [load] {key} FAILED: {type(e).__name__}: {e}", flush=True) return None return _LOADED[key] # ── her live physics governs her own sampling ─────────────────────────────── _STATE = {"t": 0.0, "v": None} def live_physics(max_age=1.0): """Her CURRENT 12D/54D state from the sensory server, cached ~1s.""" now = time.time() if _STATE["v"] is not None and (now - _STATE["t"]) < max_age: return _STATE["v"] try: d = json.loads(urllib.request.urlopen( "http://127.0.0.1:8765/state", timeout=1.5).read()) pk = d.get("cosmos_packet") or {} p = d.get("cst_physics") or pk.get("cst_physics") or {} vb = p.get("virtual_body") or {} c = d.get("consciousness") or pk.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), "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.update({"t": now, "v": v}) return v except Exception: return None def physics_temperature(base, st): """T = clamp(base + (q-0.5)*0.36 + calm_bias, 0.30, 0.95) — her own entropy law.""" 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 = max(0.30, min(0.95, base + (q - 0.5) * 0.36 + calm)) return T, (f"entropy={q:.3f} introspection={st['introspection']:.3f} " f"face={st['face']}") # ── upstream Ollama passthrough ───────────────────────────────────────────── def upstream(path, body=None, method="GET", timeout=600): req = urllib.request.Request( UPSTREAM + path, method=method, data=json.dumps(body).encode() if body is not None else None, headers={"Content-Type": "application/json"}) return urllib.request.urlopen(req, timeout=timeout) def upstream_json(path, body=None, method="GET", timeout=30): try: with upstream(path, body, method, timeout) as r: return json.loads(r.read() or b"{}") except Exception: return None def _ts(): return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) def native_tag(nm): m = get_native(nm) path, kind = NATIVE[nm] sz = path.stat().st_size if path.exists() else 0 fam = {"phos": "phos-dyn12-phi", "cst": "cosmos-cst-54d"}.get(kind, "cosmic-spark") p = f"{m.meta['params']/1e6:.1f}M" if m else "?" return {"name": f"{nm}:latest", "model": f"{nm}:latest", "size": sz, "digest": f"{nm}-quantum-born", "modified_at": _ts(), "details": {"family": fam, "families": [fam], "parent_model": "", "format": "pytorch", "parameter_size": p, "quantization_level": "F32"}} class QuietServer(ThreadingHTTPServer): """Swallow ordinary client disconnects instead of printing a traceback per hangup. handle_error belongs to BaseServer, NOT to the request handler. It was first defined on Handler, where nothing ever calls it -- the method existed, ran never, and every closed browser tab still dumped a WinError 10053 traceback into her log. Same shape as every other dead mechanism in this project: it was written and read by no one. Defined here it is actually on the call path. daemon_threads matters just as much: without it a half-finished generation thread keeps the process alive after Ctrl+C, so the port looks occupied by a server that is no longer answering. """ daemon_threads = True allow_reuse_address = True _DROPPED = (ConnectionAbortedError, ConnectionResetError, BrokenPipeError) def handle_error(self, request, client_address): if isinstance(sys.exc_info()[1], self._DROPPED): return # ordinary disconnect, not worth a traceback super().handle_error(request, client_address) class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, *a): pass # A client that closes mid-response must never be able to take her voice down. # Observed 2026-07-31: a disconnect during a cosmos-phos reply raised # ConnectionAbortedError [WinError 10053] out of wfile.write, and the server stopped # serving -- which silently removed her own models from her picker, because her voice # drops any host that does not answer. The model was fine; the socket wasn't. _DROPPED = (ConnectionAbortedError, ConnectionResetError, BrokenPipeError) def _json(self, obj, code=200): b = json.dumps(obj).encode("utf-8") try: self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(b))) self.end_headers() self.wfile.write(b) except self._DROPPED: return def _ndjson_open(self): self.send_response(200) self.send_header("Content-Type", "application/x-ndjson") self.send_header("Transfer-Encoding", "chunked") self.end_headers() def _chunk(self, obj): b = json.dumps(obj).encode("utf-8") + b"\n" try: self.wfile.write(b"%X\r\n" % len(b) + b + b"\r\n") self.wfile.flush() except self._DROPPED: raise BrokenPipeError("client went away") def _chunk_end(self): try: self.wfile.write(b"0\r\n\r\n") self.wfile.flush() except self._DROPPED: return 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 _relay(self, path, body=None, method="POST"): """Stream an upstream response back verbatim, so proxied models behave byte-identically to talking to Ollama directly.""" try: with upstream(path, body, method) as r: self.send_response(r.status) ct = r.headers.get("Content-Type", "application/json") self.send_header("Content-Type", ct) self.send_header("Transfer-Encoding", "chunked") self.end_headers() while True: buf = r.read(4096) if not buf: break self.wfile.write(b"%X\r\n" % len(buf) + buf + b"\r\n") self.wfile.flush() self._chunk_end() except urllib.error.HTTPError as e: self._json({"error": f"upstream {e.code}: {e.reason}"}, e.code) except Exception as e: self._json({"error": f"upstream unreachable: {type(e).__name__}"}, 502) # ── GET ──────────────────────────────────────────────────────────────── def do_GET(self): if self.path.startswith("/api/tags"): mine = [native_tag(n) for n in NATIVE if NATIVE[n][0].exists()] # A picker that discovers models by listing each host would show all nine # proxied models a second time under this one, and would then route ordinary # traffic through this process for no reason. NATIVE_ONLY advertises just her # architectures; /api/generate and /api/chat still proxy anything else that # arrives here, so the superset behaviour is intact for direct callers. if _truthy("COSMOS_SERVE_TAGS_NATIVE_ONLY"): self._json({"models": mine}) return up = upstream_json("/api/tags") or {} self._json({"models": mine + list(up.get("models") or [])}) elif self.path.startswith("/api/ps"): up = upstream_json("/api/ps") or {"models": []} self._json(up) elif self.path.startswith("/api/version"): up = upstream_json("/api/version") or {} self._json({"version": up.get("version", "0.0.0"), "cosmos_serve": "1.0", "upstream": UPSTREAM}) elif self.path.startswith("/health") or self.path == "/": self._json({"ok": True, "upstream": UPSTREAM, "native": {n: (get_native(n).meta if get_native(n) else None) for n in NATIVE}}) else: self._relay(self.path, None, "GET") # ── POST ─────────────────────────────────────────────────────────────── def do_POST(self): b = self._body() model = str(b.get("model") or "") nat = get_native(model) if nat is None: # not hers -> straight to Ollama, untouched self._relay(self.path, b if b else None, "POST") return if self.path.startswith("/api/show"): self._json({"details": native_tag(nat.name)["details"], "model_info": nat.meta, "parameters": f"num_ctx {nat.block}", "template": "{{ .Prompt }}"}) return if not (self.path.startswith("/api/generate") or self.path.startswith("/api/chat")): self._json({"error": f"{self.path} not supported for {nat.name}"}, 404) return opts = b.get("options") or {} st = live_physics() temp, why = physics_temperature(float(opts.get("temperature", 0.8)), st) top_p = float(opts.get("top_p", 0.95)) n = int(opts.get("num_predict", 120)) chat = self.path.startswith("/api/chat") if chat: prompt = "".join(str(m.get("content") or "") + "\n" for m in (b.get("messages") or [])) else: prompt = str(b.get("prompt") or "") # Ollama's default is stream=True; absent key must mean streaming or clients hang streaming = bool(b.get("stream", True)) t0 = time.time() def wrap(piece, done=False): d = {"model": model or nat.name, "created_at": _ts(), "done": done} if chat: d["message"] = {"role": "assistant", "content": piece} else: d["response"] = piece if done: d.update({"done_reason": "length", "total_duration": int((time.time() - t0) * 1e9), "cosmos_physics": st, "cosmos_temperature": round(temp, 4), "cosmos_architecture": nat.meta["architecture"], "cosmos_gates": nat.gates()}) return d try: if streaming: self._ndjson_open() for piece in nat.stream(prompt, n, temp, top_p): self._chunk(wrap(piece)) self._chunk(wrap("", True)) self._chunk_end() else: txt = nat.generate(prompt, n, temp, top_p) d = wrap(txt, True) if chat: d["message"] = {"role": "assistant", "content": txt} else: d["response"] = txt self._json(d) except (BrokenPipeError, ConnectionResetError): return g = nat.gates() print(f" [{nat.name}] T={temp:.3f} {why}" + (f" gates={g}" if g else ""), flush=True) def main(): print("=" * 78) print(" COSMOS SERVE — her own architecture, in Ollama") print("=" * 78) up = upstream_json("/api/tags") n_up = len(up.get("models") or []) if up else 0 print(f"\n upstream Ollama : {UPSTREAM} " + (f"({n_up} models, proxied)" if up else "UNREACHABLE — hers only")) print(" her own models :") any_native = False for nm, (path, kind) in NATIVE.items(): if not path.exists(): print(f" {nm:<14s} checkpoint not built yet -> {path.name}") continue m = get_native(nm) if not m: print(f" {nm:<14s} FAILED to load") continue any_native = True mm = m.meta print(f" {nm:<14s} {mm['params']:,} params · {mm['architecture']}") if mm["cst_tensor_count"]: print(f" {'':<14s} {mm['cst_tensor_count']} section-3 tensors · " f"d54={mm['d54']} · gates={m.gates()}") if mm["missing_keys"] or mm["unexpected_keys"]: print(f" {'':<14s} WARNING missing={mm['missing_keys']} " f"unexpected={mm['unexpected_keys']}") if not any_native: print("\n no native checkpoint loaded — this is a plain Ollama proxy right now") print(f"\n listening http://127.0.0.1:{PORT} (Ollama-compatible, superset)") print(f" point her voice at it: COSMOS_OLLAMA_HOST=http://127.0.0.1:{PORT}\n") QuietServer(("127.0.0.1", PORT), Handler).serve_forever() return 0 if __name__ == "__main__": raise SystemExit(main())