diff --git a/src/admin/mod.rs b/src/admin/mod.rs index 8916466..57a7036 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -8,8 +8,9 @@ pub mod assets; use crate::config::{ - AdminCfg, ApiKeyCfg, BreakerCfg, CacheCfg, Config, CortiqCfg, LogCfg, ModelCfg, ProtocolsCfg, - RouteCfg, RouterCfg, RoutingCfg, RoutingPolicy, StatsCfg, TelemetryCfg, TierTargets, + AdminCfg, ApiKeyCfg, BreakerCfg, CacheCfg, CmfCfg, Config, CortiqCfg, LogCfg, ModelCfg, + ProtocolsCfg, RouteCfg, RouterCfg, RoutingCfg, RoutingPolicy, StatsCfg, TelemetryCfg, + TierTargets, }; use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective}; use crate::state::SharedState; @@ -142,6 +143,7 @@ fn parse_routing(model: &str) -> RoutingDirective { pub fn api_routes(admin_token: String) -> Router { Router::new() .route("/admin/api/health", get(health)) + .route("/admin/api/router/probe", post(router_probe)) .route("/admin/api/meta", get(meta)) .route("/admin/api/config", get(get_config).put(put_config)) .route("/admin/api/models", get(list_models).post(create_model)) @@ -163,6 +165,12 @@ pub fn api_routes(admin_token: String) -> Router { get(list_secrets).put(set_secret).delete(clear_secret), ) .route("/admin/api/stats", get(get_stats)) + .route("/admin/api/shadow", get(get_shadow)) + .route("/admin/api/hf/search", get(hf_search)) + .route("/admin/api/import", get(list_imports).post(start_import)) + .route("/admin/api/import/:job", get(import_status).delete(delete_import)) + .route("/admin/api/import/:job/cancel", post(cancel_import)) + .route("/admin/api/import/:job/register", post(register_import)) .route("/admin/api/requests", get(get_requests)) .route("/admin/api/test", post(run_test)) .route("/admin/api/test/stream", post(run_test_stream)) @@ -225,12 +233,14 @@ async fn probe_router(url: &str) -> (bool, u64) { Err(_) => return (false, 0), }; let started = Instant::now(); - let target = format!("{}/healthz", url.trim_end_matches('/')); + // cortiq-router exposes health at /v1/healthz (auth-exempt); a bare /healthz + // hits the auth middleware and 401s, which would read as a false "reachable". + let target = format!("{}/v1/healthz", url.trim_end_matches('/')); let ok = client .get(&target) .send() .await - .map(|r| r.status().as_u16() < 500) + .map(|r| r.status().is_success()) .unwrap_or(false); (ok, started.elapsed().as_millis() as u64) } @@ -238,6 +248,18 @@ async fn probe_router(url: &str) -> (bool, u64) { async fn health(State(state): State) -> ApiResult { let live = state.live(); let (router_ok, router_ms) = probe_router(&live.cfg.router.url).await; + let key_source = match &live.cfg.router.api_key_env { + Some(env) => state.secrets.source(env), + None => "none", + }; + let last = state + .router_status + .load(std::sync::atomic::Ordering::Relaxed); + let last_error = if last == 0 { + Value::Null + } else { + json!({ "kind": crate::router_client::classify_status(last), "http": last }) + }; let models: Vec = live .cfg .models @@ -262,11 +284,105 @@ async fn health(State(state): State) -> ApiResult { "ok": true, "version": env!("CARGO_PKG_VERSION"), "listen": live.cfg.listen, - "router": { "url": live.cfg.router.url, "reachable": router_ok, "latency_ms": router_ms }, + "router": { + "url": live.cfg.router.url, + "reachable": router_ok, + "latency_ms": router_ms, + "key_env": live.cfg.router.api_key_env, + "key_source": key_source, + "last_error": last_error, + }, "models": models, })) } +/// `POST /admin/api/router/probe` — a real (authenticated) `/v1/route` call so the +/// panel can tell "key missing / rejected / out of quota" apart from "router down". +/// User-triggered only: it consumes one routing decision. Also fetches +/// `GET /v1/usage` best-effort — not part of the documented contract yet; the +/// panel shows balance/usage automatically once the router starts serving it. +async fn router_probe(State(state): State) -> ApiResult { + let live = state.live(); + let rcfg = live.cfg.router.clone(); + let key = rcfg + .api_key_env + .as_deref() + .and_then(|n| state.secrets.resolve(n)); + if rcfg.api_key_env.is_some() && key.is_none() { + return ok(json!({ "ok": false, "status": "no_key" })); + } + + let mut builder = reqwest::Client::builder() + .timeout(std::time::Duration::from_millis(rcfg.timeout_ms.max(5000))); + if !rcfg.verify_tls { + builder = builder + .danger_accept_invalid_certs(true) + .danger_accept_invalid_hostnames(true); + } + let client = builder.build().map_err(|e| ApiError::bad(e.to_string()))?; + let url = rcfg.url.trim_end_matches('/').to_string(); + + let mut body = json!({ + "input": { "text": "ping" }, + "options": { "policy_profile": "balanced", "allow_oracle": false } + }); + if let Some(tax) = &rcfg.taxonomy_id { + body["taxonomy_id"] = json!(tax); + } + let mut req = client.post(format!("{url}/v1/route")).json(&body); + if let Some(k) = &key { + req = req.bearer_auth(k); + } + + let started = Instant::now(); + let resp = match req.send().await { + Ok(r) => r, + Err(e) => { + let status = if e.is_timeout() { + "timeout" + } else { + "unreachable" + }; + return ok(json!({ + "ok": false, + "status": status, + "latency_ms": started.elapsed().as_millis() as u64, + })); + } + }; + let latency_ms = started.elapsed().as_millis() as u64; + let code = resp.status().as_u16(); + if !resp.status().is_success() { + let message = resp + .json::() + .await + .ok() + .and_then(|v| v["error"]["message"].as_str().map(|s| s.to_string())); + return ok(json!({ + "ok": false, + "status": crate::router_client::classify_status(code), + "http": code, + "latency_ms": latency_ms, + "message": message, + })); + } + + let mut usage = Value::Null; + if let Some(k) = &key { + if let Ok(r) = client + .get(format!("{url}/v1/usage")) + .bearer_auth(k) + .send() + .await + { + if r.status().is_success() { + usage = r.json::().await.unwrap_or(Value::Null); + } + } + } + ok(json!({ "ok": true, "status": "ok", "latency_ms": latency_ms, "usage": usage })) +} + async fn meta() -> ApiResult { ok(json!({ "version": env!("CARGO_PKG_VERSION"), @@ -478,6 +594,7 @@ async fn get_settings(State(state): State) -> ApiResult { "stats": c.stats, "admin": c.admin, "cache": c.cache, + "cmf": c.cmf, })) } @@ -493,6 +610,7 @@ struct SettingsBody { stats: Option, admin: Option, cache: Option, + cmf: Option, } async fn put_settings(State(state): State, Json(b): Json) -> ApiResult { @@ -536,6 +654,14 @@ async fn put_settings(State(state): State, Json(b): Json, Query(q): Query ok(snap) } +/// Self-warming shadow loop: per-task-type promotion state, Wilson-LB of the +/// judged pass-rate, and whether the local model is serving it yet. +async fn get_shadow(State(state): State) -> ApiResult { + let labels: Vec = state + .promotion + .snapshot() + .into_iter() + .map(|(label, st, n, pass_rate, lb)| { + json!({ + "label": label, + "state": format!("{st:?}"), + "n": n, + "pass_rate": pass_rate, + "wilson_lb": lb, + "serves_local": state.promotion.serves_local(&label), + }) + }) + .collect(); + ok(json!({ "enabled": state.promotion.enabled(), "labels": labels })) +} + +// ── CMF model factory: HF search → convert → register ── + +#[derive(Deserialize)] +struct HfQuery { + q: Option, + limit: Option, +} + +/// Proxy HuggingFace model search (server-side: avoids CORS, adds token). +async fn hf_search(State(state): State, Query(q): Query) -> ApiResult { + let live = state.live(); + let token = if live.cfg.cmf.hf_token_env.is_empty() { + None + } else { + state.secrets.resolve(&live.cfg.cmf.hf_token_env) + }; + let query = q.q.unwrap_or_default(); + let limit = q.limit.unwrap_or(24).min(50); + let models = crate::import::hf_search(&query, limit, token.as_deref()) + .await + .map_err(ApiError::bad)?; + ok(json!({ "models": models })) +} + +async fn list_imports(State(state): State) -> ApiResult { + ok(json!({ "jobs": state.imports.list() })) +} + +/// Start a conversion job (returns the job id; progress via `import/:job`). +async fn start_import( + State(state): State, + Json(p): Json, +) -> ApiResult { + let live = state.live(); + let id = crate::import::start_import(state.imports.clone(), &live.cfg.cmf, p) + .map_err(ApiError::bad)?; + ok(json!({ "job": id })) +} + +async fn import_status( + State(state): State, + Path(job): Path, +) -> ApiResult { + match state.imports.get(&job) { + Some(j) => ok(serde_json::to_value(j).unwrap_or_else(|_| json!({}))), + None => Err(ApiError::not_found(format!("job '{job}' not found"))), + } +} + +/// Cancel a running conversion: kills the converter and removes partial output. +async fn cancel_import( + State(state): State, + Path(job): Path, +) -> ApiResult { + if state.imports.get(&job).is_none() { + return Err(ApiError::not_found(format!("job '{job}' not found"))); + } + let cancelled = state.imports.cancel(&job); + ok(json!({ "ok": cancelled })) +} + +/// Delete a finished conversion and its converted `.cmf` file(s) from disk. +async fn delete_import( + State(state): State, + Path(job): Path, +) -> ApiResult { + if state.imports.get(&job).is_none() { + return Err(ApiError::not_found(format!("job '{job}' not found"))); + } + let removed = state.imports.delete(&job).map_err(ApiError::bad)?; + ok(json!({ "ok": removed })) +} + +/// Register a finished `.cmf` as a local model (OpenAI provider → cortiq-server). +async fn register_import( + State(state): State, + Path(job): Path, +) -> ApiResult { + let j = state + .imports + .get(&job) + .ok_or_else(|| ApiError::not_found(format!("job '{job}' not found")))?; + if j.state != "done" { + return Err(ApiError::bad("conversion not finished")); + } + let base = std::path::Path::new(&j.output) + .file_stem() + .map(|s| s.to_string_lossy().to_string()) + .unwrap_or_else(|| "cmf-local".into()); + let id = format!("cmf-{base}"); + let mut cfg = current_cfg(&state); + if cfg.models.iter().any(|m| m.id == id) { + return Err(ApiError::bad(format!("model '{id}' already registered"))); + } + cfg.models.push(crate::config::ModelCfg { + id: id.clone(), + provider: "openai".into(), + base_url: cfg.cmf.cortiq_server_url.clone(), + model: base, + cost_tier: "local".into(), + price_in: 0.0, + price_out: 0.0, + kind: "chat".into(), + api_key_env: None, + caps: Vec::new(), + }); + state.reload(cfg)?; + ok(json!({ "ok": true, "model_id": id })) +} + #[derive(Deserialize)] struct RecentQuery { #[serde(default)] diff --git a/src/cmf_runtime.rs b/src/cmf_runtime.rs new file mode 100644 index 0000000..d17604e --- /dev/null +++ b/src/cmf_runtime.rs @@ -0,0 +1,269 @@ +//! Local CMF model server lifecycle. +//! +//! Closes the loop with : 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. + +use crate::config::CmfCfg; +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). +#[derive(Clone, Debug, Default, serde::Serialize)] +pub struct CmfStatus { + /// `manage_server` is on and a `local_model` is configured. + pub enabled: bool, + /// Version reported by `cortiq --version` (None = not installed). + pub installed_version: Option, + /// Newest `cortiq-cli` on crates.io (populated when `auto_update` runs). + pub latest_version: Option, + /// 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, + /// Recent lifecycle log lines (bounded). + pub log: Vec, +} + +/// Holds the managed child process and its live status. +pub struct CmfRuntime { + status: Mutex, + child: Mutex>, +} + +impl CmfRuntime { + pub fn new() -> Arc { + Arc::new(Self { + status: Mutex::new(CmfStatus::default()), + child: Mutex::new(None), + }) + } + + pub fn status(&self) -> CmfStatus { + self.status.lock().unwrap().clone() + } + + fn log(&self, msg: impl Into) { + let msg = msg.into(); + tracing::info!(target: "cmf", "{msg}"); + 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); + } + } + + fn fail(&self, msg: impl Into) { + let msg = msg.into(); + tracing::warn!(target: "cmf", "{msg}"); + let mut s = self.status.lock().unwrap(); + s.last_error = Some(msg.clone()); + s.log.push(format!("✗ {msg}")); + } + + fn set_child(&self, child: Child) { + *self.child.lock().unwrap() = Some(child); + self.status.lock().unwrap().running = true; + } + + /// Stop the managed server (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 s = self.status.lock().unwrap(); + s.running = false; + s.healthy = false; + } +} + +impl Default for CmfRuntime { + fn default() -> Self { + Self { + status: Mutex::new(CmfStatus::default()), + child: Mutex::new(None), + } + } +} + +/// crates.io `newest_version` for `cortiq-cli`. +async fn latest_crates_version() -> Option { + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + .ok()?; + let body = client + .get("https://crates.io/api/v1/crates/cortiq-cli") + .header("User-Agent", "cortiq-gateway") + .send() + .await + .ok()? + .text() + .await + .ok()?; + let v: serde_json::Value = serde_json::from_str(&body).ok()?; + v["crate"]["newest_version"] + .as_str() + .map(|s| s.to_string()) +} + +/// ` --version` → the trailing version token (e.g. "cortiq 0.1.2" → "0.1.2"). +fn installed_version(bin: &str) -> Option { + let out = std::process::Command::new(bin).arg("--version").output().ok()?; + if !out.status.success() { + return None; + } + let s = String::from_utf8_lossy(&out.stdout); + s.split_whitespace().last().map(|s| s.trim().to_string()) +} + +fn parse_ver(s: &str) -> (u32, u32, u32) { + let mut it = s.split('.').map(|x| x.trim().parse::().unwrap_or(0)); + (it.next().unwrap_or(0), it.next().unwrap_or(0), it.next().unwrap_or(0)) +} + +fn version_lt(a: &str, b: &str) -> bool { + parse_ver(a) < parse_ver(b) +} + +/// `cargo install cortiq-cli [--force]` — blocking compile, run off-thread. +async fn cargo_install(force: bool) -> Result<(), String> { + let mut cmd = Command::new("cargo"); + cmd.args(["install", "cortiq-cli", "--locked"]); + if force { + cmd.arg("--force"); + } + let out = cmd.output().await.map_err(|e| e.to_string())?; + if out.status.success() { + Ok(()) + } else { + Err(String::from_utf8_lossy(&out.stderr).trim().to_string()) + } +} + +/// Spawn `cortiq serve --host --port `; the child is killed +/// when its handle is dropped (i.e. when the gateway shuts down). +fn spawn_serve(cfg: &CmfCfg) -> Result { + Command::new(&cfg.cortiq_bin) + .arg("serve") + .arg(&cfg.local_model) + .arg("--host") + .arg(&cfg.local_host) + .arg("--port") + .arg(cfg.local_port.to_string()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .kill_on_drop(true) + .spawn() + .map_err(|e| e.to_string()) +} + +/// Poll `/healthz` until the local server answers 200 (or the deadline passes). +async fn wait_healthy(host: &str, port: u16, tries: u32) -> bool { + let url = format!("http://{host}:{port}/healthz"); + let client = reqwest::Client::new(); + for _ in 0..tries { + if let Ok(r) = client + .get(&url) + .timeout(Duration::from_secs(2)) + .send() + .await + { + if r.status().is_success() { + return true; + } + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + false +} + +/// Orchestrate the whole lifecycle: ensure installed → maybe update → spawn → +/// wait healthy. Safe to call in a background task; it never panics. +pub async fn manage(rt: Arc, 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)); + return; + } + + // 1. Ensure the `cortiq` binary is available. + let mut ver = installed_version(&cfg.cortiq_bin); + if ver.is_none() { + if cfg.auto_install { + rt.log("cortiq not found — installing cortiq-cli from crates.io (this compiles, may take a few minutes)…"); + if let Err(e) = cargo_install(false).await { + rt.fail(format!("cargo install cortiq-cli failed: {e}")); + return; + } + ver = installed_version(&cfg.cortiq_bin); + if ver.is_some() { + rt.log("installed cortiq-cli"); + } + } else { + rt.fail("cortiq binary not found (set [cmf].auto_install = true, or install cortiq-cli)"); + return; + } + } + rt.status.lock().unwrap().installed_version = ver.clone(); + + // 2. Optionally update to the newest crates.io release. + if cfg.auto_update { + if let Some(latest) = latest_crates_version().await { + rt.status.lock().unwrap().latest_version = Some(latest.clone()); + if ver.as_deref().map(|c| version_lt(c, &latest)).unwrap_or(false) { + rt.log(format!("updating cortiq-cli {} → {latest}…", ver.as_deref().unwrap_or("?"))); + if let Err(e) = cargo_install(true).await { + rt.log(format!("update failed (keeping current): {e}")); + } else { + ver = installed_version(&cfg.cortiq_bin); + rt.status.lock().unwrap().installed_version = ver.clone(); + rt.log(format!("updated to {}", ver.as_deref().unwrap_or("?"))); + } + } + } + } + + // 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; + } + } + + // 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"); + } +} diff --git a/src/config.rs b/src/config.rs index 71397e5..3e48780 100644 --- a/src/config.rs +++ b/src/config.rs @@ -3,6 +3,26 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; +use std::path::PathBuf; + +/// Base directory for gateway-managed data (models, logs). Honors +/// `$CORTIQ_GATEWAY_HOME`, then `$XDG_DATA_HOME/cortiq-gateway`, then +/// `~/.cortiq-gateway` — so defaults work regardless of the process CWD +/// (important once the gateway is installed rather than run from its repo). +pub fn data_dir() -> PathBuf { + for (var, sub) in [("CORTIQ_GATEWAY_HOME", ""), ("XDG_DATA_HOME", "cortiq-gateway"), ("HOME", ".cortiq-gateway")] { + if let Ok(v) = std::env::var(var) { + if !v.is_empty() { + return if sub.is_empty() { PathBuf::from(v) } else { PathBuf::from(v).join(sub) }; + } + } + } + PathBuf::from("data") +} + +fn data_path(name: &str) -> String { + data_dir().join(name).to_string_lossy().into_owned() +} #[derive(Clone, Debug, Serialize, Deserialize)] pub struct Config { @@ -33,10 +53,116 @@ pub struct Config { pub stats: StatsCfg, #[serde(default)] pub cache: CacheCfg, + #[serde(default)] + pub shadow: ShadowCfg, + #[serde(default)] + pub cmf: CmfCfg, +} + +/// 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. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct CmfCfg { + /// Python interpreter that has torch + the converter deps. + #[serde(default = "default_cmf_python")] + pub python_bin: String, + /// Path to `convert_dtgma_to_cmf.py`. + #[serde(default = "default_cmf_converter")] + pub converter: String, + /// Directory where produced `.cmf` files are written. + #[serde(default = "default_cmf_models_dir")] + pub models_dir: String, + /// Base URL of the cortiq-server that serves the `.cmf` (OpenAI /v1), + /// used when registering a converted model as a provider. + #[serde(default = "default_cmf_server_url")] + pub cortiq_server_url: String, + /// Secret name of a HuggingFace token (higher search rate limits + + /// gated/private repos). Empty = anonymous public search. + #[serde(default)] + pub hf_token_env: String, + /// `cortiq` CLI binary for the local server (from `cargo install cortiq-cli`). + #[serde(default = "default_cortiq_bin")] + pub cortiq_bin: String, + /// Install cortiq-cli from crates.io automatically if the binary is missing. + #[serde(default)] + pub auto_install: bool, + /// On startup, check crates.io and reinstall if a newer cortiq-cli exists. + #[serde(default)] + pub auto_update: bool, + /// Spawn and manage a local `cortiq serve` for `local_model`, and register + /// it in the model pool — a local `.cmf` backend the gateway can use even + /// when no external provider (or the router) is available. + #[serde(default)] + pub manage_server: bool, + /// Path to the local `.cmf` to serve (empty = disabled). + #[serde(default)] + pub local_model: String, + /// Host the managed local server binds to. + #[serde(default = "default_cmf_local_host")] + pub local_host: String, + /// Port the managed local server binds to. + #[serde(default = "default_cmf_local_port")] + pub local_port: u16, + /// Model id the local server is registered under in the pool. + #[serde(default = "default_cmf_model_id")] + pub model_id: String, + /// Use ONLY local models: route every request to the local model, ignoring + /// the router and any external providers ("use only local models"). + #[serde(default)] + pub local_only: bool, +} +impl Default for CmfCfg { + fn default() -> Self { + Self { + python_bin: default_cmf_python(), + converter: default_cmf_converter(), + models_dir: default_cmf_models_dir(), + cortiq_server_url: default_cmf_server_url(), + hf_token_env: String::new(), + cortiq_bin: default_cortiq_bin(), + auto_install: false, + auto_update: false, + manage_server: false, + local_model: String::new(), + local_host: default_cmf_local_host(), + local_port: default_cmf_local_port(), + model_id: default_cmf_model_id(), + local_only: false, + } + } +} +fn default_cortiq_bin() -> String { + "cortiq".into() +} +fn default_cmf_local_host() -> String { + "127.0.0.1".into() +} +fn default_cmf_local_port() -> u16 { + 8081 +} +fn default_cmf_model_id() -> String { + "cmf-local".into() +} +fn default_cmf_python() -> String { + "python3".into() +} +fn default_cmf_converter() -> String { + "../cmf/converter/convert_dtgma_to_cmf.py".into() +} +fn default_cmf_models_dir() -> String { + data_path("models") +} +fn default_cmf_server_url() -> String { + "http://127.0.0.1:8081/v1".into() } #[derive(Clone, Debug, Serialize, Deserialize)] pub struct RouterCfg { + /// Master switch — when false the router is never called; the gateway + /// serves from the local model pool (local `.cmf` / configured default). + #[serde(default = "default_router_enabled")] + pub enabled: bool, pub url: String, #[serde(default = "default_router_timeout")] pub timeout_ms: u64, @@ -47,6 +173,9 @@ pub struct RouterCfg { #[serde(default, skip_serializing_if = "Option::is_none")] pub taxonomy_id: Option, } +fn default_router_enabled() -> bool { + true +} fn default_router_timeout() -> u64 { 800 } @@ -313,7 +442,7 @@ fn default_stats_enabled() -> bool { true } fn default_stats_file() -> String { - "config/stats.jsonl".into() + data_path("stats.jsonl") } fn default_stats_ring() -> usize { 500 @@ -322,6 +451,82 @@ fn default_stats_retention() -> String { "7d".into() } +/// Self-warming shadow loop (docs/SELF_WARMING_GATEWAY.md). While a task +/// label is not yet promoted, the client is served the CLOUD answer; in +/// parallel (sampled, non-blocking) the local `cmf-local` model answers and +/// a cheap `judge` model scores it against the cloud answer. The per-label +/// Wilson-LB of that pass stream gates promotion to local serving. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct ShadowCfg { + /// Master switch. Off = zero overhead, classic gateway behavior. + #[serde(default)] + pub enabled: bool, + /// model_id (from `[[models]]`) of the local CMF model (OpenAI-compatible + /// cortiq-server). Empty disables the loop. + #[serde(default)] + pub local_model_id: String, + /// model_id of the cheap judge (different family than the answerer). + #[serde(default)] + pub judge_model_id: String, + /// Fraction of eligible requests to shadow+judge (0..1). Bounds cost. + #[serde(default = "default_shadow_sample")] + pub sample_rate: f64, + /// judge.jsonl append+replay path. Empty = in-memory only. + #[serde(default = "default_shadow_file")] + pub file: String, + /// Wilson-95 lower bound of pass-rate required to promote. + #[serde(default = "default_shadow_lb")] + pub promote_lb: f64, + /// Min judgments before promotion is considered. + #[serde(default = "default_shadow_nmin")] + pub n_min: usize, + /// Rolling window of judgments per label. + #[serde(default = "default_shadow_window")] + pub window: usize, + /// Extra canary judgments before full local serving. + #[serde(default = "default_shadow_soak")] + pub soak: usize, + /// GATE past shadow-only: actually SERVE promoted labels from the local + /// model (with failover to cloud on error + a complexity veto). Default + /// OFF — the memo's 3 risks (§8) should be retired by measurement first. + #[serde(default)] + pub serve_when_promoted: bool, +} +impl Default for ShadowCfg { + fn default() -> Self { + Self { + enabled: false, + local_model_id: String::new(), + judge_model_id: String::new(), + sample_rate: default_shadow_sample(), + file: default_shadow_file(), + promote_lb: default_shadow_lb(), + n_min: default_shadow_nmin(), + window: default_shadow_window(), + soak: default_shadow_soak(), + serve_when_promoted: false, + } + } +} +fn default_shadow_sample() -> f64 { + 0.15 +} +fn default_shadow_file() -> String { + data_path("judge.jsonl") +} +fn default_shadow_lb() -> f64 { + 0.95 +} +fn default_shadow_nmin() -> usize { + 200 +} +fn default_shadow_window() -> usize { + 500 +} +fn default_shadow_soak() -> usize { + 100 +} + /// Semantic (embedding-based) response cache — returns a cached answer for prompts /// that are semantically near a previous one, skipping the (expensive) model call. #[derive(Clone, Debug, Serialize, Deserialize)] @@ -374,8 +579,12 @@ impl Config { /// Integrity validation: non-empty model pool, valid routing targets. pub fn validate(&self) -> anyhow::Result<()> { - if self.models.is_empty() { - anyhow::bail!("config: at least one [[models]] entry is required"); + // 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(); + if self.models.is_empty() && !has_local { + anyhow::bail!("config: at least one [[models]] entry (or a managed local CMF model) is required"); } // uniqueness of model ids let mut seen = std::collections::HashSet::new(); @@ -385,9 +594,12 @@ impl Config { } } // every target in [routing] must exist in the model pool - let ids: std::collections::HashSet<&str> = + let mut ids: std::collections::HashSet<&str> = self.models.iter().map(|m| m.id.as_str()).collect(); - if !ids.contains(self.routing.default.as_str()) { + if has_local { + ids.insert(self.cmf.model_id.as_str()); + } + if !self.routing.default.is_empty() && !ids.contains(self.routing.default.as_str()) { anyhow::bail!( "config: routing.default '{}' is not a known model id", self.routing.default diff --git a/src/import.rs b/src/import.rs new file mode 100644 index 0000000..c7a93df --- /dev/null +++ b/src/import.rs @@ -0,0 +1,390 @@ +//! CMF model factory: search HuggingFace, then convert a chosen repo to a +//! local quantized `.cmf` via the Python converter — as a tracked, streamed +//! background job so the admin UI can show live progress. + +use crate::config::CmfCfg; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::process::Stdio; +use std::sync::{Arc, Mutex}; +use std::time::{SystemTime, UNIX_EPOCH}; +use tokio::io::{AsyncBufReadExt, BufReader}; +use tokio::sync::Notify; + +/// Parse a `@PROGRESS <0..1> ` marker emitted by the converter. +fn parse_progress(line: &str) -> Option<(f32, String)> { + let rest = line.strip_prefix("@PROGRESS ")?; + let mut it = rest.splitn(2, ' '); + let frac: f32 = it.next()?.trim().parse().ok()?; + let phase = it.next().unwrap_or("").trim().to_string(); + Some((frac.clamp(0.0, 1.0), phase)) +} + +/// Remove a partial/aborted output — `.cmf` may be a file or a tensor-dir. +fn cleanup_output(path: &str) { + let p = std::path::Path::new(path); + if p.is_dir() { + let _ = std::fs::remove_dir_all(p); + } else if p.exists() { + let _ = std::fs::remove_file(p); + } +} + +fn now() -> u64 { + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() +} + +/// Job id from a hash of (repo, nanos) — unique without a uuid dependency. +fn gen_id(repo: &str) -> String { + use std::hash::{Hash, Hasher}; + let nanos = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos(); + let mut h = std::collections::hash_map::DefaultHasher::new(); + repo.hash(&mut h); + nanos.hash(&mut h); + format!("{:016x}", h.finish()) +} + +/// Parameters chosen in the import wizard. +#[derive(Clone, Debug, Deserialize)] +pub struct ImportParams { + pub repo: String, + #[serde(default = "default_quant")] + pub quant: String, + #[serde(default)] + pub name: String, // output basename; empty → derived from repo + // ── advanced ── + #[serde(default)] + pub linear_core: Option, // gated_delta_net | vmf_phase + #[serde(default)] + pub nphase: Option, + #[serde(default)] + pub vbit_shape: Option, // log2 | cubic + #[serde(default)] + pub mean_bits: Option, + #[serde(default)] + pub shard_max_gb: Option, + #[serde(default)] + pub skip_mtp: bool, +} +fn default_quant() -> String { + "Q8_2F".into() +} + +/// Live state of one conversion job. +#[derive(Clone, Debug, Serialize)] +pub struct Job { + pub id: String, + pub repo: String, + pub quant: String, + pub output: String, + pub state: String, // running | done | error | cancelled + pub log: Vec, + pub started: u64, + pub finished: Option, + pub size_bytes: Option, + pub progress: Option, // 0..1, parsed from converter @PROGRESS + pub phase: Option, // current phase label +} + +#[derive(Default)] +pub struct JobStore { + jobs: Mutex>, + /// Cancel handles, keyed by job id (not serialized). Notifying aborts the + /// converter process and cleans up its partial output. + cancels: Mutex>>, +} + +impl JobStore { + pub fn new() -> Arc { + Arc::new(Self::default()) + } + fn insert(&self, job: Job) { + self.jobs.lock().unwrap().insert(job.id.clone(), job); + } + fn set_cancel(&self, id: &str, n: Arc) { + self.cancels.lock().unwrap().insert(id.to_string(), n); + } + fn drop_cancel(&self, id: &str) { + self.cancels.lock().unwrap().remove(id); + } + /// Signal a running job to abort. Returns false if the job isn't cancellable. + pub fn cancel(&self, id: &str) -> bool { + let running = matches!( + self.jobs.lock().unwrap().get(id).map(|j| j.state.as_str()), + Some("running") + ); + if !running { + return false; + } + if let Some(n) = self.cancels.lock().unwrap().get(id) { + n.notify_one(); + true + } else { + false + } + } + fn push_line(&self, id: &str, line: String) { + if let Some(j) = self.jobs.lock().unwrap().get_mut(id) { + j.log.push(line); + let n = j.log.len(); + if n > 200 { + j.log.drain(0..n - 200); // keep the tail + } + } + } + fn set_progress(&self, id: &str, frac: f32, phase: String) { + if let Some(j) = self.jobs.lock().unwrap().get_mut(id) { + j.progress = Some(frac); + if !phase.is_empty() { + j.phase = Some(phase); + } + } + } + fn set_state(&self, id: &str, state: &str) { + if let Some(j) = self.jobs.lock().unwrap().get_mut(id) { + j.state = state.into(); + j.finished = Some(now()); + } + } + fn finish(&self, id: &str, ok: bool) { + if let Some(j) = self.jobs.lock().unwrap().get_mut(id) { + if j.state == "cancelled" { + return; // a concurrent cancel already settled this job + } + j.finished = Some(now()); + j.size_bytes = std::fs::metadata(&j.output).ok().map(|m| m.len()); + let done = ok && j.size_bytes.unwrap_or(0) > 0; + j.state = if done { "done" } else { "error" }.into(); + if done { + j.progress = Some(1.0); + } + } + } + pub fn get(&self, id: &str) -> Option { + self.jobs.lock().unwrap().get(id).cloned() + } + /// Delete a finished job together with its converted output file(s). + /// Refuses a running job (cancel it first). `Ok(false)` if unknown. + pub fn delete(&self, id: &str) -> Result { + let job = self.jobs.lock().unwrap().get(id).cloned(); + let Some(job) = job else { return Ok(false) }; + if job.state == "running" { + return Err("cancel the running job before deleting".into()); + } + cleanup_output(&job.output); + self.cancels.lock().unwrap().remove(id); + self.jobs.lock().unwrap().remove(id); + Ok(true) + } + pub fn list(&self) -> Vec { + let mut v: Vec = self.jobs.lock().unwrap().values().cloned().collect(); + v.sort_by(|a, b| b.started.cmp(&a.started)); + v.truncate(20); + v + } +} + +/// Proxy HuggingFace model search (avoids browser CORS + adds our token). +pub async fn hf_search(query: &str, limit: usize, token: Option<&str>) -> Result { + let q: String = url_escape(query); + let sort = if query.trim().is_empty() { "trendingScore" } else { "downloads" }; + let url = format!( + "https://huggingface.co/api/models?search={q}&sort={sort}&direction=-1&limit={limit}&full=false" + ); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(15)) + .build() + .map_err(|e| e.to_string())?; + let mut req = client.get(&url).header("User-Agent", "cortiq-gateway"); + if let Some(t) = token { + req = req.bearer_auth(t); + } + let resp = req.send().await.map_err(|e| e.to_string())?; + if !resp.status().is_success() { + return Err(format!("HuggingFace API {}", resp.status())); + } + resp.json().await.map_err(|e| e.to_string()) +} + +fn url_escape(s: &str) -> String { + s.chars() + .map(|c| match c { + 'a'..='z' | 'A'..='Z' | '0'..='9' | '-' | '_' | '.' | '/' => c.to_string(), + ' ' => "+".to_string(), + _ => format!("%{:02X}", c as u32), + }) + .collect() +} + +fn sanitize(s: &str) -> String { + s.chars() + .map(|c| if c.is_ascii_alphanumeric() || c == '-' || c == '_' { c } else { '-' }) + .collect() +} + +/// Kick off a conversion. Returns the job id immediately; progress streams +/// into the job log. Fails fast if the converter script is missing. +pub fn start_import(store: Arc, cfg: &CmfCfg, p: ImportParams) -> Result { + if p.repo.trim().is_empty() { + return Err("empty repo id".into()); + } + // Native `cortiq convert` (from crates.io, no Python) handles standard + MoE + // models; the bundled Python converter is used only for advanced options it + // doesn't support yet (linear-attention folding, v-bit shaping). + let use_python = p.linear_core.is_some() || p.vbit_shape.is_some() || p.mean_bits.is_some(); + if use_python && !std::path::Path::new(&cfg.converter).exists() { + return Err(format!("advanced options need the Python converter, not found: {}", cfg.converter)); + } + let _ = std::fs::create_dir_all(&cfg.models_dir); + let base = if p.name.trim().is_empty() { + sanitize(p.repo.rsplit('/').next().unwrap_or(&p.repo)) + } else { + sanitize(&p.name) + }; + let output = std::path::Path::new(&cfg.models_dir) + .join(format!("{base}.cmf")) + .to_string_lossy() + .to_string(); + let output_abs = std::fs::canonicalize(&cfg.models_dir) + .map(|d| d.join(format!("{base}.cmf")).to_string_lossy().to_string()) + .unwrap_or_else(|_| output.clone()); + + let id = gen_id(&p.repo); + store.insert(Job { + id: id.clone(), + repo: p.repo.clone(), + quant: p.quant.clone(), + output: output_abs.clone(), + state: "running".into(), + log: vec![format!("→ converting {} to {} ({})", p.repo, base, p.quant)], + started: now(), + finished: None, + size_bytes: None, + progress: Some(0.0), + phase: Some("starting".into()), + }); + let cancel = Arc::new(Notify::new()); + store.set_cancel(&id, cancel.clone()); + + let hf_token = if cfg.hf_token_env.is_empty() { + None + } else { + std::env::var(&cfg.hf_token_env).ok() + }; + + // Program + args: `cortiq convert` by default, or the Python converter for + // advanced options. Both stream `@PROGRESS ` markers into the log. + let (program, args, workdir): (String, Vec, std::path::PathBuf) = if use_python { + let conv = std::path::Path::new(&cfg.converter); + let mut a = vec![ + std::fs::canonicalize(conv).map(|p| p.to_string_lossy().to_string()).unwrap_or_else(|_| cfg.converter.clone()), + "--model".into(), p.repo.clone(), + "--quant".into(), p.quant.clone(), + "--output".into(), output_abs.clone(), + ]; + if let Some(lc) = &p.linear_core { a.push("--linear-core".into()); a.push(lc.clone()); } + if let Some(n) = p.nphase { a.push("--nphase".into()); a.push(n.to_string()); } + if let Some(vs) = &p.vbit_shape { a.push("--vbit-shape".into()); a.push(vs.clone()); } + if let Some(mb) = p.mean_bits { a.push("--mean-bits".into()); a.push(mb.to_string()); } + if let Some(g) = p.shard_max_gb { a.push("--shard-max-gb".into()); a.push(g.to_string()); } + if p.skip_mtp { a.push("--skip-mtp".into()); } + let wd = conv.parent().map(|d| d.to_path_buf()).unwrap_or_else(|| std::path::PathBuf::from(".")); + (cfg.python_bin.clone(), a, wd) + } else { + // native Rust converter — quant name maps to cortiq's scheme. + let quant = match p.quant.to_ascii_uppercase().as_str() { + s if s.starts_with("Q4") => "q4", + "Q8_2F" => "q8_2f", + "F16" | "FP16" => "f16", + _ => "q8", + }; + let mut a = vec![ + "convert".into(), + "--model".into(), p.repo.clone(), + "--quant".into(), quant.into(), + "--output".into(), output_abs.clone(), + ]; + if let Some(t) = &hf_token { a.push("--hf-token".into()); a.push(t.clone()); } + (cfg.cortiq_bin.clone(), a, std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."))) + }; + + let ret_id = id.clone(); + let output_for_task = output_abs.clone(); + tokio::spawn(async move { + let mut cmd = tokio::process::Command::new(&program); + cmd.args(&args) + .current_dir(&workdir) + .env("PYTHONUNBUFFERED", "1") + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + // HF token for the Python converter (native cortiq gets --hf-token). + if use_python { + if let Some(tok) = hf_token { + cmd.env("HF_TOKEN", &tok).env("HUGGING_FACE_HUB_TOKEN", tok); + } + } + let mut child = match cmd.spawn() { + Ok(c) => c, + Err(e) => { + store.push_line(&id, format!("✗ spawn failed: {e}")); + store.finish(&id, false); + store.drop_cancel(&id); + return; + } + }; + let stdout = child.stdout.take(); + let stderr = child.stderr.take(); + // stdout: parse @PROGRESS markers into progress/phase; everything else logs. + let s1 = store.clone(); + let id1 = id.clone(); + let t_out = tokio::spawn(async move { + if let Some(o) = stdout { + let mut lines = BufReader::new(o).lines(); + while let Ok(Some(l)) = lines.next_line().await { + if let Some((frac, phase)) = parse_progress(&l) { + s1.set_progress(&id1, frac, phase); + } else { + s1.push_line(&id1, l); + } + } + } + }); + let s2 = store.clone(); + let id2 = id.clone(); + let t_err = tokio::spawn(async move { + if let Some(e) = stderr { + let mut lines = BufReader::new(e).lines(); + while let Ok(Some(l)) = lines.next_line().await { + s2.push_line(&id2, l); + } + } + }); + + tokio::select! { + status = child.wait() => { + let _ = tokio::join!(t_out, t_err); + let ok = status.map(|s| s.success()).unwrap_or(false); + if !ok { + cleanup_output(&output_for_task); // drop partial/corrupt output + } + store.push_line( + &id, + if ok { "✓ done".into() } else { "✗ converter exited with error".into() }, + ); + store.finish(&id, ok); + } + _ = cancel.notified() => { + let _ = child.start_kill(); + let _ = child.wait().await; + t_out.abort(); + t_err.abort(); + cleanup_output(&output_for_task); + store.push_line(&id, "✗ cancelled by user — partial output removed".into()); + store.set_state(&id, "cancelled"); + } + } + store.drop_cancel(&id); + }); + + Ok(ret_id) +} diff --git a/src/main.rs b/src/main.rs index 3eede89..6c92bf1 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,10 +10,13 @@ mod admin; mod cache; +mod cmf_runtime; mod config; mod error; +mod import; mod model; mod pipeline; +mod promotion; mod protocols; mod providers; mod registry; @@ -79,6 +82,16 @@ async fn main() -> anyhow::Result<()> { let state = SharedState::build(cfg, args.config.clone())?; + // Close the loop with the CMF format (github.com/infosave2007/cmf): install + // / update cortiq-cli and run a local `cortiq serve` if configured. Runs in + // the background so startup is never blocked by a cargo install or a model + // load; the served model is registered in the pool and usable offline. + { + let cmf_rt = state.cmf.clone(); + let cmf_cfg = state.live().cfg.cmf.clone(); + tokio::spawn(async move { cmf_runtime::manage(cmf_rt, cmf_cfg).await }); + } + let mut app = Router::new() .route("/healthz", get(|| async { "ok" })) .route("/readyz", get(|| async { "ready" })) diff --git a/src/pipeline.rs b/src/pipeline.rs index 232efa7..b76d8ae 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -4,7 +4,8 @@ use crate::error::{GatewayError, Result}; use crate::model::{ - ChatRequest, ChatResponse, Choice, Message, RouteDecision, RouteInfo, RoutingDirective, Usage, + ChatRequest, ChatResponse, Choice, GenParams, Message, RequestMeta, RouteDecision, RouteInfo, + RoutingDirective, Usage, }; use crate::state::{Live, SharedState}; use crate::stats::RequestRecord; @@ -67,6 +68,21 @@ impl Pipeline { } } + // Self-warming shadow loop: capture the prompt BEFORE `req` is moved, + // so we can (non-blocking, sampled) judge a local answer after the + // cloud reply. Client latency is unaffected. Nothing is served locally + // here — this only feeds the promotion table. + let shadow_cfg = state.live().cfg.shadow.clone(); + let shadow_prompt = if shadow_cfg.enabled + && !shadow_cfg.local_model_id.is_empty() + && !shadow_cfg.judge_model_id.is_empty() + { + let live = state.live(); + Some(self.extract_text(&req.messages, "last_user", live.cfg.route.max_chars)) + } else { + None + }; + let result = self.run_inner(req, state).await; // record statistics (for both success and error) @@ -135,6 +151,29 @@ impl Pipeline { } } + // Fire the shadow judgment for a sampled subset (non-blocking). + if let (Some(prompt), Ok(resp)) = (shadow_prompt, &result) { + let cloud_answer = resp + .choices + .first() + .map(|c| c.message.content.clone()) + .unwrap_or_default(); + let label = resp.cortiq.task_label.clone(); + // Only judge CLOUD-served answers — the reference must be the cloud + // answer, not a local one (a promoted label served locally would + // otherwise judge local-vs-local). Canary auditing of served-local + // answers (forcing a cloud reference for a fraction) is the next step. + if !cloud_answer.is_empty() + && !label.is_empty() + && label != "cache" + && resp.cortiq.selected_model != shadow_cfg.local_model_id + && should_shadow(&prompt, shadow_cfg.sample_rate) + { + let st = state.0.clone(); + tokio::spawn(shadow_capture(st, shadow_cfg, prompt, cloud_answer, label)); + } + } + result } @@ -152,50 +191,48 @@ impl Pipeline { live.cfg.route.max_chars, ); - // 2. Obtain the router decision (or use the pinned model) - let (decision, route_source) = match &req.routing { - RoutingDirective::Pinned { model_id: _ } => { - let dec = RouteDecision { - task_label: "pinned".to_string(), - complexity_score: 0.0, - complexity_tier: "pinned".to_string(), - router_request_id: None, - source: "pinned".to_string(), - }; - (dec, "pinned".to_string()) - } - RoutingDirective::Auto { profile } => { - let prof = profile.as_deref().unwrap_or(&live.cfg.route.profile); - match live.router.route(&text, prof).await { - Ok(Some(dec)) => { - let src = dec.source.clone(); - (dec, src) - } - Ok(None) | Err(_) => { - tracing::warn!( - "Router is unavailable. Gracefully degrading to default model." - ); - let dec = RouteDecision { - task_label: "degraded".to_string(), - complexity_score: 0.5, - complexity_tier: "degraded".to_string(), - router_request_id: None, - source: "fallback".to_string(), - }; - (dec, "fallback".to_string()) - } - } - } + // 2–3. Decide the route and pick candidate models. + // + // Local-first rules, so CMF models work with no router at all: + // • `[cmf].local_only` → every request goes to the local pool, + // ignoring the router, the pin, and external providers; + // • `[router].enabled = false` → skip the router, use the local pool; + // • router unavailable at runtime → graceful degrade to the local pool; + // • an explicitly pinned model → used directly, no router call. + let mk = |label: &str, source: &str| RouteDecision { + task_label: label.to_string(), + complexity_score: 0.0, + complexity_tier: label.to_string(), + router_request_id: None, + source: source.to_string(), }; - // 3. Select candidates by complexity tier / pinned model - let candidates = match &req.routing { - RoutingDirective::Pinned { model_id } => vec![model_id.clone()], - RoutingDirective::Auto { .. } => { - if route_source == "fallback" { - vec![live.routing.default_model().to_string()] - } else { - live.routing.candidates(&decision.complexity_tier) + let (decision, route_source, candidates) = if live.cfg.cmf.local_only { + (mk("local-only", "local"), "local".to_string(), live.local_candidates()) + } else { + match &req.routing { + RoutingDirective::Pinned { model_id } => { + (mk("pinned", "pinned"), "pinned".to_string(), vec![model_id.clone()]) + } + RoutingDirective::Auto { profile } => { + if !live.cfg.router.enabled { + (mk("router-disabled", "local"), "local".to_string(), live.local_candidates()) + } else { + let prof = profile.as_deref().unwrap_or(&live.cfg.route.profile); + match live.router.route(&text, prof).await { + Ok(Some(dec)) => { + let src = dec.source.clone(); + let tier = dec.complexity_tier.clone(); + (dec, src, live.routing.candidates(&tier)) + } + Ok(None) | Err(_) => { + tracing::warn!( + "Router is unavailable. Gracefully degrading to the local/default model." + ); + (mk("degraded", "fallback"), "fallback".to_string(), live.local_candidates()) + } + } + } } } }; @@ -216,6 +253,19 @@ impl Pipeline { } } + // Local-first safety net: if routing produced no usable candidate (e.g. + // an unknown pinned model while the router is off), fall back to the + // managed local model so a configured local backend still answers. + if final_candidates.is_empty() && live.cfg.cmf.manage_server { + for id in live.local_candidates() { + if let Some(prov) = live.registry.get(&id) { + if needs_tools && !prov.caps().tools { + continue; + } + final_candidates.push(id); + } + } + } if final_candidates.is_empty() { return Err(GatewayError::UpstreamUnavailable( "No healthy model available that satisfies required capabilities (e.g. tools)" @@ -227,7 +277,23 @@ impl Pipeline { async fn run_inner(&self, req: ChatRequest, state: &SharedState) -> Result { let live = state.live(); - let (decision, route_source, final_candidates) = self.resolve(&req, &live).await?; + let (decision, route_source, mut final_candidates) = self.resolve(&req, &live).await?; + + // Serving-when-promoted (gated): a promoted task-label is served by the + // local model FIRST (failover to cloud on error). Per-request VETO uses + // the router's own complexity signal — a "high"-tier request escalates + // to cloud even in a promoted label (the local skill is for the common + // case, not the hard tail). Off by default. + let sh = &live.cfg.shadow; + if sh.serve_when_promoted + && !sh.local_model_id.is_empty() + && matches!(req.routing, RoutingDirective::Auto { .. }) + && state.promotion.serves_local(&decision.task_label) + && !matches!(decision.complexity_tier.as_str(), "high" | "hard" | "complex") + && !final_candidates.iter().any(|m| m == &sh.local_model_id) + { + final_candidates.insert(0, sh.local_model_id.clone()); + } // 4. Attempt provider call with failover (fall back to alternatives on error) let mut last_err = None; @@ -592,3 +658,141 @@ fn tap_stream( }); } } + +// ─────────────────────────── self-warming shadow loop ─────────────────────────── + +/// Deterministic sampling by prompt hash — no RNG dependency, and the same +/// prompt is always (not) sampled, so repeated office templates don't skew +/// the judged set request-by-request. +fn should_shadow(prompt: &str, rate: f64) -> bool { + if rate <= 0.0 { + return false; + } + if rate >= 1.0 { + return true; + } + use std::hash::{Hash, Hasher}; + let mut h = std::collections::hash_map::DefaultHasher::new(); + prompt.hash(&mut h); + (h.finish() % 10_000) < (rate * 10_000.0) as u64 +} + +fn clip(s: &str, max: usize) -> String { + if s.len() <= max { + s.to_string() + } else { + s.chars().take(max).collect() + } +} + +fn shadow_req(model_id: &str, content: String, max_tokens: u32) -> ChatRequest { + ChatRequest { + routing: RoutingDirective::Pinned { + model_id: model_id.to_string(), + }, + messages: vec![Message { + role: "user".to_string(), + content, + tool_calls: Vec::new(), + }], + tools: Vec::new(), + params: GenParams { + temperature: Some(0.0), + max_tokens: Some(max_tokens), + ..Default::default() + }, + stream: false, + meta: RequestMeta::default(), + } +} + +/// Extract `{"score":N,"hard_fail":bool}` from a judge reply, tolerant of +/// surrounding prose. Missing/garbled → (0, true) (fail-closed: an +/// unparseable judgment must never count as a pass). +fn parse_judge(text: &str) -> (u8, bool) { + let score = text + .split("score") + .nth(1) + .and_then(|t| t.chars().skip_while(|c| !c.is_ascii_digit()).next()) + .and_then(|c| c.to_digit(10)) + .map(|d| d.min(4) as u8); + let hard_fail = text + .split("hard_fail") + .nth(1) + .map(|t| t.contains("true")) + .unwrap_or(true); + match score { + Some(s) => (s, hard_fail), + None => (0, true), + } +} + +/// Non-blocking: answer locally, have the cheap judge score it against the +/// already-paid cloud answer, and feed the per-label promotion table. Any +/// failure (missing provider, upstream error, unparseable judgment) is +/// silently dropped — the client already got the cloud answer. +async fn shadow_capture( + st: std::sync::Arc, + cfg: crate::config::ShadowCfg, + prompt: String, + cloud_answer: String, + label: String, +) { + let live = st.live.load_full(); + let (Some(local), Some(judge)) = ( + live.registry.get(&cfg.local_model_id), + live.registry.get(&cfg.judge_model_id), + ) else { + return; + }; + + // 1. Local answer (free, CPU). + let a_local = match local.chat(shadow_req(&cfg.local_model_id, clip(&prompt, 4000), 512)).await { + Ok(r) => r + .choices + .into_iter() + .next() + .map(|c| c.message.content) + .unwrap_or_default(), + Err(_) => return, + }; + if a_local.trim().is_empty() { + return; + } + + // 2. Reference-based judgment (cheap model, different family). + let jp = format!( + "You grade a CANDIDATE answer against a REFERENCE answer for the same task.\n\ + Task type: {label}\n\n[TASK]\n{task}\n\n[REFERENCE]\n{reference}\n\n[CANDIDATE]\n{candidate}\n\n\ + Reply with ONLY compact JSON: {{\"score\":0-4,\"hard_fail\":true|false}}. \ + score>=3 means the candidate is acceptable (correct, complete, right format). \ + hard_fail=true for fabricated facts, broken format, or unsafe content.", + label = label, + task = clip(&prompt, 2000), + reference = clip(&cloud_answer, 3000), + candidate = clip(&a_local, 3000), + ); + let Ok(jr) = judge.chat(shadow_req(&cfg.judge_model_id, jp, 32)).await else { + return; + }; + let judge_tokens = jr.usage.total_tokens; + let jtext = jr + .choices + .into_iter() + .next() + .map(|c| c.message.content) + .unwrap_or_default(); + let (score, hard_fail) = parse_judge(&jtext); + + st.promotion.record(crate::promotion::JudgeRecord { + ts: now_secs(), + task_label: label, + pass: score >= 3 && !hard_fail, + hard_fail, + score, + source: "judge".to_string(), + recon_e: None, + born_conf: None, + judge_tokens, + }); +} diff --git a/src/registry.rs b/src/registry.rs index f52ec3a..e816b27 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -1,7 +1,7 @@ //! Model pool registry: builds providers from `[[models]]` and resolves //! `model_id → Arc`. -use crate::config::Config; +use crate::config::{Config, ModelCfg}; use crate::providers::{anthropic::AnthropicProvider, openai::OpenAiProvider, Provider}; use crate::secrets::SecretStore; use std::collections::HashMap; @@ -27,9 +27,37 @@ 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) + { + let m = ModelCfg { + id: cfg.cmf.model_id.clone(), + provider: "openai".into(), + base_url: format!("http://{}:{}/v1", cfg.cmf.local_host, cfg.cmf.local_port), + model: "cortiq".into(), + cost_tier: "local".into(), + price_in: 0.0, + price_out: 0.0, + kind: "chat".into(), + api_key_env: None, + caps: Vec::new(), + }; + by_id.insert(m.id.clone(), Arc::new(OpenAiProvider::new(&m, None)?)); + } Ok(Self { by_id }) } + /// Any model id, preferring a real registered one (used as a local-first + /// fallback target when routing cannot pick a specific model). + pub fn any_id(&self) -> Option { + self.by_id.keys().next().cloned() + } + pub fn get(&self, model_id: &str) -> Option> { self.by_id.get(model_id).cloned() } diff --git a/src/state.rs b/src/state.rs index c76cb9b..15694a9 100644 --- a/src/state.rs +++ b/src/state.rs @@ -8,7 +8,7 @@ use crate::cache::SemanticCache; use crate::config::Config; use crate::registry::Registry; -use crate::router_client::RouterClient; +use crate::router_client::{RouterClient, RouterLastStatus}; use crate::routing::RoutingTable; use crate::secrets::SecretStore; use crate::stats::Stats; @@ -27,8 +27,12 @@ pub struct Live { } impl Live { - pub fn build(cfg: Config, secrets: &SecretStore) -> anyhow::Result { - let router = RouterClient::new(&cfg.router)?; + pub fn build( + cfg: Config, + secrets: &SecretStore, + router_status: RouterLastStatus, + ) -> anyhow::Result { + let router = RouterClient::new(&cfg.router, secrets, router_status)?; let registry = Registry::from_config(&cfg, secrets)?; let routing = RoutingTable::from_config(&cfg); Ok(Self { @@ -38,15 +42,57 @@ impl Live { router, }) } + + /// Local-first fallback targets, in order: the managed local CMF model (if + /// configured), then the routing default, then any registered model — only + /// registered ids are returned. Used whenever the router is bypassed + /// (local-only / disabled) or unavailable, so CMF models serve with no router. + pub fn local_candidates(&self) -> Vec { + 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()); + } + let def = self.routing.default_model().to_string(); + if !def.is_empty() && !v.contains(&def) { + v.push(def); + } + v.retain(|id| self.registry.get(id).is_some()); + if v.is_empty() { + if let Some(a) = self.registry.any_id() { + v.push(a); + } + } + v + } } pub struct AppState { pub live: ArcSwap, pub stats: Arc, + pub promotion: Arc, + pub imports: Arc, + /// Managed local CMF model server (install/update/spawn lifecycle + status). + pub cmf: Arc, pub cache: Arc, pub secrets: SecretStore, pub config_path: String, pub pipeline: crate::pipeline::Pipeline, + /// Outcome of the most recent router call (survives config reloads) — + /// lets the admin panel distinguish a bad/expired key from a down router. + pub router_status: RouterLastStatus, +} + +/// Map the user-facing `[shadow]` config to the promotion-table tuning. +fn promotion_cfg(s: &crate::config::ShadowCfg) -> crate::promotion::PromotionCfg { + crate::promotion::PromotionCfg { + enabled: s.enabled && !s.local_model_id.is_empty(), + window: s.window, + n_min: s.n_min, + promote_lb: s.promote_lb, + soak: s.soak, + file: if s.file.is_empty() { None } else { Some(s.file.clone()) }, + ..Default::default() + } } impl std::ops::Deref for SharedState { @@ -61,15 +107,23 @@ impl SharedState { let secrets_path = sibling(&config_path, "secrets.toml"); let secrets = SecretStore::load(&secrets_path); let stats = Stats::new(&cfg.stats); + let promotion = crate::promotion::Promotion::new(promotion_cfg(&cfg.shadow)); + let imports = crate::import::JobStore::new(); + let cmf = crate::cmf_runtime::CmfRuntime::new(); let cache = SemanticCache::new(&cfg.cache); - let live = Live::build(cfg, &secrets)?; + let router_status = RouterLastStatus::default(); + let live = Live::build(cfg, &secrets, router_status.clone())?; Ok(SharedState(Arc::new(AppState { live: ArcSwap::from_pointee(live), stats, + promotion, + imports, + cmf, cache, secrets, config_path, pipeline: crate::pipeline::Pipeline::new(), + router_status, }))) } @@ -84,7 +138,7 @@ impl AppState { /// If any step fails, the active state and file are left unchanged. pub fn reload(&self, new_cfg: Config) -> anyhow::Result<()> { new_cfg.validate()?; - let live = Live::build(new_cfg.clone(), &self.secrets)?; + let live = Live::build(new_cfg.clone(), &self.secrets, self.router_status.clone())?; new_cfg.save(&self.config_path)?; self.live.store(Arc::new(live)); Ok(()) @@ -94,7 +148,7 @@ impl AppState { /// (e.g. after a secret change — only provider keys changed). pub fn rebuild(&self) -> anyhow::Result<()> { let cfg = self.live.load_full().cfg.clone(); - let live = Live::build(cfg, &self.secrets)?; + let live = Live::build(cfg, &self.secrets, self.router_status.clone())?; self.live.store(Arc::new(live)); Ok(()) } diff --git a/web/js/i18n/en.js b/web/js/i18n/en.js index 05f4e57..4e43e4e 100644 --- a/web/js/i18n/en.js +++ b/web/js/i18n/en.js @@ -28,12 +28,14 @@ export default { // nav "nav.dashboard": "Dashboard", "nav.models": "Models", + "nav.import": "Import model", "nav.routing": "Routing", "nav.protocols": "Protocols", "nav.keys": "API keys", "nav.playground": "Playground", "nav.settings": "Settings", "nav.foot": "Gateway · v{v}", + "nav.billing": "Router key & billing", // topbar / status "status.router": "Router", @@ -41,6 +43,7 @@ export default { "status.online": "online", "status.offline": "offline", "status.checking": "checking…", + "status.noKey": "no key", "theme.toggle": "Theme", "logout": "Log out", @@ -89,6 +92,7 @@ export default { "dash.health.title": "Health", "dash.health.reachable": "reachable", "dash.health.unreachable": "unreachable", + "dash.health.getKey": "Get a router key", // models "models.title": "Models", @@ -225,7 +229,34 @@ export default { "settings.router.keyEnv": "Router key env name", "settings.router.timeout": "Timeout (ms)", "settings.router.verifyTls": "Verify TLS certificate", + "settings.router.enabled": "Use the router (off = local models only)", + "settings.cmf": "Local models (CMF)", + "settings.cmf.localOnly": "Use only local models", + "settings.cmf.manage": "Run a local model server", + "settings.cmf.autoInstall": "Auto-install cortiq (crates.io)", + "settings.cmf.autoUpdate": "Auto-update cortiq", + "settings.cmf.model": "Local .cmf model path", + "settings.cmf.port": "Local server port", + "settings.cmf.modelId": "Model id in the pool", + "settings.cmf.note": "Serves a local .cmf via cortiq (github.com/infosave2007/cmf). Works even when the router is unavailable.", "settings.router.taxonomy": "Taxonomy id", + "settings.router.getKey": "Get or top up a router key at", + "settings.router.secret": "Router key (value)", + "settings.router.saveKey": "Save key", + "settings.router.keySaved": "Router key saved", + "settings.router.keyEmpty": "Paste the key first", + "settings.router.secretHelp": "Paste the key from your api.allaigate.com email, then Save key. This is NOT a “Keys” value — those are for clients calling your gateway.", + "settings.router.keyStoredMask": "•••••••••••• — key saved (paste again to replace)", + "settings.router.test": "Test router", + "router.probe.ok": "Router OK ({ms} ms)", + "router.probe.noKey": "Router key is not set", + "router.probe.auth": "Key rejected — invalid or revoked", + "router.probe.payment": "Payment required — plan expired or quota exhausted", + "router.probe.quota": "Rate limit exceeded", + "router.probe.timeout": "Router timed out", + "router.probe.unreachable": "Router unreachable", + "router.probe.error": "Router error", + "router.usage": "Usage", "settings.route": "Routing input", "settings.route.strategy": "Text strategy", "settings.route.maxChars": "Max chars", @@ -266,4 +297,46 @@ export default { "settings.cache.ttl": "Entry TTL", "settings.cache.embed": "Embedding model", "settings.cache.note": "Cache settings take effect after a restart.", + + // import (HuggingFace → CMF) + "import.subtitle": "Search HuggingFace, convert to a local .cmf, and serve it.", + "import.searchPlaceholder": "Search HuggingFace models (e.g. qwen3, llama)…", + "import.searching": "Searching…", + "import.noResults": "No models found.", + "import.configure": "Configure conversion", + "import.outName": "Output name", + "import.outNameHint": "The .cmf file name (letters, digits, - and _).", + "import.quant": "Quantization", + "import.convert": "Convert & load", + "import.converting": "Converting", + "import.started": "Conversion started", + "import.failed": "Conversion failed — see the log.", + "import.register": "Register as local model", + "import.registered": "Registered as '{id}'. Run cortiq-server to serve it.", + "import.jobs": "Conversions", + "import.cancel": "Cancel", + "import.cancelling": "Cancelling — cleaning up partial output…", + "import.delete": "Delete", + "import.deleteConfirm": "Delete this model and its converted .cmf file from disk?", + "import.deleted": "Deleted with its files", + "import.state.running": "running", + "import.state.done": "done", + "import.state.error": "error", + "import.state.cancelled": "cancelled", + "import.q.q8_2f": "8-bit two-field (𝒲×θ) — recommended: best quality/size.", + "import.q.q8_row": "8-bit per-row — simple, robust.", + "import.q.q4": "4-bit block — smallest, lower quality.", + "import.q.vbit": "Variable 3–8 bit — per-expert bit budgets.", + "import.q.f16": "16-bit — no quantization.", + "import.q.f32": "32-bit — full precision (large file).", + "import.adv.title": "Advanced settings", + "import.adv.auto": "auto", + "import.adv.linearCore": "Linear core", + "import.adv.linearHint": "How linear-attention layers are carried/folded.", + "import.adv.nphase": "vmf_phase phases/head", + "import.adv.vbitShape": "VBIT budget shape", + "import.adv.meanBits": "VBIT mean bits", + "import.adv.shard": "Shard max (GB)", + "import.adv.noShard": "no sharding", + "import.adv.skipMtp": "skip MTP head", }; diff --git a/web/js/i18n/ru.js b/web/js/i18n/ru.js index 2af538b..c686c05 100644 --- a/web/js/i18n/ru.js +++ b/web/js/i18n/ru.js @@ -27,12 +27,14 @@ export default { // nav "nav.dashboard": "Дашборд", "nav.models": "Модели", + "nav.import": "Импорт модели", "nav.routing": "Роутинг", "nav.protocols": "Протоколы", "nav.keys": "API-ключи", "nav.playground": "Playground", "nav.settings": "Настройки", "nav.foot": "Шлюз · v{v}", + "nav.billing": "Ключ роутера и оплата", // topbar / status "status.router": "Роутер", @@ -40,6 +42,7 @@ export default { "status.online": "в сети", "status.offline": "недоступен", "status.checking": "проверка…", + "status.noKey": "нет ключа", "theme.toggle": "Тема", "logout": "Выйти", @@ -88,6 +91,7 @@ export default { "dash.health.title": "Состояние", "dash.health.reachable": "доступен", "dash.health.unreachable": "недоступен", + "dash.health.getKey": "Получить ключ роутера", // models "models.title": "Модели", @@ -224,7 +228,34 @@ export default { "settings.router.keyEnv": "Имя env-переменной с ключом роутера", "settings.router.timeout": "Таймаут (мс)", "settings.router.verifyTls": "Проверять TLS-сертификат", + "settings.router.enabled": "Использовать роутер (выкл = только локальные модели)", + "settings.cmf": "Локальные модели (CMF)", + "settings.cmf.localOnly": "Использовать только локальные модели", + "settings.cmf.manage": "Запускать локальный сервер моделей", + "settings.cmf.autoInstall": "Автоустановка cortiq (crates.io)", + "settings.cmf.autoUpdate": "Автообновление cortiq", + "settings.cmf.model": "Путь к локальной .cmf-модели", + "settings.cmf.port": "Порт локального сервера", + "settings.cmf.modelId": "ID модели в пуле", + "settings.cmf.note": "Обслуживает локальный .cmf через cortiq (github.com/infosave2007/cmf). Работает даже когда роутер недоступен.", "settings.router.taxonomy": "ID таксономии", + "settings.router.getKey": "Получить или пополнить ключ роутера:", + "settings.router.secret": "Ключ роутера (значение)", + "settings.router.saveKey": "Сохранить ключ", + "settings.router.keySaved": "Ключ роутера сохранён", + "settings.router.keyEmpty": "Сначала вставьте ключ", + "settings.router.secretHelp": "Вставьте ключ из письма с api.allaigate.com и нажмите «Сохранить ключ». Это НЕ ключ из раздела «Ключи» — те выдаются клиентам для доступа к вашему шлюзу.", + "settings.router.keyStoredMask": "•••••••••••• — ключ сохранён (вставьте снова, чтобы заменить)", + "settings.router.test": "Проверить роутер", + "router.probe.ok": "Роутер работает ({ms} мс)", + "router.probe.noKey": "Ключ роутера не задан", + "router.probe.auth": "Ключ отклонён — недействителен или отозван", + "router.probe.payment": "Требуется оплата — тариф истёк или квота исчерпана", + "router.probe.quota": "Превышен лимит запросов", + "router.probe.timeout": "Роутер не ответил вовремя", + "router.probe.unreachable": "Роутер недоступен", + "router.probe.error": "Ошибка роутера", + "router.usage": "Использование", "settings.route": "Входные данные роутинга", "settings.route.strategy": "Стратегия текста", "settings.route.maxChars": "Макс. символов", @@ -265,4 +296,46 @@ export default { "settings.cache.ttl": "TTL записи", "settings.cache.embed": "Модель эмбеддингов", "settings.cache.note": "Настройки кэша применяются после перезапуска.", + + // импорт (HuggingFace → CMF) + "import.subtitle": "Найдите модель на HuggingFace, сконвертируйте в локальный .cmf и подключите.", + "import.searchPlaceholder": "Поиск моделей на HuggingFace (напр. qwen3, llama)…", + "import.searching": "Ищу…", + "import.noResults": "Модели не найдены.", + "import.configure": "Настройка конвертации", + "import.outName": "Имя файла", + "import.outNameHint": "Имя .cmf (буквы, цифры, - и _).", + "import.quant": "Квантизация", + "import.convert": "Сконвертировать и загрузить", + "import.converting": "Конвертация", + "import.started": "Конвертация запущена", + "import.failed": "Конвертация не удалась — смотрите лог.", + "import.register": "Подключить как локальную модель", + "import.registered": "Подключено как '{id}'. Запустите cortiq-server для обслуживания.", + "import.jobs": "Конвертации", + "import.cancel": "Отменить", + "import.cancelling": "Отмена — удаляю недописанный файл…", + "import.delete": "Удалить", + "import.deleteConfirm": "Удалить эту модель и сконвертированный .cmf-файл с диска?", + "import.deleted": "Удалено вместе с файлами", + "import.state.running": "идёт", + "import.state.done": "готово", + "import.state.error": "ошибка", + "import.state.cancelled": "отменено", + "import.q.q8_2f": "8-бит двухполевой (𝒲×θ) — рекомендуется: лучшее качество/размер.", + "import.q.q8_row": "8-бит построчно — просто, надёжно.", + "import.q.q4": "4-бит блочно — компактнее, ниже качество.", + "import.q.vbit": "Переменная 3–8 бит — per-expert бюджеты.", + "import.q.f16": "16-бит — без квантования.", + "import.q.f32": "32-бит — полная точность (большой файл).", + "import.adv.title": "Расширенные настройки", + "import.adv.auto": "авто", + "import.adv.linearCore": "Линейное ядро", + "import.adv.linearHint": "Как переносятся/сворачиваются линейные слои внимания.", + "import.adv.nphase": "vmf_phase фаз/голову", + "import.adv.vbitShape": "Форма бюджета VBIT", + "import.adv.meanBits": "VBIT средних бит", + "import.adv.shard": "Шард макс (ГБ)", + "import.adv.noShard": "без шардинга", + "import.adv.skipMtp": "пропустить MTP-голову", }; diff --git a/web/js/views/settings.js b/web/js/views/settings.js index bb7afcf..2b073d2 100644 --- a/web/js/views/settings.js +++ b/web/js/views/settings.js @@ -2,7 +2,7 @@ import { h, mount, toast } from "../ui.js"; import { t } from "../i18n.js"; import { api } from "../api.js"; -import { appState } from "../app.js"; +import { appState, SITE_URL } from "../app.js"; function field(label, control, hint) { return h("label", { class: "field" }, h("span", {}, label), control, hint ? h("div", { class: "hint" }, hint) : null); @@ -12,9 +12,45 @@ function check(label, checked) { return { node: h("label", { class: "check" }, cb, label), cb }; } +// Flatten router usage (nested account/usage objects) into readable, wrapping +// key: value rows — a raw JSON.stringify dump overflows the card horizontally. +function usageRows(obj, prefix) { + const rows = []; + for (const [k, v] of Object.entries(obj || {})) { + const key = prefix ? prefix + "." + k : k; + if (v && typeof v === "object" && !Array.isArray(v)) { + rows.push(...usageRows(v, key)); + } else { + rows.push( + h("div", { class: "mono kv" }, + h("span", { class: "kv-k" }, key + ": "), + h("span", { class: "kv-v" }, Array.isArray(v) ? JSON.stringify(v) : String(v))) + ); + } + } + return rows; +} + +// Probe status → i18n key suffix (see /admin/api/router/probe). +function probeMsgKey(status) { + const map = { + ok: "ok", + no_key: "noKey", + auth: "auth", + payment: "payment", + quota: "quota", + timeout: "timeout", + unreachable: "unreachable", + }; + return "router.probe." + (map[status] || "error"); +} + +// Statuses where the fix is on the site (get / pay for a key). +const PROBE_SITE_STATUSES = ["no_key", "auth", "payment", "quota"]; + export async function renderSettings() { const meta = appState.meta || (await api.meta()); - const s = await api.getSettings(); + const [s, secretsData] = await Promise.all([api.getSettings(), api.listSecrets()]); const root = h("div"); const listenIn = h("input", { value: s.listen || "" }); @@ -24,6 +60,101 @@ export async function renderSettings() { const rTimeout = h("input", { type: "number", value: s.router?.timeout_ms ?? 800 }); const rTax = h("input", { value: s.router?.taxonomy_id || "" }); const rVerify = check(t("settings.router.verifyTls"), s.router?.verify_tls); + const rEnabled = check(t("settings.router.enabled"), s.router?.enabled !== false); + // router key value (write-only; stored in the secret store, like model keys) + const rKeySource = + (secretsData.secrets || []).find((x) => x.name === (s.router?.api_key_env || "CORTIQ_ROUTER_KEY"))?.source || + "missing"; + const keyStored = rKeySource === "store" || rKeySource === "env"; + // When a key is already saved, show a masked "saved" hint instead of an empty + // field (an empty field reads as "nothing is set"). + const rSecret = h("input", { + type: "password", + placeholder: keyStored ? t("settings.router.keyStoredMask") : t("models.form.secretPlaceholder"), + }); + const rKeyBadge = h( + "span", + { class: "badge " + (rKeySource === "store" ? "store" : rKeySource === "env" ? "env" : "bad") }, + rKeySource === "store" + ? t("models.form.secretStored") + : rKeySource === "env" + ? t("models.form.secretEnv") + : t("models.form.secretMissing") + ); + const rSecretHint = h( + "div", + {}, + h( + "div", + {}, + rKeyBadge, + " · ", + t("settings.router.getKey") + " ", + h("a", { href: SITE_URL, target: "_blank", rel: "noopener" }, "api.allaigate.com") + ), + h("div", { class: "muted", style: "margin-top:4px" }, t("settings.router.secretHelp")) + ); + // The key lives in the secret store under this env name (config default: CORTIQ_ROUTER_KEY). + const keyName = () => rKeyEnv.value.trim() || "CORTIQ_ROUTER_KEY"; + // Persist just the pasted router key (independent of the big Save at page top). + async function saveKey() { + const v = rSecret.value.trim(); + if (!v) return false; + if (!rKeyEnv.value.trim()) rKeyEnv.value = "CORTIQ_ROUTER_KEY"; + await api.setSecret(keyName(), v); + rSecret.value = ""; + rSecret.placeholder = t("settings.router.keyStoredMask"); // show "saved", not empty + rKeyBadge.textContent = t("models.form.secretStored"); + rKeyBadge.className = "badge store"; + return true; + } + const saveKeyBtn = h("button", { class: "btn sm" }, "✓ " + t("settings.router.saveKey")); + saveKeyBtn.addEventListener("click", async () => { + saveKeyBtn.disabled = true; + try { + if (await saveKey()) toast(t("settings.router.keySaved"), "good"); + else toast(t("settings.router.keyEmpty"), "warn"); + } catch (e) { + toast(String(e.message || e), "bad"); + } + saveKeyBtn.disabled = false; + }); + // "Test router" — real /v1/route call: tells key problems apart from a down router + const probeBtn = h("button", { class: "btn" }, t("settings.router.test")); + const probeOut = h("div", { class: "hint", style: "margin-top:8px" }); + probeBtn.addEventListener("click", async () => { + probeBtn.disabled = true; + mount(probeOut, h("span", { class: "spinner" })); + try { + // If a key was just pasted but not saved, persist it first so the test uses it. + if (rSecret.value.trim()) { + try { + await saveKey(); + } catch (_) { + /* fall through — probe will report no_key/auth */ + } + } + const r = await api.probeRouter(); + const parts = [t(probeMsgKey(r.status), { ms: r.latency_ms ?? "—" })]; + if (r.message) parts.push(" — " + r.message); + const line = h("span", {}, ...parts); + const out = [line]; + if (PROBE_SITE_STATUSES.includes(r.status)) { + out.push(" · ", h("a", { href: SITE_URL, target: "_blank", rel: "noopener" }, t("dash.health.getKey") + " ↗")); + } + if (r.usage && typeof r.usage === "object") { + out.push( + h("div", { class: "usage-box", style: "margin-top:4px" }, + h("b", {}, t("router.usage")), ...usageRows(r.usage)) + ); + } + mount(probeOut, h("div", {}, h("span", { class: "badge " + (r.ok ? "ok" : "bad") }, r.ok ? "OK" : "×"), " ", ...out)); + toast(t(probeMsgKey(r.status), { ms: r.latency_ms ?? "—" }), r.ok ? "good" : "bad"); + } catch (e) { + mount(probeOut, String(e.message || e)); + } + probeBtn.disabled = false; + }); // route const strategySel = h("select", {}, ...(meta.text_strategies || []).map((x) => h("option", { value: x, selected: x === s.route?.text_strategy }, x))); const maxChars = h("input", { type: "number", value: s.route?.max_chars ?? 4000 }); @@ -54,13 +185,24 @@ export async function renderSettings() { h("option", { value: "" }, "auto"), ...embedModels.map((id) => h("option", { value: id, selected: id === s.cache?.embed_model }, id)) ); + // local models (CMF) — run a local .cmf server and/or force local-only routing + const cmfLocalOnly = check(t("settings.cmf.localOnly"), s.cmf?.local_only); + const cmfManage = check(t("settings.cmf.manage"), s.cmf?.manage_server); + const cmfAutoInstall = check(t("settings.cmf.autoInstall"), s.cmf?.auto_install); + const cmfAutoUpdate = check(t("settings.cmf.autoUpdate"), s.cmf?.auto_update); + const cmfModel = h("input", { value: s.cmf?.local_model || "", placeholder: "models/model.cmf" }); + const cmfPort = h("input", { type: "number", value: s.cmf?.local_port ?? 8081 }); + const cmfModelId = h("input", { value: s.cmf?.model_id || "cmf-local" }); const saveBtn = h("button", { class: "btn primary" }, t("common.save")); saveBtn.addEventListener("click", async () => { saveBtn.disabled = true; + // a pasted router key needs an env name to live under — default it + if (rSecret.value.trim() && !rKeyEnv.value.trim()) rKeyEnv.value = "CORTIQ_ROUTER_KEY"; const patch = { listen: listenIn.value.trim(), router: { + enabled: rEnabled.cb.checked, url: rUrl.value.trim(), api_key_env: rKeyEnv.value.trim() || null, timeout_ms: parseInt(rTimeout.value) || 800, @@ -90,6 +232,16 @@ export async function renderSettings() { max_entries: s.cache?.max_entries ?? 1000, embed_model: cacheEmbedSel.value || null, }, + cmf: { + ...(s.cmf || {}), + local_only: cmfLocalOnly.cb.checked, + manage_server: cmfManage.cb.checked, + auto_install: cmfAutoInstall.cb.checked, + auto_update: cmfAutoUpdate.cb.checked, + local_model: cmfModel.value.trim(), + local_port: parseInt(cmfPort.value) || 8081, + model_id: cmfModelId.value.trim() || "cmf-local", + }, }; // serde: drop null optionals if (!patch.router.api_key_env) delete patch.router.api_key_env; @@ -98,6 +250,13 @@ export async function renderSettings() { if (!patch.cache.embed_model) delete patch.cache.embed_model; try { const r = await api.putSettings(patch); + const secretVal = rSecret.value.trim(); + if (secretVal) { + await api.setSecret(rKeyEnv.value.trim(), secretVal); + rSecret.value = ""; + rKeyBadge.textContent = t("models.form.secretStored"); + rKeyBadge.className = "badge store"; + } toast(r.needs_restart ? t("settings.needsRestart") : t("toast.saved"), r.needs_restart ? "warn" : "good"); } catch (e) { toast(String(e.message || e), "bad"); @@ -150,8 +309,13 @@ export async function renderSettings() { h("div", { class: "card-head" }, h("h3", {}, t("settings.router"))), field(t("settings.router.url"), rUrl), field(t("settings.router.keyEnv"), rKeyEnv), + field(t("settings.router.secret"), h("div", { class: "key-row" }, rSecret, saveKeyBtn), rSecretHint), h("div", { class: "row" }, field(t("settings.router.timeout"), rTimeout), field(t("settings.router.taxonomy"), rTax)), - rVerify.node + rEnabled.node, + rVerify.node, + h("div", { class: "divider" }), + h("div", { class: "flex" }, probeBtn), + probeOut ), h( "div", @@ -199,6 +363,17 @@ export async function renderSettings() { h("div", { class: "row" }, field(t("settings.cache.threshold"), cacheThresh), field(t("settings.cache.ttl"), cacheTtlIn)), field(t("settings.cache.embed"), cacheEmbedSel), h("div", { class: "hint" }, t("settings.cache.note")) + ), + h( + "div", + { class: "card" }, + h("div", { class: "card-head" }, h("h3", {}, t("settings.cmf"))), + cmfLocalOnly.node, + cmfManage.node, + h("div", { class: "row" }, cmfAutoInstall.node, cmfAutoUpdate.node), + field(t("settings.cmf.model"), cmfModel), + h("div", { class: "row" }, field(t("settings.cmf.port"), cmfPort), field(t("settings.cmf.modelId"), cmfModelId)), + h("div", { class: "hint" }, t("settings.cmf.note")) ) ), h(