#!/usr/bin/env python3
#
# Required Notice: Copyright 2026 Matthew (matthew@coastalguardfencing.com)
# Source-available under the PolyForm Noncommercial License 1.0.0 (see LICENSE).
# Noncommercial use is free; commercial use requires a paid license — contact
# the address above.
#
"""voidllm — API-key-gated local LLM node.

A stdlib-only authenticating reverse proxy in front of the local Ollama
server's OpenAI-compatible API.  Ollama itself has NO authentication; this
node puts a user-owned API key in front of it so other tools (Continue,
scripts, remote peers reached over a tunnel or reverse proxy) can talk to the
local models with a normal `Authorization: Bearer <key>` header.

    ollama (127.0.0.1:11434, no auth)  <--  voidllm (127.0.0.1:8111, Bearer key)

Security model:
  * Keys are minted ONLY by the user via `voidllm keygen` — shown once,
    stored as sha256 hashes in ~/.config/voidllm/keys.json (0600).
  * Fail-closed: with no keys configured every authed route returns 401.
  * Constant-time compare (hmac.compare_digest) against every stored hash.
  * Auth-failure rate limit: per-source budget, then a global budget, keyed
    on the socket peer (never spoofable X-Forwarded-For).
  * Bounded request bodies, socket + backend timeouts, bounded threads.
  * /healthz is the only unauthenticated route.

Endpoints: everything under /v1/ is proxied verbatim to the backend
(chat/completions, completions, models, embeddings, ...), streaming
included (SSE / chunked responses are forwarded incrementally).  Three
UNAUTHENTICATED routes are client-facing: / (docs + quickstarts), /status
(backend health + models + uptime, HTML or JSON), /healthz.

CLI:
    voidllm serve                 # run the node (default 127.0.0.1:8111)
    voidllm keygen [NAME]         # mint a key (prints ONCE, stores hash)
    voidllm onboard CLIENT ...    # mint a scoped key + write a welcome kit
    voidllm clients               # roster: per-client limits + usage
    voidllm keys                  # list key names + fingerprints
    voidllm rm-key NAME           # revoke a key
    voidllm usage [--verify]      # per-key usage from the hash-chained ledger
    voidllm status                # node + backend health

Env overrides: VOIDLLM_HOST, VOIDLLM_PORT, VOIDLLM_BACKEND,
VOIDLLM_CONFIG_DIR, VOIDLLM_MAX_BODY, VOIDLLM_BACKEND_TIMEOUT,
VOIDLLM_PUBLIC_URL (authoritative base URL for the docs page + kits),
VOIDLLM_CONTACT (operator contact shown to clients), VOIDLLM_HIDE_MODELS,
VOIDLLM_HIDE_VERSION (strip version fingerprints from the public routes).
Keys are re-read when keys.json changes — `keygen` needs no restart.
"""

from __future__ import annotations

import argparse
import hashlib
import hmac
import html
import http.client
import json
import os
import re
import secrets
import socket
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from socketserver import ThreadingMixIn
from pathlib import Path
from urllib.parse import urlsplit

__version__ = "0.4.1"

DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8111
DEFAULT_BACKEND = "http://127.0.0.1:11434"
# Ledger + quota counters. Defaults to the XDG data dir; override with
# VOIDLLM_DATA_DIR. Keep this OUT of any cloud-synced / VCS-tracked directory —
# it holds the append-only usage ledger.
DEFAULT_DATA_DIR = os.environ.get("XDG_DATA_HOME") \
    and str(Path(os.environ["XDG_DATA_HOME"]) / "voidllm") \
    or str(Path.home() / ".local" / "share" / "voidllm")
MAX_BODY = int(os.environ.get("VOIDLLM_MAX_BODY", str(10 * 1024 * 1024)))
BACKEND_TIMEOUT = float(os.environ.get("VOIDLLM_BACKEND_TIMEOUT", "600"))
SOCKET_TIMEOUT = 30.0          # per-recv client read timeout
# absolute deadline for reading the request line + headers, independent of
# per-recv resets — this is what actually stops a slowloris trickle from
# holding a worker slot (a 1-byte-per-29s client defeats SOCKET_TIMEOUT alone)
HEADER_TIMEOUT = float(os.environ.get("VOIDLLM_HEADER_TIMEOUT", "15"))
# Body-drip slowloris defense by minimum RATE (not an absolute deadline, which
# would kill a legit large upload on a slow link). After BODY_GRACE seconds a
# client delivering the body below BODY_MIN_RATE B/s is dropped; any steady
# real link (>=1 KiB/s) completes a body of any size.
BODY_GRACE = float(os.environ.get("VOIDLLM_BODY_GRACE", "10"))
BODY_MIN_RATE = float(os.environ.get("VOIDLLM_BODY_MIN_RATE", "1024"))
STREAM_CHUNK = 8192
# concurrency cap: a slow/hostile peer pool can't exhaust worker threads (and
# stays well under the unit's TasksMax=64). Excess connections wait on the
# semaphore, bounded-queue style, rather than spawning unbounded threads.
MAX_CONNECTIONS = int(os.environ.get("VOIDLLM_MAX_CONNECTIONS", "48"))
# Trust a proxy-supplied identity header (Tailscale-User-*) for rate-limit
# keying ONLY when the operator asserts the tailscale-serve proxy is the sole
# loopback client. Off by default — a local process could otherwise forge it.
TRUST_PROXY_IDENTITY = os.environ.get("VOIDLLM_TRUST_PROXY_IDENTITY") == "1"

# --- client-facing onboarding (docs page, /status, welcome kits) ----------
# The canonical public base URL clients use (e.g. your public tunnel/proxy URL). If set it is
# AUTHORITATIVE for the docs page and generated kits — we never have to guess
# it from the (proxy-forwarded, spoofable) Host header. If unset the docs page
# derives it from the request's Host + X-Forwarded-Proto (validated/escaped).
PUBLIC_URL = (os.environ.get("VOIDLLM_PUBLIC_URL") or "").rstrip("/") or None
# Fallback base URL baked into offline-generated kits when neither --url nor
# VOIDLLM_PUBLIC_URL is given. Empty => derive a localhost URL from the port.
DEFAULT_PUBLIC_URL = ""
# How a prospective client reaches the operator to be issued a key (keys are
# owner-provisioned — there is deliberately no public self-service minting).
CONTACT = (os.environ.get("VOIDLLM_CONTACT") or "").strip()
# The public docs page lists available models by default (they aren't secret —
# an authed /v1/models already returns them). Set to 1 to omit the list from the
# unauthenticated docs AND status routes.
HIDE_MODELS = os.environ.get("VOIDLLM_HIDE_MODELS") == "1"
# Version numbers (voidllm's own + the backend Ollama's) are pure fingerprint on
# the unauthenticated routes — no client value, but they let a scanner match
# known CVEs. Set to 1 to strip them from /, /status, /healthz and the Server:
# header. The operator CLI (`voidllm status`) always shows full detail.
HIDE_VERSION = os.environ.get("VOIDLLM_HIDE_VERSION") == "1"
# Coalesced model-list / backend-health cache TTL (seconds): the unauthenticated
# docs + status routes must never make a per-request backend call (that would be
# a DoS amplifier when the node is publicly exposed).
MODELS_CACHE_TTL = float(os.environ.get("VOIDLLM_MODELS_CACHE_TTL", "60"))
# node process start, for the /status uptime figure
START_TIME = time.time()
# a Host header we're willing to echo into HTML: hostname[:port], nothing else
_HOST_RE = re.compile(r"^[A-Za-z0-9.\-]{1,253}(:[0-9]{1,5})?$")

# Auth-failure rate limiting: after FAIL_LIMIT failures inside FAIL_WINDOW
# seconds, that peer (or, past GLOBAL_FAIL_LIMIT, everyone) is locked out
# until the window slides.
FAIL_LIMIT = 10
GLOBAL_FAIL_LIMIT = 50
FAIL_WINDOW = 60.0

# A proxied request forwards these headers to the backend and no others
# (Authorization in particular must never leak through).
_FORWARD_HEADERS = ("Content-Type", "Accept")
# Hop-by-hop headers we never copy from the backend response.
_HOP_HEADERS = {
    "connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
    "te", "trailers", "transfer-encoding", "upgrade", "content-length",
}
# Never copied from the backend response onto the client response: hop-by-hop,
# PLUS Server/Date — voidllm emits its own via send_response(), so a forwarded
# backend Server would duplicate them AND leak the backend's version past
# VOIDLLM_HIDE_VERSION.
_STRIP_FROM_BACKEND = _HOP_HEADERS | {"server", "date"}


def config_dir() -> Path:
    env = os.environ.get("VOIDLLM_CONFIG_DIR")
    if env:
        return Path(env)
    return Path.home() / ".config" / "voidllm"


def keys_path() -> Path:
    return config_dir() / "keys.json"


def data_dir() -> Path:
    return Path(os.environ.get("VOIDLLM_DATA_DIR", DEFAULT_DATA_DIR))


class KeyStore:
    """sha256-hashed API keys, re-read when the file changes on disk."""

    def __init__(self, path: Path):
        self._path = path
        self._lock = threading.Lock()
        self._mtime: float | None = None
        self._entries: dict[str, dict] = {}   # name -> key entry (sha256, expires, ...)
        self._policy: dict = {}               # dual-control policy (hot-reloaded)

    def _load_locked(self) -> None:
        try:
            st = self._path.stat()
        except OSError:
            self._mtime, self._entries, self._policy = None, {}, {}
            return
        if self._mtime == st.st_mtime_ns:
            return
        try:
            raw = json.loads(self._path.read_text(encoding="utf-8"))
            keys = raw.get("keys", {})
            self._entries = {
                str(name): dict(entry)
                for name, entry in keys.items()
                if isinstance(entry, dict) and entry.get("sha256")
            }
            pol = raw.get("dual_control")
            self._policy = dict(pol) if isinstance(pol, dict) else {}
        except (OSError, ValueError, KeyError, TypeError):
            # Corrupt store: fail CLOSED (no keys accepted), don't crash.
            self._entries, self._policy = {}, {}
        self._mtime = st.st_mtime_ns

    def check(self, presented: str) -> tuple[str, dict] | None:
        """Return (name, entry) on match, else None.  Constant-time per hash."""
        digest = hashlib.sha256(presented.encode("utf-8")).hexdigest()
        with self._lock:
            self._load_locked()
            matched = None
            for name, entry in self._entries.items():
                # compare against EVERY stored hash (no early exit) so timing
                # does not reveal which entry, if any, matched
                if hmac.compare_digest(digest, str(entry["sha256"])):
                    matched = (name, dict(entry))
            return matched

    def names(self) -> dict[str, dict]:
        with self._lock:
            self._load_locked()
            return {n: dict(e) for n, e in self._entries.items()}

    def dual_policy(self) -> dict:
        """Current dual-control policy ({} = off). Hot-reloaded with the keys."""
        with self._lock:
            self._load_locked()
            return dict(self._policy)


class RateLimiter:
    """Sliding-window auth-failure limiter, keyed on the socket peer."""

    def __init__(self):
        self._lock = threading.Lock()
        self._fails: dict[str, list[float]] = {}

    def _prune(self, now: float) -> None:
        for key in list(self._fails):
            recent = [t for t in self._fails[key] if now - t < FAIL_WINDOW]
            if recent:
                self._fails[key] = recent
            else:
                del self._fails[key]
        # cap tracked peers so an address-rotating client can't grow memory
        while len(self._fails) > 1024:
            self._fails.pop(next(iter(self._fails)))

    def blocked(self, peer: str) -> bool:
        now = time.monotonic()
        with self._lock:
            self._prune(now)
            if len(self._fails.get(peer, ())) >= FAIL_LIMIT:
                return True
            total = sum(len(v) for v in self._fails.values())
            return total >= GLOBAL_FAIL_LIMIT

    def record_failure(self, peer: str) -> None:
        now = time.monotonic()
        with self._lock:
            self._prune(now)
            self._fails.setdefault(peer, []).append(now)


GENESIS = "0" * 64


class LedgerError(Exception):
    """Raised inside Ledger for corruption that must not be papered over."""


class SlowBodyError(Exception):
    """Request body was delivered too slowly (body-drip slowloris) or ended
    early — the caller must NOT forward a truncated body to the backend."""


def _now_iso() -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


def _extract_model(body: bytes) -> str | None:
    """The model the BACKEND will actually use, so scope decisions match it.

    Ollama is Go: encoding/json binds struct fields case-insensitively and
    takes the LAST matching key on duplicates.  So we resolve the model the
    same way — the last top-level key equal to "model" (any case) with a
    string value — closing a parser-differential model-smuggling bypass
    (e.g. {"model":"allowed","Model":"forbidden"}).  Anything unparseable or
    non-string yields None, which a model-scoped key treats as a denial."""
    try:
        parsed = json.loads(body)
    except (ValueError, TypeError):
        return None
    if not isinstance(parsed, dict):
        return None
    found = None
    for k, v in parsed.items():
        if isinstance(k, str) and k.lower() == "model" and isinstance(v, str):
            found = v            # last wins, matching Go
    return found


def canonical_path(raw: str) -> str | None:
    """Percent-decode + normalize a request path for the scope decision AND
    for forwarding.  Returns the clean absolute path, or None if it contains
    traversal / encoded separators that must be rejected outright.

    voidllm — not the backend — is the trust boundary, so `..`, `//`, and any
    percent-encoded slash/backslash are refused rather than passed through to
    Ollama (whose native /api/* is unauthenticated)."""
    from urllib.parse import unquote
    path = urlsplit(raw).path
    # reject encoded separators before decoding hides them (%2f, %5c, %2e%2e)
    low = path.lower()
    if "%2f" in low or "%5c" in low or "\\" in path:
        return None
    path = unquote(path)
    # any residual '%' means double-encoding (e.g. %252f -> %2f) — a real /v1
    # API path never contains a literal '%', so reject rather than forward it
    if "%" in path or "\\" in path or "//" in path:
        return None
    parts = path.split("/")
    if any(seg in ("..", ".") for seg in parts):
        return None
    return path


def _seconds_to_utc_midnight() -> int:
    now = time.gmtime()
    return max(1, ((23 - now.tm_hour) * 3600
                   + (59 - now.tm_min) * 60
                   + (60 - now.tm_sec)))


def _chain_key() -> bytes:
    """The HMAC key for the ledger chain, kept in the config dir (local root
    disk) — deliberately NOT beside the ledger, which lives under .voidllm/
    and may be backed up or synced elsewhere.  So the cloud/backup copy of the
    ledger cannot be forged by anyone who lacks this local key.  Auto-created 0600."""
    kp = config_dir() / "ledger.key"
    try:
        raw = kp.read_text(encoding="utf-8").strip()
        if raw:
            return bytes.fromhex(raw)
    except (OSError, ValueError):
        pass
    key = secrets.token_bytes(32)
    try:
        config_dir().mkdir(parents=True, exist_ok=True)
        os.chmod(config_dir(), 0o700)
        tmp = kp.with_suffix(".tmp")
        tmp.write_text(key.hex(), encoding="utf-8")
        os.chmod(tmp, 0o600)
        tmp.replace(kp)
    except OSError as exc:
        sys.stderr.write("[voidllm] could not persist ledger key: %s\n" % exc)
    return key


class Ledger:
    """Hash-chained, append-only usage ledger + per-key daily quota counters.

    Every authenticated /v1 request (and every scope/quota denial) appends one
    JSONL record to usage.jsonl:
        {ts, key, method, path, model, status, ms, bytes_out, prev, h}
    with h = HMAC-SHA256(chain_key, prev + canonical-json(record − {prev,h})).
    Because the chain is keyed by a secret held OUTSIDE the ledger dir (see
    _chain_key), a party who can read/modify the ledger file (e.g. its synced
    cloud copy) but not the local key cannot forge or silently rewrite it —
    verify() will catch any edit/insert/deletion.  (A local process that can
    read the key file too can still recompute the chain; the guarantee is
    against holders of the ledger-without-the-key, not against root-on-box.)
    Counters live in counters.json {"date": "...", "counts": {key: n}} and
    reset when the UTC date rolls over.
    """

    # bound attacker-influenced free-text fields so a single record can never
    # blow past the head-recovery tail window or bloat the ledger volume
    FIELD_CAP = 256

    def __init__(self, root: Path):
        self.root = root
        self._lock = threading.Lock()
        self._head: str | None = None       # last record's h, lazy-loaded
        self._counts: dict | None = None    # lazy-loaded quota counters
        self._key: bytes | None = None      # lazy-loaded HMAC chain key
        self.dropped = 0                    # records we failed to persist

    def _hmac_key(self) -> bytes:
        if self._key is None:
            self._key = _chain_key()
        return self._key

    @property
    def ledger_path(self) -> Path:
        return self.root / "usage.jsonl"

    @property
    def counters_path(self) -> Path:
        return self.root / "counters.json"

    # -- chain ------------------------------------------------------------

    def _record_hash(self, prev: str, payload: dict) -> str:
        canon = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        return hmac.new(self._hmac_key(), (prev + canon).encode("utf-8"),
                        hashlib.sha256).hexdigest()

    def _read_last_line(self) -> bytes:
        """The true last non-blank line.  Returns b"" ONLY for a genuinely
        empty (0-byte) file.  Grows the tail window and, if a non-empty file
        has no non-blank line in any window, scans the WHOLE file; if it is
        still all-blank that is corruption — raises LedgerError so the caller
        fails closed rather than minting GENESIS and forking the chain (a
        >window whitespace/blank-line flood was the fork vector)."""
        try:
            size = self.ledger_path.stat().st_size
        except OSError:
            return b""
        if size == 0:
            return b""
        for window in (4096, 65536, 1 << 20, size):
            try:
                with self.ledger_path.open("rb") as fh:
                    if size > window:
                        fh.seek(-window, os.SEEK_END)
                        fh.readline()  # drop the partial first line in the window
                    lines = [ln for ln in fh.read().splitlines() if ln.strip()]
            except OSError:
                return b""
            if lines:
                return lines[-1]
            if size <= window:
                break                  # whole file read, no non-blank line
        raise LedgerError(
            "ledger has %d bytes but no parseable record line — refusing to "
            "append (run `voidllm usage --verify`); %s"
            % (size, self.ledger_path))

    def _load_head_locked(self) -> str:
        if self._head is not None:
            return self._head
        last = self._read_last_line()
        if not last:
            self._head = GENESIS          # genuinely empty ledger: fresh chain
            return self._head
        try:
            head = json.loads(last)["h"]
        except (ValueError, KeyError):
            # A non-empty ledger with an unparseable tail is corruption, not a
            # fresh chain.  Do NOT cache GENESIS (that would fork every later
            # append onto prev=GENESIS); leave self._head None so each call
            # re-reads and keeps failing closed until the tail is repaired.
            raise LedgerError(
                "ledger tail unparseable — refusing to append (run "
                "`voidllm usage --verify`); %s" % self.ledger_path)
        self._head = head                 # only cache a real, parsed head
        return self._head

    def _cap(self, v):
        return v[:self.FIELD_CAP] if isinstance(v, str) else v

    def record(self, **fields) -> None:
        """Append one usage record; never raises into the request path.

        A persistent write failure is NOT silently ignored: it bumps a
        `dropped` counter and logs loudly so a broken audit trail is visible
        (the proxy keeps serving — failing the request closed here would let a
        full disk become a full outage, which on this box is the worse risk)."""
        for k in ("model", "path", "key", "key2"):
            if k in fields:
                fields[k] = self._cap(fields[k])
        try:
            with self._lock:
                prev = self._load_head_locked()
                payload = dict(fields)
                payload["prev"] = prev
                payload["h"] = self._record_hash(prev, fields)
                self.root.mkdir(parents=True, exist_ok=True)
                with self.ledger_path.open("a", encoding="utf-8") as fh:
                    fh.write(json.dumps(payload, sort_keys=True,
                                        separators=(",", ":")) + "\n")
                self._head = payload["h"]
        except (OSError, LedgerError) as exc:
            self.dropped += 1
            sys.stderr.write(
                "[voidllm] AUDIT GAP: ledger append failed (%d dropped): %s\n"
                % (self.dropped, exc))
            # persist the drop count so `voidllm usage` can flag it even after a
            # restart (best-effort — the disk may be the thing that's failing)
            try:
                self.root.mkdir(parents=True, exist_ok=True)
                (self.root / "audit-gap.count").write_text(str(self.dropped))
            except OSError:
                pass

    def verify(self) -> tuple[int, str | None]:
        """Walk the whole chain. Returns (n_records, first_error_or_None)."""
        prev, n = GENESIS, 0
        try:
            with self.ledger_path.open(encoding="utf-8") as fh:
                for i, line in enumerate(fh, 1):
                    if not line.strip():
                        continue
                    try:
                        rec = json.loads(line)
                    except ValueError:
                        return n, "line %d: not valid JSON" % i
                    body = {k: v for k, v in rec.items() if k not in ("prev", "h")}
                    if rec.get("prev") != prev:
                        return n, "line %d: chain break (prev mismatch)" % i
                    if rec.get("h") != self._record_hash(prev, body):
                        return n, "line %d: hash mismatch (record altered)" % i
                    prev = rec["h"]
                    n += 1
        except FileNotFoundError:
            return 0, None
        except OSError as exc:
            return n, str(exc)
        return n, None

    # -- quota --------------------------------------------------------------

    def _load_counts_locked(self) -> dict:
        today = time.strftime("%Y-%m-%d", time.gmtime())
        if self._counts is None:
            try:
                self._counts = json.loads(
                    self.counters_path.read_text(encoding="utf-8"))
            except (OSError, ValueError):
                self._counts = {"date": today, "counts": {}}
        if self._counts.get("date") != today:
            self._counts = {"date": today, "counts": {}}
        return self._counts

    def try_consume(self, key: str, quota: int | None) -> bool:
        """Reserve one request against KEY's daily quota (None = unlimited)."""
        with self._lock:
            counts = self._load_counts_locked()
            used = int(counts["counts"].get(key, 0))
            if quota is not None and used >= quota:
                return False
            counts["counts"][key] = used + 1
            try:
                self.root.mkdir(parents=True, exist_ok=True)
                tmp = self.counters_path.with_suffix(".tmp")
                tmp.write_text(json.dumps(counts), encoding="utf-8")
                tmp.replace(self.counters_path)
            except OSError as exc:
                sys.stderr.write("[voidllm] counter write failed: %s\n" % exc)
            return True


def _fetch_models(backend: tuple[str, int]) -> tuple[list[str], bool]:
    """Model ids from the backend's /v1/models, ([], False) on any failure.
    Broad except on purpose — this feeds an unauthenticated page and must
    never raise into it."""
    host, port = backend
    try:
        conn = http.client.HTTPConnection(host, port, timeout=3)
        try:
            conn.request("GET", "/v1/models")
            data = json.loads(conn.getresponse().read())
        finally:
            conn.close()
        ids = [str(m["id"]) for m in data.get("data", [])
               if isinstance(m, dict) and m.get("id")]
        return ids, True
    except Exception:
        return [], False


def _backend_health(backend: tuple[str, int]) -> dict:
    host, port = backend
    try:
        conn = http.client.HTTPConnection(host, port, timeout=3)
        try:
            conn.request("GET", "/api/version")
            ver = json.loads(conn.getresponse().read())
        finally:
            conn.close()
        return {"ok": True, "version": str(ver.get("version", "?"))}
    except Exception as exc:
        return {"ok": False, "error": exc.__class__.__name__}


class _BackendCache:
    """Coalesced snapshot (models + backend health) for the UNAUTHENTICATED
    docs and /status routes.  At most one backend refresh per TTL, shared
    across every request thread; a failed refresh serves the last good models
    (never a per-request backend call, never fails the page — otherwise a flood
    of `/` GETs on a public endpoint becomes a flood against the backend)."""

    def __init__(self, ttl: float):
        self._lock = threading.Lock()
        self._ttl = ttl
        self._at = 0.0
        self._snap = {"models": [], "models_ok": False,
                      "backend": {"ok": False, "error": "not-checked"}}

    def get(self, backend: tuple[str, int]) -> dict:
        now = time.monotonic()
        with self._lock:
            if self._at and now - self._at < self._ttl:
                return self._copy()
            models, mok = _fetch_models(backend)
            if mok:
                self._snap["models"] = models       # refresh only on success
            self._snap["models_ok"] = mok           # (else keep last-good list)
            self._snap["backend"] = _backend_health(backend)
            self._at = now
            return self._copy()

    def _copy(self) -> dict:
        s = self._snap
        return {"models": list(s["models"]), "models_ok": s["models_ok"],
                "backend": dict(s["backend"])}


MODEL_CACHE = _BackendCache(MODELS_CACHE_TTL)


def _base_url(handler) -> str:
    """The public base URL to show clients.  VOIDLLM_PUBLIC_URL wins outright;
    otherwise derive it from the (validated) Host + X-Forwarded-Proto, falling
    back to https for a non-loopback host.  Host is whitelist-checked so a
    hostile header can't inject markup into the page it's echoed on."""
    if PUBLIC_URL:
        return PUBLIC_URL
    host = handler.headers.get("Host", "") or ""
    if not _HOST_RE.match(host):
        h, p = handler.server.server_address[:2]
        host = "%s:%s" % (h, p)
    proto = handler.headers.get("X-Forwarded-Proto", "")
    if proto not in ("http", "https"):
        bare = host.split(":")[0]
        proto = "http" if bare in ("127.0.0.1", "localhost", "::1") else "https"
    return "%s://%s" % (proto, host)


_DOCS_CSS = """
:root{color-scheme:light dark;--bg:#fbfbfd;--fg:#1a1a1f;--mut:#5b5b66;
--card:#fff;--brd:#e6e6ec;--acc:#4f46e5;--code:#f4f4f8;--codefg:#26262e}
@media(prefers-color-scheme:dark){:root{--bg:#0e0e12;--fg:#e9e9f0;--mut:#9a9aa8;
--card:#17171e;--brd:#26262f;--acc:#8b85ff;--code:#111119;--codefg:#d6d6e2}}
*{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--fg);
font:15px/1.6 -apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,sans-serif}
.wrap{max-width:820px;margin:0 auto;padding:40px 22px 80px}
h1{font-size:26px;margin:0 0 4px}h2{font-size:17px;margin:34px 0 10px}
.badge{display:inline-block;font-size:12px;font-weight:600;color:#fff;
background:#16a34a;border-radius:20px;padding:2px 10px;vertical-align:middle}
.mut{color:var(--mut)}a{color:var(--acc)}code{background:var(--code);
color:var(--codefg);padding:1px 5px;border-radius:4px;font-size:13px}
pre{background:var(--code);color:var(--codefg);border:1px solid var(--brd);
border-radius:8px;padding:14px 16px;overflow-x:auto;font-size:13px;line-height:1.5}
pre code{background:none;padding:0}.card{background:var(--card);
border:1px solid var(--brd);border-radius:10px;padding:2px 18px;margin:14px 0}
table{border-collapse:collapse;width:100%;font-size:14px}
td,th{border-bottom:1px solid var(--brd);padding:7px 8px;text-align:left}
th{color:var(--mut);font-weight:600}.pill{font-family:monospace;font-size:12px}
ul.models{list-style:none;padding:0;margin:8px 0;display:flex;flex-wrap:wrap;gap:8px}
ul.models li{background:var(--code);border:1px solid var(--brd);border-radius:6px;
padding:4px 10px;font-family:monospace;font-size:13px}
"""


def _docs_html(base: str, models: list[str], models_ok: bool,
               contact: str) -> str:
    esc = html.escape
    base_e = esc(base)
    # When the model list is hidden the examples must not leak a real model name
    # either — clients get their scoped model in the welcome kit. Fall back to a
    # neutral placeholder rather than any live model.
    example_model = ("<model>" if HIDE_MODELS
                     else models[0] if models else "qwen2.5-coder:1.5b")

    def block(code: str) -> str:
        code = code.replace("{{BASE}}", base).replace("{{MODEL}}", example_model)
        return "<pre><code>%s</code></pre>" % esc(code)

    curl = block(
        'curl {{BASE}}/v1/chat/completions \\\n'
        '  -H "Authorization: Bearer $VOIDLLM_KEY" \\\n'
        '  -H "Content-Type: application/json" \\\n'
        '  -d \'{"model":"{{MODEL}}",'
        '"messages":[{"role":"user","content":"Hello"}]}\'')
    py = block(
        'from openai import OpenAI\n'
        'client = OpenAI(base_url="{{BASE}}/v1", api_key="$VOIDLLM_KEY")\n'
        'r = client.chat.completions.create(\n'
        '    model="{{MODEL}}",\n'
        '    messages=[{"role": "user", "content": "Hello"}])\n'
        'print(r.choices[0].message.content)')
    js = block(
        'const r = await fetch("{{BASE}}/v1/chat/completions", {\n'
        '  method: "POST",\n'
        '  headers: {\n'
        '    "Authorization": `Bearer ${process.env.VOIDLLM_KEY}`,\n'
        '    "Content-Type": "application/json" },\n'
        '  body: JSON.stringify({ model: "{{MODEL}}",\n'
        '    messages: [{ role: "user", content: "Hello" }] })\n'
        '});\n'
        'console.log((await r.json()).choices[0].message.content);')
    cont = block(
        'models:\n'
        '  - model: {{MODEL}}\n'
        '    provider: openai\n'
        '    apiBase: {{BASE}}/v1\n'
        '    apiKey: <your key>')

    if HIDE_MODELS:
        models_html = ""
    elif models:
        items = "".join("<li>%s</li>" % esc(m) for m in models)
        stale = "" if models_ok else (" <span class='mut'>(cached — backend "
                                      "not reachable right now)</span>")
        models_html = ("<h2>Available models</h2><p class='mut'>Use one of "
                       "these as the <code>model</code> field:%s</p>"
                       "<ul class='models'>%s</ul>" % (stale, items))
    else:
        models_html = ("<h2>Available models</h2><p class='mut'>The model list "
                       "is temporarily unavailable — check "
                       "<a href='/status'>/status</a>.</p>")

    ver_txt = "" if HIDE_VERSION else " (v%s)" % __version__

    if contact:
        getkey = ("<p>Keys are issued by the operator of this node — there is "
                  "no public sign-up. Contact <strong>%s</strong> to be issued "
                  "an API key scoped to your needs.</p>" % esc(contact))
    else:
        getkey = ("<p>Keys are issued by the operator of this node — there is "
                  "no public sign-up. Contact the operator to be issued an API "
                  "key.</p>")

    return (
        "<!doctype html><html lang='en'><head><meta charset='utf-8'>"
        "<meta name='viewport' content='width=device-width,initial-scale=1'>"
        "<title>voidllm — API</title><style>%s</style></head><body><div class='wrap'>"
        "<h1>voidllm <span class='badge'>voidllm is up</span></h1>"
        "<p class='mut'>An API-key-gated, OpenAI-compatible LLM endpoint"
        "%s. Point any OpenAI client at the base URL below with your key as "
        "a bearer token.</p>"
        "<div class='card'><p><strong>Base URL</strong> &nbsp;"
        "<code>%s/v1</code><br><strong>Auth</strong> &nbsp;"
        "<code>Authorization: Bearer &lt;your key&gt;</code> "
        "(<code>x-api-key</code> also accepted)<br>"
        "<strong>Health</strong> &nbsp;<a href='/status'>/status</a> &middot; "
        "<a href='/healthz'>/healthz</a></p></div>"
        "%s"
        "<h2>Quickstart — curl</h2>%s"
        "<h2>Python (openai SDK)</h2>%s"
        "<h2>JavaScript (fetch)</h2>%s"
        "<h2>Continue (VS Code / JetBrains)</h2>%s"
        "<h2>Getting a key</h2>%s"
        "<h2>Limits &amp; error codes</h2>"
        "<table><tr><th>Code</th><th>Meaning</th></tr>"
        "<tr><td class='pill'>401</td><td>Missing / invalid / expired key</td></tr>"
        "<tr><td class='pill'>403</td><td>Key not scoped for this model or path</td></tr>"
        "<tr><td class='pill'>429</td><td>Daily quota reached, or too many failed "
        "auth attempts (see <code>Retry-After</code>)</td></tr>"
        "<tr><td class='pill'>413</td><td>Request body too large</td></tr>"
        "<tr><td class='pill'>408</td><td>Request body arrived too slowly / "
        "incomplete</td></tr>"
        "<tr><td class='pill'>400</td><td>Malformed path or unsupported "
        "Transfer-Encoding</td></tr></table>"
        "<p class='mut' style='margin-top:34px'>Your key is scoped and metered; "
        "keep it secret. Usage is logged. Streaming (SSE) is supported — pass "
        "<code>\"stream\": true</code>.</p>"
        "</div></body></html>"
        % (_DOCS_CSS, ver_txt, base_e, models_html, curl, py, js, cont,
           getkey))


def _status_html(snap: dict, base: str) -> str:
    esc = html.escape
    b = snap["backend"]
    up = int(max(0, time.time() - START_TIME))
    days, rem = divmod(up, 86400)
    hrs, rem = divmod(rem, 3600)
    mins = rem // 60
    uptime = ("%dd %dh %dm" % (days, hrs, mins) if days
              else "%dh %dm" % (hrs, mins) if hrs else "%dm" % mins)
    if b.get("ok"):
        ver = ("" if HIDE_VERSION else
               " <span class='mut'>(ollama %s)</span>"
               % esc(str(b.get("version", "?"))))
        be = "<span style='color:#16a34a'>&#9679; healthy</span>%s" % ver
    else:
        be = "<span style='color:#dc2626'>&#9679; unreachable</span>"
    node_ver = "" if HIDE_VERSION else "v%s &middot; " % esc(__version__)
    if HIDE_MODELS:
        models_section = ""
    elif snap["models"]:
        stale = "" if snap["models_ok"] else " <span class='mut'>(cached)</span>"
        models_section = ("<h2>Models</h2><ul class='models'>%s</ul>%s"
                          % ("".join("<li>%s</li>" % esc(m)
                                     for m in snap["models"]), stale))
    else:
        models_section = "<h2>Models</h2><p class='mut'>none reported</p>"
    return (
        "<!doctype html><html lang='en'><head><meta charset='utf-8'>"
        "<meta name='viewport' content='width=device-width,initial-scale=1'>"
        "<title>voidllm — status</title><style>%s</style></head><body>"
        "<div class='wrap'><h1>voidllm status</h1>"
        "<div class='card'><p><strong>Node</strong> &nbsp;"
        "<span style='color:#16a34a'>&#9679; up</span> "
        "<span class='mut'>%suptime %s</span><br>"
        "<strong>Backend</strong> &nbsp;%s<br>"
        "<strong>Endpoint</strong> &nbsp;<code>%s/v1</code></p></div>"
        "%s"
        "<p class='mut' style='margin-top:30px'>Machine-readable: "
        "<code>GET /status</code> with <code>Accept: application/json</code>. "
        "See <a href='/'>the docs</a> to get started.</p>"
        "</div></body></html>"
        % (_DOCS_CSS, node_ver, esc(uptime), be, esc(base), models_section))


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    # The Server: header is another version fingerprint; drop the number when
    # VOIDLLM_HIDE_VERSION is set. (BaseHTTPRequestHandler appends the Python
    # sys_version separately; we suppress that too via sys_version below.)
    server_version = "voidllm" if HIDE_VERSION else "voidllm/" + __version__
    sys_version = ""

    # injected by serve()
    keystore: KeyStore = None      # type: ignore[assignment]
    limiter: RateLimiter = None    # type: ignore[assignment]
    ledger: Ledger = None          # type: ignore[assignment]
    backend: tuple[str, int] = ("127.0.0.1", 11434)

    # ---- plumbing -------------------------------------------------------

    def setup(self):
        super().setup()
        self.connection.settimeout(SOCKET_TIMEOUT)

    def handle_one_request(self):
        # arm an absolute header-read deadline; the route handler disarms it as
        # soon as headers are parsed, so a long streaming RESPONSE is never
        # killed — only a slow REQUEST (slowloris) trips it.
        self._arm_wd(HEADER_TIMEOUT)
        try:
            super().handle_one_request()
        finally:
            self._disarm_hdr_wd()

    def _arm_wd(self, timeout: float):
        self._disarm_hdr_wd()
        wd = threading.Timer(timeout, self._force_close)
        wd.daemon = True
        wd.start()
        self._hdr_wd = wd

    def _disarm_hdr_wd(self):
        wd = getattr(self, "_hdr_wd", None)
        if wd is not None:
            wd.cancel()
            self._hdr_wd = None

    def _force_close(self):
        try:
            self.connection.shutdown(socket.SHUT_RDWR)
        except OSError:
            pass

    def log_message(self, fmt, *args):  # stderr, journald-friendly
        cmd = getattr(self, "command", None) or "-"
        path = getattr(self, "path", None) or "-"
        sys.stderr.write("[voidllm] %s %s -> %s\n" % (cmd, path, fmt % args))

    def _send_json(self, code: int, obj: dict, extra: dict | None = None) -> None:
        body = json.dumps(obj).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        for k, v in (extra or {}).items():
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(body)

    def _send_html(self, code: int, markup: str) -> None:
        body = markup.encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _docs_page(self) -> None:
        base = _base_url(self)
        snap = MODEL_CACHE.get(self.backend)
        self._send_html(200, _docs_html(base, snap["models"],
                                        snap["models_ok"], CONTACT))

    def _status_page(self) -> None:
        base = _base_url(self)
        snap = MODEL_CACHE.get(self.backend)
        accept = self.headers.get("Accept", "") or ""
        if "text/html" in accept and "application/json" not in accept:
            self._send_html(200, _status_html(snap, base))
            return
        backend = snap["backend"]
        if HIDE_VERSION:
            backend = {k: v for k, v in backend.items() if k != "version"}
        payload = {
            "service": "voidllm",
            "uptime_seconds": int(max(0, time.time() - START_TIME)),
            "backend": backend,
            "base_url": base,
        }
        if not HIDE_VERSION:
            payload["version"] = __version__
        if not HIDE_MODELS:
            payload["models"] = snap["models"]
            payload["models_ok"] = snap["models_ok"]
        self._send_json(200, payload)

    def _drain_body(self) -> None:
        """Read and discard the request body so HTTP/1.1 keep-alive can't be
        poisoned by an unread body (anti request-smuggling).

        Rate-bounded like the proxied read: a slow-drip drain (the pre-auth /
        reject-path body slowloris) is stopped and the connection closed. If
        the body is larger than we drain (>MAX_BODY) or is delivered too
        slowly or ends early, we cannot safely reuse the connection -> close
        it rather than leave unread bytes framing a smuggled follow-up."""
        # the ONE validated length (set by _validate_framing at request start);
        # never re-parse Content-Length here — re-parsing was the ambiguity vector.
        length = getattr(self, "_req_len", 0) or 0
        if length <= 0:
            return
        if length > MAX_BODY:
            self.close_connection = True   # we won't fully drain it; don't reuse
        remaining = min(length, MAX_BODY)
        start = time.monotonic()
        read = 0
        while remaining > 0:
            try:
                # read1() returns after ONE recv (may be < requested) so the
                # rate check below runs per packet — a plain read() would block
                # until the whole chunk arrived, letting a sub-chunk drip park
                # the worker indefinitely.
                chunk = self.rfile.read1(min(remaining, STREAM_CHUNK))
            except (OSError, socket.timeout):
                self.close_connection = True
                return
            if not chunk:
                self.close_connection = True   # early EOF: framing uncertain
                return
            remaining -= len(chunk)
            read += len(chunk)
            elapsed = time.monotonic() - start
            if elapsed > BODY_GRACE and read / elapsed < BODY_MIN_RATE:
                self.close_connection = True   # slow-drip drain: stop + close
                return

    def _read_body_bounded(self, length: int) -> bytes:
        """Read exactly `length` body bytes, enforcing a minimum delivery RATE
        (not an absolute deadline — that would kill a legit large upload on a
        slow link). A drip client falls below BODY_MIN_RATE after the grace
        window and is dropped; a steady real upload of any size completes.
        Raises SlowBodyError on a stall / early EOF so the caller never
        forwards a truncated body to the backend."""
        buf = bytearray()
        start = time.monotonic()
        while len(buf) < length:
            try:
                # read1(): one recv per iteration so the rate check runs per
                # packet (a plain read() blocks until the full slice arrives,
                # so a sub-chunk drip would never reach the check — the bug
                # that made the earlier rate guard inert).
                chunk = self.rfile.read1(min(length - len(buf), STREAM_CHUNK))
            except (OSError, socket.timeout) as exc:
                raise SlowBodyError(str(exc))
            if not chunk:
                raise SlowBodyError("client closed before sending full body")
            buf += chunk
            elapsed = time.monotonic() - start
            if elapsed > BODY_GRACE and len(buf) / elapsed < BODY_MIN_RATE:
                raise SlowBodyError(
                    "body too slow (%d bytes in %.1fs)" % (len(buf), elapsed))
        return bytes(buf)

    def _validate_framing(self):
        """Single source of truth for request-body framing. Returns
        (length, None) when the body length is UNAMBIGUOUS, or (None, error)
        when it is malformed/ambiguous and the request must be rejected with a
        400 + Connection: close BEFORE auth, draining, or forwarding.

        Rejects (RFC 7230 3.3.3): any Transfer-Encoding (we don't decode it);
        and any Content-Length that isn't a single agreed non-negative decimal —
        covering duplicate CONFLICTING headers, comma-joined conflicting values,
        empty / signed / non-decimal / out-of-range tokens, and CL+TE. Repeated
        Content-Length is accepted ONLY when every value is a valid, numerically
        identical integer. This closes the keep-alive desync / request-smuggling
        vector: an ambiguous message can never leave trailing bytes to frame a
        smuggled second request. get_all() is used so a duplicated/empty first
        header can't hide a real one behind .get()."""
        if self.headers.get_all("Transfer-Encoding"):
            return None, "Transfer-Encoding not supported; use Content-Length"
        raw = self.headers.get_all("Content-Length")
        if not raw:
            return 0, None
        seen = set()
        for header_value in raw:
            for token in str(header_value).split(","):
                token = token.strip()
                # strict ASCII decimal: rejects "", "+5", "-1", "0x10", "1.5",
                # whitespace, and unicode digits (isascii() bars non-ASCII)
                if not token or not (token.isascii() and token.isdigit()):
                    return None, "malformed Content-Length"
                value = int(token)
                if value > (1 << 63) - 1:
                    return None, "Content-Length out of range"
                seen.add(value)
        if len(seen) != 1:
            return None, "conflicting Content-Length"
        return seen.pop(), None

    # ---- auth -----------------------------------------------------------

    def _peer(self) -> str:
        # Rate-limit identity for FAILED auth attempts only (valid keys are
        # never gated). Behind `tailscale serve` every client is 127.0.0.1, so
        # per-source keying needs a proxy-supplied identity — but ANY loopback
        # process can also set that header, so we trust it ONLY when the
        # operator opts in (VOIDLLM_TRUST_PROXY_IDENTITY=1), meaning the
        # deployment guarantees the tailscale-serve proxy is the sole loopback
        # client. Default: socket peer only. X-Forwarded-For is never trusted.
        sock = self.client_address[0] if self.client_address else "?"
        if TRUST_PROXY_IDENTITY and sock in ("127.0.0.1", "::1"):
            ident = (self.headers.get("Tailscale-User-Login")
                     or self.headers.get("Tailscale-User-Name"))
            if ident:
                return "ts:" + ident.strip()[:128]
        return sock

    def _authenticate(self) -> tuple[str, dict] | None:
        """Return (key_name, entry) or None (response already sent).

        A VALID, unexpired key is always honored — the failure limiter gates
        only FAILED attempts, so a flood of bad tokens can never 429 a
        legitimate key holder (the pre-fix behavior was a trivial global DoS
        behind a loopback reverse proxy).
        """
        header = self.headers.get("Authorization") or ""
        token = ""
        if header.lower().startswith("bearer "):
            token = header[7:].strip()
        if not token:
            # also accept x-api-key for anthropic-style clients
            token = (self.headers.get("x-api-key") or "").strip()
        match = self.keystore.check(token) if token else None
        if match is not None:
            name, entry = match
            if not self._key_expired(entry):
                return name, entry     # valid key: served regardless of limiter
            reason = "API key expired %s" % entry.get("expires")
        else:
            reason = "missing or invalid API key"
        # failed auth: record it, then decide 401 vs 429 for THIS caller only
        peer = self._peer()
        self.limiter.record_failure(peer)
        self._drain_body()
        if self.limiter.blocked(peer):
            self._send_json(429, {"error": "too many failed auth attempts"},
                            {"Retry-After": str(int(FAIL_WINDOW)),
                             "WWW-Authenticate": "Bearer"})
        else:
            self._send_json(401, {"error": reason},
                            {"WWW-Authenticate": "Bearer"})
        return None

    @staticmethod
    def _key_expired(entry: dict) -> bool:
        """True if the key entry has an expiry date that is now in the past.
        Single source of truth — used for BOTH the primary and second key."""
        expires = entry.get("expires")
        return bool(expires and time.strftime("%Y-%m-%d", time.gmtime())
                    > str(expires))

    def _effective_model(self, path: str, model: str | None):
        """Resolve the model a request is really touching, and whether the
        endpoint carries a model at all. POST bodies name it; GET /v1/models/{id}
        names it in the path; GET /v1/models (list) is model-less."""
        eff_model = model
        model_bearing = self.command == "POST"
        if self.command == "GET" and path.lower().startswith("/v1/models/"):
            eff_model = path[len("/v1/models/"):].strip("/")
            model_bearing = bool(eff_model)
        return eff_model, model_bearing

    @staticmethod
    def _in_scope(entry: dict, path: str, eff_model: str | None,
                  model_bearing: bool):
        """Pure scope predicate (no quota, no I/O). Returns (ok, failed_facet)
        where failed_facet is 'endpoint', 'model', or None. Used for BOTH keys
        so the primary and the co-signer are judged by identical rules."""
        allowed_paths = entry.get("paths")
        if allowed_paths:
            sub = path[len("/v1/"):] if path.startswith("/v1/") else path
            if not any(sub == p or sub.startswith(p.rstrip("/") + "/")
                       for p in allowed_paths):
                return False, "endpoint"
        allowed_models = entry.get("models")
        if allowed_models and model_bearing and eff_model not in allowed_models:
            return False, "model"
        return True, None

    def _enforce_scope(self, entry: dict, path: str, model: str | None,
                       key: str) -> bool:
        """Scope + quota gate for the PRIMARY key. Sends its own denial + ledger
        record. Scope is delegated to _in_scope so it can't drift from the
        second-key check; quota is consumed only here, only for the primary."""
        eff_model, model_bearing = self._effective_model(path, model)
        ok, facet = self._in_scope(entry, path, eff_model, model_bearing)
        if not ok:
            if facet == "endpoint":
                self._deny(403, "key scope does not allow this endpoint",
                           key, path, model)
            else:
                self._deny(403, "key scope does not allow this model",
                           key, path, eff_model)
            return False
        quota = entry.get("quota_rpd")
        if not self.ledger.try_consume(key, int(quota) if quota else None):
            self._deny(429, "daily request quota exhausted", key, path, model,
                       {"Retry-After": str(_seconds_to_utc_midnight())})
            return False
        return True

    def _requires_dual(self, path: str, eff_model: str | None,
                       model_bearing: bool) -> bool:
        """Does the store-level dual-control policy require a second key here?"""
        pol = self.keystore.dual_policy()
        if not pol:
            return False
        if pol.get("all"):
            return True
        if model_bearing and eff_model and eff_model in (pol.get("models") or []):
            return True
        paths = pol.get("paths") or []
        if paths:
            sub = path[len("/v1/"):] if path.startswith("/v1/") else path
            if any(sub == p or sub.startswith(p.rstrip("/") + "/") for p in paths):
                return True
        return False

    def _require_second(self, primary_name: str, primary_entry: dict,
                        path: str, model: str | None):
        """Two-man rule. Returns the co-signer's key NAME (proceed), None (not
        required), or False (denied — response already sent). A request that
        reaches here already has a valid PRIMARY, so a bad co-signer is a policy
        denial (401/403), never fed to the failure limiter — a valid primary is
        never turned into a 429 (the documented no-DoS invariant).

        Enforces TWO DISTINCT CREDENTIALS: the co-signer must be a valid,
        unexpired, in-scope key whose NAME and whose underlying KEY MATERIAL both
        differ from the primary's. (It proves two distinct keys, not two distinct
        people — that depends on how the keys are issued and held.)"""
        eff_model, model_bearing = self._effective_model(path, model)
        if not self._requires_dual(path, eff_model, model_bearing):
            return None
        # A duplicated X-Void-Key-2 is ambiguous — fail closed rather than pick one.
        seconds = self.headers.get_all("X-Void-Key-2") or []
        if len(seconds) > 1:
            self._deny(401, "dual-control: multiple X-Void-Key-2 headers are "
                       "ambiguous", primary_name, path, model)
            return False
        token = (seconds[0].strip() if seconds else "")
        if not token:
            self._deny(401, "dual-control: this request requires a second key "
                       "in the X-Void-Key-2 header", primary_name, path, model)
            return False
        match = self.keystore.check(token)
        if match is None:
            self._deny(401, "dual-control: invalid second key",
                       primary_name, path, model)
            return False
        second_name, second_entry = match
        if second_name == primary_name:
            self._deny(401, "dual-control: the second key must be different "
                       "from the first", primary_name, path, model)
            return False
        # Distinct key MATERIAL, not just a distinct name: the same secret twice,
        # or two names sharing one stored hash, does NOT satisfy the two-man rule.
        if str(second_entry.get("sha256")) == str(primary_entry.get("sha256")):
            self._deny(401, "dual-control: the second key must be a different "
                       "key, not the same secret under another name",
                       primary_name, path, model)
            return False
        if self._key_expired(second_entry):
            self._deny(401, "dual-control: second key expired",
                       primary_name, path, model, key2=second_name)
            return False
        ok, facet = self._in_scope(second_entry, path, eff_model, model_bearing)
        if not ok:
            self._deny(403, "dual-control: second key scope does not allow "
                       "this %s" % ("endpoint" if facet == "endpoint" else "model"),
                       primary_name, path, model, key2=second_name)
            return False
        return second_name

    def _deny(self, code: int, msg: str, key: str, path: str,
              model: str | None, extra: dict | None = None,
              key2: str | None = None) -> None:
        # callers have already consumed the request body — do not drain here.
        # Receipt BEFORE reply: the denial is on the ledger by the time the
        # client sees it (and tests can read-after-response deterministically).
        fields = dict(ts=_now_iso(), key=key, method=self.command,
                      path=path, model=model, status=code, ms=0, bytes_out=0)
        if key2:
            fields["key2"] = key2
        self.ledger.record(**fields)
        self._send_json(code, {"error": msg}, extra)

    # ---- routes ---------------------------------------------------------

    def do_GET(self):
        self._disarm_hdr_wd()          # headers are in; don't kill a slow body/stream
        length, ferr = self._validate_framing()
        if ferr is not None:
            self.close_connection = True
            self._send_json(400, {"error": ferr})
            return
        self._req_len = length
        # GET endpoints take no body; a Content-Length body left unread would
        # frame a smuggled follow-up on keep-alive — drain it (rate-bounded,
        # closes on anomaly). No-op for a normal bodyless GET.
        self._drain_body()
        path = urlsplit(self.path).path
        if path == "/healthz":
            hz = {"ok": True, "service": "voidllm"}
            if not HIDE_VERSION:
                hz["version"] = __version__
            self._send_json(200, hz)
            return
        if path == "/":
            # Unauthenticated client-facing docs page (base URL, quickstarts,
            # models, limits). Reveals only what an authed /v1/models already
            # would; all interpolated values are escaped.
            self._docs_page()
            return
        if path == "/status":
            # Unauthenticated status a client can self-check before pinging the
            # operator: node up, backend health, models, uptime. JSON or HTML
            # by Accept. Backend data comes from the coalesced cache, never a
            # per-request backend call.
            self._status_page()
            return
        if path == "/favicon.ico":
            # browsers fetch this on every visit; answer without burning an
            # auth-failure against the visitor's rate-limit budget
            self._send_json(404, {"error": "no favicon"})
            return
        auth = self._authenticate()
        if not auth:
            return
        key, entry = auth
        cpath = canonical_path(self.path)
        if cpath is None:
            self._send_json(400, {"error": "malformed path"})
            return
        if cpath.startswith("/v1/"):
            # log the path-named model on GET /v1/models/{id} for audit fidelity
            gmodel = (cpath[len("/v1/models/"):].strip("/")
                      if cpath.startswith("/v1/models/") else None)
            second = self._require_second(key, entry, cpath, None)
            if second is False:      # dual-control denial already sent
                return
            if self._enforce_scope(entry, cpath, None, key):
                self._proxy(cpath, body=b"", key=key, model=gmodel or None,
                            key2=second or None)
            return
        self._send_json(404, {"error": "not found"})

    def do_POST(self):
        self._disarm_hdr_wd()          # headers are in; don't kill a slow body/stream
        length, ferr = self._validate_framing()
        if ferr is not None:
            self.close_connection = True
            self._send_json(400, {"error": ferr})
            return
        self._req_len = length          # the ONE source of truth for the body length
        auth = self._authenticate()
        if not auth:
            return
        key, entry = auth
        cpath = canonical_path(self.path)
        if cpath is None:
            self._drain_body()
            self._send_json(400, {"error": "malformed path"})
            return
        path = cpath
        if not path.startswith("/v1/"):
            self._drain_body()
            self._send_json(404, {"error": "not found"})
            return
        if length > MAX_BODY:
            self._drain_body()
            self._send_json(413, {"error": "request body too large"})
            return
        # rate-bounded body read: stops a body-drip slowloris without killing a
        # legit large upload, and never forwards a truncated body downstream
        if length:
            try:
                body = self._read_body_bounded(length)
            except SlowBodyError:
                self.close_connection = True
                self._send_json(408, {"error": "request body too slow or "
                                               "incomplete"})
                return
        else:
            body = b""
        model = _extract_model(body)
        second = self._require_second(key, entry, path, model)
        if second is False:          # dual-control denial already sent
            return
        if self._enforce_scope(entry, path, model, key):
            self._proxy(path, body=body, key=key, model=model,
                        key2=second or None)

    # ---- backend proxy --------------------------------------------------

    def _proxy(self, path: str, body: bytes, key: str,
               model: str | None, key2: str | None = None) -> None:
        # forward the CANONICAL path + query — never the raw self.path, so a
        # traversal sequence can't reach the backend even if it slipped past
        started = time.monotonic()
        status, bytes_out = 502, 0
        query = urlsplit(self.path).query
        fwd = path + ("?" + query if query else "")
        host, port = self.backend
        conn = http.client.HTTPConnection(host, port, timeout=BACKEND_TIMEOUT)
        try:
            try:
                headers = {"Host": "%s:%d" % (host, port)}
                for name in _FORWARD_HEADERS:
                    val = self.headers.get(name)
                    if val:
                        headers[name] = val
                conn.request(self.command, fwd, body=body or None,
                             headers=headers)
                resp = conn.getresponse()
            except (OSError, http.client.HTTPException) as exc:
                conn.close()
                self._send_json(502, {"error": "backend unreachable: %s" % exc})
                return
            try:
                status = resp.status
                self.send_response(resp.status)
                for name, value in resp.getheaders():
                    if name.lower() not in _STRIP_FROM_BACKEND:
                        self.send_header(name, value)
                # Always chunk the relay: works identically for SSE streams and
                # plain JSON, without trusting the backend's framing.
                self.send_header("Transfer-Encoding", "chunked")
                self.end_headers()
                while True:
                    chunk = resp.read(STREAM_CHUNK)
                    if not chunk:
                        break
                    self.wfile.write(b"%x\r\n" % len(chunk))
                    self.wfile.write(chunk)
                    self.wfile.write(b"\r\n")
                    self.wfile.flush()
                    bytes_out += len(chunk)
                self.wfile.write(b"0\r\n\r\n")
            except (OSError, ValueError):
                # client went away mid-stream; nothing sane to send
                self.close_connection = True
            finally:
                conn.close()
        finally:
            fields = dict(
                ts=_now_iso(), key=key, method=self.command,
                path=path, model=model, status=status,
                ms=int((time.monotonic() - started) * 1000),
                bytes_out=bytes_out)
            if key2:
                fields["key2"] = key2   # co-signer, for two-man-rule audit
            self.ledger.record(**fields)


# ---- CLI ---------------------------------------------------------------


def _load_store() -> dict:
    try:
        return json.loads(keys_path().read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {"keys": {}}


def _write_store(store: dict) -> None:
    cdir = config_dir()
    cdir.mkdir(parents=True, exist_ok=True)
    os.chmod(cdir, 0o700)
    tmp = keys_path().with_suffix(".tmp")
    tmp.write_text(json.dumps(store, indent=2) + "\n", encoding="utf-8")
    os.chmod(tmp, 0o600)
    tmp.replace(keys_path())


def _describe_limits(entry: dict) -> str:
    return ", ".join(filter(None, [
        "expires %s" % entry["expires"] if entry.get("expires") else "",
        "models:%s" % ",".join(entry["models"]) if entry.get("models") else "",
        "paths:%s" % ",".join(entry["paths"]) if entry.get("paths") else "",
        "%s req/day" % entry["quota_rpd"] if entry.get("quota_rpd") else "",
    ])) or "unrestricted"


def _mint_key(name, force=False, expires_days=None, models=None, paths=None,
              quota_rpd=None):
    """Create + persist an API key. Returns (token, entry), or (None, None) if
    the name already exists and force is False. Shared by keygen and onboard."""
    store = _load_store()
    keys = store.setdefault("keys", {})
    if name in keys and not force:
        return None, None
    token = "vllm-" + secrets.token_urlsafe(32)
    entry = {
        "sha256": hashlib.sha256(token.encode("utf-8")).hexdigest(),
        "created": time.strftime("%Y-%m-%dT%H:%M:%S"),
    }
    if expires_days:
        entry["expires"] = time.strftime(
            "%Y-%m-%d", time.gmtime(time.time() + expires_days * 86400))
    if models:
        entry["models"] = [m.strip() for m in models.split(",") if m.strip()]
    if paths:
        entry["paths"] = [p.strip().strip("/") for p in paths.split(",")
                          if p.strip()]
    if quota_rpd:
        entry["quota_rpd"] = quota_rpd
    keys[name] = entry
    _write_store(store)
    return token, entry


def cmd_keygen(args) -> int:
    token, entry = _mint_key(args.name, args.force, args.expires_days,
                             args.models, args.paths, args.quota_rpd)
    if token is None:
        print("key %r already exists (use --force to rotate it)" % args.name,
              file=sys.stderr)
        return 1
    print("API key %r minted — shown ONCE, store it now:\n\n  %s\n"
          % (args.name, token))
    print("stored hash: %s  (%s)\nlimits: %s" % (
        entry["sha256"][:16], keys_path(), _describe_limits(entry)))
    return 0


def cmd_keys(_args) -> int:
    store = _load_store()
    keys = store.get("keys", {})
    if not keys:
        print("no keys configured — run `voidllm keygen` (node is fail-closed)")
        return 0
    for name, entry in sorted(keys.items()):
        print("%-20s sha256:%s…  created %s  [%s]" % (
            name, entry.get("sha256", "")[:16], entry.get("created", "?"),
            _describe_limits(entry)))
    pol = store.get("dual_control") or {}
    if pol:
        print("\ndual-control: %s" % _dual_summary(pol))
    return 0


def _dual_summary(pol: dict) -> str:
    if not pol:
        return "off"
    if pol.get("all"):
        return "ON for ALL /v1 requests"
    parts = []
    if pol.get("models"):
        parts.append("models %s" % ",".join(pol["models"]))
    if pol.get("paths"):
        parts.append("paths %s" % ",".join("/v1/" + p for p in pol["paths"]))
    return "ON for " + "; ".join(parts) if parts else "off"


def _print_dual(pol: dict) -> None:
    if not pol:
        print("dual-control: OFF — a single valid key authorizes every request.")
        return
    print("dual-control: ON. Matching requests need a second, DIFFERENT, in-scope "
          "key in the X-Void-Key-2 header.")
    print("  scope: " + _dual_summary(pol))
    print("  note: this enforces two distinct KEYS, not two distinct people — one "
          "holder of both keys satisfies it.")


def cmd_dual(args) -> int:
    store = _load_store()
    if args.off:
        store.pop("dual_control", None)
        _write_store(store)
        _print_dual({})
        return 0
    if args.all or args.models is not None or args.paths is not None:
        pol: dict = {}
        if args.all:
            pol["all"] = True
        if args.models is not None:
            ms = [m.strip() for m in args.models.split(",") if m.strip()]
            if ms:
                pol["models"] = ms
        if args.paths is not None:
            ps = [p.strip().strip("/") for p in args.paths.split(",") if p.strip()]
            if ps:
                pol["paths"] = ps
        if pol:
            store["dual_control"] = pol
        else:
            store.pop("dual_control", None)
        _write_store(store)
    _print_dual(_load_store().get("dual_control") or {})
    return 0


def _kit_markdown(client, token, entry, base, contact) -> str:
    model = (entry.get("models") or ["qwen2.5-coder:1.5b"])[0]
    lines = [
        "# voidllm API — welcome, %s" % client, "",
        "Your API access to this voidllm LLM endpoint is ready.", "",
        "| Setting | Value |", "|---|---|",
        "| Base URL | `%s/v1` |" % base,
        "| API key | `%s` |" % token,
        "| Auth header | `Authorization: Bearer <key>` |",
        "| Limits | %s |" % _describe_limits(entry), "",
        "> Keep this key secret — it is scoped and metered to you, and it "
        "appears in this file only (it cannot be recovered later).", "",
        "## curl", "```bash",
        'curl %s/v1/chat/completions \\' % base,
        '  -H "Authorization: Bearer %s" \\' % token,
        '  -H "Content-Type: application/json" \\',
        '  -d \'{"model":"%s","messages":[{"role":"user","content":"Hello"}]}\''
        % model, "```", "",
        "## Python (openai SDK)", "```python",
        "from openai import OpenAI",
        'client = OpenAI(base_url="%s/v1", api_key="%s")' % (base, token),
        "r = client.chat.completions.create(",
        '    model="%s",' % model,
        '    messages=[{"role": "user", "content": "Hello"}])',
        "print(r.choices[0].message.content)", "```", "",
        "## Continue (VS Code / JetBrains)", "```yaml", "models:",
        "  - model: %s" % model, "    provider: openai",
        "    apiBase: %s/v1" % base, "    apiKey: %s" % token, "```", "",
        "## Notes",
        '- Streaming (SSE) works: pass `"stream": true`.',
        "- Errors: 401 bad/expired key · 403 out-of-scope model or path · "
        "429 quota or rate limit (honor `Retry-After`).",
        "- Live status: %s/status · Docs: %s/" % (base, base),
    ]
    if contact:
        lines.append("- Questions or more access: %s" % contact)
    return "\n".join(lines) + "\n"


def _kit_html(client, token, entry, base, contact) -> str:
    esc = html.escape
    model = (entry.get("models") or ["qwen2.5-coder:1.5b"])[0]
    contact_html = ("<p>Questions or more access: <strong>%s</strong></p>"
                    % esc(contact)) if contact else ""
    curl = ('curl %s/v1/chat/completions \\\n'
            '  -H "Authorization: Bearer %s" \\\n'
            '  -H "Content-Type: application/json" \\\n'
            '  -d \'{"model":"%s","messages":'
            '[{"role":"user","content":"Hello"}]}\'' % (base, token, model))
    py = ('from openai import OpenAI\n'
          'client = OpenAI(base_url="%s/v1", api_key="%s")\n'
          'r = client.chat.completions.create(\n'
          '    model="%s",\n'
          '    messages=[{"role": "user", "content": "Hello"}])\n'
          'print(r.choices[0].message.content)' % (base, token, model))
    return (
        "<!doctype html><html lang='en'><head><meta charset='utf-8'>"
        "<meta name='viewport' content='width=device-width,initial-scale=1'>"
        "<title>voidllm — welcome %s</title><style>%s"
        ".key{background:#fef3c7;color:#78350f;border:1px solid #f59e0b}"
        "@media(prefers-color-scheme:dark){.key{background:#3a2c05;"
        "color:#fde68a;border-color:#a16207}}</style></head><body>"
        "<div class='wrap'><h1>Welcome, %s</h1>"
        "<p class='mut'>Your API access to this voidllm endpoint is ready.</p>"
        "<div class='card key'><p><strong>Your API key</strong><br>"
        "<code>%s</code></p><p style='font-size:13px'>Keep this secret — it is "
        "scoped and metered to you and appears here only.</p></div>"
        "<div class='card'><p><strong>Base URL</strong> <code>%s/v1</code><br>"
        "<strong>Auth</strong> <code>Authorization: Bearer &lt;key&gt;</code>"
        "<br><strong>Limits</strong> %s</p></div>"
        "<h2>curl</h2><pre><code>%s</code></pre>"
        "<h2>Python (openai SDK)</h2><pre><code>%s</code></pre>"
        "<h2>More</h2><p>Full docs &amp; JS/Continue snippets: "
        "<a href='%s/'>%s/</a> · Status: <a href='%s/status'>%s/status</a></p>"
        "%s</div></body></html>"
        % (esc(client), _DOCS_CSS, esc(client), esc(token), esc(base),
           esc(_describe_limits(entry)), esc(curl), esc(py),
           esc(base), esc(base), esc(base), esc(base), contact_html))


def cmd_onboard(args) -> int:
    token, entry = _mint_key(args.name, args.force, args.expires_days,
                             args.models, args.paths, args.quota_rpd)
    if token is None:
        print("client key %r already exists (use --force to rotate)"
              % args.name, file=sys.stderr)
        return 1
    base = (args.url or PUBLIC_URL or DEFAULT_PUBLIC_URL
            or "http://localhost:%d" % DEFAULT_PORT).rstrip("/")
    contact = args.contact or CONTACT
    # A kit bakes in a LIVE secret, so it defaults to the config dir (0700),
    # files 0600 — never a shared or cloud-synced location. Override with --out.
    outdir = (Path(args.out).expanduser() if args.out
              else config_dir() / "onboarding")
    try:
        outdir.mkdir(parents=True, exist_ok=True)
        os.chmod(outdir, 0o700)
    except OSError as exc:
        print("cannot prepare output dir %s: %s" % (outdir, exc),
              file=sys.stderr)
        return 1
    safe = re.sub(r"[^A-Za-z0-9._-]", "_", args.name) or "client"
    md_path = outdir / ("%s-welcome.md" % safe)
    html_path = outdir / ("%s-welcome.html" % safe)
    for path, content in (
            (md_path, _kit_markdown(args.name, token, entry, base, contact)),
            (html_path, _kit_html(args.name, token, entry, base, contact))):
        path.write_text(content, encoding="utf-8")
        os.chmod(path, 0o600)
    print("onboarded client %r — scoped key minted, kit written (0600):"
          % args.name)
    print("  %s\n  %s" % (md_path, html_path))
    print("\nbase URL: %s/v1\nlimits:   %s" % (base, _describe_limits(entry)))
    print("\n⚠ these files contain a LIVE API key. Send them to the client over "
          "a secure channel, then delete them — the key is not recoverable "
          "(rotate with `voidllm onboard %s --force`)." % args.name)
    if args.out:
        print("⚠ you chose a custom output dir — make sure it isn't cloud-synced "
              "or shared, so the live key isn't uploaded or exposed.")
    return 0


def cmd_clients(_args) -> int:
    keys = _load_store().get("keys", {})
    if not keys:
        print("no keys configured (node is fail-closed) — mint one with "
              "`voidllm onboard <client>`")
        return 0
    ledger = Ledger(data_dir())
    totals: dict = {}
    try:
        with ledger.ledger_path.open(encoding="utf-8") as fh:
            for line in fh:
                if not line.strip():
                    continue
                try:
                    rec = json.loads(line)
                except ValueError:
                    continue
                t = totals.setdefault(rec.get("key", "?"),
                                      {"n": 0, "denied": 0})
                t["n"] += 1
                if int(rec.get("status") or 0) in (403, 429):
                    t["denied"] += 1
    except FileNotFoundError:
        pass
    today = time.strftime("%Y-%m-%d", time.gmtime())
    used_today: dict = {}
    try:
        c = json.loads(ledger.counters_path.read_text(encoding="utf-8"))
        if c.get("date") == today:
            used_today = c.get("counts", {})
    except (OSError, ValueError):
        pass
    print("%-18s %-11s %-7s %-7s %-11s %s" % (
        "CLIENT", "TODAY", "TOTAL", "DENIED", "EXPIRES", "SCOPE"))
    for name, entry in sorted(keys.items()):
        t = totals.get(name, {"n": 0, "denied": 0})
        quota = entry.get("quota_rpd")
        used = int(used_today.get(name, 0))
        today_str = "%d/%d" % (used, int(quota)) if quota else str(used)
        exp = str(entry.get("expires") or "-")
        if entry.get("models"):
            scope = "models:" + ",".join(entry["models"])
        elif entry.get("paths"):
            scope = "paths:" + ",".join(entry["paths"])
        else:
            scope = "any"
        flag = ""
        if entry.get("expires") and today > str(entry["expires"]):
            flag = "  ⚠EXPIRED"
        elif quota and used >= int(quota):
            flag = "  ⚠QUOTA"
        print("%-18s %-11s %-7d %-7d %-11s %s%s" % (
            name[:18], today_str, t["n"], t["denied"], exp, scope, flag))
    return 0


def cmd_usage(args) -> int:
    ledger = Ledger(data_dir())
    if args.verify:
        n, err = ledger.verify()
        if err:
            print("✗ chain INVALID after %d good record(s): %s" % (n, err))
            return 1
        print("✓ chain intact: %d record(s) verified" % n)
        return 0
    totals: dict = {}
    try:
        with ledger.ledger_path.open(encoding="utf-8") as fh:
            for line in fh:
                if not line.strip():
                    continue
                try:
                    rec = json.loads(line)
                except ValueError:
                    continue
                if args.key and rec.get("key") != args.key:
                    continue
                t = totals.setdefault(rec.get("key", "?"),
                                      {"n": 0, "bytes": 0, "denied": 0})
                t["n"] += 1
                t["bytes"] += int(rec.get("bytes_out") or 0)
                if int(rec.get("status") or 0) in (403, 429):
                    t["denied"] += 1
    except FileNotFoundError:
        print("no usage recorded yet (%s)" % ledger.ledger_path)
        return 0
    for key, t in sorted(totals.items()):
        print("%-20s %6d req  %10d bytes out  %d denied" % (
            key, t["n"], t["bytes"], t["denied"]))
    if not totals:
        print("no matching records")
    gap = ledger.ledger_path.with_name("audit-gap.count")
    try:
        dropped = int(gap.read_text())
        if dropped:
            print("\n⚠ %d record(s) were DROPPED (audit gaps) — see the "
                  "service log" % dropped)
    except (OSError, ValueError):
        pass
    return 0


def cmd_rm_key(args) -> int:
    store = _load_store()
    if args.name not in store.get("keys", {}):
        print("no such key: %r" % args.name, file=sys.stderr)
        return 1
    del store["keys"][args.name]
    _write_store(store)
    print("revoked %r" % args.name)
    return 0


def _backend_hostport(url: str) -> tuple[str, int]:
    parts = urlsplit(url)
    return parts.hostname or "127.0.0.1", parts.port or 11434


def cmd_status(args) -> int:
    host, port = _backend_hostport(args.backend)
    try:
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", "/api/version")
        ver = json.loads(conn.getresponse().read())
        conn.close()
        print("backend  %s:%s  ollama %s" % (host, port, ver.get("version", "?")))
    except (OSError, ValueError, http.client.HTTPException) as exc:
        print("backend  %s:%s  UNREACHABLE (%s)" % (host, port, exc))
    n = len(_load_store().get("keys", {}))
    print("keys     %d configured (%s)" % (n, keys_path()))
    return 0


class BoundedThreadingHTTPServer(ThreadingHTTPServer):
    """ThreadingHTTPServer with a hard cap on concurrent worker threads, so a
    slowloris / connection-flood peer can't exhaust threads (or breach the
    unit's TasksMax). Over-cap connections block on the semaphore briefly and
    are dropped if they can't be served promptly."""
    _sem = threading.BoundedSemaphore(MAX_CONNECTIONS)

    def process_request(self, request, client_address):
        if not self._sem.acquire(timeout=5):
            try:
                request.close()
            except Exception:
                pass  # over-cap connection is being dropped; close errors are moot
            return
        try:
            super().process_request(request, client_address)
        except BaseException:
            # spawning the worker thread failed (e.g. can't start new thread at
            # the OS limit) — release the permit we took so the cap doesn't
            # leak a slot on every such failure, then re-raise.
            self._sem.release()
            raise

    def process_request_thread(self, request, client_address):
        try:
            super().process_request_thread(request, client_address)
        finally:
            self._sem.release()


def cmd_serve(args) -> int:
    Handler.keystore = KeyStore(keys_path())
    Handler.limiter = RateLimiter()
    Handler.ledger = Ledger(data_dir())
    Handler.backend = _backend_hostport(args.backend)
    srv = BoundedThreadingHTTPServer((args.host, args.port), Handler)
    srv.daemon_threads = True
    n_keys = len(_load_store().get("keys", {}))
    sys.stderr.write(
        "[voidllm] serving %s -> ollama %s:%s on http://%s:%s  (%d key%s%s)\n"
        % (("/v1/*",) + Handler.backend + (args.host, args.port, n_keys,
           "" if n_keys == 1 else "s",
           "" if n_keys else " — FAIL-CLOSED until `voidllm keygen`")))
    try:
        srv.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        srv.server_close()
    return 0


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(prog="voidllm", description=__doc__.split("\n")[0])
    ap.add_argument("--version", action="version", version=__version__)
    sub = ap.add_subparsers(dest="cmd", required=True)

    sp = sub.add_parser("serve", help="run the node")
    sp.add_argument("--host", default=os.environ.get("VOIDLLM_HOST", DEFAULT_HOST))
    sp.add_argument("--port", type=int,
                    default=int(os.environ.get("VOIDLLM_PORT", str(DEFAULT_PORT))))
    sp.add_argument("--backend",
                    default=os.environ.get("VOIDLLM_BACKEND", DEFAULT_BACKEND))
    sp.set_defaults(fn=cmd_serve)

    kp = sub.add_parser("keygen", help="mint an API key (prints once)")
    kp.add_argument("name", nargs="?", default="default")
    kp.add_argument("--force", action="store_true", help="rotate an existing name")
    kp.add_argument("--expires-days", type=int, metavar="N",
                    help="key stops working N days from now")
    kp.add_argument("--models", metavar="A,B",
                    help="restrict key to these model ids")
    kp.add_argument("--paths", metavar="P,Q",
                    help="restrict key to these /v1 subpaths (e.g. chat/completions,models)")
    kp.add_argument("--quota-rpd", type=int, metavar="N",
                    help="daily request quota for this key")
    kp.set_defaults(fn=cmd_keygen)

    op = sub.add_parser("onboard",
                        help="mint a scoped client key + write a welcome kit")
    op.add_argument("name", help="client name (also the key name)")
    op.add_argument("--url", help="public base URL baked into the kit "
                    "(default: $VOIDLLM_PUBLIC_URL, else http://localhost:PORT)")
    op.add_argument("--contact", help="how the client reaches you for support")
    op.add_argument("--out", metavar="DIR",
                    help="kit output dir (default ~/.config/voidllm/onboarding; "
                    "keep it private — kits hold a live key)")
    op.add_argument("--force", action="store_true",
                    help="rotate an existing client's key")
    op.add_argument("--expires-days", type=int, metavar="N")
    op.add_argument("--models", metavar="A,B")
    op.add_argument("--paths", metavar="P,Q")
    op.add_argument("--quota-rpd", type=int, metavar="N")
    op.set_defaults(fn=cmd_onboard)

    up = sub.add_parser("usage", help="per-key usage from the hash-chained ledger")
    up.add_argument("--key", help="only this key")
    up.add_argument("--verify", action="store_true",
                    help="verify the ledger hash chain end-to-end")
    up.set_defaults(fn=cmd_usage)

    lp = sub.add_parser("keys", help="list configured keys")
    lp.set_defaults(fn=cmd_keys)

    dp = sub.add_parser("dual", help="two-man rule: require a second key "
                        "(X-Void-Key-2) for sensitive requests")
    dp.add_argument("--all", action="store_true",
                    help="require a second key on EVERY /v1 request")
    dp.add_argument("--models", metavar="A,B",
                    help="require a second key only for these model ids "
                    "(the model comes from the request body — prefer --paths "
                    "when you need it to be unspoofable)")
    dp.add_argument("--paths", metavar="P,Q",
                    help="require a second key only for these /v1 subpaths "
                    "(e.g. chat/completions) — not client-spoofable")
    dp.add_argument("--off", action="store_true", help="disable dual-control")
    dp.set_defaults(fn=cmd_dual)

    cp = sub.add_parser("clients",
                        help="roster: per-client limits + usage from the ledger")
    cp.set_defaults(fn=cmd_clients)

    rp = sub.add_parser("rm-key", help="revoke a key")
    rp.add_argument("name")
    rp.set_defaults(fn=cmd_rm_key)

    st = sub.add_parser("status", help="node + backend health")
    st.add_argument("--backend",
                    default=os.environ.get("VOIDLLM_BACKEND", DEFAULT_BACKEND))
    st.set_defaults(fn=cmd_status)

    args = ap.parse_args(argv)
    return args.fn(args)


if __name__ == "__main__":
    raise SystemExit(main())