Why does the chat go offline after a while?

The default set-up uses a Cloudflare quick tunnel, whose URL is ephemeral: it changes every time the notebook restarts, and the notebook itself recycles after a period. When that happens the old URL stops working. Re-run the server cell to get a new URL and paste it into the chat settings. For a stable public chat, host the model on a persistent GPU behind a fixed domain.

Is an API key needed?

No, not by default. The server runs keyless (NO_AUTH = True), so the chat connects with no key. If you set NO_AUTH = False and an API_KEY on the server, put the same key in the chat settings.

Where does the agent's code actually run?

In your own browser tab. Agent mode runs a real Python sandbox (CPython compiled to WebAssembly) locally, so files, package installs and code execution happen on your machine, not on the model server. The model only decides what to run; you can watch every command in the Terminal panel.

Is the web search in CorX Chat real?

Yes. Agent mode's web_search tool calls a keyless DuckDuckGo-backed endpoint and returns real titles, URLs and snippets, shown as a collapsible sources list with links. fetch_url then reads an actual page's content. Nothing about search results is simulated or invented by the model.

What does the Effort setting actually change?

Real request parameters: the token budget per reply, how many tool-call rounds the agent gets before it must stop, how many search results it pulls back, and how strongly the system prompt tells it to think step by step, verify its work and search when unsure. Higher effort takes longer and uses more tokens on purpose.

Running CorX3.8, and how the chat works.

The chat talks to CorX3.8 on your own server. This page has the server code to copy, the four steps to connect it, and how Agent mode’s terminal, files and uploads work.

CorX3.8-27B is too big to run in a browser, so it runs on a GPU you provide — a notebook like molab or Colab, or a rented GPU box. The code below turns that GPU into an OpenAI-compatible API and opens a public URL. You paste that URL into the chat’s settings, and the browser talks to your model directly.

Step 1 · The server

Copy this and run it on a GPU.

Paste the whole thing into one notebook cell and run it. First run installs dependencies and loads the model (1–2 minutes), then it prints a public URL.

"""
================================================================
 CorX3.8-27B — API SERVER for the CorX Labs site
 CorX Labs / Nathan

 Serves the model as an OpenAI-compatible endpoint your website
 can call. transformers backend (vLLM crashed on Qwen3.8 GDN).
================================================================
"""

import os
os.environ["CUDA_VISIBLE_DEVICES"] = "0"
os.environ["PYTORCH_CUDA_ALLOC_CONF"] = "expandable_segments:True"

import subprocess
import sys
import json
import time
import threading
import queue
import uuid
import urllib.request
import gc
import asyncio

# ---------------- SETTINGS ----------------
MODEL_REPO   = "Sigmandndnns/CorX3.8-27B"
HF_TOKEN     = ""
PORT         = 8000
MAX_NEW_TOKENS_CAP = 32000

# Precision. "auto" measures the GPU and picks: full bf16 when the card can
# hold the whole 27B (~54GB) — an RTX PRO 6000 Blackwell, H100, A100 80GB —
# and 4-bit on smaller cards so it still fits in VRAM instead of spilling to
# CPU. Force it with "bf16" or "4bit".
PRECISION = "auto"

# auth
NO_AUTH      = True                     # True = keyless (your "unrestricted")
API_KEY      = "corx-key"               # used only if NO_AUTH = False

# concurrency
QUEUE_TIMEOUT = 120                     # seconds a request waits for the GPU
STREAM_TIMEOUT = 300                    # seconds to wait for a token before
                                        # giving up. Must cover prefill of the
                                        # whole prompt on a 27B model — too low
                                        # and you get "[Generation Error]".
# ------------------------------------------

print("Installing deps (first run only)...")
_deps = ["fastapi", "uvicorn[standard]", "transformers", "accelerate"]
if PRECISION != "bf16":          # may still be needed if auto picks 4-bit
    _deps.append("bitsandbytes")
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-U", *_deps],
               check=False)

import torch
from transformers import (AutoTokenizer, AutoModelForCausalLM,
                          TextIteratorStreamer)

if HF_TOKEN:
    os.environ["HF_TOKEN"] = HF_TOKEN

print(f"GPUs visible: {torch.cuda.device_count()}")
if torch.cuda.device_count() == 0:
    raise RuntimeError("No GPU attached to this molab session.")

# ---------------- PICK PRECISION FROM THE ACTUAL CARD ----------------
_props = [torch.cuda.get_device_properties(i) for i in range(torch.cuda.device_count())]
TOTAL_VRAM_GB = sum(p.total_memory for p in _props) / 1e9
_cap = f"{_props[0].major}.{_props[0].minor}"
print(f"GPU: {_props[0].name} — {TOTAL_VRAM_GB:.0f} GB total VRAM, compute {_cap}, torch {torch.__version__}")

# A new architecture (Blackwell is sm_120) needs a PyTorch built with kernels
# for it. On an older wheel every CUDA op dies with a bewildering "no kernel
# image is available" — so say plainly what to do before loading 54GB.
try:
    _supported = torch.cuda.get_arch_list()
    if _supported and f"sm_{_props[0].major}{_props[0].minor}" not in _supported:
        print(f"[warn] This torch was built for {_supported} — nothing for sm_{_props[0].major}{_props[0].minor}.")
        print("       If loading fails with 'no kernel image is available', install a matching build:")
        print("       pip install -U --index-url https://download.pytorch.org/whl/cu128 torch")
except Exception:
    pass

# 27B bf16 weights are ~54GB; leave headroom for the KV cache and activations.
_BF16_NEEDS_GB = 62
if PRECISION == "auto":
    LOAD_IN_4BIT = TOTAL_VRAM_GB < _BF16_NEEDS_GB
    print(f"[precision] auto → {'4-bit' if LOAD_IN_4BIT else 'full bf16'} "
          f"({TOTAL_VRAM_GB:.0f} GB {'<' if LOAD_IN_4BIT else '>='} {_BF16_NEEDS_GB} GB)")
else:
    LOAD_IN_4BIT = (PRECISION == "4bit")
    print(f"[precision] forced → {'4-bit' if LOAD_IN_4BIT else 'full bf16'}")
    if not LOAD_IN_4BIT and TOTAL_VRAM_GB < _BF16_NEEDS_GB:
        print(f"[warn] Only {TOTAL_VRAM_GB:.0f} GB of VRAM — bf16 will spill to CPU and be slow.")

# ---------------- LOAD MODEL ----------------
tok_kwargs = {"trust_remote_code": True}
if HF_TOKEN:
    tok_kwargs["token"] = HF_TOKEN

print(f"Loading CorX3.8-27B in {'4-bit' if LOAD_IN_4BIT else 'bf16'} (1-3 min)...")
tok = AutoTokenizer.from_pretrained(MODEL_REPO, **tok_kwargs)
if tok.pad_token is None:
    tok.pad_token = tok.eos_token

load_kwargs = dict(trust_remote_code=True, low_cpu_mem_usage=True)
# transformers renamed torch_dtype -> dtype; pick whichever this version takes
# so the load is clean on both.
import inspect as _inspect
_load_params = _inspect.signature(AutoModelForCausalLM.from_pretrained).parameters
load_kwargs["dtype" if "dtype" in _load_params else "torch_dtype"] = torch.bfloat16
if HF_TOKEN:
    load_kwargs["token"] = HF_TOKEN

# SDPA is PyTorch's built-in fused attention — a solid speedup over the
# eager path on any modern card, and it needs no extra package.
try:
    load_kwargs["attn_implementation"] = "sdpa"
except Exception:
    pass

if LOAD_IN_4BIT:
    # Small card: quantise so the whole model still sits in VRAM. Spilling to
    # CPU is what makes the first token take minutes and trip the timeout.
    from transformers import BitsAndBytesConfig
    load_kwargs["device_map"] = "auto"
    load_kwargs["quantization_config"] = BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_compute_dtype=torch.bfloat16,
        bnb_4bit_quant_type="nf4",
        bnb_4bit_use_double_quant=True)
elif torch.cuda.device_count() == 1:
    # Enough VRAM and one card: put the whole model on GPU 0 explicitly.
    # Pinning it beats device_map="auto", which keeps the offload machinery
    # in play and can still decide to park something on the CPU.
    load_kwargs["device_map"] = {"": 0}
else:
    load_kwargs["device_map"] = "auto"   # shard across the cards

model = AutoModelForCausalLM.from_pretrained(MODEL_REPO, **load_kwargs)
model.eval()
# NOTE: do not set generation_config.cache_implementation = "static" here.
# A static cache sends generate() down the torch.compile / _inductor path,
# which throws inside the worker thread ("Detected that you are using FX
# tracing...") — no tokens are ever produced and the request just hangs.
# The default dynamic cache is correct for this server.
print(f"Model ready. VRAM in use: {torch.cuda.memory_allocated()/1e9:.1f} GB "
      f"of {TOTAL_VRAM_GB:.0f} GB")

# Confirm nothing landed off-GPU — this is the usual root cause of a first
# token that never arrives.
try:
    _dev = set(str(d) for d in model.hf_device_map.values())
    if any(d in ("cpu", "disk") for d in _dev):
        print("[warn] Part of the model is on CPU/disk — generation will be SLOW.")
        print("       Set PRECISION = \"4bit\" so it fits entirely in VRAM.")
    else:
        print(f"[ok] Whole model resident on GPU ({sorted(_dev)}).")
except Exception:
    pass

# ---------------- GENERATION (single GPU, queued) ----------------
_job_lock = threading.Lock()

# Worker wrapper to ensure gradients are off in the background thread.
# If generation dies, end the streamer straight away and hand the reason
# back — otherwise the reader sits waiting for a token that will never come
# and the request only fails much later, on a timeout, with no explanation.
def _generate_thread_worker(kwargs, streamer=None, err_box=None):
    with torch.inference_mode():
        try:
            model.generate(**kwargs)
        except Exception as e:
            import traceback
            traceback.print_exc()
            print(f"\n[Generation Error in thread] {type(e).__name__}: {e}")
            if err_box is not None:
                err_box.append(f"{type(e).__name__}: {e}")
            if streamer is not None:
                try:
                    streamer.end()        # unblock the reader immediately
                except Exception:
                    pass

def _generate(messages, max_new_tokens, temperature, streamer=None, err_box=None):
    prompt = tok.apply_chat_template(messages, tokenize=False,
                                     add_generation_prompt=True)
    inputs = tok(prompt, return_tensors="pt").to(model.device)
    kwargs = dict(**inputs,
                  max_new_tokens=min(int(max_new_tokens), MAX_NEW_TOKENS_CAP),
                  temperature=max(float(temperature), 0.01),
                  top_p=0.9, do_sample=float(temperature) > 0,
                  repetition_penalty=1.05,
                  pad_token_id=tok.pad_token_id or tok.eos_token_id)

    if streamer is not None:
        kwargs["streamer"] = streamer
        # Must target our safe wrapper instead of model.generate directly
        threading.Thread(target=_generate_thread_worker,
                         args=(kwargs, streamer, err_box),
                         daemon=True).start()
        return None

    with torch.inference_mode():
        out = model.generate(**kwargs)
    return tok.decode(out[0][inputs["input_ids"].shape[1]:],
                      skip_special_tokens=True)

# ---------------- OPENAI-COMPATIBLE API + CORS ----------------
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import uvicorn

app = FastAPI(title="CorX3.8 API")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

def check_auth(request: Request):
    if NO_AUTH:
        return
    if request.headers.get("authorization", "") != f"Bearer {API_KEY}":
        raise HTTPException(status_code=401, detail="bad api key")

@app.get("/health")
async def health():
    return {"status": "ok", "model": "corx3.8",
            "vram_gb": round(torch.cuda.memory_allocated()/1e9, 1),
            "busy": _job_lock.locked()}

@app.get("/v1/models")
async def list_models(request: Request):
    check_auth(request)
    return {"object": "list",
            "data": [{"id": "corx3.8", "object": "model",
                      "owned_by": "corx-labs"}]}

@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    check_auth(request)
    body = await request.json()
    messages = body.get("messages", [])
    if not messages:
        raise HTTPException(status_code=400, detail="messages required")
    max_new = body.get("max_tokens", 2048)
    temperature = body.get("temperature", 0.7)
    stream = bool(body.get("stream", False))
    cid = f"chatcmpl-{uuid.uuid4().hex[:12]}"

    got = _job_lock.acquire(timeout=QUEUE_TIMEOUT)
    if not got:
        raise HTTPException(status_code=503, detail="model busy, retry shortly")

    if not stream:
        try:
            text = _generate(messages, max_new, temperature)
            return JSONResponse({
                "id": cid, "object": "chat.completion", "model": "corx3.8",
                "choices": [{"index": 0, "finish_reason": "stop",
                             "message": {"role": "assistant", "content": text}}],
            })
        except Exception as e:
            import traceback; traceback.print_exc()
            return JSONResponse(status_code=500,
                content={"error": {"message": f"{type(e).__name__}: {e}",
                                    "type": "generation_error"}})
        finally:
            # Bulletproof release & GC
            try: _job_lock.release()
            except RuntimeError: pass
            torch.cuda.empty_cache()
            gc.collect()

    async def event_stream():
        try:
            # This is the wait for the NEXT token, and the first one only
            # arrives after the whole prompt is prefilled. A 27B model with a
            # long system prompt (and any CPU offload) can take well over a
            # minute to get there, so a short timeout here shows up in the
            # chat as "[Generation Error]" even though nothing actually failed.
            streamer = TextIteratorStreamer(
                tok, skip_prompt=True, skip_special_tokens=True, timeout=STREAM_TIMEOUT)
            err_box = []          # filled by the worker if generate() throws
            _generate(messages, max_new, temperature, streamer=streamer,
                      err_box=err_box)

            _DONE = object() # Sentinel value to prevent StopIteration Exception
            sent_any = False

            while True:
                try:
                    # Pass _DONE as the default value to next() so it doesn't throw an error to asyncio
                    chunk = await asyncio.to_thread(next, streamer, _DONE)

                    if chunk is _DONE:
                        break # Stream finished successfully

                    data = {"id": cid, "object": "chat.completion.chunk",
                            "model": "corx3.8",
                            "choices": [{"index": 0, "delta": {"content": chunk},
                                         "finish_reason": None}]}
                    sent_any = True
                    yield f"data: {json.dumps(data)}\n\n"

                except queue.Empty:
                    print(f"[api] No token for {STREAM_TIMEOUT}s — giving up. "
                          f"Raise STREAM_TIMEOUT, shorten the prompt, or load in 4-bit.")
                    err = {"id": cid, "object": "chat.completion.chunk",
                           "model": "corx3.8",
                           "choices": [{"index": 0, "delta": {"content":
                                        f"\n[No token produced within {STREAM_TIMEOUT}s. "
                                        f"The model is likely still prefilling — raise STREAM_TIMEOUT "
                                        f"in the server, or load the model in 4-bit so it fits the GPU.]"},
                                        "finish_reason": "error"}]}
                    yield f"data: {json.dumps(err)}\n\n"
                    break

            # generate() died before producing anything: say what actually
            # went wrong instead of returning an empty, silent reply.
            if err_box and not sent_any:
                err = {"id": cid, "object": "chat.completion.chunk",
                       "model": "corx3.8",
                       "choices": [{"index": 0, "delta": {"content":
                                    f"\n[Generation failed on the server: {err_box[0]}]"},
                                    "finish_reason": "error"}]}
                yield f"data: {json.dumps(err)}\n\n"

            done = {"id": cid, "object": "chat.completion.chunk",
                    "model": "corx3.8",
                    "choices": [{"index": 0, "delta": {},
                                 "finish_reason": "stop"}]}
            yield f"data: {json.dumps(done)}\n\n"
            yield "data: [DONE]\n\n"
        finally:
            # Bulletproof release & GC for streaming
            try: _job_lock.release()
            except RuntimeError: pass
            torch.cuda.empty_cache()
            gc.collect()

    return StreamingResponse(event_stream(), media_type="text/event-stream")

# ---------------- START SERVER (background thread) ----------------
_server_ready = threading.Event()

def serve():
    config = uvicorn.Config(app, host="0.0.0.0", port=PORT, log_level="warning")
    server = uvicorn.Server(config)
    _server_ready.set()
    server.run()

threading.Thread(target=serve, daemon=True).start()
_server_ready.wait(timeout=10)
time.sleep(2)

try:
    urllib.request.urlopen(f"http://127.0.0.1:{PORT}/health", timeout=5)
    print(f"[api] local server healthy on :{PORT}")
except Exception as e:
    print(f"[api] WARNING: local server not responding ({type(e).__name__})")

# ---------------- CLOUDFLARE QUICK TUNNEL (zero account) ----------------
import re
import platform

def get_cloudflared():
    path = os.path.abspath("./cloudflared")
    if os.path.exists(path):
        return path
    arch = platform.machine().lower()
    if arch in ("x86_64", "amd64"):
        url = ("https://github.com/cloudflare/cloudflared/releases/latest/"
               "download/cloudflared-linux-amd64")
    elif "aarch64" in arch or "arm64" in arch:
        url = ("https://github.com/cloudflare/cloudflared/releases/latest/"
               "download/cloudflared-linux-arm64")
    else:
        url = ("https://github.com/cloudflare/cloudflared/releases/latest/"
               "download/cloudflared-linux-amd64")
    print(f"[tunnel] downloading cloudflared ({arch})...")
    urllib.request.urlretrieve(url, path)
    os.chmod(path, 0o755)
    return path

def start_cloudflare_tunnel(port, timeout=40):
    binary = get_cloudflared()
    proc = subprocess.Popen(
        [binary, "tunnel", "--no-autoupdate", "--url", f"http://localhost:{port}"],
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
    url = None
    url_re = re.compile(r"https://[-a-z0-9]+\.trycloudflare\.com")
    start = time.time()
    while time.time() - start < timeout:
        line = proc.stdout.readline()
        if not line:
            if proc.poll() is not None:
                break
            continue
        m = url_re.search(line)
        if m:
            url = m.group(0)
            break
    def _drain():
        for _ in proc.stdout:
            pass
    threading.Thread(target=_drain, daemon=True).start()
    return url, proc

PUBLIC_URL = None
tunnel_proc = None
for attempt in range(2):
    try:
        PUBLIC_URL, tunnel_proc = start_cloudflare_tunnel(PORT)
        if PUBLIC_URL:
            print(f"[tunnel] live: {PUBLIC_URL}")
            break
        time.sleep(3)
    except Exception as e:
        print(f"[tunnel] attempt {attempt+1} failed: {str(e)[:80]}")
        time.sleep(3)

print("\n" + "=" * 66)
if PUBLIC_URL:
    print(" CorX3.8-27B API IS LIVE")
    print("=" * 66)
    print(f" Public URL : {PUBLIC_URL}/v1")
    print(f" Health     : {PUBLIC_URL}/health")
    print(f" Auth       : {'NONE (open)' if NO_AUTH else 'Bearer ' + API_KEY}")
    print(f" Model      : corx3.8")
else:
    print(" TUNNEL FAILED — API is live LOCALLY")
print("=" * 66)
print("Paste the Public URL (without /v1) into the CorX chat Settings.")

_KEEP_ALIVE = tunnel_proc

# ---------------- BLOCK FOREVER so the cell never completes ----------------
while True:
    try:
        time.sleep(15)
        if tunnel_proc is not None and tunnel_proc.poll() is not None:
            print("[tunnel] cloudflared exited — restarting...")
            new_url, tunnel_proc = start_cloudflare_tunnel(PORT)
            if new_url:
                PUBLIC_URL = new_url
                print(f"[tunnel] back up: {PUBLIC_URL}/v1")
            _KEEP_ALIVE = tunnel_proc
    except KeyboardInterrupt:
        print("stopped by user.")
        break
    except Exception as _e:
        print(f"[keepalive] {type(_e).__name__}: {_e}")
        time.sleep(5)

Which GPU? CorX3.8 needs roughly 24 GB of GPU memory in 4-bit, or about 54 GB in full bf16. You don't have to choose: PRECISION = "auto" measures the card at start-up and loads full bf16 when there is room for it (RTX PRO 6000 Blackwell, H100, A100 80GB), dropping to 4-bit only on smaller cards. Force it either way with PRECISION = "bf16" or "4bit".

On a single card that fits the model, it is pinned to GPU 0 rather than left to device_map="auto" — that keeps the offload machinery out of the picture entirely. The cell prints which precision it chose and confirms the whole model is resident on the GPU; if anything lands on CPU or disk it says so, because that is what makes the first token take minutes.

On a brand-new card (Blackwell is sm_120) PyTorch must have kernels built for it, or every CUDA call fails with “no kernel image is available”. The script checks and, if your build is too old, prints the install line: pip install -U --index-url https://download.pytorch.org/whl/cu128 torch.

Step 2 · Connect

Paste the URL into the chat.

Wait for “API IS LIVE”
When the cell prints CorX3.8-27B API IS LIVE, it shows a Public URL like https://something.trycloudflare.com/v1. Copy it without the /v1 — the chat adds that for you.
Open the chat and its settings
Go to the chat and click the model name at the top (or the gear icon) to open Model & connection. With CorX3.8 selected, paste the URL into Server address and press Save.
Watch the status dot
The dot by the title turns green · Online when it connects (it checks the server’s /health). Red means it can’t reach the server — check the cell is still running and the URL is current.
Leave the cell running
The server lives as long as that cell runs. The code blocks on purpose so the notebook doesn’t tear it down, and restarts the tunnel if it drops.

Or bring your own model. The same Model & connection panel has a provider picker: pick OpenRouter, Claude, DeepSeek or OpenAI, paste that provider’s API key (stored in your browser only, sent only to that provider), and choose a model. The whole agent — search, terminal, files, plan — then runs on that model instead. Two of these call the provider straight from the browser and just work: OpenRouter (one key, most models) and Claude. DeepSeek and OpenAI block direct browser calls, so they’re marked proxy and will hit a CORS wall unless you front them yourself — use OpenRouter to reach GPT or DeepSeek models from the browser.

Step 3 · Agent tools

It runs code, files and a plan on its own.

There is no toggle to switch on — every conversation can use tools: run Python, make and edit files, install packages and plan a task. CorX3.8 decides for itself when a tool would help and calls it without asking permission first. The tools run in a real Python sandbox inside your browser tab — not on the model server — so you can watch every command in the Terminal panel and nothing touches your computer outside the tab.

What it can do

Run Python and see the real output in the Terminal
Create, edit, save and delete files in the Files panel
Install Python packages with micropip
Publish a plan and tick off each step as it works
Take a zip or file you upload and work on it, with Python's own zipfile module
Search the web and fetch a page's real content, on its own initiative
Recall relevant bits of your other saved chats
Pull real pictures into the chat with find_image (Wikimedia Commons, credited)
Read its own traceback when code fails and try to fix it on the next round

The panel, search & uploads

Open the work panel with the panel button by the title. It has three tabs: Terminal (commands as they run, and you can type Python yourself), Files (everything the sandbox holds, shared across every chat in this tab — click one to edit, save or delete), and Plan (the live checklist).

A real web search runs through /api/search (DuckDuckGo, no key) with a sliding row of site logos while it searches, then a Searched the web dropdown you can open to see every link and snippet. /api/fetch reads a page's actual content, including any code shown on it.

The upload button in the composer stages files — including .zip archives — into the sandbox at /work. Ask the agent to unzip, read or transform them.

Honest note on tools. The model server is a plain chat endpoint with no built-in tool-calling, so every system prompt teaches CorX3.8 a small text format for asking to run a tool. Whether it uses tools well depends on the model following that format — it is a Patois-focused fine-tune, not a dedicated tool-use model, so expect it to be better at conversation than at long autonomous chains. If it ever ignores a tool it should have used, ask it directly and it will call it.

Step 4 · Effort

Low to Max — a real dial, not a label.

The Effort selector in the composer changes actual request parameters: how many tokens the model is allowed per reply, how many tool-call rounds the agent gets before it must stop, how many search results it pulls back, and how hard the system prompt tells it to think, weigh alternatives, verify its own steps and search when unsure. Higher effort genuinely takes longer and spends more tokens — that trade is the point, for a task that is actually hard.

Effort levels and what each one changes
Level	Tokens	Tool rounds	Search results
Low	512	3	3
Medium	1,024	5	4
High	2,048	8	6
Extra	4,096	14	8
Max	6,144	24	10

Extra and Max also add a second pass: after producing code or a solution, the model is told to stop and deliberately review its own work for bugs or missed cases — as if critiquing someone else's pull request — before presenting it as done, and to say so plainly and correct itself if it realises mid-answer that something it already said was wrong.

Effort is set per conversation and remembered when you come back to that chat.

Conversations & profile

Saved in this browser, nowhere else.

Conversations

Every chat lives in the sidebar — switch between them, delete one, or start a New chat. Messages, the plan, the agent toggle and the effort level are all saved per conversation and restored exactly as you left them after a refresh, a closed tab, or days away. The Python sandbox itself is not per-conversation — it is one real Python process for the whole tab, so its files are shared across every chat you have open.

Interrupted runs resume themselves

If you refresh mid-task, the chat notices on reload and starts picking the run back up automatically after a few seconds — Dismiss if you'd rather not. Because the Python sandbox resets on every reload, the agent is told plainly that it did, and to check what actually exists with list_files before assuming anything survived.

Profile

Click your name at the bottom of the sidebar to set a name and a photo. Both are saved locally and used to label your messages — never sent anywhere, and not part of what the model itself receives.

Cross-chat memory

With Memory on in your profile, the agent can call search_memory to grep your other saved conversations for relevant context — a real keyword search over what is in this browser, not a hidden server-side record.

Honest limits

What this set-up is, and isn’t.

The URL is ephemeral. A Cloudflare quick tunnel gives a new address every restart, and the notebook recycles. When the chat goes Offline, re-run the cell and paste the new URL. A permanent public chat needs a persistent host and a fixed domain.
One GPU, one job at a time. The server queues requests and returns a clear “busy” rather than faking concurrency. Fine for a demo; not for real traffic.
Open by default. Keyless means anyone with the URL can use your GPU while it is up. Set NO_AUTH = False and an API_KEY to lock it down, then put the key in settings.
The model can be wrong. CorX3.8 is an open research model. Don’t rely on it for facts without checking.

Optional · Heretic mode

The uncensored build, at full precision.

Same chat, same tools, a different set of weights. Heretic runs Qwen3.8-27B-Heretic-Abliterated — the same base CorX3.8 was built on, with the refusal direction removed. Run this cell instead of the one above and paste the URL it prints into the same box.

Read this first. Abliterated means the part of the network that produces refusals has been edited out. It will answer almost anything asked of it, including things it should not, and it is no more accurate than the normal build — removing the refusals does not add knowledge. There is no filter left except you.

The tunnel URL is public and keyless by default. Anyone holding it can use this model on your GPU, under your connection. If you run Heretic, set NO_AUTH = False and an API_KEY, and do not paste the URL anywhere. Everything it generates is your responsibility, not the model’s.

"""
================================================================
 CorX3.8-27B "HERETIC" — uncensored API server
 Build 14 — molab (marimo cloud notebook), RTX PRO 6000 Blackwell
 CorX Labs / Nathan

 ---- why it was slow, and what changed ----

 * PRECISION = "auto" means "the largest variant that fits". On a
   92 GB card that is F16 — and F16 is the SLOWEST thing you can
   run, not the best. Generating a token reads every weight of a
   dense model, so speed is bandwidth-bound: at ~1.8 TB/s, 54 GB
   of F16 weights caps you around 33 tok/s, while 22 GB of Q6_K
   caps around 82. More VRAM does not mean run the biggest file.
   The default is now "fast", which picks the smallest variant
   whose quality still holds (Q6_K first, then Q5_K_M, Q8_0 ...),
   roughly 2.5x faster than F16 for a difference you cannot see on
   a 27B. "auto" and "full" still do what they always did.

 * The tuning flags were one all-or-nothing list, so an older
   llama.cpp rejecting a single flag dropped the run to stock
   settings — losing the batch sizes too. They are tiers now, and
   it steps down one at a time.

 * Added: quantised KV cache (-ctk/-ctv q8_0, the other bandwidth
   cost after the weights, and it grows with conversation length),
   and single-GPU pinning so the multi-GPU scheduling path is
   skipped entirely.

 * It now says out loud when it is NOT on CUDA. Vulkan works but
   is materially slower on an NVIDIA card, and that was invisible.

 ---- what build 11 got wrong ----

 * "The source directory /marimo/llama.cpp does not appear to
   contain CMakeLists.txt". clone_llama() asked the wrong question:
   `if not os.path.isdir(LLAMA_SRC)`. build_vulkan() calls it first,
   and when that clone failed it left an empty directory behind — so
   the CUDA build that followed saw a directory, skipped cloning,
   and handed cmake an empty folder. It now tests for CMakeLists.txt
   (the file that proves a checkout), deletes anything incomplete,
   retries three times, and leaves nothing behind when it gives up,
   so a failed run cannot poison the next one.

 * One failing rung killed the whole cell. A CUDA build that threw
   ended the run even though the NVIDIA Vulkan driver was sitting
   right there unused. Every rung is now attempted independently;
   only an exhausted ladder is fatal, and it then reports each
   reason it collected.

 * A stale CMakeCache.txt from an earlier, different configure
   poisons every later one with a message that is not obviously
   about the cache. A failed configure now wipes the build
   directory and tries once more.

 * If nvcc rejects the host compiler as too new ("unsupported GNU
   version"), the build retries with an older g++ instead of
   stopping.

 ---- what build 10 got wrong ----

 * "No GPU backend could be set up" on a box that HAS the NVIDIA
   Vulkan driver. Two separate causes, both fixed:

   - The CUDA last resort died on "OpenPGP signature verification
     failed". NVIDIA publishes no Debian 13 (trixie) repo, so the
     script borrows debian12 — but Debian 13's apt 3.0 then rejects
     that repo's InRelease signature, and the per-line error check
     dutifully marked every candidate unreachable. The repo is now
     marked trusted (trust_cuda_sources) right after the keyring is
     installed, so apt stops refusing NVIDIA's own HTTPS host, and a
     signature warning is no longer treated as "not reachable".

   - The Vulkan path should have won before CUDA was ever reached,
     since the driver is present. A prebuilt Ubuntu binary that
     will not start here — one missing shared library — used to be
     abandoned outright. ensure_runnable() now reads ldd, installs
     the library apt actually knows, and retries, instead of
     falling through a fixable link error to the CUDA mess.

 ---- what build 9 got wrong ----

 * The Vulkan build died at configure time on missing SPIR-V
   headers. glslc alone is not enough: ggml-vulkan includes
   <spirv/unified1/spirv.hpp>, which comes from the separate
   spirv-headers package. build_vulkan() now installs the whole
   shader toolchain, and vendors the headers straight from Khronos
   if the distro has no package for them.

 * Every CUDA repo candidate was rejected — debian12, ubuntu2404
   and ubuntu2204 all exist, so that was a bug here, not on
   NVIDIA's server. The check tested `"404" in blob or "Err:" in
   blob` against the entire apt-get update output, and the NVIDIA
   domain appears in ordinary successful Get: lines, so one
   unrelated broken source anywhere in sources.list condemned all
   three. Errors are now matched per line, and availability is
   settled by `apt-cache policy` rather than inferred from log
   text.

 * cmake failures raised a bare CalledProcessError whose message
   was just the argv, so the actual reason was only visible by
   scrolling. Build steps now print the tail of their output.

 * PRECISION = "auto" weighed VRAM only. On a 96 GB card that
   picks F16 (~54 GB) and then dies at the disk check. It now
   requires the variant to fit VRAM *and* free disk.

 Paste this into ONE marimo cell and run it. It serves a GGUF
 repo over an OpenAI-compatible API with open CORS, /health, and
 a public Cloudflare URL.

 ---- what build 8 got wrong, and what changed ----

 * The run died at "No GPU backend could be set up" while its own
   error text said "Vulkan: nvidia". The real NVIDIA Vulkan driver
   was present the whole time. try_prebuilt() returned None from
   inside its asset loop, so a single missing asset ended the
   search, and there was no source-build fallback behind it. There
   is now: build_vulkan() needs glslc + libvulkan-dev (~50 MB),
   not nvcc (~1 GB).

 * install_cuda_toolkit() break-ed out of the distro loop after
   the first candidate, so once debian13 failed it never tried
   ubuntu2404 or debian12. NVIDIA publishes no repo for Debian 13
   (trixie) at all — apt-get update was 404ing, which is the real
   reason all four packages were "unable to locate", and that
   output was sent to DEVNULL so it never appeared. Debian 13 now
   maps onto debian12 and apt errors are shown.

 * _works() ignored the exit code, so a binary missing libgomp1 or
   libcurl4 counted as working and failed later at model load.

 ---- what is molab-specific ----

 * The cell no longer blocks. In marimo a blocking cell freezes
   the whole notebook. The server runs in background threads and
   subprocesses, which live as long as the kernel does, so you can
   keep using other cells. Set BLOCK_CELL = True for the old
   behaviour.
 * Re-running the cell kills whatever is holding the ports first,
   instead of dying on "address already in use".
 * pip may not exist in a uv-managed molab venv, so package
   installs try pip, then uv, then ensurepip.
 * Free disk is checked against the actual size of the GGUF before
   downloading. A notebook container is not where a 54 GB file
   fits, which is why PRECISION now defaults to "auto".
 * Nothing here uses `global`. marimo compiles a cell into a
   function, so shared mutable state lives in dicts and lists.

 Abliterated means the refusal direction has been removed from the
 weights. It answers almost anything, including things it should
 not, and it is no more accurate for it. You are the only filter
 left.
================================================================
"""

import os

os.environ.setdefault("CUDA_VISIBLE_DEVICES", "0")

import asyncio
import subprocess
import sys
import json
import time
import threading
import re
import platform
import shutil
import tarfile
import zipfile
import glob
import signal
import urllib.request
import urllib.error

# ═══════════════════════════ SETTINGS ═══════════════════════════
MODEL_REPO = "0bserverx/Qwen3.8-27B-Heretic-Abliterated-Uncensored-GGUF"
MODEL_ID = "corx3.8-heretic"
HF_TOKEN = ""

PORT = 8000            # public port, what the tunnel points at
LLAMA_PORT = 8081      # llama-server, localhost only

# Context window. 4096 is too small for this agent — the system prompt, the
# tool list and a few turns of history do not fit, and llama.cpp then silently
# drops the oldest messages. On a GPU the prefill cost of a bigger window is
# milliseconds. Raise it if you have the VRAM.
CTX = 16384
MAX_NEW_TOKENS_CAP = 8192

# Which GGUF to pull. THIS IS THE SINGLE BIGGEST SPEED LEVER, so it is worth
# understanding rather than guessing at.
#
# Generating a token means reading EVERY weight of a dense model, so generation
# speed is bandwidth-bound, not compute-bound. On this card (~1.8 TB/s GDDR7)
# the ceiling is roughly bandwidth / model-size:
#
#     F16    54 GB  ->  ~33 tok/s ceiling   (~23 realistic)
#     Q8_0   29 GB  ->  ~62 tok/s           (~43)
#     Q6_K   22 GB  ->  ~82 tok/s           (~57)
#     Q4_K_M 16 GB  -> ~112 tok/s           (~79)
#
# So "more VRAM" does NOT mean "run the biggest file" — on a 92 GB card the
# biggest file is simply the slowest one. F16 is ~2.5x slower than Q6_K for a
# quality difference you cannot see on a 27B: Q8_0 is effectively lossless and
# Q6_K is very close to it.
#
#   "fast" — best speed for near-identical quality. THE DEFAULT, and what you
#            want unless you have a specific reason otherwise.
#   "auto" — largest variant that fits VRAM and disk. Maximum quality, slowest.
#   "full" — F16/BF16, nothing quantised. Slowest of all.
#   "q8_0" / "q6_k" / "q4_k_m" — force one by name.
PRECISION = "fast"

# Preference order for "fast": near-lossless first, then progressively smaller.
# F16/BF16 sit at the end because they are only worth taking when a repo ships
# nothing else.
FAST_ORDER = ["q6_k", "q5_k_m", "q8_0", "q5_k_s", "q4_k_m",
              "q4_k_s", "q5_0", "q4_0", "q3_k_m", "bf16", "f16", "fp16", "f32"]

# Memory bandwidth in GB/s, used only to print an honest expected ceiling.
# Roughly right for a Blackwell workstation card; adjust if yours differs.
GPU_BANDWIDTH_GBS = 1800

REQUIRE_GPU = True     # refuse to serve a CPU-bound 27B. Leave this True.

# How to get a GPU-capable llama-server.
#   "auto"   — CUDA if nvcc exists, else prebuilt CUDA, else Vulkan against
#              the real NVIDIA driver (prebuilt, then compiled), else install
#              a CUDA toolkit and compile. Never falls back to the CPU.
#   "cuda" / "vulkan" — insist on one. Software Vulkan is refused either way.
BACKEND = "auto"

INSTALL_CUDA_TOOLKIT = True   # may this cell apt-get a toolkit as a last resort
NO_AUTH = True
API_KEY = "corx-key"
LOAD_TIMEOUT = 1800           # seconds to wait for weights to load
# The server runs in a DETACHED process, so the cell always finishes and the
# notebook stays usable. Set True to tear the running one down and start over.
RESTART_RUNNER = False
MIN_FREE_GB = 5               # headroom to leave on the disk after download
# ════════════════════════════════════════════════════════════════

BEST_TO_WORST = ["f32", "bf16", "f16", "fp16", "q8_0", "q6_k",
                 "q5_k_m", "q5_k_s", "q5_0", "q4_k_m", "q4_k_s", "q4_0", "q3_k_m"]

# Rough VRAM for weights + a 16k KV cache on a 27B, in GB.
NEEDS_GB = {"f32": 118, "bf16": 62, "f16": 62, "fp16": 62, "q8_0": 36, "q6_k": 29,
            "q5_k_m": 25, "q5_k_s": 24, "q5_0": 24, "q4_k_m": 21, "q4_k_s": 20,
            "q4_0": 20, "q3_k_m": 17}

IS_ROOT = (os.geteuid() == 0)
_APT_UPDATED = [False]


def sh(cmd, **kw):
    return subprocess.run(cmd, shell=isinstance(cmd, str), **kw)


def apt(pkgs, quiet=True):
    """Install packages, never with recommends.

    --no-install-recommends is not tidiness: libvulkan-dev recommends
    mesa-vulkan-drivers, which is lavapipe, which is the software rasteriser
    this whole script exists to refuse."""
    if not IS_ROOT:
        return False
    if not _APT_UPDATED[0]:
        sh("apt-get -qq update", check=False,
           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        _APT_UPDATED[0] = True
    r = sh(["apt-get", "-qq", "install", "-y", "--no-install-recommends"] + list(pkgs),
           check=False,
           stdout=subprocess.DEVNULL if quiet else None,
           stderr=subprocess.DEVNULL if quiet else None)
    return r.returncode == 0


# ---------------- PACKAGES (molab may be uv-managed, with no pip) ----------
def ensure_packages(pkgs):
    missing = []
    for mod, pkg in pkgs:
        try:
            __import__(mod)
        except ImportError:
            missing.append(pkg)
    if not missing:
        return
    print(f"[env] installing {', '.join(missing)} ...")
    attempts = [
        [sys.executable, "-m", "pip", "install", "-q", "-U"] + missing,
        ["uv", "pip", "install", "-q", "--python", sys.executable] + missing,
    ]
    for cmd in attempts:
        try:
            if sh(cmd, check=False, stdout=subprocess.DEVNULL,
                  stderr=subprocess.DEVNULL).returncode == 0:
                return
        except FileNotFoundError:
            continue
    # Last resort: bootstrap pip into this interpreter, then retry.
    sh([sys.executable, "-m", "ensurepip", "--upgrade"], check=False,
       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    r = sh([sys.executable, "-m", "pip", "install", "-q", "-U"] + missing, check=False)
    if r.returncode != 0:
        raise RuntimeError(
            f"Could not install {', '.join(missing)}. Add them to the notebook's "
            "package list (molab: the packages panel) and re-run this cell.")


ensure_packages([("huggingface_hub", "huggingface_hub"),
                 ("fastapi", "fastapi"),
                 ("uvicorn", "uvicorn[standard]"),
                 ("httpx", "httpx")])


# ---------------- RE-RUN SAFETY: free the ports first ----------------
# marimo re-executes a cell on every edit. Without this, the second run dies
# on "address already in use" and leaves the first server orphaned.
def free_port(port):
    killed = []
    for cmd in (["fuser", "-k", f"{port}/tcp"], ["lsof", "-ti", f":{port}"]):
        try:
            if cmd[0] == "fuser":
                sh(cmd, check=False, stdout=subprocess.DEVNULL,
                   stderr=subprocess.DEVNULL)
            else:
                out = subprocess.check_output(cmd, text=True,
                                              stderr=subprocess.DEVNULL).split()
                for pid in out:
                    os.kill(int(pid), signal.SIGKILL)
                    killed.append(pid)
        except Exception:
            continue
    if killed:
        print(f"[env] freed port {port} (killed {', '.join(killed)})")
        time.sleep(1)


for _p in (PORT, LLAMA_PORT):
    free_port(_p)


# ---------------- LOOK AT THE ACTUAL CARD ----------------
def gpu_info():
    """Name, VRAM in GB, compute capability — straight from nvidia-smi, so
    this script needs no torch."""
    try:
        out = subprocess.check_output(
            ["nvidia-smi", "--query-gpu=name,memory.total,compute_cap",
             "--format=csv,noheader,nounits"], text=True).strip().splitlines()
        name, mem, cap = [x.strip() for x in out[0].split(",")]
        return name, float(mem) / 1024.0, cap
    except Exception as e:
        raise RuntimeError(
            f"No NVIDIA GPU visible ({type(e).__name__}). This needs a CUDA card "
            "— a 27B on CPU is minutes per reply, which is what the front-end "
            "reports as a timeout. In molab, check that the notebook was started "
            "on a GPU runtime."
        ) from e


GPU_NAME, VRAM_GB, COMPUTE_CAP = gpu_info()
CUDA_ARCH = COMPUTE_CAP.replace(".", "")      # "12.0" -> "120" (Blackwell)
print(f"GPU: {GPU_NAME} — {VRAM_GB:.0f} GB VRAM, compute {COMPUTE_CAP} "
      f"(sm_{CUDA_ARCH}) | root={IS_ROOT}")


# ════════════════════════════════════════════════════════════════
#  Getting a GPU-capable llama-server
#
#  Ground truth, because this is where every previous run died:
#
#   * A notebook container has the NVIDIA *driver* but no CUDA
#     *toolkit*. nvidia-smi works, nvcc does not exist, and a
#     source build dies at "CUDA Toolkit not found".
#   * No PyPI wheel ships nvcc. nvidia-cuda-nvcc-cu12 contains
#     ptxas and nvvm only.
#   * llama.cpp publishes no Linux CUDA binary. It does publish a
#     Linux Vulkan one, and Vulkan is a genuine GPU path *if* it
#     binds to the real NVIDIA ICD. The trap is mesa-vulkan-drivers
#     (lavapipe), a software rasteriser that still reports every
#     layer "offloaded to GPU".
#   * If no Vulkan binary is published for this release, compiling
#     one is cheap. That missing fallback is what turned a solved
#     problem into a failed run.
# ════════════════════════════════════════════════════════════════
BIN_DIR = os.path.abspath("./llama-bin")
SERVER_BIN = os.path.join(BIN_DIR, "llama-server")
LLAMA_SRC = os.path.abspath("./llama.cpp")


def install_runtime_libs():
    """Shared objects a *prebuilt* llama-server links against. Names differ
    across distros (libcurl4 vs libcurl4t64 on Debian 13), so install them one
    at a time — one wrong name must not fail the rest."""
    for p in ("libgomp1", "libcurl4t64", "libcurl4", "libvulkan1"):
        apt([p])


def _works(path):
    """Does this binary actually run here?

    The old version only caught exceptions, so a binary that started and died
    on a missing .so returned True. Check the exit code, and surface the last
    line of output so a failure is diagnosable rather than mysterious."""
    if not (os.path.isfile(path) and os.access(path, os.X_OK)):
        return False
    try:
        r = subprocess.run([path, "--version"], capture_output=True, text=True,
                           timeout=120,
                           env=dict(os.environ, LD_LIBRARY_PATH=os.path.dirname(path)))
    except Exception as e:
        print(f"[build]   won't start: {type(e).__name__}: {e}")
        return False
    if r.returncode != 0:
        out = ((r.stderr or "") + (r.stdout or "")).strip().splitlines()
        print(f"[build]   exited {r.returncode}: "
              f"{out[-1][:200] if out else '(no output)'}")
        return False
    return True


# A prebuilt binary built on Ubuntu often refuses to start on this container
# for one dull reason: a shared library it links against is absent or named
# differently here. That returns _works() == False and the whole prebuilt path
# is abandoned — which is very likely why the Vulkan binary was skipped and the
# run fell through to CUDA. So read ldd, install what is actually missing, and
# retry, instead of giving up on a fixable link error.
_SONAME_PKG = {
    "libvulkan.so": ["libvulkan1"],
    "libgomp.so": ["libgomp1"],
    "libcurl.so": ["libcurl4t64", "libcurl4"],
    "libstdc++.so": ["libstdc++6"],
    "libgcc_s.so": ["libgcc-s1"],
    "libgfortran.so": ["libgfortran5"],
    "libnuma.so": ["libnuma1"],
    "libopenblas.so": ["libopenblas0-pthread", "libopenblas0"],
    "libgomp1.so": ["libgomp1"],
}


def _missing_libs(path):
    try:
        r = subprocess.run(["ldd", path], capture_output=True, text=True,
                           env=dict(os.environ, LD_LIBRARY_PATH=os.path.dirname(path)))
    except Exception:
        return []
    miss = []
    for line in (r.stdout + r.stderr).splitlines():
        m = re.search(r"(\S+)\s*=>\s*not found", line)
        if m:
            miss.append(m.group(1))
    return miss


def ensure_runnable(path):
    """_works(), and if it fails on a missing .so, install it and try once more."""
    if _works(path):
        return True
    miss = _missing_libs(path)
    if not miss:
        return False
    print(f"[build]   binary needs libraries this box lacks: {', '.join(miss)}")
    for so in miss:
        stem = so.split(".so")[0] + ".so"
        for pkg in _SONAME_PKG.get(stem, []):
            if apt_has(pkg) and apt([pkg]):
                break
    still = _missing_libs(path)
    if still:
        print(f"[build]   still missing after install: {', '.join(still)} "
              "(no apt package maps to these).")
    return _works(path)


def _install_binary(src_dir):
    """Copy a built or extracted bin directory into BIN_DIR and check it runs.
    Also pulls a sibling lib/ if the archive uses that layout."""
    os.makedirs(BIN_DIR, exist_ok=True)
    dirs = [src_dir]
    sib = os.path.join(os.path.dirname(src_dir), "lib")
    if os.path.isdir(sib):
        dirs.append(sib)
    for d in dirs:
        for f in os.listdir(d):
            s = os.path.join(d, f)
            if os.path.isfile(s):
                shutil.copy2(s, os.path.join(BIN_DIR, f))
    if os.path.exists(SERVER_BIN):
        os.chmod(SERVER_BIN, 0o755)
    return ensure_runnable(SERVER_BIN)


# ---------------------------------------------------------------- CUDA toolkit
def find_nvcc():
    p = shutil.which("nvcc")
    if p:
        return p
    roots = [os.environ.get("CUDA_HOME"), os.environ.get("CUDA_PATH"),
             "/usr/local/cuda"] + sorted(glob.glob("/usr/local/cuda-*"), reverse=True)
    for r in roots:
        if not r:
            continue
        c = os.path.join(r, "bin", "nvcc")
        if os.path.isfile(c) and os.access(c, os.X_OK):
            return c
    return None


def nvcc_version(nvcc):
    try:
        out = subprocess.check_output([nvcc, "--version"], text=True)
        m = re.search(r"release (\d+)\.(\d+)", out)
        return (int(m.group(1)), int(m.group(2))) if m else None
    except Exception:
        return None


def distro_candidates():
    """apt repo names to try, in order.

    NVIDIA publishes no repo for Debian 13 (trixie). Your log shows debian13
    accepting the keyring and then failing to locate a single package: apt-get
    update was 404ing against a directory that does not exist. Map trixie onto
    debian12 and keep the Ubuntu repos behind it."""
    try:
        osr = dict(l.strip().split("=", 1) for l in open("/etc/os-release") if "=" in l)
        did = osr.get("ID", "").strip('"')
        ver = osr.get("VERSION_ID", "").strip('"').replace(".", "")
    except Exception:
        did, ver = "", ""
    alias = {"debian13": "debian12", "debian14": "debian12",
             "ubuntu2410": "ubuntu2404", "ubuntu2504": "ubuntu2404",
             "ubuntu2510": "ubuntu2404", "ubuntu2604": "ubuntu2404"}
    native = f"{did}{ver}"
    out = []
    for d in (alias.get(native, native), "debian12", "ubuntu2404", "ubuntu2204"):
        if d and d not in out:
            out.append(d)
    return out


def apt_has(pkg):
    """Is this package actually installable right now? apt-cache policy prints
    'Candidate: (none)' when it is not, which beats guessing from log text."""
    try:
        out = subprocess.check_output(["apt-cache", "policy", pkg], text=True,
                                      stderr=subprocess.DEVNULL)
    except Exception:
        return False
    m = re.search(r"Candidate:\s*(\S+)", out)
    return bool(m and m.group(1) != "(none)")


def drop_cuda_sources():
    """Remove NVIDIA apt sources so a failed candidate cannot poison the next
    apt-get update."""
    for f in glob.glob("/etc/apt/sources.list.d/cuda*"):
        try:
            os.remove(f)
        except OSError:
            pass
    sh("apt-get -qq update", check=False, stdout=subprocess.DEVNULL,
       stderr=subprocess.DEVNULL)


def trust_cuda_sources():
    """Mark the NVIDIA CUDA apt source trusted.

    This is the fix for "OpenPGP signature verification failed" on this box.
    NVIDIA publishes no Debian 13 (trixie) repo, so we borrow debian12; but
    Debian 13's apt (3.0) then rejects the InRelease signature and every
    candidate looks unreachable. The keyring is installed and the repo is
    NVIDIA's own host over HTTPS, so tell apt to trust it rather than fail on a
    cross-release signature check. Handles both the one-line (.list) and the
    deb822 (.sources) formats a keyring might drop."""
    touched = False
    for f in glob.glob("/etc/apt/sources.list.d/cuda*"):
        try:
            txt = open(f).read()
        except OSError:
            continue
        if f.endswith(".sources"):
            # deb822: add "Trusted: yes" to each stanza (blank-line separated)
            # that does not already carry it.
            stanzas = re.split(r"\n\s*\n", txt.strip())
            fixed = []
            for st in stanzas:
                if st.strip() and not re.search(r"(?im)^\s*trusted:", st):
                    st = st.rstrip() + "\nTrusted: yes"
                    touched = True
                fixed.append(st)
            new = "\n\n".join(fixed) + "\n"
        else:
            new_lines = []
            for line in txt.splitlines():
                s = line.strip()
                if s.startswith("deb") and "trusted=yes" not in s:
                    if "[" in s:
                        s = s.replace("[", "[trusted=yes ", 1)
                    else:
                        s = s.replace("deb ", "deb [trusted=yes] ", 1)
                    touched = True
                new_lines.append(s)
            new = "\n".join(new_lines) + "\n"
        try:
            open(f, "w").write(new)
        except OSError:
            pass
    return touched


def install_cuda_toolkit():
    """Install just the pieces llama.cpp needs: nvcc, cudart headers, cuBLAS
    dev. About a third of the full toolkit. Needs root."""
    if not IS_ROOT:
        print("[cuda] no toolkit and not root, so it cannot be installed here.")
        return None

    apt(["ca-certificates", "curl", "gnupg"])
    try:
        arch_i = int(CUDA_ARCH)
    except ValueError:
        arch_i = 0
    # sm_120 (Blackwell, incl. RTX PRO 6000) needs 12.8 or newer. Older cards
    # can fall further back.
    versions = (["13-0", "12-9", "12-8"] if arch_i >= 120
                else ["13-0", "12-9", "12-8", "12-6", "12-4"])

    for d in distro_candidates():
        url = ("https://developer.download.nvidia.com/compute/cuda/repos/"
               f"{d}/x86_64/cuda-keyring_1.1-1_all.deb")
        print(f"[cuda] trying NVIDIA's apt repo for {d} ...")
        deb = "/tmp/cuda-keyring.deb"
        try:
            urllib.request.urlretrieve(url, deb)
        except Exception as e:
            print(f"[cuda]   no keyring for {d} ({type(e).__name__})")
            continue
        if sh(["dpkg", "-i", deb], check=False, stdout=subprocess.DEVNULL,
              stderr=subprocess.DEVNULL).returncode != 0:
            print(f"[cuda]   keyring would not install for {d}")
            continue

        # Trust the repo BEFORE updating: on Debian 13 the InRelease signature
        # is rejected ("OpenPGP signature verification failed"), so without this
        # the update below fails and the candidate is discarded.
        trust_cuda_sources()

        # build 9 tested `"404" in blob or "Err:" in blob` against the WHOLE
        # apt-get update output. The NVIDIA domain appears in normal successful
        # Get: lines, so any unrelated broken source anywhere in sources.list
        # made every candidate look dead — which is why debian12, ubuntu2404
        # and ubuntu2204 were all rejected despite all three existing. Match
        # per line, and only on lines that are actually errors for this repo.
        upd = sh("apt-get update -o Acquire::Retries=2", check=False,
                 capture_output=True, text=True)
        lines = ((upd.stdout or "") + (upd.stderr or "")).splitlines()
        # A signature warning is no longer fatal now the repo is trusted; only a
        # genuine fetch failure (404/could-not-connect) means "not reachable".
        broken = [l for l in lines
                  if l.strip().startswith(("Err:", "E:"))
                  and "developer.download.nvidia.com" in l
                  and "signature" not in l.lower()
                  and "signed" not in l.lower()]
        if broken:
            print(f"[cuda]   {d} repo is not reachable:")
            for l in broken[:3]:
                print("           " + l.strip()[:150])
            drop_cuda_sources()
            continue
        _APT_UPDATED[0] = True

        # Ask apt what it can actually see, rather than inferring it from log
        # text. "Candidate: (none)" is unambiguous; parsing update output is not.
        for ver in versions:
            if not apt_has(f"cuda-nvcc-{ver}"):
                continue
            pkgs = [f"cuda-nvcc-{ver}", f"cuda-cudart-dev-{ver}",
                    f"libcublas-dev-{ver}", f"cuda-cccl-{ver}"]
            pkgs = [p for p in pkgs if apt_has(p)]
            print(f"[cuda] installing {' '.join(pkgs)} (about 1 GB, once)...")
            if apt(pkgs):
                nv = find_nvcc()
                if nv:
                    print(f"[cuda] toolkit installed: {nv}")
                    return nv
        # Fall through to the NEXT candidate, rather than break-ing out of the
        # loop the way build 8 did.
        print(f"[cuda]   {d} is reachable but ships none of {', '.join(versions)}.")
        drop_cuda_sources()

    print("[cuda] could not install a CUDA toolkit from NVIDIA's apt repo.")
    return None


# -------------------------------------------------------------------- builds
def clone_llama():
    """Get a llama.cpp source tree, and be certain it is actually a source tree.

    This is what broke build 10. The check was `if not os.path.isdir(LLAMA_SRC)`,
    which is not the same question. build_vulkan() calls this first; when its
    clone failed it left an empty /marimo/llama.cpp behind, and the CUDA build
    that followed saw a directory, skipped the clone, and handed cmake a folder
    with nothing in it — "does not appear to contain CMakeLists.txt". Test for
    the file that proves the checkout, not for the directory."""
    apt(["build-essential", "cmake", "git", "ccache", "pkg-config"])
    if not shutil.which("cmake"):
        raise RuntimeError("cmake is not installed and could not be installed "
                           "(root=%s). Nothing can be compiled here." % IS_ROOT)
    if not shutil.which("git"):
        raise RuntimeError("git is not installed and could not be installed "
                           "(root=%s)." % IS_ROOT)

    marker = os.path.join(LLAMA_SRC, "CMakeLists.txt")
    if os.path.isfile(marker):
        return

    tail = []
    for attempt in range(3):
        if os.path.exists(LLAMA_SRC):
            print("[build] llama.cpp checkout is incomplete — removing and re-cloning.")
            shutil.rmtree(LLAMA_SRC, ignore_errors=True)
        print(f"[build] cloning llama.cpp (attempt {attempt + 1}/3)...")
        r = sh(["git", "clone", "--depth", "1",
                "https://github.com/ggml-org/llama.cpp", LLAMA_SRC],
               check=False, capture_output=True, text=True)
        if os.path.isfile(marker):
            return
        tail = ((r.stdout or "") + (r.stderr or "")).strip().splitlines()[-4:]
        for l in tail:
            print("        " + l[:160])
        time.sleep(2 * (attempt + 1))

    # Leave nothing behind: a half-cloned directory here is exactly what made
    # the *next* build skip cloning and fail on a missing CMakeLists.txt.
    shutil.rmtree(LLAMA_SRC, ignore_errors=True)
    raise RuntimeError(
        "Could not clone llama.cpp after 3 attempts. Last output: "
        + (" | ".join(tail) if tail else "(none)"))


def _run_capture(cmd):
    """Run and hand back (returncode, combined output). Never hide the reason a
    build died — build 9 raised a bare CalledProcessError whose message was just
    the argv, so the real cause was only visible by scrolling the cell."""
    r = subprocess.run(cmd, capture_output=True, text=True)
    return r.returncode, (r.stdout or "") + (r.stderr or "")


def _print_tail(what, rc, out, n=30):
    print(f"[build] {what} failed (exit {rc}):")
    for l in out.strip().splitlines()[-n:]:
        print("        " + l[:160])


def cmake_build(build_dir, flags):
    base = ["-DLLAMA_CURL=OFF", "-DCMAKE_BUILD_TYPE=Release"]
    rc, out = _run_capture(["cmake", "-S", LLAMA_SRC, "-B", build_dir] + flags + base)
    if rc != 0:
        # A CMakeCache.txt from an earlier, different configure (different source
        # path or compiler) poisons every later one, and the message for that is
        # not obviously about the cache. Wipe and try once more before believing
        # the failure is real.
        print("[build] configure failed — clearing the build directory and retrying.")
        shutil.rmtree(build_dir, ignore_errors=True)
        rc, out = _run_capture(["cmake", "-S", LLAMA_SRC, "-B", build_dir] + flags + base)
    if rc != 0:
        _print_tail("cmake configure", rc, out)
        raise RuntimeError(f"cmake configure failed: {_short_reason(out)}")

    rc, out = _run_capture(["cmake", "--build", build_dir, "--config", "Release",
                            "-j", str(os.cpu_count() or 4), "--target", "llama-server"])
    if rc != 0:
        _print_tail("cmake build", rc, out)
        raise RuntimeError(f"cmake build failed: {_short_reason(out)}")

    built = os.path.join(build_dir, "bin", "llama-server")
    if not os.path.exists(built):
        raise RuntimeError("Build finished but llama-server is missing — see the log.")
    if not _install_binary(os.path.dirname(built)):
        raise RuntimeError("Built llama-server will not run here.")
    return built


def _short_reason(out):
    """The one line worth putting in an exception message."""
    for l in out.strip().splitlines():
        s = l.strip()
        if s.startswith(("CMake Error", "error:", "fatal error:")) or "unsupported" in s:
            return s[:200]
    tail = out.strip().splitlines()
    return tail[-1][:200] if tail else "no output"


def _older_gcc():
    """A host compiler nvcc will accept, if the default one is too new.

    CUDA pins the maximum supported GCC; Debian 13 ships a newer one than some
    toolkits allow, and nvcc then stops with "unsupported GNU version"."""
    for v in ("14", "13", "12", "11"):
        exe = shutil.which(f"g++-{v}")
        if exe:
            return exe
        if apt_has(f"g++-{v}") and apt([f"g++-{v}"]):
            exe = shutil.which(f"g++-{v}")
            if exe:
                return exe
    return None


def build_cuda(nvcc):
    ver = nvcc_version(nvcc)
    if ver:
        print(f"[build] CUDA toolkit {ver[0]}.{ver[1]} at {nvcc}")
        if int(CUDA_ARCH) >= 120 and ver < (12, 8):
            print(f"[warn] sm_{CUDA_ARCH} needs CUDA 12.8 or newer and this is "
                  f"{ver[0]}.{ver[1]}. The build will probably fail.")
    clone_llama()
    toolkit_root = os.path.dirname(os.path.dirname(nvcc))
    flags = ["-DGGML_CUDA=ON",
             f"-DCMAKE_CUDA_ARCHITECTURES={CUDA_ARCH}",
             f"-DCMAKE_CUDA_COMPILER={nvcc}",
             f"-DCUDAToolkit_ROOT={toolkit_root}"]
    print(f"[build] compiling for sm_{CUDA_ARCH} (5-15 min, cached afterwards)...")
    try:
        cmake_build(os.path.join(LLAMA_SRC, "build-cuda"), flags)
    except RuntimeError as e:
        # The one CUDA failure worth a second attempt: a host compiler newer than
        # this toolkit accepts. Everything else is reported as-is.
        if "unsupported GNU version" not in str(e) and "unsupported" not in str(e).lower():
            raise
        gxx = _older_gcc()
        if not gxx:
            raise
        print(f"[build] host compiler rejected by nvcc — retrying with {gxx}")
        cmake_build(os.path.join(LLAMA_SRC, "build-cuda"),
                    flags + [f"-DCMAKE_CUDA_HOST_COMPILER={gxx}"])
    print(f"[build] ok: {SERVER_BIN}")
    return "cuda"


def have_spirv_headers():
    """ggml-vulkan includes <spirv/unified1/spirv.hpp>. glslc alone does not
    supply it — build 9 installed glslc, configured, and died on SPIRV-Headers."""
    for d in ("/usr/include/spirv/unified1", "/usr/local/include/spirv/unified1"):
        if glob.glob(os.path.join(d, "spirv.*")):
            return True
    return False


def vendor_spirv_headers():
    """Fallback when the distro has no spirv-headers package: take them from
    Khronos directly. Headers only, no build, a few hundred KB."""
    dst = "/usr/local/include/spirv"
    try:
        tmp = "/tmp/SPIRV-Headers"
        shutil.rmtree(tmp, ignore_errors=True)
        sh(["git", "clone", "--depth", "1",
            "https://github.com/KhronosGroup/SPIRV-Headers", tmp], check=True,
           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        src = os.path.join(tmp, "include", "spirv")
        if not os.path.isdir(src):
            return False
        shutil.rmtree(dst, ignore_errors=True)
        shutil.copytree(src, dst)
        print(f"[build] vendored SPIRV-Headers into {dst}")
        return True
    except Exception as e:
        print(f"[build] could not vendor SPIRV-Headers ({type(e).__name__}: {e})")
        return False


def build_vulkan():
    """Compile with Vulkan. The cheap path when the NVIDIA driver is present
    but the toolkit is not: a shader toolchain instead of 1 GB of nvcc.

    Blackwell exposes NV_coopmat2 through Vulkan, so this is a fast path on an
    RTX PRO 6000, not a consolation prize."""
    try:
        clone_llama()
    except Exception as e:
        print(f"[build] cannot compile ({e})")
        return None

    # One at a time: names vary across distros and one bad name must not take
    # the whole batch down with it.
    for p in ("libvulkan-dev", "vulkan-headers", "glslc", "libshaderc-dev",
              "glslang-tools", "glslang-dev", "libglslang-dev",
              "spirv-headers", "spirv-tools", "libspirv-tools-dev"):
        apt([p])

    if not shutil.which("glslc"):
        apt(["shaderc"])
    if not shutil.which("glslc"):
        print("[build] glslc is unavailable, so the Vulkan shaders cannot be "
              "compiled. Skipping the Vulkan source build.")
        return None
    if not have_spirv_headers() and not vendor_spirv_headers():
        print("[build] no SPIR-V headers available; skipping the Vulkan build.")
        return None

    print("[build] compiling llama.cpp with Vulkan (5-12 min, cached afterwards)...")
    try:
        cmake_build(os.path.join(LLAMA_SRC, "build-vulkan"),
                    ["-DGGML_VULKAN=ON",
                     "-DCMAKE_CXX_FLAGS=-I/usr/local/include"])
    except Exception as e:
        print(f"[build] Vulkan build failed ({type(e).__name__}) — see the lines "
              "above for the actual cmake error.")
        return None
    print(f"[build] ok: {SERVER_BIN}")
    return "vulkan"


# -------------------------------------------------------------------- Vulkan
def vulkan_device_kind():
    """'nvidia', 'software', 'other' or None, read from the ICD manifests
    rather than by running vulkaninfo, which is usually not installed. The
    absence of this check is what made an earlier build look healthy while it
    ran on the CPU."""
    icds = []
    for d in ("/usr/share/vulkan/icd.d", "/etc/vulkan/icd.d",
              "/usr/local/share/vulkan/icd.d"):
        icds += glob.glob(os.path.join(d, "*.json"))
    names = " ".join(os.path.basename(i).lower() for i in icds)
    if not icds:
        return None
    if "nvidia" in names:
        return "nvidia"
    if "lvp" in names or "lavapipe" in names or "swrast" in names:
        return "software"
    return "other"


def try_prebuilt(kind):
    """Fetch a published llama.cpp binary. kind is 'cuda' or 'vulkan'.

    Tries EVERY matching asset, and prints what was on offer when none match.
    The old version returned None from inside the loop, so one bad asset ended
    the search and the log never said why."""
    if platform.machine().lower() not in ("x86_64", "amd64"):
        return None
    url = "https://api.github.com/repos/ggml-org/llama.cpp/releases/latest"
    req = urllib.request.Request(url, headers={"User-Agent": "corx-heretic"})
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            rel = json.load(r)
    except Exception as e:
        print(f"[build] could not read release info ({type(e).__name__}: {e})")
        return None

    all_names = [a["name"] for a in rel.get("assets", [])]
    matches = []
    for a in rel.get("assets", []):
        n = a["name"].lower()
        if (("ubuntu" in n or "linux" in n) and kind in n
                and ("x64" in n or "amd64" in n)
                and (n.endswith(".tar.gz") or n.endswith(".zip"))):
            matches.append(a)

    if not matches:
        print(f"[build] no {kind} binary published for Linux x64. Assets in "
              f"{rel.get('tag_name', '?')}: {', '.join(all_names[:14])}")
        return None

    install_runtime_libs()
    for a in matches:
        print(f"[build] downloading prebuilt {kind} binary ({a['name']})...")
        try:
            dl = os.path.join("/tmp", a["name"])
            urllib.request.urlretrieve(a["browser_download_url"], dl)
            ex = f"/tmp/llama-{kind}-extract"
            shutil.rmtree(ex, ignore_errors=True)
            os.makedirs(ex, exist_ok=True)
            if dl.endswith(".zip"):
                with zipfile.ZipFile(dl) as zf:
                    zf.extractall(ex)
            else:
                with tarfile.open(dl, "r:gz") as tf:
                    tf.extractall(ex)
            found = glob.glob(os.path.join(ex, "**", "llama-server"), recursive=True)
            if not found:
                print("[build]   archive contains no llama-server.")
                continue
            if _install_binary(os.path.dirname(found[0])):
                print(f"[build] using prebuilt {kind} binary.")
                return kind
            print("[build]   that archive would not run here; trying the next.")
        except Exception as e:
            print(f"[build]   fetch failed ({type(e).__name__}: {e})")
    return None


# ---------------------------------------------------------------- the decision
def acquire_backend():
    if BACKEND in ("cuda", "vulkan"):
        print(f"[build] BACKEND is pinned to {BACKEND}.")
    want_cuda = BACKEND in ("auto", "cuda")
    want_vulkan = BACKEND in ("auto", "vulkan")
    failures = []

    def attempt(label, fn):
        """A rung that fails must not kill the run — the next one may work.
        Only an exhausted ladder is fatal, and then we report every reason."""
        try:
            return fn()
        except Exception as e:
            print(f"[build] {label} failed: {e}")
            failures.append(f"{label}: {e}")
            return None

    # 1. Toolkit already present: compile CUDA, the fastest thing to run.
    if want_cuda:
        nvcc = find_nvcc()
        if nvcc:
            got = attempt("CUDA build", lambda: build_cuda(nvcc))
            if got:
                return got
        else:
            print("[build] no nvcc here — the driver is installed, the CUDA "
                  "toolkit is not.")
        got = attempt("prebuilt CUDA", lambda: try_prebuilt("cuda"))
        if got:
            return got

    # 2. Vulkan, but only against the real driver.
    if want_vulkan:
        kind = vulkan_device_kind()
        if kind is None:
            print("[build] no Vulkan driver installed; installing libvulkan1 "
                  "(NOT mesa, which is the software one)...")
            apt(["libvulkan1"])
            kind = vulkan_device_kind()
        if kind == "nvidia":
            print("[build] real NVIDIA Vulkan driver found — a genuine GPU path, "
                  "somewhat slower than CUDA.")
            got = attempt("prebuilt Vulkan", lambda: try_prebuilt("vulkan"))
            if got:
                return got
            got = attempt("Vulkan build", build_vulkan)
            if got:
                return got
        elif kind == "software":
            print("[build] the only Vulkan driver here is a software one "
                  "(lavapipe). Refusing it: that runs the model on the CPU while "
                  "reporting layers offloaded to GPU.")
        elif kind is not None:
            print(f"[build] Vulkan drivers present but not NVIDIA ({kind}); skipping.")

    # 3. Nothing free left. Pay for a toolkit and compile.
    if want_cuda and INSTALL_CUDA_TOOLKIT and not find_nvcc():
        print("[cuda] no GPU backend yet — installing a CUDA toolkit.")
        nvcc = install_cuda_toolkit()
        if nvcc:
            got = attempt("CUDA build (fresh toolkit)", lambda: build_cuda(nvcc))
            if got:
                return got

    raise RuntimeError(
        "No GPU backend could be set up.\n"
        f"  - nvcc: {find_nvcc() or 'not found'}\n"
        f"  - Vulkan ICD: {vulkan_device_kind() or 'no driver installed'}\n"
        f"  - root: {IS_ROOT}\n"
        + ("What was tried:\n" + "".join(f"  - {f}\n" for f in failures) if failures else "")
        + "Fix one of these and re-run:\n"
        "  * install glslc + libvulkan-dev so the Vulkan build can compile, or\n"
        "  * use a runtime image that ships the CUDA toolkit (nvcc), or\n"
        "  * run as root so this cell can install one.\n"
        "Do not work around this by letting it run on the CPU — a 27B there is "
        "minutes per reply.")


if _works(SERVER_BIN):
    print(f"[build] llama-server already here: {SERVER_BIN}")
    BACKEND_USED = "existing"
else:
    BACKEND_USED = acquire_backend()


# ---------------- FIND AND FETCH THE RIGHT GGUF ----------------
_SPLIT_RE = re.compile(r"^(?P<stem>.+)-(?P<idx>\d{5})-of-(?P<tot>\d{5})\.gguf$", re.I)


def repo_files_with_sizes():
    """?blobs=true so each sibling carries a byte size — needed to check the
    disk before committing to a multi-gigabyte download."""
    url = f"https://huggingface.co/api/models/{MODEL_REPO}?blobs=true"
    req = urllib.request.Request(url, headers={"User-Agent": "corx-heretic"})
    if HF_TOKEN:
        req.add_header("Authorization", f"Bearer {HF_TOKEN}")
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            sibs = json.load(r).get("siblings", [])
    except urllib.error.HTTPError as e:
        if e.code in (401, 403):
            raise RuntimeError(
                f"{MODEL_REPO} is gated or private ({e.code}). Accept the terms on "
                "the model page and set HF_TOKEN at the top of this cell.") from e
        if e.code == 404:
            raise RuntimeError(
                f"{MODEL_REPO} does not exist on the Hub (404). Check the exact "
                "owner/name — it is case-sensitive and must be a repo that "
                "actually ships .gguf files.") from e
        raise
    return {s["rfilename"]: s.get("size") for s in sibs}


def gguf_variants(files):
    """One entry per variant, split shards grouped together."""
    groups = {}
    for f in files:
        if not f.lower().endswith(".gguf"):
            continue
        base, folder = os.path.basename(f), os.path.dirname(f)
        m = _SPLIT_RE.match(base)
        stem = m.group("stem") if m else base[:-5]
        groups.setdefault((folder, stem), []).append(f)
    return {k: sorted(v) for k, v in groups.items()}


def choose_variant(variants):
    def tag_of(stem):
        low = stem.lower()
        for t in BEST_TO_WORST:
            if t in low:
                return t
        return None

    scored = []
    for (folder, stem), files in variants.items():
        t = tag_of(stem)
        if t is not None:
            scored.append((BEST_TO_WORST.index(t), t, stem, files))
    if not scored:
        (folder, stem), files = sorted(variants.items())[0]
        print(f"[model] no quant tag in any filename; using {stem}")
        return stem, files
    scored.sort()

    if PRECISION == "full":
        for rank, t, stem, files in scored:
            if t in ("f32", "bf16", "f16", "fp16"):
                if NEEDS_GB[t] > VRAM_GB:
                    print(f"[warn] {t.upper()} wants ~{NEEDS_GB[t]} GB and this card "
                          f"has {VRAM_GB:.0f} GB. It will spill to system RAM and be "
                          "unusably slow. PRECISION = \"auto\" is the right setting "
                          "in a notebook.")
                print(f"[model] full precision → {stem} ({t.upper()})")
                return stem, files
        print(f"[warn] no F16/BF16 in this repo; taking the largest: {scored[0][2]}")
        return scored[0][2], scored[0][3]

    if PRECISION == "fast":
        # Speed is bandwidth / model-size, so the fastest usable variant is the
        # SMALLEST one whose quality still holds up — the opposite of what
        # "auto" does. Walk the preference order, not the size order.
        by_tag = {}
        for rank, t, stem, files in scored:
            by_tag.setdefault(t, (stem, files))
        for t in FAST_ORDER:
            if t not in by_tag:
                continue
            stem, files = by_tag[t]
            if NEEDS_GB[t] > VRAM_GB:
                continue
            gb = variant_gb(files)
            if gb and gb + MIN_FREE_GB > FREE_GB:
                print(f"[model]   skipping {stem}: {gb:.1f} GB on disk vs "
                      f"{FREE_GB:.1f} GB free")
                continue
            if gb:
                ceil = GPU_BANDWIDTH_GBS / gb
                print(f"[model] fast → {stem} ({t.upper()}, {gb:.1f} GB) — expect "
                      f"roughly {ceil * 0.7:.0f} tok/s "
                      f"({ceil:.0f} is the bandwidth ceiling for this size).")
                big = by_tag.get("f16") or by_tag.get("bf16")
                if big and t not in ("f16", "bf16", "f32", "fp16"):
                    bg = variant_gb(big[1])
                    if bg and bg > gb:
                        print(f"[model]   (F16 would be {bg:.1f} GB and about "
                              f"{bg / gb:.1f}x slower for no visible quality gain. "
                              f"PRECISION = \"auto\" if you want it anyway.)")
            else:
                print(f"[model] fast → {stem} ({t.upper()})")
            return stem, files
        print("[warn] nothing in the fast list fits; falling back to the largest "
              "that does.")

    if PRECISION in ("auto", "fast"):
        # Two constraints, not one. With 96 GB of VRAM the card is no longer the
        # binding limit — a notebook's disk is. Checking only VRAM would pick
        # F16, then fail at the download.
        for rank, t, stem, files in scored:
            if NEEDS_GB[t] > VRAM_GB:
                continue
            gb = variant_gb(files)
            if gb and gb + MIN_FREE_GB > FREE_GB:
                print(f"[model]   skipping {stem}: {gb:.1f} GB on disk vs "
                      f"{FREE_GB:.1f} GB free")
                continue
            print(f"[model] auto → {stem} ({t.upper()}, ~{NEEDS_GB[t]} GB VRAM of "
                  f"{VRAM_GB:.0f} GB, {gb:.1f} GB on disk)")
            return stem, files
        raise RuntimeError(
            f"No variant in {MODEL_REPO} fits both {VRAM_GB:.0f} GB VRAM and "
            f"{FREE_GB:.1f} GB free disk. Point HF_HOME at a larger volume, or "
            "name a smaller quant directly, e.g. PRECISION = \"q4_k_m\".")

    want = PRECISION.lower()
    for rank, t, stem, files in scored:
        if want in stem.lower():
            print(f"[model] forced → {stem}")
            return stem, files
    raise RuntimeError(f'PRECISION = "{PRECISION}" matches nothing in {MODEL_REPO}. '
                       f"Available: {sorted({s[2] for s in scored})}")


print(f"[model] reading {MODEL_REPO} ...")
SIZES = repo_files_with_sizes()
variants = gguf_variants(SIZES.keys())
if not variants:
    raise RuntimeError(f"{MODEL_REPO} contains no .gguf files. This script serves "
                       "GGUF via llama.cpp; a safetensors repo will not work here.")

HF_HOME = os.environ.get("HF_HOME") or os.path.expanduser("~/.cache/huggingface")
os.makedirs(HF_HOME, exist_ok=True)
FREE_GB = shutil.disk_usage(HF_HOME).free / 1e9


def variant_gb(files):
    return sum(SIZES.get(f) or 0 for f in files) / 1e9


print(f"[model] variants: {', '.join(sorted(s for _, s in variants))}")
print(f"[model] {FREE_GB:.1f} GB free on {HF_HOME}")
stem, wanted_files = choose_variant(variants)

# Belt and braces: "full" and forced quants skip the check above.
need_gb = variant_gb(wanted_files)
if need_gb and need_gb + MIN_FREE_GB > FREE_GB:
    raise RuntimeError(
        f"Not enough disk: {stem} needs {need_gb:.1f} GB and only {FREE_GB:.1f} GB "
        "is free. Set PRECISION = \"auto\" to pick a variant that fits, or point "
        "HF_HOME at a larger volume before this cell runs.")

from huggingface_hub import hf_hub_download

if len(wanted_files) > 1:
    print(f"[model] {len(wanted_files)} shards to fetch — this is the slow part.")
local_paths = [hf_hub_download(repo_id=MODEL_REPO, filename=f,
                               token=HF_TOKEN or None) for f in wanted_files]
MODEL_PATH = local_paths[0]           # llama.cpp finds the rest beside it
print(f"[model] {sum(os.path.getsize(p) for p in local_paths) / 1e9:.1f} GB → "
      f"{MODEL_PATH}")


# ════════════════════════════════════════════════════════════════
#  HAND OFF TO A DETACHED RUNNER
#
#  Build 12 kept llama-server, the API shim and cloudflared alive
#  from daemon threads inside the notebook kernel. Daemon threads
#  die with the process that owns them, so the moment molab
#  recycled or idled out the kernel, the whole server went with it
#  — "it just stopped running".
#
#  Nothing that lives inside the kernel can survive the kernel. So
#  everything that has to keep running is written out as its own
#  script and launched with start_new_session=True: a separate
#  session, its own process group, stdin closed, output to a log.
#  It survives the cell finishing, the kernel restarting, and the
#  notebook being closed. It restarts llama-server and cloudflared
#  by itself and republishes the tunnel URL when it changes.
#
#  Re-running this cell finds a healthy runner and just reprints
#  the URL instead of starting a second one.
# ════════════════════════════════════════════════════════════════
RUN_DIR = os.path.abspath("./corx-run")
os.makedirs(RUN_DIR, exist_ok=True)
CFG_PATH = os.path.join(RUN_DIR, "config.json")
RUNNER_PATH = os.path.join(RUN_DIR, "runner.py")
URL_PATH = os.path.join(RUN_DIR, "url.txt")
STATUS_PATH = os.path.join(RUN_DIR, "status.json")
LOG_PATH = os.path.join(RUN_DIR, "runner.log")
PID_PATH = os.path.join(RUN_DIR, "runner.pid")

RUNNER_SRC = r'''
# CorX Heretic runner — owns llama-server, the OpenAI shim and the tunnel.
# Launched detached by the notebook cell; supervises everything for ever.
import json, os, re, signal, subprocess, sys, threading, time, urllib.request, asyncio

CFG = json.load(open(sys.argv[1]))
SERVER_BIN   = CFG["server_bin"]
MODEL_PATH   = CFG["model_path"]
BIN_DIR      = CFG["bin_dir"]
MODEL_ID     = CFG["model_id"]
PORT         = CFG["port"]
LLAMA_PORT   = CFG["llama_port"]
CTX          = CFG["ctx"]
MAX_NEW      = CFG["max_new_tokens_cap"]
NO_AUTH      = CFG["no_auth"]
API_KEY      = CFG["api_key"]
REQUIRE_GPU  = CFG["require_gpu"]
LOAD_TIMEOUT = CFG["load_timeout"]
BACKEND_USED = CFG["backend"]
PRECISION_ST = CFG["precision"]
URL_PATH     = CFG["url_file"]
STATUS_PATH  = CFG["status_file"]

STATE = {"url": None, "tunnel": None, "llama": None, "tok_s": 0.0,
         "ready": False, "tunnel_fails": 0}
DEVICE_INFO = {"device": None, "offloaded": None, "layers": None,
               "cpu_only": None, "software": False}
_LOG_TAIL = []
_inflight = [0]
_lock = threading.Lock()


def say(m):
    print(m, flush=True)


def write_status(**kw):
    d = {"url": STATE["url"], "ready": STATE["ready"], "tokens_per_sec": STATE["tok_s"],
         "device": DEVICE_INFO["device"], "gpu_layers": DEVICE_INFO["offloaded"],
         "backend": BACKEND_USED, "precision": PRECISION_ST}
    d.update(kw)
    try:
        tmp = STATUS_PATH + ".tmp"
        with open(tmp, "w") as f:
            json.dump(d, f)
        os.replace(tmp, STATUS_PATH)
    except Exception:
        pass


# ---------------- llama-server ----------------
_RE_OFFLOAD = re.compile(r"offloaded\s+(\d+)\s*/\s*(\d+)\s+layers?\s+to\s+gpu")
_RE_USING = re.compile(r"using device (\S+)\s*\((.*?)\)", re.I)
_RE_VK = re.compile(r"ggml_vulkan:\s*\d+\s*=\s*(.+?)\s*\|")
_RE_CUDA = re.compile(r"Device\s+\d+:\s*(.+?),\s*compute capability")


def note_line(line):
    low = line.lower()
    m = _RE_OFFLOAD.search(low)
    if m:
        DEVICE_INFO["offloaded"] = int(m.group(1))
        DEVICE_INFO["layers"] = int(m.group(2))
    if not DEVICE_INFO["software"]:
        for rx, fmt in ((_RE_USING, "{0} ({1})"), (_RE_VK, "Vulkan: {0}"),
                        (_RE_CUDA, "CUDA: {0}")):
            mm = rx.search(line)
            if mm and not DEVICE_INFO["device"]:
                DEVICE_INFO["device"] = fmt.format(*mm.groups())
                break
    if "llvmpipe" in low or "lavapipe" in low or "swiftshader" in low:
        DEVICE_INFO["device"] = "SOFTWARE RASTERISER (CPU)"
        DEVICE_INFO["software"] = True
    DEVICE_INFO["cpu_only"] = bool(
        DEVICE_INFO["software"]
        or (DEVICE_INFO["offloaded"] is not None and DEVICE_INFO["offloaded"] == 0))


# Performance flags. This card is an RTX PRO 6000 Blackwell class part; the
# defaults leave a lot on the table. Flash attention and a bigger micro-batch
# are the two that matter most for prefill. Not every llama.cpp build accepts
# every spelling, so if the server rejects them we retry with the plain set
# rather than refusing to run.
# Tiers, not one all-or-nothing list. An older llama.cpp that rejects a single
# flag used to drop us all the way to stock settings, losing the batch tuning
# too. Try the most aggressive set first and step down one tier at a time.
#
#   -fa              flash attention: large prefill win, and required for the
#                    quantised KV cache below.
#   -ctk/-ctv q8_0   quantise the KV cache. Halves the memory it reads per
#                    token, which is the other bandwidth cost after the weights,
#                    and matters more the longer the conversation gets.
#   -b / -ub         bigger batch and micro-batch: this is what actually feeds
#                    a card this wide during prefill. The default 512 leaves
#                    most of it idle.
#   --split-mode none / --main-gpu 0
#                    one GPU: skip the multi-GPU scheduling path entirely.
FLAG_TIERS = [
    ["-fa", "-ctk", "q8_0", "-ctv", "q8_0", "-b", "4096", "-ub", "2048",
     "--split-mode", "none", "--main-gpu", "0",
     "-t", str(os.cpu_count() or 8), "--no-warmup"],
    ["-fa", "-b", "4096", "-ub", "2048",
     "-t", str(os.cpu_count() or 8), "--no-warmup"],
    ["-b", "2048", "-ub", "1024", "-t", str(os.cpu_count() or 8)],
    [],
]


def llama_cmd(extra):
    return ([SERVER_BIN, "-m", MODEL_PATH,
             "--host", "127.0.0.1", "--port", str(LLAMA_PORT),
             "-ngl", "999", "-c", str(CTX), "--jinja", "-a", MODEL_ID] + extra)


def spawn_llama(extra):
    cmd = llama_cmd(extra)
    say("[llama] " + " ".join(cmd))
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                         text=True, bufsize=1,
                         env=dict(os.environ, LD_LIBRARY_PATH=BIN_DIR))

    def watch():
        for line in p.stdout:
            _LOG_TAIL.append(line)
            del _LOG_TAIL[:-400]
            note_line(line)
            sys.stdout.write("[llama] " + line)
            sys.stdout.flush()
    threading.Thread(target=watch, daemon=True).start()
    return p


def llama_healthy():
    try:
        with urllib.request.urlopen("http://127.0.0.1:%d/health" % LLAMA_PORT, timeout=5) as r:
            return r.status == 200
    except Exception:
        return False


def start_llama():
    # Step down the tiers until one starts; never give up tuning wholesale
    # because a single flag was not recognised.
    for extra in FLAG_TIERS:
        p = spawn_llama(extra)
        start = time.time()
        while time.time() - start < LOAD_TIMEOUT:
            if p.poll() is not None:
                bad = "".join(_LOG_TAIL[-6:]).lower()
                if extra and ("unrecognized" in bad or "invalid" in bad
                              or "usage" in bad or "unknown argument" in bad):
                    say("[llama] this build rejected one of those flags — "
                        "stepping down a tier.")
                    break
                say("[llama] exited with %s. Last lines:" % p.returncode)
                for l in _LOG_TAIL[-12:]:
                    say("        " + l.rstrip())
                break
            if llama_healthy():
                say("[llama] ready — flags: %s" % (" ".join(extra) if extra else "stock"))
                if DEVICE_INFO["device"] and "cuda" not in str(DEVICE_INFO["device"]).lower():
                    say("[warn] NOT running on CUDA (%s). Vulkan works but is "
                        "materially slower than CUDA on an NVIDIA card; set "
                        "BACKEND = \"cuda\" and re-run if you want the fast path."
                        % DEVICE_INFO["device"])
                return p
            time.sleep(2)
        try:
            p.kill()
        except Exception:
            pass
    return None


def warmup():
    body = {"model": MODEL_ID, "max_tokens": 24, "stream": False,
            "messages": [{"role": "user", "content": "Say READY and nothing else."}]}
    req = urllib.request.Request(
        "http://127.0.0.1:%d/v1/chat/completions" % LLAMA_PORT,
        data=json.dumps(body).encode(),
        headers={"content-type": "application/json"})
    t0 = time.time()
    with urllib.request.urlopen(req, timeout=600) as r:
        out = json.load(r)
    dt = time.time() - t0
    n = (out.get("usage") or {}).get("completion_tokens") or 24
    return (n / dt) if dt > 0 else 0.0


# ---------------- OpenAI shim ----------------
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import httpx, uvicorn

LLAMA = "http://127.0.0.1:%d" % LLAMA_PORT
app = FastAPI(title="CorX3.8 Heretic API")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True,
                   allow_methods=["*"], allow_headers=["*"])
_client = httpx.AsyncClient(timeout=httpx.Timeout(None, connect=30.0))


def _bump(n):
    with _lock:
        _inflight[0] += n


def check_auth(request):
    if NO_AUTH:
        return
    if request.headers.get("authorization", "") != "Bearer " + API_KEY:
        raise HTTPException(status_code=401, detail="bad api key")


@app.get("/health")
async def health():
    return {"status": "ok", "model": MODEL_ID,
            "backend": "llama.cpp/" + str(BACKEND_USED),
            "precision": PRECISION_ST,
            "device": DEVICE_INFO["device"] or BACKEND_USED,
            "gpu_layers": DEVICE_INFO["offloaded"],
            "tokens_per_sec": round(STATE["tok_s"], 1),
            "ctx": CTX, "busy": _inflight[0] > 0}


@app.get("/v1/models")
async def list_models(request: Request):
    check_auth(request)
    return {"object": "list",
            "data": [{"id": MODEL_ID, "object": "model", "owned_by": "corx-labs"}]}


def _err_chunk(text):
    return ("data: " + json.dumps({
        "object": "chat.completion.chunk", "model": MODEL_ID,
        "choices": [{"index": 0, "finish_reason": "error",
                     "delta": {"content": text}}]}) + "\n\n").encode()


@app.post("/v1/chat/completions")
async def chat_completions(request: Request):
    check_auth(request)
    body = await request.json()
    if not body.get("messages"):
        raise HTTPException(status_code=400, detail="messages required")
    body["model"] = MODEL_ID
    if body.get("max_tokens"):
        body["max_tokens"] = min(int(body["max_tokens"]), MAX_NEW)
    stream = bool(body.get("stream", False))

    _bump(1)
    if not stream:
        try:
            r = await _client.post(LLAMA + "/v1/chat/completions", json=body)
            return JSONResponse(status_code=r.status_code, content=r.json())
        except Exception as e:
            return JSONResponse(status_code=500, content={"error": {
                "message": "%s: %s" % (type(e).__name__, e), "type": "generation_error"}})
        finally:
            _bump(-1)

    # A Cloudflare quick tunnel buffers a body unless told plainly not to, which
    # is what made replies arrive in one lump minutes later instead of
    # streaming. Needs all three: the headers below, bytes immediately, and
    # bytes every few seconds while the model prefills. ":" lines are SSE
    # comments the browser ignores.
    async def relay():
        q = asyncio.Queue()
        got_content = asyncio.Event()

        async def pump():
            try:
                async with _client.stream(
                        "POST", LLAMA + "/v1/chat/completions", json=body,
                        headers={"Accept": "text/event-stream",
                                 "Accept-Encoding": "identity"}) as r:
                    if r.status_code != 200:
                        detail = (await r.aread()).decode("utf-8", "replace")[:400]
                        await q.put(_err_chunk("\n[llama.cpp returned %s: %s]"
                                               % (r.status_code, detail)))
                        await q.put(b"data: [DONE]\n\n")
                        return
                    async for chunk in r.aiter_bytes():
                        got_content.set()
                        await q.put(chunk)
            except Exception as e:
                tail = "".join(_LOG_TAIL[-6:]).strip().replace("\n", " ")[:300]
                await q.put(_err_chunk(
                    "\n[Generation failed on the server: %s: %s. Last server log: %s]"
                    % (type(e).__name__, e, tail)))
                await q.put(b"data: [DONE]\n\n")
            finally:
                await q.put(None)

        task = asyncio.create_task(pump())
        yield b": open\n\n"
        try:
            while True:
                try:
                    item = await asyncio.wait_for(q.get(), timeout=5)
                except asyncio.TimeoutError:
                    if not got_content.is_set():
                        yield b": ping\n\n"
                    continue
                if item is None:
                    break
                yield item
        finally:
            task.cancel()
            _bump(-1)

    return StreamingResponse(relay(), media_type="text/event-stream", headers={
        "Cache-Control": "no-cache, no-store, no-transform",
        "X-Accel-Buffering": "no",
        "Connection": "keep-alive"})


def serve_shim():
    uvicorn.run(app, host="0.0.0.0", port=PORT, log_level="warning")


# ---------------- cloudflared ----------------
def cloudflared_path():
    p = os.path.abspath("./cloudflared")
    if os.path.exists(p):
        return p
    import platform as _pf
    arch = _pf.machine().lower()
    suffix = "arm64" if ("aarch64" in arch or "arm64" in arch) else "amd64"
    url = ("https://github.com/cloudflare/cloudflared/releases/latest/"
           "download/cloudflared-linux-" + suffix)
    urllib.request.urlretrieve(url, p)
    os.chmod(p, 0o755)
    return p


def start_tunnel(timeout=60):
    b = cloudflared_path()
    p = subprocess.Popen([b, "tunnel", "--no-autoupdate", "--url",
                          "http://localhost:%d" % PORT],
                         stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                         text=True, bufsize=1)
    rx = re.compile(r"https://[-a-z0-9]+\.trycloudflare\.com")
    url, start = None, time.time()
    while time.time() - start < timeout:
        line = p.stdout.readline()
        if not line:
            if p.poll() is not None:
                break
            continue
        m = rx.search(line)
        if m:
            url = m.group(0)
            break
    threading.Thread(target=lambda: [None for _ in p.stdout], daemon=True).start()
    return url, p


def publish(url):
    STATE["url"] = url
    try:
        tmp = URL_PATH + ".tmp"
        with open(tmp, "w") as f:
            f.write(url or "")
        os.replace(tmp, URL_PATH)
    except Exception:
        pass
    write_status()
    say("[tunnel] " + str(url))


# ---------------- bring everything up, then keep it up ----------------
write_status(stage="starting llama-server")
STATE["llama"] = start_llama()
if STATE["llama"] is None:
    write_status(stage="failed", error="llama-server would not start")
    say("[fatal] llama-server would not start; see the log above.")
    sys.exit(1)

if DEVICE_INFO["cpu_only"] and REQUIRE_GPU:
    write_status(stage="failed", error="model landed on the CPU")
    say("[fatal] the model is on the CPU, not the GPU. Refusing to serve.")
    try:
        STATE["llama"].kill()
    except Exception:
        pass
    sys.exit(1)

write_status(stage="warming up")
try:
    STATE["tok_s"] = warmup()
    say("[llama] warm-up: about %.1f tokens/sec" % STATE["tok_s"])
except Exception as e:
    say("[warn] warm-up failed (%s: %s)" % (type(e).__name__, e))

threading.Thread(target=serve_shim, daemon=True).start()
time.sleep(2)

u, t = start_tunnel()
STATE["tunnel"] = t
if not u:
    STATE["tunnel_fails"] = 1
STATE["ready"] = True
publish(u)
write_status(stage="live" if u else "live locally — no tunnel yet")
say("[runner] live. This process is detached and will keep running.")

while True:
    time.sleep(10)
    try:
        if STATE["llama"] is None or STATE["llama"].poll() is not None:
            say("[llama] died — restarting.")
            write_status(stage="restarting llama-server", ready=False)
            STATE["ready"] = False
            STATE["llama"] = start_llama()
            if STATE["llama"] is not None:
                try:
                    STATE["tok_s"] = warmup()
                except Exception:
                    pass
                STATE["ready"] = True
                write_status(stage="live")
        if STATE["tunnel"] is not None and STATE["tunnel"].poll() is not None:
            # Back off rather than hammering: if cloudflared cannot start at all
            # (no egress, say) a tight retry loop just floods the log. The API
            # is still served locally in the meantime.
            wait = min(60, 5 * (2 ** STATE["tunnel_fails"]))
            say("[tunnel] cloudflared exited — retrying in %ds." % wait)
            time.sleep(wait)
            u, t = start_tunnel()
            STATE["tunnel"] = t
            if u:
                STATE["tunnel_fails"] = 0
            else:
                STATE["tunnel_fails"] = min(4, STATE["tunnel_fails"] + 1)
            publish(u)
    except Exception as e:
        say("[runner] %s: %s" % (type(e).__name__, e))
        time.sleep(5)
'''


def runner_pid():
    try:
        pid = int(open(PID_PATH).read().strip())
    except Exception:
        return None
    try:
        os.kill(pid, 0)
        return pid
    except OSError:
        return None


def stop_runner():
    pid = runner_pid()
    if pid:
        try:
            os.killpg(os.getpgid(pid), signal.SIGTERM)
        except Exception:
            try:
                os.kill(pid, signal.SIGTERM)
            except Exception:
                pass
        time.sleep(2)
    for p in (PORT, LLAMA_PORT):
        free_port(p)


def read_status():
    try:
        return json.load(open(STATUS_PATH))
    except Exception:
        return {}


# Already up and healthy? Then this is a re-run: report and leave it alone.
_existing = runner_pid()
if _existing and not RESTART_RUNNER:
    st = read_status()
    if st.get("ready") and st.get("url"):
        print("\n" + "=" * 66)
        print(" ALREADY RUNNING (pid %d) — nothing to do." % _existing)
        print("=" * 66)
        print(" Public URL : %s/v1" % st["url"])
        print(" Model      : %s   (%s)" % (MODEL_ID, st.get("precision")))
        print(" Device     : %s" % st.get("device"))
        print(" Speed      : ~%.1f tokens/sec" % (st.get("tokens_per_sec") or 0))
        print("=" * 66)
        print("Set RESTART_RUNNER = True and re-run to restart it.")
        raise SystemExit
print("[runner] starting a fresh detached runner...")
stop_runner()

with open(RUNNER_PATH, "w") as f:
    f.write(RUNNER_SRC)
with open(CFG_PATH, "w") as f:
    json.dump({
        "server_bin": SERVER_BIN, "model_path": MODEL_PATH, "bin_dir": BIN_DIR,
        "model_id": MODEL_ID, "port": PORT, "llama_port": LLAMA_PORT,
        "ctx": CTX, "max_new_tokens_cap": MAX_NEW_TOKENS_CAP,
        "no_auth": NO_AUTH, "api_key": API_KEY, "require_gpu": REQUIRE_GPU,
        "load_timeout": LOAD_TIMEOUT, "backend": BACKEND_USED,
        "precision": stem, "url_file": URL_PATH, "status_file": STATUS_PATH,
    }, f)
for stale in (URL_PATH, STATUS_PATH):
    try:
        os.remove(stale)
    except OSError:
        pass

_log = open(LOG_PATH, "ab", buffering=0)
_proc = subprocess.Popen(
    [sys.executable, "-u", RUNNER_PATH, CFG_PATH],
    stdout=_log, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL,
    start_new_session=True,          # its own session: survives the kernel
    cwd=os.getcwd())
with open(PID_PATH, "w") as f:
    f.write(str(_proc.pid))
print("[runner] pid %d, log %s" % (_proc.pid, LOG_PATH))

# Wait for it to publish a URL, showing what it is doing meanwhile.
_last = None
_deadline = time.time() + LOAD_TIMEOUT + 300
while time.time() < _deadline:
    if _proc.poll() is not None:
        print("[runner] exited early (%s). Last of its log:" % _proc.returncode)
        try:
            print("".join(open(LOG_PATH, errors="replace").read().splitlines(True)[-25:]))
        except Exception:
            pass
        raise SystemExit
    st = read_status()
    stage = st.get("stage")
    if stage and stage != _last:
        print("[runner] %s" % stage)
        _last = stage
    if st.get("error"):
        print("[runner] FAILED: %s" % st["error"])
        try:
            print("".join(open(LOG_PATH, errors="replace").read().splitlines(True)[-25:]))
        except Exception:
            pass
        raise SystemExit
    if st.get("ready") and st.get("url"):
        PUBLIC_URL = st["url"]
        print("\n" + "=" * 66)
        print(" CorX3.8-27B HERETIC API IS LIVE")
        print("=" * 66)
        print(" Public URL : %s/v1" % PUBLIC_URL)
        print(" Health     : %s/health" % PUBLIC_URL)
        print(" Auth       : %s" % ("NONE (open)" if NO_AUTH else "Bearer " + API_KEY))
        print(" Model      : %s   (%s)" % (MODEL_ID, stem))
        print(" Backend    : llama.cpp / %s" % BACKEND_USED)
        print(" Device     : %s" % (st.get("device") or BACKEND_USED))
        print(" Speed      : ~%.1f tokens/sec, %d ctx" % (st.get("tokens_per_sec") or 0, CTX))
        print("=" * 66)
        print("Paste the Public URL (without /v1) into the CorX chat Settings,")
        print("and set the model to  %s" % MODEL_ID)
        print("")
        print("The runner is DETACHED (pid %d). It keeps serving after this cell" % _proc.pid)
        print("finishes, after the kernel restarts, and restarts llama-server or")
        print("the tunnel by itself. Follow it with:  tail -f %s" % LOG_PATH)
        print("The current URL is always in:  %s" % URL_PATH)
        break
    time.sleep(3)
else:
    print("[runner] still not live. Check the log: %s" % LOG_PATH)

Why this one is built differently. That repo ships GGUF, not safetensors, so the backend here is llama.cpp rather than transformers. The cell compiles llama-server with CUDA for your exact card the first time (5–15 minutes, then cached), asks the Hub which files the repo actually contains, downloads the right one, and puts a thin FastAPI layer in front so the site sees the identical /health and /v1/chat/completions contract as the normal server. Nothing in the chat needs to change.

Full precision means the F16 file. A GGUF repo ships one file per quantisation, so PRECISION = "full" pulls the F16/BF16 weights — roughly 54 GB, nothing quantised, every layer on the GPU. That wants about 64 GB of VRAM once the 32k KV cache is in there. On a smaller card use PRECISION = "auto" and it takes the largest file that fits, or name a quant directly with "q8_0" or "q4_k_m". Large files come as shards; the script fetches all of them and hands llama.cpp the first.

When it prints HERETIC API IS LIVE, paste the Public URL into settings the same way, then set the model field to corx3.8-heretic. Sandbox, search, files, GitHub and personalities all work exactly as they do on the normal build.

If the chat says it “did not respond within 25s”. That is almost never the address. The usual cause is the model running on the CPU while everything still looks fine: a Vulkan build on a notebook box picks up lavapipe, a software rasteriser, so -ngl 999 reports “63/63 layers offloaded to GPU” and then generates at a few seconds per token. A 27B is unusable there, and the front end can only report the silence.

This cell will not let that happen quietly. It reads the startup log for the device it actually got and refuses to hand you a URL if the model landed on the CPU. Before printing the address it runs a warm-up generation and tells you the measured tokens/sec — so you know it is fast enough without finding out in the chat. Set REQUIRE_GPU = False if you want it to serve anyway.

If it said “CUDA Toolkit not found”, that is the opposite problem and just as common: a notebook container ships the NVIDIA driver but no toolkit, so nvidia-smi works, nvcc does not exist, and there is nothing to compile with. No pip wheel fixes it either — nvidia-cuda-nvcc-cu12 contains ptxas and nvvm only, not nvcc.

So getting a backend is a ladder, and the cell walks it for you: compile with CUDA if nvcc is already there; else use a prebuilt CUDA binary if one is published; else use the real NVIDIA Vulkan driver if it is installed, which is a genuine GPU path needing no toolkit at all; else install about 1 GB of CUDA from NVIDIA’s own apt repo and compile. A software Vulkan device is refused at every rung. If none of it is possible it stops with a list of what to change — it will never quietly serve from the CPU. Pin a rung with BACKEND = "cuda" or "vulkan", and set INSTALL_CUDA_TOOLKIT = False to be told rather than fixed.

The chat helps from its side too: when your own server misses the window it asks /health — which answers instantly even while generation crawls — and tells you which problem you have, a dead tunnel URL or a model on the wrong device, quoting the device and tok/s it reported.

Optional · Voice studio

Clone a voice with TriStream-SVS.

A second server, for the microphone button in the chat header. Give it a voice to clone and a performance to sing, and it returns the performance in that voice. Runs alongside the chat model on the same machine, or on its own.

How the two clips are used. TriStream keeps pitch, timbre and texture in three encoder streams that cannot exchange information until the fusion trunk, so a two-clip transplant is the natural operation rather than a trick. The voice clip goes through the speaker encoder into the filter stream — the only stream that carries identity. The performance clip contributes its F0 contour to the source stream, which has no pathway to timbre at all, and its texture to the residual stream. That is why the melody transfers without the original singer’s voice coming with it.

The tunnel URL is public and keyless by default, exactly like the chat server. Anyone holding it can run voice cloning on your GPU. Clone voices you have the right to use — your own, or ones you have permission for.

Press Copy, paste, run
The Copy button on the block below puts the whole script on your clipboard. Paste it into a notebook cell on a GPU machine, or into a file and run python3 tristream-server.py. It installs everything, downloads the weights and opens a Cloudflare tunnel. On Blackwell cards (RTX PRO 6000, RTX 50-series) it installs the CUDA 12.8 build of PyTorch, which is the first one carrying sm_120 kernels.
Check /health first
It reports which entry point inside the model repo it managed to bind to, and whether that gives stream-level control or the repo’s own conversion call. If ok is false it says exactly what it found and what it tried.
Open the chat and press the microphone
Paste the URL, press Connect, choose your two clips, press Generate. The result plays inline and downloads as a WAV.

Press Copy for the whole script — all 1879 lines, not just what is on screen. If you would rather not paste it by hand, the same file is at curl -O https://corx-labs.com/chat/tristream-server.py.

# =============================================================================
# TriStream-SVS server — build 14
#
# The build number is printed at startup and reported by /health. If a log does
# not say "build 14", the copy running is an older one — re-copy from
# https://corx-labs.com/chat/documentation/#tristream
#
# Serves Sigmandndnns/TriStream-SVS-300M behind an OpenAI-shaped HTTP API and a
# Cloudflare quick tunnel, so the voice studio in the CorX chat can reach it.
#
# Runs either way, unchanged: paste it into one molab / Colab cell on a GPU
# runtime, or save it as tristream-server.py on your own workstation and run
# `python3 tristream-server.py`. Nothing here needs a notebook. It prints a
# https://<random>.trycloudflare.com URL; paste that into the chat's Voice
# panel and press Connect.
#
# On Blackwell cards (RTX PRO 6000, RTX 50-series) it installs the CUDA 12.8
# build of PyTorch — the first one that ships sm_120 kernels — and /health
# reports the GPU it found and whether the build actually covers it.
#
# WHAT IT DOES WITH THE TWO CLIPS
# -------------------------------
# TriStream's whole design is that the three encoder streams are structurally
# separate — the source stream physically cannot carry timbre, and the filter
# stream is the only one that carries identity. That makes a two-clip transplant
# the natural operation rather than a hack:
#
#   VOICE clip       -> speaker encoder -> singer embedding -> FILTER stream
#                       (who it sounds like)
#   PERFORMANCE clip -> F0 contour + voiced/unvoiced flag   -> SOURCE stream
#                       (the notes, the timing, the vibrato)
#                    -> mel                                  -> RESIDUAL stream
#                       (breath and consonant texture; switchable to the voice)
#
# So: it sings the performance, in the target voice.
#
# HONEST NOTE ON THE MODEL INTERFACE
# ----------------------------------
# This script does NOT assume an inference API. It downloads the repo, looks at
# what is actually in it, and binds to whatever entry point it finds — trying
# stream-level control first so the transplant above is real, and falling back
# to the repo's own conversion call if that is all it exposes. /health reports
# exactly which path it bound to, and if it can bind to nothing it says what it
# found and what it tried instead of pretending to work. Check /health before
# you trust the output.
# =============================================================================

import os
import subprocess
import sys
import time

# Bumped whenever this file changes. Printed at startup and reported by /health,
# so a log or a screenshot says which version is actually running — guessing
# that from behaviour wastes a round every time.
BUILD        = 14

REPO_ID      = "Sigmandndnns/TriStream-SVS-300M"

# The repo publishes weights without code, so the class that defines the network
# has to come from somewhere else — the training script it was trained with is
# the obvious source, since the checkpoint records who trained it. Point this at
# a .py file (a local path or an https URL) that defines the model class, or
# drop the file at <workdir>/model_def.py and leave this empty. Whatever class
# it defines is constructed from config.json and checked tensor by tensor
# against the checkpoint before it is used.
MODEL_DEF    = "https://corx-labs.com/chat/tristream-model.py"

PORT         = 811
SAMPLE_RATE  = 24000          # TriStream decodes mel at 24 kHz
STEPS        = 32             # rectified-flow sampling steps; the card says ~32
NO_AUTH      = True           # matches the other CorX server scripts

# Keep the cell running after launch, printing a heartbeat and — the part that
# matters — the new tunnel address whenever cloudflared rotates it. Set to False
# to have the cell finish immediately as it used to.
KEEP_ALIVE   = True
WORKDIR      = os.path.abspath("./tristream-run")

os.makedirs(WORKDIR, exist_ok=True)
RUNNER_PATH = os.path.join(WORKDIR, "runner.py")
CFG_PATH    = os.path.join(WORKDIR, "config.json")
LOG_PATH    = os.path.join(WORKDIR, "server.log")
URL_PATH    = os.path.join(WORKDIR, "tunnel-url.txt")
PID_PATH    = os.path.join(WORKDIR, "server.pid")
SUPERVISOR_PATH = os.path.join(WORKDIR, "supervisor.py")
STATUS_PATH = os.path.join(WORKDIR, "supervisor.json")


def sh(cmd, **kw):
    print("$", cmd, flush=True)
    return subprocess.run(cmd, shell=True, **kw)


# -----------------------------------------------------------------------------
# 1. Dependencies
# -----------------------------------------------------------------------------
print("=" * 72)
print("TriStream-SVS server — build %d — installing dependencies" % BUILD)
print("=" * 72)

sh("pip -q install --upgrade pip")
# Blackwell cards (RTX PRO 6000, RTX 50-series) are compute capability sm_120,
# and the cu124 wheels carry no sm_120 kernels — torch imports fine, then every
# CUDA call fails with "no kernel image is available". cu128 is the first build
# that ships them, so try it first and only fall back for older cards.
sh("pip -q install torch torchaudio --index-url https://download.pytorch.org/whl/cu128 "
   "|| pip -q install torch torchaudio --index-url https://download.pytorch.org/whl/cu124 "
   "|| pip -q install torch torchaudio")
sh("pip -q install 'transformers>=4.44' huggingface_hub safetensors einops "
   "fastapi 'uvicorn[standard]' python-multipart soundfile librosa numpy scipy")

# cloudflared, for the quick tunnel
if subprocess.run("which cloudflared", shell=True, capture_output=True).returncode != 0:
    sh("wget -q -O /usr/local/bin/cloudflared "
       "https://github.com/cloudflare/cloudflared/releases/latest/download/cloudflared-linux-amd64 "
       "&& chmod +x /usr/local/bin/cloudflared")

import json    # noqa: E402
import signal  # noqa: E402
import socket  # noqa: E402


# -----------------------------------------------------------------------------
# 1b. Clear the port
#
# The server is launched detached on purpose, so it survives the notebook kernel
# dying. The cost of that is that re-running this cell collides with the copy
# still holding the port: uvicorn fails with [Errno 98] and exits, leaving a
# fresh tunnel pointing at nothing. So the old instance is stopped first.
# -----------------------------------------------------------------------------
def port_free(port):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    try:
        s.bind(("0.0.0.0", port))
        return True
    except OSError:
        return False
    finally:
        s.close()


def pids_on_port(port):
    """Who is listening on this port, using /proc alone.

    lsof and fuser are both absent from most notebook images, so relying on
    them means the port never actually gets cleared and the failure looks
    identical to not having tried. /proc/net/tcp gives the socket inode of
    every listener; /proc/<pid>/fd maps inodes back to processes.
    """
    inodes = set()
    for proto in ("tcp", "tcp6"):
        try:
            rows = open("/proc/net/" + proto).read().splitlines()[1:]
        except OSError:
            continue
        for row in rows:
            f = row.split()
            if len(f) < 10 or f[3] != "0A":        # 0A is TCP_LISTEN
                continue
            try:
                if int(f[1].rsplit(":", 1)[1], 16) == port:
                    inodes.add(f[9])
            except (IndexError, ValueError):
                continue
    if not inodes:
        return []

    found = []
    for entry in os.listdir("/proc"):
        if not entry.isdigit():
            continue
        fd_dir = "/proc/%s/fd" % entry
        try:
            names = os.listdir(fd_dir)
        except OSError:
            continue                                # gone, or not ours to read
        for fd in names:
            try:
                target = os.readlink(os.path.join(fd_dir, fd))
            except OSError:
                continue
            if target.startswith("socket:[") and target[8:-1] in inodes:
                found.append(int(entry))
                break
    return found


def is_our_runner(pid):
    """Confirm the pid is still our server before signalling it.

    Pids get recycled. A stale pid file plus an unlucky wrap-around would mean
    killing somebody else's process, so check what is actually running there.
    """
    try:
        with open("/proc/%d/cmdline" % pid, "rb") as fh:
            cmd = fh.read()
            return b"runner.py" in cmd or b"supervisor.py" in cmd
    except Exception:
        return False      # no /proc, or it is already gone — do not signal


def stop_previous():
    """Stop the instance from a previous run, by pid file then by port."""
    if os.path.exists(PID_PATH):
        try:
            old = int(open(PID_PATH).read().strip())
        except Exception:
            old = 0
        try:
            if old > 0 and old != os.getpid() and is_our_runner(old):
                for sig in (signal.SIGTERM, signal.SIGKILL):
                    try:
                        # start_new_session made it a process group leader, so
                        # this takes its cloudflared child with it.
                        os.killpg(os.getpgid(old), sig)
                        print("stopped the previous server (pid %d)" % old)
                    except ProcessLookupError:
                        break
                    except Exception:
                        try:
                            os.kill(old, sig)
                        except Exception:
                            break
                    freed = False
                    for _ in range(20):
                        if port_free(PORT):
                            freed = True
                            break
                        time.sleep(0.25)
                    if freed:
                        return
        finally:
            # Always, including on the early return above: a pid file left
            # behind points at a process that no longer exists, and eventually
            # at an unrelated one that does.
            try:
                os.remove(PID_PATH)
            except OSError:
                pass

    if port_free(PORT):
        return

    # No usable pid file — servers started by an earlier version of this script
    # never wrote one. Find the holder through /proc instead, and only kill it
    # if it is one of ours.
    holders = pids_on_port(PORT)
    for pid in holders:
        if pid == os.getpid() or not is_our_runner(pid):
            print("port %d is held by pid %d, which is not this server — "
                  "leaving it alone" % (PORT, pid))
            continue
        for sig in (signal.SIGTERM, signal.SIGKILL):
            try:
                os.killpg(os.getpgid(pid), sig)
            except Exception:
                try:
                    os.kill(pid, sig)
                except Exception:
                    break
            print("stopped an older server holding port %d (pid %d)" % (PORT, pid))
            for _ in range(20):
                if port_free(PORT):
                    return
                time.sleep(0.25)

    if holders:
        # /proc answered, and what it found is not ours. fuser -k would kill it
        # regardless of whose it is, so stop here and let the caller move to a
        # different port instead of taking down somebody else's service.
        return

    # Only when /proc told us nothing — a restricted container — is a blind
    # kill the lesser evil, and even then only because nothing else can.
    subprocess.run("fuser -k %d/tcp" % PORT, shell=True, capture_output=True)
    for _ in range(20):
        if port_free(PORT):
            return
        time.sleep(0.25)


stop_previous()
if not port_free(PORT):
    # Still occupied by something that is not ours. The tunnel URL is printed
    # either way, so the port number does not need to be memorable.
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(("0.0.0.0", 0))
    PORT = s.getsockname()[1]
    s.close()
    print("port %d was busy and would not clear — using %d instead" % (811, PORT))

json.dump({
    "build": BUILD, "model_def": MODEL_DEF,
    "repo_id": REPO_ID, "port": PORT, "sample_rate": SAMPLE_RATE,
    "steps": STEPS, "no_auth": NO_AUTH, "workdir": WORKDIR,
    "url_path": URL_PATH,
}, open(CFG_PATH, "w"))


# -----------------------------------------------------------------------------
# 2. The runner
#
# Written to disk and launched in its own session. A thread inside the notebook
# dies with the kernel; a detached process does not, which is the difference
# between "it stopped running overnight" and a server that stays up.
# -----------------------------------------------------------------------------
RUNNER_SRC = r'''
import gc, io, json, os, re, subprocess, sys, threading, time, traceback

CFG = json.load(open(sys.argv[1]))
BUILD       = CFG.get("build", 0)
MODEL_DEF   = CFG.get("model_def", "")
REPO_ID     = CFG["repo_id"]
PORT        = CFG["port"]
SAMPLE_RATE = CFG["sample_rate"]
STEPS       = CFG["steps"]
NO_AUTH     = CFG["no_auth"]
WORKDIR     = CFG["workdir"]
URL_PATH    = CFG["url_path"]

import numpy as np
import soundfile as sf
import torch

DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
DTYPE  = torch.float16 if DEVICE == "cuda" else torch.float32

GPU_NAME = ""
if DEVICE == "cpu":
    # Worth saying loudly. A 322M-parameter flow-matching decoder on CPU is not
    # slow, it is unusable — and "no GPU here" is easy to miss when the runtime
    # simply never had one attached.
    print("!! No CUDA device visible — running on CPU. torch %s. This model will "
          "not synthesise at a usable speed this way; the runtime needs a GPU "
          "attached." % torch.__version__, flush=True)
if DEVICE == "cuda":
    GPU_NAME = torch.cuda.get_device_name(0)
    cap = torch.cuda.get_device_capability(0)
    arch_ok = ("sm_%d%d" % cap) in (torch.cuda.get_arch_list() or [])
    print("GPU: %s (sm_%d%d), torch %s built for %s"
          % (GPU_NAME, cap[0], cap[1], torch.__version__,
             ",".join(torch.cuda.get_arch_list() or ["?"])), flush=True)
    if not arch_ok:
        print("!! This torch build has no kernels for sm_%d%d. Reinstall with:\n"
              "   pip install --force-reinstall torch torchaudio "
              "--index-url https://download.pytorch.org/whl/cu128" % cap, flush=True)
    # Blackwell has bf16 throughout and it is better behaved than fp16 here.
    if cap[0] >= 9:
        DTYPE = torch.bfloat16

STATE = {
    "ready": False,
    "loading": True,      # the API serves while this is true
    "binding": None,      # which entry point we bound to
    "detail": "",         # why, or why not
    "files": [],
    "tried": [],
    "ckpt_keys": [],      # first weight names, when a checkpoint was opened
    "arch": None,         # the parameter tree, when weights load but code is absent
    "model_def_version": None,
    "model": None,
    "device": DEVICE,
    "tunnel": None,
    "tunnel_since": None,
    "tunnel_restarts": 0,
    "started": time.time(),
}

def log(*a):
    print(*a, flush=True)


def supervisor_restarts():
    """How many times the supervisor has had to bring this server back."""
    try:
        with open(os.path.join(WORKDIR, "supervisor.json")) as f:
            return int(json.load(f).get("restarts", 0))
    except Exception:
        return 0


# ---------------------------------------------------------------------------
# Audio helpers — the two inputs the panel sends
# ---------------------------------------------------------------------------
# A whole song is a realistic upload and a bad idea: pitch tracking alone runs
# at roughly 3x realtime, so four minutes of audio is over a minute of CPU work
# before the model is even called. Clips are trimmed to this and the caller is
# told, rather than being left to wonder why nothing came back.
MAX_SECONDS = 60.0


def sniff_format(raw):
    """Name the container from its magic bytes.

    libsndfile reports "Format not recognised" next to a BytesIO object address,
    which tells a person nothing about the file they just picked.
    """
    if raw[:4] == b"RIFF" and raw[8:12] == b"WAVE": return "WAV"
    if raw[:4] == b"fLaC": return "FLAC"
    if raw[:4] == b"OggS": return "OGG"
    if raw[:3] == b"ID3":  return "MP3"
    if len(raw) > 1 and raw[0] == 0xFF and (raw[1] & 0xE0) == 0xE0: return "MP3"
    if raw[4:8] == b"ftyp": return "MP4/M4A/AAC"
    if raw[:4] == b"\x1aE\xdf\xa3": return "WebM/Matroska"
    if raw[:4] == b"FORM": return "AIFF"
    if raw[:4] == b"caff": return "CAF"
    return "unknown"


def decode_audio(raw):
    """Bytes -> (float32 samples, sample rate), trying more than one decoder.

    libsndfile handles WAV, FLAC, OGG and MP3, but not the MPEG-4 family — and
    .m4a is what an iPhone voice memo and most Android recorders produce, which
    makes it the single most likely file for someone cloning their own voice.
    librosa/audioread covers it when a system codec exists.
    """
    errors = []
    try:
        wav, sr = sf.read(io.BytesIO(raw), dtype="float32", always_2d=True)
        return wav.mean(axis=1), sr
    except Exception as e:
        errors.append("soundfile: %s" % str(e)[:80])

    try:
        import librosa
        wav, sr = librosa.load(io.BytesIO(raw), sr=None, mono=True)
        return np.asarray(wav, dtype="float32"), sr
    except Exception as e:
        errors.append("librosa/audioread: %s" % str(e)[:80])

    fmt = sniff_format(raw)
    # The decoder errors go to the log, not to the person: libsndfile's message
    # is "Format not recognised" next to a BytesIO object address, which reads
    # like a crash and says nothing about the file they picked.
    log("decode failed (%s): %s" % (fmt, "; ".join(errors)))
    raise ValueError("This build cannot decode %s. Re-export the clip as WAV, FLAC, "
                     "OGG or MP3 — all four work. On a phone, a voice memo is usually "
                     ".m4a, which is the case that lands here most often." % fmt)


def load_audio(raw, target_sr=None, max_seconds=MAX_SECONDS, label="clip"):
    """Bytes -> mono float32 at target_sr, trimmed and sanity-checked."""
    import librosa
    target_sr = target_sr or SAMPLE_RATE
    if not raw:
        raise ValueError("The %s was empty — no bytes arrived." % label)

    wav, sr = decode_audio(raw)
    if sr != target_sr:
        wav = librosa.resample(wav, orig_sr=sr, target_sr=target_sr)

    trimmed = 0.0
    if max_seconds and len(wav) > int(max_seconds * target_sr):
        trimmed = len(wav) / float(target_sr)
        wav = wav[:int(max_seconds * target_sr)]

    peak = float(np.max(np.abs(wav))) if wav.size else 0.0
    if wav.size < target_sr // 10:
        raise ValueError("The %s is only %.2fs long. Give it at least a second of "
                         "audio to work with." % (label, wav.size / float(target_sr)))
    if peak < 1e-4:
        raise ValueError("The %s is silent (peak %.2g). Check the right file was "
                         "picked and that it is not a muted export." % (label, peak))
    wav = wav / peak * 0.97
    return wav.astype("float32"), trimmed


def f0_contour(wav, sr=None, fmin=65.0, fmax=1200.0):
    """Log-F0 and the voiced/unvoiced flag — exactly what the source stream takes.

    The source stream receives these two arrays and nothing else. It has no path
    to the mel, the singer embedding, or any other timbre-carrying signal, which
    is why taking them from a different singer transfers the melody without
    dragging that singer's voice along with it.
    """
    import librosa
    sr = sr or SAMPLE_RATE
    f0, voiced, _ = librosa.pyin(wav, fmin=fmin, fmax=fmax, sr=sr,
                                 frame_length=1024, hop_length=256)
    f0 = np.nan_to_num(f0, nan=0.0)
    log_f0 = np.zeros_like(f0)
    nz = f0 > 0
    log_f0[nz] = np.log(f0[nz])
    return log_f0.astype("float32"), voiced.astype("float32")


def mel_spec(wav, sr=None, n_mels=100):
    """100-band mel at 24 kHz — the decoder's output format, per the model card."""
    import librosa
    sr = sr or SAMPLE_RATE
    m = librosa.feature.melspectrogram(y=wav, sr=sr, n_fft=1024, hop_length=256,
                                       n_mels=n_mels)
    return np.log(np.clip(m, 1e-5, None)).astype("float32")


def shift_pitch(log_f0, semitones):
    if not semitones:
        return log_f0
    out = log_f0.copy()
    nz = out > 0
    out[nz] = out[nz] + float(semitones) * np.log(2.0) / 12.0
    return out


def to_wav_bytes(wav, sr=None):
    buf = io.BytesIO()
    sf.write(buf, np.asarray(wav, dtype="float32"), sr or SAMPLE_RATE, format="WAV",
             subtype="PCM_16")
    return buf.getvalue()


# ---------------------------------------------------------------------------
# Binding to whatever the repo actually exposes
#
# The model card is not readable from the machine that wrote this script, so
# nothing here assumes a function signature. It downloads the repo, walks the
# whole tree and binds to the most capable entry point present, preferring one
# that lets the three streams be driven separately, because that is what makes
# the voice and the performance come from different clips rather than a generic
# any-to-any conversion.
#
# This repo turned out to be weights-first: a config.json that is NOT a
# transformers config (no model_type) and plain PyTorch checkpoints. So the
# checkpoint is itself a load path, not merely a file to hand to something
# else — and the search has to include subdirectories, because that is where
# the repo keeps everything that is not a weight file.
# ---------------------------------------------------------------------------
STREAM_CALLS  = ["convert_streams", "transplant", "convert_voice", "voice_convert"]
CONVERT_CALLS = ["convert", "inference", "infer", "synthesize", "generate", "__call__"]

CKPT_EXT = (".pt", ".pth", ".ckpt", ".bin", ".safetensors")
# Checkpoints named "best" beat "last": same training run, better validation.
CKPT_RANK = ("best", "final", "last", "ema", "g_", "model")
CLASS_HINTS = ("tristream", "svs", "singer", "sing", "synth", "voice",
               "generator", "model", "net")


def walk_repo(root):
    """Every file in the snapshot, relative to it.

    listdir() was the original bug: it sees only the top level, so a repo that
    keeps its code in a subdirectory looks like a repo with no code at all.
    """
    out = []
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in (".git", ".cache", "__pycache__")]
        for f in filenames:
            out.append(os.path.relpath(os.path.join(dirpath, f), root))
    return sorted(out)


def add_import_roots(root, files):
    """Put every directory that holds Python on sys.path.

    This matters twice over. It lets a module inside a subdirectory be imported
    at all, and it lets pickle resolve the module path recorded inside a saved
    nn.Module — unpickling fails with ModuleNotFoundError unless the defining
    module is importable under the exact name it had when it was saved.
    """
    roots = {root}
    for rel in files:
        if rel.endswith(".py"):
            d = os.path.dirname(os.path.join(root, rel))
            roots.add(d)
            # A package's parent is what makes "pkg.module" resolve.
            if os.path.exists(os.path.join(d, "__init__.py")):
                roots.add(os.path.dirname(d))
    for d in sorted(roots, key=len):
        if d and d not in sys.path:
            sys.path.insert(0, d)
    return sorted(roots)


# Files that are programs rather than model definitions. Importing one runs it:
# a training script parses argv and calls sys.exit(), a setup.py builds a
# package, a test file may assert its way out. None of them define the model.
SKIP_PY = ("setup.py", "__main__.py", "conftest.py", "train.py", "training.py",
           "preprocess.py", "app.py", "demo.py", "gradio_app.py", "webui.py")
SKIP_DIRS = ("tests", "test", "scripts", "examples", "notebooks", "data")


def worth_importing(rel):
    base = os.path.basename(rel)
    if base in SKIP_PY or base.startswith("test_"):
        return False
    parts = os.path.dirname(rel).split(os.sep)
    return not any(p in SKIP_DIRS for p in parts)


def import_file(path, root):
    """Import one .py by location, without guessing a package name.

    Running the repo's Python is the same trust you extend by running its
    weights; there is no way to load a model like this without it. What is not
    acceptable is letting it end the process — argv is blanked so a stray
    argparse cannot exit, and SystemExit is turned back into a plain error.
    """
    import importlib.util
    name = os.path.splitext(os.path.relpath(path, root))[0].replace(os.sep, "_")
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise ImportError("no loader for %s" % path)
    mod = importlib.util.module_from_spec(spec)
    saved_argv = sys.argv
    sys.argv = [os.path.basename(path)]
    sys.modules[name] = mod
    try:
        spec.loader.exec_module(mod)
    except SystemExit as e:
        sys.modules.pop(name, None)
        raise ImportError("the file called sys.exit(%s) on import — it is a "
                          "script, not a model definition" % e.code)
    except BaseException:
        sys.modules.pop(name, None)
        raise
    finally:
        sys.argv = saved_argv
    return mod


def load_checkpoint(path):
    """torch.load with the trust decision made explicitly.

    torch 2.6 flipped weights_only to True, which refuses any checkpoint holding
    more than tensors. These files predate that and may hold a pickled module,
    so the flag has to come off deliberately rather than by accident.
    """
    try:
        return torch.load(path, map_location="cpu", weights_only=False)
    except TypeError:
        return torch.load(path, map_location="cpu")


def find_state_dict(blob):
    """Pull the weights out of whatever shape the checkpoint was saved in."""
    if hasattr(blob, "state_dict") and callable(getattr(blob, "state_dict")):
        return None, "(already a module)"
    if not isinstance(blob, dict):
        return None, None

    def looks_like_weights(d):
        vals = [v for v in list(d.values())[:12]]
        return bool(vals) and all(isinstance(v, torch.Tensor) for v in vals)

    for key in ("state_dict", "model_state_dict", "model", "generator", "net",
                "module", "weights", "params", "g"):
        v = blob.get(key)
        if isinstance(v, dict) and looks_like_weights(v):
            return v, key
        if hasattr(v, "state_dict"):
            return v, key
    if looks_like_weights(blob):
        return blob, "(top level)"
    return None, None


def strip_prefix(sd):
    """DataParallel and Lightning both leave a prefix on every key."""
    for pre in ("module.", "model.", "_orig_mod."):
        if sd and all(k.startswith(pre) for k in sd):
            return {k[len(pre):]: v for k, v in sd.items()}
    return sd


def candidate_classes(mods):
    """nn.Module subclasses defined in the repo's own modules, best guess first."""
    found = []
    for mod in mods:
        for attr in dir(mod):
            obj = getattr(mod, attr, None)
            if not isinstance(obj, type) or not issubclass(obj, torch.nn.Module):
                continue
            if obj.__module__ != mod.__name__:      # imported from torch, not defined here
                continue
            if attr.startswith("_"):
                continue                      # a helper, by convention
            score = sum(2 for h in CLASS_HINTS if h in attr.lower())
            found.append((score, attr, obj))
    found.sort(key=lambda t: -t[0])
    return found


def construct(cls, cfg):
    """Try the constructor shapes a config-driven model usually takes."""
    for args, kwargs in ((( ), cfg if isinstance(cfg, dict) else {}),
                         ((cfg,), {}),
                         (( ), {})):
        try:
            return cls(*args, **kwargs), ("%s(**config)" % cls.__name__ if kwargs
                                          else "%s(config)" % cls.__name__ if args
                                          else "%s()" % cls.__name__)
        except Exception:
            continue
    return None, None


def load_model_def():
    """Import a model definition supplied from outside the repo.

    This repo publishes parameters without a network, so the class has to come
    from wherever the model was trained. Accepts a local path or an https URL,
    and falls back to <workdir>/model_def.py so a file can simply be dropped in.
    Whatever it defines is still checked against the checkpoint tensor by tensor
    before anything uses it — supplying a definition is not a promise that it is
    the right one.
    """
    src = MODEL_DEF or os.path.join(WORKDIR, "model_def.py")
    if src.startswith(("http://", "https://")):
        dest = os.path.join(WORKDIR, "model_def.py")
        try:
            import urllib.request
            urllib.request.urlretrieve(src, dest)
            log("fetched the model definition from", src)
            src = dest
        except Exception as e:
            STATE["tried"].append("model definition %s: %s" % (src, str(e)[:140]))
            return []
    if not os.path.exists(src):
        return []
    try:
        mod = import_file(src, os.path.dirname(src) or ".")
    except KeyboardInterrupt:
        raise
    except BaseException as e:
        STATE["tried"].append("model definition %s: %s: %s"
                              % (src, type(e).__name__, str(e)[:160]))
        return []
    names = [n for _, n, _ in candidate_classes([mod])]
    ver = getattr(mod, "MODEL_DEF_VERSION", None)
    STATE["model_def_version"] = ver
    log("model definition %s (version %s) defines: %s"
        % (src, ver if ver is not None else "unstamped", ", ".join(names) or "none"))
    STATE["tried"].append("model definition %s imported (version %s, classes: %s)"
                          % (src, ver if ver is not None else "unstamped",
                             ", ".join(names) or "none"))
    return [mod]


def describe_arch(sd, cfg, blob, rel):
    """Write down everything needed to rebuild the missing model class.

    A state_dict is a complete description of a network's parameter tree: every
    submodule path and every tensor shape. Together with the config and the
    checkpoint's own metadata that is enough to write a matching nn.Module —
    which is the only way forward for a repo that ships weights without code.
    Guessing the architecture instead would load some tensors, skip others, and
    synthesise noise while reporting success.
    """
    groups = {}
    for k, v in sd.items():
        top = k.split(".")[0]
        g = groups.setdefault(top, {"count": 0, "params": 0, "sample": []})
        g["count"] += 1
        try:
            g["params"] += int(v.numel())
        except Exception:
            pass
        if len(g["sample"]) < 6:
            g["sample"].append("%s %s" % (k, tuple(getattr(v, "shape", ()))))

    meta = {}
    if isinstance(blob, dict):
        for k, v in blob.items():
            if k == "model" or isinstance(v, dict) and len(v) > 50:
                continue
            if isinstance(v, (str, int, float, bool)) or v is None:
                meta[k] = v
            elif isinstance(v, dict):
                meta[k] = {kk: vv for kk, vv in list(v.items())[:40]
                           if isinstance(vv, (str, int, float, bool)) or vv is None}

    STATE["arch"] = {
        "checkpoint": rel,
        "tensors": len(sd),
        "parameters": sum(g["params"] for g in groups.values()),
        "config": cfg,
        "checkpoint_meta": meta,
        "top_level_modules": {k: {"tensors": g["count"], "parameters": g["params"],
                                  "sample": g["sample"]}
                              for k, g in sorted(groups.items())},
        "all_keys": ["%s %s" % (k, tuple(getattr(v, "shape", ()))) for k, v in sd.items()],
    }

    path = os.path.join(WORKDIR, "architecture.json")
    try:
        with open(path, "w") as fh:
            json.dump(STATE["arch"], fh, indent=2, default=str)
    except Exception as e:
        log("could not write %s: %s" % (path, e))

    log("")
    log("=" * 72)
    log("  This repo has weights but no model code.")
    log("  %d tensors, %s parameters, %d top-level modules:"
        % (len(sd), format(STATE["arch"]["parameters"], ","), len(groups)))
    for name, g in sorted(groups.items(), key=lambda kv: -kv[1]["params"]):
        log("     %-24s %4d tensors  %s params" % (name, g["count"], format(g["params"], ",")))
    log("")
    log("  The full parameter tree is in %s" % path)
    log("  and at <tunnel-url>/architecture. That is what a matching nn.Module")
    log("  has to be written against — the shapes name every layer.")
    log("=" * 72)
    log("")


def bind_model():
    from huggingface_hub import snapshot_download
    local_dir = os.path.join(WORKDIR, "model")

    # Fetch one checkpoint, not every checkpoint. This repo ships best and last,
    # 322M parameters each; pulling both doubles the download, doubles the disk,
    # and buys nothing, since "last" is the same run at an earlier validation
    # score. If the listing cannot be read, fall back to taking everything.
    allow = None
    try:
        from huggingface_hub import list_repo_files
        names = list_repo_files(REPO_ID)
        weights = [f for f in names if f.lower().endswith(CKPT_EXT)]
        if len(weights) > 1:
            weights.sort(key=lambda f: next((i for i, k in enumerate(CKPT_RANK)
                                             if k in os.path.basename(f).lower()),
                                            len(CKPT_RANK)))
            skip = set(weights[1:])
            allow = [f for f in names if f not in skip]
            log("fetching %s and skipping %s" % (weights[0], ", ".join(sorted(skip))))
    except Exception as e:
        log("could not list the repo first (%s); fetching everything" % str(e)[:80])

    local = snapshot_download(repo_id=REPO_ID, local_dir=local_dir, allow_patterns=allow)

    files = walk_repo(local)
    STATE["files"] = files
    log("repo files (%d):" % len(files))
    for f in files[:60]:
        log("   ", f)

    roots = add_import_roots(local, files)
    log("import roots:", roots)

    cfg = {}
    cfg_path = os.path.join(local, "config.json")
    if os.path.exists(cfg_path):
        try:
            with open(cfg_path) as fh:
                cfg = json.load(fh)
            log("config keys:", sorted(cfg)[:30])
        except Exception as e:
            STATE["tried"].append("config.json: %s" % str(e)[:120])

    obj = None

    # (a) torch.hub, if the repo ships a hubconf anywhere in the tree
    for rel in files:
        if os.path.basename(rel) != "hubconf.py":
            continue
        hub_dir = os.path.dirname(os.path.join(local, rel)) or local
        try:
            hub = import_file(os.path.join(local, rel), local)
            entries = [a for a in dir(hub)
                       if not a.startswith("_") and callable(getattr(hub, a))]
            log("hubconf entry points:", entries)
            for entry in entries:
                try:
                    obj = torch.hub.load(hub_dir, entry, source="local", pretrained=True)
                    STATE["detail"] = "loaded via torch.hub %s()" % entry
                    break
                except Exception as e:
                    STATE["tried"].append("hub.%s: %s" % (entry, str(e)[:120]))
        except Exception as e:
            STATE["tried"].append("hubconf: %s" % str(e)[:140])
        if obj is not None:
            break

    # (b) transformers — only when the config is actually a transformers config.
    #     Trying it without a model_type key just buries the real error under a
    #     guaranteed failure, which is exactly what happened the first time.
    if obj is None and cfg.get("model_type"):
        for loader in ("AutoModel", "AutoModelForSpeechSeq2Seq"):
            try:
                import transformers
                cls = getattr(transformers, loader)
                obj = cls.from_pretrained(local, trust_remote_code=True,
                                          torch_dtype=DTYPE).to(DEVICE).eval()
                STATE["detail"] = "loaded via transformers.%s(trust_remote_code=True)" % loader
                break
            except Exception as e:
                STATE["tried"].append("%s: %s" % (loader, str(e)[:160]))
    elif obj is None and os.path.exists(cfg_path):
        STATE["tried"].append("transformers: skipped, config.json has no model_type "
                              "(it is the model's own config, not a transformers one)")

    # (c) a module that builds the model — supplied by the user first, since a
    #     definition someone went and fetched beats anything guessed from the
    #     repo, then anything the repo itself ships.
    repo_mods = load_model_def()
    if obj is None:
        for rel in files:
            if not rel.endswith(".py") or os.path.basename(rel) == "hubconf.py":
                continue
            if not worth_importing(rel):
                STATE["tried"].append("skipped %s (a script, not a model definition)" % rel)
                continue
            try:
                repo_mods.append(import_file(os.path.join(local, rel), local))
            except KeyboardInterrupt:
                raise
            except BaseException as e:
                STATE["tried"].append("import %s: %s: %s"
                                      % (rel, type(e).__name__, str(e)[:120]))
        log("imported %d repo module(s)" % len(repo_mods))

        for mod in repo_mods:
            for factory in ("load_model", "load_pretrained", "from_pretrained",
                            "build_model", "build", "get_model", "load"):
                fn = getattr(mod, factory, None)
                if fn is None:
                    continue
                # A bare class constructor is not a loader. Calling TriStreamSVS()
                # succeeds and hands back a randomly initialised network, which
                # would then report ready and synthesise noise — far worse than
                # failing. Classes are only tried with the repo path, which at
                # least implies they read something; weights that come in through
                # a checkpoint are checked tensor by tensor further down.
                arg_sets = ((local,),) if isinstance(fn, type) else ((local,), ())
                for args in arg_sets:
                    try:
                        made = fn(*args)
                    except Exception as e:
                        STATE["tried"].append("%s.%s: %s"
                                              % (mod.__name__, factory, str(e)[:120]))
                        continue
                    if made is not None:
                        obj = made
                        STATE["detail"] = "loaded via %s.%s()" % (mod.__name__, factory)
                        break
                if obj is not None:
                    break
            if obj is not None:
                break

    # (d) the checkpoint itself. For a weights-first repo this is the real path,
    #     and the original script never opened these files at all.
    ckpts = [f for f in files if f.lower().endswith(CKPT_EXT)]
    ckpts.sort(key=lambda f: next((i for i, k in enumerate(CKPT_RANK)
                                   if k in os.path.basename(f).lower()), len(CKPT_RANK)))
    if obj is None and ckpts:
        blob = None
        for rel in ckpts:
            path = os.path.join(local, rel)
            # Release the previous one first. Python evaluates the right-hand
            # side before rebinding, so a plain reassignment holds two 322M
            # parameter checkpoints in memory at once — enough to get the
            # process OOM-killed on a modest runtime, which looks from outside
            # exactly like the server stopping for no reason.
            blob = None
            gc.collect()
            try:
                blob = load_checkpoint(path)
            except ModuleNotFoundError as e:
                # The single most useful diagnostic there is: the pickle names
                # the exact module the repo expects to be importable.
                STATE["tried"].append(
                    "%s: needs the module '%s', which is not in this repo — the "
                    "architecture code lives somewhere else" % (rel, e.name))
                continue
            except Exception as e:
                STATE["tried"].append("%s: %s" % (rel, str(e)[:160]))
                continue

            if isinstance(blob, torch.nn.Module):
                obj = blob
                STATE["detail"] = "loaded the pickled module out of %s" % rel
                break

            sd, where = find_state_dict(blob)
            if isinstance(blob, dict):
                log("%s top-level keys: %s" % (rel, sorted(blob)[:20]))
                # Checkpoints often carry the config the model was built with.
                for ck in ("config", "hyper_parameters", "hparams", "args", "cfg"):
                    if isinstance(blob.get(ck), dict) and not cfg:
                        cfg = blob[ck]
                        log("using config from the checkpoint's '%s' key" % ck)
            if sd is None:
                STATE["tried"].append("%s: no state_dict inside (keys: %s)"
                                      % (rel, sorted(blob)[:12] if isinstance(blob, dict)
                                         else type(blob).__name__))
                continue
            if hasattr(sd, "state_dict"):
                obj = sd
                STATE["detail"] = "loaded the pickled module from %s['%s']" % (rel, where)
                break

            sd = strip_prefix(sd)
            STATE["ckpt_keys"] = list(sd)[:12]
            log("%s: %d weight tensors under '%s'" % (rel, len(sd), where))

            # Record the tree first. Whether or not something below manages to
            # load it, this is the description needed to fix a mismatch.
            if STATE["arch"] is None:
                describe_arch(sd, cfg, blob, rel)

            # A definition that builds itself from the checkpoint cannot
            # mismatch it: every layer is created at the shape the weights
            # actually have. Preferred over constructing a class blind.
            builder = None
            for mod in repo_mods:
                fn = getattr(mod, "build_from_checkpoint", None)
                if callable(fn):
                    builder = (mod, fn)
                    break
            if builder is not None:
                mod, fn = builder
                try:
                    made = fn(sd, cfg)
                except Exception as e:
                    traceback.print_exc()
                    STATE["tried"].append("%s.build_from_checkpoint: %s: %s"
                                          % (mod.__name__, type(e).__name__, str(e)[:1200]))
                    made = None
                if made is not None:
                    missing, unexpected = made.load_state_dict(sd, strict=False)
                    matched = len(sd) - len(unexpected)
                    if matched == len(sd) and not missing:
                        obj = made
                        STATE["detail"] = ("built from the checkpoint by %s: all %d "
                                           "tensors loaded" % (mod.__name__, len(sd)))
                        break
                    STATE["tried"].append(
                        "%s.build_from_checkpoint: %d/%d tensors matched, %d missing "
                        "(first missing: %s) (first unexpected: %s)"
                        % (mod.__name__, matched, len(sd), len(missing),
                           ", ".join(list(missing)[:5]) or "none",
                           ", ".join(list(unexpected)[:5]) or "none"))

            classes = candidate_classes(repo_mods)
            if obj is None and not classes:
                # The weights are here and readable; the class to put them in is
                # not. That is recoverable — a state_dict names every submodule
                # and every shape — so write down the blueprint rather than only
                # the complaint. Nothing can be reconstructed from a sentence
                # saying it failed.
                STATE["tried"].append(
                    "%s: %d weight tensors, but this repo ships no architecture "
                    "code to load them into — see /architecture" % (rel, len(sd)))
                blob, sd = None, None
                gc.collect()
                break      # a second checkpoint teaches nothing and costs memory
            for _, name, cls in classes[:6]:
                made, how = construct(cls, cfg)
                if made is None:
                    STATE["tried"].append("%s(): could not construct from config" % name)
                    continue
                try:
                    missing, unexpected = made.load_state_dict(sd, strict=False)
                except Exception as e:
                    STATE["tried"].append("%s.load_state_dict: %s" % (name, str(e)[:120]))
                    continue
                matched = len(sd) - len(unexpected)
                # A class that accepts a handful of keys is the wrong class; a
                # real match takes nearly all of them.
                if matched >= max(8, int(0.6 * len(sd))):
                    obj = made
                    STATE["detail"] = ("built %s from config and loaded %d/%d tensors "
                                       "from %s (%d missing)"
                                       % (how, matched, len(sd), rel, len(missing)))
                    break
                STATE["tried"].append(
                    "%s: only %d/%d tensors matched — first missing: %s; first "
                    "unexpected: %s" % (name, matched, len(sd),
                                        ", ".join(list(missing)[:6]) or "none",
                                        ", ".join(list(unexpected)[:6]) or "none"))
            if obj is not None:
                break

    if obj is None:
        STATE["ready"] = False
        STATE["detail"] = summarise_failure(files, ckpts)
        log("!! " + STATE["detail"])
        return

    try:
        obj = obj.to(DEVICE).eval()
    except Exception as e:
        STATE["tried"].append("to(%s): %s" % (DEVICE, str(e)[:120]))

    # Which call do we have? Stream-level beats whole-clip conversion.
    for name in STREAM_CALLS:
        if callable(getattr(obj, name, None)):
            STATE["binding"] = ("streams", name)
            break
    else:
        for name in CONVERT_CALLS:
            if callable(getattr(obj, name, None)):
                STATE["binding"] = ("convert", name)
                break

    if STATE["binding"] is None:
        STATE["ready"] = False
        methods = [m for m in dir(obj) if not m.startswith("_")
                   and callable(getattr(obj, m, None))][:40]
        STATE["detail"] = ("Built %s, but none of its methods are a recognised "
                           "inference call. Public methods: %s"
                           % (type(obj).__name__, ", ".join(methods)))
        log("!! " + STATE["detail"])
        return

    STATE["model"] = obj
    STATE["ready"] = True
    log("bound to %s.%s — %s" % (type(obj).__name__, STATE["binding"][1], STATE["detail"]))


def summarise_failure(files, ckpts):
    """Say what is actually in the repo, so the next step is obvious.

    A failure here is nearly always one of two things: the architecture code is
    not in the repo, or it is there under a shape this script did not try. The
    difference is visible in the file list, so print it rather than a verdict.
    """
    py = [f for f in files if f.endswith(".py")]
    lines = ["Could not construct the model from this repo."]
    if not py and STATE["arch"]:
        a = STATE["arch"]
        lines.append(
            "The weights read fine — %d tensors, %s parameters, in %d top-level "
            "modules (%s) — but the repo ships no Python, so there is no class to "
            "load them into. A checkpoint cannot be run on its own: it is the "
            "parameters, not the network. The full parameter tree is at "
            "/architecture and in the log, which is what a matching model "
            "definition has to be written against. If you have the training "
            "script this came from, that script already contains the class — "
            "set MODEL_DEF at the top of this file to its path or URL, or drop "
            "it at %s, and it will be constructed from config.json and checked "
            "against the checkpoint."
            % (a["tensors"], format(a["parameters"], ","),
               len(a["top_level_modules"]), ", ".join(list(a["top_level_modules"])[:8]),
               os.path.join(WORKDIR, "model_def.py")))
        meta = a.get("checkpoint_meta") or {}
        if meta.get("developer") or meta.get("name"):
            lines.append("The checkpoint says it was trained by %s as \"%s\"%s."
                         % (meta.get("developer", "someone"),
                            meta.get("name", "this model"),
                            ", step %s" % meta["step"] if meta.get("step") else ""))
    elif not py:
        lines.append(
            "It contains weights (%s) but no Python at all, so there is no "
            "architecture to load them into."
            % (", ".join(os.path.basename(c) for c in ckpts) or "none found"))
    else:
        lines.append("Python found: %s." % ", ".join(py))
    lines.append("Files: %s." % ", ".join(files[:25]))
    if STATE["ckpt_keys"]:
        lines.append("First weight names: %s." % ", ".join(STATE["ckpt_keys"]))
    lines.append("Attempts: %s" % (" | ".join(STATE["tried"]) or "none"))
    return " ".join(lines)


# ---------------------------------------------------------------------------
# The two operations the panel offers
# ---------------------------------------------------------------------------
@torch.inference_mode()
def run_convert(voice_wav, perf_wav, residual_from="performance", pitch_shift=0.0,
                steps=STEPS, lyrics=""):
    """Sing the performance clip in the voice clip's voice."""
    model = STATE["model"]
    kind, name = STATE["binding"]
    fn = getattr(model, name)

    log_f0, vuv = f0_contour(perf_wav)
    log_f0 = shift_pitch(log_f0, pitch_shift)
    residual_src = perf_wav if residual_from == "performance" else voice_wav

    if kind == "streams":
        # The real transplant: each stream driven from the clip it belongs to.
        out = fn(
            source={"log_f0": log_f0, "voiced": vuv},
            filter={"reference_audio": voice_wav, "sample_rate": SAMPLE_RATE},
            residual={"audio": residual_src, "sample_rate": SAMPLE_RATE},
            lyrics=lyrics or None,
            steps=int(steps),
        )
        used = "stream transplant (%s)" % name
    else:
        # Repo exposes only whole-clip conversion. Still the right mapping —
        # content from the performance, identity from the voice — but the stream
        # routing is the repo's, not ours.
        out = _call_flexible(fn, perf_wav, voice_wav, steps, lyrics)
        used = "repo conversion call (%s)" % name

    wav, note = _as_audio(out)
    # The model may have degraded something rather than failing outright; say
    # so alongside the vocoder note instead of letting it pass unremarked.
    warn = getattr(model, "last_warning", "")
    if warn:
        note = (note + "; " + warn) if note else warn
    return wav, used, note


@torch.inference_mode()
def run_generate(voice_wav, lyrics, melody_wav=None, steps=STEPS, pitch_shift=0.0):
    """Generation mode: sing lyrics from scratch in the cloned voice."""
    model = STATE["model"]
    kind, name = STATE["binding"]
    fn = getattr(model, name)

    kwargs = {"lyrics": lyrics, "steps": int(steps)}
    if melody_wav is not None:
        log_f0, vuv = f0_contour(melody_wav)
        log_f0 = shift_pitch(log_f0, pitch_shift)
        kwargs["source"] = {"log_f0": log_f0, "voiced": vuv}
    if kind == "streams":
        kwargs["filter"] = {"reference_audio": voice_wav, "sample_rate": SAMPLE_RATE}
        out = fn(**kwargs)
    else:
        out = _call_flexible(fn, None, voice_wav, steps, lyrics, melody_wav)
    wav, note = _as_audio(out)
    return wav, "generation (%s)" % name, note


def _call_flexible(fn, content_wav, reference_wav, steps, lyrics, melody_wav=None):
    """Try the argument spellings a repo of this kind plausibly uses.

    Every attempt and its error is kept, so a failure reports what was actually
    tried rather than a bare traceback.
    """
    import inspect
    attempts = []
    base = {"sample_rate": SAMPLE_RATE, "sr": SAMPLE_RATE, "steps": int(steps),
            "num_steps": int(steps), "n_steps": int(steps)}
    names_content = ["source_audio", "content_audio", "audio", "wav", "src", "source"]
    names_ref     = ["reference_audio", "speaker_audio", "target_audio", "ref", "reference",
                     "prompt_audio", "speaker_wav"]
    try:
        sig = set(inspect.signature(fn).parameters)
    except (TypeError, ValueError):
        sig = set()

    kwargs = {}
    if content_wav is not None:
        for n in names_content:
            if n in sig:
                kwargs[n] = content_wav
                break
    for n in names_ref:
        if n in sig:
            kwargs[n] = reference_wav
            break
    if lyrics and "lyrics" in sig:
        kwargs["lyrics"] = lyrics
    if melody_wav is not None and "melody" in sig:
        kwargs["melody"] = melody_wav
    for k, v in base.items():
        if k in sig:
            kwargs[k] = v

    if kwargs:
        try:
            return fn(**kwargs)
        except Exception as e:
            attempts.append("kwargs %s: %s" % (sorted(kwargs), str(e)[:200]))

    for args in ([content_wav, reference_wav] if content_wav is not None else [reference_wav],
                 [reference_wav, content_wav] if content_wav is not None else None):
        if args is None:
            continue
        try:
            return fn(*[a for a in args if a is not None])
        except Exception as e:
            attempts.append("positional %d: %s" % (len(args), str(e)[:200]))

    raise RuntimeError("No call signature worked. Tried: " + " | ".join(attempts))


N_MELS = 100          # what the model card says the decoder emits


def looks_like_mel(arr):
    """Is this a mel spectrogram rather than a waveform?

    It matters because the two are not interchangeable and the difference is
    silent: flattening a (n_mels, frames) mel into a 1-D array interleaves
    frequency bins as if they were time samples, which plays as noise. The
    decoder in this model emits a 100-band mel, so this is the expected shape,
    not an edge case.
    """
    if arr.ndim != 2:
        return False
    a, b = arr.shape
    # One axis the size of the mel band count, the other much longer.
    return (a == N_MELS and b >= 8) or (b == N_MELS and a >= 8)


def mel_to_wave(mel):
    """Vocode a mel with Griffin-Lim, since no neural vocoder ships here.

    Griffin-Lim is phase-blind, so this sounds thinner and slightly metallic
    next to a trained vocoder. It is still recognisably the right voice singing
    the right notes, which is the whole point, and it is honest about what it is
    — the response says a fallback vocoder was used.
    """
    import librosa
    if mel.shape[0] != N_MELS:
        mel = mel.T
    power = np.exp(mel.astype("float64"))      # mel_spec() stored log-magnitude
    wav = librosa.feature.inverse.mel_to_audio(
        power, sr=SAMPLE_RATE, n_fft=1024, hop_length=256, n_iter=48)
    return np.asarray(wav, dtype="float32")


def _as_audio(out):
    """Whatever came back -> (mono float32 waveform, note about how).

    Batch and channel dimensions are squeezed before anything else. The previous
    version reshaped any multi-dimensional output into one long row, which is
    correct for (1, samples) and destroys a (1, n_mels, frames) mel.
    """
    note = ""
    if isinstance(out, dict):
        for k in ("audio", "wav", "waveform", "samples", "output", "mel", "mel_spectrogram"):
            if k in out:
                out = out[k]
                break
    if isinstance(out, (list, tuple)) and out:
        out = out[0]
    if torch.is_tensor(out):
        out = out.detach().float().cpu().numpy()
    out = np.asarray(out, dtype="float32")

    while out.ndim > 1 and out.shape[0] == 1:      # drop batch/channel dims of 1
        out = out[0]

    if looks_like_mel(out):
        out = mel_to_wave(out)
        note = "mel output vocoded with Griffin-Lim (no neural vocoder in this build)"
    elif out.ndim > 1:
        # Genuinely multi-channel audio: average the channels, keeping time.
        time_axis = int(np.argmax(out.shape))
        out = out.mean(axis=1 - time_axis) if out.ndim == 2 else out.reshape(-1)

    out = np.asarray(out, dtype="float32").reshape(-1)
    peak = float(np.max(np.abs(out))) if out.size else 0.0
    if peak > 1.0:
        out = out / peak * 0.97
    return out, note


# ---------------------------------------------------------------------------
# HTTP
# ---------------------------------------------------------------------------
from fastapi import FastAPI, File, Form, UploadFile, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, Response

app = FastAPI(title="TriStream-SVS")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"],
                   allow_headers=["*"], expose_headers=["*"])


@app.get("/health")
def health():
    return {
        "ok": bool(STATE["ready"]),
        "build": BUILD,
        "model": REPO_ID,
        "device": STATE["device"],
        "gpu": GPU_NAME,
        "sample_rate": SAMPLE_RATE,
        "loading": bool(STATE["loading"]),
        "uptime": round(time.time() - STATE["started"], 1),
        # A server that quietly died and came back looks identical to one that
        # never moved, right up until you notice the work in flight was lost.
        "restarts": supervisor_restarts(),
        # Which copy of the model definition is loaded. Without this, a fix
        # that has not reached the machine is indistinguishable from a fix
        # that did not work.
        "model_def_version": STATE["model_def_version"],
        # If the tunnel has rotated, the address the caller used to reach this
        # is not the address that works now — and they cannot see that from here.
        "tunnel": STATE["tunnel"],
        "tunnel_restarts": STATE["tunnel_restarts"],
        "binding": STATE["binding"][1] if STATE["binding"] else None,
        "mode": STATE["binding"][0] if STATE["binding"] else None,
        "detail": STATE["detail"] or ("Still loading — downloading weights and "
                                      "importing the repo." if STATE["loading"] else ""),
        "files": STATE["files"],
        "tried": STATE["tried"],
        "modes": ["convert", "generate"],
        # A summary only — the full parameter tree is at /architecture, since
        # several hundred tensor names do not belong in a status payload.
        "architecture": None if not STATE["arch"] else {
            "tensors": STATE["arch"]["tensors"],
            "parameters": STATE["arch"]["parameters"],
            "config": STATE["arch"]["config"],
            "checkpoint_meta": STATE["arch"]["checkpoint_meta"],
            "top_level_modules": {k: v["tensors"] for k, v
                                  in STATE["arch"]["top_level_modules"].items()},
        },
    }


@app.get("/architecture")
def architecture():
    """The full parameter tree, when the weights loaded but the code was absent.

    This is the thing a replacement nn.Module has to match, key for key.
    """
    if not STATE["arch"]:
        return JSONResponse({"detail": "No architecture dump — either the model "
                                       "loaded, or no checkpoint could be read."},
                            status_code=404)
    return STATE["arch"]


def _guard():
    if STATE["loading"]:
        raise HTTPException(status_code=503, detail="Still loading — the weights are "
                                                    "downloading or the repo is importing.")
    if not STATE["ready"]:
        raise HTTPException(status_code=503, detail=STATE["detail"] or "Model did not load.")


@app.post("/v1/voice/convert")
async def convert(voice: UploadFile = File(...),
                  performance: UploadFile = File(...),
                  residual_from: str = Form("performance"),
                  pitch_shift: float = Form(0.0),
                  steps: int = Form(STEPS),
                  lyrics: str = Form("")):
    _guard()
    try:
        v, v_trim = load_audio(await voice.read(), label="voice clip")
        p, p_trim = load_audio(await performance.read(), label="performance clip")
        t0 = time.time()
        wav, used, notes = run_convert(v, p, residual_from, pitch_shift, steps, lyrics)
        headers = {"X-TriStream-Path": used,
                   "X-TriStream-Seconds": "%.1f" % (time.time() - t0),
                   "Content-Disposition": 'attachment; filename="tristream.wav"'}
        trims = ["%s %.0fs" % (n, t) for n, t in (("voice", v_trim), ("performance", p_trim)) if t]
        if trims:
            headers["X-TriStream-Trimmed"] = "first %.0fs used (%s)" % (MAX_SECONDS, ", ".join(trims))
        if notes:
            headers["X-TriStream-Note"] = notes
        return Response(content=to_wav_bytes(wav), media_type="audio/wav", headers=headers)
    except HTTPException:
        raise
    except ValueError as e:
        # Bad input, not a broken server. Say so with a 400 so the panel can
        # show the reason instead of "the server had an internal error".
        return JSONResponse(status_code=400, content={"error": str(e)})
    except Exception as e:
        traceback.print_exc()
        return JSONResponse(status_code=500, content={"error": str(e)})


@app.post("/v1/voice/generate")
async def generate(voice: UploadFile = File(...),
                   lyrics: str = Form(...),
                   melody: UploadFile = File(None),
                   pitch_shift: float = Form(0.0),
                   steps: int = Form(STEPS)):
    _guard()
    try:
        v, v_trim = load_audio(await voice.read(), label="voice clip")
        m, m_trim = (None, 0.0)
        if melody is not None:
            raw = await melody.read()
            if raw:
                m, m_trim = load_audio(raw, label="melody guide")
        t0 = time.time()
        wav, used, notes = run_generate(v, lyrics, m, steps, pitch_shift)
        headers = {"X-TriStream-Path": used,
                   "X-TriStream-Seconds": "%.1f" % (time.time() - t0),
                   "Content-Disposition": 'attachment; filename="tristream.wav"'}
        trims = ["%s %.0fs" % (n, t) for n, t in (("voice", v_trim), ("melody", m_trim)) if t]
        if trims:
            headers["X-TriStream-Trimmed"] = "first %.0fs used (%s)" % (MAX_SECONDS, ", ".join(trims))
        if notes:
            headers["X-TriStream-Note"] = notes
        return Response(content=to_wav_bytes(wav), media_type="audio/wav", headers=headers)
    except HTTPException:
        raise
    except ValueError as e:
        # Bad input, not a broken server. Say so with a 400 so the panel can
        # show the reason instead of "the server had an internal error".
        return JSONResponse(status_code=400, content={"error": str(e)})
    except Exception as e:
        traceback.print_exc()
        return JSONResponse(status_code=500, content={"error": str(e)})


# ---------------------------------------------------------------------------
# Cloudflare quick tunnel
# ---------------------------------------------------------------------------
def start_tunnel():
    """Keep a quick tunnel up, and announce the address every time it changes.

    A quick tunnel is free and unguaranteed — cloudflared exits on its own often
    enough that treating it as permanent is wrong. The previous version read
    lines until it saw a URL, then drained the pipe until cloudflared exited, at
    which point the thread simply ended. The server carried on serving perfectly
    on localhost while its public address answered nothing, which from the
    outside is indistinguishable from the server having stopped — and a
    supervisor watching the server process sees nothing wrong, because nothing
    is wrong with the server.
    """
    pat = re.compile(r"https://[a-z0-9-]+\.trycloudflare\.com")
    attempt = 0
    while True:
        attempt += 1
        began = time.time()
        try:
            proc = subprocess.Popen(
                ["cloudflared", "tunnel", "--url", "http://127.0.0.1:%d" % PORT,
                 "--no-autoupdate"],
                stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
        except Exception as e:
            log("!! could not start cloudflared: %s" % str(e)[:160])
            time.sleep(15)
            continue

        seen = None
        for line in proc.stdout:
            m = pat.search(line)
            if m and m.group(0) != seen:
                seen = m.group(0)
                STATE["tunnel"] = seen
                STATE["tunnel_since"] = time.time()
                try:
                    with open(URL_PATH, "w") as f:
                        f.write(seen)
                except Exception:
                    pass
                log("")
                log("=" * 72)
                log(("TUNNEL URL: " if attempt == 1 else "NEW TUNNEL URL: ") + seen)
                if attempt > 1:
                    log("The previous address is dead. Paste this one into the Voice panel.")
                log("=" * 72)
                log("")

        code = proc.wait()
        lived = time.time() - began
        STATE["tunnel"] = None
        STATE["tunnel_restarts"] += 1
        log("!! the cloudflare tunnel exited (%s) after %.0fs — restarting it. "
            "The server itself is fine; only the public address was lost."
            % ("code %d" % code if code >= 0 else "signal %d" % -code, lived))
        # A tunnel that lasted a while is not a failing tunnel, so do not let a
        # long-lived one inherit the backoff of an early crash.
        if lived > 120:
            attempt = 1
        time.sleep(min(30, 2 ** min(attempt, 4)))


threading.Thread(target=start_tunnel, daemon=True).start()


def load_in_background():
    """Load the model without the API depending on it finishing, or succeeding.

    This used to run inline, before uvicorn.run(). Loading a model means
    importing the repo's own Python, and a repo script that calls sys.exit()
    — argparse does exactly that when a training script is missing an argument
    — raises SystemExit, which is a BaseException and walks straight through
    `except Exception`. The process died, the tunnel stayed up with nothing
    behind it, and the browser got a Cloudflare 502 carrying no CORS headers,
    which surfaces as a bare "Failed to fetch" with nothing to diagnose.

    So: serve first, load second, and catch everything. A failed or slow load
    is now something /health can describe instead of a server that vanished.
    """
    try:
        bind_model()
    except KeyboardInterrupt:
        raise
    except BaseException as e:
        STATE["ready"] = False
        STATE["detail"] = "Load failed: %s: %s" % (type(e).__name__, e)
        traceback.print_exc()
    finally:
        STATE["loading"] = False
        if STATE["ready"]:
            log("model ready")
        else:
            log("!! model not ready — /health has the detail")


threading.Thread(target=load_in_background, daemon=True).start()

import uvicorn
uvicorn.run(app, host="0.0.0.0", port=PORT, log_level="info")
'''

with open(RUNNER_PATH, "w") as f:
    f.write(RUNNER_SRC)


# -----------------------------------------------------------------------------
# 2b. Supervisor
#
# A detached server that dies stays dead, and from the outside that is
# indistinguishable from never having started: the tunnel URL keeps resolving
# for a while, requests just stop being answered. The common killers in a
# notebook are an OOM kill (exit -9) and the runtime reclaiming background
# processes, neither of which the server can prevent or even report on its own
# way out.
#
# So the thing that gets launched is this supervisor, and it launches the
# server. It restarts on an unexpected exit, backs off if the exit is
# immediate, records the exit code and the tail of the log, and stops trying
# only on a clean stop signal or after repeated instant failures — because a
# server that cannot survive two seconds will not survive the twentieth
# attempt either, and a restart loop hides the real error.
# -----------------------------------------------------------------------------
SUPERVISOR_SRC = r"""
import json, os, signal, subprocess, sys, time

CFG_PATH = sys.argv[1]
CFG      = json.load(open(CFG_PATH))
WORKDIR  = CFG["workdir"]
RUNNER   = os.path.join(WORKDIR, "runner.py")
LOG      = os.path.join(WORKDIR, "server.log")
STATUS   = os.path.join(WORKDIR, "supervisor.json")

MAX_INSTANT_FAILURES = 5     # consecutive exits inside MIN_HEALTHY_SECONDS
MIN_HEALTHY_SECONDS  = 30.0  # ran at least this long => the attempt counted

state = {"state": "starting", "restarts": 0, "started": time.time(),
         "last_exit": None, "detail": "", "pid": os.getpid()}
child = None


def save():
    tmp = STATUS + ".tmp"
    with open(tmp, "w") as f:
        json.dump(state, f)
    os.replace(tmp, STATUS)


def describe(code):
    if code is None:
        return "still running"
    if code < 0:
        name = {9: "SIGKILL — almost certainly out of memory",
                15: "SIGTERM — asked to stop",
                1: "SIGHUP — the session went away",
                2: "SIGINT — interrupted"}.get(-code, "signal %d" % -code)
        return name
    return "exit code %d" % code


def tail(n=25):
    try:
        with open(LOG) as f:
            return "".join(f.readlines()[-n:])
    except Exception:
        return ""


def stop(signum, frame):
    state["state"] = "stopped"
    state["detail"] = "supervisor asked to stop"
    save()
    # Signal the child, but do NOT wait for it here. The main thread is already
    # inside child.wait(), and Popen.wait() is not re-entrant — it takes a lock
    # it is already holding, so waiting from the handler deadlocks the
    # supervisor and nothing ever stops. A stop that hangs is worse than no
    # stop at all, because the port stays held.
    if child is not None:
        for sig in (signal.SIGTERM, signal.SIGKILL):
            try:
                if child.poll() is None:
                    child.send_signal(sig)
            except Exception:
                break
    # os._exit rather than sys.exit: SystemExit would unwind through the same
    # wait() and interpreter shutdown can block on it for the same reason.
    os._exit(0)


signal.signal(signal.SIGTERM, stop)
signal.signal(signal.SIGINT, stop)

instant = 0
save()

while True:
    began = time.time()
    with open(LOG, "a") as log:
        log.write("\n--- supervisor: starting the server (attempt %d) ---\n"
                  % (state["restarts"] + 1))
        log.flush()
        child = subprocess.Popen([sys.executable, "-u", RUNNER, CFG_PATH],
                                 stdout=log, stderr=subprocess.STDOUT,
                                 stdin=subprocess.DEVNULL, cwd=WORKDIR)
        state["state"] = "running"
        state["child_pid"] = child.pid
        save()
        code = child.wait()

    ran = time.time() - began
    reason = describe(code)
    state["last_exit"] = {"code": code, "reason": reason,
                          "ran_seconds": round(ran, 1), "at": time.time()}

    if code == -signal.SIGTERM or code == 0:
        # A deliberate stop, or the server choosing to exit. Not ours to undo.
        state["state"] = "stopped"
        state["detail"] = "server exited cleanly (%s)" % reason
        save()
        with open(LOG, "a") as log:
            log.write("--- supervisor: %s after %.0fs; not restarting ---\n" % (reason, ran))
        break

    instant = instant + 1 if ran < MIN_HEALTHY_SECONDS else 0
    state["restarts"] += 1
    state["detail"] = "%s after %.0fs" % (reason, ran)

    if instant >= MAX_INSTANT_FAILURES:
        state["state"] = "giving up"
        state["detail"] = ("%s, and it has failed %d times in a row within %ds. "
                           "Restarting again would just hide the error — the log "
                           "tail is above." % (reason, instant, int(MIN_HEALTHY_SECONDS)))
        save()
        with open(LOG, "a") as log:
            log.write("--- supervisor: giving up after %d immediate failures ---\n%s\n"
                      % (instant, tail()))
        break

    delay = min(60, 2 ** min(instant, 5))
    state["state"] = "restarting"
    save()
    with open(LOG, "a") as log:
        log.write("--- supervisor: %s after %.0fs; restarting in %ds "
                  "(restart %d) ---\n" % (reason, ran, delay, state["restarts"]))
        if code == -9:
            log.write("--- that exit code is the kernel OOM killer. The runtime ran "
                      "out of RAM; a smaller model or a bigger machine is the fix. ---\n")
    time.sleep(delay)
"""

with open(SUPERVISOR_PATH, "w") as f:
    f.write(SUPERVISOR_SRC)

# -----------------------------------------------------------------------------
# 3. Launch, detached
# -----------------------------------------------------------------------------
print("\nStarting the supervisor in its own session (it outlives this cell,\n      and restarts the server if it dies)…")
if os.path.exists(URL_PATH):
    os.remove(URL_PATH)

# Truncated, not appended. An accumulating log replays every previous run's
# errors on every read, and a stale [Errno 98] from three runs ago is
# indistinguishable from a live one.
_log = open(LOG_PATH, "w")
_log.write("TriStream-SVS build %d — started %s — port %d\n"
           % (BUILD, time.strftime("%Y-%m-%d %H:%M:%S"), PORT))
_log.flush()
if os.path.exists(STATUS_PATH):
    os.remove(STATUS_PATH)
_proc = subprocess.Popen(
    [sys.executable, "-u", SUPERVISOR_PATH, CFG_PATH],
    stdout=_log, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL,
    start_new_session=True, cwd=WORKDIR)
open(PID_PATH, "w").write(str(_proc.pid))
print("pid", _proc.pid, "· port", PORT, "· log:", LOG_PATH)

print("\nWaiting for the Cloudflare tunnel…")
url = None
for i in range(180):
    if os.path.exists(URL_PATH):
        url = open(URL_PATH).read().strip()
        if url:
            break
    if _proc.poll() is not None:
        print("\n!! The supervisor exited. Last 40 log lines:\n")
        print("".join(open(LOG_PATH).readlines()[-40:]))
        try:
            st = json.load(open(STATUS_PATH))
            print("\nsupervisor status: %s — %s" % (st.get("state"), st.get("detail")))
        except Exception:
            pass
        raise SystemExit(1)
    time.sleep(1)

print("\n" + "=" * 72)
if url:
    print("  TriStream is up.")
    print()
    print("  URL:  " + url)
    print()
    print("  1. Open https://corx-labs.com/chat/")
    print("  2. Press the microphone button in the header")
    print("  3. Paste the URL above, press Connect")
    print()
    print("  Check " + url + "/health first — it reports which entry point the")
    print("  model bound to. If `ok` is false, the detail field says exactly what")
    print("  the repo contained and what was tried.")
else:
    print("  No tunnel URL after 180s. Log tail:")
    print("".join(open(LOG_PATH).readlines()[-40:]))
print("=" * 72)
print("\nFollow the log with:  !tail -f " + LOG_PATH)


# -----------------------------------------------------------------------------
# 4. Stay with it
#
# The cell used to print the URL and finish. That looks like the server stopped
# — and worse, a quick tunnel rotates its address, so when it does there is no
# longer anything running that could tell you the new one. Staying here prints
# the new address the moment it changes, shows a heartbeat so "still up" is
# visible rather than assumed, and keeps the notebook busy so the runtime is
# less likely to be reclaimed out from under it.
#
# Stopping this cell does not stop the server: it is detached on purpose and
# keeps serving. Re-running the cell adopts it.
# -----------------------------------------------------------------------------
if KEEP_ALIVE and url:
    import urllib.request as _u

    def _health():
        try:
            with _u.urlopen("http://127.0.0.1:%d/health" % PORT, timeout=4) as r:
                return json.loads(r.read())
        except Exception:
            return None

    print("\nWatching. The server keeps running if you stop this cell.\n")
    current, last_beat, quiet = url, 0.0, 0
    try:
        while True:
            time.sleep(2)

            if os.path.exists(URL_PATH):
                try:
                    fresh = open(URL_PATH).read().strip()
                except Exception:
                    fresh = ""
                if fresh and fresh != current:
                    current = fresh
                    print("\n" + "=" * 72)
                    print("  THE TUNNEL ADDRESS CHANGED. The old one is dead.")
                    print()
                    print("  URL:  " + current)
                    print()
                    print("  Paste this into the Voice panel and press Connect.")
                    print("=" * 72 + "\n")

            if time.time() - last_beat < 30:
                continue
            last_beat = time.time()

            h = _health()
            if h is None:
                quiet += 1
                print("[%s] no answer from the server (%d in a row) — check the log"
                      % (time.strftime("%H:%M:%S"), quiet))
                if _proc.poll() is not None:
                    print("The supervisor has exited. Last 20 log lines:\n")
                    print("".join(open(LOG_PATH).readlines()[-20:]))
                    break
                continue

            quiet = 0
            state = ("ready" if h.get("ok") else
                     "loading" if h.get("loading") else "not ready")
            extra = ""
            if h.get("restarts"):
                extra += " · %d server restart(s)" % h["restarts"]
            if h.get("tunnel_restarts"):
                extra += " · %d tunnel restart(s)" % h["tunnel_restarts"]
            print("[%s] %s · up %.0fs%s" % (time.strftime("%H:%M:%S"), state,
                                            h.get("uptime", 0), extra))
    except KeyboardInterrupt:
        print("\nStopped watching. The server is still running at " + current)
        print("Re-run this cell to watch it again, or check " + LOG_PATH)

Ready

Start the server, then open the chat.

Once the cell prints a URL and you have pasted it into settings, you can talk to CorX3.8.

Open the chat About CorX3.8

Running CorX3.8, and how the chat works.

Copy this and run it on a GPU.

Paste the URL into the chat.

Wait for “API IS LIVE”

Open the chat and its settings

Watch the status dot

Leave the cell running

It runs code, files and a plan on its own.

What it can do

The panel, search & uploads

Low to Max — a real dial, not a label.

Saved in this browser, nowhere else.

Conversations

Interrupted runs resume themselves

Profile

Cross-chat memory

What this set-up is, and isn’t.

The uncensored build, at full precision.

Clone a voice with TriStream-SVS.

Press Copy, paste, run

Check /health first

Open the chat and press the microphone

Start the server, then open the chat.

Check `/health` first