diff --git a/CHANGELOG.md b/CHANGELOG.md index e7d56c0..5cf0bff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,48 @@ All notable changes to this project are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.9.0] - 2026-05-28 + +### Added + +- **Fallback model + single front OpenAI gateway ("fleet").** A new + scaffold-based deployment runs **two always-warm vLLM backends behind one + stdlib gateway** that model-gear manages as three containers + (`model-gear-gateway`, `model-gear-vllm-primary`, `model-gear-vllm-fallback`). + The gateway routes each request by its `model` field, defaults an + unknown/missing name to the primary, and fails over to the other backend when + the chosen one refuses the connection or returns a 5xx **before** the response + body (4xx is returned verbatim; no mid-stream retry). SSE streams are relayed + chunk-by-chunk. Default fallback: the MoE `mmangkad/Qwen3.6-35B-A3B-NVFP4`. +- **New gateway package `model_gear/gateway/`** — a pure-stdlib + (`http.server` + `http.client`, no runtime deps) reverse proxy: `_routing.py` + (pure name/alias/default routing + failover ordering), `_config.py` (env → + routing table + server config), `server.py` (the `handle_post` failover seam, + upstream client, and `ThreadingHTTPServer` handler), run as + `python -m model_gear.gateway`. +- **`model init --fleet`** scaffolds the fleet templates + (`docker-compose.yml` + `.env` + `Dockerfile.gateway`) and pins + `MODEL_GEAR_VERSION` to the running release; **`model fleet up | down | + status`** drives the deployment (`up`/`down` dry-run by default, `--apply` to + commit; `status` is read-only and reports all three containers + the gateway + `/health` + `/v1/models`). +- **Docs:** `docs/gateway-fleet.md` (topology, routing/failover, memory, + verbs), `docs/qwen3.6-35b-a3b-nvfp4.md` (the MoE fallback), a README "fleet" + section, and `model explain fleet` / `model explain gateway` entries. + +### Changed + +- `model_gear/runtime/_compose.py` gained a template registry + (`SINGLE_TEMPLATES` / `FLEET_TEMPLATES`), a `templates=` argument on + `scaffold_plan` / `write_scaffold` (single-model stays the default — existing + callers unchanged), a `compose_up_build` helper, and `FLEET_CONTAINERS`. +- The fleet `.env` mirrors `VLLM_MODEL` / `VLLM_SERVED_NAME` / + `VLLM_TOOL_CALL_PARSER` (= the primary) so the read-only single-model verbs + (`status` / `whoami` / `doctor`) stay coherent on a fleet deployment. + `model switch` remains single-model only. + +### Fixed + ## [0.8.1] - 2026-05-27 ### Changed diff --git a/README.md b/README.md index 4e6ec44..366c103 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,36 @@ model whose repo ships custom modeling code. If vLLM rejects the `nvidia/` ModelOpt checkpoint, set `VLLM_MODEL` to the vLLM-native `RedHatAI/Qwen3-32B-NVFP4` and drop `--quantization` from the compose `command`. +## Running two models behind one gateway (fleet) + +`model init --fleet` scaffolds a **three-container** deployment instead of one: +two always-warm vLLM backends (a primary + an MoE fallback) and a single stdlib +**gateway** that fronts them on the host port the acp `vllm-local` provider +already expects. The gateway routes each request by its `model` field, defaults an +unknown/missing name to the primary, and fails over to the other backend if the +chosen one is down — so existing single-model clients keep working unchanged while +a second model becomes addressable by name. + +```bash +model init --fleet --apply # ~/.model-gear/{docker-compose.yml,.env,Dockerfile.gateway} +docker login nvcr.io # NGC API key for the vLLM image +model fleet up --apply # builds the gateway image + starts all three +model fleet status # container states + gateway /health + /v1/models +``` + +```bash +curl -s http://localhost:8000/v1/models # lists BOTH served models +# route explicitly by name; an unknown/missing model falls back to the primary +curl -s http://localhost:8000/v1/chat/completions -d '{"model":"mmangkad/Qwen3.6-35B-A3B-NVFP4","messages":[...]}' +``` + +Both models stay loaded, so set `PRIMARY_GPU_MEM_UTIL` + `FALLBACK_GPU_MEM_UTIL` +in the fleet `.env` to sum well under 1.0 (they share the 128 GB unified memory). +`model switch` is single-model only — change fleet models by editing the fleet +`.env` and re-running `model fleet up --apply`. See `model explain fleet` / +`model explain gateway` for the routing and failover semantics, and +[`docs/gateway-fleet.md`](docs/gateway-fleet.md) for the full topology. + ### Per-model notes Each runtime model has a doc under `docs/` recording how to run it, live test @@ -87,6 +117,9 @@ results, and caveats: - [`docs/qwen3.6-27b-nvfp4.md`](docs/qwen3.6-27b-nvfp4.md) — a **candidate** (`mmangkad/Qwen3.6-27B-NVFP4`), load-tested on DGX Spark; loads under the current vLLM image but is slower on decode, so the 32B stays. +- [`docs/qwen3.6-35b-a3b-nvfp4.md`](docs/qwen3.6-35b-a3b-nvfp4.md) — the **MoE + fallback** (`mmangkad/Qwen3.6-35B-A3B-NVFP4`) the gateway fleet pairs with the + 32B; ~3B active params decode much faster on this box. The numbers in each doc come from `model switch --apply` then `model assess` (correctness) and `model benchmark` (throughput). `model overview --list` diff --git a/docs/gateway-fleet.md b/docs/gateway-fleet.md new file mode 100644 index 0000000..337fcaf --- /dev/null +++ b/docs/gateway-fleet.md @@ -0,0 +1,108 @@ +# Fleet: two models behind one OpenAI gateway + +The **fleet** runs two always-warm vLLM models behind a single stdlib +OpenAI-compatible gateway, managed by model-gear as three Docker containers. It is +an alternative to the single-model deployment — scaffold it with +`model init --fleet` (the single-model `model init` is unchanged and remains the +default). + +## Why + +The single-model deployment serves one model on `:8000` and `model switch` swaps +it (freeing the prior model). The fleet instead keeps **both** models loaded and +puts one OpenAI endpoint in front of them, so: + +- existing clients (the acp `vllm-local` provider, `curl`, …) keep pointing at + `:8000` and keep working — an unknown/missing `model` defaults to the primary; +- a second model is addressable by name in the same `/v1/...` calls; +- if the chosen backend is down, the gateway fails over to the other one. + +On the DGX Spark (GB10, 128 GB unified memory) both ~30B-class NVFP4 models fit at +once; the fleet pairs the dense primary with an **MoE** fallback (`A3B` ≈ 3B active +params) that decodes much faster, so the fast model stays fast. + +## Topology + +```text +client / acp ──:8000──▶ model-gear-gateway (python -m model_gear.gateway) + │ route by `model` → default → failover + ├──▶ model-gear-vllm-primary :8000 (internal) + └──▶ model-gear-vllm-fallback :8000 (internal) +``` + +Three containers, all `restart: unless-stopped`: + +| Container | Role | Host port | +|---|---|---| +| `model-gear-gateway` | stdlib reverse proxy (the single OpenAI front) | `${VLLM_PORT:-8000}` | +| `model-gear-vllm-primary` | primary model (default: `nvidia/Qwen3-32B-NVFP4`) | internal only | +| `model-gear-vllm-fallback` | MoE fallback (default: `mmangkad/Qwen3.6-35B-A3B-NVFP4`) | internal only | + +The backends are reachable only on the compose network +(`http://vllm-primary:8000`, `http://vllm-fallback:8000`); only the gateway is +published to the host. The gateway needs no Docker socket access — compose owns +the lifecycle; the gateway only routes. + +## The gateway + +A pure-stdlib (`http.server` + `http.client`, no third-party deps) reverse proxy: + +- **Name routing** — a request's `model` routes to the backend that serves it, + plus any `GATEWAY_ALIASES`. The forwarded body's `model` is rewritten to the + backend's `--served-model-name` so the backend accepts aliased/default routes. +- **Default model** — a missing or unknown `model` routes to + `GATEWAY_DEFAULT_MODEL` (the primary). +- **Failover** — if the chosen backend refuses the connection or returns a 5xx + **before any response body**, the request is retried against the other backend. + A 4xx is a client error (returned verbatim, no failover). Once a 2xx body starts + streaming there is no retry — the client already has bytes. +- **Streaming** — `"stream": true` (SSE) is relayed chunk-by-chunk with per-chunk + flushing; normal JSON is buffered with `Content-Length`. +- **Endpoints** — `/v1/chat/completions`, `/v1/completions`, `/v1/embeddings` + (proxied), `/v1/models` (lists both backends), `/health` (gateway liveness). + +The gateway image is built from the scaffolded `Dockerfile.gateway` +(`pip install model-gear==${MODEL_GEAR_VERSION}`, as a non-root user); `model init +--fleet` pins `MODEL_GEAR_VERSION` to the running model-gear release. The version +is required (pinning keeps the image reproducible); from-source/dev boxes that run +ahead of a PyPI release point `MODEL_GEAR_VERSION` at a published TestPyPI `.devN` +build. + +## Verbs + +```bash +model init --fleet --apply # scaffold compose + .env + Dockerfile.gateway +model fleet up --apply # docker compose up -d --build, wait for gateway /health +model fleet status # each container's state + gateway /health + /v1/models +model fleet down --apply # docker compose down +``` + +`model fleet up` / `down` are **dry-run by default**; pass `--apply` to commit. +`--compose-dir` overrides the deployment dir (default `$MODEL_GEAR_DIR` or +`~/.model-gear`). `model fleet status` is read-only. + +**`model switch` does not drive the fleet** — it rewrites the single-model +`VLLM_*` keys. Change fleet models by editing the fleet `.env` +(`PRIMARY_MODEL` / `FALLBACK_MODEL` and their `*_SERVED_NAME` / `*_GPU_MEM_UTIL` +/ `*_TOOL_CALL_PARSER` / `*_QUANTIZATION`) and re-running `model fleet up --apply`. + +## Memory (both warm) + +Both models stay resident, so `PRIMARY_GPU_MEM_UTIL` + `FALLBACK_GPU_MEM_UTIL` +must sum well under 1.0 of the 128 GB. The scaffolded defaults are **0.40** + +**0.35** (≈ 96 GB reserved, leaving headroom for the OS and KV growth). These are +estimates for a dense 32B + a 35B-A3B MoE — **validate live** (watch `nvidia-smi` +at `model fleet up`; OOM is the top operational risk) and tune the two values. + +Note the throughput trade-off: decode is memory-bandwidth bound and the bandwidth +(~273 GB/s) is **shared**. The MoE reads only its active experts per token, so it +stays fast; two backends decoding *simultaneously* split the bandwidth. The +gateway routes one request to one backend, so a single client sees full speed. + +## Coherence with the single-model verbs + +The fleet `.env` mirrors `VLLM_MODEL` / `VLLM_SERVED_NAME` / `VLLM_TOOL_CALL_PARSER` +(= the primary's) so the read-only single-model verbs (`model status`, +`model whoami`, `model doctor`'s `env_coherence` check) stay sensible on a fleet +deployment. `culture.yaml` needs no change: its `model: vllm-local/nvidia/Qwen3-32B-NVFP4` +resolves through the gateway on `:8000` as the default. diff --git a/docs/qwen3.6-35b-a3b-nvfp4.md b/docs/qwen3.6-35b-a3b-nvfp4.md new file mode 100644 index 0000000..11ce8a2 --- /dev/null +++ b/docs/qwen3.6-35b-a3b-nvfp4.md @@ -0,0 +1,82 @@ +# Fallback model: `mmangkad/Qwen3.6-35B-A3B-NVFP4` + +The **MoE fallback** the gateway fleet pairs with the dense primary +(`nvidia/Qwen3-32B-NVFP4`). See [`docs/gateway-fleet.md`](gateway-fleet.md) for the +fleet topology; this doc records what the model is and how it is configured in the +fleet. + +Source: . + +> **Status: configured, not yet load-tested on this hardware.** The numbers below +> are *expectations* from the architecture, not measured values. Fill in the +> Benchmark table from a live `model fleet up` → `model assess` / `model benchmark` +> run (and confirm the quantization/parser caveats) before relying on them. + +## What it is + +- An **NVFP4 (Mixture-of-Experts)** checkpoint: ~35B total parameters, **~3B + active per token** (`A3B`). vLLM loads *all* experts into memory; the small + active set only reduces per-token compute. +- Decode is memory-bandwidth bound on the GB10 (~273 GB/s shared). Reading only + ~3B active params per token (≈1.5 GB at 4-bit) gives an **expected decode + ceiling far above the dense 32B** (which reads ~18 GB/token) — the reason it is + the fast fallback. *Confirm live.* + +## How it runs in the fleet + +Configured via the `FALLBACK_*` keys in the fleet `.env` (scaffolded by +`model init --fleet`); served by the `model-gear-vllm-fallback` container: + +```dotenv +FALLBACK_MODEL=mmangkad/Qwen3.6-35B-A3B-NVFP4 +FALLBACK_SERVED_NAME=mmangkad/Qwen3.6-35B-A3B-NVFP4 +FALLBACK_MAX_MODEL_LEN=32768 +FALLBACK_GPU_MEM_UTIL=0.35 # both models warm: keep primary+fallback well under 1.0 +FALLBACK_TOOL_CALL_PARSER=qwen3_coder +FALLBACK_QUANTIZATION=modelopt_fp4 +``` + +Address it through the gateway by name (or set `GATEWAY_ALIASES` for a short +alias): + +```bash +curl -s http://localhost:8000/v1/chat/completions \ + -d '{"model":"mmangkad/Qwen3.6-35B-A3B-NVFP4","messages":[{"role":"user","content":"hi"}]}' +``` + +## Caveats to confirm on first load + +1. **Tool-call format.** Qwen3.6 emits the Qwen3-Coder **XML** function format, so + the backend is served with `--tool-call-parser=qwen3_coder` (not the `hermes` + parser the dense Qwen3-32B uses). `model_gear.runtime._parser.infer_parser` + already maps `qwen3.6` → `qwen3_coder`. Verify a `tool_choice:"auto"` probe + returns a `finish` tool call. +2. **Quantization format.** The fleet defaults `FALLBACK_QUANTIZATION=modelopt_fp4` + (as for the `nvidia/` checkpoints). This community (`mmangkad`) checkpoint may + instead be a compressed-tensors NVFP4 — if vLLM rejects `modelopt_fp4`, drop or + change `FALLBACK_QUANTIZATION`. +3. **`--trust-remote-code`.** The fleet compose omits it (as the single-model + template does). If this checkpoint ships custom modeling code, vLLM will say so + on load; add it back deliberately (it lets repo code run in-container alongside + `HF_TOKEN` and the mounted cache). +4. **Architecture support.** Confirm the engine registers the checkpoint's + architecture, as done for the 27B sibling: + `docker exec model-gear-vllm-fallback python3 -c "from + vllm.model_executor.models.registry import ModelRegistry; + print(ModelRegistry.get_supported_archs())"`. + +## Benchmark — pending + +Fill from a live run (`model fleet up --apply`, then `model assess` / `model +benchmark` against `:8000` with this model's name): + +| Property | Value | +|---|---| +| Health / `max_model_len` | *pending* | +| Correctness (`17×23`, train 14:45→17:10) | *pending* | +| Reasoning trace field | *pending* | +| Tool calling (`tool_choice:auto`, `qwen3_coder`) | *pending* | +| **Decode throughput** | *pending* (expected ≫ the dense 32B's ~9.7 tok/s) | +| Prefill (~2K tokens) | *pending* | +| GPU memory reserved (at `FALLBACK_GPU_MEM_UTIL`) | *pending* | +| Co-resident total (primary + fallback) | *pending* — watch for OOM | diff --git a/model_gear/cli/__init__.py b/model_gear/cli/__init__.py index 33b840c..b4c1c4a 100644 --- a/model_gear/cli/__init__.py +++ b/model_gear/cli/__init__.py @@ -66,6 +66,7 @@ def _build_parser() -> argparse.ArgumentParser: from model_gear.cli._commands import cli as _cli_group from model_gear.cli._commands import doctor as _doctor_cmd from model_gear.cli._commands import explain as _explain_cmd + from model_gear.cli._commands import fleet as _fleet_cmd from model_gear.cli._commands import init as _init_cmd from model_gear.cli._commands import learn as _learn_cmd from model_gear.cli._commands import overview as _overview_cmd @@ -96,6 +97,7 @@ def _build_parser() -> argparse.ArgumentParser: _assess_cmd.register(sub) _benchmark_cmd.register(sub) _init_cmd.register(sub) + _fleet_cmd.register(sub) # Agent-first / introspection verbs (sibling rubric). _whoami_cmd.register(sub) diff --git a/model_gear/cli/_commands/fleet.py b/model_gear/cli/_commands/fleet.py new file mode 100644 index 0000000..651e56e --- /dev/null +++ b/model_gear/cli/_commands/fleet.py @@ -0,0 +1,173 @@ +"""``model fleet up | down | status`` — drive the 3-container gateway deployment. + +The fleet is two always-warm vLLM backends behind one stdlib gateway (scaffolded +by ``model init --fleet``). These verbs are the fleet-lane counterparts of the +single-model ``serve`` / ``stop`` / ``status``: + +- ``model fleet up`` — ``docker compose up -d --build`` (builds the gateway image), + then waits for the gateway ``/health``. Dry-run by default; ``--apply`` commits. +- ``model fleet down`` — ``docker compose down``. Dry-run by default; ``--apply``. +- ``model fleet status`` — read-only: each container's state, the gateway's + ``/health``, and the routed model list (``/v1/models``). + +``model switch`` does NOT drive the fleet (it rewrites the single-model ``VLLM_*`` +keys); change fleet models by editing the fleet ``.env`` and re-running ``up``. +""" + +from __future__ import annotations + +import argparse + +from model_gear import assess +from model_gear.cli import _runtime_ops +from model_gear.cli._output import emit_diagnostic, emit_result +from model_gear.runtime import _compose, _health + +_UNSET = "(unset)" +_JSON_HELP = "Emit structured JSON." +_PORT_HELP = "Gateway host port (default: VLLM_PORT in .env)." + + +def _fleet_models(port: int) -> list[str] | None: + """Best-effort ``/v1/models`` ids via the gateway; ``None`` if unreachable.""" + if not _health.is_healthy(port): + return None + try: + _, payload = assess._get(f"http://localhost:{port}", "/v1/models") + data = payload.get("data") if isinstance(payload, dict) else None + return [m.get("id") for m in data] if isinstance(data, list) else None + except OSError: + return None + + +def cmd_fleet_up(args: argparse.Namespace) -> int: + json_mode = bool(getattr(args, "json", False)) + deploy_dir = _runtime_ops.deployment_dir(args) + env_path = deploy_dir / _compose.ENV_FILE + port = _runtime_ops.resolve_port(args, env_path) + + if not args.apply: + msg = ( + f"DRY RUN — would run: docker compose up -d --build in {deploy_dir}, " + f"then wait for the gateway /health on :{port}.\nRe-run with --apply to execute." + ) + payload = {"dry_run": True, "deployment_dir": str(deploy_dir), "port": port} + emit_result(payload if json_mode else msg, json_mode=json_mode) + else: + emit_diagnostic(f">> building + starting the fleet in {deploy_dir}") + _runtime_ops.compose_check( + _compose.compose_up_build(deploy_dir), "docker compose up -d --build" + ) + # The gateway answers /health within seconds (it doesn't block on backends); + # the vLLM backends load in the background — check them via 'model fleet status'. + _health.wait_health( + port, deadline_seconds=120, interval=5, container=_compose.FLEET_GATEWAY + ) + result = { + "serving": True, + "port": port, + "deployment_dir": str(deploy_dir), + "containers": list(_compose.FLEET_CONTAINERS), + } + text = ( + f">> gateway up on :{port}. Backends load in the background — " + f"check: model fleet status --compose-dir {deploy_dir}" + ) + emit_result(result if json_mode else text, json_mode=json_mode) + return 0 + + +def cmd_fleet_down(args: argparse.Namespace) -> int: + json_mode = bool(getattr(args, "json", False)) + deploy_dir = _runtime_ops.deployment_dir(args) + + if not args.apply: + dry = ( + f"DRY RUN — would run: docker compose down in {deploy_dir}.\n" + "Re-run with --apply to execute." + ) + payload = {"dry_run": True, "deployment_dir": str(deploy_dir)} + emit_result(payload if json_mode else dry, json_mode=json_mode) + else: + emit_diagnostic(f">> stopping the fleet in {deploy_dir}") + _runtime_ops.compose_check(_compose.compose_down(deploy_dir), "docker compose down") + result = {"stopped": True, "deployment_dir": str(deploy_dir)} + emit_result( + result if json_mode else f">> fleet stopped in {deploy_dir}", json_mode=json_mode + ) + return 0 + + +def cmd_fleet_status(args: argparse.Namespace) -> int: + json_mode = bool(getattr(args, "json", False)) + deploy_dir = _runtime_ops.deployment_dir(args) + env_path = deploy_dir / _compose.ENV_FILE + port = _runtime_ops.resolve_port(args, env_path) + + containers = [ + {"name": name, "state": _compose.inspect_state(name)} for name in _compose.FLEET_CONTAINERS + ] + report = { + "deployment_dir": str(deploy_dir), + "port": port, + "gateway_health": "ok" if _health.is_healthy(port) else "not responding", + "containers": containers, + "models": _fleet_models(port), + } + + if json_mode: + emit_result(report, json_mode=True) + else: + lines = [ + f"dir: {report['deployment_dir']}", + f"gateway: {report['gateway_health']} (:{port})", + ] + for c in containers: + lines.append(f" {c['name']} — {c['state']}") + models = report["models"] + lines.append("models: " + (", ".join(models) if models else _UNSET)) + emit_result("\n".join(lines), json_mode=False) + return 0 + + +def _no_verb(args: argparse.Namespace) -> int: + # Bare `model fleet` → the read-only status (safe default). + return cmd_fleet_status(args) + + +def _add_compose_dir(p: argparse.ArgumentParser) -> None: + p.add_argument( + "--compose-dir", help="Deployment dir (default: $MODEL_GEAR_DIR or ~/.model-gear)." + ) + + +def register(sub: argparse._SubParsersAction) -> None: + p = sub.add_parser( + "fleet", + help="Drive the gateway fleet (up / down / status). See 'model fleet status'.", + ) + _add_compose_dir(p) + p.add_argument("--port", type=int, help=_PORT_HELP) + p.add_argument("--json", action="store_true", help=_JSON_HELP) + p.set_defaults(func=_no_verb, json=False) + # Propagate the structured-error parser class to the noun's subparsers. + noun = p.add_subparsers(dest="fleet_command", parser_class=type(p)) + + up = noun.add_parser("up", help="Build + start the fleet (dry-run; --apply).") + _add_compose_dir(up) + up.add_argument("--port", type=int, help=_PORT_HELP) + up.add_argument("--apply", action="store_true", help="Actually build + start the fleet.") + up.add_argument("--json", action="store_true", help=_JSON_HELP) + up.set_defaults(func=cmd_fleet_up) + + down = noun.add_parser("down", help="Stop the fleet (dry-run; --apply).") + _add_compose_dir(down) + down.add_argument("--apply", action="store_true", help="Actually stop the fleet.") + down.add_argument("--json", action="store_true", help=_JSON_HELP) + down.set_defaults(func=cmd_fleet_down) + + st = noun.add_parser("status", help="Read-only: container states, gateway /health, /v1/models.") + _add_compose_dir(st) + st.add_argument("--port", type=int, help=_PORT_HELP) + st.add_argument("--json", action="store_true", help=_JSON_HELP) + st.set_defaults(func=cmd_fleet_status) diff --git a/model_gear/cli/_commands/init.py b/model_gear/cli/_commands/init.py index 293d544..1dfa6af 100644 --- a/model_gear/cli/_commands/init.py +++ b/model_gear/cli/_commands/init.py @@ -2,7 +2,9 @@ Copies the packaged ``docker-compose.yml`` + ``env.example``→``.env`` into ``TARGET`` (default ``~/.model-gear``; ``model init .`` for the local folder). -Mutating: dry-run by default; ``--apply`` writes, ``--force`` overwrites. +``--fleet`` scaffolds the 3-container gateway deployment instead (two always-warm +vLLM backends + a single OpenAI front). Mutating: dry-run by default; ``--apply`` +writes, ``--force`` overwrites. """ from __future__ import annotations @@ -10,23 +12,29 @@ import argparse from pathlib import Path +from model_gear import __version__ from model_gear.cli._output import emit_result -from model_gear.runtime import _compose +from model_gear.runtime import _compose, _env -def _emit_dry_run(target: Path, json_mode: bool) -> None: - plan = _compose.scaffold_plan(target) +def _templates(fleet: bool) -> dict[str, str]: + return _compose.FLEET_TEMPLATES if fleet else _compose.SINGLE_TEMPLATES + + +def _emit_dry_run(target: Path, fleet: bool, json_mode: bool) -> None: + plan = _compose.scaffold_plan(target, _templates(fleet)) if json_mode: emit_result( { "dry_run": True, + "fleet": fleet, "target": str(target), "files": [{"name": name, "exists": exists} for name, exists in plan], }, json_mode=True, ) return - lines = [f"DRY RUN — would scaffold into {target}:"] + lines = [f"DRY RUN — would scaffold {'fleet ' if fleet else ''}into {target}:"] for name, exists in plan: note = " (exists; needs --force to overwrite)" if exists else "" lines.append(f" {name}{note}") @@ -34,26 +42,38 @@ def _emit_dry_run(target: Path, json_mode: bool) -> None: emit_result("\n".join(lines), json_mode=False) -def _emit_apply(target: Path, force: bool, json_mode: bool) -> None: - written = _compose.write_scaffold(target, force=force) +def _emit_apply(target: Path, fleet: bool, force: bool, json_mode: bool) -> None: + written = _compose.write_scaffold(target, force=force, templates=_templates(fleet)) + if fleet: + # Pin the gateway image to the model-gear release that scaffolded this. + _env.set_env(target / _compose.ENV_FILE, "MODEL_GEAR_VERSION", __version__) if json_mode: - emit_result({"scaffolded": str(target), "files": [p.name for p in written]}, json_mode=True) + emit_result( + {"scaffolded": str(target), "fleet": fleet, "files": [p.name for p in written]}, + json_mode=True, + ) return + next_step = ( + "docker login nvcr.io && model fleet up --apply" + if fleet + else "docker login nvcr.io && model serve --apply" + ) emit_result( f">> scaffolded {target}:\n" + "\n".join(f" {p.name}" for p in written) - + "\n>> next: docker login nvcr.io && model serve --apply", + + f"\n>> next: {next_step}", json_mode=False, ) def cmd_init(args: argparse.Namespace) -> int: json_mode = bool(getattr(args, "json", False)) + fleet = bool(getattr(args, "fleet", False)) target = Path(args.target).expanduser() if args.target else _compose.default_deployment_dir() if args.apply: - _emit_apply(target, args.force, json_mode) + _emit_apply(target, fleet, args.force, json_mode) else: - _emit_dry_run(target, json_mode) + _emit_dry_run(target, fleet, json_mode) return 0 @@ -67,6 +87,12 @@ def register(sub: argparse._SubParsersAction) -> None: nargs="?", help="Where to scaffold (default ~/.model-gear; '.' for the current folder).", ) + p.add_argument( + "--fleet", + action="store_true", + help="Scaffold the 3-container gateway deployment (2 vLLM backends + 1 front) " + "instead of a single model.", + ) p.add_argument("--force", action="store_true", help="Overwrite existing files.") p.add_argument("--apply", action="store_true", help="Actually write the files.") p.add_argument("--json", action="store_true", help="Emit structured JSON.") diff --git a/model_gear/cli/_commands/learn.py b/model_gear/cli/_commands/learn.py index d211dfb..48f9f23 100644 --- a/model_gear/cli/_commands/learn.py +++ b/model_gear/cli/_commands/learn.py @@ -30,6 +30,9 @@ model stop Stop the vLLM server. Dry-run; --apply. model switch Switch the served model. Dry-run; --apply recreates the container and waits for /health. + model fleet up|down|status + Drive the 2-model gateway deployment (scaffold it with + 'model init --fleet'). up/down are dry-run; --apply. model status Read-only: current model, container state, /health. model assess Read-only: correctness probes + reasoning-trace field. model benchmark Read-only: decode throughput + prefill latency. @@ -85,6 +88,10 @@ def _as_json_payload() -> dict[str, object]: }, {"path": ["stop"], "summary": "Stop the vLLM server (dry-run; --apply)."}, {"path": ["switch"], "summary": "Switch the served model (dry-run; --apply)."}, + { + "path": ["fleet"], + "summary": "Drive the 2-model gateway deployment (up/down/status; --apply).", + }, {"path": ["status"], "summary": "Current model, container state, /health."}, {"path": ["assess"], "summary": "Correctness probes + reasoning-trace field."}, {"path": ["benchmark"], "summary": "Decode throughput + prefill latency."}, @@ -94,7 +101,7 @@ def _as_json_payload() -> dict[str, object]: {"path": ["doctor"], "summary": "Diagnose docker/compose/.env/health."}, ], "mutation_safety": { - "write_verbs": ["switch", "serve", "stop", "init"], + "write_verbs": ["switch", "serve", "stop", "init", "fleet up", "fleet down"], "rule": "dry-run by default; require --apply to commit", }, "exit_codes": { diff --git a/model_gear/cli/_commands/overview.py b/model_gear/cli/_commands/overview.py index f507420..ea3d1fa 100644 --- a/model_gear/cli/_commands/overview.py +++ b/model_gear/cli/_commands/overview.py @@ -20,9 +20,10 @@ from model_gear.cli._output import emit_result _VERBS = [ - "init [TARGET] — scaffold a deployment dir (dry-run; --apply)", + "init [TARGET] — scaffold a deployment dir (--fleet for the gateway; dry-run; --apply)", "serve / stop — start / stop the vLLM server (dry-run; --apply)", "switch — switch the served model (dry-run; --apply)", + "fleet up / down / status — drive the 2-model gateway deployment (dry-run; --apply)", "status — current model, container state, /health", "assess — correctness probes against the served model", "benchmark — decode throughput + prefill latency", @@ -37,6 +38,7 @@ "assess — correctness probes against the served model", "switch — change the served model (dry-run by default)", "benchmark — decode throughput + prefill latency", + "fleet — front two always-warm models with one OpenAI gateway (routing + failover)", ] diff --git a/model_gear/explain/catalog.py b/model_gear/explain/catalog.py index 0978f0a..51e3b1f 100644 --- a/model_gear/explain/catalog.py +++ b/model_gear/explain/catalog.py @@ -27,6 +27,9 @@ Dry-run by default; `--apply` to commit. - `model switch ` — switch the served model. Dry-run by default; `--apply` recreates the container and waits for `/health`. +- `model fleet up|down|status` — drive the 2-model gateway deployment (one + OpenAI front over two always-warm models). Scaffold it with + `model init --fleet`. `up`/`down` are dry-run by default; `--apply` to commit. - `model status` — read-only: current model, container state, `/health`. - `model assess` — read-only correctness probes + reasoning-trace detection. - `model benchmark` — read-only decode throughput + prefill latency. @@ -50,6 +53,8 @@ ## See also - `model explain switch` +- `model explain fleet` +- `model explain gateway` - `model explain assess` - `model explain backend` - `model explain models` @@ -178,8 +183,64 @@ - `docs/qwen3-32b-nvfp4.md` — `nvidia/Qwen3-32B-NVFP4`, the current runtime model. - `docs/qwen3.6-27b-nvfp4.md` — `mmangkad/Qwen3.6-27B-NVFP4`, a candidate that load-tested slower on decode, so the 32B stays. +- `docs/qwen3.6-35b-a3b-nvfp4.md` — `mmangkad/Qwen3.6-35B-A3B-NVFP4`, the MoE + fallback the gateway fleet pairs with the 32B (~3B active → fast decode). -`model overview --list` lists these and flags which one is currently served. +`model overview --list` lists these and flags which one is currently served. To +run two side-by-side behind one OpenAI endpoint, see `model explain fleet`. +""" + +_FLEET = """\ +# model fleet + +The fleet runs **two always-warm models behind one OpenAI-compatible gateway**, +managed as three containers: `model-gear-vllm-primary`, `model-gear-vllm-fallback`, +and `model-gear-gateway`. Scaffold it with `model init --fleet` (writes the fleet +`docker-compose.yml`, `.env`, and `Dockerfile.gateway`), then: + +- `model fleet up` — `docker compose up -d --build` (builds the gateway image), + then waits for the gateway `/health`. The vLLM backends load in the background. +- `model fleet down` — `docker compose down`. +- `model fleet status` — read-only: each container's state, the gateway `/health`, + and the routed model list (`/v1/models`). + +`up`/`down` are **dry-run by default**; pass `--apply` to commit. `--compose-dir` +overrides the deployment dir. Both backends stay loaded — set their +`PRIMARY_GPU_MEM_UTIL` / `FALLBACK_GPU_MEM_UTIL` to sum well under 1.0 (both share +the 128 GB unified memory). + +Note: `model switch` does **not** drive the fleet (it rewrites the single-model +`VLLM_*` keys). Change fleet models by editing the fleet `.env` and re-running +`model fleet up --apply`. See `model explain gateway` for routing/failover. +""" + +_GATEWAY = """\ +# model-gear gateway + +The gateway is a stdlib (no third-party deps) OpenAI-compatible reverse proxy +that fronts the fleet's two vLLM backends on one port — the host port the acp +`vllm-local` provider already expects. It runs as the `model-gear-gateway` +container (`python -m model_gear.gateway`). + +## Routing + +- **By name** — a request's `model` field routes to the backend that serves it + (plus any `GATEWAY_ALIASES`). The forwarded body's `model` is rewritten to the + backend's `--served-model-name` so the backend accepts it. +- **Default** — a missing or unknown `model` routes to `GATEWAY_DEFAULT_MODEL` + (the primary), so existing single-model clients keep working unchanged. +- **Failover** — if the chosen backend refuses the connection or returns a 5xx + **before any response body**, the gateway retries the request against the other + backend. A 4xx (client error) is returned verbatim — no failover. Once a 2xx + body starts streaming, there is no retry. SSE streams (`"stream": true`) are + relayed chunk-by-chunk with per-chunk flushing. + +## Endpoints + +`/v1/chat/completions`, `/v1/completions`, `/v1/embeddings` (proxied); `/v1/models` +(lists both backends); `/health` (gateway liveness). Configured via the `gateway` +service's environment in the fleet compose (`PRIMARY_URL` / `FALLBACK_URL` / +`*_SERVED_NAME` / `GATEWAY_DEFAULT_MODEL` / `GATEWAY_ALIASES` / timeouts). """ _WHOAMI = """\ @@ -239,6 +300,8 @@ ("switch",): _SWITCH, ("serve",): _SERVE, ("stop",): _SERVE, + ("fleet",): _FLEET, + ("gateway",): _GATEWAY, ("status",): _STATUS, ("assess",): _ASSESS, ("benchmark",): _BENCHMARK, diff --git a/model_gear/gateway/__init__.py b/model_gear/gateway/__init__.py new file mode 100644 index 0000000..9c9d50e --- /dev/null +++ b/model_gear/gateway/__init__.py @@ -0,0 +1,19 @@ +"""model-gear gateway — a stdlib OpenAI-compatible reverse proxy for the fleet. + +Fronts two always-warm vLLM backends on one port: routes each request by its +``model`` field, defaults unknown/missing names to the primary, and fails over to +the other backend when the chosen one is down. Runs as the ``gateway`` container +in a ``model init --fleet`` deployment (``python -m model_gear.gateway``). + +Public surface: + +- :func:`build_config` — env → ``(RoutingTable, ServerConfig)`` +- :func:`serve` — bind and serve forever +""" + +from __future__ import annotations + +from model_gear.gateway._config import build_config +from model_gear.gateway.server import serve + +__all__ = ["build_config", "serve"] diff --git a/model_gear/gateway/__main__.py b/model_gear/gateway/__main__.py new file mode 100644 index 0000000..b56b656 --- /dev/null +++ b/model_gear/gateway/__main__.py @@ -0,0 +1,19 @@ +"""``python -m model_gear.gateway`` — the gateway container entrypoint. + +Builds the routing table + server config from the environment (set by the fleet +compose ``gateway`` service) and serves forever. +""" + +from __future__ import annotations + +from model_gear.gateway._config import build_config +from model_gear.gateway.server import serve + + +def main() -> None: # pragma: no cover - process entrypoint + table, cfg = build_config() + serve(table, cfg) + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/model_gear/gateway/_config.py b/model_gear/gateway/_config.py new file mode 100644 index 0000000..fa4909e --- /dev/null +++ b/model_gear/gateway/_config.py @@ -0,0 +1,83 @@ +"""Build the gateway's :class:`RoutingTable` + :class:`ServerConfig` from env vars. + +Reads a mapping (``os.environ`` by default) and constructs frozen config objects. +No sockets — pass a plain ``dict`` to unit-test it offline. The env keys mirror +the ``gateway`` service's ``environment:`` block in the fleet compose template. +""" + +from __future__ import annotations + +import os +from collections.abc import Mapping +from dataclasses import dataclass + +from model_gear.gateway._routing import Backend, RoutingTable + +_DEFAULT_PRIMARY = "nvidia/Qwen3-32B-NVFP4" +_DEFAULT_FALLBACK = "mmangkad/Qwen3.6-35B-A3B-NVFP4" + + +@dataclass(frozen=True) +class ServerConfig: + """Where the gateway listens and how patient it is with backends.""" + + host: str + port: int + connect_timeout: float # short: a refused/down backend fails over fast + read_timeout: float # long: a reasoning model's first token is slow + + +def _parse_aliases(raw: str | None) -> dict[str, str]: + """Parse ``alias=served,other=served`` into a dict; skip blank/malformed pairs.""" + out: dict[str, str] = {} + for pair in (raw or "").split(","): + pair = pair.strip() + if "=" not in pair: + continue + alias, _, target = pair.partition("=") + alias, target = alias.strip(), target.strip() + if alias and target: + out[alias] = target + return out + + +def _as_float(env: Mapping[str, str], key: str, default: float) -> float: + try: + return float(env.get(key) or default) + except (TypeError, ValueError): + return float(default) + + +def _as_int(env: Mapping[str, str], key: str, default: int) -> int: + try: + return int(env.get(key) or default) + except (TypeError, ValueError): + return int(default) + + +def build_config(env: Mapping[str, str] | None = None) -> tuple[RoutingTable, ServerConfig]: + """Construct the routing table and server config from environment variables.""" + env = os.environ if env is None else env + + primary = Backend( + name="primary", + base_url=(env.get("PRIMARY_URL") or "http://vllm-primary:8000").rstrip("/"), + served_name=env.get("PRIMARY_SERVED_NAME") or _DEFAULT_PRIMARY, + ) + fallback = Backend( + name="fallback", + base_url=(env.get("FALLBACK_URL") or "http://vllm-fallback:8000").rstrip("/"), + served_name=env.get("FALLBACK_SERVED_NAME") or _DEFAULT_FALLBACK, + ) + table = RoutingTable( + backends=(primary, fallback), + default_model=env.get("GATEWAY_DEFAULT_MODEL") or primary.served_name, + aliases=_parse_aliases(env.get("GATEWAY_ALIASES")), + ) + server = ServerConfig( + host=env.get("GATEWAY_HOST") or "0.0.0.0", # nosec B104 — bind all inside the container + port=_as_int(env, "GATEWAY_PORT", 8000), + connect_timeout=_as_float(env, "GATEWAY_CONNECT_TIMEOUT", 5.0), + read_timeout=_as_float(env, "GATEWAY_READ_TIMEOUT", 600.0), + ) + return table, server diff --git a/model_gear/gateway/_routing.py b/model_gear/gateway/_routing.py new file mode 100644 index 0000000..2f7a0c9 --- /dev/null +++ b/model_gear/gateway/_routing.py @@ -0,0 +1,75 @@ +"""Pure routing / failover logic for the gateway — no sockets, no I/O. + +Kept isolated from :mod:`model_gear.gateway.server` so the gateway's +decision-making core is fully unit-testable offline. ``server`` is the only +module that touches ``http.client`` / sockets. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class Backend: + """One upstream vLLM server in the fleet.""" + + name: str # logical role: "primary" / "fallback" + base_url: str # e.g. "http://vllm-primary:8000" + served_name: str # the OpenAI model id this backend serves + + +@dataclass(frozen=True) +class RoutingTable: + """How the gateway maps a requested model to a backend (frozen → thread-safe).""" + + backends: tuple[Backend, ...] + default_model: str # served_name used for a missing/unknown request model + aliases: dict[str, str] # alias -> served_name + + +def resolve_model(table: RoutingTable, requested: str | None) -> str: + """Map a requested model name to a served model name. + + An alias resolves to its target; a name some backend already serves resolves + to itself; anything else (``None`` or unknown) resolves to ``default_model``. + """ + if requested: + if requested in table.aliases: + return table.aliases[requested] + for backend in table.backends: + if backend.served_name == requested: + return requested + return table.default_model + + +def _backend_for(table: RoutingTable, served_name: str) -> Backend | None: + for backend in table.backends: + if backend.served_name == served_name: + return backend + return None + + +def order_backends(table: RoutingTable, served_name: str) -> list[Backend]: + """Attempt order for ``served_name``: its owner first, then the rest. + + The owner is tried first; the remaining backends are failover targets. An + unmatched ``served_name`` falls back to the default model's owner first. + """ + owner = _backend_for(table, served_name) or _backend_for(table, table.default_model) + ordered: list[Backend] = [] + if owner is not None: + ordered.append(owner) + ordered.extend(b for b in table.backends if b is not owner) + return ordered + + +def list_models_payload(table: RoutingTable) -> dict: + """OpenAI ``/v1/models`` shape listing every backend's served model.""" + return { + "object": "list", + "data": [ + {"id": backend.served_name, "object": "model", "owned_by": "model-gear"} + for backend in table.backends + ], + } diff --git a/model_gear/gateway/server.py b/model_gear/gateway/server.py new file mode 100644 index 0000000..c9353e1 --- /dev/null +++ b/model_gear/gateway/server.py @@ -0,0 +1,392 @@ +"""The gateway HTTP server: a stdlib reverse proxy fronting the fleet backends. + +``ThreadingHTTPServer`` + ``BaseHTTPRequestHandler``; the only module that opens +sockets. Routing/failover *decisions* live in :func:`handle_post` (a seam that +takes an ``open_upstream`` callable, so it's unit-testable without sockets) and in +:mod:`model_gear.gateway._routing` (pure). The handler just reads the request, +calls :func:`handle_post`, and relays the chosen upstream response — buffered for +normal JSON, re-chunked for SSE streams. + +Failover is intentionally narrow: a backend is retried only when it refuses the +connection or returns a 5xx **before any response body reaches the client**. A +4xx is a client error (returned verbatim, no failover); once a 2xx body starts +streaming, there is no retry (the client already has bytes). +""" + +from __future__ import annotations + +import http.client +import json +import sys +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Callable, Iterable +from urllib.parse import urlsplit + +from model_gear.gateway._config import ServerConfig +from model_gear.gateway._routing import ( + Backend, + RoutingTable, + list_models_payload, + order_backends, + resolve_model, +) + +_CHUNK = 65536 + +# Hop-by-hop headers must not be forwarded across a proxy (RFC 7230 §6.1). We also +# drop Content-Length/Transfer-Encoding in both directions and recompute framing. +_HOP_BY_HOP = frozenset( + { + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailers", + "transfer-encoding", + "upgrade", + "host", + "content-length", + } +) + + +# --- request-body helpers (pure, testable) --------------------------------- + + +def _parse_body(body: bytes) -> dict | None: + try: + data = json.loads(body) + except (ValueError, TypeError): + return None + return data if isinstance(data, dict) else None + + +def extract_model(body: bytes) -> str | None: + """The request's ``model`` field, or ``None`` (missing / malformed JSON).""" + data = _parse_body(body) + model = data.get("model") if data else None + return model if isinstance(model, str) and model else None + + +def is_streaming(body: bytes) -> bool: + """True when the request asked for an SSE stream (``"stream": true``).""" + data = _parse_body(body) + return bool(data and data.get("stream") is True) + + +def rewrite_model(body: bytes, served_name: str) -> bytes: + """Rewrite the body's ``model`` to ``served_name`` so the backend accepts it. + + Aliases and default-routing change the model the *gateway* picked; the + backend only knows its own ``--served-model-name``, so the forwarded body + must carry that name. Non-JSON bodies pass through untouched. + """ + data = _parse_body(body) + if data is None: + return body + data["model"] = served_name + return json.dumps(data).encode("utf-8") + + +def filter_headers(headers: Iterable[tuple[str, str]]) -> list[tuple[str, str]]: + """Drop hop-by-hop headers (used both for the forwarded request and response).""" + return [(k, v) for k, v in headers if k.lower() not in _HOP_BY_HOP] + + +def frame_chunk(chunk: bytes) -> bytes: + """Wrap ``chunk`` in HTTP chunked-transfer framing (``\\r\\n\\r\\n``).""" + return b"%X\r\n" % len(chunk) + chunk + b"\r\n" + + +CHUNK_TERMINATOR = b"0\r\n\r\n" + + +def read_chunked_body(rfile, max_bytes: int = 64 * 1024 * 1024) -> bytes: + """Decode an HTTP/1.1 ``Transfer-Encoding: chunked`` request body from ``rfile``. + + Clients/proxies may send a chunked body with no ``Content-Length``; reading + only by length would forward an empty payload. Stops at the zero-length + chunk, ignores chunk extensions, and caps the total at ``max_bytes`` so a + malformed/huge stream can't exhaust memory. + """ + body = bytearray() + while len(body) <= max_bytes: + size_line = rfile.readline() + if not size_line: + break # stream ended early + size_field = size_line.split(b";", 1)[0].strip() # drop chunk extensions + try: + size = int(size_field, 16) + except ValueError: + break # malformed size → stop rather than misread + if size == 0: + rfile.readline() # consume the trailing CRLF after the last chunk + break + body += rfile.read(size) + rfile.readline() # consume the CRLF following each chunk + return bytes(body) + + +# --- upstream client ------------------------------------------------------- + + +class UpstreamError(Exception): + """Connecting to a backend failed before any response (→ try the next one).""" + + +@dataclass +class _Upstream: + """An opened upstream response. Duck-typed: tests substitute their own.""" + + status: int + headers: list[tuple[str, str]] + _resp: object # http.client.HTTPResponse + _conn: object # http.client.HTTPConnection + + def read(self, n: int) -> bytes: + return self._resp.read(n) + + def read_all(self) -> bytes: + return self._resp.read() + + def close(self) -> None: + try: + self._conn.close() + except OSError: + pass + + +def open_upstream( + backend: Backend, + path: str, + body: bytes, + headers: list[tuple[str, str]], + *, + connect_timeout: float, + read_timeout: float, +) -> _Upstream: + """POST ``body`` to ``backend`` and return the opened response. + + Uses a short ``connect_timeout`` for establishing the socket (so a down + backend fails over fast) then a long ``read_timeout`` for the response (a + reasoning model's first token is slow). Raises :class:`UpstreamError` if the + backend can't be reached — including a malformed ``base_url`` (a non-numeric + port makes ``parts.port`` raise ``ValueError``; a bad path/host raises + ``http.client.InvalidURL``) — so the caller fails over instead of 500ing. An + HTTP error *status* is returned as a normal response (the caller decides + whether a 5xx triggers failover). + """ + conn = None + try: + parts = urlsplit(backend.base_url) + if parts.scheme == "https": + conn = http.client.HTTPSConnection( + parts.hostname, parts.port or 443, timeout=connect_timeout + ) + else: + conn = http.client.HTTPConnection( + parts.hostname, parts.port or 80, timeout=connect_timeout + ) + conn.connect() + if conn.sock is not None: + conn.sock.settimeout(read_timeout) + conn.request("POST", path, body=body, headers=dict(headers)) + resp = conn.getresponse() + except (OSError, http.client.HTTPException, ValueError) as exc: + if conn is not None: + conn.close() + raise UpstreamError(f"{backend.name}: {exc}") from exc + return _Upstream( + status=resp.status, headers=filter_headers(resp.getheaders()), _resp=resp, _conn=conn + ) + + +OpenUpstream = Callable[..., _Upstream] + + +# --- routing + failover decision (pure seam) ------------------------------- + + +@dataclass +class GatewayResponse: + """What the handler should send. Either a gateway-generated body, or an + upstream to relay (buffered or streaming).""" + + status: int + headers: list[tuple[str, str]] + body: bytes | None = None + upstream: _Upstream | None = None + streaming: bool = False + attempts: list[str] = field(default_factory=list) + + +def _error_body(message: str, attempts: list[str]) -> bytes: + return json.dumps( + {"error": {"message": message, "type": "upstream_unavailable", "attempts": attempts}} + ).encode("utf-8") + + +def handle_post( + table: RoutingTable, + cfg: ServerConfig, + path: str, + req_headers: Iterable[tuple[str, str]], + body: bytes, + open_upstream: OpenUpstream, +) -> GatewayResponse: + """Resolve the model, then try backends in failover order. + + Returns the first backend that produces a response **before the body** (2xx + or 4xx — committed), or a 502 if every backend refused / 5xx'd. ``open_upstream`` + is injected so this is unit-testable without sockets. + """ + served = resolve_model(table, extract_model(body)) + streaming = is_streaming(body) + fwd_body = rewrite_model(body, served) + fwd_headers = filter_headers(req_headers) + attempts: list[str] = [] + + for backend in order_backends(table, served): + try: + up = open_upstream( + backend, + path, + fwd_body, + fwd_headers, + connect_timeout=cfg.connect_timeout, + read_timeout=cfg.read_timeout, + ) + except UpstreamError as exc: + attempts.append(str(exc)) + continue + if up.status >= 500: + attempts.append(f"{backend.name}: HTTP {up.status}") + up.close() + continue + # 2xx or 4xx → commit to this backend (4xx is a client error; no failover). + return GatewayResponse( + status=up.status, + headers=up.headers, + upstream=up, + streaming=streaming, + attempts=attempts, + ) + + return GatewayResponse( + status=502, + headers=[("Content-Type", "application/json")], + body=_error_body("all fleet backends are unavailable", attempts), + attempts=attempts, + ) + + +# --- the HTTP handler ------------------------------------------------------ + + +class _Handler(BaseHTTPRequestHandler): + """Bound to a ``table`` + ``server_config`` by :func:`_make_handler`.""" + + # Set per-server by _make_handler (frozen dataclasses → safe to share). + table: RoutingTable + server_config: ServerConfig + # HTTP/1.1 so we can stream with chunked transfer encoding. + protocol_version = "HTTP/1.1" + + # --- GET: /health, /v1/models --- + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + route = self.path.split("?", 1)[0] + if route == "/health": + self._send_json(200, {"status": "ok", "service": "model-gear-gateway"}) + elif route == "/v1/models": + self._send_json(200, list_models_payload(self.table)) + else: + self._send_json(404, {"error": {"message": f"not found: {route}", "type": "not_found"}}) + + # --- POST: proxy /v1/* to a backend --- + def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + body = self._read_body() + resp = handle_post( + self.table, + self.server_config, + self.path, + list(self.headers.items()), + body, + open_upstream, + ) + if resp.upstream is None: + self._send_simple(resp.status, resp.headers, resp.body or b"") + return + try: + if resp.streaming: + self._relay_streaming(resp) + else: + self._relay_buffered(resp) + finally: + resp.upstream.close() + + # --- relay helpers --- + def _read_body(self) -> bytes: + cl = self.headers.get("Content-Length") + if cl is not None: + try: + length = int(cl) + except ValueError: + length = 0 + return self.rfile.read(length) if length > 0 else b"" + if "chunked" in (self.headers.get("Transfer-Encoding") or "").lower(): + return read_chunked_body(self.rfile) + return b"" + + def _relay_buffered(self, resp: GatewayResponse) -> None: + data = resp.upstream.read_all() + self.send_response(resp.status) + for key, value in resp.headers: + self.send_header(key, value) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + if data: + self.wfile.write(data) + + def _relay_streaming(self, resp: GatewayResponse) -> None: + self.send_response(resp.status) + for key, value in resp.headers: + self.send_header(key, value) + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + while True: + chunk = resp.upstream.read(_CHUNK) + if not chunk: + break + self.wfile.write(frame_chunk(chunk)) + self.wfile.flush() # SSE must flush per chunk or it buffers until EOF + self.wfile.write(CHUNK_TERMINATOR) + self.wfile.flush() + + def _send_json(self, status: int, obj: dict) -> None: + self._send_simple(status, [("Content-Type", "application/json")], json.dumps(obj).encode()) + + def _send_simple(self, status: int, headers: list[tuple[str, str]], body: bytes) -> None: + self.send_response(status) + for key, value in headers: + self.send_header(key, value) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + if body: + self.wfile.write(body) + + def log_message(self, fmt: str, *args) -> None: # keep request logs tidy in docker logs + sys.stderr.write("[gateway] %s\n" % (fmt % args)) + + +def _make_handler(table: RoutingTable, cfg: ServerConfig) -> type[_Handler]: + bound = type("_BoundHandler", (_Handler,), {"table": table, "server_config": cfg}) + return bound + + +def serve(table: RoutingTable, cfg: ServerConfig) -> None: # pragma: no cover + """Bind and serve forever (the long-lived gateway process).""" + httpd = ThreadingHTTPServer((cfg.host, cfg.port), _make_handler(table, cfg)) + sys.stderr.write(f"[gateway] listening on {cfg.host}:{cfg.port}\n") + httpd.serve_forever() diff --git a/model_gear/runtime/_compose.py b/model_gear/runtime/_compose.py index 2ab74cd..ebe71e9 100644 --- a/model_gear/runtime/_compose.py +++ b/model_gear/runtime/_compose.py @@ -18,9 +18,26 @@ CONTAINER = "model-gear-vllm" COMPOSE_FILE = "docker-compose.yml" ENV_FILE = ".env" - -# Template filename -> destination filename written by the scaffold. -_TEMPLATES = {"docker-compose.yml": COMPOSE_FILE, "env.example": ENV_FILE} +DOCKERFILE_GATEWAY = "Dockerfile.gateway" + +# Fleet container names (model init --fleet / model fleet ...): two always-warm +# vLLM backends plus the stdlib gateway that fronts them on one OpenAI port. +FLEET_PRIMARY = "model-gear-vllm-primary" +FLEET_FALLBACK = "model-gear-vllm-fallback" +FLEET_GATEWAY = "model-gear-gateway" +FLEET_CONTAINERS = (FLEET_PRIMARY, FLEET_FALLBACK, FLEET_GATEWAY) + +# Template filename -> destination filename written by the scaffold. The single +# template set is the default (every existing caller stays unchanged); the fleet +# set scaffolds the 3-container gateway deployment (model init --fleet). +SINGLE_TEMPLATES = {"docker-compose.yml": COMPOSE_FILE, "env.example": ENV_FILE} +FLEET_TEMPLATES = { + "fleet/docker-compose.yml": COMPOSE_FILE, + "fleet/env.example": ENV_FILE, + "fleet/Dockerfile.gateway": DOCKERFILE_GATEWAY, +} +# Back-compat alias: the single set was the only one before the fleet existed. +_TEMPLATES = SINGLE_TEMPLATES def default_deployment_dir() -> Path: @@ -58,20 +75,38 @@ def resolve_deployment_dir(explicit: os.PathLike | str | None) -> Path: # --- scaffolding ----------------------------------------------------------- -def scaffold_plan(target: Path) -> list[tuple[str, bool]]: +def _read_template(template_root, name: str) -> str: + """Read a packaged template by its (possibly ``fleet/``-prefixed) name. + + ``importlib.resources`` traversables join one path segment per ``/`` call, + so split the name and chain rather than passing ``"fleet/foo"`` in one go. + """ + node = template_root + for part in name.split("/"): + node = node / part + return node.read_text(encoding="utf-8") + + +def scaffold_plan( + target: Path, templates: dict[str, str] = SINGLE_TEMPLATES +) -> list[tuple[str, bool]]: """Return ``(dest_name, already_exists)`` for each file ``init`` would write.""" - return [(dest, (target / dest).exists()) for dest in _TEMPLATES.values()] + return [(dest, (target / dest).exists()) for dest in templates.values()] -def write_scaffold(target: os.PathLike | str, *, force: bool) -> list[Path]: +def write_scaffold( + target: os.PathLike | str, *, force: bool, templates: dict[str, str] = SINGLE_TEMPLATES +) -> list[Path]: """Copy the packaged templates into ``target``. Returns written paths. - Refuses to overwrite an existing file unless ``force`` is set. + Refuses to overwrite an existing file unless ``force`` is set. ``templates`` + selects the template set (single-model by default, ``FLEET_TEMPLATES`` for + the gateway deployment). """ dest_dir = Path(target).expanduser() template_root = files("model_gear.templates") written: list[Path] = [] - for tname, dest_name in _TEMPLATES.items(): + for tname, dest_name in templates.items(): dest = dest_dir / dest_name if dest.exists() and not force: raise ModelGearError( @@ -80,8 +115,8 @@ def write_scaffold(target: os.PathLike | str, *, force: bool) -> list[Path]: remediation="re-run with --force to overwrite", ) dest_dir.mkdir(parents=True, exist_ok=True) - for tname, dest_name in _TEMPLATES.items(): - content = (template_root / tname).read_text(encoding="utf-8") + for tname, dest_name in templates.items(): + content = _read_template(template_root, tname) dest = dest_dir / dest_name dest.write_text(content, encoding="utf-8") # .env is meant to hold secrets (HF_TOKEN); keep it owner-only on shared @@ -130,6 +165,13 @@ def compose_up_detached(deploy_dir: os.PathLike | str): return _run(["docker", "compose", "up", "-d"], cwd=str(deploy_dir)) +def compose_up_build(deploy_dir: os.PathLike | str): + """``docker compose up -d --build`` — used by the fleet, whose gateway service + is built from a local ``Dockerfile.gateway`` (``--build`` picks up a new image + on a re-run; first run builds either way).""" + return _run(["docker", "compose", "up", "-d", "--build"], cwd=str(deploy_dir)) + + def docker_available() -> bool: """True if both ``docker`` and ``docker compose`` resolve.""" d = _probe(["docker", "version"]) diff --git a/model_gear/templates/fleet/Dockerfile.gateway b/model_gear/templates/fleet/Dockerfile.gateway new file mode 100644 index 0000000..0736b37 --- /dev/null +++ b/model_gear/templates/fleet/Dockerfile.gateway @@ -0,0 +1,20 @@ +# model-gear gateway — the stdlib OpenAI-compatible front of the fleet. +# +# `model init --fleet` scaffolds this next to the fleet docker-compose.yml, which +# builds it as the `gateway` service. The gateway is pure Python stdlib (no +# runtime deps), so the slim image stays tiny and `pip install` is fast. +# +# MODEL_GEAR_VERSION pins the gateway to the model-gear release that scaffolded +# the deployment (init --fleet writes the running version into .env). It is +# required — pinning keeps the image reproducible. Dev / from-source boxes that +# run ahead of a PyPI release set it to a published TestPyPI `.devN` build. +FROM python:3.12-slim +ARG MODEL_GEAR_VERSION +# One layer: install model-gear (system-wide, as root) and create the unprivileged +# runtime user. The gateway only binds :8000 and makes outbound HTTP calls to the +# backends on the compose network, so it needs no elevated privileges. +RUN pip install --no-cache-dir "model-gear==${MODEL_GEAR_VERSION}" \ + && useradd --create-home --uid 10001 gateway +USER gateway +EXPOSE 8000 +ENTRYPOINT ["python", "-m", "model_gear.gateway"] diff --git a/model_gear/templates/fleet/__init__.py b/model_gear/templates/fleet/__init__.py new file mode 100644 index 0000000..b0dffc1 --- /dev/null +++ b/model_gear/templates/fleet/__init__.py @@ -0,0 +1,6 @@ +"""Packaged *fleet* templates (``model init --fleet``). + +A subpackage so ``importlib.resources.files("model_gear.templates.fleet")`` +resolves; the files themselves are copied verbatim by +:func:`model_gear.runtime._compose.write_scaffold`. +""" diff --git a/model_gear/templates/fleet/docker-compose.yml b/model_gear/templates/fleet/docker-compose.yml new file mode 100644 index 0000000..35a6548 --- /dev/null +++ b/model_gear/templates/fleet/docker-compose.yml @@ -0,0 +1,144 @@ +# model-gear FLEET — two always-warm vLLM backends behind one stdlib gateway. +# +# `model init --fleet` copies this file (+ env.example -> .env, + Dockerfile.gateway) +# into a deployment directory; `model fleet up --apply` drives `docker compose +# up -d --build` there. The gateway serves an OpenAI-compatible API on the host +# port the acp `vllm-local` provider expects; it routes by the request's `model` +# field, defaults unknown/missing names to GATEWAY_DEFAULT_MODEL, and fails over +# to the other backend when the chosen one is down. +# +# model init --fleet --apply # scaffold (writes the running version into .env) +# docker login nvcr.io # NGC API key for the vLLM image +# model fleet up --apply # build the gateway + start all three +# model fleet status # container states + gateway /health + /v1/models +# +# Tuned for DGX Spark (GB10 Grace Blackwell, 128 GB unified memory). Both models +# stay loaded — keep PRIMARY_GPU_MEM_UTIL + FALLBACK_GPU_MEM_UTIL well under 1.0. +services: + vllm-primary: + image: nvcr.io/nvidia/vllm:26.04-py3 # NGC ARM64/Blackwell build (see README to verify the tag) + container_name: model-gear-vllm-primary + restart: unless-stopped + deploy: + resources: + reservations: + devices: + - { driver: nvidia, count: all, capabilities: [gpu] } + ipc: host + ulimits: + memlock: { soft: -1, hard: -1 } + stack: { soft: 67108864, hard: 67108864 } + env_file: + - path: .env + required: false + environment: + - HF_HOME=/root/.cache/huggingface + - TOKENIZERS_PARALLELISM=false + volumes: + - ${HF_CACHE:-${HOME:-/root}/.cache/huggingface}:/root/.cache/huggingface + # No host port — reachable only as http://vllm-primary:8000 on the compose net. + expose: + - "8000" + command: + - vllm + - serve + - ${PRIMARY_MODEL:-nvidia/Qwen3-32B-NVFP4} + - --served-model-name=${PRIMARY_SERVED_NAME:-nvidia/Qwen3-32B-NVFP4} + - --host=0.0.0.0 + - --port=8000 + - --quantization=${PRIMARY_QUANTIZATION:-modelopt_fp4} + - --kv-cache-dtype=fp8 + - --max-model-len=${PRIMARY_MAX_MODEL_LEN:-32768} + - --gpu-memory-utilization=${PRIMARY_GPU_MEM_UTIL:-0.40} + - --reasoning-parser=qwen3 + - --enable-auto-tool-choice + - --tool-call-parser=${PRIMARY_TOOL_CALL_PARSER:-hermes} + - --enable-prefix-caching + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8000/health"] + interval: 30s + timeout: 10s + retries: 5 + start_period: 600s + + vllm-fallback: + image: nvcr.io/nvidia/vllm:26.04-py3 + container_name: model-gear-vllm-fallback + restart: unless-stopped + deploy: + resources: + reservations: + devices: + - { driver: nvidia, count: all, capabilities: [gpu] } + ipc: host + ulimits: + memlock: { soft: -1, hard: -1 } + stack: { soft: 67108864, hard: 67108864 } + env_file: + - path: .env + required: false + environment: + - HF_HOME=/root/.cache/huggingface + - TOKENIZERS_PARALLELISM=false + volumes: + - ${HF_CACHE:-${HOME:-/root}/.cache/huggingface}:/root/.cache/huggingface + expose: + - "8000" + command: + - vllm + - serve + - ${FALLBACK_MODEL:-mmangkad/Qwen3.6-35B-A3B-NVFP4} + - --served-model-name=${FALLBACK_SERVED_NAME:-mmangkad/Qwen3.6-35B-A3B-NVFP4} + - --host=0.0.0.0 + - --port=8000 + - --quantization=${FALLBACK_QUANTIZATION:-modelopt_fp4} + - --kv-cache-dtype=fp8 + - --max-model-len=${FALLBACK_MAX_MODEL_LEN:-32768} + - --gpu-memory-utilization=${FALLBACK_GPU_MEM_UTIL:-0.35} + - --reasoning-parser=qwen3 + - --enable-auto-tool-choice + - --tool-call-parser=${FALLBACK_TOOL_CALL_PARSER:-qwen3_coder} + - --enable-prefix-caching + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8000/health"] + interval: 30s + timeout: 10s + retries: 5 + start_period: 600s + + gateway: + build: + context: . + dockerfile: Dockerfile.gateway + args: + MODEL_GEAR_VERSION: ${MODEL_GEAR_VERSION:-} + container_name: model-gear-gateway + restart: unless-stopped + # Start order only — the gateway tolerates a backend still loading (it routes + # and fails over per request), so it must not block on backend health. + depends_on: + - vllm-primary + - vllm-fallback + environment: + - GATEWAY_PORT=8000 + - PRIMARY_URL=http://vllm-primary:8000 + - PRIMARY_SERVED_NAME=${PRIMARY_SERVED_NAME:-nvidia/Qwen3-32B-NVFP4} + - FALLBACK_URL=http://vllm-fallback:8000 + - FALLBACK_SERVED_NAME=${FALLBACK_SERVED_NAME:-mmangkad/Qwen3.6-35B-A3B-NVFP4} + - GATEWAY_DEFAULT_MODEL=${GATEWAY_DEFAULT_MODEL:-nvidia/Qwen3-32B-NVFP4} + - GATEWAY_ALIASES=${GATEWAY_ALIASES:-} + - GATEWAY_CONNECT_TIMEOUT=${GATEWAY_CONNECT_TIMEOUT:-5} + - GATEWAY_READ_TIMEOUT=${GATEWAY_READ_TIMEOUT:-600} + ports: + - "${VLLM_PORT:-8000}:8000" + healthcheck: + # python:3.12-slim has no curl — probe /health with stdlib urllib instead. + test: + - CMD + - python + - -c + - import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://localhost:8000/health',timeout=3).status==200 else 1) + interval: 30s + timeout: 10s + retries: 5 + start_period: 20s diff --git a/model_gear/templates/fleet/env.example b/model_gear/templates/fleet/env.example new file mode 100644 index 0000000..c6d4206 --- /dev/null +++ b/model_gear/templates/fleet/env.example @@ -0,0 +1,56 @@ +# Copy to .env (model init --fleet does this) and adjust. Consumed by the fleet +# docker-compose.yml: two always-warm vLLM backends + a stdlib gateway that +# fronts them on one OpenAI-compatible port. +# +# Both models stay loaded; the gateway routes by the request's `model` field, +# falls back to the default model for an unknown/missing name, and fails over to +# the other backend if the chosen one is down. Keep the two GPU_MEM_UTIL values +# summing well under 1.0 so both fit in the 128 GB unified memory (DGX Spark). + +# HuggingFace token. Required only for gated repos — accept the license first, +# then paste a read token. Injected into the two vLLM backends, not the gateway. +HF_TOKEN= + +# Host path for the HuggingFace weight cache (shared by both backends). Defaults +# to ~/.cache/huggingface (or /root/.cache/huggingface if $HOME is unset). +HF_CACHE= + +# --- Primary backend (the gateway's default model) ------------------------ +PRIMARY_MODEL=nvidia/Qwen3-32B-NVFP4 +PRIMARY_SERVED_NAME=nvidia/Qwen3-32B-NVFP4 +PRIMARY_MAX_MODEL_LEN=32768 +PRIMARY_GPU_MEM_UTIL=0.40 +PRIMARY_TOOL_CALL_PARSER=hermes # Qwen3 dense → Hermes-style JSON tool calls +PRIMARY_QUANTIZATION=modelopt_fp4 # nvidia/ checkpoint is ModelOpt FP4 + +# --- Fallback backend (MoE: ~3B active → ~10x faster decode) -------------- +FALLBACK_MODEL=mmangkad/Qwen3.6-35B-A3B-NVFP4 +FALLBACK_SERVED_NAME=mmangkad/Qwen3.6-35B-A3B-NVFP4 +FALLBACK_MAX_MODEL_LEN=32768 +FALLBACK_GPU_MEM_UTIL=0.35 +FALLBACK_TOOL_CALL_PARSER=qwen3_coder # Qwen3.6 emits the XML function format +FALLBACK_QUANTIZATION=modelopt_fp4 # verify for the mmangkad checkpoint (may be compressed-tensors) + +# --- Gateway (single front OpenAI endpoint) ------------------------------- +# Host port acp/culture.yaml expects → mapped to the gateway. The two backends +# are reachable only on the internal compose network (vllm-primary/vllm-fallback). +VLLM_PORT=8000 +GATEWAY_DEFAULT_MODEL=nvidia/Qwen3-32B-NVFP4 +# Extra name -> served-name routes, comma-separated (e.g. +# qwen3-32b=nvidia/Qwen3-32B-NVFP4,fast=mmangkad/Qwen3.6-35B-A3B-NVFP4). +GATEWAY_ALIASES= +# Short connect timeout (failover-relevant) vs long read timeout (a reasoning +# model's first token is slow — too tight would falsely trip failover). +GATEWAY_CONNECT_TIMEOUT=5 +GATEWAY_READ_TIMEOUT=600 + +# Pins the gateway image to this model-gear release (model init --fleet fills in +# the running version). Required — dev/from-source boxes set a TestPyPI .devN. +MODEL_GEAR_VERSION= + +# Coherence mirrors so the read-only verbs (status / whoami / doctor) — which read +# the single-model VLLM_* keys — stay sensible on a fleet deployment. Keep equal +# to PRIMARY_* (the gateway's default). +VLLM_MODEL=nvidia/Qwen3-32B-NVFP4 +VLLM_SERVED_NAME=nvidia/Qwen3-32B-NVFP4 +VLLM_TOOL_CALL_PARSER=hermes diff --git a/pyproject.toml b/pyproject.toml index 48c2ea6..735329a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "model-gear" -version = "0.8.1" +version = "0.9.0" description = "model-gear — run, assess, and switch the local vLLM model." readme = "README.md" license = "MIT" diff --git a/tests/test_cli.py b/tests/test_cli.py index 48cf667..7ea62f4 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -85,7 +85,15 @@ def test_learn_json(capsys: pytest.CaptureFixture[str]) -> None: verbs = {tuple(c["path"]) for c in payload["commands"]} assert ("switch",) in verbs assert ("assess",) in verbs - assert set(payload["mutation_safety"]["write_verbs"]) == {"switch", "serve", "stop", "init"} + assert ("fleet",) in verbs + assert set(payload["mutation_safety"]["write_verbs"]) == { + "switch", + "serve", + "stop", + "init", + "fleet up", + "fleet down", + } # --- explain -------------------------------------------------------------- diff --git a/tests/test_cli_fleet.py b/tests/test_cli_fleet.py new file mode 100644 index 0000000..4c00b88 --- /dev/null +++ b/tests/test_cli_fleet.py @@ -0,0 +1,119 @@ +"""Tests for the ``model fleet`` verbs (up / down / status) and ``init --fleet``.""" + +from __future__ import annotations + +import json +import types + +from model_gear.cli import main +from model_gear.runtime import _compose, _health + + +def _ok() -> types.SimpleNamespace: + return types.SimpleNamespace(returncode=0, stdout="", stderr="") + + +def _scaffold_fleet(path): + _compose.write_scaffold(path, force=True, templates=_compose.FLEET_TEMPLATES) + return path + + +# --- fleet up ------------------------------------------------------------- + + +def test_fleet_up_dry_run_changes_nothing(tmp_path, monkeypatch, capsys) -> None: + _scaffold_fleet(tmp_path) + + def boom(*a, **k): + raise AssertionError("compose ran during dry-run") + + monkeypatch.setattr(_compose, "compose_up_build", boom) + rc = main(["fleet", "up", "--compose-dir", str(tmp_path)]) + assert rc == 0 + assert "DRY RUN" in capsys.readouterr().out + + +def test_fleet_up_apply_builds_and_waits(tmp_path, monkeypatch) -> None: + _scaffold_fleet(tmp_path) + calls: list[str] = [] + monkeypatch.setattr( + _compose, "compose_up_build", lambda d: (calls.append("up-build"), _ok())[1] + ) + waited: dict = {} + + def fake_wait(port, **kw): + waited["port"] = port + waited["container"] = kw.get("container") + + monkeypatch.setattr(_health, "wait_health", fake_wait) + rc = main(["fleet", "up", "--compose-dir", str(tmp_path), "--apply", "--json"]) + assert rc == 0 + assert calls == ["up-build"] + assert waited["container"] == _compose.FLEET_GATEWAY # waits on the gateway front + + +# --- fleet down ----------------------------------------------------------- + + +def test_fleet_down_dry_run(tmp_path, capsys) -> None: + _scaffold_fleet(tmp_path) + rc = main(["fleet", "down", "--compose-dir", str(tmp_path)]) + assert rc == 0 + assert "DRY RUN" in capsys.readouterr().out + + +def test_fleet_down_apply(tmp_path, monkeypatch) -> None: + _scaffold_fleet(tmp_path) + calls: list[str] = [] + monkeypatch.setattr(_compose, "compose_down", lambda d: (calls.append("down"), _ok())[1]) + rc = main(["fleet", "down", "--compose-dir", str(tmp_path), "--apply"]) + assert rc == 0 + assert calls == ["down"] + + +# --- fleet status --------------------------------------------------------- + + +def test_fleet_status_json_reports_three_containers(tmp_path, capsys) -> None: + _scaffold_fleet(tmp_path) + rc = main(["fleet", "status", "--compose-dir", str(tmp_path), "--json"]) + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + names = [c["name"] for c in payload["containers"]] + assert names == list(_compose.FLEET_CONTAINERS) + # offline fixture: _probe → None (state "not created"), is_healthy → False. + assert all(c["state"] == "not created" for c in payload["containers"]) + assert payload["gateway_health"] == "not responding" + assert payload["models"] is None # not healthy → no /v1/models fetch + assert payload["port"] == 8000 + + +def test_bare_fleet_defaults_to_status(tmp_path, capsys) -> None: + _scaffold_fleet(tmp_path) + rc = main(["fleet", "--compose-dir", str(tmp_path)]) + assert rc == 0 + out = capsys.readouterr().out + assert "gateway:" in out + assert _compose.FLEET_GATEWAY in out + + +def test_fleet_status_fetches_models_when_healthy(tmp_path, monkeypatch, capsys) -> None: + _scaffold_fleet(tmp_path) + monkeypatch.setattr(_health, "is_healthy", lambda *a, **k: True) + from model_gear import assess + + monkeypatch.setattr( + assess, "_get", lambda url, path, timeout=10: (200, {"data": [{"id": "P"}, {"id": "F"}]}) + ) + rc = main(["fleet", "status", "--compose-dir", str(tmp_path), "--json"]) + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["gateway_health"] == "ok" + assert payload["models"] == ["P", "F"] + + +def test_fleet_status_unscaffolded_errors(capsys) -> None: + # No deployment scaffolded (autouse fixture points the home at an empty dir). + rc = main(["fleet", "status"]) + assert rc == 2 # EXIT_ENV_ERROR + assert "hint:" in capsys.readouterr().err diff --git a/tests/test_gateway_routing.py b/tests/test_gateway_routing.py new file mode 100644 index 0000000..1f792af --- /dev/null +++ b/tests/test_gateway_routing.py @@ -0,0 +1,167 @@ +"""Pure (no-socket) tests for the gateway: routing, config, and body helpers.""" + +from __future__ import annotations + +from model_gear.gateway import server as S +from model_gear.gateway._config import _parse_aliases, build_config +from model_gear.gateway._routing import ( + Backend, + RoutingTable, + list_models_payload, + order_backends, + resolve_model, +) + + +def _table() -> RoutingTable: + return RoutingTable( + backends=( + Backend("primary", "http://vllm-primary:8000", "P"), + Backend("fallback", "http://vllm-fallback:8000", "F"), + ), + default_model="P", + aliases={"fast": "F", "big": "P"}, + ) + + +# --- resolve_model -------------------------------------------------------- + + +def test_resolve_model_exact_alias_default() -> None: + t = _table() + assert resolve_model(t, "P") == "P" + assert resolve_model(t, "F") == "F" + assert resolve_model(t, "fast") == "F" # alias + assert resolve_model(t, "big") == "P" # alias + assert resolve_model(t, None) == "P" # missing → default + assert resolve_model(t, "who-knows") == "P" # unknown → default + assert resolve_model(t, "") == "P" # empty → default + + +# --- order_backends ------------------------------------------------------- + + +def test_order_backends_owner_first_then_failover() -> None: + t = _table() + assert [b.name for b in order_backends(t, "P")] == ["primary", "fallback"] + assert [b.name for b in order_backends(t, "F")] == ["fallback", "primary"] + # an unmatched served name falls back to the default model's owner first + assert [b.name for b in order_backends(t, "nope")] == ["primary", "fallback"] + + +def test_list_models_payload_shape() -> None: + payload = list_models_payload(_table()) + assert payload["object"] == "list" + assert [m["id"] for m in payload["data"]] == ["P", "F"] + assert all(m["object"] == "model" for m in payload["data"]) + + +# --- build_config / aliases ---------------------------------------------- + + +def test_build_config_defaults() -> None: + table, cfg = build_config({}) + assert table.backends[0].served_name == "nvidia/Qwen3-32B-NVFP4" + assert table.backends[1].served_name == "mmangkad/Qwen3.6-35B-A3B-NVFP4" + assert table.default_model == "nvidia/Qwen3-32B-NVFP4" # defaults to primary + assert table.backends[0].base_url == "http://vllm-primary:8000" + assert cfg.host == "0.0.0.0" + assert cfg.port == 8000 + assert cfg.connect_timeout == 5.0 + assert cfg.read_timeout == 600.0 + + +def test_build_config_overrides_and_url_normalised() -> None: + table, cfg = build_config( + { + "PRIMARY_URL": "http://a:9000/", # trailing slash stripped + "PRIMARY_SERVED_NAME": "alpha", + "FALLBACK_URL": "http://b:9001", + "FALLBACK_SERVED_NAME": "beta", + "GATEWAY_DEFAULT_MODEL": "beta", + "GATEWAY_PORT": "9999", + "GATEWAY_CONNECT_TIMEOUT": "2.5", + "GATEWAY_READ_TIMEOUT": "120", + } + ) + assert table.backends[0].base_url == "http://a:9000" + assert table.default_model == "beta" + assert cfg.port == 9999 + assert cfg.connect_timeout == 2.5 + assert cfg.read_timeout == 120.0 + + +def test_build_config_bad_numbers_fall_back_to_defaults() -> None: + _, cfg = build_config({"GATEWAY_PORT": "abc", "GATEWAY_READ_TIMEOUT": "nan?"}) + assert cfg.port == 8000 + assert cfg.read_timeout == 600.0 + + +def test_parse_aliases() -> None: + assert _parse_aliases("a=b, c=d") == {"a": "b", "c": "d"} + assert _parse_aliases("") == {} + assert _parse_aliases(None) == {} + # blank / malformed / half-empty pairs are skipped + assert _parse_aliases("bad, =x, y=, ok=fine") == {"ok": "fine"} + + +# --- request-body helpers ------------------------------------------------- + + +def test_extract_model() -> None: + assert S.extract_model(b'{"model": "foo"}') == "foo" + assert S.extract_model(b'{"no": "model"}') is None + assert S.extract_model(b'{"model": 5}') is None # non-string + assert S.extract_model(b"not json") is None + assert S.extract_model(b"[1,2]") is None # non-dict json + + +def test_is_streaming() -> None: + assert S.is_streaming(b'{"stream": true}') is True + assert S.is_streaming(b'{"stream": false}') is False + assert S.is_streaming(b'{"x": 1}') is False + assert S.is_streaming(b"garbage") is False + + +def test_rewrite_model() -> None: + import json + + out = S.rewrite_model(b'{"model": "fast", "messages": []}', "served-x") + assert json.loads(out)["model"] == "served-x" + assert json.loads(out)["messages"] == [] + # non-JSON / non-dict pass through untouched + assert S.rewrite_model(b"not json", "x") == b"not json" + assert S.rewrite_model(b"[1]", "x") == b"[1]" + + +def test_filter_headers_drops_hop_by_hop() -> None: + out = dict( + S.filter_headers( + [ + ("Host", "x"), + ("Connection", "keep-alive"), + ("Content-Length", "10"), + ("Transfer-Encoding", "chunked"), + ("Content-Type", "application/json"), + ("Authorization", "Bearer t"), + ] + ) + ) + assert out == {"Content-Type": "application/json", "Authorization": "Bearer t"} + + +def test_frame_chunk() -> None: + assert S.frame_chunk(b"hello") == b"5\r\nhello\r\n" + assert S.frame_chunk(b"") == b"0\r\n\r\n" # (matches CHUNK_TERMINATOR for empty) + + +def test_read_chunked_body() -> None: + import io + + # two chunks + zero terminator; chunk extensions on the first are ignored + raw = b"5;ext=1\r\nhello\r\n6\r\n world\r\n0\r\n\r\n" + assert S.read_chunked_body(io.BytesIO(raw)) == b"hello world" + # a malformed size stops the read rather than misreading + assert S.read_chunked_body(io.BytesIO(b"zz\r\n")) == b"" + # empty / truncated stream → empty body + assert S.read_chunked_body(io.BytesIO(b"")) == b"" diff --git a/tests/test_gateway_server.py b/tests/test_gateway_server.py new file mode 100644 index 0000000..440ff56 --- /dev/null +++ b/tests/test_gateway_server.py @@ -0,0 +1,293 @@ +"""Gateway server tests: handle_post failover decisions (no sockets) + a loopback +integration covering the handler relay (buffered + chunked streaming) and the +``open_upstream`` http.client path.""" + +from __future__ import annotations + +import json +import socket +import threading +import urllib.error +import urllib.request +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + +from model_gear.gateway import server as S +from model_gear.gateway._config import build_config + + +def _cfg(**over): + env = {"PRIMARY_SERVED_NAME": "P", "FALLBACK_SERVED_NAME": "F", "GATEWAY_DEFAULT_MODEL": "P"} + env.update(over) + return build_config(env) + + +class _FakeUpstream: + """Duck-typed stand-in for server._Upstream (no socket).""" + + def __init__(self, status, body=b'{"ok":1}', chunks=None): + self.status = status + self.headers = [("Content-Type", "application/json")] + self._body = body + self._chunks = list(chunks) if chunks is not None else None + self.closed = False + + def read_all(self): + return self._body + + def read(self, _n): + if self._chunks is None: + data, self._body = self._body, b"" + return data + return self._chunks.pop(0) if self._chunks else b"" + + def close(self): + self.closed = True + + +def _opener(behavior): + """behavior: {backend_name: status_int | Exception}. Records (name, body).""" + calls = [] + + def opener(backend, path, body, headers, *, connect_timeout, read_timeout): + calls.append((backend.name, body)) + outcome = behavior[backend.name] + if isinstance(outcome, Exception): + raise outcome + return _FakeUpstream(outcome) + + return opener, calls + + +# --- handle_post: failover / default / rewrite (no sockets) --------------- + + +def test_failover_on_connection_refused() -> None: + table, cfg = _cfg() + opener, calls = _opener({"primary": S.UpstreamError("refused"), "fallback": 200}) + resp = S.handle_post(table, cfg, "/v1/chat/completions", [], b'{"model":"P"}', opener) + assert [c[0] for c in calls] == ["primary", "fallback"] + assert resp.status == 200 and resp.upstream is not None + + +def test_failover_on_5xx() -> None: + table, cfg = _cfg() + opener, calls = _opener({"primary": 503, "fallback": 200}) + resp = S.handle_post(table, cfg, "/v1/chat/completions", [], b'{"model":"P"}', opener) + assert [c[0] for c in calls] == ["primary", "fallback"] + assert resp.status == 200 + + +def test_no_failover_on_4xx() -> None: + table, cfg = _cfg() + opener, calls = _opener({"primary": 400, "fallback": 200}) + resp = S.handle_post(table, cfg, "/v1/chat/completions", [], b'{"model":"P"}', opener) + assert [c[0] for c in calls] == ["primary"] # 4xx is a client error → returned verbatim + assert resp.status == 400 + + +def test_all_backends_down_returns_502() -> None: + table, cfg = _cfg() + opener, _ = _opener({"primary": S.UpstreamError("x"), "fallback": S.UpstreamError("y")}) + resp = S.handle_post(table, cfg, "/v1/chat/completions", [], b'{"model":"P"}', opener) + assert resp.status == 502 and resp.upstream is None + assert json.loads(resp.body)["error"]["attempts"] == ["x", "y"] + + +def test_missing_model_routes_to_default() -> None: + table, cfg = _cfg() + opener, calls = _opener({"primary": 200, "fallback": 200}) + resp = S.handle_post(table, cfg, "/v1/chat/completions", [], b"{}", opener) + assert calls[0][0] == "primary" # default model's owner first + assert resp.status == 200 + + +def test_explicit_fallback_routes_to_fallback_first() -> None: + table, cfg = _cfg() + opener, calls = _opener({"primary": 200, "fallback": 200}) + resp = S.handle_post(table, cfg, "/v1/chat/completions", [], b'{"model":"F"}', opener) + assert calls[0][0] == "fallback" + assert resp.status == 200 + + +def test_alias_model_is_rewritten_in_forwarded_body() -> None: + table, cfg = _cfg(GATEWAY_ALIASES="fast=F") + opener, calls = _opener({"fallback": 200, "primary": 200}) + S.handle_post(table, cfg, "/v1/chat/completions", [], b'{"model":"fast"}', opener) + name, fwd_body = calls[0] + assert name == "fallback" # alias resolved → fallback owns it + assert json.loads(fwd_body)["model"] == "F" # body rewritten to the served name + + +def test_streaming_flag_propagates() -> None: + table, cfg = _cfg() + opener, _ = _opener({"primary": 200, "fallback": 200}) + resp = S.handle_post( + table, cfg, "/v1/chat/completions", [], b'{"model":"P","stream":true}', opener + ) + assert resp.streaming is True + + +# --- loopback integration: the real handler relay + open_upstream --------- + + +@pytest.fixture +def gateway(monkeypatch): + """A real ThreadingHTTPServer on an ephemeral port; open_upstream is stubbed + so no real backend is needed. Yields the base URL.""" + table, cfg = _cfg() + + def fake_open(backend, path, body, headers, *, connect_timeout, read_timeout): + if S.is_streaming(body): + return _FakeUpstream(200, chunks=[b"data: a\n\n", b"data: b\n\n"]) + return _FakeUpstream(200, body=b'{"echo": "' + backend.name.encode() + b'"}') + + monkeypatch.setattr(S, "open_upstream", fake_open) + httpd = ThreadingHTTPServer(("127.0.0.1", 0), S._make_handler(table, cfg)) + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + host, port = httpd.server_address + try: + yield f"http://{host}:{port}" + finally: + httpd.shutdown() + httpd.server_close() + + +def test_integration_health_and_models(gateway) -> None: + with urllib.request.urlopen(gateway + "/health", timeout=5) as r: + assert r.status == 200 + assert json.load(r)["status"] == "ok" + with urllib.request.urlopen(gateway + "/v1/models", timeout=5) as r: + payload = json.load(r) + assert [m["id"] for m in payload["data"]] == ["P", "F"] + + +def test_integration_unknown_get_404(gateway) -> None: + with pytest.raises(urllib.error.HTTPError) as exc: + urllib.request.urlopen(gateway + "/nope", timeout=5) + assert exc.value.code == 404 + + +def test_integration_buffered_post(gateway) -> None: + req = urllib.request.Request( + gateway + "/v1/chat/completions", + data=b'{"model":"P"}', + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=5) as r: + assert r.status == 200 + assert r.headers.get("Content-Length") is not None # buffered → Content-Length + assert json.load(r)["echo"] == "primary" + + +def test_integration_chunked_request_body_is_decoded(gateway) -> None: + # A chunked request body (no Content-Length) must reach the backend intact, + # not be forwarded as empty. The stub echoes the backend it routed to; with a + # valid `model` the body must parse and route to the primary (default). + host, port = gateway.removeprefix("http://").split(":") + body = b'{"model":"P"}' + chunked = b"%X\r\n%s\r\n0\r\n\r\n" % (len(body), body) + request = ( + b"POST /v1/chat/completions HTTP/1.1\r\n" + b"Host: x\r\n" + b"Content-Type: application/json\r\n" + b"Transfer-Encoding: chunked\r\n\r\n" + ) + chunked + with socket.create_connection((host, int(port)), timeout=5) as sock: + sock.sendall(request) + buf = b"" + while b"\r\n\r\n" not in buf or b'"echo"' not in buf: + data = sock.recv(4096) + if not data: + break + buf += data + assert b"200" in buf.split(b"\r\n", 1)[0] + assert b'"echo": "primary"' in buf # body decoded → default route, not empty + + +def test_integration_streaming_post_is_chunked(gateway) -> None: + # Raw socket so we can see the chunked framing on the wire (urllib would decode it). + host, port = gateway.removeprefix("http://").split(":") + body = b'{"model":"P","stream":true}' + request = ( + b"POST /v1/chat/completions HTTP/1.1\r\n" + b"Host: x\r\n" + b"Content-Type: application/json\r\n" + b"Content-Length: %d\r\n\r\n" % len(body) + ) + body + with socket.create_connection((host, int(port)), timeout=5) as sock: + sock.sendall(request) + buf = b"" + while b"0\r\n\r\n" not in buf: # read until the chunked terminator + chunk = sock.recv(4096) + if not chunk: + break + buf += chunk + assert b"Transfer-Encoding: chunked" in buf + assert b"data: a\n\n" in buf and b"data: b\n\n" in buf + assert buf.rstrip().endswith(b"0") # final zero-length chunk terminates the body + + +# --- open_upstream over a real loopback backend --------------------------- + + +class _Backend(BaseHTTPRequestHandler): + def do_POST(self): # noqa: N802 + self.rfile.read(int(self.headers.get("Content-Length") or 0)) + code = 503 if self.path == "/boom" else 200 + body = b'{"served": true}' + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *a): # silence + pass + + +@pytest.fixture +def backend(): + httpd = ThreadingHTTPServer(("127.0.0.1", 0), _Backend) + threading.Thread(target=httpd.serve_forever, daemon=True).start() + host, port = httpd.server_address + try: + yield f"http://{host}:{port}" + finally: + httpd.shutdown() + httpd.server_close() + + +def test_open_upstream_success_and_5xx(backend) -> None: + from model_gear.gateway._routing import Backend + + b = Backend("primary", backend, "P") + up = S.open_upstream(b, "/v1/chat/completions", b"{}", [], connect_timeout=2, read_timeout=5) + assert up.status == 200 + assert json.loads(up.read_all())["served"] is True + up.close() + + up = S.open_upstream(b, "/boom", b"{}", [], connect_timeout=2, read_timeout=5) + assert up.status == 503 # returned (not raised) so handle_post can fail over + up.close() + + +def test_open_upstream_refused_raises_upstream_error() -> None: + from model_gear.gateway._routing import Backend + + # Nothing is listening on this port → connect fails fast. + b = Backend("primary", "http://127.0.0.1:1", "P") + with pytest.raises(S.UpstreamError): + S.open_upstream(b, "/x", b"{}", [], connect_timeout=1, read_timeout=2) + + +def test_open_upstream_malformed_url_raises_upstream_error() -> None: + from model_gear.gateway._routing import Backend + + # A non-numeric port makes urlsplit's .port raise ValueError — must surface as + # UpstreamError (→ failover), not an uncaught 500. + b = Backend("primary", "http://host:not-a-port", "P") + with pytest.raises(S.UpstreamError): + S.open_upstream(b, "/x", b"{}", [], connect_timeout=1, read_timeout=2) diff --git a/tests/test_init.py b/tests/test_init.py index 030e075..9f9071c 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -76,3 +76,39 @@ def test_init_local_folder(tmp_path, monkeypatch) -> None: rc = main(["init", ".", "--apply"]) assert rc == 0 assert (tmp_path / "docker-compose.yml").is_file() + + +# --- fleet scaffold ------------------------------------------------------- + + +def test_init_fleet_apply_writes_three_files(tmp_path) -> None: + from model_gear import __version__ + + target = tmp_path / "fleet" + rc = main(["init", "--fleet", str(target), "--apply"]) + assert rc == 0 + assert (target / "docker-compose.yml").is_file() + assert (target / ".env").is_file() + assert (target / "Dockerfile.gateway").is_file() + compose = (target / "docker-compose.yml").read_text() + assert "vllm-primary" in compose + assert "vllm-fallback" in compose + assert "model-gear-gateway" in compose + env = (target / ".env").read_text() + assert "PRIMARY_MODEL=nvidia/Qwen3-32B-NVFP4" in env + assert "FALLBACK_MODEL=mmangkad/Qwen3.6-35B-A3B-NVFP4" in env + # init --fleet pins the gateway image to the running model-gear version. + assert f"MODEL_GEAR_VERSION={__version__}" in env + # coherence mirror keeps the single-model read-only verbs sensible. + assert "VLLM_SERVED_NAME=nvidia/Qwen3-32B-NVFP4" in env + + +def test_init_fleet_dry_run_json(tmp_path, capsys) -> None: + target = tmp_path / "fleet" + rc = main(["init", "--fleet", str(target), "--json"]) + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["fleet"] is True + names = {f["name"] for f in payload["files"]} + assert names == {"docker-compose.yml", ".env", "Dockerfile.gateway"} + assert not target.exists() diff --git a/uv.lock b/uv.lock index 958bd28..d9eb05b 100644 --- a/uv.lock +++ b/uv.lock @@ -236,7 +236,7 @@ wheels = [ [[package]] name = "model-gear" -version = "0.8.0" +version = "0.9.0" source = { editable = "." } [package.dev-dependencies]