//! Admin API for the web management panel (`/admin/api/*`) + SPA serving (`/admin/*`). //! //! All endpoints under `/admin/api` are protected by a Bearer admin token. Config changes //! go through `AppState::reload` (validation → build `Live` → atomic TOML write //! → swap), so they are applied without a restart. Secrets are never returned — //! only their presence status. pub mod assets; use crate::config::{ 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; use crate::stats::parse_duration_secs; use axum::{ extract::{Path, Query, Request, State}, http::{header, StatusCode}, middleware::{self, Next}, response::{IntoResponse, Response}, routing::{delete, get, post}, Json, Router, }; use serde::Deserialize; use serde_json::{json, Value}; use std::collections::HashMap; use std::net::ToSocketAddrs; use std::time::Instant; // ───────────────────────────── common types / helpers ─────────────────────── type ApiResult = Result, ApiError>; pub struct ApiError { status: StatusCode, message: String, } impl ApiError { fn bad(msg: impl Into) -> Self { Self { status: StatusCode::BAD_REQUEST, message: msg.into(), } } fn not_found(msg: impl Into) -> Self { Self { status: StatusCode::NOT_FOUND, message: msg.into(), } } } impl From for ApiError { fn from(e: anyhow::Error) -> Self { Self::bad(e.to_string()) } } impl IntoResponse for ApiError { fn into_response(self) -> Response { ( self.status, Json(json!({"error": {"message": self.message, "type": "admin_error"}})), ) .into_response() } } fn ok(v: Value) -> ApiResult { Ok(Json(v)) } fn current_cfg(state: &SharedState) -> Config { state.live().cfg.clone() } /// Random hex token from `n_bytes` bytes of cryptographic randomness. /// Uses the OS CSPRNG via `getrandom` (works on Linux/macOS/Windows); falls back /// to a time-based generator only in the extremely unlikely event that fails. pub fn random_token(n_bytes: usize) -> String { let mut buf = vec![0u8; n_bytes]; if getrandom::getrandom(&mut buf).is_err() { let t = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_nanos(); for (i, b) in buf.iter_mut().enumerate() { *b = ((t >> ((i % 16) * 8)) as u8) ^ (i as u8).wrapping_mul(31); } } buf.iter().map(|b| format!("{b:02x}")).collect() } fn constant_time_eq(a: &[u8], b: &[u8]) -> bool { if a.len() != b.len() { return false; } let mut diff = 0u8; for (x, y) in a.iter().zip(b.iter()) { diff |= x ^ y; } diff == 0 } fn mask_key(key: &str) -> String { let n = key.chars().count(); if n <= 8 { return "•".repeat(n.max(3)); } let head: String = key.chars().take(6).collect(); let tail: String = key.chars().skip(n - 4).collect(); format!("{head}…{tail}") } fn tiers_to_map(routing: &RoutingCfg) -> HashMap> { routing .tiers .iter() .map(|(k, v)| { let list = match v { TierTargets::List(l) => l.clone(), TierTargets::One(s) => vec![s.clone()], }; (k.clone(), list) }) .collect() } fn parse_routing(model: &str) -> RoutingDirective { if let Some(rest) = model.strip_prefix("cortiq-auto") { let profile = rest.strip_prefix(':').map(|p| p.to_string()); RoutingDirective::Auto { profile } } else { RoutingDirective::Pinned { model_id: model.to_string(), } } } // ───────────────────────────── router ─────────────────────────────────────── /// Admin API routes. If `admin_token` is empty, the panel is open (no auth); /// otherwise every `/admin/api/*` request needs `Authorization: Bearer `. /// The token is captured in a middleware closure so it does not depend on State. pub fn api_routes(admin_token: String) -> Router { let 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)) .route( "/admin/api/models/:id", axum::routing::put(update_model).delete(delete_model), ) .route("/admin/api/models/:id/probe", post(probe_model)) .route("/admin/api/routing", get(get_routing).put(put_routing)) .route("/admin/api/routing/labels", get(get_routing_labels)) .route( "/admin/api/protocols", get(get_protocols).put(put_protocols), ) .route("/admin/api/harnesses", get(get_harnesses)) .route("/admin/api/settings", get(get_settings).put(put_settings)) .route("/admin/api/keys", get(list_keys).post(create_key)) .route("/admin/api/keys/:key", axum::routing::delete(delete_key)) .route( "/admin/api/secrets", get(list_secrets).put(set_secret).delete(clear_secret), ) .route("/admin/api/stats", get(get_stats).delete(clear_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/cmf", get(cmf_status)) .route("/admin/api/cmf/install", post(cmf_install)) .route("/admin/api/cmf/port", get(cmf_port_check)) .route("/admin/api/cmf/files", get(cmf_files)) .route("/admin/api/provider/models", post(provider_models)) .route("/admin/api/requests", get(get_requests)) .route("/admin/api/test", post(run_test)) .route("/admin/api/test/stream", post(run_test_stream)) .route("/admin/api/logs", get(get_logs).delete(clear_logs)) .route("/admin/api/system", get(get_system)) .route("/admin/api/restart", post(restart_gateway)) .route("/admin/api/bench", get(bench_list).post(bench_start)) .route("/admin/api/config/export", get(config_export)) .route("/admin/api/config/import", post(config_import)) .route("/admin/api/media/models", get(media_models)) .route("/admin/api/media", get(media_jobs).post(media_generate)) .route( "/admin/api/media/:job", get(media_job_status).delete(media_delete), ) .route("/admin/api/media/:job/cancel", post(media_cancel)) .route("/admin/api/media/:job/file/:kind", get(media_file)) .route("/admin/api/setup/status", get(setup_status)) .route("/admin/api/setup/complete", post(setup_complete)) .route("/admin/api/setup/featured", get(setup_featured)) .route("/admin/api/setup/file", delete(delete_setup_file)); if admin_token.is_empty() { router } else { router.route_layer(middleware::from_fn(move |req: Request, next: Next| { let expected = admin_token.clone(); async move { auth(expected, req, next).await } })) } } /// Extract Bearer admin token from the `Authorization` header. fn bearer_token(req: &Request) -> String { req.headers() .get(header::AUTHORIZATION) .and_then(|v| v.to_str().ok()) .map(|v| { v.strip_prefix("Bearer ") .or_else(|| v.strip_prefix("bearer ")) .unwrap_or(v) .trim() }) .unwrap_or("") .to_string() } /// Admin authentication middleware (constant-time comparison). async fn auth(expected: String, req: Request, next: Next) -> Response { let token = bearer_token(&req); if !expected.is_empty() && constant_time_eq(token.as_bytes(), expected.as_bytes()) { next.run(req).await } else { ( StatusCode::UNAUTHORIZED, Json(json!({"error": {"message": "invalid or missing admin token", "type": "authentication_error", "code": "unauthorized"}})), ) .into_response() } } // ───────────────────────────── metrics ────────────────────────────────────── /// `GET /metrics` (Prometheus). Exposed when `telemetry.metrics = true`. pub async fn metrics(State(state): State) -> Response { let live = state.live(); if !live.cfg.telemetry.metrics { return (StatusCode::NOT_FOUND, "metrics disabled").into_response(); } let body = state.stats.prometheus(); ([(header::CONTENT_TYPE, "text/plain; version=0.0.4")], body).into_response() } // ───────────────────────────── health / meta ─────────────────────────────── async fn probe_router(url: &str) -> (bool, u64) { let client = match reqwest::Client::builder() .timeout(std::time::Duration::from_millis(1500)) .danger_accept_invalid_certs(true) .build() { Ok(c) => c, Err(_) => return (false, 0), }; let started = Instant::now(); // 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().is_success()) .unwrap_or(false); (ok, started.elapsed().as_millis() as 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 .iter() .map(|m| { let key = match &m.api_key_env { Some(env) => state.secrets.source(env), None => "none", }; json!({ "id": m.id, "provider": m.provider, "model": m.model, "kind": m.kind, "cost_tier": m.cost_tier, "key_source": key, "in_registry": live.registry.get(&m.id).is_some(), }) }) .collect(); ok(json!({ "ok": true, "version": env!("CARGO_PKG_VERSION"), "listen": live.cfg.listen, "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 })) } #[derive(Deserialize)] struct ProviderModelsBody { provider: String, base_url: String, /// Key typed into the form (takes priority)… #[serde(default)] api_key: Option, /// …or the name of a stored secret to reuse if no key was typed. #[serde(default)] api_key_env: Option, } /// Concise error line from a provider's JSON error body. fn provider_err(body: &str, status: u16) -> String { if let Ok(v) = serde_json::from_str::(body) { if let Some(m) = v["error"]["message"] .as_str() .or_else(|| v["error"].as_str()) .or_else(|| v["message"].as_str()) { return format!("{status}: {m}"); } } format!("HTTP {status}") } /// Validate a provider API key and list its models by calling the provider's /// `/models` endpoint — powers the "Test key & load models" button. The key is /// used transiently (from the form) or resolved from a stored secret; it is not /// persisted here. async fn provider_models( State(state): State, Json(b): Json, ) -> ApiResult { let base = b.base_url.trim_end_matches('/').to_string(); if base.is_empty() { return Err(ApiError::bad("base URL is required")); } let key = b.api_key.filter(|k| !k.trim().is_empty()).or_else(|| { b.api_key_env .as_deref() .and_then(|e| state.secrets.resolve(e)) }); let key = key.as_deref(); let client = reqwest::Client::builder() .timeout(std::time::Duration::from_secs(15)) .build() .map_err(|e| ApiError::bad(e.to_string()))?; // Anthropic uses x-api-key + anthropic-version and /v1/models; every other // (OpenAI-compatible) provider uses a Bearer token and /models. let resp = if b.provider == "anthropic" { let mut r = client .get(format!("{base}/v1/models")) .header("anthropic-version", "2023-06-01"); if let Some(k) = key { r = r.header("x-api-key", k); } r.send().await } else { let mut r = client.get(format!("{base}/models")); if let Some(k) = key { r = r.header("authorization", format!("Bearer {k}")); } r.send().await }; let resp = resp.map_err(|e| ApiError::bad(format!("request failed: {e}")))?; let status = resp.status().as_u16(); let body = resp.text().await.unwrap_or_default(); if !(200..300).contains(&status) { return ok(json!({ "ok": false, "status": status, "error": provider_err(&body, status) })); } // OpenAI + Anthropic both return { "data": [ { "id": ... } ] }. let v: Value = serde_json::from_str(&body).unwrap_or(json!({})); let mut models: Vec = v["data"] .as_array() .map(|arr| { arr.iter() .filter_map(|m| { let id = m["id"].as_str()?.to_string(); let mut obj = json!({ "id": id }); let mut caps = vec![]; if let Some(pricing) = m.get("pricing").filter(|v| !v.is_null()) { if let Some(prompt) = pricing .get("prompt") .and_then(|s| s.as_str()) .and_then(|s| s.parse::().ok()) { obj["price_in"] = json!(prompt * 1_000_000.0); } if let Some(comp) = pricing .get("completion") .and_then(|s| s.as_str()) .and_then(|s| s.parse::().ok()) { obj["price_out"] = json!(comp * 1_000_000.0); } if pricing .get("input_cache_read") .filter(|v| !v.is_null()) .is_some() || pricing .get("input_cache_write") .filter(|v| !v.is_null()) .is_some() { caps.push("caching"); } } if let Some(params) = m.get("supported_parameters").and_then(|p| p.as_array()) { let has_tools = params.iter().any(|p| { p.as_str() == Some("tools") || p.as_str() == Some("tool_choice") }); if has_tools { caps.push("tools"); } let has_json = params.iter().any(|p| p.as_str() == Some("response_format")); if has_json { caps.push("json"); } let has_reasoning = params.iter().any(|p| { p.as_str() == Some("reasoning") || p.as_str() == Some("include_reasoning") }); if has_reasoning || m.get("reasoning").filter(|v| !v.is_null()).is_some() { caps.push("reasoning"); } } if let Some(arch) = m.get("architecture").filter(|v| !v.is_null()) { if let Some(inputs) = arch.get("input_modalities").and_then(|i| i.as_array()) { if inputs.iter().any(|i| i.as_str() == Some("image")) { caps.push("vision"); } if inputs.iter().any(|i| i.as_str() == Some("audio")) { caps.push("audio"); } } } if !caps.is_empty() { obj["caps"] = json!(caps); } Some(obj) }) .collect() }) .unwrap_or_default(); models.sort_by(|a, b| a["id"].as_str().cmp(&b["id"].as_str())); let mut resp = json!({ "ok": true, "count": models.len(), "models": models }); if b.provider == "openrouter" { if let Some(k) = key { if let Ok(cred_resp) = client .get("https://openrouter.ai/api/v1/credits") .header("authorization", format!("Bearer {k}")) .send() .await { if let Ok(cred_body) = cred_resp.text().await { let cred_v: Value = serde_json::from_str(&cred_body).unwrap_or(json!({})); if let Some(data) = cred_v.get("data") { let total = data .get("total_credits") .and_then(|v| v.as_f64()) .unwrap_or(0.0); let usage = data .get("total_usage") .and_then(|v| v.as_f64()) .unwrap_or(0.0); resp["balance"] = json!(total - usage); } } } } } ok(resp) } async fn meta(State(state): State) -> ApiResult { // process start marker: open tabs poll this and offer a reload when the // gateway (and thus the embedded SPA) has been restarted/updated static STARTED: std::sync::OnceLock = std::sync::OnceLock::new(); let started = *STARTED.get_or_init(|| { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_secs() }); ok(json!({ "version": env!("CARGO_PKG_VERSION"), "started": started, "update_available": state.gateway_update.lock().unwrap().clone(), "admin_open_public": state .admin_open_public .load(std::sync::atomic::Ordering::Relaxed), "providers": ["openai", "anthropic", "openrouter", "lmstudio", "ollama", "http"], "provider_impl": { "openai": true, "anthropic": true, "openrouter": true, "lmstudio": true, "ollama": true, "http": true }, "cost_tiers": ["cheap", "mid", "expensive", "local"], "kinds": ["chat", "embedding"], "profiles": ["cost-saver", "balanced", "quality-first"], "text_strategies": ["last_user", "last_user_plus_system", "concat_all"], "policy_modes": ["fixed_table", "cost_aware"], "tiers": ["low", "medium", "high"], "caps": ["tools", "vision", "audio", "reasoning", "json", "caching"], "protocols_impl": { "openai_chat": true, "openai_completions": true, "openai_embeddings": true, "openai_models": true, "anthropic_messages": true, "mcp": true, "native_passthrough": true }, "languages": ["en", "ru", "de", "fr", "es", "zh", "tr"], })) } // ───────────────────────────── config ────────────────────────────────────── async fn get_config(State(state): State) -> ApiResult { let cfg = current_cfg(&state); ok(serde_json::to_value(&cfg).map_err(|e| ApiError::bad(e.to_string()))?) } async fn put_config(State(state): State, Json(cfg): Json) -> ApiResult { state.reload(cfg)?; ok(json!({ "ok": true })) } // ───────────────────────────── models ────────────────────────────────────── async fn list_models(State(state): State) -> ApiResult { let live = state.live(); let mut models: Vec = live .cfg .models .iter() .map(|m| { let mut v = serde_json::to_value(m).unwrap_or(json!({})); let key_source = match &m.api_key_env { Some(env) => state.secrets.source(env), None => "none", }; v["key_source"] = json!(key_source); v }) .collect(); // 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; for s in cmf.effective_servers() { if models.iter().any(|m| m["id"] == json!(s.id)) { continue; } models.push(json!({ "id": s.id, "provider": "openai", "base_url": format!("http://{}:{}/v1", cmf.local_host, s.port), "model": "cortiq", "cost_tier": "local", "price_in": 0.0, "price_out": 0.0, "kind": "chat", "caps": [], "key_source": "none", "managed": true, })); } ok(json!({ "models": models })) } async fn create_model(State(state): State, Json(m): Json) -> ApiResult { let mut cfg = current_cfg(&state); if cfg.models.iter().any(|x| x.id == m.id) { return Err(ApiError::bad(format!("model id '{}' already exists", m.id))); } cfg.models.push(m); state.reload(cfg)?; ok(json!({ "ok": true })) } async fn update_model( State(state): State, Path(id): Path, Json(m): Json, ) -> ApiResult { let mut cfg = current_cfg(&state); let idx = cfg .models .iter() .position(|x| x.id == id) .ok_or_else(|| ApiError::not_found(format!("model '{id}' not found")))?; // if the id was renamed — must not conflict with an existing one if m.id != id && cfg.models.iter().any(|x| x.id == m.id) { return Err(ApiError::bad(format!("model id '{}' already exists", m.id))); } cfg.models[idx] = m; state.reload(cfg)?; ok(json!({ "ok": true })) } /// Remove a `.cmf` path — it may be a plain file or a tensor-dir. Returns whether /// something was actually removed. fn remove_cmf_path(path: &std::path::Path) -> bool { if path.is_dir() { std::fs::remove_dir_all(path).is_ok() } else if path.is_file() { std::fs::remove_file(path).is_ok() } else { false } } /// The local `.cmf` backing a pool model, if it is a local CMF model whose file /// lives under `models_dir`. Cloud models (real remote base_url) return None. /// Registered imports store `model` = the file stem under models_dir. fn cmf_file_for_model( cmf: &crate::config::CmfCfg, m: &crate::config::ModelCfg, ) -> Option { let local = m.base_url == cmf.cortiq_server_url || m.base_url .contains(&format!("{}:{}", cmf.local_host, cmf.local_port)); let stem = m.model.trim(); if !local || stem.is_empty() || stem.contains('/') || stem.contains("..") { return None; // not local, or an unsafe/underivable path } Some(std::path::Path::new(&cmf.models_dir).join(format!("{stem}.cmf"))) } /// Drop a model id from routing (default + every tier) so `validate()` still /// passes after the model is gone. fn strip_id_from_routing(routing: &mut RoutingCfg, id: &str) { if routing.default == id { routing.default = String::new(); } for tt in routing.tiers.values_mut().chain(routing.tasks.values_mut()) { match tt { crate::config::TierTargets::List(v) => v.retain(|x| x != id), crate::config::TierTargets::One(s) if s == id => { *tt = crate::config::TierTargets::List(Vec::new()); } crate::config::TierTargets::One(_) => {} } } // an emptied task rule means "no rule", not "no candidates" routing.tasks.retain(|_, tt| match tt { crate::config::TierTargets::List(v) => !v.is_empty(), crate::config::TierTargets::One(s) => !s.is_empty(), }); } /// 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, Path(id): Path) -> ApiResult { let mut cfg = current_cfg(&state); // 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(); let before = cfg.models.len(); cfg.models.retain(|x| x.id != id); if cfg.models.len() == before { return Err(ApiError::not_found(format!("model '{id}' not found"))); } strip_id_from_routing(&mut cfg.routing, &id); 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)?; let file_removed = file.map(|p| remove_cmf_path(&p)).unwrap_or(false); ok(json!({ "ok": true, "file_removed": file_removed })) } async fn probe_model(State(state): State, Path(id): Path) -> ApiResult { let live = state.live(); // Provider string: from the static pool, or "openai" for a managed local // model (which is synthesized into the registry, not [[models]]). let provider_str = live .cfg .models .iter() .find(|m| m.id == id) .map(|m| m.provider.clone()) .or_else(|| { live.cfg .cmf .effective_servers() .iter() .find(|s| s.id == id) .map(|_| "openai".to_string()) }) .ok_or_else(|| ApiError::not_found(format!("model '{id}' not found")))?; if provider_str == "anthropic" { return ok(json!({ "ok": false, "error": "anthropic provider not yet implemented (planned v0.2)", })); } let provider = live .registry .get(&id) .ok_or_else(|| ApiError::bad(format!("model '{id}' has no runtime provider")))?; let req = ChatRequest { routing: RoutingDirective::Pinned { model_id: id.clone(), }, messages: vec![Message { role: "user".into(), content: "ping".into(), tool_calls: vec![], }], tools: vec![], params: GenParams { temperature: Some(0.0), max_tokens: Some(1), ..Default::default() }, stream: false, meta: RequestMeta { protocol: "probe".into(), ..Default::default() }, }; let started = Instant::now(); let res = provider.chat(req).await; let latency_ms = started.elapsed().as_millis() as u64; match res { Ok(_) => ok(json!({ "ok": true, "latency_ms": latency_ms })), Err(e) => { let msg = e.to_string(); // Make auth failures actionable — a raw "User not found" (OpenRouter's // 401) usually means the API key is wrong or the wrong type. let hint = if msg.contains("401") || msg.contains("Unauthorized") || msg.contains("User not found") || msg.contains("No auth") { " — check the API key (for OpenRouter use an inference key, not a provisioning/management key)" } else { "" }; ok(json!({ "ok": false, "latency_ms": latency_ms, "error": format!("{msg}{hint}") })) } } } // ───────────────────────────── routing ───────────────────────────────────── async fn get_routing(State(state): State) -> ApiResult { let live = state.live(); let tasks: HashMap> = live .cfg .routing .tasks .iter() .map(|(k, v)| { ( k.clone(), match v { TierTargets::List(l) => l.clone(), TierTargets::One(s) => vec![s.clone()], }, ) }) .collect(); ok(json!({ "tiers": tiers_to_map(&live.cfg.routing), "tasks": tasks, "default": live.cfg.routing.default, "policy": live.cfg.routing.policy, })) } /// Task labels of the active taxonomy, straight from the router (`GET /// /v1/taxonomies`) — the UI builds the per-task rule editor from this, so a /// custom taxonomy shows its own labels without a gateway update. async fn get_routing_labels(State(state): State) -> ApiResult { let live = state.live(); match live.router.taxonomies().await { Ok(Some(v)) => { let want = live.cfg.router.taxonomy_id.clone().unwrap_or_default(); let mut labels: Vec = Vec::new(); if let Some(arr) = v["taxonomies"].as_array() { for tx in arr { let id = tx["taxonomy_id"].as_str().unwrap_or(""); if want.is_empty() || id == want || labels.is_empty() { if let Some(ls) = tx["labels"].as_array() { let this: Vec = ls .iter() .filter_map(|l| l.as_str().map(String::from)) .collect(); if id == want || labels.is_empty() { labels = this; } } } } } if labels.is_empty() { ok(json!({ "labels": Value::Null, "source": "router-empty" })) } else { ok(json!({ "labels": labels, "source": "router" })) } } _ => ok(json!({ "labels": Value::Null, "source": "unavailable" })), } } #[derive(Deserialize)] struct RoutingBody { #[serde(default)] tiers: HashMap>, #[serde(default)] tasks: HashMap>, default: String, #[serde(default)] policy: RoutingPolicy, } async fn put_routing(State(state): State, Json(body): Json) -> ApiResult { let mut cfg = current_cfg(&state); cfg.routing = RoutingCfg { tiers: body .tiers .into_iter() .map(|(k, v)| (k, TierTargets::List(v))) .collect(), tasks: body .tasks .into_iter() .filter(|(_, v)| !v.is_empty()) // empty rule = "по сложности", not a rule .map(|(k, v)| (k, TierTargets::List(v))) .collect(), default: body.default, policy: body.policy, }; state.reload(cfg)?; ok(json!({ "ok": true })) } // ──────────────────────────── harness detection ──────────────────────────── // Which coding harnesses are present on the machine the gateway runs on, so the // API-keys view can mark tabs "installed" / "already configured for cortiq". // Detection is best-effort: binaries in PATH + common install dirs, config // files under the user's home. Works on Linux/macOS/Windows. fn harness_home() -> Option { std::env::var_os("HOME") .or_else(|| std::env::var_os("USERPROFILE")) .map(std::path::PathBuf::from) } fn find_bin(name: &str) -> bool { let exts: &[&str] = if cfg!(windows) { &[".exe", ".cmd", ".bat", ""] } else { &[""] }; let mut dirs: Vec = std::env::var_os("PATH") .map(|p| std::env::split_paths(&p).collect()) .unwrap_or_default(); if let Some(home) = harness_home() { // GUI-launched processes often miss the shell PATH — add usual suspects for extra in [ ".local/bin", ".cargo/bin", "bin", ".bun/bin", ".npm-global/bin", "AppData/Roaming/npm", ] { dirs.push(home.join(extra)); } } dirs.extend(["/usr/local/bin", "/opt/homebrew/bin"].map(std::path::PathBuf::from)); dirs.iter() .any(|d| exts.iter().any(|e| d.join(format!("{name}{e}")).is_file())) } /// Does the harness config mention cortiq (i.e. was it pointed at this gateway)? fn config_mentions_cortiq(rel: &str) -> bool { harness_home() .map(|h| h.join(rel)) .and_then(|p| std::fs::read_to_string(p).ok()) .map(|c| c.to_lowercase().contains("cortiq")) .unwrap_or(false) } async fn get_harnesses() -> ApiResult { let harnesses = tokio::task::spawn_blocking(|| { let vscode_ext_glob = |marker: &str| -> bool { let Some(home) = harness_home() else { return false; }; for extdir in [".vscode/extensions", ".vscode-server/extensions"] { if let Ok(rd) = std::fs::read_dir(home.join(extdir)) { for e in rd.flatten() { if e.file_name().to_string_lossy().starts_with(marker) { return true; } } } } false }; json!({ "Claude Code": { "installed": find_bin("claude"), "configured": serde_json::Value::Null }, "Codex": { "installed": find_bin("codex"), "configured": config_mentions_cortiq(".codex/config.toml") }, "OpenCode": { "installed": find_bin("opencode"), "configured": config_mentions_cortiq(".config/opencode/opencode.json") }, "Continue": { "installed": harness_home().map(|h| h.join(".continue").is_dir()).unwrap_or(false), "configured": config_mentions_cortiq(".continue/config.yaml") }, "Cline": { "installed": vscode_ext_glob("saoudrizwan.claude-dev"), "configured": serde_json::Value::Null }, }) }) .await .unwrap_or_else(|_| json!({})); ok(harnesses) } // ───────────────────────────── protocols ─────────────────────────────────── async fn get_protocols(State(state): State) -> ApiResult { let live = state.live(); ok(serde_json::to_value(&live.cfg.protocols).unwrap_or(json!({}))) } async fn put_protocols(State(state): State, Json(p): Json) -> ApiResult { let mut cfg = current_cfg(&state); cfg.protocols = p; state.reload(cfg)?; ok(json!({ "ok": true })) } // ───────────────────────────── settings ──────────────────────────────────── async fn get_settings(State(state): State) -> ApiResult { let live = state.live(); let c = &live.cfg; // admin section without token value (not stored here — only the env name) ok(json!({ "listen": c.listen, "router": c.router, "route": c.route, "breaker": c.breaker, "log": c.log, "telemetry": c.telemetry, "cortiq": c.cortiq, "stats": c.stats, "admin": c.admin, "cache": c.cache, "cmf": c.cmf, })) } #[derive(Deserialize)] struct SettingsBody { listen: Option, router: Option, route: Option, breaker: Option, log: Option, telemetry: Option, cortiq: Option, stats: Option, admin: Option, cache: Option, cmf: Option, } async fn put_settings(State(state): State, Json(b): Json) -> ApiResult { let mut cfg = current_cfg(&state); let mut needs_restart = false; if let Some(v) = b.listen { if v != cfg.listen { needs_restart = true; } cfg.listen = v; } if let Some(v) = b.router { cfg.router = v; } if let Some(v) = b.route { cfg.route = v; } if let Some(v) = b.breaker { cfg.breaker = v; } if let Some(v) = b.log { cfg.log = v; } if let Some(v) = b.telemetry { cfg.telemetry = v; } if let Some(v) = b.cortiq { cfg.cortiq = v; } if let Some(v) = b.stats { cfg.stats = v; } if let Some(v) = b.admin { if v.listen != cfg.admin.listen { needs_restart = true; } cfg.admin = v; } if let Some(v) = b.cache { // the cache is built at startup; persist now, apply on restart needs_restart = true; cfg.cache = v; } if let Some(v) = b.cmf { // manage_server / local_model / threads / gpu apply on restart (the server // is spawned once at startup with those env vars); local_only / router // routing apply immediately. if v.manage_server != cfg.cmf.manage_server || 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; } cfg.cmf = v; } state.reload(cfg)?; ok(json!({ "ok": true, "needs_restart": needs_restart })) } // ───────────────────────────── cmf runtime ───────────────────────────────── /// Status of the `cortiq` CLI (the CMF format runtime): is it installed, which /// version, and whether a managed server is running. Drives the "Install cortiq" /// button + status in the Settings → Local models card. async fn cmf_status(State(state): State) -> 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 = 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, "latest": st.latest_version, "manage_server": cmf.manage_server, "local_host": cmf.local_host, "servers": servers, "log": st.log, })) } /// Install/upgrade the `cortiq` CLI from crates.io on demand (spawned in the /// background; poll `GET /admin/api/cmf` for `installed`/`version`). async fn cmf_install(State(state): State) -> ApiResult { let bin = state.live().cfg.cmf.cortiq_bin.clone(); tokio::spawn(crate::cmf_runtime::install_now(state.cmf.clone(), bin)); ok(json!({ "started": true })) } /// List the `.cmf` model files available under `models_dir` so the UI can offer /// them as a dropdown instead of a hand-typed path. Includes tensor-dir `.cmf`. async fn cmf_files(State(state): State) -> ApiResult { let dir = state.live().cfg.cmf.models_dir.clone(); let mut entries: Vec<(String, String, Option, bool)> = Vec::new(); if let Ok(rd) = std::fs::read_dir(&dir) { for e in rd.flatten() { let name = e.file_name().to_string_lossy().to_string(); if !name.ends_with(".cmf") { continue; } let p = e.path(); let is_dir = p.is_dir(); let size = if is_dir { None } else { std::fs::metadata(&p).ok().map(|m| m.len()) }; let path = format!("{}/{}", dir.trim_end_matches('/'), name); entries.push((name, path, size, is_dir)); } } entries.sort_by(|a, b| a.0.cmp(&b.0)); let files: Vec = entries .into_iter() .map(|(name, path, size, is_dir)| json!({ "name": name, "path": path, "size": size, "is_dir": is_dir })) .collect(); ok(json!({ "dir": dir, "files": files })) } #[derive(Deserialize)] struct PortQuery { port: u16, } /// Check whether a TCP port is free to bind on the CMF local host, so a user can /// pick an available port before enabling the managed server (avoids clashing /// with something already on the default 8081). If the port is the one this /// gateway's own managed server already runs on, that is reported distinctly. /// When occupied, `suggested` holds the next free port so the UI can offer it. async fn cmf_port_check(State(state): State, Query(q): Query) -> ApiResult { let cmf = state.live().cfg.cmf.clone(); let host = if cmf.local_host.trim().is_empty() { "127.0.0.1".to_string() } else { cmf.local_host.clone() }; let st = state.cmf.status(); let port = q.port; 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 || { if running_here { return (false, "in use by this gateway's managed CMF server", None); } if port_is_free(&host, port) { (true, "free", None) } else { ( false, "occupied by another process", next_free_port(&host, port), ) } }) .await .unwrap_or((false, "port check failed", None)); ok(json!({ "port": port, "available": available, "detail": detail, "suggested": suggested, })) } /// Whether `port` is truly free on `host`. A bind test alone is not enough: a /// Docker-published / dual-stack listener can leave an IPv4 bind succeeding even /// though connections to the port are forwarded to a container. So a port counts /// as free only if it is BOTH bindable AND nothing answers a connect attempt. fn port_is_free(host: &str, port: u16) -> bool { if std::net::TcpListener::bind((host, port)).is_err() { return false; // something is bound to exactly this address } // catch proxied/forwarded ports a bind test misses (e.g. Docker on macOS) let reachable = (host, port) .to_socket_addrs() .ok() .and_then(|mut it| it.next()) .map(|addr| { std::net::TcpStream::connect_timeout(&addr, std::time::Duration::from_millis(200)) .is_ok() }) .unwrap_or(false); !reachable } /// First free port in (from, from+50], or None if all are taken. fn next_free_port(host: &str, from: u16) -> Option { (from.saturating_add(1)..=from.saturating_add(50)).find(|p| port_is_free(host, *p)) } // ───────────────────────────── keys ──────────────────────────────────────── async fn list_keys(State(state): State) -> ApiResult { let live = state.live(); let keys: Vec = live .cfg .api_keys .iter() .map(|k| { json!({ "key": k.key, "key_masked": mask_key(&k.key), "account": k.account, "rate_per_min": k.rate_per_min, "allow_models": k.allow_models, }) }) .collect(); ok(json!({ "keys": keys, "open_mode": live.cfg.api_keys.is_empty() })) } #[derive(Deserialize)] struct NewKey { #[serde(default)] key: String, account: String, #[serde(default)] rate_per_min: u32, #[serde(default)] allow_models: Vec, } async fn create_key(State(state): State, Json(b): Json) -> ApiResult { let key = if b.key.trim().is_empty() { format!("sk-gw-{}", random_token(18)) } else { b.key.trim().to_string() }; let mut cfg = current_cfg(&state); if cfg.api_keys.iter().any(|k| k.key == key) { return Err(ApiError::bad("key already exists")); } cfg.api_keys.push(ApiKeyCfg { key: key.clone(), account: b.account, rate_per_min: b.rate_per_min, allow_models: b.allow_models, }); state.reload(cfg)?; // show the full key only once ok(json!({ "ok": true, "key": key })) } async fn delete_key(State(state): State, Path(key): Path) -> ApiResult { let mut cfg = current_cfg(&state); let before = cfg.api_keys.len(); cfg.api_keys.retain(|k| k.key != key); if cfg.api_keys.len() == before { return Err(ApiError::not_found("key not found")); } state.reload(cfg)?; ok(json!({ "ok": true })) } // ───────────────────────────── secrets ───────────────────────────────────── async fn list_secrets(State(state): State) -> ApiResult { let live = state.live(); // collect all env names referenced in the config let mut names: Vec = Vec::new(); for m in &live.cfg.models { if let Some(env) = &m.api_key_env { if !names.contains(env) { names.push(env.clone()); } } } if let Some(env) = &live.cfg.router.api_key_env { if !names.contains(env) { names.push(env.clone()); } } let secrets: Vec = names .iter() .map(|n| json!({ "name": n, "source": state.secrets.source(n) })) .collect(); ok(json!({ "secrets": secrets })) } #[derive(Deserialize)] struct SecretBody { name: String, value: String, } async fn set_secret(State(state): State, Json(b): Json) -> ApiResult { if b.name.trim().is_empty() { return Err(ApiError::bad("secret name is required")); } state.secrets.set(b.name.trim(), &b.value)?; // rebuild providers with the new key (without writing TOML) state.rebuild()?; ok(json!({ "ok": true })) } #[derive(Deserialize)] struct NameQuery { name: String, } async fn clear_secret(State(state): State, Query(q): Query) -> ApiResult { state.secrets.clear(q.name.trim())?; state.rebuild()?; ok(json!({ "ok": true })) } // ───────────────────────────── stats / requests ──────────────────────────── #[derive(Deserialize)] struct StatsQuery { #[serde(default)] range: Option, #[serde(default)] groupby: Option, } async fn get_stats(State(state): State, Query(q): Query) -> ApiResult { let range_secs = q .range .as_deref() .and_then(parse_duration_secs) .unwrap_or(24 * 3600); let groupby = q.groupby.as_deref().unwrap_or("model"); let mut snap = state.stats.snapshot(range_secs, groupby); snap["cache"] = state.cache.snapshot(); ok(snap) } /// Clear all recorded stats/logs (in-memory aggregates, series, recent requests) /// and truncate the JSONL log — the dashboard "Clear logs" button. async fn clear_stats(State(state): State) -> ApiResult { state.stats.clear(); ok(json!({ "ok": true })) } /// 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 })) } // ── first-run onboarding ── fn setup_marker(state: &SharedState) -> std::path::PathBuf { std::path::Path::new(&state.config_path) .parent() .unwrap_or_else(|| std::path::Path::new(".")) .join(".setup_done") } async fn setup_status(State(state): State) -> ApiResult { let live = state.live(); let marker = setup_marker(&state); let done = marker.exists(); let has_models = !live.cfg.models.is_empty() || !live.cfg.cmf.effective_servers().is_empty(); let has_imports = !state.imports.list().is_empty(); ok(json!({ "setup_done": done, "has_models": has_models, "has_imports": has_imports, "needs_setup": !done && !has_models, "listen": live.cfg.listen, "version": env!("CARGO_PKG_VERSION"), })) } #[derive(Deserialize)] struct SetupCompleteBody { #[serde(default)] lang: Option, /// Connect finished downloads as local models (default: yes). #[serde(default)] register: Option, } /// Finish the wizard. Everything that finished downloading is connected as a /// local model here — the wizard promises exactly that, and leaving it to a /// separate button is how a first run ends with an empty model pool. async fn setup_complete( State(state): State, Json(b): Json, ) -> ApiResult { let marker = setup_marker(&state); if let Some(dir) = marker.parent() { let _ = std::fs::create_dir_all(dir); } let lang = b.lang.unwrap_or_else(|| "en".into()); let _ = std::fs::write( &marker, format!( "done lang={lang} ts={}", std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_secs() ), ); let mut registered: Vec = Vec::new(); if b.register.unwrap_or(true) { let finished: Vec = state .imports .list() .into_iter() .filter(|j| j.state == "done") .map(|j| j.output) .collect(); if !finished.is_empty() { let cortiq_bin = state.live().cfg.cmf.cortiq_bin.clone(); let mut cfg = current_cfg(&state); for output in &finished { // media packs (image/video) are used by the media API directly, // not served — connecting them as chat servers would only fail let modality = crate::media::classify_file(std::path::Path::new(output), &cortiq_bin).await; if modality != crate::media::Modality::Chat { continue; } if let Some(id) = register_cmf_output(&mut cfg, output) { registered.push(id); } } if cfg.routing.default.is_empty() { if let Some(first) = registered.first() { cfg.routing.default = first.clone(); } } if !registered.is_empty() { state.reload(cfg)?; } } } let runtime = crate::cmf_runtime::installed_version(&state.live().cfg.cmf.cortiq_bin); ok(json!({ "ok": true, "registered": registered, "runtime_installed": runtime.is_some(), "runtime_version": runtime, })) } /// Curated ready-to-run .cmf models from infosave (HF author) — proxied server-side. async fn setup_featured(State(state): State) -> 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) }; // Try HF author search; on failure return curated fallback let hf_result = crate::import::hf_search_author("infosave", 30, token.as_deref()).await; let models = match hf_result { Ok(v) => v, Err(_) => serde_json::json!([]), }; // curated fallback with real sizes (usedStorage from HF) let fallback = json!([ {"id": "infosave/Bonsai-1.7Bcmf", "label": "Bonsai 1.7B", "badge": "1.7B · Q1", "desc": "Small, fast local model — fits 4GB RAM", "size": "334 MB", "size_bytes": 334185278}, {"id": "infosave/Bonsai-8B_2bit_cmf", "label": "Bonsai 8B 2-bit", "badge": "8B · Q1T", "desc": "Balanced quality/size — 2-bit ternary", "size": "2.3 GB", "size_bytes": 2322861074_u64}, {"id": "infosave/Bonsai-27Bcmf", "label": "Bonsai 27B", "badge": "27B · Q1", "desc": "High quality — needs 16GB RAM", "size": "5.1 GB", "size_bytes": 5103462190_u64}, {"id": "infosave/Qwen3.6-27Bcmf", "label": "Qwen 3.6 27B CMF", "badge": "27B · Q4 · MoE", "desc": "MoE model, strong reasoning", "size": "14.3 GB", "size_bytes": 14264207150_u64}, {"id": "infosave/Nanbeige4.2-3Bcmf", "label": "Nanbeige 4.2 3B", "badge": "3B · looped", "desc": "Looped transformer, compact", "size": "2.4 GB", "size_bytes": 2364287523_u64}, {"id": "infosave/KAT-Coder-V2.5-CMF", "label": "KAT Coder 2.5", "badge": "code · MoE", "desc": "Code specialist", "size": "12.7 GB", "size_bytes": 12653975221_u64}, ]); let arr = if models.as_array().map(|a| a.is_empty()).unwrap_or(true) { fallback } else { // enrich HF data with label/badge/size for UI — fetch usedStorage concurrently let mut ids: Vec = Vec::new(); if let Some(arr) = models.as_array() { for m in arr { if let Some(id) = m["id"].as_str() { if id.to_ascii_lowercase().contains("cmf") { ids.push(id.to_string()); } } } } if ids.is_empty() { return ok(json!({ "models": fallback })); } // fetch per-variant sizes in parallel (tree API, 10s each) let token_owned = token.map(|s| s.to_string()); let variant_lists: Vec<(String, Vec<(String, u64)>)> = futures::future::join_all(ids.iter().map(|id| { let tid = token_owned.clone(); let iid = id.clone(); async move { let vars = crate::import::hf_repo_variants(&iid, tid.as_deref()).await; (iid, vars) } })) .await; let mut variant_map: std::collections::HashMap> = std::collections::HashMap::new(); let mut fallback_map: std::collections::HashMap = std::collections::HashMap::new(); let mut fallback_badge: std::collections::HashMap = std::collections::HashMap::new(); // also keep fallback sizes/badges for when tree is unreachable for v in fallback.as_array().unwrap_or(&vec![]) { if let (Some(id), Some(sz)) = (v["id"].as_str(), v["size_bytes"].as_u64()) { if let Some(s) = v["size"].as_str() { fallback_map.insert(id.to_string(), (s.to_string(), sz)); } } if let (Some(id), Some(b)) = (v["id"].as_str(), v["badge"].as_str()) { fallback_badge.insert(id.to_string(), b.to_string()); } } for (id, vars) in variant_lists { if !vars.is_empty() { variant_map.insert(id, vars); } } let mut out = Vec::new(); if let Some(arr) = models.as_array() { for m in arr { let id = m["id"].as_str().unwrap_or("").to_string(); if !id.to_ascii_lowercase().contains("cmf") { continue; } let likes = m["likes"].as_u64().unwrap_or(0); let vars = variant_map.get(&id); // extract size token from id let lower = id.to_ascii_lowercase(); let size_token = ["1.7b", "48b", "35b", "27b", "12b", "8b", "7b", "3b", "4b"] .iter() .find(|c| lower.contains(*c)) .map(|c| match *c { "1.7b" => "1.7B", "48b" => "48B", "35b" => "35B", "27b" => "27B", "12b" => "12B", "8b" => "8B", "7b" => "7B", "3b" => "3B", "4b" => "4B", _ => "CMF", }) .unwrap_or("CMF"); // quant from variant name if available (more accurate for Bonsai q1), else id let quant_src = vars .and_then(|vs| vs.first().map(|(n, _)| n.as_str())) .unwrap_or(&id) .to_ascii_lowercase(); let quant = if quant_src.contains("q1t") || quant_src.contains("2bit") || lower.contains("q1t") || lower.contains("2bit") { "Q1T" } else if quant_src.contains("q2tp") || lower.contains("q2tp") { "Q2TP" } else if quant_src.contains("q4tp") || lower.contains("q4tp") { "Q4TP" } else if quant_src.contains("q1") || lower.contains("q1") { "Q1" } else if quant_src.contains("q4") || lower.contains("q4") { "Q4" } else if quant_src.contains("q8") || lower.contains("q8") { "Q8" } else { "" }; let mut badge = if quant.is_empty() || size_token == "CMF" { if size_token == "CMF" { "CMF · ready".to_string() } else { size_token.to_string() } } else { format!("{size_token} · {quant}") }; // fallback has more accurate quant (e.g. Bonsai 1.7B Q1) when tree fetch failed if let Some(fb) = fallback_badge.get(&id) { if badge == size_token && fb.contains('·') { badge = fb.clone(); } } let (size, size_bytes, variants_json) = if let Some(vs) = vars { if vs.len() == 1 { let (name, sz) = &vs[0]; let s = crate::import::format_bytes(*sz); ( Some(s.clone()), Some(*sz), json!([{"name": name, "size": s, "size_bytes": sz}]), ) } else { // dedupe same-size variants (e.g. q2tp v2/v3 both 112GB → keep latest v3) let mut seen_sz = std::collections::HashSet::new(); let mut uniq_vs: Vec<(String, u64)> = Vec::new(); // iterate reversed to keep latest (v3) over v2 when same size for (n, sz) in vs.iter().rev() { let s = crate::import::format_bytes(*sz); if seen_sz.insert(s) { uniq_vs.push((n.clone(), *sz)); } } uniq_vs.reverse(); let parts: Vec = uniq_vs .iter() .map(|(_, sz)| crate::import::format_bytes(*sz)) .collect(); let max_sz = uniq_vs.iter().map(|(_, sz)| *sz).max(); let vars_json: Vec = uniq_vs.iter().map(|(n, sz)| json!({"name": n, "size": crate::import::format_bytes(*sz), "size_bytes": sz})).collect(); (Some(parts.join(" / ")), max_sz, json!(vars_json)) } } else if let Some((s, sz)) = fallback_map.get(&id) { ( Some(s.clone()), Some(*sz), json!([{"name": id.split('/').next_back().unwrap_or(&id), "size": s, "size_bytes": sz}]), ) } else { (None::, None::, Value::Null) }; // check if already downloaded locally (case-insensitive, handles sharded dirs) let models_dir = live.cfg.cmf.models_dir.clone(); let base = crate::import::sanitize_for_id(id.split('/').next_back().unwrap_or(&id)); let mut downloaded = false; let mut downloaded_variants: Vec = Vec::new(); if let Some(vs) = variants_json.as_array() { for v in vs { let vname = v["name"].as_str().unwrap_or(""); let vbase = if vs.len() == 1 { base.clone() } else { format!("{}-{}", base, crate::import::sanitize_for_id(vname)) }; let expected = format!("{vbase}.cmf").to_lowercase(); // case-insensitive search via read_dir (handles /tmp symlink and case) let mut exists = false; let mut local_sz: u64 = 0; if let Ok(rd) = std::fs::read_dir(&models_dir) { for entry in rd.filter_map(|e| e.ok()) { let name = entry.file_name().to_string_lossy().to_lowercase(); if name == expected { exists = true; local_sz = entry.metadata().map(|m| m.len()).unwrap_or(0); if local_sz == 0 && entry.path().is_dir() { local_sz = std::fs::read_dir(entry.path()) .ok() .map(|r| { r.filter_map(|e| e.ok()) .filter_map(|e| e.metadata().ok()) .map(|m| m.len()) .sum() }) .unwrap_or(0); } break; } } } // fallback direct path check (for non-listed or still being written) if !exists { let p = std::path::Path::new(&models_dir).join(format!("{vbase}.cmf")); if p.exists() || p.is_dir() { exists = true; local_sz = std::fs::metadata(&p).map(|m| m.len()).unwrap_or_else(|_| { std::fs::read_dir(&p) .ok() .map(|r| { r.filter_map(|e| e.ok()) .filter_map(|e| e.metadata().ok()) .map(|m| m.len()) .sum() }) .unwrap_or(0) }); } } let remote_sz = v["size_bytes"].as_u64().unwrap_or(0); let is_complete = exists && local_sz != 0 && remote_sz != 0 && local_sz == remote_sz; // downloaded = only when complete (not partial) downloaded = downloaded || is_complete; let mut vv = v.clone(); vv["downloaded"] = json!(is_complete); vv["local_size"] = json!(local_sz); // the exact output stem the import will produce, and the id of // the job that produces it — the UI passes both back verbatim so // download / progress / delete can never disagree on the file. vv["local_name"] = json!(vbase); vv["job"] = json!(crate::import::gen_id(&format!("{id}::{vname}"))); if exists && !is_complete && local_sz != 0 { // partial or outdated — show as update available if not currently downloading // we don't have job state here, so just mark update if sizes differ vv["update_available"] = json!(true); } else if is_complete && remote_sz != 0 && local_sz != remote_sz { vv["update_available"] = json!(true); } downloaded_variants.push(vv); } } let final_variants = if downloaded_variants.is_empty() { variants_json } else { json!(downloaded_variants) }; let any_update = downloaded_variants .iter() .any(|v| v["update_available"].as_bool().unwrap_or(false)); out.push(json!({ "id": id, "label": id.split('/').next_back().unwrap_or(&id), "local_name": base, "badge": badge, "desc": m["pipeline_tag"].as_str().unwrap_or("text-generation"), "likes": likes, "size": size, "size_bytes": size_bytes, "variants": final_variants, "downloaded": downloaded, "update_available": any_update, "raw": m, })); } } if out.is_empty() { fallback } else { json!(out) } }; ok(json!({ "models": arr })) } #[derive(Deserialize)] struct SetupFileQuery { repo: String, #[serde(default)] variant: Option, /// Exact output stem (`local_name` from `setup/featured`). Preferred over the /// repo/variant guess, which cannot know how many variants a repo has. #[serde(default)] name: Option, } /// Drop a downloaded `.cmf` from the config: any managed server or pool model /// backed by that stem. Returns the ids that were unregistered. fn unregister_cmf_stem(state: &SharedState, stem: &str) -> Result, ApiError> { let mut cfg = current_cfg(state); let stem_of = |path: &str| { std::path::Path::new(path) .file_stem() .map(|s| s.to_string_lossy().to_lowercase()) }; let mut ids: Vec = cfg .cmf .servers .iter() .filter(|s| stem_of(&s.model).as_deref() == Some(stem)) .map(|s| s.id.clone()) .collect(); for m in &cfg.models { if cmf_file_for_model(&cfg.cmf, m) .and_then(|p| stem_of(&p.to_string_lossy())) .as_deref() == Some(stem) { ids.push(m.id.clone()); } } if ids.is_empty() { return Ok(Vec::new()); } cfg.cmf .servers .retain(|s| stem_of(&s.model).as_deref() != Some(stem)); if stem_of(&cfg.cmf.local_model).as_deref() == Some(stem) { cfg.cmf.local_model = String::new(); cfg.cmf.manage_server = false; } cfg.models.retain(|m| !ids.contains(&m.id)); for id in &ids { strip_id_from_routing(&mut cfg.routing, id); } state.reload(cfg)?; Ok(ids) } async fn delete_setup_file( State(state): State, Query(q): Query, ) -> ApiResult { let repo = q.repo.trim(); if repo.is_empty() { return Err(ApiError::bad("repo is required")); } let live = state.live(); let models_dir = live.cfg.cmf.models_dir.clone(); let base = crate::import::sanitize_for_id(repo.split('/').next_back().unwrap_or(repo)); let variant = q .variant .as_deref() .map(|s| s.trim()) .filter(|s| !s.is_empty()); // Exact stem wins; sanitizing it also keeps the delete inside models_dir. let vbase = match q.name.as_deref().map(str::trim).filter(|s| !s.is_empty()) { Some(n) => crate::import::sanitize_for_id(n), // for single-file the file is {base}.cmf, for sharded {base}-{variant}.cmf None => match variant { // only sharded variants need suffix; single file ignores Some(v) if v.starts_with("parts-") => { format!("{}-{}", base, crate::import::sanitize_for_id(v)) } _ => base.clone(), }, }; if vbase.is_empty() { return Err(ApiError::bad("cannot derive a file name to delete")); } // Stop serving it before the file goes away, so a managed server doesn't keep // a deleted .cmf open (and the pool doesn't advertise a model that is gone). let unregistered = unregister_cmf_stem(&state, &vbase)?; for id in &unregistered { state.cmf.stop_one(id).await; } let p = std::path::Path::new(&models_dir).join(format!("{vbase}.cmf")); let mut removed = false; if p.exists() || p.is_dir() { removed = remove_cmf_path(&p); } else { // try case-insensitive search (macOS /tmp symlink, etc.) if let Ok(rd) = std::fs::read_dir(&models_dir) { let expected = format!("{vbase}.cmf").to_lowercase(); for entry in rd.filter_map(|e| e.ok()) { if entry.file_name().to_string_lossy().to_lowercase() == expected { removed = remove_cmf_path(&entry.path()); break; } } } } // also remove any job with same repo+variant id (so list doesn't keep stale "done") let jid = crate::import::gen_id(&format!("{}::{}", repo, variant.unwrap_or(""))); let _ = state.imports.delete(&jid); // also try to delete by repo alone (old non-variant ids) if variant.is_none() { let jid2 = crate::import::gen_id(repo); let _ = state.imports.delete(&jid2); } ok(json!({ "ok": true, "removed": removed, "unregistered": unregistered })) } 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 { let Some(j) = state.imports.get(&job) else { return Err(ApiError::not_found(format!("job '{job}' not found"))); }; // If it was already registered, drop it from the config first — otherwise the // pool keeps serving a model whose .cmf is about to disappear. let stem = std::path::Path::new(&j.output) .file_stem() .map(|s| s.to_string_lossy().to_lowercase()) .unwrap_or_default(); let unregistered = if stem.is_empty() { Vec::new() } else { unregister_cmf_stem(&state, &stem)? }; for id in &unregistered { state.cmf.stop_one(id).await; } let removed = state.imports.delete(&job).map_err(ApiError::bad)?; ok(json!({ "ok": removed, "unregistered": unregistered })) } /// Add a finished `.cmf` to the config as a *managed* local model: the gateway /// spawns and supervises its own `cortiq serve` for it. Registering it into the /// static pool instead would point the model at `cortiq_server_url` with nothing /// listening there, which is an upstream error the moment anyone calls it. /// /// A stale static-pool entry for the same file is migrated rather than rejected, /// so pressing Register again repairs a model left dead by an earlier version. /// `None` = already registered and served. fn register_cmf_output(cfg: &mut crate::config::Config, output: &str) -> Option { let base = std::path::Path::new(output) .file_stem() .map(|s| s.to_string_lossy().to_string()) .unwrap_or_else(|| "cmf-local".into()); let id = format!("cmf-{base}"); let served = cfg.cmf.manage_server && cfg .cmf .servers .iter() .any(|s| s.id == id || s.model == output); if served { return None; } // Same id sitting in the static pool → it is the dead form of this model. // Routing keeps referencing the id, so nothing needs re-pointing. cfg.models.retain(|m| m.id != id); cfg.cmf.servers.retain(|s| s.id != id && s.model != output); cfg.cmf.manage_server = true; let host = if cfg.cmf.local_host.trim().is_empty() { "127.0.0.1" } else { &cfg.cmf.local_host }; let base_port = cfg.cmf.servers.iter().map(|s| s.port).max().unwrap_or(8089); let port = next_free_port(host, base_port).unwrap_or_else(|| base_port + 1); cfg.cmf.servers.push(crate::config::CmfServer { id: id.clone(), model: output.to_string(), port, threads: cfg.cmf.threads, gpu: cfg.cmf.gpu, temperature: None, top_p: None, max_tokens: None, think_budget: None, system_prompt: None, o1: None, o1_m: None, o1_window: None, o1_sink: None, skip_mtp: false, }); Some(id) } /// 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 live = state.live(); // Image/video packs are not served by `cortiq serve` — they are used // directly by the media API. Nothing to register; tell the UI what it is. let modality = crate::media::classify_file(std::path::Path::new(&j.output), &live.cfg.cmf.cortiq_bin) .await; if modality != crate::media::Modality::Chat { return ok(json!({ "ok": true, "media_kind": modality, "model_id": null })); } let mut cfg = current_cfg(&state); let id = register_cmf_output(&mut cfg, &j.output).ok_or_else(|| { ApiError::bad(format!( "model 'cmf-{}' is already registered and running", std::path::Path::new(&j.output) .file_stem() .map(|s| s.to_string_lossy().to_string()) .unwrap_or_default() )) })?; if cfg.routing.default.is_empty() { cfg.routing.default = id.clone(); } state.reload(cfg)?; ok(json!({ "ok": true, "model_id": id })) } #[derive(Deserialize)] struct RecentQuery { #[serde(default)] limit: Option, #[serde(default)] offset: Option, } async fn get_requests(State(state): State, Query(q): Query) -> ApiResult { let limit = q.limit.unwrap_or(50).min(500); let offset = q.offset.unwrap_or(0); let items = state.stats.recent(limit, offset); ok(json!({ "requests": items })) } // ───────────────────────── system / logs / lifecycle ───────────────────────── /// Recent gateway log lines (in-memory ring) — for users who never see the /// terminal: Docker, autostart, a double-clicked exe. async fn get_logs(Query(q): Query) -> ApiResult { let limit = q.limit.unwrap_or(300).min(500); ok(json!({ "lines": crate::logbuf::snapshot(limit) })) } async fn clear_logs() -> ApiResult { crate::logbuf::clear(); ok(json!({ "ok": true })) } /// Host metrics snapshot: CPU, RAM, free disk at the models dir, gateway RSS. async fn get_system(State(state): State) -> ApiResult { let snap = state.sysmon.snapshot(); ok(serde_json::to_value(snap).unwrap_or_else(|_| json!({}))) } /// Restart the gateway in place: stop managed model servers, spawn a fresh /// copy of ourselves (it retries the bind while our port frees), then exit. /// In Docker we only exit — we are PID 1, a spawned child would die with the /// container anyway; the container's restart policy brings the gateway back. async fn restart_gateway(State(state): State) -> ApiResult { state.cmf.stop().await; let in_docker = std::path::Path::new("/.dockerenv").exists(); let exe = std::env::current_exe().map_err(|e| ApiError::bad(e.to_string()))?; let args: Vec = std::env::args().skip(1).collect(); tokio::spawn(async move { // let the 200 reach the browser first tokio::time::sleep(std::time::Duration::from_millis(300)).await; if !in_docker { let _ = std::process::Command::new(exe) .args(args) .env("CORTIQ_BIND_RETRY", "1") .spawn(); tokio::time::sleep(std::time::Duration::from_millis(200)).await; } std::process::exit(0); }); ok(json!({ "ok": true })) } #[derive(Deserialize)] struct BenchBody { model: String, } /// Benchmark a local .cmf with `cortiq bench --json` — real tok/s on this box. async fn bench_start(State(state): State, Json(b): Json) -> ApiResult { let live = state.live(); crate::bench::start(state.bench.clone(), &live.cfg.cmf, b.model).map_err(ApiError::bad)?; ok(json!({ "ok": true })) } async fn bench_list(State(state): State) -> ApiResult { ok(json!({ "runs": state.bench.list() })) } /// The config file as-is (comments included) for download. Secrets live in /// secrets.toml and are deliberately NOT part of the export. async fn config_export(State(state): State) -> Response { match tokio::fs::read_to_string(&state.config_path).await { Ok(text) => ( [ (header::CONTENT_TYPE, "application/toml".to_string()), ( header::CONTENT_DISPOSITION, "attachment; filename=\"gateway.toml\"".to_string(), ), ], text, ) .into_response(), Err(e) => ApiError::bad(format!("cannot read config: {e}")).into_response(), } } /// Replace the whole config from an uploaded TOML. Validated and hot-reloaded /// exactly like a panel edit; on any error the current config stays untouched. async fn config_import(State(state): State, body: String) -> ApiResult { let cfg: crate::config::Config = toml::from_str(&body).map_err(|e| ApiError::bad(format!("invalid TOML: {e}")))?; state.reload(cfg)?; ok(json!({ "ok": true })) } // ───────────────────────────── media generation ───────────────────────────── /// Local `.cmf` files with their modality (chat / image / video) — the /// playground's media tabs read this to know what can generate what. async fn media_models(State(state): State) -> ApiResult { let live = state.live(); let models = crate::media::list_media_models(&live.cfg.cmf.models_dir, &live.cfg.cmf.cortiq_bin).await; ok(json!({ "models": models })) } async fn media_jobs(State(state): State) -> ApiResult { ok(json!({ "jobs": state.media.list() })) } async fn media_generate( State(state): State, Json(p): Json, ) -> ApiResult { let live = state.live(); let id = crate::media::start_generation(state.media.clone(), &live.cfg.cmf, p) .map_err(ApiError::bad)?; ok(json!({ "job": id })) } async fn media_job_status(State(state): State, Path(job): Path) -> ApiResult { match state.media.get(&job) { Some(j) => { let mut v = serde_json::to_value(&j).unwrap_or_else(|_| json!({})); if j.state == "queued" { v["queue_position"] = json!(state.media.queue_position(&job)); } ok(v) } None => Err(ApiError::not_found(format!("media job '{job}' not found"))), } } async fn media_cancel(State(state): State, Path(job): Path) -> ApiResult { if state.media.get(&job).is_none() { return Err(ApiError::not_found(format!("media job '{job}' not found"))); } ok(json!({ "ok": state.media.cancel(&job, &crate::media::media_dir()) })) } async fn media_delete(State(state): State, Path(job): Path) -> ApiResult { if state.media.get(&job).is_none() { return Err(ApiError::not_found(format!("media job '{job}' not found"))); } let removed = state .media .delete(&job, &crate::media::media_dir()) .map_err(ApiError::bad)?; ok(json!({ "ok": removed })) } /// Serve a finished job's output bytes (ppm / avi / wav / mp4). The job id is /// server-generated hex and the file name is fixed per kind, so no path from the /// client ever reaches the filesystem. async fn media_file( State(state): State, Path((job, kind)): Path<(String, String)>, ) -> Response { let Some(j) = state.media.get(&job) else { return ApiError::not_found(format!("media job '{job}' not found")).into_response(); }; if j.state != "done" { return ApiError::bad("generation not finished").into_response(); } let Some((file, mime)) = crate::media::output_file(&kind) else { return ApiError::bad(format!("unknown output kind '{kind}'")).into_response(); }; let path = crate::media::media_dir().join(&job).join(file); match tokio::fs::read(&path).await { Ok(bytes) => ( [ (header::CONTENT_TYPE, mime.to_string()), ( header::CONTENT_DISPOSITION, format!( "inline; filename=\"cortiq-{job}.{}\"", file.rsplit('.').next().unwrap_or("bin") ), ), ], bytes, ) .into_response(), Err(_) => ApiError::not_found("output file missing on disk").into_response(), } } // ───────────────────────────── playground (test) ─────────────────────────── #[derive(Deserialize)] struct TestBody { #[serde(default = "default_test_model")] model: String, messages: Vec, #[serde(default)] temperature: Option, #[serde(default)] max_tokens: Option, } fn default_test_model() -> String { "cortiq-auto".into() } async fn run_test(State(state): State, Json(b): Json) -> ApiResult { if b.messages.is_empty() { return Err(ApiError::bad("messages must not be empty")); } let req = ChatRequest { routing: parse_routing(&b.model), messages: b.messages, tools: vec![], params: GenParams { temperature: b.temperature, max_tokens: b.max_tokens, ..Default::default() }, stream: false, meta: RequestMeta { account: "playground".into(), protocol: "playground".into(), ..Default::default() }, }; let started = Instant::now(); let res = state.pipeline.run(req, &state).await; let latency_ms = started.elapsed().as_millis() as u64; match res { Ok(resp) => { let answer = resp .choices .first() .map(|c| c.message.content.clone()) .unwrap_or_default(); ok(json!({ "ok": true, "latency_ms": latency_ms, "answer": answer, "model_used": resp.model_used, "usage": { "prompt_tokens": resp.usage.prompt_tokens, "completion_tokens": resp.usage.completion_tokens, "total_tokens": resp.usage.total_tokens, }, "cortiq": { "task_label": resp.cortiq.task_label, "complexity": { "score": resp.cortiq.complexity_score, "tier": resp.cortiq.complexity_tier }, "selected_model": resp.cortiq.selected_model, "route_source": resp.cortiq.route_source, "router_request_id": resp.cortiq.router_request_id, "cost_usd": resp.cortiq.cost_usd, "failover": resp.cortiq.failover, }, })) } Err(e) => ok(json!({ "ok": false, "latency_ms": latency_ms, "error": e.to_string() })), } } /// Streaming playground: forwards provider SSE chunks; routing decision is exposed /// via X-Cortiq-* response headers (the SPA reads them after the stream ends). async fn run_test_stream(State(state): State, Json(b): Json) -> Response { if b.messages.is_empty() { return ApiError::bad("messages must not be empty").into_response(); } let req = ChatRequest { routing: parse_routing(&b.model), messages: b.messages, tools: vec![], params: GenParams { temperature: b.temperature, max_tokens: b.max_tokens, ..Default::default() }, stream: true, meta: RequestMeta { account: "playground".into(), protocol: "playground".into(), ..Default::default() }, }; match state.pipeline.run_stream(req, &state).await { Ok((info, stream)) => { let mut headers = crate::protocols::openai_chat::cortiq_headers(&info); headers.insert( header::CONTENT_TYPE, axum::http::HeaderValue::from_static("text/event-stream"), ); (headers, axum::body::Body::from_stream(stream)).into_response() } Err(e) => ApiError::bad(e.to_string()).into_response(), } }