From e3c9b5c7462d3be345cf44a952fbf5fd56a28061 Mon Sep 17 00:00:00 2001 From: Jeongkyu Shin Date: Sun, 30 Aug 2026 14:06:44 +0900 Subject: [PATCH] feat(server): give the b10621 router a model cache source and INI presets Closes the four deferred manifest entries issue #1438 recorded after PR #1495: POST /models, DELETE /models, GET /models/sse, and --models-preset, all of which needed a model cache source that did not exist. The router pool now reconciles three model sources with b10621's precedence (cache < models-dir < preset). The cache is the mlxcel model store (--model-store-root / MLXCEL_MODELS_DIR / MLXCEL_CACHE_DIR): its snapshots list as removable source "cache" entries, POST /models validates a requested name (non-empty, normalized repo id, synchronous metadata probe, upstream's validate-then-download order) and downloads the MLX snapshot on a background task while the entry lists as "downloading" with per-url progress, and DELETE /models cancels an in-flight download or unloads a running instance, deletes the snapshot with containment re-asserted against the store root, and emits model_remove. GET /models/sse now serves the full b10621 event vocabulary: models_reload, model_status, status_change, download_progress, download_finished, download_failed, model_remove. --models-preset loads b10621 INI presets: [name] sections define models (model= checkpoint directory, hf-repo= cache snapshot) or overlay configuration onto discovered ones, [*] cascades under every model, keys accept long-option or LLAMA_ARG_* spellings, and each section overlays a per-model clone of the router's startup config re-resolved through build_server_config so preset keys and CLI flags cannot drift (explicitly-given CLI flags win, upstream's overlay order). load-on-startup and dedup-cache-models are honored, stop-timeout parses inert, and an untranslated key fails startup naming the key and section. Router mode now also starts from --models-preset alone, matching upstream's model-less router condition for the surfaces mlxcel exposes. Supporting changes: the downloader gains DownloadHooks (per-chunk progress callbacks and cooperative cancellation with a typed DownloadCancelled marker) plus probe_repo for the metadata probe; ServerStartupConfig derives Clone and carries model_store_root; per-model configs are rebuilt per entry, so preset sampling overlays and per-checkpoint generation_config defaults resolve exactly as on the CLI; dispatch resolves preset aliases and addresses the load path by the entry's real name (found live: the alias path previously answered "not found" from ensure_ready). Manifest: POST /models and GET /models/sse flip to by_design with rationale and pinning tests (bare-name default-org expansion as policy; in-process payload residue as architectural, mirroring --models-dir), DELETE /models flips to supported, --models-preset flips to aliased with its refusal divergences recorded. --models-dir itself stays by_design as #1499 recorded. Validated by unit and route tests (fake downloader driving the same hooks: download, progress payloads, failure, cancel-on-delete, dedup hiding, alias routing, auth on the new surfaces) and a real-checkpoint E2E on Apple Silicon: router started with a preset-defined smollm-135m-4bit (load-on-startup, temp overlay, alias chat), a pre-seeded store snapshot deleted through DELETE /models, and mlx-community/SmolLM-135M-Instruct-4bit downloaded twice through POST /models over the real network with 19 download_progress events, download_finished, model_remove, and a chat completion served from the downloaded cache entry. Refs #1438 --- compat/llama-server/b10621/router-models.json | 47 +- docs/llama-server-compat.md | 8 +- src/bin/mlx_server.rs | 1 + src/commands/serve.rs | 1 + src/downloader/mod.rs | 132 +++- src/downloader/tests.rs | 12 +- src/server/cli_input.rs | 5 + src/server/cli_input_tests.rs | 1 + src/server/llama_compat_tests.rs | 15 +- src/server/mod.rs | 2 + src/server/router_cache.rs | 169 +++++ src/server/router_models.rs | 615 ++++++++++++++++-- src/server/router_models_tests.rs | 478 +++++++++++++- src/server/router_presets.rs | 413 ++++++++++++ src/server/router_presets_tests.rs | 158 +++++ src/server/router_server.rs | 172 +++-- src/server/router_server_tests.rs | 257 +++++++- src/server/startup.rs | 43 +- 18 files changed, 2362 insertions(+), 167 deletions(-) create mode 100644 src/server/router_cache.rs create mode 100644 src/server/router_presets.rs create mode 100644 src/server/router_presets_tests.rs diff --git a/compat/llama-server/b10621/router-models.json b/compat/llama-server/b10621/router-models.json index 5cf0471b8..6a0808f73 100644 --- a/compat/llama-server/b10621/router-models.json +++ b/compat/llama-server/b10621/router-models.json @@ -120,12 +120,13 @@ "default": "disabled", "description": "path to INI file containing model presets for the router server (default: disabled) (env: LLAMA_ARG_MODELS_PRESET)", "discovery": "help-entry-head", - "state": "deferred", + "state": "aliased", "issue": 1438, - "test": "src/server/startup.rs", - "notes": "Accepted on both binaries with the b10621 spelling and env binding, and refused at startup with a diagnostic: mlxcel cannot yet translate llama-server INI presets into per-model configuration, and loading un-preset models while the operator believes presets apply would be the silent-ignore failure epic #1431 exists to remove.", + "test": "src/server/router_presets_tests.rs", + "notes": "Implemented since #1438's cache-source PR: the INI file is parsed with upstream's grammar ([name] sections, [*] global preset cascaded under every model, ; and # comments, version key skipped, unknown keys a startup error), keys accept both the long-option spelling without dashes and the LLAMA_ARG_* environment spelling, and each section overlays a clone of the router's own startup config which is re-resolved through the same pipeline as the CLI, so preset keys and CLI flags cannot drift (explicitly-given router CLI flags win over preset values, upstream's own overlay order). Sections define models via model= (checkpoint directory) or hf-repo= (a cache snapshot), or overlay configuration onto a discovered model; the preset-only load-on-startup, stop-timeout, and dedup-cache-models options parse, with load-on-startup and dedup-cache-models honored and stop-timeout inert (in-process unload needs no force-kill escalation). Translated keys: model, hf-repo, alias, tags, ctx-size, parallel, temp, top-k, top-p, min-p, seed, n-predict.", "divergence": [ - "Any configured preset file fails startup with the migration diagnostic; b10621 loads INI presets and applies them per model (#1438)." + "Keys outside the translated set fail startup with a diagnostic naming the key and section; b10621 can apply any server option per model. The refusal is the epic's no-silent-ignore rule, and the translated set grows with mlxcel's per-model configurability (#1438).", + "A preset hf-repo= must already be a cache snapshot at startup; b10621 downloads a preset's hf-repo on demand when the model loads (#1438)." ], "rationale": null, "mlxcel": { @@ -173,13 +174,11 @@ "source": "tools/server/server.cpp", "condition": "router-mode", "discovery": "source-route-registration", - "state": "deferred", + "state": "supported", "issue": 1438, - "test": "src/server/router_server_tests.rs", - "notes": "Mounted with upstream's reachable refusal surface for a models-dir pool: a missing name answers 400 \"model must be a non-empty string\", an unknown one 500 \"model name=X is not found\", and a discovered one 500 \"model name=X is not removable (not from cache)\", which is exactly what b10621 answers for its models_dir-sourced entries.", - "divergence": [ - "No cache source exists, so no model is ever removable; b10621 can delete cache-sourced models from disk (#1438)." - ], + "test": "src/server/router_server_tests.rs::delete_models_removes_a_cache_entry_from_disk", + "notes": "Implemented since #1438's cache-source PR: a cache-sourced model is removed from disk (an in-flight download is cancelled first, a loaded instance unloaded first), the entry leaves the list, and the SSE stream carries model_remove. The refusal surface matches upstream exactly: a missing name answers 400 \"model must be a non-empty string\", an unknown one 500 \"model name=X is not found\", and a models-dir or preset entry 500 \"model name=X is not removable (not from cache)\". Deletion is containment-checked against the store root (the same sanitized / composition and re-assertion the mlxcel rm verb uses), so an HTTP-supplied name can never delete outside the cache.", + "divergence": [], "rationale": null, "mlxcel": { "route": "DELETE /models" @@ -193,14 +192,18 @@ "source": "tools/server/server.cpp", "condition": "router-mode", "discovery": "source-route-registration", - "state": "deferred", + "state": "by_design", "issue": 1438, - "test": "src/server/router_models_tests.rs", - "notes": "Router mode serves the SSE stream with b10621's event envelope ({\"model\", \"event\"[, \"data\"]}) for models_reload, model_status on load begin, and status_change on load/unload transitions.", + "test": "src/server/router_models_tests.rs::a_download_emits_the_b10621_event_sequence_and_lands_in_the_cache", + "notes": "Router mode serves the SSE stream with b10621's event envelope ({\"model\", \"event\"[, \"data\"]}) and, since #1438's cache-source PR, the full event vocabulary: models_reload, model_status on load begin, status_change on load/unload transitions, download_progress with per-url {done, total} while POST /models fetches into the cache, download_finished / download_failed at each download's end, and model_remove when DELETE /models removes a cache entry.", "divergence": [ - "The event vocabulary is the in-process subset: download_progress / download_finished / download_failed and model_remove never occur (no cache downloads, no removable models), and status_change payloads carry status and failed only, without a child's info/progress blocks (#1438)." + "status_change payloads carry no child exit_code and no child /props-derived info block for loaded models: router mode serves models in-process, so there is no child process to exit or to report loading facts over a pipe. Download progress info is served; the child-lifecycle payload fields have no in-process referent (#1438)." ], - "rationale": null, + "rationale": { + "kind": "architectural", + "reason": "The in-process pool is the same architectural property already argued on --models-dir: models are served inside mlxcel-server rather than as child llama-server processes, so a process exit code and a child's self-reported loading info do not exist to be forwarded. Fabricating an exit_code for a load that never had a process would be less honest than omitting the field.", + "revisit_if": null + }, "mlxcel": { "route": "GET /models/sse" } @@ -213,14 +216,18 @@ "source": "tools/server/server.cpp", "condition": "router-mode", "discovery": "source-route-registration", - "state": "deferred", + "state": "by_design", "issue": 1438, - "test": "src/server/router_server_tests.rs", - "notes": "Mounted with upstream's validation surface: a missing name answers 400 \"model must be a non-empty string\" and an existing one 400 \"model 'X' already exists\".", + "test": "src/server/router_server_tests.rs::post_models_expands_bare_names_to_the_default_org", + "notes": "Implemented since #1438's cache-source PR: the name is validated (non-empty, well-formed repo id, synchronous metadata probe against the hub, exactly upstream's validate-then-download order), an existing name answers 400 \"model 'X' already exists\", a missing one 400 \"model must be a non-empty string\", and a valid new one answers {\"success\": true} while a background task downloads the MLX snapshot into the model cache (the mlxcel store that --model-store-root / MLXCEL_MODELS_DIR / MLXCEL_CACHE_DIR resolve). The transient entry is listed with status \"downloading\" and per-url progress, the SSE stream carries download_progress / download_finished / download_failed, and a finished snapshot lists as a removable cache entry.", "divergence": [ - "A new model name answers a 500 diagnostic instead of starting a HuggingFace download into the cache; mlxcel's router has no cache source yet (#1438)." + "A bare, owner-less model name expands to mlxcel's default organization (mlx-community/) before validation and download, the same expansion -m and mlxcel download apply, so the cache entry lists under the expanded repo id where b10621 lists the verbatim name (#1438)." ], - "rationale": null, + "rationale": { + "kind": "policy", + "reason": "mlxcel's model-reference domain is uniform across every surface that accepts a repo id (-m, mlxcel download, the store layout, and POST /models): a bare name means / everywhere, and the store's / layout has exactly one spelling for each snapshot. Accepting a verbatim owner-less id on this one route would create cache entries the rest of mlxcel cannot name. Pinned by post_models_expands_bare_names_to_the_default_org in src/server/router_server_tests.rs.", + "revisit_if": "mlxcel's model-reference resolution stops expanding bare names, or the store gains a verbatim-id layout addressable by the other surfaces." + }, "mlxcel": { "route": "POST /models" } diff --git a/docs/llama-server-compat.md b/docs/llama-server-compat.md index 73a0559b3..671e8d097 100644 --- a/docs/llama-server-compat.md +++ b/docs/llama-server-compat.md @@ -207,9 +207,13 @@ Concretely, on both server binaries: `--no-ui`, `--no-webui`, `--no-agent`, `--n `GET`/`POST /tools` and `GET`/`POST /cors-proxy` are mounted as b10621's own disabled-feature stub: 403 with `{"error":{"message":"this feature is disabled","type":"feature_disabled"}}`, byte-shaped like upstream's `res_403`. A client of a llama-server deployment that had these features off sees the identical answer here; a deployment that had them on is the recorded divergence. Because the CORS proxy is refused rather than implemented, its SSRF surface (loopback, link-local, metadata-service and DNS-rebinding reachability) does not exist here by construction. Client-declared `tools` in chat-completion requests are unrelated and keep working: mlxcel parses model output for tool calls, it just never executes anything server-side. A server-side MCP tool loop expressed through the OpenAI Responses built-in tool contract is tracked separately in #1457 as a product feature, outside b10621 compatibility. ### `--models-dir` is router mode; the store root moved to `--model-store-root` (#1438) -`--models-dir` / `LLAMA_ARG_MODELS_DIR` now carries b10621's router semantics on `mlxcel-server` and `mlxcel serve`: started without a model argument it serves the router surface over the directory, an in-process model pool where b10621 spawns child processes. Discovery treats each direct subdirectory holding a `config.json` as one model named by the directory; a symlink that resolves outside the models directory is skipped, request `model` names resolve only against the discovered registry, and the management routes sit behind the same API keys as everything else. `GET /models` (and `/v1/models`) reports the b10621 router inventory with `status`, `architecture`, `source: "models_dir"` and `can_remove`, `?reload=1` rescans, `POST /models/load` / `POST /models/unload` answer upstream's exact refusals, `GET /models/sse` streams `{"model", "event", "data"}` status events, `--models-max` (default 4) bounds the loaded set with LRU eviction, and `--models-autoload` plus the per-request `?autoload=` switch control load-on-demand. Requests route by the JSON body's `model` on POST and `?model=` on GET, so `GET /slots?model=X` or `GET /props?model=X` reach the named model's own endpoint, exactly upstream's proxy contract. +`--models-dir` / `LLAMA_ARG_MODELS_DIR` now carries b10621's router semantics on `mlxcel-server` and `mlxcel serve`: started without a model argument it serves the router surface, an in-process model pool where b10621 spawns child processes. The pool reconciles three sources with b10621's precedence (cache < models-dir < preset): the model cache (the mlxcel store that `--model-store-root` / `MLXCEL_MODELS_DIR` / `MLXCEL_CACHE_DIR` resolve), whose `/` snapshots list as removable `source: "cache"` entries; the `--models-dir` directory, where each direct subdirectory holding a `config.json` is one model named by the directory (a symlink resolving outside the directory is skipped); and `--models-preset` INI sections. Request `model` names resolve only against the discovered registry (plus preset aliases), and the management routes sit behind the same API keys as everything else. `GET /models` (and `/v1/models`) reports the b10621 router inventory with `status`, `aliases`, `tags`, `architecture`, `source` and `can_remove`, `?reload=1` rescans, `POST /models/load` / `POST /models/unload` answer upstream's exact refusals, `--models-max` (default 4) bounds the loaded set with LRU eviction, and `--models-autoload` plus the per-request `?autoload=` switch control load-on-demand. Requests route by the JSON body's `model` on POST and `?model=` on GET, so `GET /slots?model=X` or `GET /props?model=X` reach the named model's own endpoint, exactly upstream's proxy contract. -**The old mlxcel meaning moved.** `--models-dir` used to name the local model-store root that `-m ` resolution and auto-download use; that is now `--model-store-root` on both server binaries (`MLXCEL_MODELS_DIR` still works, and the `mlxcel download` / `list` / `rm` / `generate` subcommands keep their `--models-dir` spelling, which was never a llama-server surface). A server command line that combines `--models-dir` with a model argument fails startup with a diagnostic naming the replacement rather than silently picking either meaning. `--models-preset` is accepted and refused at startup: mlxcel cannot yet translate llama-server INI presets, and pretending otherwise would serve un-preset models silently. `--tags` sets the informational tags the `/v1/models` model object reports. +`POST /models` downloads a new model into the cache: the name is validated (non-empty, well-formed repo id, then a synchronous metadata probe against the hub, upstream's validate-then-download order; a bare owner-less name expands to `mlx-community/` first, the same expansion `-m` applies), the route answers `{"success": true}`, and a background task fetches the MLX snapshot while the entry lists as `status: "downloading"` with per-url progress. `DELETE /models?model=X` removes a cache-sourced model from disk (cancelling an in-flight download or unloading a running instance first) and answers upstream's exact refusals for everything else; deletion is containment-checked against the store root, so an HTTP-supplied name can never delete outside the cache. `GET /models/sse` streams the full `{"model", "event"[, "data"]}` vocabulary: `models_reload`, `model_status`, `status_change`, `download_progress`, `download_finished`, `download_failed`, and `model_remove`. + +`--models-preset` loads b10621's INI presets: `[name]` sections define models (`model =` names a checkpoint directory, `hf-repo =` a cache snapshot) or overlay configuration onto discovered ones, `[*]` is the global preset cascaded under every model, keys accept the long-option spelling without dashes or the `LLAMA_ARG_*` environment spelling, and each section overlays a per-model clone of the router's startup config that is re-resolved through the same pipeline as the CLI (explicitly-given router CLI flags win, upstream's own overlay order). The preset-only `load-on-startup` and `dedup-cache-models` options are honored and `stop-timeout` parses inert. A key outside the translated set (`model`, `hf-repo`, `alias`, `tags`, `ctx-size`, `parallel`, `temp`, `top-k`, `top-p`, `min-p`, `seed`, `n-predict`) fails startup with a diagnostic naming the key and section rather than being silently ignored; giving `--models-preset` together with a model argument fails startup the same way, since presets steer nothing outside router mode. + +**The old mlxcel meaning moved.** `--models-dir` used to name the local model-store root that `-m ` resolution and auto-download use; that is now `--model-store-root` on both server binaries (`MLXCEL_MODELS_DIR` still works, and the `mlxcel download` / `list` / `rm` / `generate` subcommands keep their `--models-dir` spelling, which was never a llama-server surface). A server command line that combines `--models-dir` with a model argument fails startup with a diagnostic naming the replacement rather than silently picking either meaning. `--tags` sets the informational tags the `/v1/models` model object reports. Single-model `GET /models` / `GET /v1/models` also moved to b10621's shape: the OpenAI `data` entry carries `aliases` (the full `--alias` list), `tags`, `owned_by: "llamacpp"` (it used to say `"user"`), and a `meta` block of checkpoint facts derived from `config.json` and the safetensors headers (`n_params` unpacks quantized `U32` payloads by their declared bit width), next to b10621's Ollama-compatible `models` block whose `details.format` honestly reads `"safetensors"`. diff --git a/src/bin/mlx_server.rs b/src/bin/mlx_server.rs index a6f32ff3b..d6f482ffd 100644 --- a/src/bin/mlx_server.rs +++ b/src/bin/mlx_server.rs @@ -2369,6 +2369,7 @@ fn build_startup_input(mut args: ServerArgs) -> anyhow::Result anyhow::Result St /// /// Writes to a sibling tempfile first, then atomically renames to `dest`. /// On error, the tempfile is removed and both progress bars are abandoned. +#[allow(clippy::too_many_arguments)] async fn stream_file( client: &reqwest::Client, url: &str, @@ -541,6 +542,8 @@ async fn stream_file( filename: &str, file_pb: &indicatif::ProgressBar, aggregate_pb: &indicatif::ProgressBar, + expected_size: u64, + hooks: &DownloadHooks, ) -> Result { let tmp_name = format!( ".mlxcel-partial.{}.{}", @@ -552,7 +555,18 @@ async fn stream_file( ); let tmp = dest.with_file_name(tmp_name); - let result = stream_to_tempfile(client, url, &tmp, dest, filename, file_pb, aggregate_pb).await; + let result = stream_to_tempfile( + client, + url, + &tmp, + dest, + filename, + file_pb, + aggregate_pb, + expected_size, + hooks, + ) + .await; if result.is_err() { // Best-effort cleanup — ignore errors from remove_file (file may not // exist yet if `File::create` itself failed). The original error from @@ -610,6 +624,7 @@ async fn open_tempfile_no_symlink(tmp: &Path) -> Result { /// Inner implementation of [`stream_file`]: stream bytes into `tmp`, then /// atomically rename to `dest`. Callers are responsible for cleaning up `tmp` /// on error. +#[allow(clippy::too_many_arguments)] async fn stream_to_tempfile( client: &reqwest::Client, url: &str, @@ -618,6 +633,8 @@ async fn stream_to_tempfile( filename: &str, file_pb: &indicatif::ProgressBar, aggregate_pb: &indicatif::ProgressBar, + expected_size: u64, + hooks: &DownloadHooks, ) -> Result { // Open the tempfile FIRST so that an adversary cannot pre-stage a symlink // at `tmp` between the HTTP response and the actual write. `O_NOFOLLOW` @@ -641,10 +658,16 @@ async fn stream_to_tempfile( )); } + // Prefer the response's own Content-Length over the HEAD-pass estimate: + // it reflects the entity actually being served. + let total_size = response.content_length().unwrap_or(expected_size); + let mut stream = response.bytes_stream(); let mut bytes_written: u64 = 0; while let Some(chunk) = stream.next().await { + check_cancelled(hooks) + .with_context(|| format!("Cancelled while downloading {filename}"))?; let chunk = chunk.with_context(|| format!("Stream error while downloading {filename}"))?; out.write_all(&chunk) .await @@ -653,6 +676,9 @@ async fn stream_to_tempfile( bytes_written += chunk_len; file_pb.inc(chunk_len); aggregate_pb.inc(chunk_len); + if let Some(progress) = &hooks.progress { + progress(url, bytes_written, total_size); + } } out.flush() @@ -713,6 +739,85 @@ pub fn ensure_online(what: &str) -> Result<()> { Ok(()) } +/// Programmatic observation and control of one [`download_repo_with_hooks`] +/// run (issue #1438). +/// +/// The CLI download surface reports progress through indicatif bars on +/// stderr; the router server's `POST /models` flow instead forwards progress +/// into its `GET /models/sse` event stream and needs to cancel an in-flight +/// download when `DELETE /models` removes the model. Both needs are optional +/// observations of the same download loop, so they ride along as hooks rather +/// than forking the implementation. +#[derive(Clone, Default)] +pub struct DownloadHooks { + /// Called as bytes arrive for each file: `(url, downloaded, total)`. + /// `total` is `0` when the size is unknown. Files already present on disk + /// report one terminal call with `downloaded == total`. Called from the + /// download worker thread; keep it fast and non-blocking. + pub progress: Option>, + /// Cooperative cancellation flag, checked between streamed chunks and + /// between files. When it becomes `true`, the download aborts with an + /// error wrapping [`DownloadCancelled`], leaving no partial tempfiles + /// behind (completed files stay, exactly like any other mid-run failure). + pub cancel: Option>, +} + +impl std::fmt::Debug for DownloadHooks { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DownloadHooks") + .field("progress", &self.progress.is_some()) + .field("cancel", &self.cancel.is_some()) + .finish() + } +} + +/// Marker error produced when a [`DownloadHooks::cancel`] flag aborts a +/// download. Callers distinguish cancellation from genuine failures with +/// `err.chain().any(|c| c.is::())` (the marker may sit +/// below added context) rather than by matching message text. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DownloadCancelled; + +impl std::fmt::Display for DownloadCancelled { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "download cancelled") + } +} + +impl std::error::Error for DownloadCancelled {} + +/// True when `err`'s chain contains the [`DownloadCancelled`] marker. +#[must_use] +pub fn is_download_cancelled(err: &anyhow::Error) -> bool { + err.chain().any(|cause| cause.is::()) +} + +fn check_cancelled(hooks: &DownloadHooks) -> Result<()> { + if let Some(cancel) = &hooks.cancel + && cancel.load(Ordering::Relaxed) + { + return Err(anyhow::Error::new(DownloadCancelled)); + } + Ok(()) +} + +/// Probe that `repo_id` names a reachable HuggingFace model repository +/// (issue #1438). Resolves the ambient token, builds the API client, and +/// fetches the repository manifest without downloading any file. The router +/// server's `POST /models` validates a requested name with this before +/// starting the background download, mirroring b10621's synchronous metadata +/// fetch in its own handler. Blocking: call from a blocking-capable thread. +pub fn probe_repo(repo_id: &str, revision: Option<&str>) -> Result<()> { + ensure_online(&format!("the repository manifest for '{repo_id}'"))?; + let token = resolve_token(None); + let api = build_api(token)?; + let repo = build_repo_handle(repo_id, revision); + api.repo(repo) + .info() + .map(|_| ()) + .map_err(|err| map_hf_error(err, repo_id, revision, None)) +} + /// Download a HuggingFace model repository snapshot into a local directory. /// /// On success, every allow-listed file from the upstream repository is present @@ -724,6 +829,12 @@ pub fn ensure_online(what: &str) -> Result<()> { /// invalid repo id, missing authentication on a gated repo, missing revision, /// network failure, and on-disk I/O errors. pub fn download_repo(opts: DownloadOptions) -> Result<()> { + download_repo_with_hooks(opts, DownloadHooks::default()) +} + +/// [`download_repo`] with progress observation and cooperative cancellation +/// (issue #1438). See [`DownloadHooks`]. +pub fn download_repo_with_hooks(opts: DownloadOptions, hooks: DownloadHooks) -> Result<()> { // Offline mode forbids every snapshot fetch (issue #1434). The resolver // already reports a cache miss with a repo-specific diagnostic before // reaching here; this is the backstop for any other caller, including @@ -741,17 +852,17 @@ pub fn download_repo(opts: DownloadOptions) -> Result<()> { if tokio::runtime::Handle::try_current().is_ok() { let handle = std::thread::Builder::new() .name("mlxcel-download".to_string()) - .spawn(move || download_repo_blocking(opts)) + .spawn(move || download_repo_blocking(opts, hooks)) .context("Failed to spawn the download worker thread")?; return match handle.join() { Ok(result) => result, Err(panic) => std::panic::resume_unwind(panic), }; } - download_repo_blocking(opts) + download_repo_blocking(opts, hooks) } -fn download_repo_blocking(opts: DownloadOptions) -> Result<()> { +fn download_repo_blocking(opts: DownloadOptions, hooks: DownloadHooks) -> Result<()> { // Issue #171: expand a bare, prefix-less model name (e.g. `Qwen3-4B-4bit`) // to `/` BEFORE anything is derived from `opts.repo_id` — // the HF-cache reuse probe below, the store destination @@ -931,7 +1042,11 @@ fn download_repo_blocking(opts: DownloadOptions) -> Result<()> { // rate limit. Sizes are best-effort: if a HEAD fails we fall back to 0 // (indeterminate bar). We skip the HEAD pass entirely when progress bars // are suppressed. - let size_map: std::collections::HashMap = if show_bars { + // The HEAD-request size pass also runs when a progress hook is installed + // (issue #1438): the router's SSE `download_progress` events carry + // per-file totals, which come from nowhere else. + let size_map: std::collections::HashMap = if show_bars || hooks.progress.is_some() + { rt.block_on(async { let client_ref = &client; let endpoint_ref = &endpoint; @@ -972,6 +1087,7 @@ fn download_repo_blocking(opts: DownloadOptions) -> Result<()> { let mut total_bytes: u64 = 0; for (idx, filename) in wanted.iter().enumerate() { + check_cancelled(&hooks)?; let dest = local_dir.join(filename); // Defense in depth: even if a malicious sibling slipped through the @@ -1005,6 +1121,10 @@ fn download_repo_blocking(opts: DownloadOptions) -> Result<()> { // instead we advance it by the expected file size to keep the total // accurate. let cached_size = fs::metadata(&dest).map(|m| m.len()).unwrap_or(0); + if let Some(progress) = &hooks.progress { + let url = file_url(&endpoint, &opts.repo_id, revision, filename); + progress(&url, cached_size, cached_size); + } println!("[{}/{total}] cached: {filename}", idx + 1,); // Advance aggregate bar by the cached file size so total progress // reflects what is on disk, not just what was downloaded this session. @@ -1031,6 +1151,8 @@ fn download_repo_blocking(opts: DownloadOptions) -> Result<()> { filename, &file_pb, &aggregate_pb, + file_size, + &hooks, )); match result { diff --git a/src/downloader/tests.rs b/src/downloader/tests.rs index b8887c918..37fdcf4dc 100644 --- a/src/downloader/tests.rs +++ b/src/downloader/tests.rs @@ -491,7 +491,17 @@ async fn stream_file_cleans_up_tempfile_on_error() { let file_pb = mp.add(indicatif::ProgressBar::hidden()); let agg_pb = mp.add(indicatif::ProgressBar::hidden()); - let result = stream_file(&client, &url, &dest, "model.safetensors", &file_pb, &agg_pb).await; + let result = stream_file( + &client, + &url, + &dest, + "model.safetensors", + &file_pb, + &agg_pb, + 0, + &DownloadHooks::default(), + ) + .await; assert!( result.is_err(), "expected stream_file to return Err on dropped connection" diff --git a/src/server/cli_input.rs b/src/server/cli_input.rs index ad821c931..5a1f8f27d 100644 --- a/src/server/cli_input.rs +++ b/src/server/cli_input.rs @@ -179,6 +179,10 @@ pub struct ServerStartupInput { pub models_preset: Option, /// `--tags` (#1438). pub tags: Option, + /// `--model-store-root`: the mlxcel model-store root override (#1438's + /// replacement for the pre-router `--models-dir` meaning). In router + /// mode this is also the model cache `POST /models` downloads into. + pub model_store_root: Option, pub warmup: bool, pub no_warmup: bool, pub temperature: f32, @@ -958,6 +962,7 @@ impl ServerStartupInput { models_autoload: self.models_autoload, models_preset: self.models_preset, tags: self.tags, + model_store_root: self.model_store_root, spm_infill: self.infill.spm_infill, embd_normalize: embedding_compat.embd_normalize, embedding_serving_mode: embedding_serving_mode(&embedding_compat), diff --git a/src/server/cli_input_tests.rs b/src/server/cli_input_tests.rs index 19873d403..f2960758c 100644 --- a/src/server/cli_input_tests.rs +++ b/src/server/cli_input_tests.rs @@ -32,6 +32,7 @@ fn sample_input() -> ServerStartupInput { ServerStartupInput { chat_compat: Default::default(), model_path: PathBuf::from("models/foo"), + model_store_root: None, adapter_path: Some(PathBuf::from("adapters/bar")), lora: None, lora_scaled: None, diff --git a/src/server/llama_compat_tests.rs b/src/server/llama_compat_tests.rs index d5ae52bc3..44dae6391 100644 --- a/src/server/llama_compat_tests.rs +++ b/src/server/llama_compat_tests.rs @@ -127,8 +127,19 @@ fn router_probe_app() -> axum::Router { std::fs::create_dir_all(&model_dir).expect("probe models dir"); std::fs::write(model_dir.join("config.json"), "{}").expect("probe config"); let pool = std::sync::Arc::new( - crate::server::router_models::RouterPool::new(dir, ServerConfig::default(), 4, true) - .expect("probe pool"), + crate::server::router_models::RouterPool::new( + crate::server::router_models::RouterSources { + models_dir: Some(dir), + cache: None, + presets: Default::default(), + }, + crate::server::ServerStartupConfig::default(), + Default::default(), + crate::server::router_presets::PresetCliOverrides::default(), + 4, + true, + ) + .expect("probe pool"), ); crate::server::router_server::create_router_app( crate::server::router_server::RouterServerState { diff --git a/src/server/mod.rs b/src/server/mod.rs index 6fb67ec6b..44041a33b 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -54,8 +54,10 @@ pub mod rerank_model; pub mod rerank_worker; pub mod responses_store; pub mod responses_translator; +pub mod router_cache; pub mod router_front; pub mod router_models; +pub mod router_presets; pub mod router_server; pub mod routes; pub mod slot_persist; diff --git a/src/server/router_cache.rs b/src/server/router_cache.rs new file mode 100644 index 000000000..13bf227d1 --- /dev/null +++ b/src/server/router_cache.rs @@ -0,0 +1,169 @@ +// Copyright 2025-2026 Lablup Inc. and Jeongkyu Shin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Router-mode model cache source (llama-server b10621, issue #1438). +//! +//! b10621's router lists its download cache as removable model entries +//! (`source: "cache"`, `can_remove: true`), lets `POST /models` download a +//! new HuggingFace repository into that cache, and lets `DELETE /models` +//! remove a cache entry from disk. mlxcel's equivalent cache is its model +//! store (the directory `--model-store-root` / `MLXCEL_MODELS_DIR` / +//! `MLXCEL_CACHE_DIR` resolve, holding `/` MLX snapshots), so +//! the router's cache source wraps that store. +//! +//! Confinement: every path in and out of the store goes through +//! [`crate::downloader::store`]'s sanitized `/` composition, and +//! removal re-asserts containment under the store root before deleting +//! (`remove_model_with_override`), so an HTTP-supplied model name can never +//! escape the cache directory in either direction. The downloader is behind a +//! trait so tests exercise the full download flow without the network. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use crate::downloader::{self, DownloadHooks, DownloadOptions}; + +/// How a router downloads a repository into the cache. The production +/// implementation is [`HfRouterDownloader`]; tests substitute a fake that +/// writes files locally and drives the same hooks. +pub trait RouterDownloader: Send + Sync { + /// Synchronously validate that `repo_id` names a fetchable repository + /// (b10621 validates by fetching metadata before answering `POST + /// /models`). Blocking; called from a blocking-capable thread. + fn validate(&self, repo_id: &str) -> anyhow::Result<()>; + + /// Download `repo_id` into the models root at `dest_root`, reporting + /// progress and honoring cancellation through `hooks`. Blocking. + fn download(&self, repo_id: &str, dest_root: &Path, hooks: DownloadHooks) + -> anyhow::Result<()>; +} + +/// The real HuggingFace-backed downloader. +#[derive(Debug, Default, Clone, Copy)] +pub struct HfRouterDownloader; + +impl RouterDownloader for HfRouterDownloader { + fn validate(&self, repo_id: &str) -> anyhow::Result<()> { + downloader::probe_repo(repo_id, None) + } + + fn download( + &self, + repo_id: &str, + dest_root: &Path, + hooks: DownloadHooks, + ) -> anyhow::Result<()> { + downloader::download_repo_with_hooks( + DownloadOptions { + repo_id: repo_id.to_string(), + local_dir: None, + models_dir: Some(dest_root.to_path_buf()), + revision: None, + token: None, + force: false, + }, + hooks, + ) + } +} + +/// The router's model cache: the mlxcel model store plus a downloader. +pub struct CacheSource { + root: PathBuf, + downloader: Arc, +} + +impl std::fmt::Debug for CacheSource { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("CacheSource") + .field("root", &self.root) + .finish() + } +} + +impl CacheSource { + pub fn new(root: PathBuf, downloader: Arc) -> Self { + Self { root, downloader } + } + + /// The models root this cache lists, downloads into, and removes from. + pub fn root(&self) -> &Path { + &self.root + } + + /// Enumerate complete snapshots as `(repo_id, path)` pairs. Uses the + /// store's own listing (a directory counts only when it holds a + /// `config.json`), so a half-finished download is not offered as a model. + pub fn list(&self) -> Vec<(String, PathBuf)> { + crate::downloader::list_models_with_override(Some(&self.root)) + .into_iter() + .map(|m| (m.repo_id, m.path)) + .collect() + } + + /// The snapshot directory `repo_id` would occupy (whether or not it + /// exists yet). Repo-id segments are sanitized by the store so the + /// composed path cannot escape the root. + pub fn snapshot_dir(&self, repo_id: &str) -> PathBuf { + crate::downloader::model_dir_with_override(repo_id, Some(&self.root)) + .unwrap_or_else(|| self.root.join(repo_id)) + } + + /// Normalize a requested model name into the repo id used as the cache + /// entry name (bare names expand to the default organization, exactly as + /// `-m ` resolution does). + pub fn normalize_name(&self, name: &str) -> anyhow::Result { + downloader::normalize_repo_id(name) + } + + /// Validate that `repo_id` is fetchable (metadata probe, no file + /// downloads). Blocking. + pub fn validate(&self, repo_id: &str) -> anyhow::Result<()> { + self.downloader.validate(repo_id) + } + + /// Download `repo_id` into the cache. Blocking; run on a worker thread. + pub fn download(&self, repo_id: &str, hooks: DownloadHooks) -> anyhow::Result<()> { + self.downloader.download(repo_id, &self.root, hooks) + } + + /// Remove `repo_id`'s snapshot from the cache. Deleting is contained to + /// the store root by `remove_model_under`'s re-assertion; a missing + /// snapshot (for example a cancelled download that never completed a + /// file) is not an error, matching b10621's best-effort + /// `common_download_remove`. + pub fn remove(&self, repo_id: &str) -> anyhow::Result<()> { + use crate::downloader::{RemoveOutcome, remove_model_with_override}; + match remove_model_with_override(repo_id, None, Some(&self.root)) { + Ok(RemoveOutcome::Removed { path, size_bytes }) => { + tracing::info!( + "router: removed cache model '{repo_id}' ({size_bytes} bytes) at {}", + path.display() + ); + Ok(()) + } + Ok(RemoveOutcome::HfCacheOnly { hf_path }) => { + // The read-only HuggingFace cache is not ours to manage; the + // router's cache never lists it, so this arm is unreachable + // from HTTP. Refuse rather than pretend. + anyhow::bail!( + "model '{repo_id}' only exists in the read-only HuggingFace cache at {}", + hf_path.display() + ) + } + Ok(RemoveOutcome::NotFound) => Ok(()), + Err(err) => Err(anyhow::anyhow!(err.to_string())), + } + } +} diff --git a/src/server/router_models.rs b/src/server/router_models.rs index 416b26998..0d2290d73 100644 --- a/src/server/router_models.rs +++ b/src/server/router_models.rs @@ -14,37 +14,48 @@ //! Router-mode model pool (llama-server b10621 compatible, issue #1438). //! -//! b10621's router server discovers models under `--models-dir`, spawns one -//! child llama-server process per loaded model, and proxies requests to the -//! child the request's `model` field names. mlxcel serves the same HTTP -//! surface from one process: each pool entry, once loaded, owns a full -//! [`AppState`] plus its axum [`Router`], and the dispatcher forwards the -//! request into that sub-app in-process instead of over a child socket. The -//! state machine mirrors upstream's `UNLOADED -> LOADING -> LOADED` (a +//! b10621's router server discovers models from three sources: its download +//! cache (removable, `POST /models` downloads into it), the `--models-dir` +//! directory, and `--models-preset` INI sections, with name collisions +//! resolved cache < models-dir < preset. It spawns one child llama-server +//! process per loaded model and proxies requests to the child the request's +//! `model` field names. mlxcel serves the same HTTP surface from one +//! process: each pool entry, once loaded, owns a full [`AppState`] plus its +//! axum [`Router`], and the dispatcher forwards the request into that +//! sub-app in-process instead of over a child socket. The state machine +//! mirrors upstream's `UNLOADED -> LOADING -> LOADED` plus the transient +//! `DOWNLOADING` while `POST /models` fetches a repository into the cache (a //! failed load returns to `unloaded` with a failure recorded, which is -//! b10621's failed shape), `--models-max` bounds the concurrently loaded -//! set with LRU eviction, and `--models-autoload` (plus the per-request +//! b10621's failed shape), `--models-max` bounds the concurrently loaded set +//! with LRU eviction, and `--models-autoload` (plus the per-request //! `?autoload=` override) loads on demand. //! -//! Confinement: entry names come only from discovery, requests resolve names -//! through the registry (a request can never smuggle a path), and a -//! discovered entry whose canonical path escapes the canonical models -//! directory (a symlink pointing outside it) is skipped at scan time. +//! Confinement: entry names come only from discovery, the cache store, or +//! operator-authored presets; requests resolve names through the registry (a +//! request can never smuggle a path); a discovered entry whose canonical +//! path escapes the canonical models directory (a symlink pointing outside +//! it) is skipped at scan time; and cache downloads/removals go through the +//! store's sanitized `/` composition with containment +//! re-asserted before any deletion (see [`super::router_cache`]). use std::collections::BTreeMap; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI64, Ordering}; use std::sync::{Arc, Mutex, RwLock}; -use super::config::ServerConfig; -use super::{AppState, ChatTemplateProcessor, ModelProvider}; +use super::router_cache::CacheSource; +use super::router_presets::{PresetCliOverrides, PresetSection, RouterPresets}; +use super::{AppState, ChatTemplateProcessor, ModelProvider, ServerStartupConfig}; -/// b10621 `server_model_status` (the subset an in-process pool reaches). +/// b10621 `server_model_status` (the subset an in-process pool reaches; +/// `downloaded` is upstream's "erase on next reload" marker, which the +/// in-process pool replaces with an immediate rescan). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RouterModelStatus { Unloaded, Loading, Loaded, + Downloading, } impl RouterModelStatus { @@ -53,6 +64,26 @@ impl RouterModelStatus { Self::Unloaded => "unloaded", Self::Loading => "loading", Self::Loaded => "loaded", + Self::Downloading => "downloading", + } + } +} + +/// b10621 `server_model_source`: where an entry came from, which decides +/// `can_remove` (only cache entries are removable). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RouterModelSource { + Cache, + ModelsDir, + Preset, +} + +impl RouterModelSource { + pub fn as_str(&self) -> &'static str { + match self { + Self::Cache => "cache", + Self::ModelsDir => "models_dir", + Self::Preset => "preset", } } } @@ -61,6 +92,19 @@ impl RouterModelStatus { pub struct RouterModelEntry { pub name: String, pub path: PathBuf, + pub source: RouterModelSource, + /// Preset aliases (b10621 lists them on the model object and resolves + /// them for request routing). + pub aliases: Vec, + pub tags: Vec, + /// Hidden by a preset's `dedup-cache-models` (still resolvable by name). + pub hidden: bool, + /// The preset section that shaped this entry, for the `status.preset` + /// INI block; `None` when no named section applies. + preset: Option, + /// Per-model server config: the router's own CLI config overlaid with + /// this model's preset (issue #1438). + config: super::config::ServerConfig, state: Mutex, last_used: AtomicI64, } @@ -69,8 +113,18 @@ pub struct RouterModelEntry { struct EntryState { /// The loaded sub-app; `Some` while loading or loaded. app: Option, - /// Set when the last load failed; cleared by the next load attempt. + /// Set when the last load or download failed; cleared by the next + /// attempt. failed: bool, + /// `Some` while `POST /models` is downloading this entry into the cache. + download: Option, +} + +struct DownloadInFlight { + cancel: Arc, + /// b10621 `loaded_info` during a download: + /// `{"progress": {url: {"done": n, "total": n}}}`. + progress: serde_json::Value, } #[derive(Clone)] @@ -85,6 +139,9 @@ impl RouterModelEntry { Ok(guard) => guard, Err(_) => return RouterModelStatus::Unloaded, }; + if guard.download.is_some() { + return RouterModelStatus::Downloading; + } match &guard.app { None => RouterModelStatus::Unloaded, Some(app) => { @@ -112,7 +169,8 @@ impl RouterModelEntry { .is_some_and(|app| app.state.model_provider.is_chat_unavailable()) } - /// b10621 `is_running`: loading or loaded. + /// b10621 `is_running`: loading or loaded (a downloading entry is not + /// running; it has no weights to serve). pub fn is_running(&self) -> bool { matches!( self.status(), @@ -120,6 +178,13 @@ impl RouterModelEntry { ) } + pub fn is_downloading(&self) -> bool { + self.state + .lock() + .map(|guard| guard.download.is_some()) + .unwrap_or(false) + } + fn router(&self) -> Option { self.state .lock() @@ -128,6 +193,15 @@ impl RouterModelEntry { .as_ref() .map(|app| app.router.clone()) } + + fn download_progress_json(&self) -> Option { + self.state + .lock() + .ok()? + .download + .as_ref() + .map(|d| d.progress.clone()) + } } /// A status snapshot for `GET /models` and the SSE stream. @@ -138,15 +212,37 @@ pub struct RouterModelSnapshot { pub failed: bool, pub vision: bool, pub audio: bool, + pub source: RouterModelSource, + pub aliases: Vec, + pub tags: Vec, + pub hidden: bool, + /// `{"progress": {...}}` while downloading (b10621 `loaded_info`). + pub download_info: Option, + /// The preset section as INI text (b10621 `status.preset`). + pub preset_ini: Option, +} + +/// Model sources the pool reconciles on every rescan (issue #1438). +#[derive(Default)] +pub struct RouterSources { + /// `--models-dir` discovery root. + pub models_dir: Option, + /// The model cache (mlxcel model store) with its downloader. + pub cache: Option, + /// Parsed `--models-preset` sections. + pub presets: RouterPresets, } /// The pool shared by every router-mode handler. pub struct RouterPool { entries: RwLock>>, - models_dir: PathBuf, - /// Template for each per-model [`ServerConfig`]; the router's own CLI - /// arguments apply to every model, the b10621 base-preset overlay. - base_config: ServerConfig, + sources: RouterSources, + /// Template for each per-model [`ServerStartupConfig`]; the router's own + /// CLI arguments apply to every model (the b10621 base-preset overlay), + /// and each model's preset section overlays underneath them. + base_startup: ServerStartupConfig, + api_keys: super::ApiKeys, + cli_overrides: PresetCliOverrides, pub models_max: usize, pub autoload_default: bool, events: tokio::sync::broadcast::Sender, @@ -155,7 +251,7 @@ pub struct RouterPool { load_lock: tokio::sync::Mutex<()>, } -/// Why a name failed to resolve or load. +/// Why a name failed to resolve, load, download, or be removed. #[derive(Debug, Clone, PartialEq, Eq)] pub enum RouterPoolError { MissingName, @@ -163,20 +259,28 @@ pub enum RouterPoolError { NotLoaded, LoadFailed(String), Capacity(String), + /// b10621: only cache-sourced models are removable. + NotRemovable(String), + /// `POST /models` on a name the pool already has. + AlreadyExists(String), } impl RouterPool { pub fn new( - models_dir: PathBuf, - base_config: ServerConfig, + sources: RouterSources, + base_startup: ServerStartupConfig, + api_keys: super::ApiKeys, + cli_overrides: PresetCliOverrides, models_max: usize, autoload_default: bool, ) -> anyhow::Result { let (events, _) = tokio::sync::broadcast::channel(256); let pool = Self { entries: RwLock::new(BTreeMap::new()), - models_dir, - base_config, + sources, + base_startup, + api_keys, + cli_overrides, models_max, autoload_default, events, @@ -186,27 +290,150 @@ impl RouterPool { Ok(pool) } - /// Scan `models_dir` for checkpoint directories and reconcile the - /// registry (b10621 `load_models`). New directories appear as unloaded - /// entries; removed directories drop their entry unless the model is - /// still running. + /// Build the per-model [`super::config::ServerConfig`] by overlaying the + /// entry's preset section onto a clone of the router's startup config and + /// re-running the CLI's own resolution pipeline, so preset keys and CLI + /// flags resolve identically (per-model `generation_config.json` sampling + /// defaults included). + fn build_entry_config( + &self, + name: &str, + path: &Path, + section: &PresetSection, + ) -> super::config::ServerConfig { + let mut startup = self.base_startup.clone(); + startup.model_path = path.to_path_buf(); + super::router_presets::apply_section_to_startup(&mut startup, section, &self.cli_overrides); + let mut config = super::startup::build_server_config(&startup, self.api_keys.clone()); + // The name a model answers with is its directory name / repo id; the + // router's own --alias never leaks into models, the b10621 + // preset-strip rule. Preset aliases do apply. + config.model_alias = Some(name.to_string()); + let mut aliases = vec![name.to_string()]; + aliases.extend(section.aliases.iter().cloned()); + config.model_aliases = aliases; + config + } + + /// Scan every source and reconcile the registry (b10621 `load_models`): + /// cache snapshots, then `--models-dir` directories (overriding cache + /// names), then preset sections (defining new models or re-sourcing + /// discovered ones). New names appear as unloaded entries; removed names + /// drop their entry unless the model is still running or downloading. pub fn rescan(&self) -> anyhow::Result<()> { - let discovered = discover_models(&self.models_dir)?; - let mut entries = self + // Phase 1: enumerate sources and build replacement entries without + // holding the registry lock (config building reads the checkpoint's + // generation_config.json). + let mut discovered: BTreeMap = BTreeMap::new(); + if let Some(cache) = &self.sources.cache { + for (name, path) in cache.list() { + discovered.insert(name, (path, RouterModelSource::Cache)); + } + } + if let Some(dir) = &self.sources.models_dir { + for (name, path) in discover_models(dir)? { + discovered.insert(name, (path, RouterModelSource::ModelsDir)); + } + } + for (name, section) in &self.sources.presets.models { + if let Some(path) = §ion.model_path { + discovered.insert(name.clone(), (path.clone(), RouterModelSource::Preset)); + } else if let Some(repo) = §ion.hf_repo { + let Some(cache) = &self.sources.cache else { + tracing::warn!( + "router: preset '[{name}]' names hf-repo '{repo}' but no model cache \ + is configured; skipping" + ); + continue; + }; + let path = cache.snapshot_dir(repo); + if path.join("config.json").is_file() { + discovered.insert(name.clone(), (path, RouterModelSource::Preset)); + } else { + tracing::warn!( + "router: preset '[{name}]' names hf-repo '{repo}', which is not in the \ + cache; download it first (POST /models or `mlxcel download {repo}`)" + ); + } + } else if let Some((path, _)) = discovered.get(name.as_str()).cloned() { + // Overlay-only section: the model keeps its discovered path + // but is re-sourced as preset, upstream's merge rule. + discovered.insert(name.clone(), (path, RouterModelSource::Preset)); + } else { + tracing::warn!( + "router: preset '[{name}]' names no checkpoint (model= / hf-repo=) and \ + matches no discovered model" + ); + } + } + + // A preset with `dedup-cache-models` hides cache entries that resolve + // to the snapshot the preset itself serves. + let mut hidden_names: Vec = Vec::new(); + for (pname, section) in &self.sources.presets.models { + if !section.dedup_cache_models { + continue; + } + let preset_path = discovered.get(pname.as_str()).map(|(p, _)| p.clone()); + let Some(preset_path) = preset_path else { + continue; + }; + for (cname, (cpath, csource)) in &discovered { + if cname != pname && *csource == RouterModelSource::Cache && *cpath == preset_path { + hidden_names.push(cname.clone()); + } + } + } + + let existing: BTreeMap> = self .entries - .write() - .map_err(|_| anyhow::anyhow!("router pool poisoned"))?; - for (name, path) in &discovered { - entries.entry(name.clone()).or_insert_with(|| { + .read() + .map_err(|_| anyhow::anyhow!("router pool poisoned"))? + .clone(); + + let mut rebuilt: BTreeMap> = BTreeMap::new(); + for (name, (path, source)) in &discovered { + if let Some(entry) = existing.get(name) + && (entry.is_running() || entry.is_downloading()) + { + // A running model keeps serving its current configuration; + // source changes apply on its next load (upstream unloads on + // preset change, which the in-process pool defers to the + // operator's own unload/load cycle). + rebuilt.insert(name.clone(), entry.clone()); + continue; + } + let section = self.sources.presets.for_model(name); + let config = self.build_entry_config(name, path, §ion); + rebuilt.insert( + name.clone(), Arc::new(RouterModelEntry { name: name.clone(), path: path.clone(), + source: *source, + aliases: section.aliases.clone(), + tags: section.tags.clone(), + hidden: hidden_names.contains(name), + preset: self.sources.presets.models.get(name).cloned(), + config, state: Mutex::new(EntryState::default()), last_used: AtomicI64::new(0), - }) - }); + }), + ); + } + // Keep running or downloading entries whose source vanished (or, for + // downloads, never existed yet). + for (name, entry) in &existing { + if !rebuilt.contains_key(name) && (entry.is_running() || entry.is_downloading()) { + rebuilt.insert(name.clone(), entry.clone()); + } } - entries.retain(|name, entry| discovered.contains_key(name) || entry.is_running()); + + let mut entries = self + .entries + .write() + .map_err(|_| anyhow::anyhow!("router pool poisoned"))?; + *entries = rebuilt; drop(entries); self.notify("models_reload", "*", serde_json::Value::Null); Ok(()) @@ -231,12 +458,18 @@ impl RouterPool { if entry.failed() { data["failed"] = true.into(); } + if let Some(progress) = entry.download_progress_json() { + // b10621 carries the download progress as the entry's + // `loaded_info`/`progress` blocks on `status_change`. + data["info"] = progress; + } self.notify("status_change", &entry.name, data); } /// Resolve a request's model name (b10621 `router_validate_model`): - /// resolves through the registry only, so a path can never be smuggled - /// in, and enforces the not-loaded refusal when autoload is off. + /// resolves through the registry only (name or preset alias), so a path + /// can never be smuggled in, and enforces the not-loaded refusal when + /// autoload is off. pub fn resolve( &self, name: &str, @@ -246,10 +479,7 @@ impl RouterPool { return Err(RouterPoolError::MissingName); } let entry = self - .entries - .read() - .ok() - .and_then(|entries| entries.get(name).cloned()) + .lookup(name) .ok_or_else(|| RouterPoolError::NotFound(name.to_string()))?; if !autoload && !entry.is_running() { return Err(RouterPoolError::NotLoaded); @@ -257,10 +487,23 @@ impl RouterPool { Ok(entry) } + /// Exact-name lookup (load/unload/delete address models by name only). pub fn get(&self, name: &str) -> Option> { self.entries.read().ok()?.get(name).cloned() } + /// Name-or-alias lookup (request routing). + pub fn lookup(&self, name: &str) -> Option> { + let entries = self.entries.read().ok()?; + if let Some(entry) = entries.get(name) { + return Some(entry.clone()); + } + entries + .values() + .find(|e| e.aliases.iter().any(|a| a == name)) + .cloned() + } + pub fn snapshot(&self) -> Vec { let entries = match self.entries.read() { Ok(entries) => entries, @@ -288,11 +531,31 @@ impl RouterPool { failed: entry.failed(), vision, audio, + source: entry.source, + aliases: entry.aliases.clone(), + tags: entry.tags.clone(), + hidden: entry.hidden, + download_info: entry.download_progress_json(), + preset_ini: entry + .preset + .as_ref() + .map(|section| section.to_ini(&entry.name)), } }) .collect() } + /// Entries whose preset asked for `load-on-startup`. + pub fn load_on_startup_names(&self) -> Vec { + self.sources + .presets + .models + .iter() + .filter(|(_, section)| section.load_on_startup) + .map(|(name, _)| name.clone()) + .collect() + } + fn running_count(&self) -> usize { self.entries .read() @@ -307,6 +570,11 @@ impl RouterPool { let entry = self .get(name) .ok_or_else(|| RouterPoolError::NotFound(name.to_string()))?; + if entry.is_downloading() { + return Err(RouterPoolError::LoadFailed(format!( + "model '{name}' is still downloading" + ))); + } let _permit = self.load_lock.lock().await; if entry.is_running() { return Ok(entry); @@ -349,15 +617,10 @@ impl RouterPool { // an observable state; the tokenizer and chat-template reads are // still filesystem work, so they run on the blocking pool rather // than stalling the async runtime. - let (name, path, base_config) = ( - entry.name.clone(), - entry.path.clone(), - self.base_config.clone(), - ); - let built = - tokio::task::spawn_blocking(move || build_model_app(&name, &path, &base_config)) - .await - .map_err(|join_err| RouterPoolError::LoadFailed(join_err.to_string()))?; + let (path, config) = (entry.path.clone(), entry.config.clone()); + let built = tokio::task::spawn_blocking(move || build_model_app(&path, config)) + .await + .map_err(|join_err| RouterPoolError::LoadFailed(join_err.to_string()))?; match built { Ok((state, router)) => { if let Ok(mut guard) = entry.state.lock() { @@ -403,7 +666,7 @@ impl RouterPool { self.notify_status(&entry); return Ok(entry); } - RouterModelStatus::Unloaded => { + RouterModelStatus::Unloaded | RouterModelStatus::Downloading => { self.notify_status(&entry); return Err(RouterPoolError::LoadFailed(format!( "model '{name}' failed to load" @@ -435,11 +698,18 @@ impl RouterPool { } /// Unload `name` (b10621 `server_models::unload` through - /// `POST /models/unload`). + /// `POST /models/unload`). Unloading a downloading model cancels the + /// download, upstream's own unload-during-download behavior. pub fn unload(&self, name: &str) -> Result<(), RouterPoolError> { let entry = self .get(name) .ok_or_else(|| RouterPoolError::NotFound(name.to_string()))?; + if let Ok(guard) = entry.state.lock() + && let Some(download) = &guard.download + { + download.cancel.store(true, Ordering::Relaxed); + return Ok(()); + } if !entry.is_running() { return Err(RouterPoolError::NotLoaded); } @@ -470,20 +740,233 @@ impl RouterPool { } } +/// Cache download and removal (issue #1438). Split from the core pool impl +/// so the state machine above stays readable. +impl RouterPool { + /// Whether a model cache (the mlxcel store) is configured for this pool. + pub fn has_cache(&self) -> bool { + self.sources.cache.is_some() + } + + /// Normalize a requested `POST /models` name into the cache's repo-id + /// spelling. Errors when no cache is configured or the name cannot form a + /// valid repository id. + pub fn normalize_cache_name(&self, name: &str) -> anyhow::Result { + let cache = self + .sources + .cache + .as_ref() + .ok_or_else(|| anyhow::anyhow!("no model cache is configured"))?; + cache.normalize_name(name) + } + + /// Synchronously validate that `repo_id` is fetchable (b10621's metadata + /// probe before it starts a download). Blocking. + pub fn validate_cache_repo(&self, repo_id: &str) -> anyhow::Result<()> { + let cache = self + .sources + .cache + .as_ref() + .ok_or_else(|| anyhow::anyhow!("no model cache is configured"))?; + cache.validate(repo_id) + } + + /// Start downloading `name` into the cache (`POST /models`). The name + /// must already be validated (non-empty, normalized, not present, + /// metadata probe passed). Registers a transient `downloading` entry, + /// then fetches on a background task, forwarding progress into the SSE + /// stream and rescanning when the snapshot lands. + pub fn start_download(self: &Arc, name: &str) -> Result<(), RouterPoolError> { + let Some(cache) = &self.sources.cache else { + return Err(RouterPoolError::LoadFailed( + "no model cache is configured (set --model-store-root, MLXCEL_MODELS_DIR, or \ + MLXCEL_CACHE_DIR)" + .to_string(), + )); + }; + if self.lookup(name).is_some() { + return Err(RouterPoolError::AlreadyExists(name.to_string())); + } + let cancel = Arc::new(AtomicBool::new(false)); + let path = cache.snapshot_dir(name); + let section = PresetSection::default(); + let config = self.build_entry_config(name, &path, §ion); + let entry = Arc::new(RouterModelEntry { + name: name.to_string(), + path, + source: RouterModelSource::Cache, + aliases: Vec::new(), + tags: Vec::new(), + hidden: false, + preset: None, + config, + state: Mutex::new(EntryState { + app: None, + failed: false, + download: Some(DownloadInFlight { + cancel: cancel.clone(), + progress: serde_json::json!({ "progress": {} }), + }), + }), + last_used: AtomicI64::new(0), + }); + { + let mut entries = self + .entries + .write() + .map_err(|_| RouterPoolError::LoadFailed("router pool poisoned".into()))?; + entries.insert(entry.name.clone(), entry.clone()); + } + self.notify_status(&entry); + + let pool = self.clone(); + let task_entry = entry; + let repo = name.to_string(); + tokio::spawn(async move { + let hooks = crate::downloader::DownloadHooks { + progress: Some(pool.download_progress_hook(&task_entry)), + cancel: Some(cancel), + }; + let blocking_pool = pool.clone(); + let blocking_repo = repo.clone(); + let result = tokio::task::spawn_blocking(move || { + let Some(cache) = &blocking_pool.sources.cache else { + return Err(anyhow::anyhow!("no model cache configured")); + }; + cache.download(&blocking_repo, hooks) + }) + .await + .unwrap_or_else(|join_err| Err(anyhow::anyhow!(join_err.to_string()))); + + let ok = result.is_ok(); + if let Ok(mut guard) = task_entry.state.lock() { + guard.download = None; + guard.failed = !ok; + } + match &result { + Ok(()) => tracing::info!("router: download of '{repo}' finished"), + Err(err) if crate::downloader::is_download_cancelled(err) => { + tracing::info!("router: download of '{repo}' cancelled"); + } + Err(err) => tracing::warn!("router: download of '{repo}' failed: {err:#}"), + } + // b10621 order: the terminal download event first, then the + // reload that reconciles the entry (a finished snapshot becomes a + // regular cache entry; a failed one drops out of the list). + pool.notify( + if ok { + "download_finished" + } else { + "download_failed" + }, + &repo, + serde_json::Value::Null, + ); + if let Err(err) = pool.rescan() { + tracing::warn!("router: rescan after download of '{repo}' failed: {err:#}"); + } + }); + Ok(()) + } + + /// The per-chunk progress hook: updates the entry's progress block and + /// forwards a throttled `download_progress` event (unthrottled SSE at + /// chunk granularity would flood every subscriber). + fn download_progress_hook( + self: &Arc, + entry: &Arc, + ) -> Arc { + let pool = self.clone(); + let entry = entry.clone(); + let last_emit: Mutex> = Mutex::new(None); + Arc::new(move |url: &str, done: u64, total: u64| { + let snapshot = { + let Ok(mut guard) = entry.state.lock() else { + return; + }; + let Some(download) = guard.download.as_mut() else { + return; + }; + download.progress["progress"][url] = + serde_json::json!({ "done": done, "total": total }); + download.progress.clone() + }; + let terminal = total > 0 && done >= total; + let due = { + let Ok(mut last) = last_emit.lock() else { + return; + }; + let now = std::time::Instant::now(); + let due = terminal + || last.is_none_or(|t| { + now.duration_since(t) >= std::time::Duration::from_millis(250) + }); + if due { + *last = Some(now); + } + due + }; + if due { + pool.notify("download_progress", &entry.name, snapshot); + } + }) + } + + /// Remove `name` from the cache (`DELETE /models`, b10621 + /// `server_models::remove`): cancel an in-flight download or stop a + /// running instance, delete the snapshot from disk (containment-checked + /// against the store root), drop the entry, and emit `model_remove`. + pub async fn remove(&self, name: &str) -> Result<(), RouterPoolError> { + let entry = self + .get(name) + .ok_or_else(|| RouterPoolError::NotFound(name.to_string()))?; + if entry.source != RouterModelSource::Cache { + return Err(RouterPoolError::NotRemovable(name.to_string())); + } + let Some(cache) = &self.sources.cache else { + return Err(RouterPoolError::NotRemovable(name.to_string())); + }; + + if let Ok(guard) = entry.state.lock() + && let Some(download) = &guard.download + { + tracing::info!("router: cancelling download for model '{name}'"); + download.cancel.store(true, Ordering::Relaxed); + } + // Wait for the download worker to acknowledge the cancel (it clears + // the download state in its completion block). + let waited = std::time::Instant::now(); + while entry.is_downloading() { + if waited.elapsed() > std::time::Duration::from_secs(120) { + return Err(RouterPoolError::LoadFailed(format!( + "model '{name}' download did not stop in time" + ))); + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + if entry.is_running() { + tracing::info!("router: stopping model instance '{name}' before removal"); + self.unload_entry(&entry); + } + + cache + .remove(name) + .map_err(|err| RouterPoolError::LoadFailed(err.to_string()))?; + if let Ok(mut entries) = self.entries.write() { + entries.remove(name); + } + self.notify("model_remove", name, serde_json::Value::Null); + Ok(()) + } +} + /// Build the per-model serving stack: tokenizer, chat template, provider, /// [`AppState`], and the model's own axum app (without the CORS layer; the /// router's top level owns CORS so headers are emitted exactly once). fn build_model_app( - name: &str, model_path: &Path, - base_config: &ServerConfig, + config: super::config::ServerConfig, ) -> anyhow::Result<(AppState, axum::Router)> { - let mut config = base_config.clone(); - // The name a child answers with is its directory name; the router's own - // --alias never leaks into children, the b10621 preset-strip rule. - config.model_alias = Some(name.to_string()); - config.model_aliases = vec![name.to_string()]; - let tokenizer = crate::tokenizer::load_tokenizer(model_path)?; let chat_template = ChatTemplateProcessor::from_model_path(model_path)?.unwrap_or_default(); let batch_metrics = Arc::new(super::state::BatchMetrics::new()); diff --git a/src/server/router_models_tests.rs b/src/server/router_models_tests.rs index 64bcca7b3..bda32e665 100644 --- a/src/server/router_models_tests.rs +++ b/src/server/router_models_tests.rs @@ -14,12 +14,21 @@ //! Unit tests for the router-mode pool (issue #1438): discovery and its //! confinement boundary, name resolution, the autoload gate, failed loads, -//! and the SSE event stream. +//! the SSE event stream, and the cache source (list / download / remove with +//! the full b10621 event vocabulary). -use std::path::PathBuf; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; -use super::{RouterModelStatus, RouterPool, RouterPoolError, discover_models}; -use crate::server::config::ServerConfig; +use super::{ + RouterModelSource, RouterModelStatus, RouterPool, RouterPoolError, RouterSources, + discover_models, +}; +use crate::downloader::DownloadHooks; +use crate::server::ServerStartupConfig; +use crate::server::router_cache::{CacheSource, RouterDownloader}; +use crate::server::router_presets::{PresetCliOverrides, parse_preset_text}; fn temp_models_dir(tag: &str) -> PathBuf { let dir = std::env::temp_dir().join(format!( @@ -40,8 +49,133 @@ fn add_fake_model(root: &std::path::Path, name: &str) { std::fs::write(dir.join("config.json"), "{}").expect("config.json"); } +/// A downloader that materializes a fake snapshot locally, driving the same +/// hooks the HuggingFace downloader drives. `delay_until_cancel` makes the +/// download hang until the cancel flag flips, for cancellation tests. +struct FakeDownloader { + fail: bool, + delay_until_cancel: bool, + downloads: AtomicUsize, +} + +impl FakeDownloader { + fn ok() -> Arc { + Arc::new(Self { + fail: false, + delay_until_cancel: false, + downloads: AtomicUsize::new(0), + }) + } + + fn failing() -> Arc { + Arc::new(Self { + fail: true, + delay_until_cancel: false, + downloads: AtomicUsize::new(0), + }) + } + + fn hanging() -> Arc { + Arc::new(Self { + fail: false, + delay_until_cancel: true, + downloads: AtomicUsize::new(0), + }) + } +} + +impl RouterDownloader for FakeDownloader { + fn validate(&self, _repo_id: &str) -> anyhow::Result<()> { + Ok(()) + } + + fn download( + &self, + repo_id: &str, + dest_root: &Path, + hooks: DownloadHooks, + ) -> anyhow::Result<()> { + self.downloads.fetch_add(1, Ordering::SeqCst); + let url = format!("https://example.invalid/{repo_id}/config.json"); + if let Some(progress) = &hooks.progress { + progress(&url, 1, 2); + } + if self.delay_until_cancel { + let cancel = hooks.cancel.clone().expect("cancel flag"); + let started = std::time::Instant::now(); + while !cancel.load(Ordering::Relaxed) { + if started.elapsed() > std::time::Duration::from_secs(10) { + anyhow::bail!("fake download was never cancelled"); + } + std::thread::sleep(std::time::Duration::from_millis(10)); + } + return Err(anyhow::Error::new(crate::downloader::DownloadCancelled)); + } + if self.fail { + anyhow::bail!("fake network failure"); + } + let dest = dest_root.join(repo_id); + std::fs::create_dir_all(&dest)?; + std::fs::write(dest.join("config.json"), "{}")?; + if let Some(progress) = &hooks.progress { + progress(&url, 2, 2); + } + Ok(()) + } +} + +fn sources_dir_only(root: PathBuf) -> RouterSources { + RouterSources { + models_dir: Some(root), + cache: None, + presets: Default::default(), + } +} + +fn pool_from(sources: RouterSources, max: usize, autoload: bool) -> RouterPool { + RouterPool::new( + sources, + ServerStartupConfig::default(), + Default::default(), + PresetCliOverrides::default(), + max, + autoload, + ) + .expect("pool") +} + fn pool(root: PathBuf, max: usize, autoload: bool) -> RouterPool { - RouterPool::new(root, ServerConfig::default(), max, autoload).expect("pool") + pool_from(sources_dir_only(root), max, autoload) +} + +/// Drain currently queued events into a Vec of (event, model) pairs. +fn drain_events( + events: &mut tokio::sync::broadcast::Receiver, +) -> Vec<(String, String)> { + let mut out = Vec::new(); + while let Ok(event) = events.try_recv() { + out.push(( + event["event"].as_str().unwrap_or_default().to_string(), + event["model"].as_str().unwrap_or_default().to_string(), + )); + } + out +} + +async fn wait_for_event( + events: &mut tokio::sync::broadcast::Receiver, + wanted: &str, +) -> serde_json::Value { + tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + let event = events.recv().await.expect("event stream open"); + if event["event"] == wanted { + return event; + } + } + }) + .await + .unwrap_or_else(|_| panic!("timed out waiting for '{wanted}'")) } #[test] @@ -166,3 +300,337 @@ async fn rescan_picks_up_new_directories_and_drops_removed_ones() { assert!(pool.get("second").is_some()); assert!(pool.get("first").is_none(), "removed dir drops its entry"); } + +// ── Cache source (#1438: POST /models, DELETE /models, SSE vocabulary) ────── + +#[tokio::test] +async fn cache_snapshots_list_as_removable_cache_entries() { + let cache_root = temp_models_dir("cache-list"); + add_fake_model(&cache_root.join("mlx-community"), "tiny"); + let models_dir = temp_models_dir("cache-list-dir"); + add_fake_model(&models_dir, "local"); + let pool = pool_from( + RouterSources { + models_dir: Some(models_dir), + cache: Some(CacheSource::new(cache_root, FakeDownloader::ok())), + presets: Default::default(), + }, + 4, + true, + ); + let snapshots = pool.snapshot(); + let cache_entry = snapshots + .iter() + .find(|s| s.name == "mlx-community/tiny") + .expect("cache entry listed"); + assert_eq!(cache_entry.source, RouterModelSource::Cache); + let dir_entry = snapshots + .iter() + .find(|s| s.name == "local") + .expect("dir entry"); + assert_eq!(dir_entry.source, RouterModelSource::ModelsDir); +} + +#[tokio::test] +async fn models_dir_wins_a_name_collision_with_the_cache() { + let cache_root = temp_models_dir("collide-cache"); + add_fake_model(&cache_root, "same-name"); + let models_dir = temp_models_dir("collide-dir"); + add_fake_model(&models_dir, "same-name"); + let pool = pool_from( + RouterSources { + models_dir: Some(models_dir.clone()), + cache: Some(CacheSource::new(cache_root, FakeDownloader::ok())), + presets: Default::default(), + }, + 4, + true, + ); + let entry = pool.get("same-name").expect("entry"); + assert_eq!(entry.source, RouterModelSource::ModelsDir); + assert_eq!( + entry.path, + models_dir.join("same-name").canonicalize().unwrap() + ); +} + +#[tokio::test] +async fn a_download_emits_the_b10621_event_sequence_and_lands_in_the_cache() { + let cache_root = temp_models_dir("dl-ok"); + let downloader = FakeDownloader::ok(); + let pool = Arc::new(pool_from( + RouterSources { + models_dir: None, + cache: Some(CacheSource::new(cache_root.clone(), downloader.clone())), + presets: Default::default(), + }, + 4, + true, + )); + let mut events = pool.subscribe(); + + pool.start_download("mlx-community/new-model") + .expect("start"); + // Transient entry is visible as `downloading` immediately. + let entry = pool + .get("mlx-community/new-model") + .expect("transient entry"); + assert_eq!(entry.status(), RouterModelStatus::Downloading); + assert!(entry.is_downloading()); + + wait_for_event(&mut events, "download_finished").await; + // The rescan after the download lists the snapshot as a cache entry. + wait_for_event(&mut events, "models_reload").await; + let entry = pool.get("mlx-community/new-model").expect("cache entry"); + assert_eq!(entry.source, RouterModelSource::Cache); + assert_eq!(entry.status(), RouterModelStatus::Unloaded); + assert!(!entry.is_downloading()); + assert!( + cache_root + .join("mlx-community/new-model/config.json") + .is_file(), + "snapshot landed in the cache" + ); + assert_eq!(downloader.downloads.load(Ordering::SeqCst), 1); +} + +#[tokio::test] +async fn download_progress_events_carry_per_url_done_and_total() { + let cache_root = temp_models_dir("dl-progress"); + let pool = Arc::new(pool_from( + RouterSources { + models_dir: None, + cache: Some(CacheSource::new(cache_root, FakeDownloader::ok())), + presets: Default::default(), + }, + 4, + true, + )); + let mut events = pool.subscribe(); + pool.start_download("mlx-community/progress-model") + .expect("start"); + let event = wait_for_event(&mut events, "download_progress").await; + assert_eq!(event["model"], "mlx-community/progress-model"); + let progress = event["data"]["progress"] + .as_object() + .expect("per-url progress map"); + let (_, first) = progress.iter().next().expect("one url"); + assert!(first.get("done").is_some() && first.get("total").is_some()); +} + +#[tokio::test] +async fn a_failed_download_emits_download_failed_and_drops_the_entry() { + let cache_root = temp_models_dir("dl-fail"); + let pool = Arc::new(pool_from( + RouterSources { + models_dir: None, + cache: Some(CacheSource::new(cache_root, FakeDownloader::failing())), + presets: Default::default(), + }, + 4, + true, + )); + let mut events = pool.subscribe(); + pool.start_download("mlx-community/broken-model") + .expect("start"); + wait_for_event(&mut events, "download_failed").await; + wait_for_event(&mut events, "models_reload").await; + assert!( + pool.get("mlx-community/broken-model").is_none(), + "failed download drops out of the list" + ); +} + +#[tokio::test] +async fn start_download_refuses_an_existing_name() { + let cache_root = temp_models_dir("dl-dup"); + add_fake_model(&cache_root.join("mlx-community"), "present"); + let pool = Arc::new(pool_from( + RouterSources { + models_dir: None, + cache: Some(CacheSource::new(cache_root, FakeDownloader::ok())), + presets: Default::default(), + }, + 4, + true, + )); + assert!(matches!( + pool.start_download("mlx-community/present"), + Err(RouterPoolError::AlreadyExists(_)) + )); +} + +#[tokio::test] +async fn remove_deletes_a_cache_model_from_disk_and_emits_model_remove() { + let cache_root = temp_models_dir("rm-ok"); + add_fake_model(&cache_root.join("mlx-community"), "doomed"); + let pool = pool_from( + RouterSources { + models_dir: None, + cache: Some(CacheSource::new(cache_root.clone(), FakeDownloader::ok())), + presets: Default::default(), + }, + 4, + true, + ); + let mut events = pool.subscribe(); + pool.remove("mlx-community/doomed").await.expect("remove"); + assert!(pool.get("mlx-community/doomed").is_none()); + assert!( + !cache_root.join("mlx-community/doomed").exists(), + "snapshot removed from disk" + ); + let names: Vec<(String, String)> = drain_events(&mut events); + assert!( + names.contains(&( + "model_remove".to_string(), + "mlx-community/doomed".to_string() + )), + "model_remove emitted: {names:?}" + ); +} + +#[tokio::test] +async fn remove_refuses_models_dir_and_unknown_entries() { + let models_dir = temp_models_dir("rm-refuse"); + add_fake_model(&models_dir, "local"); + let cache_root = temp_models_dir("rm-refuse-cache"); + let pool = pool_from( + RouterSources { + models_dir: Some(models_dir), + cache: Some(CacheSource::new(cache_root, FakeDownloader::ok())), + presets: Default::default(), + }, + 4, + true, + ); + assert!(matches!( + pool.remove("local").await, + Err(RouterPoolError::NotRemovable(name)) if name == "local" + )); + assert!(matches!( + pool.remove("ghost").await, + Err(RouterPoolError::NotFound(_)) + )); +} + +#[tokio::test] +async fn remove_cancels_an_in_flight_download() { + let cache_root = temp_models_dir("rm-cancel"); + let pool = Arc::new(pool_from( + RouterSources { + models_dir: None, + cache: Some(CacheSource::new(cache_root, FakeDownloader::hanging())), + presets: Default::default(), + }, + 4, + true, + )); + let mut events = pool.subscribe(); + pool.start_download("mlx-community/hanging").expect("start"); + assert!( + pool.get("mlx-community/hanging") + .expect("entry") + .is_downloading() + ); + + pool.remove("mlx-community/hanging") + .await + .expect("remove cancels"); + assert!(pool.get("mlx-community/hanging").is_none()); + let names = drain_events(&mut events); + assert!( + names.iter().any(|(e, _)| e == "download_failed"), + "cancelled download reports download_failed: {names:?}" + ); + assert!( + names + .iter() + .any(|(e, m)| e == "model_remove" && m == "mlx-community/hanging"), + "model_remove emitted: {names:?}" + ); +} + +// ── Presets (#1438: --models-preset translation) ──────────────────────────── + +#[tokio::test] +async fn preset_sections_define_models_with_aliases_and_tags() { + let checkpoint_root = temp_models_dir("preset-def"); + add_fake_model(&checkpoint_root, "ckpt"); + let ini = format!( + "[my-model]\nmodel = {}\nalias = short, alt\ntags = prod\n", + checkpoint_root.join("ckpt").display() + ); + let presets = parse_preset_text(&ini).expect("parse"); + let pool = pool_from( + RouterSources { + models_dir: None, + cache: None, + presets, + }, + 4, + true, + ); + let entry = pool.get("my-model").expect("preset entry"); + assert_eq!(entry.source, RouterModelSource::Preset); + assert_eq!(entry.aliases, vec!["short", "alt"]); + assert_eq!(entry.tags, vec!["prod"]); + // Aliases resolve for request routing. + assert!(pool.lookup("short").is_some()); + // The model object reproduces the section as INI text. + let snapshot = pool + .snapshot() + .into_iter() + .find(|s| s.name == "my-model") + .expect("snapshot"); + let ini_text = snapshot.preset_ini.expect("preset ini"); + assert!(ini_text.starts_with("[my-model]\n"), "{ini_text}"); + assert!(!ini_text.contains("alias"), "alias stripped: {ini_text}"); +} + +#[tokio::test] +async fn preset_overlay_resources_a_discovered_model_and_dedup_hides_cache_twins() { + let cache_root = temp_models_dir("preset-dedup"); + add_fake_model(&cache_root.join("mlx-community"), "twin"); + let ini = "[served]\nhf-repo = mlx-community/twin\ndedup-cache-models = 1\n"; + let presets = parse_preset_text(ini).expect("parse"); + let pool = pool_from( + RouterSources { + models_dir: None, + cache: Some(CacheSource::new(cache_root, FakeDownloader::ok())), + presets, + }, + 4, + true, + ); + let served = pool.get("served").expect("preset entry"); + assert_eq!(served.source, RouterModelSource::Preset); + let twin = pool + .snapshot() + .into_iter() + .find(|s| s.name == "mlx-community/twin") + .expect("cache twin"); + assert!(twin.hidden, "cache twin hidden by dedup-cache-models"); +} + +#[tokio::test] +async fn load_on_startup_names_come_from_the_preset() { + let checkpoint_root = temp_models_dir("preset-los"); + add_fake_model(&checkpoint_root, "ckpt"); + let ini = format!( + "[eager]\nmodel = {}\nload-on-startup = 1\n[lazy]\nmodel = {}\n", + checkpoint_root.join("ckpt").display(), + checkpoint_root.join("ckpt").display() + ); + let presets = parse_preset_text(&ini).expect("parse"); + let pool = pool_from( + RouterSources { + models_dir: None, + cache: None, + presets, + }, + 4, + true, + ); + assert_eq!(pool.load_on_startup_names(), vec!["eager"]); +} diff --git a/src/server/router_presets.rs b/src/server/router_presets.rs new file mode 100644 index 000000000..4c55e53e4 --- /dev/null +++ b/src/server/router_presets.rs @@ -0,0 +1,413 @@ +// Copyright 2025-2026 Lablup Inc. and Jeongkyu Shin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! b10621 `--models-preset` INI translation (issue #1438). +//! +//! b10621 loads an INI file whose sections name router models: `[*]` is a +//! global preset merged into every model, and any other `[name]` section +//! either defines a new model (when it carries `model =` or `hf-repo =`) or +//! overlays configuration onto a model discovered from the cache or +//! `--models-dir`. Keys are llama-server option spellings without leading +//! dashes (`ctx-size`, `temp`) or their `LLAMA_ARG_*` environment names, plus +//! the three preset-only options (`load-on-startup`, `stop-timeout`, +//! `dedup-cache-models`). Reference: +//! https://github.com/ggml-org/llama.cpp/blob/main/common/preset.cpp +//! +//! mlxcel translates each section onto its own per-model +//! [`super::ServerStartupConfig`] overlay and rebuilds the model's +//! `ServerConfig` through the same resolution pipeline the CLI uses, so +//! preset keys and CLI flags cannot drift apart. A key outside the translated +//! set fails startup with a diagnostic naming the key and section: loading +//! un-preset models while the operator believes presets apply is the +//! silent-ignore failure epic #1431 exists to remove. Router CLI arguments +//! that were explicitly given override preset values, which is b10621's own +//! overlay order (`preset.merge(base_preset)` with the CLI-args preset +//! winning). + +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; + +/// One parsed preset section, already translated into typed directives. +#[derive(Debug, Clone, Default, PartialEq)] +pub struct PresetSection { + /// `model =`: checkpoint directory (MLX SafeTensors snapshot). + pub model_path: Option, + /// `hf-repo =`: HuggingFace repo id resolved against the model cache. + pub hf_repo: Option, + /// `alias =`: comma-separated alias list. + pub aliases: Vec, + /// `tags =`: comma-separated informational tags. + pub tags: Vec, + pub ctx_size: Option, + pub n_parallel: Option, + pub temperature: Option, + pub top_k: Option, + pub top_p: Option, + pub min_p: Option, + /// `seed =`: b10621 semantics, negative means "random per request". + pub seed: Option, + pub n_predict: Option, + /// Preset-only `load-on-startup`: begin loading this model as the router + /// starts. + pub load_on_startup: bool, + /// Preset-only `stop-timeout` (seconds). Accepted for INI compatibility; + /// the in-process pool unloads by dropping the model's app state, which + /// needs no force-kill escalation, so the value steers nothing. + pub stop_timeout: Option, + /// Preset-only `dedup-cache-models`: hide a cache model whose snapshot + /// path this preset also resolves to. + pub dedup_cache_models: bool, + /// The raw `key = value` pairs in file order, kept so the model object's + /// `status.preset` block can reproduce the section as INI text the way + /// b10621 does. + pub raw: Vec<(String, String)>, +} + +impl PresetSection { + /// Merge `other` (the `[*]` global section) underneath `self`: values + /// already present in `self` win, which is upstream's cascade order + /// (global first, then the section's own options overwrite). + fn merge_under_global(&mut self, global: &PresetSection) { + macro_rules! fill { + ($field:ident) => { + if self.$field.is_none() { + self.$field = global.$field.clone(); + } + }; + } + fill!(model_path); + fill!(hf_repo); + fill!(ctx_size); + fill!(n_parallel); + fill!(temperature); + fill!(top_k); + fill!(top_p); + fill!(min_p); + fill!(seed); + fill!(n_predict); + fill!(stop_timeout); + if self.aliases.is_empty() { + self.aliases = global.aliases.clone(); + } + if self.tags.is_empty() { + self.tags = global.tags.clone(); + } + self.load_on_startup |= global.load_on_startup; + self.dedup_cache_models |= global.dedup_cache_models; + } + + /// Render the section back as INI text for the b10621 `status.preset` + /// block (host/port/alias/tags are stripped upstream; mlxcel sections + /// never carry host/port, and alias/tags are stripped here). + pub fn to_ini(&self, name: &str) -> String { + let mut out = format!("[{name}]\n"); + for (key, value) in &self.raw { + if key == "alias" || key == "tags" { + continue; + } + out.push_str(&format!("{key} = {value}\n")); + } + out + } +} + +/// The parsed `--models-preset` file. +#[derive(Debug, Clone, Default)] +pub struct RouterPresets { + /// `[*]`: merged into every model of every source. + pub global: PresetSection, + /// Named sections in file order (BTreeMap: name collisions replace, like + /// upstream's map semantics). + pub models: BTreeMap, +} + +impl RouterPresets { + /// The section that applies to `name`, with the global section merged + /// underneath. For a model with no named section, the global section + /// alone applies. + pub fn for_model(&self, name: &str) -> PresetSection { + let mut section = self.models.get(name).cloned().unwrap_or_default(); + section.merge_under_global(&self.global); + section + } + + pub fn is_empty(&self) -> bool { + self.models.is_empty() && self.global == PresetSection::default() + } +} + +/// Which router CLI flags were explicitly given, so they win over preset +/// values (b10621 overlays the CLI-args preset on top of every model preset). +/// Detected once at startup from the process argument list. +#[derive(Debug, Clone, Copy, Default)] +pub struct PresetCliOverrides { + pub ctx_size: bool, + pub n_parallel: bool, + pub min_p: bool, + pub seed: bool, + pub n_predict: bool, + pub alias: bool, + pub tags: bool, +} + +impl PresetCliOverrides { + /// Detect from the live process arguments. Long-form flags only; the + /// `temperature` / `top-k` / `top-p` flags need no detection here because + /// [`super::ServerStartupConfig`] already carries their `*_was_set` bits. + pub fn detect() -> Self { + use super::long_cli_flag_was_set; + Self { + ctx_size: long_cli_flag_was_set("ctx-size"), + n_parallel: long_cli_flag_was_set("parallel"), + min_p: long_cli_flag_was_set("min-p"), + seed: long_cli_flag_was_set("seed"), + n_predict: long_cli_flag_was_set("n-predict") || long_cli_flag_was_set("predict"), + alias: long_cli_flag_was_set("alias"), + tags: long_cli_flag_was_set("tags"), + } + } +} + +/// Apply `section` onto a per-model startup-config clone, honoring the +/// CLI-wins overlay order. `startup.model_path` is set by the caller to the +/// entry's checkpoint directory before `build_server_config` re-resolves the +/// per-model defaults. +pub fn apply_section_to_startup( + startup: &mut super::ServerStartupConfig, + section: &PresetSection, + cli: &PresetCliOverrides, +) { + if let Some(ctx) = section.ctx_size + && !cli.ctx_size + { + startup.ctx_size = ctx; + } + if let Some(par) = section.n_parallel + && !cli.n_parallel + { + startup.n_parallel = par; + } + if let Some(temp) = section.temperature + && !startup.temperature_was_set + { + startup.temperature = temp; + startup.temperature_was_set = true; + } + if let Some(top_k) = section.top_k + && !startup.top_k_was_set + { + startup.top_k = top_k; + startup.top_k_was_set = true; + } + if let Some(top_p) = section.top_p + && !startup.top_p_was_set + { + startup.top_p = top_p; + startup.top_p_was_set = true; + } + if let Some(min_p) = section.min_p + && !cli.min_p + { + startup.min_p = min_p; + } + if let Some(seed) = section.seed + && !cli.seed + { + startup.seed = u64::try_from(seed).ok(); + } + if let Some(n_predict) = section.n_predict + && !cli.n_predict + { + startup.n_predict = n_predict; + } + if !section.aliases.is_empty() && !cli.alias { + startup.model_aliases = section.aliases.clone(); + startup.model_alias = section.aliases.first().cloned(); + } + if !section.tags.is_empty() && !cli.tags { + startup.tags = Some(section.tags.join(",")); + } +} + +fn parse_truthy(key: &str, section: &str, value: &str) -> anyhow::Result { + match value.to_ascii_lowercase().as_str() { + "1" | "true" | "on" | "yes" | "enabled" => Ok(true), + "0" | "false" | "off" | "no" | "disabled" => Ok(false), + other => anyhow::bail!( + "option '{key}' in preset '[{section}]' expects a boolean, found '{other}'" + ), + } +} + +fn parse_num(key: &str, section: &str, value: &str) -> anyhow::Result { + value.trim().parse::().map_err(|_| { + anyhow::anyhow!("option '{key}' in preset '[{section}]' has an invalid value '{value}'") + }) +} + +fn split_list(value: &str) -> Vec { + value + .split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) + .collect() +} + +/// Canonicalize an INI key: strip leading dashes and map `LLAMA_ARG_*` +/// environment spellings onto the long-option spelling, the same dual +/// key space upstream's `key_to_opt` map accepts. +fn canonical_key(raw: &str) -> String { + let stripped = raw.trim_start_matches('-'); + let lower = stripped.to_ascii_lowercase(); + match stripped { + "LLAMA_ARG_MODEL" => "model".into(), + "LLAMA_ARG_HF_REPO" => "hf-repo".into(), + "LLAMA_ARG_ALIAS" => "alias".into(), + "LLAMA_ARG_TAGS" => "tags".into(), + "LLAMA_ARG_CTX_SIZE" => "ctx-size".into(), + "LLAMA_ARG_N_PARALLEL" => "parallel".into(), + "LLAMA_ARG_N_PREDICT" => "n-predict".into(), + _ => match lower.as_str() { + "m" => "model".into(), + "hf" | "hfr" => "hf-repo".into(), + "a" => "alias".into(), + "c" => "ctx-size".into(), + "np" => "parallel".into(), + "n" => "n-predict".into(), + "temperature" => "temp".into(), + _ => lower, + }, + } +} + +/// Parse one section's key/value pair into `section`. Unknown keys are a +/// hard error, matching upstream's strict `load_from_ini` (and keeping the +/// epic's no-silent-ignore rule: a key mlxcel cannot honor per model must +/// stop startup, not vanish). +fn apply_key( + section: &mut PresetSection, + section_name: &str, + raw_key: &str, + value: &str, +) -> anyhow::Result<()> { + let key = canonical_key(raw_key); + match key.as_str() { + "version" => return Ok(()), // reserved, skipped upstream too + "model" => section.model_path = Some(PathBuf::from(value)), + "hf-repo" => section.hf_repo = Some(value.trim().to_string()), + "alias" => section.aliases = split_list(value), + "tags" => section.tags = split_list(value), + "ctx-size" => section.ctx_size = Some(parse_num(&key, section_name, value)?), + "parallel" => section.n_parallel = Some(parse_num(&key, section_name, value)?), + "temp" => section.temperature = Some(parse_num(&key, section_name, value)?), + "top-k" => section.top_k = Some(parse_num(&key, section_name, value)?), + "top-p" => section.top_p = Some(parse_num(&key, section_name, value)?), + "min-p" => section.min_p = Some(parse_num(&key, section_name, value)?), + "seed" => section.seed = Some(parse_num(&key, section_name, value)?), + "n-predict" => section.n_predict = Some(parse_num(&key, section_name, value)?), + "load-on-startup" => { + section.load_on_startup = parse_truthy(&key, section_name, value)?; + } + "stop-timeout" => section.stop_timeout = Some(parse_num(&key, section_name, value)?), + "dedup-cache-models" => { + section.dedup_cache_models = parse_truthy(&key, section_name, value)?; + } + other => anyhow::bail!( + "option '{other}' in preset '[{section_name}]' is not translated by mlxcel presets \ + (#1438). Remove it from the preset, or configure it through the router's own \ + command line, which applies to every model." + ), + } + section.raw.push((key, value.to_string())); + Ok(()) +} + +/// Parse a b10621 `--models-preset` INI file. +/// +/// Grammar (upstream `common/preset.cpp`): `[name]` section headers, `key = +/// value` pairs, `;` / `#` comment lines, blank lines. Keys before the first +/// header belong to the implicit `default` section, which upstream drops when +/// empty; `[*]` is the global preset merged into every model. +pub fn parse_preset_file(path: &Path) -> anyhow::Result { + let text = std::fs::read_to_string(path).map_err(|err| { + anyhow::anyhow!( + "--models-preset {}: cannot read preset file: {err}", + path.display() + ) + })?; + parse_preset_text(&text) + .map_err(|err| anyhow::anyhow!("--models-preset {}: {err}", path.display())) +} + +/// Parse preset INI text. Split from [`parse_preset_file`] for tests. +pub fn parse_preset_text(text: &str) -> anyhow::Result { + let mut presets = RouterPresets::default(); + let mut current_name = String::from("default"); + let mut current = PresetSection::default(); + + let finish = |name: &str, section: PresetSection, presets: &mut RouterPresets| { + if name == "*" { + presets.global = section; + } else if name == "default" { + // Upstream drops an empty implicit default section; a + // non-empty one is a model named "default". + if section != PresetSection::default() { + presets.models.insert(name.to_string(), section); + } + } else { + presets.models.insert(name.to_string(), section); + } + }; + + for (idx, raw_line) in text.lines().enumerate() { + let line = raw_line.trim(); + if line.is_empty() || line.starts_with(';') || line.starts_with('#') { + continue; + } + if let Some(rest) = line.strip_prefix('[') { + let Some(name) = rest.strip_suffix(']') else { + anyhow::bail!("line {}: malformed section header '{raw_line}'", idx + 1); + }; + let name = name.trim(); + if name.is_empty() { + anyhow::bail!("line {}: empty section name", idx + 1); + } + finish( + &std::mem::replace(&mut current_name, name.to_string()), + std::mem::take(&mut current), + &mut presets, + ); + continue; + } + let Some((key, value)) = line.split_once('=') else { + anyhow::bail!( + "line {}: expected 'key = value', found '{raw_line}'", + idx + 1 + ); + }; + let key = key.trim(); + if key.is_empty() { + anyhow::bail!("line {}: empty option key", idx + 1); + } + apply_key(&mut current, ¤t_name, key, value.trim()) + .map_err(|err| anyhow::anyhow!("line {}: {err}", idx + 1))?; + } + finish(¤t_name, current, &mut presets); + Ok(presets) +} + +#[cfg(test)] +#[path = "router_presets_tests.rs"] +mod router_presets_tests; diff --git a/src/server/router_presets_tests.rs b/src/server/router_presets_tests.rs new file mode 100644 index 000000000..16f940b84 --- /dev/null +++ b/src/server/router_presets_tests.rs @@ -0,0 +1,158 @@ +// Copyright 2025-2026 Lablup Inc. and Jeongkyu Shin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Unit tests for the `--models-preset` INI translation (issue #1438): +//! grammar, key canonicalization, the `[*]` global cascade, the strict +//! unknown-key refusal, and the CLI-wins overlay onto per-model startup +//! configs. + +use super::{PresetCliOverrides, apply_section_to_startup, parse_preset_text}; +use crate::server::ServerStartupConfig; + +#[test] +fn sections_keys_comments_and_globals_parse() { + let presets = parse_preset_text( + "; file comment\n\ + [*]\n\ + ctx-size = 4096\n\ + \n\ + [chat]\n\ + # inline section\n\ + model = /models/chat\n\ + temp = 0.5\n\ + alias = c1,c2\n\ + version = 1\n\ + [coder]\n\ + model = /models/coder\n\ + top-k = 20\n", + ) + .expect("parse"); + assert_eq!(presets.global.ctx_size, Some(4096)); + assert_eq!(presets.models.len(), 2); + let chat = &presets.models["chat"]; + assert_eq!( + chat.model_path.as_deref(), + Some(std::path::Path::new("/models/chat")) + ); + assert_eq!(chat.temperature, Some(0.5)); + assert_eq!(chat.aliases, vec!["c1", "c2"]); + // The global section cascades underneath each model's own options. + let merged = presets.for_model("chat"); + assert_eq!(merged.ctx_size, Some(4096)); + assert_eq!(merged.temperature, Some(0.5)); + // A model with no named section still receives the global preset. + assert_eq!(presets.for_model("unlisted").ctx_size, Some(4096)); +} + +#[test] +fn env_spellings_and_short_forms_canonicalize() { + let presets = parse_preset_text( + "[m]\n\ + LLAMA_ARG_MODEL = /m\n\ + LLAMA_ARG_CTX_SIZE = 2048\n\ + temperature = 0.1\n\ + LLAMA_ARG_ALIAS = a1\n", + ) + .expect("parse"); + let m = &presets.models["m"]; + assert_eq!(m.model_path.as_deref(), Some(std::path::Path::new("/m"))); + assert_eq!(m.ctx_size, Some(2048)); + assert_eq!(m.temperature, Some(0.1)); + assert_eq!(m.aliases, vec!["a1"]); +} + +#[test] +fn unknown_keys_fail_loudly_with_key_and_section() { + let err = parse_preset_text("[m]\nn-gpu-layers = 99\n").expect_err("must refuse"); + let text = err.to_string(); + assert!(text.contains("n-gpu-layers"), "{text}"); + assert!(text.contains("[m]"), "{text}"); +} + +#[test] +fn malformed_lines_and_bad_values_are_errors() { + assert!( + parse_preset_text("[m\nmodel = /m\n").is_err(), + "unclosed header" + ); + assert!( + parse_preset_text("[m]\njust-a-word\n").is_err(), + "no equals" + ); + assert!( + parse_preset_text("[m]\nctx-size = lots\n").is_err(), + "bad number" + ); + assert!( + parse_preset_text("[m]\nload-on-startup = maybe\n").is_err(), + "bad boolean" + ); +} + +#[test] +fn preset_only_options_parse() { + let presets = parse_preset_text( + "[m]\nmodel = /m\nload-on-startup = true\nstop-timeout = 30\ndedup-cache-models = 1\n", + ) + .expect("parse"); + let m = &presets.models["m"]; + assert!(m.load_on_startup); + assert_eq!(m.stop_timeout, Some(30)); + assert!(m.dedup_cache_models); +} + +#[test] +fn apply_overlays_preset_values_where_the_cli_did_not_speak() { + let presets = parse_preset_text( + "[m]\nctx-size = 8192\ntemp = 0.25\ntop-k = 7\nseed = 42\nn-predict = 64\n", + ) + .expect("parse"); + let section = presets.for_model("m"); + let mut startup = ServerStartupConfig::default(); + apply_section_to_startup(&mut startup, §ion, &PresetCliOverrides::default()); + assert_eq!(startup.ctx_size, 8192); + assert_eq!(startup.temperature, 0.25); + assert!(startup.temperature_was_set); + assert_eq!(startup.top_k, 7); + assert_eq!(startup.seed, Some(42)); + assert_eq!(startup.n_predict, 64); +} + +#[test] +fn explicit_cli_flags_win_over_preset_values() { + let presets = parse_preset_text("[m]\nctx-size = 8192\ntemp = 0.25\n").expect("parse"); + let section = presets.for_model("m"); + let mut startup = ServerStartupConfig { + ctx_size: 1024, + temperature: 0.9, + temperature_was_set: true, + ..Default::default() + }; + let cli = PresetCliOverrides { + ctx_size: true, + ..Default::default() + }; + apply_section_to_startup(&mut startup, §ion, &cli); + assert_eq!(startup.ctx_size, 1024, "CLI ctx-size wins"); + assert_eq!(startup.temperature, 0.9, "explicit CLI temp wins"); +} + +#[test] +fn a_negative_seed_means_random_per_request() { + let presets = parse_preset_text("[m]\nseed = -1\n").expect("parse"); + let section = presets.for_model("m"); + let mut startup = ServerStartupConfig::default(); + apply_section_to_startup(&mut startup, §ion, &PresetCliOverrides::default()); + assert_eq!(startup.seed, None); +} diff --git a/src/server/router_server.rs b/src/server/router_server.rs index 2d3b57ad2..8a4fd1f6c 100644 --- a/src/server/router_server.rs +++ b/src/server/router_server.rs @@ -79,6 +79,14 @@ fn pool_error_response(err: RouterPoolError) -> Response { "unavailable_error", &message, ), + // b10621 raises these two as runtime errors, which its handler + // wrapper answers as 500s with the exception text. + RouterPoolError::NotRemovable(name) => llama_server_error(&format!( + "model name={name} is not removable (not from cache)" + )), + RouterPoolError::AlreadyExists(name) => { + llama_invalid_request(&format!("model '{name}' already exists")) + } } } @@ -133,9 +141,12 @@ fn router_model_json( let mut status = serde_json::json!({ "value": snapshot.status.as_str(), // b10621 reports the child process argv; the in-process pool has - // none. + // none (recorded as a by_design divergence on --models-dir). "args": [], }); + if let Some(preset_ini) = &snapshot.preset_ini { + status["preset"] = serde_json::Value::String(preset_ini.clone()); + } if snapshot.failed { status["failed"] = true.into(); } @@ -146,10 +157,10 @@ fn router_model_json( if snapshot.audio { input_modalities.push("audio"); } - serde_json::json!({ + let mut model = serde_json::json!({ "id": snapshot.name, - "aliases": [], - "tags": [], + "aliases": snapshot.aliases, + "tags": snapshot.tags, "object": "model", "owned_by": "llamacpp", "created": created, @@ -158,11 +169,18 @@ fn router_model_json( "input_modalities": input_modalities, "output_modalities": ["text"], }, - "source": "models_dir", - // Only cache-sourced models are removable in b10621; a models-dir - // pool has none. - "can_remove": false, - }) + "source": snapshot.source.as_str(), + // Only cache-sourced models are removable, b10621's own rule. + "can_remove": snapshot.source == super::router_models::RouterModelSource::Cache, + }); + // b10621 reflects download progress through the entry's loaded_info. + if let Some(info) = &snapshot.download_info { + model["progress"] = info + .get("progress") + .cloned() + .unwrap_or(serde_json::Value::Null); + } + model } /// GET /models, GET /v1/models (router list). `?reload=1` rescans the @@ -181,6 +199,8 @@ async fn router_models_list( .pool .snapshot() .iter() + // b10621 skips cache models a preset deduplicated. + .filter(|snapshot| !snapshot.hidden) .map(|snapshot| router_model_json(snapshot, created)) .collect(); Json(serde_json::json!({ "data": data, "object": "list" })).into_response() @@ -225,7 +245,9 @@ async fn router_models_unload( let Some(entry) = state.pool.get(&name) else { return llama_invalid_request("model is not found"); }; - if !entry.is_running() { + // b10621 accepts unload for running AND downloading models (unloading a + // downloading model cancels the download). + if !entry.is_running() && !entry.is_downloading() { return llama_invalid_request("model is not running"); } match state.pool.unload(&name) { @@ -234,8 +256,10 @@ async fn router_models_unload( } } -/// POST /models (b10621 downloads the named HF repo into its cache; mlxcel -/// does not yet, and says so instead of pretending). +/// POST /models (b10621 `post_router_models`): validate the name as a +/// fetchable HuggingFace repository, then download it into the model cache +/// in the background, reporting progress through `GET /models/sse` +/// (issue #1438). async fn router_models_add( State(state): State, body: axum::body::Bytes, @@ -244,18 +268,47 @@ async fn router_models_add( if name.is_empty() { return llama_invalid_request("model must be a non-empty string"); } - if state.pool.get(&name).is_some() { + if state.pool.lookup(&name).is_some() { return llama_invalid_request(&format!("model '{name}' already exists")); } - llama_server_error( - "adding a model by download is not supported yet; place the checkpoint directory under \ - --models-dir and call GET /models?reload=1", - ) + if !state.pool.has_cache() { + return llama_server_error( + "adding a model by download requires a model cache; set --model-store-root, \ + MLXCEL_MODELS_DIR, or MLXCEL_CACHE_DIR", + ); + } + // Normalize into the `/` repo id the cache stores (a bare + // name expands to the default organization, like `-m `). The + // normalized spelling is the entry name the SSE events and the model + // list will carry. + let repo_id = match state.pool.normalize_cache_name(&name) { + Ok(repo_id) => repo_id, + Err(err) => return llama_invalid_request(&err.to_string()), + }; + if repo_id != name && state.pool.lookup(&repo_id).is_some() { + return llama_invalid_request(&format!("model '{repo_id}' already exists")); + } + // b10621 validates by fetching repository metadata before answering; a + // failed probe is a 500 from its handler wrapper. The probe blocks on + // the network, so it runs on the blocking pool. + let probe_pool = state.pool.clone(); + let probe_repo = repo_id.clone(); + let probed = tokio::task::spawn_blocking(move || probe_pool.validate_cache_repo(&probe_repo)) + .await + .unwrap_or_else(|join_err| Err(anyhow::anyhow!(join_err.to_string()))); + if let Err(err) = probed { + return llama_server_error(&format!("model validation failed: {err:#}")); + } + match state.pool.start_download(&repo_id) { + Ok(()) => Json(serde_json::json!({ "success": true })).into_response(), + Err(err) => pool_error_response(err), + } } -/// DELETE /models (b10621 `del_router_models`): only cache-sourced models are -/// removable, and a models-dir pool has none, so the refusals are the whole -/// reachable surface. +/// DELETE /models (b10621 `del_router_models`): only cache-sourced models +/// are removable. Cancels an in-flight download or stops a running +/// instance, deletes the snapshot from disk, and emits `model_remove` +/// (issue #1438). async fn router_models_delete( State(state): State, Query(query): Query>, @@ -267,9 +320,10 @@ async fn router_models_delete( if state.pool.get(&name).is_none() { return llama_server_error(&format!("model name={name} is not found")); } - llama_server_error(&format!( - "model name={name} is not removable (not from cache)" - )) + match state.pool.remove(&name).await { + Ok(()) => Json(serde_json::json!({ "success": true })).into_response(), + Err(err) => pool_error_response(err), + } } /// GET /models/sse (b10621 `get_router_models_sse`): the model-event stream. @@ -362,7 +416,9 @@ async fn dispatch(state: RouterServerState, request: Request) -> Response Ok(entry) => entry, Err(err) => return pool_error_response(err), }; - if autoload && let Err(err) = state.pool.ensure_ready(&name, AUTOLOAD_WAIT).await { + // `resolve` accepts preset aliases; the load path addresses the entry by + // its real name. + if autoload && let Err(err) = state.pool.ensure_ready(&entry.name, AUTOLOAD_WAIT).await { return pool_error_response(err); } @@ -435,34 +491,68 @@ pub fn create_router_app(state: RouterServerState) -> axum::Router { #[path = "router_server_tests.rs"] mod router_server_tests; -/// Run the router server: discover models, build the pool, and serve the -/// b10621 router surface (issue #1438). Reached from -/// [`super::startup::start_server`] when `--models-dir` is set and no model -/// argument was given, exactly b10621's `is_router_server` condition. +/// Run the router server: discover models (cache, `--models-dir`, presets), +/// build the pool, and serve the b10621 router surface (issue #1438). +/// Reached from [`super::startup::start_server`] when `--models-dir` or +/// `--models-preset` is set and no model argument was given. pub async fn run_router_server(startup: super::ServerStartupConfig) -> anyhow::Result<()> { - let models_dir = startup - .router_models_dir - .clone() - .ok_or_else(|| anyhow::anyhow!("router mode requires --models-dir"))?; + let models_dir = startup.router_models_dir.clone(); let api_keys = super::resolve_api_keys(&startup.api_keys, &startup.api_key_files)?; if !api_keys.is_empty() { tracing::info!("API-key authentication enabled ({} keys)", api_keys.len()); } - let base_config = super::startup::build_server_config(&startup, api_keys); + + // b10621 loads INI presets per model; a parse error or untranslatable + // key fails startup rather than serving un-preset models (#1438). + let presets = match startup.models_preset.as_ref() { + Some(path) => super::router_presets::parse_preset_file(path)?, + None => super::router_presets::RouterPresets::default(), + }; + + // The router's model cache is the mlxcel model store (the b10621 cache + // equivalent): removable entries, POST /models downloads into it. + let cache = crate::downloader::models_root(startup.model_store_root.as_deref()).map(|root| { + super::router_cache::CacheSource::new( + root, + Arc::new(super::router_cache::HfRouterDownloader), + ) + }); + + let base_config = super::startup::build_server_config(&startup, api_keys.clone()); + let sources = super::router_models::RouterSources { + models_dir: models_dir.clone(), + cache, + presets, + }; + let cli_overrides = super::router_presets::PresetCliOverrides::detect(); + let models_max = startup.models_max; + let models_autoload = startup.models_autoload; let pool = Arc::new(RouterPool::new( - models_dir.clone(), - base_config.clone(), - startup.models_max, - startup.models_autoload, + sources, + startup.clone(), + api_keys, + cli_overrides, + models_max, + models_autoload, )?); let discovered = pool.snapshot().len(); tracing::info!( - "Router mode: {} models discovered under {} (models_max {}, autoload {})", + "Router mode: {} models discovered (models_dir {}, models_max {}, autoload {})", discovered, - models_dir.display(), - startup.models_max, - startup.models_autoload, + models_dir + .as_ref() + .map(|d| d.display().to_string()) + .unwrap_or_else(|| "disabled".to_string()), + models_max, + models_autoload, ); + // Preset-only `load-on-startup`: begin those loads before serving. + for name in pool.load_on_startup_names() { + tracing::info!("router: load-on-startup '{name}'"); + if let Err(err) = pool.begin_load(&name).await { + tracing::warn!("router: load-on-startup '{name}' failed: {err:?}"); + } + } let state = RouterServerState { pool, config: Arc::new(base_config), diff --git a/src/server/router_server_tests.rs b/src/server/router_server_tests.rs index 5cd0e5868..cbcefb07b 100644 --- a/src/server/router_server_tests.rs +++ b/src/server/router_server_tests.rs @@ -17,7 +17,7 @@ //! contract (missing / unknown / not-loaded model), the router `/props` //! block, and authorization on the management routes. -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; use axum::Router; @@ -26,8 +26,12 @@ use axum::http::{Method, Request, StatusCode, header}; use tower::ServiceExt; use super::{RouterServerState, create_router_app}; +use crate::downloader::DownloadHooks; +use crate::server::ServerStartupConfig; use crate::server::config::ServerConfig; -use crate::server::router_models::RouterPool; +use crate::server::router_cache::{CacheSource, RouterDownloader}; +use crate::server::router_models::{RouterPool, RouterSources}; +use crate::server::router_presets::PresetCliOverrides; fn temp_models_dir(tag: &str) -> PathBuf { let dir = std::env::temp_dir().join(format!( @@ -48,12 +52,74 @@ fn add_fake_model(root: &std::path::Path, name: &str) { std::fs::write(dir.join("config.json"), "{}").expect("config.json"); } -fn router_app_with(root: PathBuf, config: ServerConfig, autoload: bool) -> Router { - let pool = Arc::new(RouterPool::new(root, config.clone(), 4, autoload).expect("pool")); - create_router_app(RouterServerState { +/// Instant local "downloader" for route tests: materializes the snapshot and +/// reports one terminal progress tick. +struct InstantDownloader; + +impl RouterDownloader for InstantDownloader { + fn validate(&self, _repo_id: &str) -> anyhow::Result<()> { + Ok(()) + } + + fn download( + &self, + repo_id: &str, + dest_root: &Path, + hooks: DownloadHooks, + ) -> anyhow::Result<()> { + let dest = dest_root.join(repo_id); + std::fs::create_dir_all(&dest)?; + std::fs::write(dest.join("config.json"), "{}")?; + if let Some(progress) = &hooks.progress { + progress(&format!("https://example.invalid/{repo_id}"), 1, 1); + } + Ok(()) + } +} + +fn router_state_from( + sources: RouterSources, + config: ServerConfig, + autoload: bool, +) -> RouterServerState { + let pool = Arc::new( + RouterPool::new( + sources, + ServerStartupConfig::default(), + config.api_keys.clone(), + PresetCliOverrides::default(), + 4, + autoload, + ) + .expect("pool"), + ); + RouterServerState { pool, config: Arc::new(config), - }) + } +} + +fn router_app_with(root: PathBuf, config: ServerConfig, autoload: bool) -> Router { + let sources = RouterSources { + models_dir: Some(root), + cache: None, + presets: Default::default(), + }; + create_router_app(router_state_from(sources, config, autoload)) +} + +fn router_app_with_cache( + root: PathBuf, + cache_root: PathBuf, + config: ServerConfig, +) -> (Router, RouterServerState) { + let sources = RouterSources { + models_dir: Some(root), + cache: Some(CacheSource::new(cache_root, Arc::new(InstantDownloader))), + presets: Default::default(), + }; + let state = router_state_from(sources, config, true); + (create_router_app(state.clone()), state) } /// Status without reading the body, for endpoints whose body never ends @@ -229,7 +295,8 @@ async fn add_and_delete_answer_their_refusal_surface() { assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(body["error"]["message"], "model 'alpha' already exists"); - let (status, _) = send( + // Without a configured cache, a download request is a server error. + let (status, body) = send( app.clone(), Method::POST, "/models", @@ -237,10 +304,13 @@ async fn add_and_delete_answer_their_refusal_surface() { None, ) .await; - assert_eq!( - status, - StatusCode::INTERNAL_SERVER_ERROR, - "download flow is deferred" + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR); + assert!( + body["error"]["message"] + .as_str() + .unwrap_or_default() + .contains("requires a model cache"), + "{body}" ); let (status, body) = send(app.clone(), Method::DELETE, "/models", "", None).await; @@ -394,3 +464,168 @@ async fn model_names_cannot_be_paths() { ); } } + +// ── Cache-backed routes (#1438: POST /models, DELETE /models, SSE) ────────── + +#[tokio::test] +async fn post_models_downloads_into_the_cache_and_lists_it_removable() { + let root = temp_models_dir("dl-route"); + let cache_root = temp_models_dir("dl-route-cache"); + let (app, state) = router_app_with_cache(root, cache_root.clone(), ServerConfig::default()); + let mut events = state.pool.subscribe(); + + let (status, body) = send( + app.clone(), + Method::POST, + "/models", + r#"{"model":"mlx-community/fresh"}"#, + None, + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["success"], true); + + // Wait for the background download to finish (the fake is instant, but + // it still crosses the spawned task). + tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + let event = events.recv().await.expect("events open"); + if event["event"] == "download_finished" { + assert_eq!(event["model"], "mlx-community/fresh"); + break; + } + } + }) + .await + .expect("download_finished"); + + assert!(cache_root.join("mlx-community/fresh/config.json").is_file()); + let (status, body) = send(app, Method::GET, "/models", "", None).await; + assert_eq!(status, StatusCode::OK); + let entry = body["data"] + .as_array() + .expect("data") + .iter() + .find(|m| m["id"] == "mlx-community/fresh") + .cloned() + .expect("downloaded model listed"); + assert_eq!(entry["source"], "cache"); + assert_eq!(entry["can_remove"], true); +} + +#[tokio::test] +async fn delete_models_removes_a_cache_entry_from_disk() { + let root = temp_models_dir("rm-route"); + let cache_root = temp_models_dir("rm-route-cache"); + add_fake_model(&cache_root.join("mlx-community"), "doomed"); + let (app, state) = router_app_with_cache(root, cache_root.clone(), ServerConfig::default()); + let mut events = state.pool.subscribe(); + + let (status, body) = send( + app.clone(), + Method::DELETE, + "/models?model=mlx-community%2Fdoomed", + "", + None, + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["success"], true); + assert!(!cache_root.join("mlx-community/doomed").exists()); + + let mut saw_remove = false; + while let Ok(event) = events.try_recv() { + if event["event"] == "model_remove" && event["model"] == "mlx-community/doomed" { + saw_remove = true; + } + } + assert!(saw_remove, "model_remove reached the SSE stream"); + + // Removing it again is upstream's not-found 500. + let (status, body) = send( + app, + Method::DELETE, + "/models?model=mlx-community%2Fdoomed", + "", + None, + ) + .await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR); + assert_eq!( + body["error"]["message"], + "model name=mlx-community/doomed is not found" + ); +} + +/// Pins the by_design policy divergence on `POST /models`: a bare, +/// owner-less name expands to mlxcel's default organization (the same +/// expansion `-m ` and `mlxcel download ` apply), so the cache +/// entry lists under the expanded repo id where b10621 would list the +/// verbatim name. +#[tokio::test] +async fn post_models_expands_bare_names_to_the_default_org() { + let root = temp_models_dir("dl-bare"); + let cache_root = temp_models_dir("dl-bare-cache"); + let (app, state) = router_app_with_cache(root, cache_root, ServerConfig::default()); + let mut events = state.pool.subscribe(); + + let (status, body) = send( + app.clone(), + Method::POST, + "/models", + r#"{"model":"Bare-Name-4bit"}"#, + None, + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + let event = events.recv().await.expect("events open"); + if event["event"] == "download_finished" { + assert_eq!(event["model"], "mlx-community/Bare-Name-4bit"); + break; + } + } + }) + .await + .expect("download_finished"); + assert!(state.pool.get("mlx-community/Bare-Name-4bit").is_some()); +} + +/// Dispatch accepts preset aliases end-to-end: the alias resolves to the +/// entry and the load path addresses the entry by its real name (a broken +/// checkpoint therefore answers a load failure, never "not found"). +#[tokio::test] +async fn dispatch_reaches_the_load_path_through_an_alias() { + let checkpoint_root = temp_models_dir("alias-dispatch"); + add_fake_model(&checkpoint_root, "ckpt"); + let ini = format!( + "[aliased-model]\nmodel = {}\nalias = nickname\n", + checkpoint_root.join("ckpt").display() + ); + let presets = crate::server::router_presets::parse_preset_text(&ini).expect("parse"); + let state = router_state_from( + RouterSources { + models_dir: None, + cache: None, + presets, + }, + ServerConfig::default(), + true, + ); + let app = create_router_app(state); + let (status, body) = send( + app, + Method::POST, + "/v1/chat/completions", + r#"{"model":"nickname","messages":[]}"#, + None, + ) + .await; + assert_ne!( + status, + StatusCode::BAD_REQUEST, + "alias must resolve: {body}" + ); + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{body}"); +} diff --git a/src/server/startup.rs b/src/server/startup.rs index bd05ac458..24bf6816e 100644 --- a/src/server/startup.rs +++ b/src/server/startup.rs @@ -57,7 +57,12 @@ struct ResolvedDistributedStartup { pub const MIN_PARALLEL_CONTEXT_SIZE: usize = 512; /// Startup configuration for the server (shared between `mlxcel serve` and `mlxcel-server`). -#[derive(Debug)] +/// +/// `Clone` exists for router mode (#1438): the pool clones this per model, +/// swaps in the model's own path and preset overlay, and re-runs +/// [`build_server_config`] so per-model presets resolve exactly like CLI +/// flags. +#[derive(Debug, Clone)] pub struct ServerStartupConfig { // llama-server b10621 chat-template / reasoning / parsing settings // (issue #1447). @@ -158,6 +163,11 @@ pub struct ServerStartupConfig { pub models_preset: Option, /// `--tags`: comma-separated informational model tags (#1438). pub tags: Option, + /// `--model-store-root`: mlxcel model-store root override (#1438's + /// replacement spelling for the pre-router `--models-dir` meaning). In + /// router mode this store is the b10621 cache source: its snapshots are + /// listed as removable models and `POST /models` downloads into it. + pub model_store_root: Option, /// `--spm-infill`: the Suffix/Prefix/Middle ordering for `POST /infill` /// (#1442). Forwarded to [`super::config::ServerConfig::spm_infill`]. @@ -584,6 +594,7 @@ impl Default for ServerStartupConfig { models_autoload: true, models_preset: None, tags: None, + model_store_root: None, spm_infill: false, embd_normalize: None, embedding_serving_mode: crate::server::config::EmbeddingServingMode::Any, @@ -2105,25 +2116,29 @@ pub async fn start_server(mut startup: ServerStartupConfig) -> Result<()> { ); } - // b10621 `--models-preset` names an INI file of per-model presets. - // mlxcel has no preset translation yet; accepting the file and ignoring - // it would silently serve un-preset models, so it fails loudly (#1438). + // b10621 router mode: no model argument plus `--models-dir` or + // `--models-preset` serves the model-management surface instead of + // loading one checkpoint (#1438). Preset files are parsed and applied + // per model inside the router startup; a parse error or untranslatable + // key fails startup there. + if startup.model_path.as_os_str().is_empty() + && (startup.router_models_dir.is_some() || startup.models_preset.is_some()) + { + return crate::server::router_server::run_router_server(startup).await; + } + + // Outside router mode `--models-preset` steers nothing; accepting and + // ignoring it would be the silent-ignore failure epic #1431 exists to + // remove, so it fails loudly (#1438). if let Some(preset) = startup.models_preset.as_ref() { anyhow::bail!( - "--models-preset {} is accepted but not implemented yet: mlxcel cannot translate \ - llama-server INI presets into per-model configuration. Start the router with \ - --models-dir alone, or configure models through the router's own CLI defaults \ - (#1438 tracks preset support)", + "--models-preset {} applies to the b10621 router server and cannot be combined \ + with a model argument. Drop the model argument to start the router (optionally \ + with --models-dir), or remove --models-preset", preset.display() ); } - // b10621 router mode: no model argument plus `--models-dir` serves the - // model-management surface instead of loading one checkpoint (#1438). - if startup.router_models_dir.is_some() && startup.model_path.as_os_str().is_empty() { - return crate::server::router_server::run_router_server(startup).await; - } - // A sampler-order flag that asks for anything but the fixed b10621 // default chain would silently sample in a different order than the // operator requested; fail here, before the multi-GB model load, since