llm-proxy/llm-proxy.py

#!/usr/bin/env python3
"""LLM router: model-prefix → OpenRouter (cloud) or local model-aware fanout.

Local backends: model-name routed when possible, first-up failover otherwise.
  - On POST with a known model id, the request goes to the backend whose
    /v1/models advertises it (cached 30s).
  - On unknown model or empty body, falls back to first-up wins,
    preserving legacy semantics.

Cloud routing: HTTPS to openrouter.ai, server-side Bearer auth from OPENROUTER_API_KEY.

/v1/models: merged cloud catalog (live-filtered against OpenRouter's
/api/v1/models) + union of every reachable local backend's model list.
Entries are visually tagged in their id:
  - "[$] vendor/model"     paid cloud (manually curated or auto-discovered)
  - "[free] vendor/model"  free cloud (auto-detected by :free suffix)
  - "[local] alias"        local backend

A periodic background thread (every 3 h) re-fetches the full OpenRouter model
list and:
  a) adds newly free models >7 B parameters to the curated free catalog
  b) removes models that are no longer free
  c) probes each free model with a tiny 1-token request; if it returns 429 the
     model id gets a "-na" suffix (not available).
  d) auto-discovers cheap paid models (prompt < $0.50/MTok) from known
     prefixes and adds them to the paid catalog.

Paid models can also be added manually by editing cloud-catalog.json's "paid"
section — the background checker never removes them.

Inbound requests have the tag + optional "-na" suffix stripped before routing,
so OWUI sending back the displayed id Just Works.

Config:
  /etc/llm-proxy/local-backends.json  — list of {name, host, port} dicts
  /etc/llm-proxy/cloud-catalog.json   — persisted cloud model catalog + status

Signals:
  SIGHUP   reload local-backends.json
  SIGUSR1  drop discover caches + trigger an immediate background re-check
"""
import http.server
import http.client
import ssl
import select
import socket
import sys
import os
import os.path
import time
import json
import threading
import signal
import re
import copy
import traceback

CONFIG_DIR          = os.environ.get("LLM_PROXY_CONFIG_DIR", "/etc/llm-proxy")
LOCAL_BACKENDS_FILE = os.path.join(CONFIG_DIR, "local-backends.json")
CLOUD_CATALOG_FILE  = os.path.join(CONFIG_DIR, "cloud-catalog.json")

# Fallback only, when local-backends.json is absent. Real backends belong in
# $LLM_PROXY_CONFIG_DIR/local-backends.json (see local-backends.example.json).
DEFAULT_LOCAL_BACKENDS = [
    ("local", "127.0.0.1", 8080),
]
# Optional display aliases {"<model-id>": "<label>"} from
# $LLM_PROXY_CONFIG_DIR/friendly-names.json (see friendly-names.example.json).
def _load_friendly_names():
    try:
        with open(os.path.join(CONFIG_DIR, "friendly-names.json")) as f:
            return json.load(f)
    except Exception:
        return {}
FRIENDLY_NAMES = _load_friendly_names()


# Optional list of local model-id substrings whose thinking should be split into a
# separate `reasoning` stream (via the chat-template `enable_thinking` kwarg), so
# clients can surface the model's reasoning instead of having it inlined or dropped.
# From $LLM_PROXY_CONFIG_DIR/reasoning-models.json (a JSON list of substrings).
# Absent / empty = feature off, so this changes nothing unless configured.
def _load_reasoning_models():
    try:
        with open(os.path.join(CONFIG_DIR, "reasoning-models.json")) as f:
            data = json.load(f)
        return [s for s in data if isinstance(s, str)] if isinstance(data, list) else []
    except Exception:
        return []
REASONING_LOCAL_MODELS = _load_reasoning_models()


def _is_reasoning_local(model):
    return bool(model) and any(sub in model for sub in REASONING_LOCAL_MODELS)


CONNECT_TIMEOUT      = 3
READ_TIMEOUT         = 1800
DISCOVERY_TTL        = 30
CLOUD_DISCOVERY_TTL  = 600
CHECK_INTERVAL       = 10800
PROBE_TIMEOUT        = 15
NA_SUFFIX            = "-na"
DOWN_SUFFIX          = "-down"
LAST_KNOWN_FILE      = os.path.join(CONFIG_DIR, "last-known-models.json")

OPENROUTER_HOST = "openrouter.ai"
OPENROUTER_PORT = 443
CLOUD_PREFIXES = (
    "moonshotai/", "poolside/", "z-ai/", "cognitivecomputations/", "liquid/",
    "openrouter/", "anthropic/", "openai/", "google/", "deepseek/", "meta-llama/",
    "mistralai/", "qwen/", "x-ai/", "nvidia/", "cohere/", "perplexity/",
    "amazon/", "microsoft/", "nousresearch/",
)

PAID_DISCOVER_THRESHOLD = 0.50  # max prompt $/MTok for auto-discover

TAG_PAID  = "[$] "
TAG_FREE  = "[free] "
TAG_LOCAL = "[local] "
TAGS = (TAG_PAID, TAG_FREE, TAG_LOCAL)

OR_KEY = os.environ.get("OPENROUTER_API_KEY", "")
ZEN_HOST = "opencode.ai"
ZEN_PORT = 443
LOG = sys.stdout

_disc_cache    = {}
_disc_lock     = threading.Lock()

_cloud_disc    = {"ids": None, "expiry": 0.0}
_cloud_lock    = threading.Lock()

LOCAL_BACKENDS = []
_backends_lock = threading.Lock()

_free_catalog        = {}
_paid_catalog        = {}
_cloud_catalog_lock  = threading.Lock()
_catalog_dirty       = False
_model_metadata       = {}     # model_id -> {name, ctx, reasoning}
_backends_up         = set()
_backends_up_lock    = threading.Lock()
_last_known          = {}    # backend_name -> [model_ids]
_last_known_lock     = threading.Lock()

_checker_pending = threading.Event()
_checker_exit    = threading.Event()

# ---- admission control: protect single-slot local backends from pile-up ----
ADMIT_TIMEOUT = 90       # max seconds a request waits for a backend slot
DEFAULT_SLOTS = 1        # local llama.cpp servers are single-slot by default
QUEUE_FACTOR  = 4        # max in-system (waiting+running) per backend = slots*QUEUE_FACTOR
_backend_sems     = {}   # name -> BoundedSemaphore(slots)
_backend_slots    = {}   # name -> int
_backend_inflight = {}   # name -> waiting+running count
_admit_lock       = threading.Lock()


def _build_sems(slots_map):
    global _backend_sems, _backend_slots
    sems, slots = {}, {}
    for name, n in slots_map.items():
        n = max(1, int(n))
        sems[name] = threading.BoundedSemaphore(n)
        slots[name] = n
    _backend_sems, _backend_slots = sems, slots


def _admit(name):
    """Reserve a slot on backend `name`. True=admitted (caller must _release),
    False=rejected (queue full or waited too long)."""
    sem = _backend_sems.get(name)
    if sem is None:
        return True
    cap = _backend_slots.get(name, DEFAULT_SLOTS) * QUEUE_FACTOR
    with _admit_lock:
        if _backend_inflight.get(name, 0) >= cap:
            return False
        _backend_inflight[name] = _backend_inflight.get(name, 0) + 1
    if sem.acquire(timeout=ADMIT_TIMEOUT):
        return True
    with _admit_lock:
        _backend_inflight[name] = max(0, _backend_inflight.get(name, 0) - 1)
    return False


def _release(name):
    sem = _backend_sems.get(name)
    if sem is None:
        return
    try:
        sem.release()
    except ValueError:
        pass
    with _admit_lock:
        _backend_inflight[name] = max(0, _backend_inflight.get(name, 0) - 1)


def log(msg):
    LOG.write(f"{time.strftime('%H:%M:%S')} {msg}\n")
    LOG.flush()


def parse_param_count(text):
    if not text:
        return None
    m = re.search(r'(?<![\d.])(\d+\.?\d*)\s*[bB](?![a-zA-Z])', text)
    if m:
        return float(m.group(1))
    m = re.search(r'(\d+)\s*[bB]\s*(?:a\d+\s*[bB])?', text, re.IGNORECASE)
    if m:
        return float(m.group(1))
    return None


def _save_catalog():
    with _cloud_catalog_lock:
        meta_count = len(_model_metadata)
        data = {
            "last_update": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "free": dict(_free_catalog),
            "paid": dict(_paid_catalog),
            "metadata": dict(_model_metadata),
        }
    log(f"save catalog: {len(data['free'])} free + {len(data['paid'])} paid + {meta_count} metadata")
    tmp = CLOUD_CATALOG_FILE + ".tmp"
    try:
        with open(tmp, "w") as f:
            json.dump(data, f, indent=2)
        os.rename(tmp, CLOUD_CATALOG_FILE)
    except Exception as e:
        log(f"save catalog FAIL: {type(e).__name__}: {e}")


def _save_last_known():
    with _last_known_lock:
        data = {
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "backends": {k: list(v) for k, v in _last_known.items()},
        }
    tmp = LAST_KNOWN_FILE + ".tmp"
    try:
        with open(tmp, "w") as f:
            json.dump(data, f, indent=2)
        os.rename(tmp, LAST_KNOWN_FILE)
    except Exception as e:
        log(f"save last-known FAIL: {type(e).__name__}: {e}")


def _load_last_known():
    try:
        with open(LAST_KNOWN_FILE) as f:
            data = json.load(f)
        return data.get("backends", {})
    except FileNotFoundError:
        return {}
    except Exception as e:
        log(f"load last-known FAIL ({type(e).__name__}: {e})")
        return {}


def _load_catalog():
    try:
        with open(CLOUD_CATALOG_FILE) as f:
            data = json.load(f)
        free_m = data.get("free") or data.get("models") or {}
        paid_m = data.get("paid") or {}
        meta = data.get("metadata") or {}
        log(f"loaded {len(free_m)} free + {len(paid_m)} paid + {len(meta)} metadata from {CLOUD_CATALOG_FILE}")
        return free_m, paid_m, meta
    except FileNotFoundError:
        log(f"{CLOUD_CATALOG_FILE} not found, starting empty")
        return {}, {}, {}
    except Exception as e:
        log(f"load catalog FAIL ({type(e).__name__}: {e}), starting empty")
        return {}, {}, {}


def reload_backends(signum, frame):
    global LOCAL_BACKENDS
    new = _load_local_backends()
    with _backends_lock:
        LOCAL_BACKENDS = new
    log(f"SIGHUP: reloaded local backends ({len(new)} entries)")


def schedule_recheck(signum, frame):
    with _disc_lock:
        n_local = len(_disc_cache)
        _disc_cache.clear()
    with _cloud_lock:
        had_cloud = _cloud_disc["ids"] is not None
        _cloud_disc["ids"] = None
        _cloud_disc["expiry"] = 0.0
    with _backends_up_lock:
        _backends_up.clear()
    log(f"SIGUSR1: cleared caches (local={n_local}, backends_up=0) + scheduling re-check")
    _checker_pending.set()


def _load_local_backends():
    try:
        with open(LOCAL_BACKENDS_FILE) as f:
            data = json.load(f)
        result = []
        slots_map = {}
        for entry in data:
            result.append((entry["name"], entry["host"], int(entry["port"])))
            slots_map[entry["name"]] = int(entry.get("slots", DEFAULT_SLOTS))
        _build_sems(slots_map)
        log(f"loaded {len(result)} local backends from {LOCAL_BACKENDS_FILE}: "
            f"{[(b[0], slots_map[b[0]]) for b in result]}")
        return result
    except FileNotFoundError:
        _build_sems({n: DEFAULT_SLOTS for n, _, _ in DEFAULT_LOCAL_BACKENDS})
        log(f"{LOCAL_BACKENDS_FILE} not found, using built-in defaults")
        return list(DEFAULT_LOCAL_BACKENDS)
    except Exception as e:
        _build_sems({n: DEFAULT_SLOTS for n, _, _ in DEFAULT_LOCAL_BACKENDS})
        log(f"local-backends config error ({type(e).__name__}: {e}), using defaults")
        return list(DEFAULT_LOCAL_BACKENDS)


def is_cloud_model(model):
    return any(model.startswith(p) for p in CLOUD_PREFIXES)


def strip_tag(mid):
    for t in TAGS:
        if mid.startswith(t):
            return mid[len(t):]
    return mid


def _fetch_backend_models(backend):
    name, host, port = backend
    conn = None
    try:
        conn = http.client.HTTPConnection(host, port, timeout=CONNECT_TIMEOUT)
        conn.request("GET", "/v1/models")
        resp = conn.getresponse()
        if resp.status != 200:
            return None
        data = json.loads(resp.read())
        ids = []
        for m in data.get("data", []):
            mid = m.get("id") or m.get("name") or m.get("model")
            if mid:
                ids.append(mid)
        for m in data.get("models", []):
            mid = m.get("id") or m.get("name") or m.get("model")
            if mid and mid not in ids:
                ids.append(mid)
        return ids
    except Exception:
        return None
    finally:
        try:
            if conn:
                conn.close()
        except Exception:
            pass


def discover(backend):
    name = backend[0]
    now = time.monotonic()
    with _disc_lock:
        cached = _disc_cache.get(name)
        if cached and cached[0] > now:
            return cached[1]
    ids = _fetch_backend_models(backend)
    with _disc_lock:
        _disc_cache[name] = (now + DISCOVERY_TTL, ids)
    if ids is not None:
        with _last_known_lock:
            _last_known[name] = ids
        _save_last_known()
        with _backends_up_lock:
            _backends_up.add(name)
    else:
        with _backends_up_lock:
            _backends_up.discard(name)
        with _last_known_lock:
            ids = _last_known.get(name)
    return ids


def _fetch_openrouter_models():
    if not OR_KEY:
        return None
    conn = None
    try:
        conn = http.client.HTTPSConnection(
            OPENROUTER_HOST, OPENROUTER_PORT,
            timeout=10, context=ssl.create_default_context(),
        )
        conn.request("GET", "/api/v1/models",
                     headers={"Authorization": f"Bearer {OR_KEY}",
                              "Accept": "application/json"})
        resp = conn.getresponse()
        if resp.status != 200:
            log(f"openrouter discovery: HTTP {resp.status}")
            return None
        data = json.loads(resp.read())
        ids = frozenset(m["id"] for m in data.get("data", []) if "id" in m)
        return ids, data.get("data", [])
    except Exception as e:
        log(f"openrouter discovery FAIL: {type(e).__name__}: {e}")
        return None
    finally:
        try:
            if conn:
                conn.close()
        except Exception:
            pass


def discover_cloud():
    now = time.monotonic()
    with _cloud_lock:
        if _cloud_disc["ids"] is not None and _cloud_disc["expiry"] > now:
            return _cloud_disc["ids"]
    result = _fetch_openrouter_models()
    with _cloud_lock:
        if result is not None:
            ids, _ = result
            _cloud_disc["ids"] = ids
            _cloud_disc["expiry"] = now + CLOUD_DISCOVERY_TTL
        return _cloud_disc["ids"]


def find_backend_for_model(model_id):
    backends = list(LOCAL_BACKENDS)
    # 1. Check if it's a direct match in any backend
    for b in backends:
        ids = discover(b)
        if ids and model_id in ids:
            return b
            
    # 2. Check if it's a friendly name
    friendly_to_backend = {v: k for k, v in FRIENDLY_NAMES.items()}
    if model_id in friendly_to_backend:
        target_name = friendly_to_backend[model_id]
        for b in backends:
            if b[0] == target_name:
                ids = discover(b)
                if ids:
                    for mid in ids:
                        if model_id in mid:
                            return b
    return None


def _probe_model(api_key, model_id):
    if not api_key:
        return None
    conn = None
    try:
        body = json.dumps({
            "model": model_id,
            "messages": [{"role": "user", "content": "hi"}],
            "max_tokens": 1,
            "stream": False,
        }).encode()
        conn = http.client.HTTPSConnection(
            OPENROUTER_HOST, OPENROUTER_PORT,
            timeout=PROBE_TIMEOUT,
            context=ssl.create_default_context(),
        )
        conn.request("POST", "/api/v1/chat/completions",
                     body=body,
                     headers={
                         "Authorization": f"Bearer {api_key}",
                         "Content-Type": "application/json",
                         "Accept": "application/json",
                     })
        resp = conn.getresponse()
        resp.read()
        if resp.status == 429:
            return False
        if resp.status == 200:
            return True
        return None
    except Exception:
        return None
    finally:
        try:
            if conn:
                conn.close()
        except Exception:
            pass


def _run_catalog_check():
    result = _fetch_openrouter_models()
    if result is None:
        log("check: openrouter fetch failed, skipping cycle")
        return
    _, raw_models = result

    # --- Free catalog (auto-discovered :free models) ---
    free_ids = {}
    paid_info = {}
    new_metadata = {}
    for m in raw_models:
        mid = m["id"]
        name = m.get("name", "")
        ctx = m.get("context_length", 0)
        pricing = m.get("pricing", {})
        supported_params = m.get("supported_parameters", [])
        has_reasoning = "reasoning" in supported_params
        meta_entry = {"name": name, "ctx": ctx, "reasoning": has_reasoning}
        new_metadata[mid] = meta_entry
        if mid.endswith(":free"):
            free_ids[mid] = {"name": name, "ctx": ctx}
        elif any(mid.startswith(p) for p in CLOUD_PREFIXES):
            prompt_price = float(pricing.get("prompt", 999))
            if prompt_price <= PAID_DISCOVER_THRESHOLD:
                paid_info[mid] = {"name": name, "ctx": ctx, "prompt_price": prompt_price}

    current_free = dict(_free_catalog)
    new_free = {}
    changes = []
    for mid, info in free_ids.items():
        param = parse_param_count(mid) or parse_param_count(info["name"])
        if param is not None and param <= 7.0:
            continue
        if param is None and info["ctx"] < 32768:
            continue
        status = current_free.get(mid, "ok")
        new_free[mid] = status
        if mid not in current_free:
            changes.append(f"+{mid} (param={param}, ctx={info['ctx']})")

    for mid in current_free:
        if mid not in new_free:
            changes.append(f"-{mid}")

    # --- Paid catalog (auto-discover cheap known-prefix models) ---
    current_paid = dict(_paid_catalog)
    new_paid = dict(current_paid)  # preserve manually-added paid entries
    paid_changes = []
    for mid, info in paid_info.items():
        if mid not in new_paid:
            paid_changes.append(f"+{mid} (${info['prompt_price']}/MTok)")
            new_paid[mid] = "ok"

    if changes or paid_changes:
        log(f"check: free changes: {', '.join(changes) if changes else 'none'}; "
            f"paid changes: {', '.join(paid_changes) if paid_changes else 'none'}")
    else:
        log("check: no catalog changes")

    # Probe free models for rate-limit status
    for mid in list(new_free.keys()):
        if not OR_KEY:
            break
        old_status = new_free[mid]
        available = _probe_model(OR_KEY, mid)
        if available is False:
            new_free[mid] = "na"
            if old_status != "na":
                log(f"check: {mid} -> RATE-LIMITED (na)")
        elif available is True:
            new_free[mid] = "ok"
            if old_status != "ok":
                log(f"check: {mid} -> recovered (ok)")

    with _cloud_catalog_lock:
        _free_catalog.clear()
        _free_catalog.update(new_free)
        _paid_catalog.clear()
        _paid_catalog.update(new_paid)
        _model_metadata.clear()
        _model_metadata.update(new_metadata)
    _save_catalog()
    n_na = sum(1 for v in new_free.values() if v == "na")
    log(f"check: done ({len(new_free)} free/{n_na} na, {len(new_paid)} paid)")


def _checker_loop():
    _run_catalog_check()
    while not _checker_exit.is_set():
        triggered = _checker_pending.wait(timeout=CHECK_INTERVAL)
        if _checker_exit.is_set():
            break
        if triggered:
            _checker_pending.clear()
        _run_catalog_check()


def _is_in_any_catalog(model):
    with _cloud_catalog_lock:
        return model in _free_catalog or model in _paid_catalog


def _get_model_status(model):
    with _cloud_catalog_lock:
        st = _free_catalog.get(model)
        if st is not None:
            return st
        return _paid_catalog.get(model)


def _merge_catalogs():
    with _cloud_catalog_lock:
        return {**_free_catalog, **_paid_catalog}


class Handler(http.server.BaseHTTPRequestHandler):
    def _read_body(self):
        length = int(self.headers.get("Content-Length", 0))
        return self.rfile.read(length) if length else b""

    def _fwd_headers_local(self, body_len):
        h = {}
        for k in ("Content-Type", "Authorization", "Accept", "Accept-Encoding"):
            v = self.headers.get(k)
            if v:
                h[k] = v
        if body_len:
            h["Content-Length"] = str(body_len)
        return h

    def _fwd_headers_cloud(self, body_len):
        h = {
            "Content-Type": self.headers.get("Content-Type", "application/json"),
            "Authorization": f"Bearer {OR_KEY}",
            "Accept": self.headers.get("Accept", "application/json"),
            "Accept-Encoding": "identity",
            "HTTP-Referer": os.environ.get("LLM_PROXY_REFERER", "http://localhost"),
            "X-Title": os.environ.get("LLM_PROXY_TITLE", "llm-proxy"),
        }
        if body_len:
            h["Content-Length"] = str(body_len)
        return h

    def _stream_response(self, resp, backend_name, t0, t_try, hint, method, path, is_streaming=False):
        t_hdr = time.monotonic()
        self.send_response(resp.status)
        content_type = resp.getheader("Content-Type", "").lower()
        # For SSE/streaming responses, NEVER forward Content-Length —
        # it breaks chunked framing and causes clients to hang waiting
        # for exactly N bytes that arrive incrementally.
        if is_streaming or "text/event-stream" in content_type:
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Cache-Control", "no-cache")
            self.send_header("Connection", "close")
            # Don't forward Content-Length or Transfer-Encoding
        else:
            for hname in ("Content-Type", "Content-Encoding", "Content-Length"):
                v = resp.getheader(hname)
                if v:
                    self.send_header(hname, v)
        self.send_header("X-LLM-Backend", backend_name)
        self.end_headers()
        bytes_out = 0
        try:
            while True:
                # read1(): return as soon as ANY data is available (one underlying read),
                # instead of read()'s "block until 64KB or EOF" -- otherwise a small
                # streaming response (a few KB of SSE) only flushes at EOF, so reasoning
                # + content land in one burst at the end instead of streaming through.
                chunk = resp.read1(65536)
                if not chunk:
                    break
                self.wfile.write(chunk)
                self.wfile.flush()
                bytes_out += len(chunk)
        except http.client.IncompleteRead as e:
            if e.partial:
                try:
                    self.wfile.write(e.partial)
                    self.wfile.flush()
                    bytes_out += len(e.partial)
                except (OSError, ValueError):
                    pass
        except (OSError, ValueError):
            pass  # client disconnected, or backend conn closed by the cancel watchdog
        dt_total = time.monotonic() - t0
        dt_ttfb = t_hdr - t_try
        log(f"{method} {path} -> {backend_name} status={resp.status} ttfb={dt_ttfb:.2f}s total={dt_total:.2f}s bytes={bytes_out} stream={is_streaming}{hint}")

    def _send_error(self, status, body):
        try:
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
        except (BrokenPipeError, ConnectionResetError):
            pass

    def _client_watchdog(self, get_conn, gone_evt, stop_evt):
        """Watch the CLIENT socket while a backend request is in flight. If the client
        hangs up (pi-agent stop / tab close), close the backend connection so the backend
        aborts generation and the admission slot is freed — instead of running a single-slot
        backend to completion for a request nobody is waiting on. Covers both the streaming
        read loop and the blocking non-streaming getresponse(). Returns the started thread."""
        cs = self.connection

        def _run():
            while not stop_evt.is_set():
                try:
                    r, _, _ = select.select([cs], [], [], 1.0)
                except (OSError, ValueError):
                    return
                if stop_evt.is_set() or not r:
                    continue
                try:
                    peek = cs.recv(1, socket.MSG_PEEK)
                except OSError:
                    peek = b""
                if peek:
                    continue                    # client sent data (pipelined) — still there
                gone_evt.set()                  # EOF -> client hung up; cut the backend loose
                for _ in range(100):            # conn may not exist yet; wait up to ~10s
                    if stop_evt.is_set():
                        return
                    c = get_conn()
                    if c is not None:
                        try: c.close()
                        except Exception: pass
                        return
                    time.sleep(0.1)
                return

        t = threading.Thread(target=_run, daemon=True)
        t.start()
        return t

    def _try_backend(self, backend, method, path, body, fwd, t0, hint, is_last, is_streaming=False):
        name, host, port = backend
        if not _admit(name):
            log(f"{method} {path} -> {name} BUSY (admission gate, slots={_backend_slots.get(name)}) total={time.monotonic()-t0:.2f}s")
            return "busy"
        t_try = time.monotonic()
        conn = None
        gone = threading.Event()
        stop = threading.Event()
        self._client_watchdog(lambda: conn, gone, stop)   # frees the slot if the client bails
        try:
            conn = http.client.HTTPConnection(host, port, timeout=CONNECT_TIMEOUT if not is_last else READ_TIMEOUT)
            conn.request(method, path, body=body, headers=fwd)
            resp = conn.getresponse()
            self._stream_response(resp, name, t0, t_try, hint, method, path, is_streaming=is_streaming)
            conn.close()
            return "ok"
        except (ConnectionRefusedError, TimeoutError, OSError, http.client.HTTPException, ValueError) as e:
            try:
                if conn:
                    conn.close()
            except Exception:
                pass
            if gone.is_set():   # the watchdog closed conn: client cancelled, not a backend fault
                log(f"{method} {path} -> {name} CLIENT-CANCELLED, backend cut loose, slot freed after {time.monotonic()-t_try:.2f}s")
                return "aborted"
            dt = time.monotonic() - t_try
            log(f"{method} {path} -> {name} FAIL {type(e).__name__} {dt:.2f}s")
            return "down"
        finally:
            stop.set()
            _release(name)

    def _proxy_local(self, method, path, body, hint, t0, model=None, is_streaming=False):
        fwd = self._fwd_headers_local(len(body))
        backends = list(LOCAL_BACKENDS)

        chosen = None
        if model:
            chosen = find_backend_for_model(model)
            if chosen is None:
                available = set()
                for b in backends:
                    for m in (discover(b) or []):
                        available.add(m)
                payload = json.dumps({"error": {
                    "message": f"model '{model}' not available on any local backend",
                    "type": "model_not_found",
                    "available": sorted(available),
                }}).encode()
                self._send_error(404, payload)
                log(f"{method} {path} -> model={model!r} 404 (not in any local backend) total={time.monotonic()-t0:.2f}s")
                return
        if chosen is not None:
            r = self._try_backend(chosen, method, path, body, fwd, t0, hint + " (routed)", is_last=True, is_streaming=is_streaming)
            if r == "ok":
                return
            if r == "aborted":
                return          # client hung up; nothing to send, don't fail over
            if r == "busy":
                self._send_error(503, b'{"error":{"message":"backend busy (admission gate), retry shortly","type":"backend_busy"}}')
                return
            log(f"{method} {path} -> {chosen[0]} routed-FAIL, falling back to legacy chain")

        chain = [b for b in backends if b is not chosen]
        busy_seen = False
        for i, b in enumerate(chain):
            is_last = (i == len(chain) - 1)
            r = self._try_backend(b, method, path, body, fwd, t0, hint, is_last, is_streaming=is_streaming)
            if r == "ok":
                return
            if r == "aborted":
                return          # client hung up; nothing to send, don't fail over
            if r == "busy":
                busy_seen = True
        if busy_seen:
            self._send_error(503, b'{"error":{"message":"all local backends busy (admission gate), retry shortly","type":"backend_busy"}}')
            log(f"{method} {path} -> ALL_BUSY total={time.monotonic()-t0:.2f}s")
        else:
            self._send_error(502, b'{"error":"all local backends down"}')
            log(f"{method} {path} -> ALL_DOWN total={time.monotonic()-t0:.2f}s")

    def _proxy_zen(self, method, path, body, hint, t0, is_streaming=False):
        upstream_path = "/zen/v1/chat/completions"
        try:
            j = json.loads(body)
            raw_model = j.get("model", "")
            if raw_model.startswith("opencode/"):
                j["model"] = raw_model[len("opencode/"):]
            for key in ("max_tokens", "max_completion_tokens"):
                val = j.get(key)
                if val is not None and (not isinstance(val, int) or val < 1):
                    j.pop(key, None)
            # Zen API does not support "developer" role; map to "system"
            for msg in j.get("messages", []):
                if msg.get("role") == "developer":
                    msg["role"] = "system"
            body = json.dumps(j).encode()
        except Exception:
            pass
        fwd = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Accept-Encoding": "identity",
        }
        t_try = time.monotonic()
        conn = None
        try:
            conn = http.client.HTTPSConnection(
                ZEN_HOST, ZEN_PORT,
                timeout=READ_TIMEOUT,
                context=ssl.create_default_context(),
            )
            conn.request(method, upstream_path, body=body, headers=fwd)
            resp = conn.getresponse()
            if resp.status >= 400:
                resp_body = resp.read()
                sent = body[:500].decode("utf-8", errors="replace")
                log(f"ZEN 400: status={resp.status} body={resp_body[:300]} sent={sent}")
                self._send_error(502, resp_body)
                log(f"{method} {path} -> opencode-zen status={resp.status} ttfb={time.monotonic()-t_try:.2f}s")
                conn.close()
                return
            self._stream_response(resp, "opencode-zen", t0, t_try, hint, method, path, is_streaming=is_streaming)
            conn.close()
        except Exception as e:
            err = json.dumps({"error": f"opencode-zen: {type(e).__name__}: {e}"}).encode()
            self._send_error(502, err)
            log(f"{method} {path} -> opencode-zen FAIL {type(e).__name__}: {e}")
            try:
                if conn:
                    conn.close()
            except Exception:
                pass

    def _proxy_cloud(self, method, path, body, hint, t0, is_streaming=False):
        if not OR_KEY:
            self._send_error(500, b'{"error":"OPENROUTER_API_KEY not set"}')
            log(f"{method} {path} -> openrouter NO_KEY")
            return
        upstream_path = "/api" + path if path.startswith("/v1") else path
        fwd = self._fwd_headers_cloud(len(body))
        t_try = time.monotonic()
        conn = None
        try:
            conn = http.client.HTTPSConnection(
                OPENROUTER_HOST, OPENROUTER_PORT,
                timeout=READ_TIMEOUT,
                context=ssl.create_default_context(),
            )
            conn.request(method, upstream_path, body=body, headers=fwd)
            resp = conn.getresponse()
            self._stream_response(resp, "openrouter", t0, t_try, hint, method, path, is_streaming=is_streaming)
            conn.close()
        except Exception as e:
            err = json.dumps({"error": f"openrouter: {type(e).__name__}: {e}"}).encode()
            self._send_error(502, err)
            log(f"{method} {path} -> openrouter FAIL {type(e).__name__}: {e}")
            try:
                if conn:
                    conn.close()
            except Exception:
                pass

    def _handle_models(self, t0):
        backends = list(LOCAL_BACKENDS)
        local_entries = []
        seen_local = set()
        up_count = 0
        for backend in backends:
            name = backend[0]
            ids = discover(backend)
            if ids is None:
                continue
            is_down = name not in _backends_up
            if is_down:
                ids = list(ids)
            if not is_down:
                up_count += 1
            for mid in ids:
                if mid in seen_local:
                    continue
                if is_down:
                    display_id = mid + DOWN_SUFFIX
                    owned_by = f"local-{name} (temporarily unavailable)"
                else:
                    display_id = mid
                    owned_by = "local-" + name
                    seen_local.add(mid)
                local_entries.append({
                    "id": TAG_LOCAL + display_id,
                    "object": "model",
                    "owned_by": owned_by,
                })

        live_or = discover_cloud()
        catalog_snapshot = _merge_catalogs()

        cloud_models = []
        with _cloud_catalog_lock:
            meta_snapshot = dict(_model_metadata)
        for mid, status in catalog_snapshot.items():
            if live_or is not None and mid not in live_or:
                continue
            display = mid
            if status == "na":
                base, suffix = mid.rsplit(":", 1) if ":" in mid else (mid, "")
                display = f"{base}-na:{suffix}" if suffix else f"{mid}{NA_SUFFIX}"
            if mid.endswith(":free"):
                tag = TAG_FREE
                owned = "openrouter-free"
                display = display[:-5] if display.endswith(":free") else display
            else:
                tag = TAG_PAID
                owned = "openrouter-paid"
            entry = {
                "id": tag + display,
                "object": "model",
                "owned_by": owned,
            }
            meta_info = meta_snapshot.get(mid, {})
            if "reasoning" in meta_info:
                entry["reasoning"] = meta_info["reasoning"]
            cloud_models.append(entry)

        # OpenCode Zen models
        cloud_models.append({
            "id": TAG_FREE + "opencode/big-pickle",
            "object": "model",
            "owned_by": "opencode-zen",
            "reasoning": True,
        })

        n_na = sum(1 for v in catalog_snapshot.values() if v == "na")
        merged = {"object": "list", "data": cloud_models + local_entries}
        body = json.dumps(merged).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
        or_state = "live" if live_or is not None else "cached/down"
        log(f"GET /v1/models -> merged cloud={len(cloud_models)} (na={n_na}, OR={or_state}) local={len(local_entries)} (from {up_count} backends) total={time.monotonic()-t0:.2f}s")

    def _proxy(self, method):
        t0 = time.monotonic()
        path = self.path
        try:
            if method == "GET" and path in ("/health", "/v1/health"):
                resp_body = json.dumps({"status":"ok"}).encode()
                self.send_response(200)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(resp_body)))
                self.end_headers()
                self.wfile.write(resp_body)
                return

            if method == "GET" and path in ("/v1/models", "/models"):
                self._handle_models(t0)
                return

            body = self._read_body()
            hint = ""
            model = ""
            stream = False
            if body and b"messages" in body:
                try:
                    j = json.loads(body)
                    raw_model = j.get("model", "")
                    model = strip_tag(raw_model)
                    model = re.sub(r"\-na(?=:|\$)", "", model)
                    # Restore :free suffix for free-tier models (catalog keys include it)
                    if ":" not in model:
                        with _cloud_catalog_lock:
                            if model + ":free" in _free_catalog:
                                model = model + ":free"
                    if model != raw_model:
                        j["model"] = model
                        body = json.dumps(j).encode()
                    n_msgs = len(j.get("messages", []))
                    max_t = j.get("max_tokens") or j.get("max_completion_tokens") or "-"
                    total_chars = sum(len(str(m.get("content", ""))) for m in j.get("messages", []))
                    stream = j.get("stream", False)
                    hint = f" model={model} msgs={n_msgs} in_chars={total_chars} max_tok={max_t} stream={stream}"
                    # reasoning passthrough: ask reasoning-capable local models to split
                    # their thinking into a separate `reasoning` stream (chat-template kwarg),
                    # so clients can surface it. The byte-relay forwards the field untouched.
                    if _is_reasoning_local(model):
                        ck = j.get("chat_template_kwargs")
                        ck = ck if isinstance(ck, dict) else {}
                        if "enable_thinking" not in ck:
                            ck["enable_thinking"] = True
                            j["chat_template_kwargs"] = ck
                            body = json.dumps(j).encode()
                except Exception:
                    pass

            # /v1/compress: proxy to compress-server backend
            if method in ("POST",) and path in ("/v1/compress", "/compress"):
                compress_backend = None
                for b in LOCAL_BACKENDS:
                    if b[0] == "compress":
                        compress_backend = b
                        break
                if compress_backend:
                    fwd = self._fwd_headers_local(len(body))
                    name, host, port = compress_backend
                    t_try = time.monotonic()
                    conn = None
                    try:
                        conn = http.client.HTTPConnection(host, port, timeout=600)
                        conn.request("POST", "/compress", body=body, headers=fwd)
                        resp = conn.getresponse()
                        data = resp.read()
                        self.send_response(resp.status)
                        for hn in ("Content-Type",):
                            v = resp.getheader(hn)
                            if v:
                                self.send_header(hn, v)
                        self.send_header("Content-Length", str(len(data)))
                        self.end_headers()
                        self.wfile.write(data)
                        conn.close()
                        log(f"POST /v1/compress -> compress-server status={resp.status} total={time.monotonic()-t0:.2f}s")
                        return
                    except Exception as e:
                        log(f"POST /v1/compress -> compress-server FAIL: {e}")
                self._send_error(502, b'{"error":"compress backend unavailable"}')
                return

            if model and model.startswith("opencode/"):
                self._proxy_zen(method, path, body, hint, t0, is_streaming=stream)
                return

            if model and (is_cloud_model(model) or _is_in_any_catalog(model)):
                is_na = _get_model_status(model) == "na"
                if is_na:
                    payload = json.dumps({"error": {
                        "message": f"model '{model}' is rate-limited and currently not available",
                        "type": "rate_limited",
                    }}).encode()
                    self._send_error(429, payload)
                    log(f"{method} {path} -> model={model!r} 429 (na) total={time.monotonic()-t0:.2f}s")
                    return
                self._proxy_cloud(method, path, body, hint, t0, is_streaming=stream)
            else:
                self._proxy_local(method, path, body, hint, t0, model=model or None, is_streaming=stream)
        except BrokenPipeError:
            log(f"{method} {path} client-disconnect total={time.monotonic()-t0:.2f}s")
        except Exception as e:
            log(f"{method} {path} EXC {type(e).__name__}: {e}")
            traceback.print_exc(file=LOG)
            try:
                self._send_error(500, b'{"error":"internal"}')
            except Exception:
                pass

    def do_POST(self): self._proxy("POST")
    def do_GET(self): self._proxy("GET")
    def do_OPTIONS(self):
        self.send_response(200)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "*")
        self.end_headers()

    def log_message(self, fmt, *args): pass


if __name__ == "__main__":
    port = int(sys.argv[1]) if len(sys.argv) > 1 else 8082

    free_cat, paid_cat, meta = _load_catalog()
    _free_catalog.clear()
    _free_catalog.update(free_cat)
    _paid_catalog.clear()
    _paid_catalog.update(paid_cat)
    _model_metadata.clear()
    _model_metadata.update(meta)
    _last_known = _load_last_known()

    reload_backends(None, None)  # populates global LOCAL_BACKENDS
    signal.signal(signal.SIGHUP, reload_backends)
    signal.signal(signal.SIGUSR1, schedule_recheck)

    _checker_pending.clear()
    _checker_exit.clear()
    checker = threading.Thread(target=_checker_loop, daemon=True, name="catalog-checker")
    checker.start()

    server = http.server.ThreadingHTTPServer(("0.0.0.0", port), Handler)
    n_free = len(_free_catalog)
    n_paid = len(_paid_catalog)
    n_na = sum(1 for v in _free_catalog.values() if v == "na")
    log(f"llm-proxy starting on :{port} local={[b[0] for b in LOCAL_BACKENDS]} "
        f"cloud={'enabled' if OR_KEY else 'DISABLED (no key)'} "
        f"catalog={n_free} free (na={n_na}) + {n_paid} paid "
        f"routing=model-aware+first-up-fallback+live-cloud-filter+auto-check")
    server.serve_forever()