v0.26.0:0 - disk-driven model menu (scan sparks; recipes; needs-setup)

The dashboard menu is now the set of models actually downloaded on the Sparks, not a hard-coded catalog. models.yaml + overrides are reframed as launch recipes matched to an on-disk model by repo; an on-disk model with no recipe is flagged needs_setup and its launch settings are inferred from its config.json for a one-time operator confirmation (discovery.py). - delete now removes weights AND the menu card (delete_from_disk sweeps all hosts; the delete endpoint resolves keys via the live menu) - new GET /api/models/suggest; /api/models returns the menu + a recipes list (download autocomplete); GET /api/models/disk-status removed - dropped the two legacy Qwen recipes (235B FP8, 2.5 72B) - tests: +test_discovery.py (cache parsing, infer_recipe, build_menu merge)
docs: trim Current state to live status — coordination epic shipped
2026-06-18 11:09:56 -05:00 · 2026-06-18 08:09:59 -05:00 · 2026-06-18 07:58:25 -05:00 · 2026-06-18 07:11:33 -05:00 · 2026-06-18 07:07:08 -05:00 · 2026-06-17 23:11:14 -05:00
23 changed files with 1826 additions and 243 deletions
@@ -33,7 +33,7 @@ Subsystem guidance lives in `docs/guides/` and loads when matching files are tou
 - `image/app/` — FastAPI app (`server.py` entry, routers in sibling modules, `static/` dashboard UI).
 - `package/startos/` — StartOS manifest, interfaces, actions, version + release notes.
- `docs/` — `AUDIO_API.md`, `EMBEDDINGS.md`, `REDACTION_GATEWAY.md` (consumer-facing API refs; update with API changes).
+- `docs/` — `AUDIO_API.md`, `EMBEDDINGS.md`, `REDACTION_GATEWAY.md`, `COORDINATION.md` (consumer-facing API refs; update with API changes).
 - `README.md` (overview), `HANDOFF.md` (fresh-user install guide), `runbook.md` (ops notes), `known-issues.md`, `ROADMAP.md` (longer-term backlog — items move into "Current state" below when picked up).
 ## Conventions
@@ -55,12 +55,16 @@ Subsystem guidance lives in `docs/guides/` and loads when matching files are tou
 ## Current state
- **Live service runs v0.22.0:0** (installed and serving); **v0.23.0:0 is built, committed (`e783653`), tagged, and published to Gitea Releases but its live install is PENDING** — see the P3 line below. Working features: swap dashboard; chat / transcribe / diarize(+chunk) / TTS proxies; embeddings + rerank + hybrid search (Qdrant); `/scrub` + `/rehydrate`; label-merge incl. dual-channel; per-Spark SSH-key copy + WireGuard `VPN <ip>` hardware-card badge; configurable vLLM port (Configure Sparks field, blank ⇒ 8888). Local/fine-tuned model support lands live once v0.23.0:0 installs. Spark 2 audio stack healthy. Security hardening (v0.19.0:0 — shellsafe SSH-injection guard, Qdrant path-injection, same-origin CSRF guard) shipped and stable; evidence in `EVALUATION.md`.
+- **Built, install pending: v0.26.0:0 — disk-driven model menu.** The dashboard now lists what's *actually downloaded* on the Sparks instead of a hard-coded catalog. `models.yaml` + overrides are reframed as **launch recipes** matched to an on-disk model by `repo` (no longer "the menu"); `image/app/discovery.py` does the merge: `build_menu` scans both Sparks (`disk.list_cached_models`, one `du` per host) ∪ recipes; an on-disk model with no recipe is `needs_setup` and `infer_recipe` reads its `config.json` to prefill a one-time setup form (operator confirms; saved to `/data` overrides). Delete now removes weights **and** the card (`delete_from_disk` sweeps all hosts; the delete endpoint resolves keys via the live menu so discovered models are deletable). New `GET /api/models/suggest`; `/api/models` returns the menu + a `recipes` list (download-box autocomplete); `GET /api/models/disk-status` removed (folded into `/api/models`). Dropped the two legacy Qwen recipes (235B FP8, 2.5 72B). Build/typecheck clean; **install (live-service restart) needs go/no-go.** Why a recipe layer survives a "menu = disk" redesign: a folder can't tell you parsers / solo-vs-cluster / MoE backend (Gemma MoE needs `marlin` on GB10) — disk drives *presence*, recipes drive *launch*.
- **matrix-bridge bot tile (done, v0.21.0:1, verified live):** `bot`-kind service tile — status badge from docker-state only (no HTTP port), plus **Update** / Restart / Stop/Start / **View logs**. Code: `app/matrix_bridge.py` + `/api/matrix-bridge/{update,logs}` (update streams; 25-min cap; fail-loud). Driven directly as `modelo` on Spark 2 (**no `sudo -iu`** — spark2 has no passwordless sudo). User is a blank-default Configure-Sparks field (`matrix_bridge_user`); blank → tile hidden (portable). Host reuses `spark2_host` (`192.168.1.87` = the bot's box `spark-32d0`); container/dir/branch are env-overridable defaults. **Load-bearing ops dep:** Update's `git fetch` runs as `modelo`, which needs `modelo`'s `~/.ssh/config` pinning the Gitea deploy key with `IdentitiesOnly yes` — else the wrong key is offered and Gitea denies (publickey). Optional next, only if the bot dev asks: Docker `HEALTHCHECK` for running-but-disconnected detection (spec §Note).
+- **Live: v0.25.0:0** (installed 2026-06-18). The OpenClaw/Johnny-5 coexistence epic is fully shipped & live: configurable `VLLM_PORT` (v0.22, blank ⇒ 8888), local/fine-tuned models (v0.23), configurable topology (v0.24 — `VLLM_CONTAINER`, `DISABLED_SERVICES` hide-list, second-Spark `kind: vllm` monitor), coordination layer (v0.25 — swap reservation lock with `423`-enforced manual-swap pause + `?force=true` Release override, `swap_complete`/`swap_failed` webhook, read-only schedule registry; consumer API in `docs/COORDINATION.md`).
- **Tests:** offline pytest harness in `image/tests/` — `cd image && .venv/bin/python -m pytest` (102 passing). Covers `build_launch_command` (incl. the shell-injection round-trip + local-model bind-mount), the transcript↔diarizer label-merge, the `shellsafe` validators, `matrix_bridge.build_update_command` (+ phase detection), and the configurable-topology layer (`test_topology.py`: `DISABLED_SERVICES` parsing, `vllm_container` override, disabled-service skip in `services_from_settings` + `check_*`, `probe_vllm_endpoint`). Mock-heavy swap/proxy tests deliberately skipped (low ROI). Redaction + live-audio suites remain standalone scripts.
+- **Other live features:** swap dashboard; chat / transcribe / diarize(+chunk) / TTS proxies; embeddings + rerank + hybrid search (Qdrant); `/scrub` + `/rehydrate`; label-merge incl. dual-channel; per-Spark SSH-key copy + WireGuard `VPN <ip>` hardware badge. Security hardening (v0.19 — shellsafe SSH-injection guard, Qdrant path-injection, same-origin CSRF guard) stable (`EVALUATION.md`). Spark 2 audio/embeddings stack healthy.
 - **matrix-bridge bot tile (v0.21.0:1, live):** `bot`-kind tile (docker-state badge; Update/Restart/Stop-Start/View-logs) for the Matrix bot on Spark 2, driven as `modelo` (no `sudo -iu`; blank `matrix_bridge_user` ⇒ tile hidden; host reuses `spark2_host`). Code: `app/matrix_bridge.py` + `/api/matrix-bridge/{update,logs}`. **Load-bearing:** Update's `git fetch` runs as `modelo` and needs `modelo`'s `~/.ssh/config` pinning the Gitea deploy key with `IdentitiesOnly yes` (else publickey denial). Optional next only if the bot dev asks: Docker `HEALTHCHECK`.
 - **Tests:** offline pytest harness in `image/tests/` — `cd image && .venv/bin/python -m pytest` (137 passing). Covers `build_launch_command` (incl. the shell-injection round-trip + local-model bind-mount), the transcript↔diarizer label-merge, the `shellsafe` validators, `matrix_bridge.build_update_command` (+ phase detection), the configurable-topology layer (`test_topology.py`), the coordination layer (`test_coordination.py`: swap-lock lifecycle/expiry/token-auth, schedule-registry CRUD, webhook payload + HMAC signature — `now` is injected into the lock so expiry is tested without sleeping), and the disk-driven menu (`test_discovery.py`: cache-dirname↔repo parsing, the cache-listing parser incl. incomplete-download filtering, and `infer_recipe` family/mode mapping — Qwen3-MoE→flashinfer_cutlass, Gemma-MoE→marlin, vision caps, solo-vs-cluster by size/host-count). The `build_menu` merge + `/api/models/suggest` are exercised by hand against the live cluster (mock-heavy unit tests there would test the mocks). Redaction + live-audio suites remain standalone scripts.
 - **Signal Engine "flakiness":** diagnosed as *not* a server bug — transient 1–4s unresponsiveness while the single GPU is busy. Client-side remedy (in-flight cap 2 / ceiling 3 / retry-on-timeout+503) drafted and **forwarded to that dev (owner confirmed 2026-06-15)**. Awaiting whether they want the measured concurrency knee.
 - **Stance (decided, not built):** no public interface / no API-token auth — LAN + WireGuard/Tailscale split-tunnel only; the CSRF guard covers the browser-driven vector.
 - **Known limits:** `/health` blips while the GPU is busy (mitigated client-side); dual-channel can miss a quiet local word under loud remote bleed; connectivity log misses sub-5s outages between 5s polls; diarizer caps at 4 speakers; matrix-bridge badge won't visibly flip on a fast `docker restart` (status re-checked only after the command returns).
 - **Infra gotcha (safety):** passwordless sudo is NOT configured on spark2 — design unprivileged probes for any Spark feature (the badge uses `ip`, not `sudo wg show`). spark2 sits on the `starttunnel` WireGuard subnet (`10.59.211.6/24`, survives reboot). Owner declined SSH-key rotation after the 2026-06-12 history scrub (only the key *name* leaked) — don't re-flag.
 - **Hosting:** self-hosted Gitea — remote `gitea`, branch `master`, over SSH; push after committing. (Wart: commit `8d839e3` is mislabeled `v0.13.0:4` but contains through v0.18.0:0.)
- **Next — committed 2026-06-17: OpenClaw/Johnny-5 coexistence epic (full plan + design stance in `ROADMAP.md` → "Cluster coordination").** Stance: Spark Control = control plane / GPU arbiter, **not** a job runner; business cron jobs live in separate services that *call* its swap API (swaps are already API-driven via `POST /api/swap`). Sequence: (1) **configurable `VLLM_PORT`** — SHIPPED **v0.22.0:0** (Configure-Sparks field, blank ⇒ 8888; + `_env_int` hardening in `config.py` so a blank/bad port no longer crashes startup, killing a P3 tech-debt item). Committed `136a471`, pushed, tagged `v0.22.0`, rebuilt clean, installed, and **published to the self-hosted Gitea Releases** 2026-06-17 (`make release` → `scripts/gitea-release.sh`, takes `GITEA_URL` + a write token). **Distribution model (decided 2026-06-17):** Gitea Releases + a read-only token the adopter's agent uses to pull the latest s9pk (`GET /api/v1/repos/grant/spark-control/releases/latest` → download the `.s9pk` asset → sideload). Note: Gitea returns `browser_download_url` on its `.local` ROOT_URL, which won't resolve off-LAN — a remote adopter pulls via whatever address reaches the Gitea (the WireGuard IP). (2) **local-path/fine-tuned models** — DONE in tree, staged as **v0.23.0:0** (`ModelDef.local_path` + exactly-one-source validator; swap bind-mounts the dir at the same container path via the launch script's `VLLM_SPARK_EXTRA_DOCKER_ARGS` hook, **no `launch-cluster.sh` change**; "+ Add local model" UI form + `local` badge; `validate_local_path`; disk-delete refused for local; 94 tests pass. Reviewer-agent pass done, findings addressed (path validation + chat-template-location guard folded into the `ModelDef` validator so YAML/override entries are checked too; `_merge_overrides` skips a bad entry instead of failing the whole catalog; `VLLM_SPARK_EXTRA_DOCKER_ARGS` contract documented in `runbook.md`). **Committed `e783653`, tagged `v0.23.0`, built clean, published to Gitea Releases — but `make install` to the live Start9 FAILED: `immense-voyage.local` wasn't resolving via mDNS from the Mac (server up at `192.168.1.72`; `start-cli -H <ip>` reaches it but returns UNAUTHORIZED, auth bound to the registered `.local` host). FINISH-HERE: flush mDNS (`sudo dscacheutil -flushcache && sudo killall -HUP mDNSResponder`) or add a hosts entry, then re-run `cd package && make install`** (details in runbook → "Sideload can't reach the server"). (3) **configurable topology** — DONE in tree, staged as **v0.24.0:0** (built clean, not yet committed/installed). Three optional Configure-Sparks knobs: vLLM container name (`VLLM_CONTAINER`, blank ⇒ `vllm_node`, threaded into the swap log-tail + validator exec via `quote_arg`); "services to hide" (`DISABLED_SERVICES` comma list → `Settings.disabled_services` frozenset, skipped by `services_from_settings`, the `check_*` probes, deep-health `run_all`, and connectivity logging — kills the Parakeet-on-8000 collision); second-Spark vLLM monitor via a `kind: vllm` custom service in `services-overrides.yaml` (`probe_vllm_endpoint` shared with `check_vllm`). `/api/endpoints` gained a `disabled` flag; the health-dot hides when disabled. 102 tests pass (+8 in `test_topology.py`). Swap mechanism deliberately NOT generalized to raw `docker run` (that's coordination, item 4). Install pending — same mDNS situation as v0.23.0. Next: (4) coordination layer (swap lock + swap webhook + schedule visibility) — only when our own automation lands. Still-open older threads: audio concurrency sweep (only if the Signal Engine dev wants the knee; needs a quiet window); optional matrix-bridge Docker `HEALTHCHECK` if the bot dev asks; Parakeet long-audio guard deferred (rationale in ROADMAP).
+- **Design stance (decided):** Spark Control = control plane / GPU arbiter, **not** a job runner; recurring business jobs live in separate services that *call* the swap API (`POST /api/swap`). Full epic history (v0.22→v0.25) is in git log + `ROADMAP.md` → "Cluster coordination".
 - **Usage note (2026-06-18):** owner's daily driver is the solo **Qwen3.6 35B**; the 235B `cluster` models are dormant. Keeping `launch-cluster.sh` (the `eugr/spark-vllm-docker` community standard, mirrors NVIDIA's `dgx-spark-playbooks` Ray+RoCE design) is still correct even single-node — it supplies the maintained, hardware-tuned vLLM images; raw docker would mean DIY image upkeep for no gain. Spark 2 stays the speech/embeddings box regardless.
 - **Next steps (all low-priority / externally gated; P2/P3 tech-debt backlog in `ROADMAP.md`):** (1) raw-`docker run` swap generalization — **DEFERRED** (rationale in ROADMAP; revisit only if an adopter wants Spark Control to *drive*, not just monitor, raw-docker swaps — cleanest fix is the adopter adopting `launch-cluster.sh`). (2) audio concurrency knee — only if the Signal Engine dev wants it (needs a quiet window). (3) matrix-bridge Docker `HEALTHCHECK` — only if the bot dev asks. (4) Parakeet long-audio guard — deferred (rationale in ROADMAP).
@@ -92,7 +92,7 @@ Now that hosts are configured, Show Public Key will give you the paste-ready ins
 From the Spark Control service page, click the Web UI button. You should see:
 - A **top status bar** with the currently loaded LLM (or "no model loaded" if Spark 1's vLLM container is fresh).
- An **LLM tab** with cards for each model in the bundled catalog. Models you've downloaded show "on disk" badges; others show "not downloaded".
+- An **LLM tab** whose cards are the models actually downloaded on your Sparks (the dashboard scans them on load). A model Spark Control doesn't yet know how to launch shows a "needs setup" card; the first switch reads its files, proposes settings, and asks you to confirm once. Use **+ Download a new model** to fetch one — it appears here when it finishes.
 - An **Audio / Speech tab** with health status and Install / Start / Stop / Restart buttons for Parakeet and Kokoro.
 If the dashboard loads and both Spark hardware cards show CPU/RAM/GPU stats, **you're in**.
@@ -112,14 +112,14 @@ Fields: `service` (required), `ok` (required), `source` (optional, free-form), `
 ## Status
-**v0.2.3 / s9pk version 0.13.0:4** — installed and verified on a Start9 server. Five bundled LLMs in the catalog (qwen3-vl, gemma4, qwen36, qwen3-235b-fp8, qwen2.5-72b), plus any custom models added through the UI.
+**s9pk version 0.26.0:0** — installed and verified on a Start9 server. The LLM menu is whatever's downloaded on the Sparks (scanned live, not hard-coded); bundled *launch recipes* (qwen3-vl, gemma4, gemma4-26b, qwen36) tell it how to launch known models, and anything else gets a "needs setup" card that infers + saves its settings on first use.
 ### What v0.2 added on top of v0.1
 - **Service discovery API** (`/api/endpoints`) for other LAN services
 - **Kokoro-82M TTS** replaces Magpie/Riva NIM as the default TTS backend (v0.14.0). Magpie's decoder had a ~30-50% truncation rate on multi-sentence inputs and ate 49 GB of GPU memory; Kokoro is 24/24 reliable at every input length tested, uses 1.3 GB GPU, and renders in ~1s. See HANDOFF.md and the release notes for the migration story.
 - **Always-on services panel** with Start/Stop/Restart for Parakeet + Kokoro, plus per-service host configuration in Configure Sparks (so they can live on Spark 1, Spark 2, or anywhere)
- **Model download** from the dashboard — paste an HF repo, pick solo or cluster, watch percent progress with bytes/rate/ETA. After completion, an "Add to catalog" dialog appears pre-filled.
+- **Model download** from the dashboard — paste an HF repo (with autocomplete for known models), pick solo or cluster, watch percent progress with bytes/rate/ETA. After completion the model appears on the menu automatically; if it's unrecognized, a pre-filled "set up this model" dialog offers to configure it.
 - **spark-vllm-docker update check** — banner shows "N commits behind upstream"; Apply Update runs `git pull && ./build-and-copy.sh -c` over SSH with a streamed log
 - **Per-model Advanced settings** — knobs for max context, GPU memory %, and three optimization toggles (fastsafetensors, prefix caching, FP8 KV cache). Persisted to `/data/models-overrides.yaml` so they survive package updates. Bundled and custom models alike.
 - **Diarization with speaker fingerprints** via Sortformer + TitaNet, exposed at `/api/audio/diarize-chunk` for chunked workflows
@@ -12,10 +12,23 @@ Sequenced:
 1. **Configurable `VLLM_PORT`** — DONE, v0.22.0:0. Field in Configure Sparks (blank ⇒ 8888); numeric-setting parsing hardened so a blank/bad value falls back instead of crashing startup. Was the immediate "vLLM unreachable" bug for an adopter on port 8000.
 2. **Local-path / fine-tuned model support** — DONE, v0.23.0:0. Catalog/`ModelDef` gained `local_path` (exactly one of `repo`/`local_path`); swap bind-mounts the dir into the vLLM container at the same path via the launch script's `VLLM_SPARK_EXTRA_DOCKER_ARGS` hook (no `launch-cluster.sh` change); "+ Add local model" form + `local` badge; disk-delete refused for local models; `validate_local_path` boundary check. His merged `ten31-v2` was the motivating case.
 3. **Configurable topology** — DONE, v0.24.0:0. Three optional Configure-Sparks knobs: vLLM container name (`VLLM_CONTAINER`, blank ⇒ `vllm_node`; threaded through the swap log-tail + pre-flight validator via `quote_arg`); "services to hide" (`DISABLED_SERVICES`, comma list — hidden services show no tile and are skipped by status/deep-health/connectivity probes, killing the Parakeet-on-8000 collision); and a second-Spark vLLM monitor via a `kind: vllm` custom service in `services-overrides.yaml` (read-only tile probed through the shared `probe_vllm_endpoint`). `/api/endpoints` gained a `disabled` flag. Covers report P4/P5/#6. (Generalizing the *swap* mechanism to the adopter's raw `docker run` was deliberately left out — that's coordination, item 4; he swaps via his own crons and uses Spark Control to monitor.)
-4. **Coordination layer** — build when our own automation actually lands (zero value until something other than the dashboard swaps models):
+4. **Coordination layer** — DONE in tree, staged as **v0.25.0:0** (built/typechecked clean; install pending). All three primitives shipped; `image/app/coordination.py` + `docs/COORDINATION.md`. Brought forward 2026-06-17 on request rather than waiting for our own automation.
-   - **Swap lock** with holder + TTL (`POST` / `GET` / `DELETE /api/swap/lock`). An external scheduler acquires it before swapping; the dashboard then refuses manual swaps and shows who holds the GPU and until when. Enforced by the swap path, not advisory.
+   - **Swap lock** with holder + TTL (`POST` / `GET` / `DELETE /api/swap/lock`). Acquire returns a secret token; the swap endpoint refuses any real swap (`423`) that doesn't present it in `X-Swap-Lock-Token`, so the dashboard's manual swap is paused while a scheduler holds it (with a `?force=true` human override). In-memory + TTL-bounded → resets to unlocked on restart; re-acquire with the token extends. Enforced in `post_swap`, not advisory.
-   - **Swap-event webhook** (`swap_complete` / `swap_failed`) to a configurable URL, so downstream consumers update their provider config when the running model changes.
+   - **Swap-event webhook** (`swap_complete` / `swap_failed`) to a configurable URL (Configure-Sparks field), fired from `SwapManager._run` *outside* the swap lock; optional shared secret ⇒ `X-Spark-Signature` HMAC. Fire-and-forget (5 s, no retries); dry runs don't fire.
-   - **Schedule visibility** — read-only view the dashboard surfaces, *registered by* external schedulers (Spark Control does not own the schedule).
+   - **Schedule visibility** — `GET/POST/DELETE /api/schedule`; read-only "Scheduled jobs" dashboard panel, registered by external schedulers. Spark Control stores and displays, never executes.
   - Tests: `image/tests/test_coordination.py` (22 cases — lock lifecycle/expiry/token, the single-read swap gate, schedule CRUD + id validation, webhook payload+signature). Known limit: lock + schedules are in-memory (a restart frees the lock and empties the registry until schedulers re-register) — persist to `/data` only if that bites.
 ### Generalizing the swap mechanism to raw `docker run` — DEFERRED (decided 2026-06-18, research-backed; was item 4's last open thread)
 Our swap drives `~/spark-vllm-docker/launch-cluster.sh` over SSH on Spark 1 (`./launch-cluster.sh stop`, then `[VLLM_SPARK_EXTRA_DOCKER_ARGS=…] ./launch-cluster.sh [--solo ]-d exec vllm serve <model> <args>`, then `docker logs -f` until the ready marker). The OpenClaw adopter launches vLLM with a plain `docker run` instead, so the swap button can't drive his cluster — only monitor it. The portability fix would be a configurable "swap backend": keep `launch-cluster.sh` as the default and add a "bring your own command" mode (operator-authored stop/launch templates in `services-overrides.yaml` with quoted `{model}`/`{container}`/`{port}`/`{extra_args}` substitution; ready-detection unchanged; the vLLM-argparse pre-flight disabled for that backend).
 **Why deferred, not built:**
 - **Raw docker is not an upgrade for *us* — for half our catalog it's impossible.** `launch-cluster.sh` is the `eugr/spark-vllm-docker` community project (de-facto DGX Spark standard; mirrors NVIDIA's own `dgx-spark-playbooks` Ray+RDMA architecture). Its headline job is **multi-node** serving: our 235B `cluster` models (Qwen3-VL 235B, Qwen3 235B) exceed one Spark's 128 GB and *must* shard across both Sparks via Ray over the 200 Gbps ConnectX/RoCE link — plumbing (NCCL/MTU/per-node env) that a single-node `docker run` cannot do. So we keep the helper script; switching our own cluster to raw docker is off the table.
 - **The feature is therefore portability-only** (for differently-wired adopters), and the one known adopter doesn't need it — he swaps via his own crons and uses Spark Control to watch.
 - **Untestable on our hardware** — our cluster uses the helper script, so we can't validate a real raw-docker swap without risking the live vLLM.
 - The one real standing risk is eugr's single-maintainer status; fallback is community forks or migrating to NVIDIA's official `dgx-spark-playbooks` launcher (same design). No reason to switch now.
 **Revisit only if** an adopter explicitly wants Spark Control to *drive* (not just monitor) swaps on a raw-`docker run` cluster. At that point, get their actual working `docker run` command and build the command-template backend to it.
 ## Near term
 - parakeet-asr long-audio memory guard — **deferred 2026-06-15, low priority.** A duration cap on `/v1/audio/diarize`: Sortformer runs the whole file in one pass (`diarizer.py:128-135`) over Spark 2's *shared* 128 GB unified memory (also feeding Kokoro/embeddings/Qdrant), so one giant single file can thrash into swap. **Precautionary — no observed incident**, and the production consumer (Recap Relay) already chunks via `/diarize-chunk` (~5-min, already bounded), so the only exposed path is a consumer POSTing one huge file to the full `/diarize`. When picked up: add a configurable `MAX_DIARIZE_SECONDS` guard in `diarizer.py` right after `duration` is computed (~line 130) → raise → HTTP 413 in `main.py` (mirrors the existing `MAX_UPLOAD_MB` 413); ship via the Reapply-patches action (restarts the live parakeet-asr container → needs go/no-go). Leave transcription out of v1 (upstream/un-patched file; parakeet-TDT handles long audio better). Revisit only if a consumer starts sending long single files.
@@ -0,0 +1,157 @@
 # Cluster coordination through Spark Control (v0.25.0)
 Spark Control is the **GPU arbiter, not a job runner.** Your recurring pipelines
 (model-warming crons, "daily X" generators, batch jobs) live in your own
 services and *drive Spark Control's swap API*. This page documents the safety
 layer around that: a **swap reservation lock**, a **swap-event webhook**, and a
 **read-only schedule registry**.
 If only the dashboard ever swaps models, you don't need any of this — it's for
 when something automated also swaps.
 All endpoints are on the Spark Control host (same LAN/VPN URL as the LLM, audio,
 and embeddings proxies). There is no API-token auth by design (LAN + split-tunnel
 VPN only); a non-browser client passes the same-origin guard automatically.
 ---
 ## 1. Swap reservation lock
 A short, TTL-bounded reservation of the swap path. While a lock is held, **any
 real swap that doesn't present the holder's token is refused with `423 Locked`**
 — including the dashboard's manual swap. The holder *name* is descriptive; the
 returned **token** is the secret that authorises swaps and the release.
 The lock is in-memory: it resets to *unlocked* if Spark Control restarts (the
 safe-for-availability default), and the swap engine's own in-progress guard
 still prevents two swaps running at once.
 ### `POST /api/swap/lock` — acquire (or extend)
 ```json
 // request
 { "holder": "openclaw-daily-vol", "ttl_seconds": 900, "note": "daily vol run" }
 // 200 response
 {
  "held": true,
  "holder": "openclaw-daily-vol",
  "acquired_at": "2026-06-17T12:00:00+00:00",
  "expires_at":  "2026-06-17T12:15:00+00:00",
  "seconds_remaining": 900,
  "note": "daily vol run",
  "token": "a1b2c3…"        // SECRET — store it; needed to swap and to release
 }
 ```
 - `ttl_seconds` is optional (default 900) and clamped to `[1, 86400]`.
 - **`409`** if a *different* holder already holds it (body includes the current
  `lock` state). To **extend** your own lock, POST again with the same `holder`
  **and** your `token` — the token is preserved and the window slides forward.
 ### `GET /api/swap/lock` — status (no token)
 ```json
 { "held": true, "holder": "openclaw-daily-vol", "expires_at": "…", "seconds_remaining": 612, "note": "…" }
 // or
 { "held": false }
 ```
 ### `DELETE /api/swap/lock` — release
 Send your token in the `X-Swap-Lock-Token` header (or `?token=`):
 ```
 DELETE /api/swap/lock
 X-Swap-Lock-Token: a1b2c3…
 ```
 - **`403`** if the token doesn't match. The dashboard's human override is
  `DELETE /api/swap/lock?force=true` (no token).
 ### Swapping while you hold the lock
 Pass the token on the swap call; the dashboard (no token) is then blocked:
 ```
 POST /api/swap
 X-Swap-Lock-Token: a1b2c3…
 { "model_key": "gemma-3-27b" }
 ```
 Recommended scheduler flow: **acquire → swap (with token) → poll `/api/swap/{id}`
 → release**. Always release in a `finally`; if you crash, the TTL frees it.
 > `POST /api/swap/{key}/validate` (pre-flight) and dry-run swaps are **not**
 > blocked by the lock — they don't touch the cluster.
 ---
 ## 2. Swap-event webhook
 Configure a URL in **Configure Sparks → "Swap webhook URL"**. After every real
 swap, Spark Control POSTs:
 ```json
 {
  "event": "swap_complete",   // or "swap_failed"
  "job_id": "1a2b3c4d",
  "model_key": "gemma-3-27b",
  "state": "ready",           // or "failed"
  "returncode": 0,
  "started_at": "2026-06-17T12:00:00+00:00",
  "finished_at": "2026-06-17T12:03:11+00:00",
  "dry_run": false
 }
 ```
 Headers: `X-Spark-Event: swap_complete`. If you set a **webhook secret**, the
 body is signed: `X-Spark-Signature: sha256=<hmac>` (HMAC-SHA256 of the raw body
 with the shared secret). Verify it like:
 ```python
 import hmac, hashlib
 expected = "sha256=" + hmac.new(secret.encode(), raw_body, hashlib.sha256).hexdigest()
 assert hmac.compare_digest(expected, request.headers["X-Spark-Signature"])
 ```
 Delivery is best-effort and fire-and-forget (5 s timeout, no retries) — a
 webhook failure never affects the swap itself. Dry runs don't fire.
 ---
 ## 3. Schedule registry (read-only display)
 So the dashboard can show *what's scheduled to touch the GPU and when*, your
 schedulers register their jobs here. **Spark Control only displays these — it
 never executes them.**
 ### `POST /api/schedule` — register / update
 ```json
 // request (pass a stable `id` to update in place on re-register)
 { "id": "daily-vol", "name": "Daily Vol", "owner": "openclaw",
  "cron": "0 6 * * *", "next_run": "2026-06-18T06:00:00Z",
  "description": "Swaps to the big model, generates the vol report" }
 // response: the stored entry (generates an id if you omit one)
 ```
 `name` is required; `id` (if given) must match `[A-Za-z0-9_.-]` (≤64 chars).
 ### `GET /api/schedule` — list
 ```json
 { "schedules": [ { "id": "daily-vol", "name": "Daily Vol", "owner": "openclaw",
  "cron": "0 6 * * *", "next_run": "…", "description": "…",
  "registered_at": "…", "updated_at": "…" } ] }
 ```
 ### `DELETE /api/schedule/{id}` — deregister
 ```json
 { "deleted": true }
 ```
 The registry is in-memory — re-register your schedules on your own startup so
 they survive a Spark Control restart.
@@ -39,6 +39,7 @@ Two kinds, both run with the `image/.venv` interpreter (system python3 has no de
 ## Layout
 - `image/app/server.py` — FastAPI entry; routers live in sibling modules (`audio_proxy.py`, `llm_proxy.py`, `embeddings_proxy.py`, `redaction_gateway.py`, `swap.py`, `health.py`, `deep_health.py`, `connectivity.py`, …).
 - `image/app/discovery.py` — the disk-driven model menu. `/api/models` lists what's actually downloaded on the Sparks (via `disk.list_cached_models`); `models.yaml`/overrides are *launch recipes* matched by repo, not the menu. An on-disk model with no recipe is `needs_setup` → `infer_recipe` reads its `config.json` to prefill a setup form the operator confirms once.
 - `image/app/static/` — the dashboard UI.
- `image/models.yaml` — vLLM model catalog bundled into the image.
+- `image/models.yaml` — bundled vLLM **launch recipes** (how to launch a known model), NOT the dashboard menu — the menu is the on-disk scan.
 - `image/spark_embed/` — Dockerfile + app for the embeddings container; built ON a Spark (ARM64, NGC PyTorch base — see the audio/cluster rule for NGC torch-pinning caveats).
@@ -103,6 +103,8 @@ class Settings:
    bind_port: int
    open_webui_url: str
    ngc_api_key: str
    swap_webhook_url: str
    swap_webhook_secret: str
    @classmethod
    def from_env(cls) -> "Settings":
@@ -165,6 +167,11 @@ class Settings:
            bind_port=_env_int("BIND_PORT", 9999),
            open_webui_url=_env("OPEN_WEBUI_URL", ""),
            ngc_api_key=_env("NGC_API_KEY", ""),
            # Coordination layer: fire a swap-lifecycle webhook to this URL so
            # downstream consumers re-point their model config on a swap. Blank
            # ⇒ disabled. The optional secret HMAC-signs the body (X-Spark-Signature).
            swap_webhook_url=_env("SWAP_WEBHOOK_URL", ""),
            swap_webhook_secret=_env("SWAP_WEBHOOK_SECRET", ""),
        )
    @property
@@ -0,0 +1,342 @@
 """Cluster-coordination layer: the GPU swap lock, swap-event webhook, and the
 read-only schedule registry.
 Spark Control is the **control plane / GPU arbiter, not a job runner.** Recurring
 business pipelines live in separate services that *call* the swap API. These
 three primitives add the *safety* layer around that:
 - **Swap lock** — a TTL-bounded reservation of the swap path. An external
  scheduler acquires it before swapping; while held by someone else the
  dashboard's manual swap is refused (enforced in the swap endpoint, not
  advisory). Holder name is descriptive; the returned token is the secret that
  authorises a swap or a release.
 - **Webhook** — fires `swap_complete` / `swap_failed` to a configurable URL so
  downstream consumers re-point their provider config when the running model
  changes. Optionally HMAC-signed.
 - **Schedule registry** — a read-only view the dashboard surfaces, *registered
  by* external schedulers. Spark Control stores what it's told; it does not own
  or execute any schedule.
 All state is in-memory (mirroring the swap/download/NIM job managers). On a
 restart the lock resets to *unlocked* — the available-by-default failure mode;
 the swap manager's own in-progress guard still prevents two swaps at once —
 and schedulers re-register their schedules.
 """
 from __future__ import annotations
 import hashlib
 import hmac
 import json
 import logging
 import re
 import uuid
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import Optional
 import httpx
 log = logging.getLogger(__name__)
 # A lock reserves the GPU for a window; clamp the TTL so a buggy client can
 # neither pin the cluster forever nor take a zero-length (useless) lock.
 LOCK_TTL_MIN = 1
 LOCK_TTL_MAX = 86_400      # 24h
 LOCK_TTL_DEFAULT = 900     # 15 min
 # Schedule ids are reflected to the dashboard and used as a URL path segment on
 # delete, so a caller-supplied id is whitelist-checked. Generated ids are hex.
 _SCHEDULE_ID_RE = re.compile(r"^[A-Za-z0-9_.-]{1,64}$")
 def valid_schedule_id(value: str) -> bool:
    """Whitelist check for a caller-supplied schedule id (register and delete)."""
    return bool(_SCHEDULE_ID_RE.match(value or ""))
 def _now() -> datetime:
    return datetime.now(timezone.utc)
 def _iso(dt: datetime) -> str:
    return dt.isoformat()
 # ---------------------------------------------------------------- swap lock ----
 class LockHeld(Exception):
    """The lock is held by a different holder. Carries the public lock state so
    the endpoint can return holder + expiry in the 409 body."""
    def __init__(self, state: dict) -> None:
        self.state = state
        super().__init__("swap lock is held by another holder")
@dataclass
 class LockState:
    holder: str
    token: str
    acquired_at: datetime
    expires_at: datetime
    note: str = ""
    def public(self, now: datetime) -> dict:
        """Token-free view safe to expose on GET / in error bodies."""
        return {
            "held": True,
            "holder": self.holder,
            "acquired_at": _iso(self.acquired_at),
            "expires_at": _iso(self.expires_at),
            "seconds_remaining": max(0, int((self.expires_at - now).total_seconds())),
            "note": self.note,
        }
 class SwapLockManager:
    """In-memory, TTL-bounded reservation of the GPU swap path.
    `now` is injectable on every method purely so the expiry logic is testable
    without sleeping; production calls omit it and get wall-clock UTC.
    """
    def __init__(self) -> None:
        self._lock: Optional[LockState] = None
    def _active(self, now: Optional[datetime] = None) -> Optional[LockState]:
        """The current lock if one is held and unexpired; lazily clears an
        expired lock so it never lingers."""
        now = now or _now()
        if self._lock is not None and self._lock.expires_at <= now:
            self._lock = None
        return self._lock
    def status(self, now: Optional[datetime] = None) -> dict:
        now = now or _now()
        active = self._active(now)
        return active.public(now) if active else {"held": False}
    def acquire(
        self,
        holder: str,
        ttl_seconds: Optional[int] = None,
        note: str = "",
        token: Optional[str] = None,
        *,
        now: Optional[datetime] = None,
    ) -> LockState:
        """Acquire a free lock (new token), or extend one already held by
        presenting its token. A request without the token is refused even if the
        holder name matches — the name is descriptive, the token is the secret.
        """
        now = now or _now()
        holder = (holder or "").strip()
        if not holder:
            raise ValueError("holder is required")
        ttl = ttl_seconds if ttl_seconds is not None else LOCK_TTL_DEFAULT
        try:
            ttl = int(ttl)
        except (TypeError, ValueError):
            ttl = LOCK_TTL_DEFAULT
        ttl = max(LOCK_TTL_MIN, min(LOCK_TTL_MAX, ttl))
        active = self._active(now)
        if active is not None:
            # Held — only the token-holder may extend/re-acquire.
            if not (token and hmac.compare_digest(active.token, token)):
                raise LockHeld(active.public(now))
            self._lock = LockState(
                holder=holder or active.holder,
                token=active.token,
                acquired_at=active.acquired_at,
                expires_at=now + timedelta(seconds=ttl),
                note=note or active.note,
            )
            return self._lock
        self._lock = LockState(
            holder=holder,
            token=uuid.uuid4().hex,
            acquired_at=now,
            expires_at=now + timedelta(seconds=ttl),
            note=note,
        )
        return self._lock
    def verify(self, token: Optional[str], now: Optional[datetime] = None) -> bool:
        """True iff `token` matches the currently-active lock."""
        active = self._active(now)
        return bool(active and token and hmac.compare_digest(active.token, token))
    def is_blocked_by(self, token: Optional[str], now: Optional[datetime] = None) -> Optional[dict]:
        """Single-read swap gate. Returns the public lock state if an active
        lock blocks a swap carrying this token, else None. Does exactly one
        `_active()` read so the decision can't straddle a TTL expiry the way a
        separate status()+verify() pair could (which, at the expiry tick, would
        spuriously refuse a swap that should now be allowed)."""
        now = now or _now()
        active = self._active(now)
        if active is None:
            return None
        if token and hmac.compare_digest(active.token, token):
            return None
        return active.public(now)
    def release(
        self,
        token: Optional[str] = None,
        *,
        force: bool = False,
        now: Optional[datetime] = None,
    ) -> bool:
        """Release the lock. Returns False if nothing was held. Requires the
        matching token unless `force` (the human override from the dashboard)."""
        active = self._active(now)
        if active is None:
            return False
        if not force and not self.verify(token, now):
            raise PermissionError("token does not hold the lock")
        self._lock = None
        return True
 # ----------------------------------------------------------------- webhook ----
 def build_webhook_payload(
    *,
    event: str,
    job_id: str,
    model_key: str,
    state: str,
    returncode: Optional[int],
    started_at: Optional[str],
    finished_at: Optional[str],
    dry_run: bool,
 ) -> dict:
    return {
        "event": event,            # swap_complete | swap_failed
        "job_id": job_id,
        "model_key": model_key,
        "state": state,
        "returncode": returncode,
        "started_at": started_at,
        "finished_at": finished_at,
        "dry_run": dry_run,
    }
 def sign_payload(secret: str, body: bytes) -> str:
    """`X-Spark-Signature` value: sha256 HMAC of the exact JSON body the
    consumer receives, so they can recompute and trust it."""
    return "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
 class WebhookNotifier:
    """Fire-and-forget POST of swap-lifecycle events. A webhook failure is
    logged and swallowed — it must never affect the swap outcome."""
    def __init__(self, url: str, secret: str = "", timeout: float = 5.0) -> None:
        self.url = (url or "").strip()
        self.secret = secret or ""
        self.timeout = timeout
    @property
    def enabled(self) -> bool:
        return bool(self.url)
    async def fire(self, event: str, payload: dict) -> None:
        if not self.enabled:
            return
        body = json.dumps(payload).encode()
        headers = {
            "content-type": "application/json",
            "user-agent": "spark-control-webhook",
            "x-spark-event": event,
        }
        if self.secret:
            headers["x-spark-signature"] = sign_payload(self.secret, body)
        try:
            async with httpx.AsyncClient(timeout=self.timeout) as client:
                await client.post(self.url, content=body, headers=headers)
        except Exception as e:  # noqa: BLE001 — best-effort, never propagate
            log.warning("swap webhook to %s failed: %s", self.url, e)
 # -------------------------------------------------------- schedule registry ----
@dataclass
 class ScheduleEntry:
    id: str
    name: str
    owner: str = ""
    cron: str = ""
    next_run: str = ""
    description: str = ""
    registered_at: str = ""
    updated_at: str = ""
    def public(self) -> dict:
        return {
            "id": self.id,
            "name": self.name,
            "owner": self.owner,
            "cron": self.cron,
            "next_run": self.next_run,
            "description": self.description,
            "registered_at": self.registered_at,
            "updated_at": self.updated_at,
        }
 class ScheduleRegistry:
    """What external schedulers tell us about their cron jobs. Read-only from the
    dashboard's side; Spark Control never executes any of it."""
    def __init__(self) -> None:
        self._items: dict[str, ScheduleEntry] = {}
    def list(self) -> list[dict]:
        return [e.public() for e in self._items.values()]
    def register(
        self,
        *,
        name: str,
        id: Optional[str] = None,
        owner: str = "",
        cron: str = "",
        next_run: str = "",
        description: str = "",
    ) -> ScheduleEntry:
        name = (name or "").strip()
        if not name:
            raise ValueError("name is required")
        if id is not None:
            id = id.strip()
            if id and not valid_schedule_id(id):
                raise ValueError("id must match [A-Za-z0-9_.-] (max 64 chars)")
        ts = _iso(_now())
        existing = self._items.get(id) if id else None
        if existing is not None:
            existing.name = name
            existing.owner = owner.strip()
            existing.cron = cron
            existing.next_run = next_run
            existing.description = description
            existing.updated_at = ts
            return existing
        sid = id or uuid.uuid4().hex[:8]
        entry = ScheduleEntry(
            id=sid,
            name=name,
            owner=owner.strip(),
            cron=cron,
            next_run=next_run,
            description=description,
            registered_at=ts,
            updated_at=ts,
        )
        self._items[sid] = entry
        return entry
    def delete(self, schedule_id: str) -> bool:
        return self._items.pop(schedule_id, None) is not None
@@ -0,0 +1,209 @@
 """Disk-driven model menu + launch-recipe inference.
 The dashboard's model list is whatever is actually downloaded on the Sparks
 (see `disk.list_cached_models`), NOT a hard-coded catalog. The bundled/overridden
 catalog entries are *launch recipes*: matched to an on-disk model by repo, they
 say HOW to launch it. A completed model on disk with no matching recipe shows up
 as `needs_setup` — the first switch reads its `config.json`, proposes a recipe
 (`infer_recipe`) the operator confirms once, and that confirmed recipe is saved
 to /data so it's a normal card from then on.
 Why a recipe layer at all, if the menu is the disk? Because a folder on disk
 doesn't say how to launch it: the per-family parsers (`--reasoning-parser`,
 `--tool-call-parser`), the MoE backend (some Gemma MoE checkpoints need
 `marlin` on GB10), and solo-vs-cluster topology can't be read off a directory.
 We infer a best guess from the model's own config + size, but the operator
 confirms it — a wrong guess is cheap, a wrong launch is not.
 """
 from __future__ import annotations
 import asyncio
 import re
 from .config import Settings
 from .disk import list_cached_models, probe_disk
 from .overrides import extract_knobs_from_args
 # A model whose weights exceed this can't fit one Spark's 128 GB beside a KV
 # cache, so it must shard across both via Ray. A heuristic prefill only — the
 # operator confirms mode in the setup form, so the exact cutoff isn't critical.
 SINGLE_SPARK_BYTES = 115 * 1000 ** 3
 # Generic knob defaults applied to every inferred recipe (the operator can tweak
 # these in the setup form). Family-specific flags (parsers, MoE backend) are
 # layered on separately by `_detect_family`.
 _COMMON_KNOBS = {
    "max_model_len": 32768,
    "gpu_memory_utilization": 0.85,
    "fastsafetensors": True,
    "prefix_caching": True,
    "kv_cache_dtype": "fp8",
 }
 def repo_to_key(repo: str) -> str:
    """Stable, URL-safe menu key for a discovered model with no recipe key yet.
    'RedHatAI/Qwen3.6-35B-A3B-NVFP4' -> 'redhatai-qwen3-6-35b-a3b-nvfp4'. The same
    slug is used by the menu, the setup form, and `_identify_current_model`, so a
    loaded-but-unconfigured model still highlights as active."""
    return re.sub(r"[^a-z0-9_-]+", "-", repo.lower()).strip("-")
 def _detect_family(config: dict) -> tuple[str, list[str], list[str]]:
    """Return (family_label, vllm_flags, capabilities) inferred from config.json.
    Only family-specific, non-knob flags (parsers, MoE backend) go in vllm_flags;
    generic knob defaults are handled by the caller. Best-effort and operator-
    confirmed, so a wrong guess is cheap."""
    arch = " ".join(config.get("architectures") or [])
    mtype = str(config.get("model_type") or "")
    s = (arch + " " + mtype).lower()
    is_moe = (
        "moe" in s
        or any(config.get(k) for k in ("num_experts", "n_routed_experts", "num_local_experts"))
    )
    is_vision = (
        "conditionalgeneration" in s
        or "vision" in s
        or "vlforcausallm" in s
        or "vision_config" in config
        or "image_token_index" in config
    )
    flags: list[str] = []
    caps: list[str] = []
    label = "Generic"
    if mtype.startswith("qwen3") or "qwen3" in s:
        label = "Qwen3 (MoE)" if is_moe else "Qwen3"
        flags.append("--reasoning-parser=qwen3")
        caps.append("reasoning")
        if is_moe:
            flags.append("--moe_backend=flashinfer_cutlass")
    elif "gemma" in s:
        label = "Gemma (MoE)" if is_moe else "Gemma"
        flags += ["--reasoning-parser=gemma4", "--tool-call-parser=gemma4", "--enable-auto-tool-choice"]
        caps += ["reasoning", "tools"]
        if is_moe:
            # The fast flashinfer/CUTLASS FP4 path errors on GB10 for Gemma MoE;
            # marlin is the working fallback (see the Gemma 26B trial notes).
            flags.append("--moe_backend=marlin")
    if is_vision and "vision" not in caps:
        caps.append("vision")
    return label, flags, caps
 def _infer_mode(total_bytes: int, on_host_count: int) -> str:
    """Solo unless the weights are present on both Sparks or too big for one."""
    if on_host_count >= 2 or total_bytes > SINGLE_SPARK_BYTES:
        return "cluster"
    return "solo"
 def infer_recipe(repo: str, config: dict, total_bytes: int, on_host_count: int) -> dict:
    """Propose a launch recipe for a discovered model — prefills the setup form."""
    label, flags, caps = _detect_family(config or {})
    mode = _infer_mode(total_bytes, on_host_count)
    vllm_args = list(flags)
    vllm_args.append("--max-num-batched-tokens=16384")
    knobs = dict(_COMMON_KNOBS)
    if mode == "cluster":
        # Large models shard across both Sparks via Ray; leave more headroom.
        vllm_args += ["-tp=2", "--distributed-executor-backend=ray"]
        knobs["gpu_memory_utilization"] = 0.7
    return {
        "key": repo_to_key(repo),
        "repo": repo,
        "display_name": repo.split("/")[-1],
        "mode": mode,
        "capabilities": caps,
        "vllm_args": vllm_args,
        "knobs": knobs,
        "family": label,
    }
 def _menu_entry_from_recipe(m, *, on_disk: bool, total_bytes: int, per_host: list[dict]) -> dict:
    d = m.model_dump()
    d["effective_knobs"] = {**extract_knobs_from_args(m.vllm_args), **(m.knobs or {})}
    d["needs_setup"] = False
    d["on_disk"] = on_disk
    d["total_bytes"] = total_bytes
    d["per_host"] = per_host
    return d
 async def build_menu(settings: Settings, catalog) -> dict[str, dict]:
    """The disk-driven model menu: every completed model on the Sparks, annotated
    with its launch recipe (matched by repo) or flagged `needs_setup` if none.
    Two SSH scans total (one per Spark), run in parallel — much cheaper than the
    old per-recipe disk probe. A host that errors is skipped, not fatal."""
    hosts = [(settings.spark1_host, settings.spark1_user)]
    if settings.spark2_host:
        hosts.append((settings.spark2_host, settings.spark2_user))
    scans = await asyncio.gather(
        *(list_cached_models(h, u, settings) for h, u in hosts),
        return_exceptions=True,
    )
    by_repo: dict[str, dict] = {}
    for (h, _u), res in zip(hosts, scans):
        if isinstance(res, Exception):
            continue
        for repo, size, complete in res:
            e = by_repo.setdefault(repo, {"total_bytes": 0, "per_host": [], "complete": False})
            e["total_bytes"] += size
            e["per_host"].append({"host": h, "size_bytes": size})
            e["complete"] = e["complete"] or complete
    recipe_by_repo = {m.repo: (k, m) for k, m in catalog.models.items() if m.repo}
    menu: dict[str, dict] = {}
    for repo, info in by_repo.items():
        # Skip half-fetched / corrupt caches (no finished snapshot) — they'd show
        # as broken cards. In-flight downloads surface in the download panel.
        if not info["complete"]:
            continue
        if repo in recipe_by_repo:
            key, m = recipe_by_repo[repo]
            menu[key] = _menu_entry_from_recipe(
                m, on_disk=True, total_bytes=info["total_bytes"], per_host=info["per_host"]
            )
        else:
            key = repo_to_key(repo)
            menu[key] = {
                "display_name": repo.split("/")[-1],
                "repo": repo,
                "local_path": None,
                "size_gb": round(info["total_bytes"] / 1e9, 1),
                "mode": _infer_mode(info["total_bytes"], len(info["per_host"])),
                "capabilities": [],
                "expected_ready_seconds": 300,
                "vllm_args": [],
                "description": None,
                "knobs": None,
                "custom": False,
                "needs_setup": True,
                "effective_knobs": {},
                "on_disk": True,
                "total_bytes": info["total_bytes"],
                "per_host": info["per_host"],
            }
    # Local/fine-tuned recipes live as a directory, not an HF cache entry — probe
    # each by path and include it if present. Their keys are unique catalog keys
    # (and local models carry repo="" per ModelDef), so they never collide with a
    # discovered repo's slug or an HF recipe key above.
    for key, m in catalog.models.items():
        if not m.local_path:
            continue
        st = await probe_disk(m.repo, m.mode, settings, local_path=m.local_path)
        if not st.on_disk:
            continue
        menu[key] = _menu_entry_from_recipe(
            m,
            on_disk=True,
            total_bytes=st.total_bytes,
            per_host=[{"host": r.host, "size_bytes": r.size_bytes} for r in st.per_host if r.on_disk],
        )
    return menu
@@ -10,6 +10,7 @@ model or one tied to an in-flight swap/download.
 """
 from __future__ import annotations
 import asyncio
 import json
 import re
 from dataclasses import dataclass
 from typing import Optional
@@ -36,6 +37,87 @@ def repo_to_cache_dirname(repo: str) -> str:
    return dn
 def cache_dirname_to_repo(dirname: str) -> Optional[str]:
    """Inverse of `repo_to_cache_dirname`: 'models--org--name' -> 'org/name'.
    A repo has exactly one '/', so the org is the first '--'-segment and the name
    is everything after (names may themselves contain single dashes). Returns
    None for anything that isn't a model cache dir."""
    if not dirname.startswith("models--"):
        return None
    parts = dirname[len("models--"):].split("--")
    if len(parts) < 2 or not parts[0] or not parts[1]:
        return None
    return f"{parts[0]}/{'--'.join(parts[1:])}"
 def parse_cache_listing(out: str) -> list[tuple[str, int, bool]]:
    """Parse the 'size|complete|dirname' lines from `list_cached_models`'s scan.
    Returns [(repo, size_bytes, complete), ...], skipping non-model lines. Pure
    function so the parsing is unit-testable without SSH."""
    items: list[tuple[str, int, bool]] = []
    for line in out.splitlines():
        line = line.strip()
        if line.count("|") < 2:
            continue
        size_s, complete_s, dirname = line.split("|", 2)
        repo = cache_dirname_to_repo(dirname.strip())
        if not repo:
            continue
        try:
            size = int(size_s)
        except ValueError:
            size = 0
        items.append((repo, size, complete_s.strip() == "1"))
    return items
 async def list_cached_models(host: str, user: str, settings: Settings) -> list[tuple[str, int, bool]]:
    """Enumerate every Hugging Face model cached on a host: (repo, size_bytes, complete).
    'complete' = the cache has at least one snapshot carrying a config.json (a
    finished download, not a half-fetched/corrupt dir). One SSH round-trip; the
    glob's no-match case is handled by the `[ -d ]` guard."""
    if not host or not user:
        return []
    cmd = (
        'HUB="$HOME/.cache/huggingface/hub"; '
        'for d in "$HUB"/models--*; do '
        '[ -d "$d" ] || continue; '
        'n=$(basename "$d"); '
        'sz=$(du -sb "$d" 2>/dev/null | cut -f1); sz=${sz:-0}; '
        'if ls "$d"/snapshots/*/config.json >/dev/null 2>&1; then c=1; else c=0; fi; '
        'echo "${sz}|${c}|${n}"; '
        'done'
    )
    rc, out, err = await ssh_run(host, user, cmd, settings, timeout=30.0)
    if rc != 0:
        return []
    return parse_cache_listing(out)
 async def read_model_config(host: str, user: str, repo: str, settings: Settings) -> Optional[dict]:
    """Read a cached model's config.json (first snapshot) for launch inference.
    Returns the parsed dict, or None if absent/unreadable. The dirname is
    whitelisted (repo_to_cache_dirname) so it's safe to embed unquoted."""
    if not host or not user:
        return None
    dn = repo_to_cache_dirname(repo)
    cmd = (
        f'D=$(ls -d "$HOME/.cache/huggingface/hub/{dn}/snapshots/"*/ 2>/dev/null | head -1); '
        f'[ -n "$D" ] && cat "${{D}}config.json" 2>/dev/null'
    )
    rc, out, err = await ssh_run(host, user, cmd, settings, timeout=20.0)
    if rc != 0 or not out.strip():
        return None
    try:
        return json.loads(out)
    except (ValueError, TypeError):
        return None
@dataclass
 class HostDiskResult:
    host: str
@@ -159,10 +241,14 @@ async def delete_host(host: str, user: str, repo: str, settings: Settings) -> Ho
    return HostDiskResult(host=host, on_disk=False, size_bytes=freed)
-async def delete_from_disk(repo: str, mode: str, settings: Settings) -> DiskStatus:
+async def delete_from_disk(repo: str, settings: Settings) -> DiskStatus:
-    """rm -rf the model's cache dir on the relevant Sparks. Idempotent."""
+    """rm -rf the model's cache dir on ALL configured Sparks. Idempotent.
    We sweep both Sparks regardless of the model's declared mode: a 'remove from
    disk & menu' must leave nothing behind, and rm of an absent dir reports 0
    bytes freed (FREED 0), so an extra host is harmless."""
    hosts: list[tuple[str, str]] = [(settings.spark1_host, settings.spark1_user)]
-    if mode == "cluster" and settings.spark2_host:
+    if settings.spark2_host:
        hosts.append((settings.spark2_host, settings.spark2_user))
    results = await asyncio.gather(*(delete_host(h, u, repo, settings) for h, u in hosts))
@@ -11,10 +11,12 @@ from typing import Literal
 from .config import Settings
 from .connectivity import get_mac, record_report, record_state, summary as connectivity_summary
 from .coordination import LockHeld, ScheduleRegistry, SwapLockManager, WebhookNotifier, valid_schedule_id
 from .custom_services import add_custom_service, delete_custom_service
 from .audio_proxy import build_router as build_audio_router
 from .deep_health import DeepHealth
-from .disk import delete_from_disk, probe_disk
+from .discovery import build_menu, infer_recipe, repo_to_key
 from .disk import delete_from_disk, probe_host, read_model_config
 from .download import DownloadManager
 from .llm_proxy import build_router as build_llm_router
 from .embeddings_proxy import build_router as build_embeddings_router
@@ -24,7 +26,7 @@ from .health import check_kokoro, check_parakeet, check_vllm, check_embeddings,
 from .matrix_bridge import MatrixBridgeManager
 from .models import ModelDef, load_catalog
 from .nim import SUGGESTED_NIMS, CATALOG_URL, NimManager
-from .overrides import add_custom, delete_custom, extract_knobs_from_args, load_overrides, set_knobs
+from .overrides import add_custom, delete_custom, load_overrides, set_knobs
 from .services import docker_state, run_action, services_from_settings
 from .shellsafe import validate_container, validate_image, validate_repo
 from .speech_models import SpeechModelsManager
@@ -37,7 +39,12 @@ from .wol import send_local_broadcast, send_via_peer
 settings = Settings.from_env()
 catalog = load_catalog(settings.models_yaml)
-swap_manager = SwapManager(settings, catalog)
+# Coordination layer (GPU arbiter): swap-lifecycle webhook, the swap reservation
 # lock, and the read-only schedule registry. See coordination.py.
 swap_webhook = WebhookNotifier(settings.swap_webhook_url, settings.swap_webhook_secret)
 swap_lock = SwapLockManager()
 schedule_registry = ScheduleRegistry()
 swap_manager = SwapManager(settings, catalog, notifier=swap_webhook)
 download_manager = DownloadManager(settings)
 update_manager = UpdateManager(settings)
 hardware_probe = HardwareProbe(settings)
@@ -67,6 +74,10 @@ _CSRF_EXEMPT_PREFIXES = (
    "/api/audio/",        # diarize-chunk / label-merge / transcribe-with-speakers
    "/api/health-event",  # health reports posted by consumer apps
 )
 # Note: the coordination endpoints (/api/swap/lock, /api/schedule) are
 # intentionally NOT exempt. External schedulers are non-browser clients (no
 # Origin header) so they pass the guard already — same as /api/swap — while a
 # malicious page can't drive them from the operator's browser. Don't add them.
@app.middleware("http")
@@ -151,20 +162,65 @@ def _reload_catalog() -> None:
    swap_manager.reload_catalog(catalog)
 def _recipe_summaries() -> list[dict]:
    """Known launch recipes (bundled + saved), for the download panel's autocomplete.
    These are NOT the menu — the menu is what's on disk. This is just the set of
    repos Spark Control already knows how to launch, so the download box can
    suggest them by name without putting phantom cards on the dashboard."""
    out = []
    for m in catalog.models.values():
        if m.repo:
            out.append({"repo": m.repo, "display_name": m.display_name, "mode": m.mode})
    return out
@app.get("/api/models")
 async def get_models() -> dict:
-    out_models: dict[str, dict] = {}
+    """The model menu = what's actually downloaded on the Sparks (one scan per
-    for key, m in catalog.models.items():
+    Spark), each annotated with its launch recipe or flagged `needs_setup`.
-        d = m.model_dump()
+
-        # Always include effective knobs for the UI (defaults from base args + any overrides)
+    Does SSH, so it's the slower of the model endpoints; the front-end calls it on
-        d["effective_knobs"] = {**extract_knobs_from_args(m.vllm_args), **(m.knobs or {})}
+    load, after a swap/download/delete, and on a slow timer — not every poll."""
-        out_models[key] = d
+    if not settings.configured:
        return {"configured": False, "defaults": catalog.defaults.model_dump(), "models": {}, "recipes": []}
    menu = await build_menu(settings, catalog)
    return {
        "configured": True,
        "defaults": catalog.defaults.model_dump(),
-        "models": out_models,
+        "models": menu,
        "recipes": _recipe_summaries(),
    }
@app.get("/api/models/suggest")
 async def suggest_model(repo: str = Query(...)) -> dict:
    """Read a downloaded model's config.json + size and propose a launch recipe.
    Prefills the 'set up this model' form for an on-disk model that has no recipe
    yet. The operator confirms/edits, then POSTs it to /api/models to save."""
    if not settings.configured:
        raise HTTPException(503, "spark1 not configured")
    try:
        validate_repo(repo)
    except ValueError as e:
        raise HTTPException(400, str(e))
    hosts = [(settings.spark1_host, settings.spark1_user)]
    if settings.spark2_host:
        hosts.append((settings.spark2_host, settings.spark2_user))
    # Config from whichever Spark has it; size summed across the Sparks that do.
    sizes = await asyncio.gather(*(probe_host(h, u, repo, settings) for h, u in hosts))
    total = sum(r.size_bytes for r in sizes if r.on_disk)
    on_hosts = sum(1 for r in sizes if r.on_disk)
    config = None
    for (h, u), r in zip(hosts, sizes):
        if r.on_disk:
            config = await read_model_config(h, u, repo, settings)
            if config is not None:
                break
    return infer_recipe(repo, config or {}, total, on_hosts)
 class KnobsBody(BaseModel):
    knobs: dict
@@ -228,71 +284,43 @@ async def del_model(key: str) -> dict:
    return {"ok": True, "key": key}
@app.get("/api/models/disk-status")
 async def get_models_disk_status() -> dict:
    """Probe each catalog model's HF cache on the appropriate Spark(s) in parallel.
    Result is keyed by model key: {on_disk, total_bytes, per_host:[{host,on_disk,size_bytes,error?}]}.
    Designed to be called once on dashboard load; takes ~1–3s depending on Spark count.
    """
    if not settings.configured:
        return {"configured": False, "models": {}}
    keys = list(catalog.models.keys())
    statuses = await asyncio.gather(*(
        probe_disk(
            catalog.models[k].repo,
            catalog.models[k].mode,
            settings,
            local_path=catalog.models[k].local_path,
        )
        for k in keys
    ), return_exceptions=True)
    out: dict[str, dict] = {}
    for k, s in zip(keys, statuses):
        if isinstance(s, Exception):
            out[k] = {"on_disk": False, "total_bytes": 0, "per_host": [], "error": str(s)}
            continue
        out[k] = {
            "on_disk": s.on_disk,
            "total_bytes": s.total_bytes,
            "per_host": [
                {"host": r.host, "on_disk": r.on_disk, "size_bytes": r.size_bytes, **({"error": r.error} if r.error else {})}
                for r in s.per_host
            ],
        }
    return {"configured": True, "models": out}
@app.delete("/api/models/{key}/disk")
 async def del_model_disk(key: str) -> dict:
-    """Delete a model's weights from the Spark filesystem(s). The catalog entry stays.
+    """Remove a model's weights from the Sparks — and thus from the menu, since the
    menu IS the disk. Resolves the key against the live menu, so a discovered
    model (no saved recipe) is deletable too.
    Safety rails:
      - Refuses a local/fine-tuned directory (hand-placed, not re-downloadable).
      - Refuses if the model is currently loaded on vLLM.
-      - Refuses if a swap or download is in flight.
+      - Refuses if a swap or this model's own download is in flight.
-      - Idempotent: if the cache dir is already gone on a host, that host reports 0 bytes freed.
+      - Idempotent across both Sparks: an already-absent cache dir frees 0 bytes.
    """
-    if key not in catalog.models:
+    if not settings.configured:
        raise HTTPException(503, "spark1 not configured")
    menu = await build_menu(settings, catalog)
    entry = menu.get(key)
    if entry is None:
        raise HTTPException(404, f"unknown model: {key}")
    m = catalog.models[key]
    # Never rm a local fine-tune directory from the dashboard — it's irreplaceable
    # training output the user placed by hand, not a re-downloadable HF cache.
-    if m.local_path:
+    if entry.get("local_path"):
        raise HTTPException(
            400,
            "this is a local model; its directory must be managed on the Spark, not deleted from here",
        )
    repo = entry["repo"]
    # Refuse if currently loaded
    try:
        vllm = await check_vllm(settings)
    except Exception:
        vllm = {}
-    if vllm.get("ok") and vllm.get("current_model") == m.repo:
+    if vllm.get("ok") and vllm.get("current_model") == repo:
        raise HTTPException(
            409,
-            f"'{m.display_name}' is the currently loaded model. Switch to a different model first, then try again."
+            f"'{entry['display_name']}' is the currently loaded model. Switch to a different model first, then try again."
        )
    # Refuse if a swap is in flight
@@ -302,10 +330,10 @@ async def del_model_disk(key: str) -> dict:
    # Refuse if a download is in flight for this same repo (a different model's download is fine)
    if download_manager.current_job_id:
        job = download_manager.get(download_manager.current_job_id)
-        if job and job.repo == m.repo:
+        if job and job.repo == repo:
            raise HTTPException(409, "this model is currently downloading; cancel or wait for it to finish")
-    status = await delete_from_disk(m.repo, m.mode, settings)
+    status = await delete_from_disk(repo, settings)
    # Audit log
    record_report(
        f"disk:{key}",
@@ -316,7 +344,7 @@ async def del_model_disk(key: str) -> dict:
    return {
        "ok": True,
        "key": key,
-        "repo": m.repo,
+        "repo": repo,
        "bytes_freed": status.total_bytes,
        "per_host": [
            {"host": r.host, "size_bytes": r.size_bytes, **({"error": r.error} if r.error else {})}
@@ -871,10 +899,13 @@ async def get_status() -> dict:
 def _identify_current_model(repo: str | None) -> str | None:
    if not repo:
        return None
    # A recipe-backed model keys by its recipe key; a discovered model (loaded but
    # not yet set up) keys by the same slug build_menu uses, so it still
    # highlights as the active card.
    for key, m in catalog.models.items():
        if m.repo == repo:
            return key
-    return None
+    return repo_to_key(repo)
 class SwapRequest(BaseModel):
@@ -892,9 +923,21 @@ async def validate_swap(key: str) -> dict:
@app.post("/api/swap")
-async def post_swap(req: SwapRequest) -> dict:
+async def post_swap(req: SwapRequest, request: Request) -> dict:
    if not settings.configured and not req.dry_run:
        raise HTTPException(503, "spark1 not configured")
    # Enforce the swap reservation lock (the GPU arbiter). A held lock blocks any
    # real swap that doesn't present the holder's token in X-Swap-Lock-Token — so
    # an external scheduler that holds the lock can swap, but the dashboard (no
    # token) is refused while someone else holds it. Dry runs don't touch the
    # cluster, so they're exempt.
    if not req.dry_run:
        blocked = swap_lock.is_blocked_by(request.headers.get("x-swap-lock-token"))
        if blocked is not None:
            raise HTTPException(status_code=423, detail={
                "error": "the GPU swap path is reserved by another holder",
                "lock": blocked,
            })
    try:
        job = await swap_manager.trigger(req.model_key, dry_run=req.dry_run)
    except KeyError:
@@ -949,6 +992,89 @@ async def stream_swap(job_id: str):
    return StreamingResponse(gen(), media_type="text/event-stream")
 # ---- Coordination layer: swap lock + schedule registry ----
 # Endpoints are control-surface, not browser-exempt: an external scheduler is a
 # non-browser client (no Origin header) so it passes the CSRF guard already, the
 # same way it calls /api/swap today; the dashboard is same-origin.
 class LockAcquireRequest(BaseModel):
    holder: str
    ttl_seconds: int | None = None
    note: str = ""
    token: str | None = None   # present only to extend an existing hold
@app.post("/api/swap/lock")
 async def acquire_swap_lock(req: LockAcquireRequest) -> dict:
    """Reserve the GPU swap path. Returns a secret token used to swap (header
    X-Swap-Lock-Token) and to release. 409 if held by another holder."""
    try:
        lock = swap_lock.acquire(req.holder, req.ttl_seconds, req.note, token=req.token)
    except ValueError as e:
        raise HTTPException(422, str(e))
    except LockHeld as e:
        raise HTTPException(status_code=409, detail={
            "error": "swap lock is held by another holder",
            "lock": e.state,
        })
    return {**swap_lock.status(), "token": lock.token}
@app.get("/api/swap/lock")
 async def get_swap_lock() -> dict:
    """Public, token-free view of the reservation: held? who? until when?"""
    return swap_lock.status()
@app.delete("/api/swap/lock")
 async def release_swap_lock(request: Request, force: bool = Query(False)) -> dict:
    """Release the reservation. Needs the matching X-Swap-Lock-Token unless
    ?force=true (the human override from the dashboard)."""
    token = request.headers.get("x-swap-lock-token") or request.query_params.get("token")
    try:
        released = swap_lock.release(token, force=force)
    except PermissionError as e:
        raise HTTPException(403, str(e))
    return {"released": released, **swap_lock.status()}
 class ScheduleRequest(BaseModel):
    name: str
    id: str | None = None
    owner: str = ""
    cron: str = ""
    next_run: str = ""
    description: str = ""
@app.get("/api/schedule")
 async def list_schedules() -> dict:
    return {"schedules": schedule_registry.list()}
@app.post("/api/schedule")
 async def register_schedule(req: ScheduleRequest) -> dict:
    """Register (or update, by id) a schedule an external scheduler owns. Spark
    Control only stores it for the dashboard — it never executes it."""
    try:
        entry = schedule_registry.register(
            name=req.name, id=req.id, owner=req.owner,
            cron=req.cron, next_run=req.next_run, description=req.description,
        )
    except ValueError as e:
        raise HTTPException(422, str(e))
    return entry.public()
@app.delete("/api/schedule/{schedule_id}")
 async def delete_schedule(schedule_id: str) -> dict:
    # Whitelist the path segment at the boundary (repo convention), even though
    # it's only ever a dict key — keeps it from being reflected or logged raw.
    if not valid_schedule_id(schedule_id):
        raise HTTPException(422, "invalid schedule id")
    return {"deleted": schedule_registry.delete(schedule_id)}
 class DownloadRequest(BaseModel):
    repo: str
    mode: Literal["spark1", "spark2", "cluster"] = "spark1"
@@ -19,13 +19,21 @@ const state = {
  configured: true,
  timer_handle: null,
  deep_health: {},
-  disk_status: {},         // keyed by model key: { on_disk, total_bytes, per_host }
+  models_loaded: false,    // true once the first disk scan (/api/models) returns
-  disk_status_loaded: false,
+  recipes: [],             // known launch recipes (for the download autocomplete)
  lock: { held: false },   // GPU swap reservation (coordination layer)
  schedules: [],           // schedules external automation has registered
 };
 const el = (sel) => document.querySelector(sel);
 const $$ = (sel) => document.querySelectorAll(sel);
 // ISO timestamp -> local clock string (e.g. "2:45:10 PM"); '' if unparseable.
 function fmtClock(iso) {
  const t = Date.parse(iso);
  return isNaN(t) ? '' : new Date(t).toLocaleTimeString();
 }
 function escapeHtml(s) {
  if (s == null) return '';
  return String(s)
@@ -51,65 +59,75 @@ function renderCards() {
  const root = el('#cards');
  root.innerHTML = '';
  const isSwapping = !!state.swap_job_id;
-  for (const key of Object.keys(state.models)) {
+  // GPU reserved by external automation — manual swaps are refused server-side
  // (423); reflect that in the buttons so the click never bounces.
  const locked = !!(state.lock && state.lock.held);
  const lockTip = locked
    ? `Reserved by ${state.lock.holder || 'automation'}${state.lock.expires_at ? ' until ' + fmtClock(state.lock.expires_at) : ''}`
    : '';
  const keys = Object.keys(state.models);
  if (keys.length === 0) {
    // The menu is the disk: nothing downloaded (or the scan hasn't returned yet).
    root.innerHTML = state.models_loaded
      ? `<div class="empty-menu muted">No models downloaded on the Sparks yet. Use <strong>+ Download a new model</strong> above to fetch one — it'll appear here when it's done.</div>`
      : `<div class="empty-menu muted">Scanning the Sparks for downloaded models…</div>`;
    return;
  }
  for (const key of keys) {
    const m = state.models[key];
    const isActive = key === state.current_model_key;
    const card = document.createElement('div');
-    card.className = 'card' + (isActive ? ' active' : '');
+    card.className = 'card' + (isActive ? ' active' : '') + (m.needs_setup ? ' needs-setup' : '');
    const desc = m.description
      ? `<div class="desc">${escapeHtml(m.description)}</div>`
      : '';
    const customPill = m.custom ? `<span class="tag custom-pill">custom</span>` : '';
    const localPill = m.local_path ? `<span class="tag local-pill" title="Served from a directory on the Spark, not Hugging Face">local</span>` : '';
-    // Disk-presence pill + trash button. Until /api/models/disk-status comes back,
+    // Every card on the menu is on disk by definition — show its real size.
-    // we don't know — render a neutral placeholder.
+    const gb = (m.total_bytes || 0) / 1e9;
-    const disk = state.disk_status[key];
+    const diskPill = gb > 0
-    let diskPill = '';
+      ? `<span class="tag on-disk" title="Weights present on the Spark(s)">on disk · ${gb.toFixed(1)} GB</span>`
-    if (state.disk_status_loaded) {
+      : '';
-      if (disk && disk.on_disk) {
+    const setupPill = m.needs_setup
-        const gb = (disk.total_bytes / 1e9);
+      ? `<span class="tag setup-pill" title="On disk, but Spark Control hasn't been told how to launch it">needs setup</span>`
-        diskPill = `<span class="tag on-disk" title="Weights present on disk">on disk · ${gb.toFixed(1)} GB</span>`;
+      : '';
-      } else {
+    // Trash = remove weights from disk AND from the menu. Disabled if active / mid-swap.
        diskPill = `<span class="tag not-on-disk" title="Weights not downloaded">not downloaded</span>`;
      }
    }
    // Trash button — hidden if not on disk; disabled (with tooltip) if currently loaded.
    // Never offered for local models: their directory is hand-placed training output,
    // not a re-downloadable HF cache (the server refuses the delete too).
    let trashBtn = '';
-    if (state.disk_status_loaded && disk && disk.on_disk && !m.local_path) {
+    if (!m.local_path) {
      const disabled = isActive || isSwapping;
      const tip = isActive
        ? 'Currently loaded — switch to another model first'
        : isSwapping
        ? 'A swap is in progress'
-        : 'Delete weights from disk';
+        : 'Remove weights from disk & menu';
-      trashBtn = `<button class="icon-btn danger" data-disk-del-key="${key}" title="${escapeHtml(tip)}" aria-label="Delete from disk" ${disabled ? 'disabled' : ''}>${trashIcon}</button>`;
+      trashBtn = `<button class="icon-btn danger" data-disk-del-key="${key}" title="${escapeHtml(tip)}" aria-label="Remove from disk and menu" ${disabled ? 'disabled' : ''}>${trashIcon}</button>`;
    }
-    // Primary card action: "Switch to this" (green) when on disk; "Download" (blue) when not.
+    // Primary action: "Current" / "Switch to this", or "Set up & switch" for a
-    // Before disk-status loads we render the swap button as a sensible default.
+    // model on disk that has no launch recipe yet.
-    const isOnDisk = !state.disk_status_loaded || (disk && disk.on_disk);
+    const swapBlocked = isSwapping || locked;
-    const dlInFlight = !!(typeof dlState !== 'undefined' && dlState && dlState.job_id);
+    const lockTipAttr = locked ? ` title="${escapeHtml(lockTip)}"` : '';
    let primaryBtn = '';
    if (isActive) {
      primaryBtn = `<button class="btn" disabled>Current</button>`;
-    } else if (isOnDisk) {
+    } else if (m.needs_setup) {
-      primaryBtn = `<button class="btn primary" data-swap-key="${key}" ${isSwapping ? 'disabled' : ''}>Switch to this</button>`;
+      primaryBtn = `<button class="btn primary" data-setup-key="${key}"${lockTipAttr} ${swapBlocked ? 'disabled' : ''}>Set up &amp; switch</button>`;
    } else if (m.local_path) {
      // A local model can't be "downloaded" — its directory has to exist on the Spark.
      primaryBtn = `<button class="btn" disabled title="Directory not found on the Spark — create it there, then refresh">Not found on Spark</button>`;
    } else {
-      const tip = dlInFlight ? 'A download is already in progress' : 'Download weights to the Spark(s)';
+      primaryBtn = `<button class="btn primary" data-swap-key="${key}"${lockTipAttr} ${swapBlocked ? 'disabled' : ''}>Switch to this</button>`;
      primaryBtn = `<button class="btn info" data-download-key="${key}" title="${escapeHtml(tip)}" ${dlInFlight ? 'disabled' : ''}>Download</button>`;
    }
    // The Test/Advanced controls need a saved recipe; hide them until setup is done.
    const recipeActions = m.needs_setup ? '' : `
        <button class="btn test-btn" data-test-key="${key}" title="Pre-flight check the launch command without starting the engine">Test</button>
        <button class="btn adv-btn" data-adv-key="${key}" title="Advanced settings">Advanced</button>`;
    card.innerHTML = `
      <div class="name">${escapeHtml(m.display_name)}</div>
      <div class="meta">
        <span class="tag mode-${m.mode}">${m.mode}</span>
-        <span class="tag">${m.size_gb} GB</span>
+        ${diskPill}
        ${setupPill}
        ${customPill}
        ${localPill}
        ${diskPill}
        ${(m.capabilities || []).map(c => `<span class="tag cap">${escapeHtml(c)}</span>`).join('')}
      </div>
      ${desc}
@@ -120,9 +138,7 @@ function renderCards() {
      </div>
      <div class="spacer"></div>
      <div class="card-actions">
-        ${primaryBtn}
+        ${primaryBtn}${recipeActions}
        <button class="btn test-btn" data-test-key="${key}" title="Pre-flight check the launch command without starting the engine">Test</button>
        <button class="btn adv-btn" data-adv-key="${key}" title="Advanced settings">Advanced</button>
        ${trashBtn}
      </div>
      <div class="test-result hidden" data-test-result-for="${key}"></div>
@@ -132,8 +148,8 @@ function renderCards() {
  for (const btn of root.querySelectorAll('[data-swap-key]')) {
    btn.addEventListener('click', () => triggerSwap(btn.dataset.swapKey));
  }
-  for (const btn of root.querySelectorAll('[data-download-key]')) {
+  for (const btn of root.querySelectorAll('[data-setup-key]')) {
-    btn.addEventListener('click', () => triggerDownloadForKey(btn.dataset.downloadKey));
+    btn.addEventListener('click', () => openSetupForKey(btn.dataset.setupKey));
  }
  for (const btn of root.querySelectorAll('[data-adv-key]')) {
    btn.addEventListener('click', () => openAdvanced(btn.dataset.advKey));
@@ -1154,24 +1170,44 @@ async function pollStatus() {
  }
 }
 let menuLoadInFlight = false;
 async function loadModels() {
-  const data = await fetchJSON('/api/models');
+  // The menu is whatever's downloaded on the Sparks — /api/models does the scan
-  state.defaults = data.defaults || {};
+  // (SSH), so this is the slower model call. Best-effort: a transient failure
-  state.models = data.models || {};
+  // leaves the previous menu in place rather than blanking the dashboard.
  // Guard against overlap: init() fires this un-awaited and pollStatus()'s
  // empty-menu fallback may call it again before the scan returns.
  if (menuLoadInFlight) return;
  menuLoadInFlight = true;
  try {
    const data = await fetchJSON('/api/models');
    state.defaults = data.defaults || {};
    state.models = data.models || {};
    state.recipes = data.recipes || [];
    state.models_loaded = true;
    populateDownloadSuggestions();
    renderCards();
  } catch (e) {
    console.warn('model menu load failed:', e.message);
  } finally {
    menuLoadInFlight = false;
  }
 }
-async function loadDiskStatus() {
+// Populate the download box's autocomplete with known recipes not currently on
-  // Probes each catalog model's HF cache over SSH; takes a beat. Best-effort.
+// disk — so common/bundled models stay discoverable without phantom menu cards.
-  try {
+function populateDownloadSuggestions() {
-    const r = await fetchJSON('/api/models/disk-status');
+  const dl = el('#dl-suggestions');
-    if (r && r.models) {
+  if (!dl) return;
-      state.disk_status = r.models;
+  const onDiskRepos = new Set(Object.values(state.models).map(m => m.repo).filter(Boolean));
-      state.disk_status_loaded = true;
+  dl.innerHTML = '';
-      renderCards();
+  for (const r of state.recipes || []) {
-    }
+    if (onDiskRepos.has(r.repo)) continue;
-  } catch (e) {
+    const opt = document.createElement('option');
-    // Silent — pills just won't render. Don't block dashboard.
+    opt.value = r.repo;
-    console.warn('disk-status probe failed:', e.message);
+    opt.label = `${r.display_name} (${r.mode})`;
    dl.appendChild(opt);
  }
 }
@@ -1185,14 +1221,12 @@ function fmtBytesShort(n) {
 function openDiskDeleteDialog(key) {
  const m = state.models[key];
-  const disk = state.disk_status[key];
+  if (!m || !m.on_disk) return;
  if (!m || !disk || !disk.on_disk) return;
  const dlg = el('#disk-delete-dialog');
-  el('#dd-summary').innerHTML = `Free <strong>${fmtBytesShort(disk.total_bytes)}</strong> by removing <strong>${escapeHtml(m.display_name)}</strong> (<code>${escapeHtml(m.repo)}</code>) from disk.`;
+  el('#dd-summary').innerHTML = `Free <strong>${fmtBytesShort(m.total_bytes)}</strong> by removing <strong>${escapeHtml(m.display_name)}</strong> (<code>${escapeHtml(m.repo)}</code>) from the Sparks. This also takes it off the menu.`;
  const hostsEl = el('#dd-hosts');
  hostsEl.innerHTML = '';
-  for (const h of (disk.per_host || [])) {
+  for (const h of (m.per_host || [])) {
    if (!h.on_disk) continue;
    const li = document.createElement('li');
    li.innerHTML = `<code>${escapeHtml(h.host)}</code> — ${fmtBytesShort(h.size_bytes)}`;
    hostsEl.appendChild(li);
@@ -1211,20 +1245,19 @@ function openDiskDeleteDialog(key) {
    try {
      const r = await fetchJSON(`/api/models/${encodeURIComponent(key)}/disk`, { method: 'DELETE' });
      dlg.close();
-      // Optimistically clear local disk state for this key, then refresh.
+      // Optimistically drop the card, then re-scan the menu (it's gone from disk).
-      delete state.disk_status[key];
+      delete state.models[key];
      renderCards();
-      // Eagerly re-probe so size is accurate (and shows "not downloaded" pill).
+      await loadModels();
      loadDiskStatus();
      const freed = r && typeof r.bytes_freed === 'number' ? fmtBytesShort(r.bytes_freed) : '';
-      console.log(`Deleted ${m.display_name} from disk${freed ? ` — freed ${freed}` : ''}.`);
+      console.log(`Removed ${m.display_name} from disk${freed ? ` — freed ${freed}` : ''}.`);
    } catch (e) {
      errEl.textContent = e.message || 'Delete failed';
      errEl.classList.remove('hidden');
    } finally {
      confirm.disabled = false;
      cancel.disabled = false;
-      confirm.textContent = 'Delete from disk';
+      confirm.textContent = 'Remove from disk & menu';
    }
  };
  cancel.onclick = onCancel;
@@ -1234,6 +1267,11 @@ function openDiskDeleteDialog(key) {
 async function triggerSwap(modelKey) {
  if (state.swap_job_id) return;
  if (state.lock && state.lock.held) {
    const until = state.lock.expires_at ? ' until ' + fmtClock(state.lock.expires_at) : '';
    alert(`The GPU swap path is reserved by ${state.lock.holder || 'automation'}${until}. Use "Release" on the reservation banner to override.`);
    return;
  }
  try {
    const r = await fetchJSON('/api/swap', {
      method: 'POST',
@@ -1242,42 +1280,84 @@ async function triggerSwap(modelKey) {
    });
    attachToSwap(r.job_id, /*needsBackfill=*/false);
  } catch (e) {
-    alert('Failed to start swap: ' + e.message);
+    // 423 Locked: a reservation was acquired between our last poll and this click.
    if (e.message && e.message.startsWith('423')) {
      alert('The GPU swap path was just reserved by automation. Refreshing…');
      pollCoordination();
    } else {
      alert('Failed to start swap: ' + e.message);
    }
  }
 }
-async function triggerDownloadForKey(modelKey) {
+// ---- coordination layer: swap lock + schedule registry ----
-  const m = state.models[modelKey];
+
-  if (!m) return;
+async function pollCoordination() {
  if (dlState.job_id) {
    alert('A download is already in progress; wait for it to finish.');
    return;
  }
  // Pick the download target from the model's mode:
  //   solo    -> spark1 only
  //   cluster -> both Sparks (fetch on Spark 1, rsync to Spark 2 in parallel)
  const dlMode = m.mode === 'cluster' ? 'cluster' : 'spark1';
  const sizeNote = m.size_gb ? ` (~${m.size_gb} GB)` : '';
  const target = m.mode === 'cluster' ? 'both Sparks' : 'Spark 1';
  if (!confirm(`Download "${m.display_name}"${sizeNote} to ${target}? Large models can take a while; you can watch progress in the download panel.`)) {
    return;
  }
  dlState.last_repo = m.repo;
  dlState.last_mode = dlMode;
  try {
-    const r = await fetchJSON('/api/download', {
+    state.lock = await fetchJSON('/api/swap/lock');
-      method: 'POST',
+  } catch { state.lock = { held: false }; }
-      headers: { 'content-type': 'application/json' },
+  try {
-      body: JSON.stringify({ repo: m.repo, mode: dlMode }),
+    const r = await fetchJSON('/api/schedule');
-    });
+    state.schedules = r.schedules || [];
-    // Open the download panel + attach to progress stream
+  } catch { state.schedules = []; }
-    openDownloadForm();
+  renderLockBanner();
-    attachToDownload(r.job_id);
+  renderSchedules();
-  } catch (e) {
+  renderCards();  // reflect lock state on the swap buttons
-    alert('Failed to start download: ' + e.message);
+}
 function renderLockBanner() {
  const banner = el('#lock-banner');
  if (!banner) return;
  const lock = state.lock;
  if (lock && lock.held) {
    const until = lock.expires_at ? ` until ${fmtClock(lock.expires_at)}` : '';
    const note = lock.note ? ` — ${escapeHtml(lock.note)}` : '';
    el('#lock-text').innerHTML =
      `GPU swap path reserved by <strong>${escapeHtml(lock.holder || 'automation')}</strong>${until}${note}. Manual swaps are paused.`;
    banner.classList.remove('hidden');
  } else {
    banner.classList.add('hidden');
  }
 }
 function renderSchedules() {
  const panel = el('#schedule-panel');
  const list = el('#schedule-list');
  if (!panel || !list) return;
  const items = state.schedules || [];
  if (!items.length) {
    panel.classList.add('hidden');
    list.innerHTML = '';
    return;
  }
  list.innerHTML = items.map((s) => {
    const meta = [
      s.cron ? `<code>${escapeHtml(s.cron)}</code>` : '',
      s.next_run ? `next: ${escapeHtml(s.next_run)}` : '',
      s.owner ? `by ${escapeHtml(s.owner)}` : '',
    ].filter(Boolean).join(' · ');
    const desc = s.description ? `<div class="desc">${escapeHtml(s.description)}</div>` : '';
    return `<div class="schedule-item">
      <div class="name">${escapeHtml(s.name)}</div>
      <div class="muted small">${meta}</div>
      ${desc}
    </div>`;
  }).join('');
  panel.classList.remove('hidden');
 }
 async function releaseLock() {
  const lock = state.lock || {};
  const who = lock.holder || 'automation';
  if (!confirm(`Force-release the GPU reservation held by ${who}? Any job relying on it may then collide with a manual swap.`)) return;
  try {
    await fetchJSON('/api/swap/lock?force=true', { method: 'DELETE' });
  } catch (e) {
    alert('Failed to release: ' + e.message);
  }
  pollCoordination();
 }
 async function attachToSwap(jobId, needsBackfill) {
  if (state.swap_eventsource) {
    state.swap_eventsource.close();
@@ -1508,12 +1588,14 @@ function handleDownloadDone(d) {
    el('#dl-title').textContent = 'Done';
    el('#dl-phase').textContent = 'Done ✓';
    el('#dl-progress-fill').style.width = '100%';
-    // Offer to add to catalog
+    // The new model now appears on the menu (the menu is the disk). If it matched
    // a known recipe it's ready to switch to; if not, offer to set it up.
    const repo = dlState.last_repo;
-    const mode = dlState.last_mode;
+    loadModels().then(() => {
-    if (repo) {
+      if (!repo) return;
-      setTimeout(() => openCatalogDialog(repo, mode), 600);
+      const entry = Object.values(state.models).find(m => m.repo === repo);
-    }
+      if (entry && entry.needs_setup) setTimeout(() => openSetupDialog(repo, { thenSwap: false }), 600);
    });
  }
  dlState.job_id = null;
 }
@@ -1626,21 +1708,67 @@ function openAdvanced(key) {
  dlg.showModal();
 }
-function openCatalogDialog(repo, mode) {
+// Context carried from openSetupDialog -> the submit handler: the inferred
 // launch flags (parsers/MoE backend) and whether to swap right after saving.
 let setupCtx = { key: '', repo: '', vllm_args: [], thenSwap: false };
 // "Set up & switch" on a needs-setup card.
 async function openSetupForKey(key) {
  const m = state.models[key];
  if (!m) return;
  if (state.lock && state.lock.held) {
    const until = state.lock.expires_at ? ' until ' + fmtClock(state.lock.expires_at) : '';
    alert(`The GPU swap path is reserved by ${state.lock.holder || 'automation'}${until}. Use "Release" on the reservation banner to override.`);
    return;
  }
  await openSetupDialog(m.repo, { thenSwap: true });
 }
 // Open the "set up this model" dialog, prefilled from inference (config.json +
 // size). The operator confirms once; on save the recipe persists and (if
 // thenSwap) we switch to it.
 async function openSetupDialog(repo, opts = {}) {
  const dlg = el('#catalog-dialog');
-  const key = repo.split('/').pop().toLowerCase().replace(/[^a-z0-9_-]/g, '-');
+  let sug = null;
-  el('#cd-key').value = key;
+  try {
-  el('#cd-name').value = repo.split('/').pop();
+    sug = await fetchJSON(`/api/models/suggest?repo=${encodeURIComponent(repo)}`);
  } catch (e) {
    console.warn('recipe suggestion failed:', e.message);
  }
  const fallbackKey = repo.toLowerCase().replace(/[^a-z0-9_-]+/g, '-').replace(/^-+|-+$/g, '');
  setupCtx = {
    key: (sug && sug.key) || fallbackKey,
    repo,
    vllm_args: (sug && sug.vllm_args) || [],
    thenSwap: !!opts.thenSwap,
  };
  el('#cd-key').value = setupCtx.key;
  el('#cd-name').value = (sug && sug.display_name) || repo.split('/').pop();
  el('#cd-repo').value = repo;
  el('#cd-size').value = '';
-  el('#cd-mode').value = mode || 'solo';
+  el('#cd-mode').value = (sug && sug.mode) || 'solo';
  el('#cd-desc').value = '';
-  el('#cd-mml').value = 32768;
+  const knobs = (sug && sug.knobs) || {};
-  el('#cd-gmu').value = 0.85;
+  el('#cd-mml').value = knobs.max_model_len || 32768;
-  el('#cd-gmu-out').value = '0.85';
+  el('#cd-gmu').value = knobs.gpu_memory_utilization || 0.85;
-  el('#cd-fst').checked = true;
+  el('#cd-gmu-out').value = parseFloat(el('#cd-gmu').value).toFixed(2);
-  el('#cd-pcache').checked = true;
+  el('#cd-fst').checked = knobs.fastsafetensors !== false;
-  el('#cd-fp8').checked = true;
+  el('#cd-pcache').checked = knobs.prefix_caching !== false;
  el('#cd-fp8').checked = (knobs.kv_cache_dtype || 'fp8') === 'fp8';
  const det = el('#cd-detected');
  if (det) {
    if (sug) {
      const caps = (sug.capabilities || []).join(', ');
      const flags = setupCtx.vllm_args.length ? `: <code>${escapeHtml(setupCtx.vllm_args.join(' '))}</code>` : '';
      det.innerHTML = `Detected <strong>${escapeHtml(sug.family || 'Generic')}</strong>${caps ? ` · ${escapeHtml(caps)}` : ''}. Launch flags set automatically${flags}.`;
    } else {
      det.textContent = "Couldn't auto-detect this model's settings — pick mode and knobs manually.";
    }
    det.classList.remove('hidden');
  }
  const submit = el('#cd-submit');
  if (submit) submit.textContent = setupCtx.thenSwap ? 'Save & switch' : 'Save settings';
  dlg.showModal();
 }
@@ -1650,13 +1778,15 @@ function setupCatalogDialog() {
  el('#catalog-form').addEventListener('submit', async (e) => {
    e.preventDefault();
    const body = {
-      key: el('#cd-key').value.trim(),
+      key: el('#cd-key').value.trim() || setupCtx.key,
      display_name: el('#cd-name').value.trim(),
      repo: el('#cd-repo').value.trim(),
      size_gb: parseFloat(el('#cd-size').value) || 0,
      mode: el('#cd-mode').value,
      description: el('#cd-desc').value.trim() || null,
-      vllm_args: [],
+      // The inferred family flags (parsers / MoE backend); knob-controlled flags
      // are layered on by the server from `knobs`, so no duplication.
      vllm_args: setupCtx.vllm_args || [],
      knobs: {
        max_model_len: parseInt(el('#cd-mml').value, 10) || 32768,
        gpu_memory_utilization: parseFloat(el('#cd-gmu').value),
@@ -1674,8 +1804,9 @@ function setupCatalogDialog() {
      el('#catalog-dialog').close();
      closeDownloadPanel();
      await loadModels();
      if (setupCtx.thenSwap) triggerSwap(body.key);
      pollStatus();
-    } catch (e) { alert('Add to catalog failed: ' + e.message); }
+    } catch (e) { alert('Saving the model setup failed: ' + e.message); }
  });
 }
@@ -2102,6 +2233,7 @@ async function init() {
  });
  el('#sshkey-close').addEventListener('click', () => el('#sshkey-dialog').close());
  el('#open-local').addEventListener('click', openLocalModelDialog);
  el('#lock-release').addEventListener('click', releaseLock);
  setupCatalogDialog();
  setupAdvancedDialog();
  setupLocalModelDialog();
@@ -2116,19 +2248,22 @@ async function init() {
  } catch {}
  setupDashboardTabs();
  setupEndpointCollapse();
-  await loadModels();
+  // Fire the (SSH-backed) menu scan without awaiting — it self-renders a
  // "Scanning…" state and fills in when it returns, so a slow/unreachable
  // cluster never blocks first paint. pollStatus() below paints the rest.
  loadModels();
  await pollStatus();
  await renderServices();
  pollCoordination();
  pollHardware();
  pollUpdates();
  // Disk-status probe runs after first paint — slow over SSH and not blocking.
  loadDiskStatus();
  // Speech-model patches panel — slow over SSH, runs after first paint.
  renderSpeechModels();
  setInterval(pollStatus, 5000);
  setInterval(pollCoordination, 5000); // swap lock + schedule registry
  setInterval(pollHardware, 8000);    // every 8s
  setInterval(pollUpdates, 300000);  // every 5 min
-  setInterval(loadDiskStatus, 60000); // every 60s — disk state changes rarely
+  setInterval(loadModels, 60000); // every 60s — re-scan the Sparks for added/removed models
  setInterval(renderSpeechModels, 120000); // every 2 min — patches change rarely
 }
@@ -96,6 +96,13 @@
      </details>
    </section>
    <section id="lock-banner" class="banner lock-banner hidden">
      <span class="lock-icon" aria-hidden="true">🔒</span>
      <span id="lock-text">GPU swap path reserved</span>
      <span class="spacer"></span>
      <button id="lock-release" class="btn small-btn">Release</button>
    </section>
    <nav id="dashboard-tabs" class="dashboard-tabs hidden" role="tablist">
      <button type="button" class="dashboard-tab" data-tab="llm" role="tab" aria-selected="true">LLM</button>
      <button type="button" class="dashboard-tab" data-tab="audio" role="tab" aria-selected="false">Audio / Speech</button>
@@ -234,9 +241,10 @@
      <dialog id="catalog-dialog" class="modal">
        <form method="dialog" class="modal-form" id="catalog-form">
-          <h3>Add downloaded model to catalog</h3>
+          <h3>Set up this model</h3>
-          <p class="muted small">It will appear as a new card you can swap to. Knob values become its default launch flags — you can tweak later via the model's "Advanced" panel.</p>
+          <p class="muted small">This model is downloaded, but Spark Control needs to know how to launch it. We've guessed from the model's own files — confirm or adjust, and it's saved so you're never asked again.</p>
-          <label class="modal-row"><span>Key (URL-safe id)</span><input type="text" id="cd-key" required pattern="[a-zA-Z0-9_-]+"></label>
+          <p id="cd-detected" class="muted small cd-detected hidden"></p>
          <label class="modal-row"><span>Key (URL-safe id)</span><input type="text" id="cd-key" required pattern="[a-zA-Z0-9_-]+" readonly></label>
          <label class="modal-row"><span>Display name</span><input type="text" id="cd-name" required></label>
          <label class="modal-row"><span>Repo (read-only)</span><input type="text" id="cd-repo" readonly></label>
          <label class="modal-row"><span>Size (GB)</span><input type="number" id="cd-size" step="0.1" min="0"></label>
@@ -257,7 +265,7 @@
          </fieldset>
          <div class="modal-actions">
            <button type="button" id="cd-cancel" class="btn">Cancel</button>
-            <button type="submit" class="btn primary">Add to catalog</button>
+            <button type="submit" id="cd-submit" class="btn primary">Save settings</button>
          </div>
        </form>
      </dialog>
@@ -295,14 +303,14 @@
      <dialog id="disk-delete-dialog" class="modal">
        <form method="dialog" class="modal-form">
-          <h3>Delete model weights from disk?</h3>
+          <h3>Remove this model from the Sparks?</h3>
          <p id="dd-summary" class="muted small"></p>
          <ul class="muted small dd-hosts" id="dd-hosts"></ul>
-          <p class="muted small">This is reversible — you can re-download from the catalog at any time. The catalog entry stays intact.</p>
+          <p class="muted small">This deletes the weights and removes the card from the menu. You can always download it again later (re-downloading restores its saved settings).</p>
          <p id="dd-error" class="muted small dd-error hidden"></p>
          <div class="modal-actions">
            <button type="button" id="dd-cancel" class="btn">Cancel</button>
-            <button type="button" id="dd-confirm" class="btn danger">Delete from disk</button>
+            <button type="button" id="dd-confirm" class="btn danger">Remove from disk &amp; menu</button>
          </div>
        </form>
      </dialog>
@@ -347,11 +355,12 @@
        <div class="download-form" id="download-form">
          <label class="dl-row">
            <span class="dl-label">HuggingFace repo</span>
-            <input type="text" id="dl-repo" placeholder="e.g. RedHatAI/Qwen3.6-35B-A3B-NVFP4" autocomplete="off">
+            <input type="text" id="dl-repo" placeholder="e.g. RedHatAI/Qwen3.6-35B-A3B-NVFP4" autocomplete="off" list="dl-suggestions">
            <datalist id="dl-suggestions"></datalist>
            <a id="dl-hf-link" class="dl-hf-link hidden" href="#" target="_blank" rel="noopener" title="Open on Hugging Face">↗</a>
          </label>
          <div class="dl-help muted small">
-            <a href="https://huggingface.co/models?other=vllm" target="_blank" rel="noopener">Browse vLLM-compatible models</a>
+            Type any repo, or pick a known one from the list. <a href="https://huggingface.co/models?other=vllm" target="_blank" rel="noopener">Browse vLLM-compatible models</a>
            · NVFP4-quantized models (e.g. <code>RedHatAI/...</code>) are best for Blackwell hardware
          </div>
          <div class="dl-row">
@@ -394,6 +403,14 @@
      <section id="cards" class="cards"></section>
    </section>
    <section id="schedule-panel" class="schedule-panel hidden">
      <div class="section-header">
        <h2 class="section-title">Scheduled jobs</h2>
      </div>
      <p class="muted small">Registered by your own automation. Spark Control only displays these — it doesn't run them.</p>
      <div id="schedule-list" class="schedule-list"></div>
    </section>
    <section id="update-banner" class="update-banner hidden">
      <div class="ub-context muted small">
        Updates to <strong><a href="https://github.com/eugr/spark-vllm-docker" target="_blank" rel="noopener">eugr/spark-vllm-docker</a></strong>
@@ -74,6 +74,42 @@ main {
 }
 .banner em { font-style: normal; background: rgba(245, 158, 11, 0.15); padding: 2px 6px; border-radius: 4px; }
 /* GPU swap reservation (coordination layer) — informational, not a warning. */
 .lock-banner {
  display: flex;
  align-items: center;
  gap: 10px;
  border-color: var(--info);
  color: var(--info);
 }
 .lock-banner .lock-icon { font-size: 16px; }
 .lock-banner strong { color: var(--text); }
 .lock-banner .spacer { flex: 1; }
 /* Scheduled-jobs panel — read-only view of what external automation registered. */
 .schedule-panel { margin-top: 8px; }
 .schedule-list {
  display: grid;
  grid-template-columns: repeat(auto-fill, minmax(240px, 1fr));
  gap: 12px;
  margin-top: 8px;
 }
 .schedule-item {
  background: var(--surface);
  border: 1px solid var(--border);
  border-radius: var(--radius);
  padding: 12px 14px;
 }
 .schedule-item .name { font-weight: 600; margin-bottom: 4px; }
 .schedule-item code {
  background: var(--surface-2);
  border: 1px solid var(--border);
  border-radius: 4px;
  padding: 1px 5px;
  font-size: 12px;
 }
 .schedule-item .desc { margin-top: 6px; color: var(--muted); font-size: 13px; }
 /* ===== Endpoint panel ===== */
 .endpoint-panel {
@@ -742,6 +778,12 @@ main {
 .card .local-pill { color: var(--warn); border-color: rgba(245, 158, 11, 0.4); }
 .tag.on-disk { color: var(--accent); border-color: rgba(74, 222, 128, 0.4); }
 .tag.not-on-disk { color: var(--muted); border-color: var(--border); opacity: 0.7; }
 .tag.setup-pill { color: var(--warn); border-color: rgba(245, 158, 11, 0.4); }
 .card.needs-setup { border-style: dashed; }
 .card-actions .btn[data-setup-key] { flex: 1; }
 .empty-menu { grid-column: 1 / -1; padding: 28px 16px; text-align: center; border: 1px dashed var(--border); border-radius: 10px; }
 .cd-detected { padding: 8px 10px; border: 1px solid var(--border); border-radius: 8px; background: rgba(255,255,255,0.02); }
 .cd-detected code { word-break: break-all; }
 .card-actions .icon-btn.danger { color: var(--error); border-color: rgba(239, 68, 68, 0.3); margin-left: auto; }
 .card-actions .icon-btn.danger:hover:not(:disabled) { background: rgba(239, 68, 68, 0.08); border-color: var(--error); color: var(--error); }
 .card-actions .icon-btn.danger:disabled { opacity: 0.35; cursor: not-allowed; }
@@ -6,6 +6,7 @@ from datetime import datetime, timezone
 from typing import Optional
 from .config import Settings
 from .coordination import WebhookNotifier, build_webhook_payload
 from .models import Catalog, build_launch_command
 from .shellsafe import quote_arg
 from .ssh import ssh_run, ssh_stream, StreamHandle
@@ -33,9 +34,15 @@ class SwapJob:
 class SwapManager:
-    def __init__(self, settings: Settings, catalog: Catalog) -> None:
+    def __init__(
        self,
        settings: Settings,
        catalog: Catalog,
        notifier: Optional[WebhookNotifier] = None,
    ) -> None:
        self.settings = settings
        self.catalog = catalog
        self.notifier = notifier
        self.lock = asyncio.Lock()
        self.jobs: dict[str, SwapJob] = {}
        self.current_job_id: Optional[str] = None
@@ -78,6 +85,21 @@ class SwapManager:
                job.finished_at = datetime.now(timezone.utc).isoformat()
                if self.current_job_id == job.id:
                    self.current_job_id = None
        # Outside the swap lock (so a webhook POST can't stall a queued swap) and
        # only for real swaps — a dry run never changes the running model. A
        # webhook failure is logged inside fire(), never raised.
        if self.notifier is not None and self.notifier.enabled and not job.dry_run:
            event = "swap_complete" if job.state == "ready" else "swap_failed"
            await self.notifier.fire(event, build_webhook_payload(
                event=event,
                job_id=job.id,
                model_key=job.model_key,
                state=job.state,
                returncode=job.returncode,
                started_at=job.started_at,
                finished_at=job.finished_at,
                dry_run=job.dry_run,
            ))
    async def _do(self, job: SwapJob) -> None:
        model = self.catalog.models[job.model_key]
@@ -1,9 +1,14 @@
-# spark-control model catalog
+# spark-control launch recipes
 #
-# Edit this file (or override at runtime via the StartOS "Edit Model Catalog"
+# These are NOT the dashboard menu. The menu is whatever is actually downloaded
-# action) to add or change available models.
+# on the Sparks — Spark Control scans the Hugging Face cache on each load and
 # shows what it finds. These entries are launch *recipes*: matched to an on-disk
 # model by `repo`, they say HOW to launch it. A downloaded model with no recipe
 # here shows up as "needs setup", and the dashboard infers + saves one on first
 # use (from the model's own config.json). Add a recipe to make a known model
 # launch correctly the moment it's downloaded, with no setup prompt.
 #
-# Each model entry produces this command on Spark 1:
+# Each recipe produces this command on Spark 1:
 #   cd ~/spark-vllm-docker
 #   ./launch-cluster.sh [--solo] -d exec vllm serve <repo> \
 #     --port=<defaults.port> --host=<defaults.host> <vllm_args...>
@@ -54,6 +59,34 @@ models:
      - --enable-prefix-caching
      - --kv-cache-dtype=fp8
  gemma4-26b:
    display_name: "Gemma 4 26B-A4B (vision, light)"
    description: >-
      Lighter, faster sibling of the Gemma 4 31B above: a Mixture-of-Experts
      model with 26B total parameters but only ~4B active per token, so it
      generates quickly. Takes images as well as text (good for tasks like
      reading a business card into structured text). Reasoning is a bit
      shallower than the dense 31B. Runs solo on one Spark.
    repo: nvidia/Gemma-4-26B-A4B-NVFP4
    size_gb: 17
    mode: solo
    capabilities: [vision, reasoning, tools]
    expected_ready_seconds: 240
    vllm_args:
      - --gpu-memory-utilization=0.8
      - --max-model-len=32768
      - --max-num-batched-tokens=16384
      - --reasoning-parser=gemma4
      - --tool-call-parser=gemma4
      - --enable-auto-tool-choice
      # MoE backend: research found this model's expert layers fall back to
      # 'marlin' on GB10 (the fast flashinfer_cutlass path errors on sm_121).
      # If a swap fails to start, this flag is the first thing to flip.
      - --moe_backend=marlin
      - --load-format=fastsafetensors
      - --enable-prefix-caching
      - --kv-cache-dtype=fp8
  qwen36:
    display_name: "Qwen3.6 35B-A3B (daily driver)"
    description: >-
@@ -74,36 +107,3 @@ models:
      - --load-format=fastsafetensors
      - --enable-prefix-caching
      - --kv-cache-dtype=fp8
  qwen3-235b-fp8:
    display_name: "Qwen3 235B-A22B FP8 (legacy)"
    description: >-
      Earlier generation of the Qwen 235B family in native FP8 precision.
      Runs across both Sparks. Mostly superseded by Qwen3-VL above; keep
      around for text-only baseline comparisons.
    repo: Qwen/Qwen3-235B-A22B-FP8
    size_gb: 220
    mode: cluster
    capabilities: []
    expected_ready_seconds: 360
    vllm_args:
      - --gpu-memory-utilization=0.7
      - -tp=2
      - --distributed-executor-backend=ray
      - --max-model-len=32768
  qwen25-72b:
    display_name: "Qwen2.5 72B (legacy)"
    description: >-
      Last-generation 72B dense model. Cluster mode required due to size.
      Kept for compatibility and baseline comparison against newer Qwens.
    repo: Qwen/Qwen2.5-72B-Instruct
    size_gb: 145
    mode: cluster
    capabilities: []
    expected_ready_seconds: 360
    vllm_args:
      - --gpu-memory-utilization=0.7
      - -tp=2
      - --distributed-executor-backend=ray
      - --max-model-len=32768
@@ -0,0 +1,201 @@
 """Coordination layer: swap lock lifecycle/expiry, schedule registry CRUD, and
 the webhook payload+signature. All offline — the lock takes an injectable `now`
 so expiry is tested without sleeping, and the webhook is exercised only on the
 disabled (no-network) path plus its pure payload/signature helpers.
 """
 import asyncio
 from datetime import datetime, timedelta, timezone
 import pytest
 from app.coordination import (
    LOCK_TTL_MAX,
    LOCK_TTL_MIN,
    LockHeld,
    ScheduleRegistry,
    SwapLockManager,
    WebhookNotifier,
    build_webhook_payload,
    sign_payload,
    valid_schedule_id,
 )
 T0 = datetime(2026, 6, 17, 12, 0, 0, tzinfo=timezone.utc)
 # ----------------------------------------------------------------- swap lock ----
 def test_acquire_free_lock_returns_token_and_status_held():
    mgr = SwapLockManager()
    lock = mgr.acquire("openclaw", ttl_seconds=60, note="daily vol", now=T0)
    assert lock.token
    st = mgr.status(now=T0)
    assert st["held"] is True
    assert st["holder"] == "openclaw"
    assert st["note"] == "daily vol"
    assert st["seconds_remaining"] == 60
    assert "token" not in st  # public view never leaks the token
 def test_acquire_requires_holder():
    with pytest.raises(ValueError):
        SwapLockManager().acquire("  ", now=T0)
 def test_acquire_held_by_other_raises_lockheld_with_state():
    mgr = SwapLockManager()
    mgr.acquire("openclaw", ttl_seconds=60, now=T0)
    with pytest.raises(LockHeld) as ei:
        mgr.acquire("johnny5", ttl_seconds=60, now=T0)
    assert ei.value.state["holder"] == "openclaw"
 def test_reacquire_with_token_extends_and_keeps_token():
    mgr = SwapLockManager()
    first = mgr.acquire("openclaw", ttl_seconds=60, now=T0)
    later = T0 + timedelta(seconds=30)
    second = mgr.acquire("openclaw", ttl_seconds=60, token=first.token, now=later)
    assert second.token == first.token
    # window extended from the later moment, not the original
    assert mgr.status(now=later)["seconds_remaining"] == 60
    assert second.acquired_at == first.acquired_at  # acquired_at preserved
 def test_reacquire_without_token_is_refused_even_for_same_holder_name():
    # Holder name is descriptive, not a secret — matching it must not grant access.
    mgr = SwapLockManager()
    mgr.acquire("openclaw", ttl_seconds=60, now=T0)
    with pytest.raises(LockHeld):
        mgr.acquire("openclaw", ttl_seconds=60, now=T0)
 def test_ttl_is_clamped():
    mgr = SwapLockManager()
    mgr.acquire("a", ttl_seconds=0, now=T0)
    assert mgr.status(now=T0)["seconds_remaining"] == LOCK_TTL_MIN
    mgr2 = SwapLockManager()
    mgr2.acquire("b", ttl_seconds=10**9, now=T0)
    assert mgr2.status(now=T0)["seconds_remaining"] == LOCK_TTL_MAX
 def test_lock_expires_and_clears_lazily():
    mgr = SwapLockManager()
    tok = mgr.acquire("openclaw", ttl_seconds=10, now=T0).token
    after = T0 + timedelta(seconds=11)
    assert mgr.status(now=after) == {"held": False}
    assert mgr.verify(tok, now=after) is False
    # an expired lock is free to re-take by anyone
    mgr.acquire("johnny5", ttl_seconds=10, now=after)
    assert mgr.status(now=after)["holder"] == "johnny5"
 def test_verify_matches_only_active_token():
    mgr = SwapLockManager()
    tok = mgr.acquire("openclaw", ttl_seconds=60, now=T0).token
    assert mgr.verify(tok, now=T0) is True
    assert mgr.verify("nope", now=T0) is False
    assert mgr.verify(None, now=T0) is False
 def test_release_requires_token_then_frees():
    mgr = SwapLockManager()
    tok = mgr.acquire("openclaw", ttl_seconds=60, now=T0).token
    with pytest.raises(PermissionError):
        mgr.release("wrong", now=T0)
    assert mgr.release(tok, now=T0) is True
    assert mgr.status(now=T0) == {"held": False}
 def test_force_release_skips_token_and_release_of_free_lock_is_false():
    mgr = SwapLockManager()
    mgr.acquire("openclaw", ttl_seconds=60, now=T0)
    assert mgr.release(force=True, now=T0) is True
    assert mgr.release(force=True, now=T0) is False  # nothing held now
 def test_is_blocked_by_is_the_swap_gate():
    # Mirrors the single-read decision the /api/swap endpoint makes.
    mgr = SwapLockManager()
    assert mgr.is_blocked_by(None, now=T0) is None          # free lock blocks nobody
    tok = mgr.acquire("openclaw", ttl_seconds=10, now=T0).token
    blocked = mgr.is_blocked_by(None, now=T0)               # no token -> blocked
    assert blocked is not None and blocked["holder"] == "openclaw"
    assert mgr.is_blocked_by("wrong", now=T0) is not None   # wrong token -> blocked
    assert mgr.is_blocked_by(tok, now=T0) is None           # holder's token -> allowed
    # At/after expiry the gate is open even without a token (the bug a separate
    # status()+verify() pair would get wrong).
    assert mgr.is_blocked_by(None, now=T0 + timedelta(seconds=11)) is None
 # ------------------------------------------------------------------- webhook ----
 def test_build_webhook_payload_shape():
    p = build_webhook_payload(
        event="swap_complete", job_id="abc123", model_key="gemma",
        state="ready", returncode=0, started_at="t0", finished_at="t1",
        dry_run=False,
    )
    assert p == {
        "event": "swap_complete", "job_id": "abc123", "model_key": "gemma",
        "state": "ready", "returncode": 0, "started_at": "t0",
        "finished_at": "t1", "dry_run": False,
    }
 def test_sign_payload_is_deterministic_and_prefixed():
    body = b'{"event":"swap_complete"}'
    sig = sign_payload("s3cr3t", body)
    assert sig.startswith("sha256=")
    assert sig == sign_payload("s3cr3t", body)
    assert sig != sign_payload("other", body)
 def test_disabled_webhook_fire_is_noop():
    n = WebhookNotifier("", "")
    assert n.enabled is False
    # Must not attempt any network call or raise when no URL is configured.
    assert asyncio.run(n.fire("swap_complete", {"x": 1})) is None
 # --------------------------------------------------------- schedule registry ----
 def test_register_and_list_schedule():
    reg = ScheduleRegistry()
    e = reg.register(name="Daily Vol", owner="openclaw", cron="0 6 * * *")
    assert e.id and e.registered_at and e.updated_at
    listed = reg.list()
    assert len(listed) == 1 and listed[0]["name"] == "Daily Vol"
 def test_register_with_id_updates_in_place():
    reg = ScheduleRegistry()
    reg.register(name="Daily Vol", id="dv", owner="openclaw", cron="0 6 * * *")
    reg.register(name="Daily Vol v2", id="dv", owner="openclaw", cron="0 7 * * *")
    listed = reg.list()
    assert len(listed) == 1
    assert listed[0]["name"] == "Daily Vol v2" and listed[0]["cron"] == "0 7 * * *"
 def test_register_requires_name_and_validates_id():
    reg = ScheduleRegistry()
    with pytest.raises(ValueError):
        reg.register(name="  ")
    with pytest.raises(ValueError):
        reg.register(name="ok", id="bad id; rm -rf")
 def test_delete_schedule():
    reg = ScheduleRegistry()
    reg.register(name="Daily Vol", id="dv")
    assert reg.delete("dv") is True
    assert reg.delete("dv") is False
    assert reg.list() == []
 def test_valid_schedule_id():
    assert valid_schedule_id("daily-vol")
    assert valid_schedule_id("a.b_c-1")
    assert not valid_schedule_id("")
    assert not valid_schedule_id("../etc")
    assert not valid_schedule_id("has space")
    assert not valid_schedule_id("x" * 65)
@@ -0,0 +1,190 @@
 """Disk-driven menu helpers: cache-dir parsing + launch-recipe inference.
 All offline — pure functions over a fake cache listing and fake config.json
 dicts. The SSH scan, the menu merge, and the suggest endpoint that wire these
 together are exercised by hand against the live cluster (mock-heavy unit tests of
 those would test the mocks).
 """
 import asyncio
 from app import discovery
 from app.config import Settings
 from app.disk import DiskStatus, cache_dirname_to_repo, parse_cache_listing
 from app.discovery import repo_to_key, infer_recipe, _detect_family
 from app.models import load_catalog
 # ---- cache dirname <-> repo ----
 def test_cache_dirname_to_repo_roundtrip():
    assert cache_dirname_to_repo("models--RedHatAI--Qwen3.6-35B-A3B-NVFP4") == "RedHatAI/Qwen3.6-35B-A3B-NVFP4"
 def test_cache_dirname_name_with_double_dash():
    # The org is the first segment; everything after is the name (single '/').
    assert cache_dirname_to_repo("models--org--weird--name") == "org/weird--name"
 def test_cache_dirname_rejects_non_model_dirs():
    assert cache_dirname_to_repo("datasets--foo--bar") is None
    assert cache_dirname_to_repo("models--onlyorg") is None
    assert cache_dirname_to_repo("random") is None
 # ---- parse_cache_listing ----
 def test_parse_cache_listing_complete_and_incomplete():
    out = (
        "20000000000|1|models--RedHatAI--Qwen3.6-35B-A3B-NVFP4\n"
        "5000000000|0|models--some--half-downloaded\n"
        "\n"
        "garbage line with no pipes\n"
        "123|1|not-a-model-dir\n"
    )
    items = parse_cache_listing(out)
    assert items == [
        ("RedHatAI/Qwen3.6-35B-A3B-NVFP4", 20000000000, True),
        ("some/half-downloaded", 5000000000, False),
    ]
 def test_parse_cache_listing_bad_size_defaults_zero():
    items = parse_cache_listing("notanumber|1|models--a--b")
    assert items == [("a/b", 0, True)]
 # ---- repo_to_key ----
 def test_repo_to_key_is_url_safe_and_stable():
    assert repo_to_key("RedHatAI/Qwen3.6-35B-A3B-NVFP4") == "redhatai-qwen3-6-35b-a3b-nvfp4"
    # Idempotent enough to be a stable id across calls.
    assert repo_to_key("nvidia/Gemma-4-26B-A4B-NVFP4") == "nvidia-gemma-4-26b-a4b-nvfp4"
 # ---- family detection ----
 def test_detect_qwen3_moe():
    cfg = {"architectures": ["Qwen3MoeForCausalLM"], "model_type": "qwen3_moe", "num_experts": 128}
    label, flags, caps = _detect_family(cfg)
    assert "--reasoning-parser=qwen3" in flags
    assert "--moe_backend=flashinfer_cutlass" in flags
    assert "reasoning" in caps
    assert "MoE" in label
 def test_detect_gemma_moe_uses_marlin():
    cfg = {"architectures": ["Gemma4MoeForConditionalGeneration"], "model_type": "gemma4_moe", "num_local_experts": 8}
    label, flags, caps = _detect_family(cfg)
    assert "--reasoning-parser=gemma4" in flags
    assert "--tool-call-parser=gemma4" in flags
    assert "--moe_backend=marlin" in flags        # NOT flashinfer_cutlass — GB10 footgun
    assert "vision" in caps                         # ConditionalGeneration => multimodal
    assert "tools" in caps
 def test_detect_generic_has_no_family_flags():
    label, flags, caps = _detect_family({"architectures": ["LlamaForCausalLM"], "model_type": "llama"})
    assert flags == []
    assert label == "Generic"
 def test_detect_vision_from_config_keys():
    _, _, caps = _detect_family({"model_type": "qwen3", "vision_config": {"x": 1}})
    assert "vision" in caps
 # ---- infer_recipe (the prefill the setup form receives) ----
 def test_infer_recipe_solo_small_model():
    cfg = {"architectures": ["Qwen3ForCausalLM"], "model_type": "qwen3"}
    rec = infer_recipe("RedHatAI/Qwen3.6-35B-A3B-NVFP4", cfg, total_bytes=20_000_000_000, on_host_count=1)
    assert rec["mode"] == "solo"
    assert rec["key"] == "redhatai-qwen3-6-35b-a3b-nvfp4"
    assert rec["repo"] == "RedHatAI/Qwen3.6-35B-A3B-NVFP4"
    assert "--reasoning-parser=qwen3" in rec["vllm_args"]
    assert "-tp=2" not in rec["vllm_args"]
    assert rec["knobs"]["kv_cache_dtype"] == "fp8"
 def test_infer_recipe_cluster_when_on_both_hosts():
    rec = infer_recipe("org/big", {}, total_bytes=10_000_000_000, on_host_count=2)
    assert rec["mode"] == "cluster"
    assert "-tp=2" in rec["vllm_args"]
    assert "--distributed-executor-backend=ray" in rec["vllm_args"]
    assert rec["knobs"]["gpu_memory_utilization"] == 0.7
 def test_infer_recipe_cluster_when_too_big_for_one_spark():
    rec = infer_recipe("org/huge", {}, total_bytes=200_000_000_000, on_host_count=1)
    assert rec["mode"] == "cluster"
 # ---- build_menu merge (disk scan ∪ recipes) ----
 def _both_spark_settings(monkeypatch) -> Settings:
    for k in ("SPARK1_HOST", "SPARK1_USER", "SPARK2_HOST", "SPARK2_USER"):
        monkeypatch.delenv(k, raising=False)
    monkeypatch.setenv("SPARK1_HOST", "1.1.1.1")
    monkeypatch.setenv("SPARK1_USER", "u")
    monkeypatch.setenv("SPARK2_HOST", "2.2.2.2")
    monkeypatch.setenv("SPARK2_USER", "u")
    return Settings.from_env()
 def test_build_menu_merges_recipe_discovered_and_hides_incomplete(monkeypatch):
    cat = load_catalog("models.yaml")  # bundled recipes incl. qwen36 + gemma4
    settings = _both_spark_settings(monkeypatch)
    async def fake_list(host, user, s):
        if host == "1.1.1.1":
            return [
                ("RedHatAI/Qwen3.6-35B-A3B-NVFP4", 20_000_000_000, True),  # recipe match
                ("someorg/mystery-7B", 7_000_000_000, True),               # needs setup
                ("broken/half", 1_000_000_000, False),                     # incomplete -> hidden
            ]
        return []  # spark2 empty
    async def fake_probe(repo, mode, s, *, local_path=None):
        return DiskStatus(repo=local_path or repo, on_disk=False, total_bytes=0, per_host=[])
    monkeypatch.setattr(discovery, "list_cached_models", fake_list)
    monkeypatch.setattr(discovery, "probe_disk", fake_probe)
    menu = asyncio.run(discovery.build_menu(settings, cat))
    # Recipe-matched: keyed by recipe key, ready (not needs_setup), real size.
    assert "qwen36" in menu
    assert menu["qwen36"]["needs_setup"] is False
    assert menu["qwen36"]["total_bytes"] == 20_000_000_000
    # Discovered-without-recipe: slug key, needs_setup.
    slug = repo_to_key("someorg/mystery-7B")
    assert menu[slug]["needs_setup"] is True
    # Incomplete download is filtered out entirely.
    assert all("half" not in k for k in menu)
    # A recipe with nothing on disk (e.g. gemma4) must NOT appear — the menu is the disk.
    assert "gemma4" not in menu
 def test_build_menu_sums_cluster_model_across_both_sparks(monkeypatch):
    cat = load_catalog("models.yaml")
    settings = _both_spark_settings(monkeypatch)
    async def fake_list(host, user, s):
        # Same repo present on BOTH Sparks — one card, sizes summed (not two cards).
        return [("org/sharded-235B", 70_000_000_000, True)]
    async def fake_probe(repo, mode, s, *, local_path=None):
        return DiskStatus(repo=repo, on_disk=False, total_bytes=0, per_host=[])
    monkeypatch.setattr(discovery, "list_cached_models", fake_list)
    monkeypatch.setattr(discovery, "probe_disk", fake_probe)
    menu = asyncio.run(discovery.build_menu(settings, cat))
    key = repo_to_key("org/sharded-235B")
    assert list(menu) == [key]                       # exactly one card
    assert menu[key]["total_bytes"] == 140_000_000_000  # summed across both hosts
    assert len(menu[key]["per_host"]) == 2
    assert menu[key]["mode"] == "cluster"            # present on 2 hosts -> cluster
@@ -173,6 +173,24 @@ const inputSpec = InputSpec.of({
    placeholder: 'starts with "nvapi-..."',
    masked: true,
  }),
  swap_webhook_url: Value.text({
    name: 'Swap webhook URL (optional)',
    description:
      'If you run automation that needs to know when the loaded model changes, paste a URL here. Spark Control POSTs a small JSON event (swap_complete / swap_failed) to it after every model swap, so the consumer can re-point its config to the new model. Leave blank to disable. Only needed if something other than this dashboard cares about swaps.',
    required: false,
    default: null,
    placeholder: 'e.g. https://my-service.local/spark-swap',
    masked: false,
  }),
  swap_webhook_secret: Value.text({
    name: 'Swap webhook secret (optional)',
    description:
      'Optional shared secret. If set, each webhook is signed with an "X-Spark-Signature: sha256=…" header (HMAC of the body) so the receiver can verify it really came from Spark Control. Leave blank to send the webhook unsigned.',
    required: false,
    default: null,
    placeholder: 'a random string the receiver also knows',
    masked: true,
  }),
 })
 export const configureSparks = sdk.Action.withInput(
@@ -35,6 +35,11 @@ export const sparkConfigSchema = z.object({
  open_webui_url: z.string().catch(''),
  // Optional NGC API key for pulling NIM containers from nvcr.io/nim/...
  ngc_api_key: z.string().catch(''),
  // Optional coordination webhook: POSTed on swap_complete/swap_failed so
  // downstream consumers re-point their model config. Blank => disabled.
  swap_webhook_url: z.string().catch(''),
  // Optional shared secret; if set, the webhook body is HMAC-signed.
  swap_webhook_secret: z.string().catch(''),
 })
 export type SparkConfig = z.infer<typeof sparkConfigSchema>
@@ -32,6 +32,8 @@ export const main = sdk.setupMain(async ({ effects }) => {
    matrix_bridge_user: '',
    open_webui_url: '',
    ngc_api_key: '',
    swap_webhook_url: '',
    swap_webhook_secret: '',
  }
  return sdk.Daemons.of(effects).addDaemon('primary', {
@@ -75,6 +77,8 @@ export const main = sdk.setupMain(async ({ effects }) => {
        CONNECTIVITY_LOG: '/data/connectivity.json',
        OPEN_WEBUI_URL: cfg.open_webui_url,
        NGC_API_KEY: cfg.ngc_api_key,
        SWAP_WEBHOOK_URL: cfg.swap_webhook_url,
        SWAP_WEBHOOK_SECRET: cfg.swap_webhook_secret,
        BIND_PORT: String(uiPort),
      },
    },
@@ -1,10 +1,10 @@
 import { VersionInfo, IMPOSSIBLE } from '@start9labs/start-sdk'
 export const v0_1_0 = VersionInfo.of({
-  version: '0.24.0:0',
+  version: '0.26.0:0',
  releaseNotes: {
    en_US:
-      "v0.24.0:0 — configurable cluster topology. Spark Control no longer assumes our exact layout, so a cluster that's wired differently can be monitored without forking. Three new optional settings in Configure Sparks: (1) vLLM container name — defaults to \"vllm_node\"; set it if your swappable vLLM runs under a different container name (the swap log view and pre-flight validator exec into it by name). (2) Services to hide — a comma-separated list of built-in services your cluster doesn't run (parakeet, kokoro, embeddings, qdrant); hidden ones show no tile and are never probed, so e.g. a vLLM sharing Parakeet's default port 8000 no longer gets a confusing Parakeet probe. (3) Monitor a second vLLM — register a vLLM on another Spark as a custom service with kind \"vllm\" (in /data/services-overrides.yaml); it gets a read-only health tile (loaded model + container state + start/stop/restart) alongside the swappable one. API: /api/endpoints now reports a `disabled` flag per service.",
+      "v0.26.0:0 — the model menu is now what's actually on your Sparks. The dashboard scans both Sparks for downloaded models and shows exactly those — no more hard-coded list. (1) Delete means delete: removing a model frees its weights AND takes the card off the menu (re-download later to bring it back, with its saved settings). (2) Download a new model and it appears on the menu by itself when it finishes. (3) Models Spark Control doesn't recognize show a \"needs setup\" card — the first time you switch to one, it reads the model's own files, guesses how to launch it (which family, solo vs both Sparks, the right vLLM flags), and asks you to confirm once; after that it's a normal card. (4) The download box now autocompletes known-good models. (5) Each install shows its own Sparks' models, so a shared copy no longer displays someone else's list. Removed the two legacy Qwen entries (235B FP8, 2.5 72B) — they'll still appear if you actually have them downloaded. No consumer-API changes; the /v1 proxy and swap API are unchanged.",
  },
  migrations: {
    up: async ({ effects }) => {},
@@ -74,11 +74,15 @@ For a cluster wired differently from the reference layout, three optional knobs
 ## Adding a new model
-1. Add an entry to `image/models.yaml`. Required fields: `display_name`, `repo`, `size_gb`, `mode` (`solo` or `cluster`), `vllm_args`. Optional but recommended: `description` (one paragraph — what the model is, what it's good for, how it differs from others; renders below the meta tags in each card), `capabilities` (tags like `[vision, reasoning, tools]`), `expected_ready_seconds`.
+The menu is whatever's downloaded on the Sparks, so the normal path is just:
-2. Confirm the weights are on the Spark: `ssh <spark-user>@<spark-1-host> 'ls ~/.cache/huggingface/hub/'`. If not, download with `./hf-download.sh <repo>` on Spark 1.
+**download it, then set it up once.**
 3. Rebuild + redeploy the package: `cd package && make x86 && make install`.
-If `description` is omitted, the card simply hides that section — no need to populate it for every model. Keep descriptions generic (not user-specific) so the catalog stays portable.
+1. **Download** from the dashboard (**+ Download a new model**, paste the HF repo) or on Spark 1 with `./hf-download.sh <repo>`. When it finishes it appears on the menu by itself.
 2. **Set it up.** If Spark Control already has a recipe for it (see below), it's ready to switch to. Otherwise it shows a **"needs setup"** card: the first switch reads the model's `config.json`, proposes how to launch it (family/parsers, solo vs cluster, vLLM flags), and you confirm once. The confirmed recipe persists to `/data/models-overrides.yaml` (survives package updates).
 ### Bundling a launch recipe (optional — skips the setup prompt)
 To make a known model launch correctly the instant it's downloaded, add a *recipe* to `image/models.yaml`. These are **not** the menu — they're matched to an on-disk model by `repo`. Required: `display_name`, `repo`, `size_gb`, `mode` (`solo`/`cluster`), `vllm_args`. Optional: `description`, `capabilities` (e.g. `[vision, reasoning, tools]`), `expected_ready_seconds`. Then rebuild + redeploy: `cd package && make x86 && make install`. Keep descriptions generic (not user-specific) so the recipes stay portable.
 ### Local / fine-tuned models (v0.23.0+)
Author	SHA1	Message	Date
Keysat	df9f244eae	v0.26.0:0 - disk-driven model menu (scan sparks; recipes; needs-setup) The dashboard menu is now the set of models actually downloaded on the Sparks, not a hard-coded catalog. models.yaml + overrides are reframed as launch recipes matched to an on-disk model by repo; an on-disk model with no recipe is flagged needs_setup and its launch settings are inferred from its config.json for a one-time operator confirmation (discovery.py). - delete now removes weights AND the menu card (delete_from_disk sweeps all hosts; the delete endpoint resolves keys via the live menu) - new GET /api/models/suggest; /api/models returns the menu + a recipes list (download autocomplete); GET /api/models/disk-status removed - dropped the two legacy Qwen recipes (235B FP8, 2.5 72B) - tests: +test_discovery.py (cache parsing, infer_recipe, build_menu merge)	2026-06-18 11:09:56 -05:00
Keysat	c0b35184ba	docs: trim Current state to live status — coordination epic shipped	2026-06-18 08:09:59 -05:00
Keysat	7ecd77f1e5	docs: defer raw-docker swap generalization — multi-node rationale recorded	2026-06-18 07:58:25 -05:00
Keysat	6bcda6e348	docs: v0.25.0:0 installed live — update Current state	2026-06-18 07:11:33 -05:00
Keysat	7ae6ab3ba8	v0.25.0:0 - cluster coordination layer (swap lock + webhook + schedule registry) GPU-arbiter safety layer for when automation, not just the dashboard, swaps models: - swap reservation lock (POST/GET/DELETE /api/swap/lock); 423-enforced in post_swap via a single-read gate, TTL-bounded, secret-token auth, human force-release override + dashboard banner - swap webhook (swap_complete/swap_failed) fired outside the swap lock, optional HMAC signature, configurable URL+secret - read-only schedule registry (GET/POST/DELETE /api/schedule) + dashboard panel New module image/app/coordination.py; docs/COORDINATION.md for consumers; 22 offline tests in test_coordination.py.	2026-06-18 07:07:08 -05:00
Keysat	dd3d1412d4	docs: v0.24.0:0 committed/tagged/pushed — Gitea release asset + live install still pending	2026-06-17 23:11:14 -05:00