llmproxy/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 sys
import os
import os.path
import time
import json
import threading
import signal
import re
import copy
import traceback

LOCAL_BACKENDS_FILE = "/etc/llm-proxy/local-backends.json"
CLOUD_CATALOG_FILE  = "/etc/llm-proxy/cloud-catalog.json"

DEFAULT_LOCAL_BACKENDS = [
    ("clevo",     "clevo.fritz.box", 8081),
    ("dirac",     "dirac.fritz.box", 8080),
    ("boltzmann", "boltzmann.fritz.box", 8082),
]
FRIENDLY_NAMES = {
    "gemma4": "escher",
    "gemma4-12b-q4km": "gemma4",
    "qwen3.5-9b-npu": "qwen3.5",
    "qwen3-30b-a3b": "qwen3.6",
}


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      = "/etc/llm-proxy/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": "http://hossenfelder.lxd",
            "X-Title": "hossenfelder",
        }
        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:
                chunk = resp.read(65536)
                if not chunk:
                    break
                self.wfile.write(chunk)
                self.wfile.flush()
                bytes_out += len(chunk)
        except (BrokenPipeError, ConnectionResetError):
            pass  # client disconnected, stop forwarding
        except http.client.IncompleteRead as e:
            if e.partial:
                self.wfile.write(e.partial)
                self.wfile.flush()
                bytes_out += len(e.partial)
        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 _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
        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) as e:
            dt = time.monotonic() - t_try
            log(f"{method} {path} -> {name} FAIL {type(e).__name__} {dt:.2f}s")
            try:
                if conn:
                    conn.close()
            except Exception:
                pass
            return "down"
        finally:
            _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 == "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 == "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 on hossenfelder"}')
            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
            # Filter out bogus deepseek-v4-pro alias on ds4-escher
            if name == "ds4-escher":
                ids = [m for m in ids if m != "deepseek-v4-pro"]
            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}"
                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()