CorX Labs
Open the chat
Open the chat

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

# 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
# ------------------------------------------

print("Installing deps (first run only)...")
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-U",
                "fastapi", "uvicorn[standard]", "transformers",
                "accelerate"], 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.")

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

print("Loading CorX3.8-27B in bf16 (1-2 min)...")
tok = AutoTokenizer.from_pretrained(MODEL_REPO, **tok_kwargs)
if tok.pad_token is None:
    tok.pad_token = tok.eos_token

load_kwargs = dict(device_map="auto", trust_remote_code=True,
                   low_cpu_mem_usage=True, torch_dtype=torch.bfloat16)
if HF_TOKEN:
    load_kwargs["token"] = HF_TOKEN

model = AutoModelForCausalLM.from_pretrained(MODEL_REPO, **load_kwargs)
model.eval()
print(f"Model ready. VRAM: {torch.cuda.memory_allocated()/1e9:.1f} GB")

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

# Worker wrapper to ensure gradients are off in the background thread
def _generate_thread_worker(kwargs):
    with torch.inference_mode():
        try:
            model.generate(**kwargs)
        except Exception as e:
            print(f"\n[Generation Error in thread] {e}")

def _generate(messages, max_new_tokens, temperature, streamer=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,),
                         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:
            streamer = TextIteratorStreamer(
                tok, skip_prompt=True, skip_special_tokens=True, timeout=15.0)
            _generate(messages, max_new, temperature, streamer=streamer)

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

            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}]}
                    yield f"data: {json.dumps(data)}\n\n"

                except queue.Empty:
                    print("[api] Streamer timed out, returning error to client.")
                    err = {"id": cid, "object": "chat.completion.chunk",
                           "model": "corx3.8",
                           "choices": [{"index": 0, "delta": {"content": "\n[Generation Error]"},
                                        "finish_reason": "error"}]}
                    yield f"data: {json.dumps(err)}\n\n"
                    break

            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 56 GB in full bf16. The code loads it in bf16, so use a big GPU (A100/H100 class) or switch the load to 4-bit if your card is smaller.

Step 2 · Connect

Paste the URL into the chat.

  1. 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.

  2. Open the chat and its settings

    Go to the chat and click the gear icon at the top. Paste the URL into Model endpoint and press Save.

  3. 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.

  4. 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.

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
  • 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.

LevelTokensTool roundsSearch results
Low51233
Medium1,02454
High2,04886
Extra4,096148
Max6,1442410

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.

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