serving/cosmos_serve.py

#!/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
        # <unk> 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())