chore(release): v0.2.8 — multiple managed local models

- The gateway can now run several local .cmf models at once: a [[cmf.servers]]
  list, one supervised `cortiq serve` per entry on its own port, each registered
  in the pool and independently routable/pinnable. The single legacy
  local_model/local_port still works (back-compat via effective_servers()).
- Settings → Local models is now a rows editor (id · path · port + check · threads
  · GPU) with add/remove and per-server serving status.
- Deleting a managed model stops just its server and removes its file — unless
  another model still points at the same file (shared-file guard).
This commit is contained in:
infosave2007
2026-07-09 12:00:10 +03:00
parent f8c1bb0914
commit 0c20eb34f0
10 changed files with 405 additions and 190 deletions
+73 -36
View File
@@ -441,20 +441,20 @@ async fn list_models(State(state): State<SharedState>) -> ApiResult {
})
.collect();
// The managed local CMF server is registered in the routing registry from
// the [cmf] flags, not from [[models]]. Surface it here (marked managed, so
// the UI shows it read-only) so the real running model is visible and
// pinnable without needing a hand-maintained static pool entry. A fresh
// install with no managed model still returns an empty list.
// Managed local CMF servers are registered in the routing registry from the
// [cmf] config, not from [[models]]. Surface each here (marked managed, so
// the UI shows it read-only) so the real running models are visible and
// pinnable without a hand-maintained static pool entry. A fresh install with
// no managed model still returns an empty list.
let cmf = &live.cfg.cmf;
if cmf.manage_server
&& !cmf.local_model.trim().is_empty()
&& !models.iter().any(|m| m["id"] == json!(cmf.model_id))
{
for s in cmf.effective_servers() {
if models.iter().any(|m| m["id"] == json!(s.id)) {
continue;
}
models.push(json!({
"id": cmf.model_id,
"id": s.id,
"provider": "openai",
"base_url": format!("http://{}:{}/v1", cmf.local_host, cmf.local_port),
"base_url": format!("http://{}:{}/v1", cmf.local_host, s.port),
"model": "cortiq",
"cost_tier": "local",
"price_in": 0.0,
@@ -544,24 +544,46 @@ fn strip_id_from_routing(routing: &mut RoutingCfg, id: &str) {
}
}
/// Whether any managed server or static model in `cfg` still uses `path` as its
/// local `.cmf`. Guards against deleting a file that another model shares.
fn cmf_path_referenced(cfg: &crate::config::Config, path: &std::path::Path) -> bool {
cfg.cmf
.effective_servers()
.iter()
.any(|s| std::path::Path::new(&s.model) == path)
|| cfg
.models
.iter()
.any(|m| cmf_file_for_model(&cfg.cmf, m).as_deref() == Some(path))
}
async fn delete_model(State(state): State<SharedState>, Path(id): Path<String>) -> ApiResult {
let mut cfg = current_cfg(&state);
// The managed local model lives in [cmf], not [[models]]. Deleting it stops
// the running server, clears the managed config, and removes its file.
if id == cfg.cmf.model_id
&& cfg.cmf.manage_server
&& !cfg.cmf.local_model.trim().is_empty()
&& !cfg.models.iter().any(|x| x.id == id)
{
let file = cfg.cmf.local_model.clone();
cfg.cmf.local_model = String::new();
cfg.cmf.manage_server = false;
strip_id_from_routing(&mut cfg.routing, &id);
state.cmf.stop().await;
state.reload(cfg)?;
let file_removed = remove_cmf_path(std::path::Path::new(&file));
return ok(json!({ "ok": true, "file_removed": file_removed }));
// Managed local models live in [cmf], not [[models]]. Deleting one stops its
// server, removes it from the managed config, and deletes its file — unless
// another model still points at the same file.
let managed = cfg.cmf.effective_servers().into_iter().find(|s| s.id == id);
if let Some(server) = managed {
if !cfg.models.iter().any(|x| x.id == id) {
if cfg.cmf.servers.iter().any(|s| s.id == id) {
cfg.cmf.servers.retain(|s| s.id != id); // multi-model list entry
} else {
cfg.cmf.local_model = String::new(); // legacy single model
cfg.cmf.manage_server = false;
}
strip_id_from_routing(&mut cfg.routing, &id);
let path = std::path::PathBuf::from(&server.model);
let shared = cmf_path_referenced(&cfg, &path);
state.cmf.stop_one(&id).await;
state.reload(cfg)?;
let file_removed = if shared {
false
} else {
remove_cmf_path(&path)
};
return ok(json!({ "ok": true, "file_removed": file_removed }));
}
}
let victim = cfg.models.iter().find(|x| x.id == id).cloned();
@@ -574,8 +596,9 @@ async fn delete_model(State(state): State<SharedState>, Path(id): Path<String>)
let file = victim
.as_ref()
.and_then(|m| cmf_file_for_model(&cfg.cmf, m));
// Only remove the file if no remaining model still references it.
let file = file.filter(|p| !cmf_path_referenced(&cfg, p));
state.reload(cfg)?;
// Physically remove the local .cmf after the config is persisted.
let file_removed = file.map(|p| remove_cmf_path(&p)).unwrap_or(false);
ok(json!({ "ok": true, "file_removed": file_removed }))
}
@@ -765,6 +788,7 @@ async fn put_settings(State(state): State<SharedState>, Json(b): Json<SettingsBo
|| v.local_model != cfg.cmf.local_model
|| v.threads != cfg.cmf.threads
|| v.gpu != cfg.cmf.gpu
|| v.servers != cfg.cmf.servers
{
needs_restart = true;
}
@@ -782,19 +806,32 @@ async fn cmf_status(State(state): State<SharedState>) -> ApiResult {
let cmf = state.live().cfg.cmf.clone();
let ver = crate::cmf_runtime::installed_version(&cmf.cortiq_bin);
let st = state.cmf.status();
// Each configured managed server, merged with its live runtime status.
let servers: Vec<Value> = cmf
.effective_servers()
.iter()
.map(|s| {
let live = st.servers.iter().find(|x| x.id == s.id);
json!({
"id": s.id,
"model": s.model,
"port": s.port,
"threads": s.threads,
"gpu": s.gpu,
"running": live.map(|l| l.running).unwrap_or(false),
"healthy": live.map(|l| l.healthy).unwrap_or(false),
"last_error": live.and_then(|l| l.last_error.clone()),
})
})
.collect();
ok(json!({
"installed": ver.is_some(),
"version": ver,
"running": st.running,
"healthy": st.healthy,
"last_error": st.last_error,
"log": st.log,
// serving info so the card can show "serving <model> on :<port>"
"latest": st.latest_version,
"manage_server": cmf.manage_server,
"local_model": cmf.local_model,
"model_id": cmf.model_id,
"local_host": cmf.local_host,
"local_port": cmf.local_port,
"servers": servers,
"log": st.log,
}))
}
@@ -825,7 +862,7 @@ async fn cmf_port_check(State(state): State<SharedState>, Query(q): Query<PortQu
};
let st = state.cmf.status();
let port = q.port;
let running_here = port == cmf.local_port && st.running;
let running_here = st.servers.iter().any(|s| s.port == port && s.running);
// The bind+connect probe is blocking std::net; run it off the async runtime.
let (available, detail, suggested) = tokio::task::spawn_blocking(move || {
+179 -75
View File
@@ -1,51 +1,68 @@
//! Local CMF model server lifecycle.
//!
//! Closes the loop with <https://github.com/infosave2007/cmf>: install the
//! `cortiq` CLI from crates.io, keep it up to date, and run `cortiq serve` as a
//! managed child process. The served model exposes an OpenAI-compatible API, so
//! the registry registers it as an ordinary provider and the gateway routes to
//! it like any other backend.
//! `cortiq` CLI from crates.io, keep it up to date, and run one `cortiq serve`
//! per configured local model as a managed child process. Each served model
//! exposes an OpenAI-compatible API, so the registry registers it as an ordinary
//! provider and the gateway routes to it like any other backend.
use crate::config::CmfCfg;
use crate::config::{CmfCfg, CmfServer};
use std::process::Stdio;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::process::{Child, Command};
/// Observable state of the managed local CMF server (surfaced to the admin API).
/// Observable state of one managed local model server.
#[derive(Clone, Debug, Default, serde::Serialize)]
pub struct CmfServerStatus {
/// Pool id it is registered under.
pub id: String,
/// Path of the `.cmf` being served.
pub model: String,
/// Port the server binds to.
pub port: u16,
/// OpenAI base URL the local server is registered under.
pub base_url: String,
/// The child `cortiq serve` process has been spawned.
pub running: bool,
/// `/healthz` on the local server answered 200.
pub healthy: bool,
/// Last fatal error for this server, if any.
pub last_error: Option<String>,
}
/// Observable state of the managed local CMF layer (surfaced to the admin API).
#[derive(Clone, Debug, Default, serde::Serialize)]
pub struct CmfStatus {
/// `manage_server` is on and a `local_model` is configured.
/// `manage_server` is on and at least one local model is configured.
pub enabled: bool,
/// Version reported by `cortiq --version` (None = not installed).
pub installed_version: Option<String>,
/// Newest `cortiq-cli` on crates.io (populated when `auto_update` runs).
pub latest_version: Option<String>,
/// The child `cortiq serve` process has been spawned.
pub running: bool,
/// `/healthz` on the local server answered 200.
pub healthy: bool,
/// OpenAI base URL the local server is registered under.
pub base_url: String,
/// Path of the `.cmf` being served.
pub model: String,
/// Last fatal error, if any.
pub last_error: Option<String>,
/// Per-model server status.
pub servers: Vec<CmfServerStatus>,
/// Recent lifecycle log lines (bounded).
pub log: Vec<String>,
}
/// Holds the managed child process and its live status.
/// A spawned server: its pool id and the child process handle.
struct Managed {
id: String,
child: Child,
}
/// Holds the managed child processes and their live status.
pub struct CmfRuntime {
status: Mutex<CmfStatus>,
child: Mutex<Option<Child>>,
children: Mutex<Vec<Managed>>,
}
impl CmfRuntime {
pub fn new() -> Arc<Self> {
Arc::new(Self {
status: Mutex::new(CmfStatus::default()),
child: Mutex::new(None),
children: Mutex::new(Vec::new()),
})
}
@@ -59,32 +76,115 @@ impl CmfRuntime {
let mut s = self.status.lock().unwrap();
s.log.push(msg);
let n = s.log.len();
if n > 50 {
s.log.drain(0..n - 50);
if n > 80 {
s.log.drain(0..n - 80);
}
}
/// A global (install/update) failure — not tied to a single server.
fn fail(&self, msg: impl Into<String>) {
let msg = msg.into();
tracing::warn!(target: "cmf", "{msg}");
self.status.lock().unwrap().log.push(format!("✗ {msg}"));
}
/// Record a fatal error against a specific server (and the log).
fn fail_server(&self, id: &str, msg: impl Into<String>) {
let msg = msg.into();
tracing::warn!(target: "cmf", "[{id}] {msg}");
let mut s = self.status.lock().unwrap();
s.last_error = Some(msg.clone());
s.log.push(format!("✗ {msg}"));
if let Some(sv) = s.servers.iter_mut().find(|x| x.id == id) {
sv.last_error = Some(msg.clone());
}
s.log.push(format!("✗ [{id}] {msg}"));
}
fn set_child(&self, child: Child) {
*self.child.lock().unwrap() = Some(child);
self.status.lock().unwrap().running = true;
/// Seed the per-server status slots from the effective server list.
fn init_servers(&self, servers: &[CmfServer], host: &str) {
let mut s = self.status.lock().unwrap();
s.enabled = true;
s.servers = servers
.iter()
.map(|sv| CmfServerStatus {
id: sv.id.clone(),
model: sv.model.clone(),
port: sv.port,
base_url: format!("http://{host}:{}/v1", sv.port),
running: false,
healthy: false,
last_error: None,
})
.collect();
}
/// Stop the managed server (used on shutdown / reconfigure).
fn set_running(&self, id: &str, running: bool) {
if let Some(sv) = self
.status
.lock()
.unwrap()
.servers
.iter_mut()
.find(|x| x.id == id)
{
sv.running = running;
}
}
fn set_healthy(&self, id: &str, healthy: bool) {
if let Some(sv) = self
.status
.lock()
.unwrap()
.servers
.iter_mut()
.find(|x| x.id == id)
{
sv.healthy = healthy;
}
}
fn add_child(&self, id: String, child: Child) {
self.children.lock().unwrap().push(Managed {
id: id.clone(),
child,
});
self.set_running(&id, true);
}
fn is_running(&self, id: &str) -> bool {
self.status
.lock()
.unwrap()
.servers
.iter()
.any(|s| s.id == id && s.running)
}
/// Stop all managed servers (used on shutdown / reconfigure).
pub async fn stop(&self) {
if let Some(mut child) = self.child.lock().unwrap().take() {
let _ = child.start_kill();
let mut kids = std::mem::take(&mut *self.children.lock().unwrap());
for m in kids.iter_mut() {
let _ = m.child.start_kill();
}
let mut s = self.status.lock().unwrap();
s.running = false;
s.healthy = false;
for sv in s.servers.iter_mut() {
sv.running = false;
sv.healthy = false;
}
}
/// Stop and forget a single managed server (used when a model is deleted).
pub async fn stop_one(&self, id: &str) {
let removed = {
let mut kids = self.children.lock().unwrap();
kids.iter()
.position(|m| m.id == id)
.map(|pos| kids.remove(pos))
};
if let Some(mut m) = removed {
let _ = m.child.start_kill();
}
self.status.lock().unwrap().servers.retain(|sv| sv.id != id);
}
}
@@ -92,7 +192,7 @@ impl Default for CmfRuntime {
fn default() -> Self {
Self {
status: Mutex::new(CmfStatus::default()),
child: Mutex::new(None),
children: Mutex::new(Vec::new()),
}
}
}
@@ -175,24 +275,24 @@ pub async fn install_now(rt: Arc<CmfRuntime>, bin: String) {
}
}
/// Spawn `cortiq serve <model> --host <host> --port <port>`; the child is killed
/// when its handle is dropped (i.e. when the gateway shuts down).
fn spawn_serve(cfg: &CmfCfg) -> Result<Child, String> {
/// Spawn `cortiq serve <model> --host <host> --port <port>` for one server; the
/// child is killed when its handle is dropped (i.e. when the gateway shuts down).
fn spawn_serve(server: &CmfServer, cfg: &CmfCfg) -> Result<Child, String> {
let mut cmd = Command::new(&cfg.cortiq_bin);
cmd.arg("serve")
.arg(&cfg.local_model)
.arg(&server.model)
.arg("--host")
.arg(&cfg.local_host)
.arg("--port")
.arg(cfg.local_port.to_string());
// Parallelise decode matvecs across cores (CMF_THREADS). 0 = leave the
.arg(server.port.to_string());
// Parallelise decode matvecs across cores (CMF_THREADS). 0/1 = leave the
// runtime's own default; N>1 gave ~2.8x on a 15B CPU model.
if cfg.threads > 1 {
cmd.env("CMF_THREADS", cfg.threads.to_string());
if server.threads > 1 {
cmd.env("CMF_THREADS", server.threads.to_string());
}
// Offload large matvecs to the GPU (Metal on macOS). Memory-bandwidth-bound
// on unified memory, so mostly a prefill win; opt-in.
if cfg.gpu {
if server.gpu {
cmd.env("CMF_GPU", "1");
}
cmd.stdout(Stdio::null())
@@ -222,24 +322,17 @@ async fn wait_healthy(host: &str, port: u16, tries: u32) -> bool {
false
}
/// Orchestrate the whole lifecycle: ensure installed → maybe update → spawn →
/// wait healthy. Safe to call in a background task; it never panics.
/// Orchestrate the whole lifecycle: ensure installed → maybe update → spawn each
/// configured local model → wait healthy. Safe to call in a background task; it
/// never panics.
pub async fn manage(rt: Arc<CmfRuntime>, cfg: CmfCfg) {
if !cfg.manage_server || cfg.local_model.trim().is_empty() {
return;
}
{
let mut s = rt.status.lock().unwrap();
s.enabled = true;
s.base_url = format!("http://{}:{}/v1", cfg.local_host, cfg.local_port);
s.model = cfg.local_model.clone();
}
if !std::path::Path::new(&cfg.local_model).exists() {
rt.fail(format!("local_model not found: {}", cfg.local_model));
let servers = cfg.effective_servers();
if servers.is_empty() {
return;
}
rt.init_servers(&servers, &cfg.local_host);
// 1. Ensure the `cortiq` binary is available.
// 1. Ensure the `cortiq` binary is available (once for all servers).
let mut ver = installed_version(&cfg.cortiq_bin);
if ver.is_none() {
if cfg.auto_install {
@@ -285,27 +378,38 @@ pub async fn manage(rt: Arc<CmfRuntime>, cfg: CmfCfg) {
}
}
// 3. Spawn the local server.
rt.log(format!(
"starting: cortiq serve {} --host {} --port {}",
cfg.local_model, cfg.local_host, cfg.local_port
));
match spawn_serve(&cfg) {
Ok(child) => rt.set_child(child),
Err(e) => {
rt.fail(format!("failed to spawn cortiq serve: {e}"));
return;
// 3. Spawn each server (fast).
for server in &servers {
if !std::path::Path::new(&server.model).exists() {
rt.fail_server(
&server.id,
format!("model file not found: {}", server.model),
);
continue;
}
rt.log(format!(
"starting: cortiq serve {} --host {} --port {}",
server.model, cfg.local_host, server.port
));
match spawn_serve(server, &cfg) {
Ok(child) => rt.add_child(server.id.clone(), child),
Err(e) => rt.fail_server(&server.id, format!("failed to spawn: {e}")),
}
}
// 4. Wait until it is serving (model load can take a while for big models).
if wait_healthy(&cfg.local_host, cfg.local_port, 240).await {
rt.status.lock().unwrap().healthy = true;
rt.log(format!(
"local CMF server ready — registered as model '{}' at http://{}:{}/v1",
cfg.model_id, cfg.local_host, cfg.local_port
));
} else {
rt.fail("local CMF server did not become healthy in time");
// 4. Wait until each spawned server is serving (model load can be slow).
for server in &servers {
if !rt.is_running(&server.id) {
continue; // failed to spawn / file missing
}
if wait_healthy(&cfg.local_host, server.port, 240).await {
rt.set_healthy(&server.id, true);
rt.log(format!(
"local CMF server ready — '{}' at http://{}:{}/v1",
server.id, cfg.local_host, server.port
));
} else {
rt.fail_server(&server.id, "did not become healthy in time");
}
}
}
+59 -3
View File
@@ -67,6 +67,24 @@ pub struct Config {
pub cmf: CmfCfg,
}
/// One managed local model: the gateway runs a dedicated `cortiq serve` for it
/// on its own port and registers it in the pool under `id`.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct CmfServer {
/// Pool id it is registered under (e.g. "cmf-qwen3-5-2b").
pub id: String,
/// Path to the `.cmf` to serve.
pub model: String,
/// TCP port the dedicated `cortiq serve` binds to (must be unique).
pub port: u16,
/// Worker threads (CMF_THREADS). 0 = runtime default.
#[serde(default = "default_cmf_threads")]
pub threads: usize,
/// Offload matvecs to the GPU (CMF_GPU=1).
#[serde(default)]
pub gpu: bool,
}
/// CMF-format model factory: import a HuggingFace model, convert it to a
/// local `.cmf` (quantized), and serve it via cortiq-server. Points the
/// gateway at the Python converter + runtime on the host.
@@ -129,6 +147,35 @@ pub struct CmfCfg {
/// (helps prefill of long prompts). Off by default.
#[serde(default)]
pub gpu: bool,
/// Managed local models. When non-empty (and `manage_server` is on), the
/// gateway spawns and supervises one `cortiq serve` per entry — each on its
/// own port — and registers each in the pool. Empty = fall back to the single
/// legacy `local_model` / `local_port` / `model_id` below (back-compat).
#[serde(default)]
pub servers: Vec<CmfServer>,
}
impl CmfCfg {
/// The managed local models actually in effect: the `servers` list when set,
/// otherwise the single legacy `local_model` (as a one-element list). Empty
/// when `manage_server` is off or nothing is configured.
pub fn effective_servers(&self) -> Vec<CmfServer> {
if !self.manage_server {
return Vec::new();
}
if !self.servers.is_empty() {
return self.servers.clone();
}
if !self.local_model.trim().is_empty() {
return vec![CmfServer {
id: self.model_id.clone(),
model: self.local_model.clone(),
port: self.local_port,
threads: self.threads,
gpu: self.gpu,
}];
}
Vec::new()
}
}
impl Default for CmfCfg {
fn default() -> Self {
@@ -149,6 +196,7 @@ impl Default for CmfCfg {
local_only: false,
threads: default_cmf_threads(),
gpu: false,
servers: Vec::new(),
}
}
}
@@ -605,7 +653,8 @@ impl Config {
// A managed local CMF model counts as an available model even though it
// is registered dynamically rather than via a [[models]] entry — this
// is what lets the gateway run on local models alone (no cloud, no router).
let has_local = self.cmf.manage_server && !self.cmf.local_model.trim().is_empty();
let servers = self.cmf.effective_servers();
let has_local = !servers.is_empty();
if self.models.is_empty() && !has_local {
// Fresh install: start with an EMPTY model pool so the admin console
// comes up — the user adds models in /admin → Models (or points [cmf]
@@ -624,8 +673,15 @@ impl Config {
// every target in [routing] must exist in the model pool
let mut ids: std::collections::HashSet<&str> =
self.models.iter().map(|m| m.id.as_str()).collect();
if has_local {
ids.insert(self.cmf.model_id.as_str());
// managed local servers are known ids too; also enforce unique ports/ids
let mut ports = std::collections::HashSet::new();
for s in &servers {
if !ids.insert(s.id.as_str()) {
anyhow::bail!("config: duplicate managed model id '{}'", s.id);
}
if !ports.insert(s.port) {
anyhow::bail!("config: two managed models share port {}", s.port);
}
}
if !self.routing.default.is_empty() && !ids.contains(self.routing.default.as_str()) {
anyhow::bail!(
+9 -9
View File
@@ -28,17 +28,17 @@ impl Registry {
by_id.insert(m.id.clone(), provider);
}
// Managed local CMF server: register it as an OpenAI-compatible provider
// so it can be routed to like any other backend — including when the
// router is unavailable or the gateway is in local-only mode.
if cfg.cmf.manage_server
&& !cfg.cmf.local_model.trim().is_empty()
&& !by_id.contains_key(&cfg.cmf.model_id)
{
// Managed local CMF servers: register each as an OpenAI-compatible
// provider so it can be routed to like any other backend — including when
// the router is unavailable or the gateway is in local-only mode.
for s in cfg.cmf.effective_servers() {
if by_id.contains_key(&s.id) {
continue; // a static [[models]] entry with the same id wins
}
let m = ModelCfg {
id: cfg.cmf.model_id.clone(),
id: s.id.clone(),
provider: "openai".into(),
base_url: format!("http://{}:{}/v1", cfg.cmf.local_host, cfg.cmf.local_port),
base_url: format!("http://{}:{}/v1", cfg.cmf.local_host, s.port),
model: "cortiq".into(),
cost_tier: "local".into(),
price_in: 0.0,
+2 -2
View File
@@ -49,8 +49,8 @@ impl Live {
/// (local-only / disabled) or unavailable, so CMF models serve with no router.
pub fn local_candidates(&self) -> Vec<String> {
let mut v = Vec::new();
if self.cfg.cmf.manage_server && !self.cfg.cmf.local_model.trim().is_empty() {
v.push(self.cfg.cmf.model_id.clone());
for s in self.cfg.cmf.effective_servers() {
v.push(s.id);
}
let def = self.routing.default_model().to_string();
if !def.is_empty() && !v.contains(&def) {