release: v0.2.37 — zero-argument startup, first-run wizard, real download progress
Startup no longer needs arguments: the config is created on first launch, a busy port is swapped for a free one, the console is open by default, and the browser lands on a setup wizard that downloads a ready-made .cmf model and wires it into routing (managed server + routing.default) when you press Finish. Download progress was reported from curl's own meter, which prints a full 0→100% bar for the -L redirect hop before the transfer starts — so every job showed 100% and then hung. Progress now comes from bytes on disk against the size from the HuggingFace tree (or a HEAD request). The UI and the gateway also disagreed about the output file name, so a finished download never marked its card as downloaded; the gateway now returns the exact stem and the UI sends it back verbatim for download, progress and delete. Featured models are read live from the infosave HF account in both the wizard and Import, and all 432 UI strings now exist in every one of the 7 languages.
This commit is contained in:
infosave2007
+539
-68
@@ -20,7 +20,7 @@ use axum::{
|
||||
http::{header, StatusCode},
|
||||
middleware::{self, Next},
|
||||
response::{IntoResponse, Response},
|
||||
routing::{get, post},
|
||||
routing::{delete, get, post},
|
||||
Json, Router,
|
||||
};
|
||||
use serde::Deserialize;
|
||||
@@ -139,10 +139,11 @@ fn parse_routing(model: &str) -> RoutingDirective {
|
||||
|
||||
// ───────────────────────────── router ───────────────────────────────────────
|
||||
|
||||
/// Admin API routes (token-protected). Mounted into the main router.
|
||||
/// Admin API routes. If `admin_token` is empty, the panel is open (no auth);
|
||||
/// otherwise every `/admin/api/*` request needs `Authorization: Bearer <token>`.
|
||||
/// The token is captured in a middleware closure so it does not depend on State.
|
||||
pub fn api_routes(admin_token: String) -> Router<SharedState> {
|
||||
Router::new()
|
||||
let router = Router::new()
|
||||
.route("/admin/api/health", get(health))
|
||||
.route("/admin/api/router/probe", post(router_probe))
|
||||
.route("/admin/api/meta", get(meta))
|
||||
@@ -183,10 +184,18 @@ pub fn api_routes(admin_token: String) -> Router<SharedState> {
|
||||
.route("/admin/api/requests", get(get_requests))
|
||||
.route("/admin/api/test", post(run_test))
|
||||
.route("/admin/api/test/stream", post(run_test_stream))
|
||||
.route_layer(middleware::from_fn(move |req: Request, next: Next| {
|
||||
.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.
|
||||
@@ -1344,6 +1353,450 @@ async fn hf_search(State(state): State<SharedState>, Query(q): Query<HfQuery>) -
|
||||
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<SharedState>) -> 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<String>,
|
||||
/// Connect finished downloads as local models (default: yes).
|
||||
#[serde(default)]
|
||||
register: Option<bool>,
|
||||
}
|
||||
|
||||
/// 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<SharedState>,
|
||||
Json(b): Json<SetupCompleteBody>,
|
||||
) -> 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<String> = Vec::new();
|
||||
if b.register.unwrap_or(true) {
|
||||
let finished: Vec<String> = state
|
||||
.imports
|
||||
.list()
|
||||
.into_iter()
|
||||
.filter(|j| j.state == "done")
|
||||
.map(|j| j.output)
|
||||
.collect();
|
||||
if !finished.is_empty() {
|
||||
let mut cfg = current_cfg(&state);
|
||||
for output in &finished {
|
||||
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<SharedState>) -> 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<String> = 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<String, Vec<(String, u64)>> =
|
||||
std::collections::HashMap::new();
|
||||
let mut fallback_map: std::collections::HashMap<String, (String, u64)> =
|
||||
std::collections::HashMap::new();
|
||||
let mut fallback_badge: std::collections::HashMap<String, String> =
|
||||
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<String> = 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<Value> = 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('/').last().unwrap_or(&id), "size": s, "size_bytes": sz}]))
|
||||
} else {
|
||||
(None::<String>, None::<u64>, 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('/').last().unwrap_or(&id));
|
||||
let mut downloaded = false;
|
||||
let mut downloaded_variants: Vec<Value> = 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('/').last().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<String>,
|
||||
/// 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<String>,
|
||||
}
|
||||
|
||||
/// 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<Vec<String>, 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<String> = 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<SharedState>,
|
||||
Query(q): Query<SetupFileQuery>,
|
||||
) -> 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('/').last().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<SharedState>) -> ApiResult {
|
||||
ok(json!({ "jobs": state.imports.list() }))
|
||||
}
|
||||
@@ -1377,11 +1830,80 @@ async fn cancel_import(State(state): State<SharedState>, Path(job): Path<String>
|
||||
|
||||
/// Delete a finished conversion and its converted `.cmf` file(s) from disk.
|
||||
async fn delete_import(State(state): State<SharedState>, Path(job): Path<String>) -> ApiResult {
|
||||
if state.imports.get(&job).is_none() {
|
||||
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 }))
|
||||
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<String> {
|
||||
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).
|
||||
@@ -1393,69 +1915,18 @@ async fn register_import(State(state): State<SharedState>, Path(job): Path<Strin
|
||||
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);
|
||||
|
||||
let is_in_models = cfg.models.iter().any(|m| m.id == id);
|
||||
let is_in_servers = cfg
|
||||
.cmf
|
||||
.servers
|
||||
.iter()
|
||||
.any(|s| s.id == id || s.model == j.output);
|
||||
|
||||
if is_in_models || is_in_servers {
|
||||
return Err(ApiError::bad(format!("model '{id}' is already registered")));
|
||||
}
|
||||
|
||||
if cfg.cmf.manage_server {
|
||||
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: j.output.clone(),
|
||||
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,
|
||||
});
|
||||
} else {
|
||||
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(),
|
||||
temperature: None,
|
||||
top_p: None,
|
||||
max_tokens: None,
|
||||
think_budget: None,
|
||||
system_prompt: None,
|
||||
o1: None,
|
||||
skip_mtp: false,
|
||||
});
|
||||
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 }))
|
||||
|
||||
@@ -9,6 +9,9 @@ use std::path::PathBuf;
|
||||
/// `$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).
|
||||
/// On Windows also checks `%APPDATA%`, `%LOCALAPPDATA%` and `%USERPROFILE%`
|
||||
/// so double-clicking the exe from `Downloads` does not try to write into
|
||||
/// `Program Files`.
|
||||
pub fn data_dir() -> PathBuf {
|
||||
for (var, sub) in [
|
||||
("CORTIQ_GATEWAY_HOME", ""),
|
||||
@@ -25,6 +28,20 @@ pub fn data_dir() -> PathBuf {
|
||||
}
|
||||
}
|
||||
}
|
||||
#[cfg(windows)]
|
||||
{
|
||||
for (var, sub) in [
|
||||
("APPDATA", "cortiq-gateway"),
|
||||
("LOCALAPPDATA", "cortiq-gateway"),
|
||||
("USERPROFILE", ".cortiq-gateway"),
|
||||
] {
|
||||
if let Ok(v) = std::env::var(var) {
|
||||
if !v.is_empty() {
|
||||
return PathBuf::from(v).join(sub);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
PathBuf::from("data")
|
||||
}
|
||||
|
||||
@@ -702,6 +719,64 @@ fn default_semcache_max() -> usize {
|
||||
1000
|
||||
}
|
||||
|
||||
/// Sensible defaults for a fresh install (empty pool — user fills via wizard/UI).
|
||||
impl Default for Config {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
listen: "0.0.0.0:9000".to_string(),
|
||||
router: RouterCfg {
|
||||
enabled: true,
|
||||
url: "http://localhost:8080".to_string(),
|
||||
timeout_ms: default_router_timeout(),
|
||||
verify_tls: true,
|
||||
api_key_env: Some("CORTIQ_ROUTER_KEY".to_string()),
|
||||
taxonomy_id: Some("general.task-type".to_string()),
|
||||
},
|
||||
route: RouteCfg::default(),
|
||||
models: Vec::new(),
|
||||
routing: RoutingCfg {
|
||||
tiers: HashMap::new(),
|
||||
default: String::new(),
|
||||
policy: RoutingPolicy::default(),
|
||||
},
|
||||
breaker: BreakerCfg::default(),
|
||||
protocols: ProtocolsCfg {
|
||||
openai_chat: true,
|
||||
openai_completions: true,
|
||||
openai_embeddings: true,
|
||||
openai_models: true,
|
||||
anthropic_messages: false,
|
||||
mcp: false,
|
||||
native_passthrough: true,
|
||||
},
|
||||
cortiq: CortiqCfg::default(),
|
||||
api_keys: Vec::new(),
|
||||
idempotency: IdempotencyCfg::default(),
|
||||
log: LogCfg::default(),
|
||||
telemetry: TelemetryCfg::default(),
|
||||
admin: AdminCfg::default(),
|
||||
stats: StatsCfg::default(),
|
||||
cache: CacheCfg::default(),
|
||||
shadow: ShadowCfg::default(),
|
||||
cmf: CmfCfg::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Where the config lives when `--config` is not given.
|
||||
/// Repo checkout (`config/gateway.toml` exists or `config/` dir exists) → keep it there;
|
||||
/// installed binary (no repo) → `$data_dir/gateway.toml`.
|
||||
pub fn default_config_path() -> String {
|
||||
let repo_path = "config/gateway.toml";
|
||||
if std::path::Path::new(repo_path).exists() {
|
||||
return repo_path.to_string();
|
||||
}
|
||||
if std::path::Path::new("Cargo.toml").exists() && std::path::Path::new("config").exists() {
|
||||
return repo_path.to_string();
|
||||
}
|
||||
data_dir().join("gateway.toml").to_string_lossy().into_owned()
|
||||
}
|
||||
|
||||
impl Config {
|
||||
/// Load and validate the config from a file.
|
||||
pub fn load(path: &str) -> anyhow::Result<Self> {
|
||||
@@ -712,6 +787,22 @@ impl Config {
|
||||
Ok(cfg)
|
||||
}
|
||||
|
||||
/// Load if exists, otherwise create a fresh default config at `path` and return it.
|
||||
/// Never requires the user to create or copy a file by hand.
|
||||
pub fn load_or_create(path: &str) -> anyhow::Result<Self> {
|
||||
if std::path::Path::new(path).exists() {
|
||||
return Self::load(path);
|
||||
}
|
||||
let cfg = Self::default();
|
||||
// ensure parent dir exists
|
||||
if let Some(dir) = std::path::Path::new(path).parent() {
|
||||
let _ = std::fs::create_dir_all(dir);
|
||||
}
|
||||
cfg.save(path)?;
|
||||
eprintln!("created default config at {path} — open /admin to finish setup");
|
||||
Ok(cfg)
|
||||
}
|
||||
|
||||
/// Integrity validation: non-empty model pool, valid routing targets.
|
||||
pub fn validate(&self) -> anyhow::Result<()> {
|
||||
// A managed local CMF model counts as an available model even though it
|
||||
|
||||
+483
-43
@@ -30,6 +30,47 @@ fn cleanup_output(path: &str) {
|
||||
}
|
||||
}
|
||||
|
||||
/// Bytes an output occupies: a file's length, or the recursive sum for a sharded
|
||||
/// `.cmf` directory. 0 when nothing has been written yet.
|
||||
pub fn path_size(p: &std::path::Path) -> u64 {
|
||||
match std::fs::metadata(p) {
|
||||
Ok(m) if m.is_file() => m.len(),
|
||||
Ok(m) if m.is_dir() => std::fs::read_dir(p)
|
||||
.map(|rd| {
|
||||
rd.filter_map(|e| e.ok())
|
||||
.map(|e| path_size(&e.path()))
|
||||
.sum()
|
||||
})
|
||||
.unwrap_or(0),
|
||||
_ => 0,
|
||||
}
|
||||
}
|
||||
|
||||
/// Aborts the wrapped task when it goes out of scope — used for the progress
|
||||
/// poller so every early return in a download path stops it.
|
||||
struct AbortOnDrop(tokio::task::JoinHandle<()>);
|
||||
impl Drop for AbortOnDrop {
|
||||
fn drop(&mut self) {
|
||||
self.0.abort();
|
||||
}
|
||||
}
|
||||
|
||||
/// Drive progress from bytes on disk. curl's own meter is unreliable here (its
|
||||
/// bar is redrawn with `\r` into a pipe, and a sharded download runs one curl
|
||||
/// per part), but the output size never lies.
|
||||
fn spawn_size_poller(store: Arc<JobStore>, id: String, path: String) -> AbortOnDrop {
|
||||
AbortOnDrop(tokio::spawn(async move {
|
||||
let p = std::path::PathBuf::from(&path);
|
||||
loop {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(600)).await;
|
||||
let done = path_size(&p);
|
||||
if done > 0 {
|
||||
store.set_bytes(&id, done);
|
||||
}
|
||||
}
|
||||
}))
|
||||
}
|
||||
|
||||
fn now() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
@@ -37,16 +78,12 @@ fn now() -> u64 {
|
||||
.as_secs()
|
||||
}
|
||||
|
||||
/// Job id from a hash of (repo, nanos) — unique without a uuid dependency.
|
||||
fn gen_id(repo: &str) -> String {
|
||||
/// Job id from repo (deterministic) — same repo+variant reuses same id, so
|
||||
/// the job list doesn't fill with duplicates and "already downloaded" can be detected.
|
||||
pub 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())
|
||||
}
|
||||
|
||||
@@ -58,6 +95,10 @@ pub struct ImportParams {
|
||||
pub quant: String,
|
||||
#[serde(default)]
|
||||
pub name: String, // output basename; empty → derived from repo
|
||||
/// For sharded CMF repos (e.g. DeepSeek with `parts-q2tp-v2`/`parts-q4tp`), which
|
||||
/// variant directory to download. Empty = auto (first .cmf or first variant).
|
||||
#[serde(default)]
|
||||
pub variant: Option<String>,
|
||||
// ── advanced ──
|
||||
#[serde(default)]
|
||||
pub linear_core: Option<String>, // gated_delta_net | vmf_phase
|
||||
@@ -97,6 +138,11 @@ pub struct Job {
|
||||
pub repo: String,
|
||||
pub quant: String,
|
||||
pub output: String,
|
||||
/// Output basename without `.cmf` — the id the UI uses to match a job to the
|
||||
/// featured card that started it, and to the file on disk.
|
||||
pub name: String,
|
||||
/// Sharded-variant directory / `.cmf` filename this job downloads, if any.
|
||||
pub variant: Option<String>,
|
||||
pub state: String, // running | done | error | cancelled
|
||||
pub log: Vec<String>,
|
||||
pub started: u64,
|
||||
@@ -104,6 +150,10 @@ pub struct Job {
|
||||
pub size_bytes: Option<u64>,
|
||||
pub progress: Option<f32>, // 0..1, parsed from converter @PROGRESS
|
||||
pub phase: Option<String>, // current phase label
|
||||
/// Expected download size (from the HF tree) and bytes already on disk —
|
||||
/// what makes the percentage in the UI real rather than a guess.
|
||||
pub total_bytes: Option<u64>,
|
||||
pub done_bytes: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
@@ -154,12 +204,53 @@ impl JobStore {
|
||||
}
|
||||
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);
|
||||
let f = frac.clamp(0.0, 1.0);
|
||||
let cur = j.progress.unwrap_or(0.0);
|
||||
// monotonic only — never go backwards (curl reprints, re-download)
|
||||
if f > cur + 0.001 {
|
||||
j.progress = Some(f);
|
||||
} else if j.progress.is_none() {
|
||||
j.progress = Some(f);
|
||||
}
|
||||
if !phase.is_empty() {
|
||||
j.phase = Some(phase);
|
||||
}
|
||||
}
|
||||
}
|
||||
/// Expected total download size, once the HF tree is known.
|
||||
fn set_total(&self, id: &str, total: u64) {
|
||||
if let Some(j) = self.jobs.lock().unwrap().get_mut(id) {
|
||||
if total > 0 {
|
||||
j.total_bytes = Some(total);
|
||||
}
|
||||
}
|
||||
}
|
||||
/// Report bytes on disk; derives the percentage whenever the total is known.
|
||||
fn set_bytes(&self, id: &str, done: u64) {
|
||||
let total = {
|
||||
let mut g = self.jobs.lock().unwrap();
|
||||
let Some(j) = g.get_mut(id) else { return };
|
||||
if j.state != "running" {
|
||||
return;
|
||||
}
|
||||
if done > j.done_bytes.unwrap_or(0) {
|
||||
j.done_bytes = Some(done);
|
||||
}
|
||||
j.total_bytes
|
||||
};
|
||||
if let Some(t) = total.filter(|t| *t > 0) {
|
||||
self.set_progress(id, done as f32 / t as f32, "downloading".into());
|
||||
}
|
||||
}
|
||||
fn set_phase(&self, id: &str, phase: String) {
|
||||
if let Some(j) = self.jobs.lock().unwrap().get_mut(id) {
|
||||
j.phase = Some(phase);
|
||||
// keep progress as is; if it was 0.0, make it indeterminate (None) so bar animates
|
||||
if j.progress == Some(0.0) {
|
||||
j.progress = None;
|
||||
}
|
||||
}
|
||||
}
|
||||
fn update_quant(&self, id: &str, quant: String) {
|
||||
if let Some(j) = self.jobs.lock().unwrap().get_mut(id) {
|
||||
j.quant = quant.clone();
|
||||
@@ -180,11 +271,14 @@ impl JobStore {
|
||||
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;
|
||||
let on_disk = path_size(std::path::Path::new(&j.output));
|
||||
j.size_bytes = if on_disk > 0 { Some(on_disk) } else { None };
|
||||
let done = ok && on_disk > 0;
|
||||
j.state = if done { "done" } else { "error" }.into();
|
||||
if done {
|
||||
j.progress = Some(1.0);
|
||||
j.done_bytes = Some(on_disk);
|
||||
j.total_bytes = Some(j.total_bytes.unwrap_or(on_disk).max(on_disk));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -205,7 +299,17 @@ impl JobStore {
|
||||
Ok(true)
|
||||
}
|
||||
pub fn list(&self) -> Vec<Job> {
|
||||
let mut v: Vec<Job> = self.jobs.lock().unwrap().values().cloned().collect();
|
||||
let mut map: HashMap<String, Job> = HashMap::new();
|
||||
for j in self.jobs.lock().unwrap().values().cloned() {
|
||||
// dedupe by repo+variant — keep the latest (so re-download doesn't create
|
||||
// duplicate rows) while two variants of one repo stay separate rows
|
||||
let key = format!("{}::{}", j.repo, j.variant.clone().unwrap_or_default());
|
||||
match map.get(&key) {
|
||||
Some(prev) if prev.started >= j.started => {}
|
||||
_ => { map.insert(key, j); }
|
||||
}
|
||||
}
|
||||
let mut v: Vec<Job> = map.into_values().collect();
|
||||
v.sort_by_key(|x| std::cmp::Reverse(x.started));
|
||||
v.truncate(20);
|
||||
v
|
||||
@@ -242,6 +346,149 @@ pub async fn hf_search(
|
||||
resp.json().await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn hf_search_author(
|
||||
author: &str,
|
||||
limit: usize,
|
||||
token: Option<&str>,
|
||||
) -> Result<serde_json::Value, String> {
|
||||
let url = format!(
|
||||
"https://huggingface.co/api/models?author={}&limit={}&sort=trendingScore&direction=-1",
|
||||
url_escape(author),
|
||||
limit
|
||||
);
|
||||
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())
|
||||
}
|
||||
|
||||
pub async fn hf_model_size(id: &str, token: Option<&str>) -> Option<u64> {
|
||||
let url = format!("https://huggingface.co/api/models/{}", url_escape(id));
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(std::time::Duration::from_secs(8))
|
||||
.build()
|
||||
.ok()?;
|
||||
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.ok()?;
|
||||
if !resp.status().is_success() {
|
||||
return None;
|
||||
}
|
||||
let v: serde_json::Value = resp.json().await.ok()?;
|
||||
v.get("usedStorage").and_then(|x| x.as_u64())
|
||||
}
|
||||
|
||||
/// Per-variant sizes for a repo (for CMF repos with sharded directories like
|
||||
/// `parts-q2tp-v2`, `parts-q4tp`). Returns `[(variant_name, bytes)]`. For a
|
||||
/// single-file repo returns one entry. Empty = tree unreachable.
|
||||
pub async fn hf_repo_variants(id: &str, token: Option<&str>) -> Vec<(String, u64)> {
|
||||
let url = format!(
|
||||
"https://huggingface.co/api/models/{}/tree/main?recursive=true",
|
||||
url_escape(id)
|
||||
);
|
||||
let client = match reqwest::Client::builder()
|
||||
.timeout(std::time::Duration::from_secs(10))
|
||||
.build()
|
||||
{
|
||||
Ok(c) => c,
|
||||
Err(_) => return Vec::new(),
|
||||
};
|
||||
let mut req = client.get(&url).header("User-Agent", "cortiq-gateway");
|
||||
if let Some(t) = token {
|
||||
req = req.bearer_auth(t);
|
||||
}
|
||||
let resp = match req.send().await {
|
||||
Ok(r) if r.status().is_success() => r,
|
||||
_ => return Vec::new(),
|
||||
};
|
||||
let arr: serde_json::Value = match resp.json().await {
|
||||
Ok(v) => v,
|
||||
Err(_) => return Vec::new(),
|
||||
};
|
||||
let list = match arr.as_array() {
|
||||
Some(a) => a,
|
||||
None => return Vec::new(),
|
||||
};
|
||||
use std::collections::HashMap;
|
||||
let mut dir_sizes: HashMap<String, u64> = HashMap::new();
|
||||
let mut single_cmf: Vec<(String, u64)> = Vec::new();
|
||||
for item in list {
|
||||
if item.get("type").and_then(|t| t.as_str()) != Some("file") {
|
||||
continue;
|
||||
}
|
||||
let path = match item.get("path").and_then(|p| p.as_str()) {
|
||||
Some(p) => p,
|
||||
None => continue,
|
||||
};
|
||||
let sz = item
|
||||
.get("size")
|
||||
.and_then(|v| v.as_u64())
|
||||
.or_else(|| item.get("lfs").and_then(|l| l.get("size")).and_then(|v| v.as_u64()))
|
||||
.unwrap_or(0);
|
||||
if sz == 0 {
|
||||
continue;
|
||||
}
|
||||
if path.to_ascii_lowercase().ends_with(".cmf") {
|
||||
single_cmf.push((path.to_string(), sz));
|
||||
} else if path.contains('/') && path.starts_with("parts-") {
|
||||
let top = path.split('/').next().unwrap_or(path).to_string();
|
||||
*dir_sizes.entry(top).or_default() += sz;
|
||||
}
|
||||
}
|
||||
if !single_cmf.is_empty() {
|
||||
return single_cmf;
|
||||
}
|
||||
let mut out: Vec<(String, u64)> = dir_sizes.into_iter().collect();
|
||||
out.sort_by(|a, b| a.0.cmp(&b.0));
|
||||
out
|
||||
}
|
||||
|
||||
/// `Content-Length` of a download URL (redirects followed) — the fallback total
|
||||
/// when the repo tree didn't tell us how big the file is.
|
||||
async fn head_content_length(url: &str, token: Option<&str>) -> Option<u64> {
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(std::time::Duration::from_secs(10))
|
||||
.build()
|
||||
.ok()?;
|
||||
let mut req = client.head(url).header("User-Agent", "cortiq-gateway");
|
||||
if let Some(t) = token {
|
||||
req = req.bearer_auth(t);
|
||||
}
|
||||
let resp = req.send().await.ok()?;
|
||||
if !resp.status().is_success() {
|
||||
return None;
|
||||
}
|
||||
resp.headers()
|
||||
.get(reqwest::header::CONTENT_LENGTH)?
|
||||
.to_str()
|
||||
.ok()?
|
||||
.parse()
|
||||
.ok()
|
||||
}
|
||||
|
||||
pub fn format_bytes(n: u64) -> String {
|
||||
if n >= 1_000_000_000 {
|
||||
format!("{:.1} GB", n as f64 / 1_000_000_000.0)
|
||||
} else if n >= 1_000_000 {
|
||||
format!("{} MB", n / 1_000_000)
|
||||
} else if n >= 1_000 {
|
||||
format!("{} KB", n / 1_000)
|
||||
} else {
|
||||
format!("{n} B")
|
||||
}
|
||||
}
|
||||
|
||||
fn url_escape(s: &str) -> String {
|
||||
s.chars()
|
||||
.map(|c| match c {
|
||||
@@ -263,6 +510,9 @@ fn sanitize(s: &str) -> String {
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
pub fn sanitize_for_id(s: &str) -> String {
|
||||
sanitize(&s.to_lowercase())
|
||||
}
|
||||
|
||||
/// Kick off a conversion. Returns the job id immediately; progress streams
|
||||
/// Map the UI quant label to the native `cortiq` quant flag.
|
||||
@@ -293,26 +543,44 @@ async fn download_cmf_repo(
|
||||
store: Arc<JobStore>,
|
||||
id: String,
|
||||
cancel: Arc<Notify>,
|
||||
variant: Option<String>,
|
||||
) {
|
||||
store.set_progress(&id, 0.0, "listing files".into());
|
||||
// don't show 0% "listing files" — it looks stuck; show indeterminate "preparing"
|
||||
store.set_phase(&id, "preparing".into());
|
||||
|
||||
let tree_url = format!("https://huggingface.co/api/models/{repo}/tree/main?recursive=true");
|
||||
let client = reqwest::Client::new();
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(std::time::Duration::from_secs(10))
|
||||
.build()
|
||||
.unwrap_or_else(|_| reqwest::Client::new());
|
||||
let mut req = client.get(&tree_url).header("User-Agent", "cortiq-gateway");
|
||||
if let Some(t) = &hf_token {
|
||||
req = req.bearer_auth(t);
|
||||
}
|
||||
|
||||
let mut cmf_rel_path = None;
|
||||
let mut cmf_files: Vec<(String, u64)> = Vec::new();
|
||||
let mut sharded_dirs: std::collections::HashMap<String, Vec<String>> = std::collections::HashMap::new();
|
||||
let mut dir_sizes: std::collections::HashMap<String, u64> = std::collections::HashMap::new();
|
||||
if let Ok(resp) = req.send().await {
|
||||
if resp.status().is_success() {
|
||||
if let Ok(json) = resp.json::<serde_json::Value>().await {
|
||||
if let Some(arr) = json.as_array() {
|
||||
for item in arr {
|
||||
if item.get("type").and_then(|t| t.as_str()) != Some("file") {
|
||||
continue;
|
||||
}
|
||||
if let Some(path) = item.get("path").and_then(|p| p.as_str()) {
|
||||
let sz = item
|
||||
.get("size")
|
||||
.and_then(|v| v.as_u64())
|
||||
.or_else(|| item.get("lfs").and_then(|l| l.get("size")).and_then(|v| v.as_u64()))
|
||||
.unwrap_or(0);
|
||||
if path.to_ascii_lowercase().ends_with(".cmf") {
|
||||
cmf_rel_path = Some(path.to_string());
|
||||
break;
|
||||
cmf_files.push((path.to_string(), sz));
|
||||
} else if path.contains('/') && path.starts_with("parts-") {
|
||||
let top = path.split('/').next().unwrap_or(path).to_string();
|
||||
sharded_dirs.entry(top.clone()).or_default().push(path.to_string());
|
||||
*dir_sizes.entry(top).or_default() += sz;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -321,12 +589,147 @@ async fn download_cmf_repo(
|
||||
}
|
||||
}
|
||||
|
||||
let cmf_path = match cmf_rel_path {
|
||||
Some(p) => p,
|
||||
None => {
|
||||
let repo_name = repo.rsplit('/').next().unwrap_or(&repo);
|
||||
format!("{repo_name}.cmf")
|
||||
// Decide what to download: single .cmf file vs sharded variant directory
|
||||
let is_sharded = !sharded_dirs.is_empty();
|
||||
let chosen_variant = variant.as_deref().map(|v| v.trim()).filter(|v| !v.is_empty());
|
||||
|
||||
if is_sharded {
|
||||
// pick variant: explicit > first alphabetically
|
||||
let target_dir = if let Some(v) = chosen_variant {
|
||||
if sharded_dirs.contains_key(v) { v.to_string() } else { sharded_dirs.keys().next().cloned().unwrap_or_default() }
|
||||
} else {
|
||||
// prefer smallest? use first sorted
|
||||
let mut keys: Vec<String> = sharded_dirs.keys().cloned().collect();
|
||||
keys.sort();
|
||||
keys.into_iter().next().unwrap_or_default()
|
||||
};
|
||||
let files = sharded_dirs.get(&target_dir).cloned().unwrap_or_default();
|
||||
if files.is_empty() {
|
||||
store.push_line(&id, format!("✗ variant '{target_dir}' not found"));
|
||||
store.finish(&id, false);
|
||||
store.drop_cancel(&id);
|
||||
return;
|
||||
}
|
||||
let total_sz = dir_sizes.get(&target_dir).copied().unwrap_or(0);
|
||||
let q = if target_dir.contains("q2tp") { "Q2TP" } else if target_dir.contains("q4tp") { "Q4TP" } else { "READY_CMF" };
|
||||
store.update_quant(&id, q.to_string());
|
||||
store.push_line(&id, format!("→ downloading {} ({} files, {}) from HuggingFace...", target_dir, files.len(), format_bytes(total_sz)));
|
||||
store.set_total(&id, total_sz);
|
||||
store.set_progress(&id, 0.001, "downloading".into());
|
||||
|
||||
// output is a directory: ensure {base}.cmf is a directory
|
||||
let out_dir = std::path::Path::new(&output_abs);
|
||||
// Drop a previous copy first (re-download / update). There is no resume,
|
||||
// so it would be overwritten anyway — and leaving it there would make the
|
||||
// size poller read the old bytes and report 100% before we transfer any.
|
||||
cleanup_output(&output_abs);
|
||||
// if output ends with .cmf and variant is sharded, keep it as dir (e.g. model.cmf/parts-*)
|
||||
// create base dir
|
||||
let _ = std::fs::create_dir_all(out_dir);
|
||||
// live percentage from bytes landing on disk (curl runs quiet, per part)
|
||||
let _poller = spawn_size_poller(store.clone(), id.clone(), output_abs.clone());
|
||||
// download each part file
|
||||
for (idx, rel) in files.iter().enumerate() {
|
||||
let filename = rel.rsplit('/').next().unwrap_or(rel);
|
||||
let dest = out_dir.join(filename);
|
||||
let url = format!("https://huggingface.co/{repo}/resolve/main/{rel}");
|
||||
store.push_line(&id, format!(" [{}/{}] {filename}", idx + 1, files.len()));
|
||||
let mut cmd = tokio::process::Command::new("curl");
|
||||
cmd.arg("-f").arg("-L").arg("-s").arg("-S").arg("-o").arg(&dest).arg(&url);
|
||||
if let Some(t) = &hf_token {
|
||||
cmd.arg("-H").arg(format!("Authorization: Bearer {t}"));
|
||||
}
|
||||
cmd.stdout(Stdio::null());
|
||||
cmd.stderr(Stdio::piped());
|
||||
let mut child = match cmd.spawn() {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
store.push_line(&id, format!("✗ failed {filename}: {e}"));
|
||||
cleanup_output(&output_abs);
|
||||
store.finish(&id, false);
|
||||
store.drop_cancel(&id);
|
||||
return;
|
||||
}
|
||||
};
|
||||
let t_err = child.stderr.take().map(|s| {
|
||||
let st = store.clone();
|
||||
let jid = id.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut lines = BufReader::new(s).lines();
|
||||
while let Ok(Some(l)) = lines.next_line().await {
|
||||
let l = l.trim().to_string();
|
||||
if !l.is_empty() {
|
||||
st.push_line(&jid, l);
|
||||
}
|
||||
}
|
||||
})
|
||||
});
|
||||
// Race the part against a cancel — a single part can be many GB, so
|
||||
// waiting for it to finish first would make Cancel look dead.
|
||||
let status = tokio::select! {
|
||||
r = child.wait() => {
|
||||
if let Some(t) = t_err { t.abort(); }
|
||||
r.map_err(|e| e.to_string())
|
||||
}
|
||||
_ = cancel.notified() => {
|
||||
let _ = child.start_kill();
|
||||
let _ = child.wait().await;
|
||||
if let Some(t) = t_err { t.abort(); }
|
||||
cleanup_output(&output_abs);
|
||||
store.push_line(&id, "✗ cancelled by user — partial output removed".into());
|
||||
store.set_state(&id, "cancelled");
|
||||
store.drop_cancel(&id);
|
||||
return;
|
||||
}
|
||||
};
|
||||
match status {
|
||||
Ok(s) if s.success() => {
|
||||
// exact byte count after each part (the poller covers the gaps)
|
||||
let on_disk = path_size(out_dir);
|
||||
if total_sz > 0 {
|
||||
store.set_bytes(&id, on_disk);
|
||||
} else {
|
||||
let frac = (idx + 1) as f32 / files.len() as f32;
|
||||
store.set_progress(&id, frac, "downloading".into());
|
||||
}
|
||||
}
|
||||
Ok(s) => {
|
||||
store.push_line(&id, format!("✗ failed {filename}: exit {:?}", s.code()));
|
||||
cleanup_output(&output_abs);
|
||||
store.finish(&id, false);
|
||||
store.drop_cancel(&id);
|
||||
return;
|
||||
}
|
||||
Err(e) => {
|
||||
store.push_line(&id, format!("✗ failed {filename}: {e}"));
|
||||
cleanup_output(&output_abs);
|
||||
store.finish(&id, false);
|
||||
store.drop_cancel(&id);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
store.push_line(&id, "✓ done".into());
|
||||
store.finish(&id, true);
|
||||
store.drop_cancel(&id);
|
||||
return;
|
||||
}
|
||||
|
||||
// single-file path (original logic)
|
||||
let (cmf_path, cmf_size) = if !cmf_files.is_empty() {
|
||||
// if variant specified and matches a file prefix, prefer it
|
||||
if let Some(v) = chosen_variant {
|
||||
cmf_files
|
||||
.iter()
|
||||
.find(|(p, _)| p.contains(v))
|
||||
.cloned()
|
||||
.unwrap_or_else(|| cmf_files[0].clone())
|
||||
} else {
|
||||
cmf_files[0].clone()
|
||||
}
|
||||
} else {
|
||||
let repo_name = repo.rsplit('/').next().unwrap_or(&repo);
|
||||
(format!("{repo_name}.cmf"), 0)
|
||||
};
|
||||
|
||||
let path_lower = cmf_path.to_ascii_lowercase();
|
||||
@@ -344,13 +747,33 @@ async fn download_cmf_repo(
|
||||
};
|
||||
store.update_quant(&id, detected_quant);
|
||||
|
||||
store.push_line(&id, format!("→ downloading {cmf_path} from HuggingFace..."));
|
||||
store.set_progress(&id, 0.05, "downloading".into());
|
||||
|
||||
let download_url = format!("https://huggingface.co/{repo}/resolve/main/{cmf_path}");
|
||||
// The tree listing is the usual source of the size; ask the CDN directly when
|
||||
// it wasn't (private repo, tree fetch failed) so the bar still gets a total.
|
||||
let total = if cmf_size > 0 {
|
||||
cmf_size
|
||||
} else {
|
||||
head_content_length(&download_url, hf_token.as_deref())
|
||||
.await
|
||||
.unwrap_or(0)
|
||||
};
|
||||
store.push_line(&id, {
|
||||
let sz = if total > 0 { format!(" ({})", format_bytes(total)) } else { String::new() };
|
||||
format!("→ downloading {cmf_path}{sz} from HuggingFace...")
|
||||
});
|
||||
store.set_total(&id, total);
|
||||
store.set_progress(&id, 0.001, "downloading".into());
|
||||
// Drop a previous copy first (re-download / update). There is no resume, so
|
||||
// it would be overwritten anyway — and leaving it there would make the size
|
||||
// poller read the old bytes and report 100% before we transfer any.
|
||||
cleanup_output(&output_abs);
|
||||
// Progress comes from bytes on disk only. curl's own meter is worse than
|
||||
// useless here: with -L it prints a complete 0→100% bar for the redirect hop
|
||||
// before the real transfer starts, which used to pin the job at 100%.
|
||||
let _poller = spawn_size_poller(store.clone(), id.clone(), output_abs.clone());
|
||||
|
||||
let mut cmd = tokio::process::Command::new("curl");
|
||||
cmd.arg("-f").arg("-L").arg("-#").arg("-o").arg(&output_abs);
|
||||
cmd.arg("-f").arg("-L").arg("-s").arg("-S").arg("-o").arg(&output_abs);
|
||||
|
||||
if let Some(t) = &hf_token {
|
||||
cmd.arg("-H").arg(format!("Authorization: Bearer {t}"));
|
||||
@@ -370,27 +793,19 @@ async fn download_cmf_repo(
|
||||
}
|
||||
};
|
||||
|
||||
// -sS leaves stderr for real errors only — drain it into the log so a failure
|
||||
// says what went wrong instead of just an exit code.
|
||||
let stderr = child.stderr.take();
|
||||
let mut t_err = None;
|
||||
if let Some(err_stream) = stderr {
|
||||
let store_clone = store.clone();
|
||||
let id_clone = id.clone();
|
||||
t_err = Some(tokio::spawn(async move {
|
||||
use tokio::io::AsyncReadExt;
|
||||
let mut reader = err_stream;
|
||||
let mut buf = [0u8; 256];
|
||||
while let Ok(n) = reader.read(&mut buf).await {
|
||||
if n == 0 {
|
||||
break;
|
||||
}
|
||||
let text = String::from_utf8_lossy(&buf[..n]);
|
||||
for word in text.split_whitespace() {
|
||||
if let Some(num_str) = word.strip_suffix('%') {
|
||||
if let Ok(pct) = num_str.parse::<f32>() {
|
||||
let frac = (pct / 100.0).clamp(0.0, 1.0);
|
||||
store_clone.set_progress(&id_clone, frac, "downloading".into());
|
||||
}
|
||||
}
|
||||
let mut lines = BufReader::new(err_stream).lines();
|
||||
while let Ok(Some(l)) = lines.next_line().await {
|
||||
let l = l.trim().to_string();
|
||||
if !l.is_empty() {
|
||||
store_clone.push_line(&id_clone, l);
|
||||
}
|
||||
}
|
||||
}));
|
||||
@@ -484,11 +899,20 @@ pub fn start_import(store: Arc<JobStore>, cfg: &CmfCfg, p: ImportParams) -> Resu
|
||||
cfg.models_dir
|
||||
)
|
||||
})?;
|
||||
let base = if p.name.trim().is_empty() {
|
||||
let mut base = if p.name.trim().is_empty() {
|
||||
sanitize(p.repo.rsplit('/').next().unwrap_or(&p.repo))
|
||||
} else {
|
||||
sanitize(&p.name)
|
||||
};
|
||||
// for sharded variants make output distinct (e.g. DeepSeek parts-q2tp vs q4tp); single-file CMF ignores variant
|
||||
if let Some(v) = p.variant.as_deref().map(|s| s.trim()).filter(|s| !s.is_empty()) {
|
||||
if v.starts_with("parts-") {
|
||||
let vs = sanitize(v);
|
||||
if !base.ends_with(&vs) {
|
||||
base = format!("{base}-{vs}");
|
||||
}
|
||||
}
|
||||
}
|
||||
let output = std::path::Path::new(&cfg.models_dir)
|
||||
.join(format!("{base}.cmf"))
|
||||
.to_string_lossy()
|
||||
@@ -497,7 +921,13 @@ pub fn start_import(store: Arc<JobStore>, cfg: &CmfCfg, p: ImportParams) -> Resu
|
||||
.map(|d| d.join(format!("{base}.cmf")).to_string_lossy().to_string())
|
||||
.unwrap_or_else(|_| output.clone());
|
||||
|
||||
let id = gen_id(&p.repo);
|
||||
let id = gen_id(&format!("{}::{}", p.repo, p.variant.as_deref().unwrap_or("")));
|
||||
// if same repo/variant is already running, don't reset progress — just return it
|
||||
if let Some(existing) = store.get(&id) {
|
||||
if existing.state == "running" {
|
||||
return Ok(id);
|
||||
}
|
||||
}
|
||||
let is_cmf = is_cmf_repo(&p.repo);
|
||||
let quant_display = if is_cmf {
|
||||
let r = p.repo.to_lowercase();
|
||||
@@ -524,6 +954,13 @@ pub fn start_import(store: Arc<JobStore>, cfg: &CmfCfg, p: ImportParams) -> Resu
|
||||
repo: p.repo.clone(),
|
||||
quant: quant_display.clone(),
|
||||
output: output_abs.clone(),
|
||||
name: base.clone(),
|
||||
variant: p
|
||||
.variant
|
||||
.as_deref()
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(str::to_string),
|
||||
state: "running".into(),
|
||||
log: if is_cmf {
|
||||
vec![format!(
|
||||
@@ -538,6 +975,8 @@ pub fn start_import(store: Arc<JobStore>, cfg: &CmfCfg, p: ImportParams) -> Resu
|
||||
size_bytes: None,
|
||||
progress: Some(0.0),
|
||||
phase: Some("starting".into()),
|
||||
total_bytes: None,
|
||||
done_bytes: None,
|
||||
});
|
||||
let cancel = Arc::new(Notify::new());
|
||||
store.set_cancel(&id, cancel.clone());
|
||||
@@ -550,8 +989,9 @@ pub fn start_import(store: Arc<JobStore>, cfg: &CmfCfg, p: ImportParams) -> Resu
|
||||
|
||||
if is_cmf && !use_python {
|
||||
let ret_id = id.clone();
|
||||
let variant = p.variant.clone();
|
||||
tokio::spawn(async move {
|
||||
download_cmf_repo(p.repo, output_abs, hf_token, store, id, cancel).await;
|
||||
download_cmf_repo(p.repo, output_abs, hf_token, store, id, cancel, variant).await;
|
||||
});
|
||||
return Ok(ret_id);
|
||||
}
|
||||
|
||||
+171
-18
@@ -38,42 +38,160 @@ use state::SharedState;
|
||||
about = "Universal LLM gateway with intelligent routing"
|
||||
)]
|
||||
struct Args {
|
||||
/// Path to the TOML config file.
|
||||
#[arg(long, default_value = "config/gateway.toml")]
|
||||
config: String,
|
||||
/// Path to the TOML config file. If omitted, uses `config/gateway.toml` when running
|
||||
/// inside the repo, otherwise `~/.cortiq-gateway/gateway.toml` (or `$CORTIQ_GATEWAY_HOME`).
|
||||
/// Missing file is created automatically — no manual copying needed.
|
||||
#[arg(long)]
|
||||
config: Option<String>,
|
||||
/// Admin token for the web panel (overrides `[admin].token_env`).
|
||||
#[arg(long)]
|
||||
admin_token: Option<String>,
|
||||
/// Do not automatically open the browser on first run.
|
||||
#[arg(long)]
|
||||
no_browser: bool,
|
||||
/// Do not auto-select a free port if the configured one is busy.
|
||||
#[arg(long)]
|
||||
no_port_search: bool,
|
||||
}
|
||||
|
||||
/// Resolve the admin token: env from `[admin].token_env` → `--admin-token` →
|
||||
/// generate a random one. Returns `(token, generated)`.
|
||||
fn resolve_admin_token(cfg: &Config, args: &Args) -> (String, bool) {
|
||||
///
|
||||
/// By default no token is required (open mode). User sets it explicitly via
|
||||
/// `--admin-token`, `[admin].token` or `[admin].token_env` if they want auth.
|
||||
fn resolve_admin_token(cfg: &Config, args: &Args) -> String {
|
||||
if let Some(env) = &cfg.admin.token_env {
|
||||
if let Ok(v) = std::env::var(env) {
|
||||
if !v.is_empty() {
|
||||
return (v, false);
|
||||
return v;
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Some(t) = &args.admin_token {
|
||||
if !t.is_empty() {
|
||||
return (t.clone(), false);
|
||||
return t.clone();
|
||||
}
|
||||
}
|
||||
if let Some(t) = &cfg.admin.token {
|
||||
if !t.is_empty() {
|
||||
return (t.clone(), false);
|
||||
return t.clone();
|
||||
}
|
||||
}
|
||||
(admin::random_token(24), true)
|
||||
String::new()
|
||||
}
|
||||
|
||||
fn is_docker() -> bool {
|
||||
std::path::Path::new("/.dockerenv").exists()
|
||||
|| std::fs::read_to_string("/proc/1/cgroup")
|
||||
.map(|s| s.contains("docker") || s.contains("kubepods"))
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
fn should_open_browser(args: &Args) -> bool {
|
||||
if args.no_browser {
|
||||
return false;
|
||||
}
|
||||
if std::env::var("CORTIQ_NO_BROWSER").map(|v| !v.is_empty() && v != "0").unwrap_or(false) {
|
||||
return false;
|
||||
}
|
||||
if std::env::var("CI").is_ok() || std::env::var("NO_BROWSER").is_ok() {
|
||||
return false;
|
||||
}
|
||||
if is_docker() {
|
||||
return false;
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
fn parse_listen(listen: &str) -> (String, u16) {
|
||||
if let Some((host, port_s)) = listen.rsplit_once(':') {
|
||||
if let Ok(p) = port_s.parse::<u16>() {
|
||||
return (host.to_string(), p);
|
||||
}
|
||||
}
|
||||
(listen.to_string(), 9000)
|
||||
}
|
||||
|
||||
fn is_port_free(host: &str, port: u16) -> bool {
|
||||
let addr = format!("{host}:{port}");
|
||||
// bind test + connect test (docker port forwarding can make bind succeed while port is still forwarded)
|
||||
if std::net::TcpListener::bind(&addr).is_err() {
|
||||
return false;
|
||||
}
|
||||
use std::net::ToSocketAddrs;
|
||||
let probe_host = if host == "0.0.0.0" { "127.0.0.1" } else { host };
|
||||
if let Ok(mut addrs) = (probe_host, port).to_socket_addrs() {
|
||||
if let Some(a) = addrs.next() {
|
||||
if std::net::TcpStream::connect_timeout(&a, std::time::Duration::from_millis(200)).is_ok() {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
fn resolve_listen(cfg: &mut Config, no_search: bool) -> String {
|
||||
if no_search {
|
||||
return cfg.listen.clone();
|
||||
}
|
||||
let (host, port) = parse_listen(&cfg.listen);
|
||||
let clean_host = if host.is_empty() { "0.0.0.0".to_string() } else { host };
|
||||
if is_port_free(&clean_host, port) {
|
||||
return cfg.listen.clone();
|
||||
}
|
||||
for p in (port + 1)..=port.saturating_add(50) {
|
||||
if is_port_free(&clean_host, p) {
|
||||
let new_listen = format!("{clean_host}:{p}");
|
||||
eprintln!("port {port} busy — switching to free port {p} ({new_listen})");
|
||||
cfg.listen = new_listen.clone();
|
||||
return new_listen;
|
||||
}
|
||||
}
|
||||
if !is_port_free(&clean_host, port) {
|
||||
eprintln!("warning: port {port} busy and no free port found in next 50, will try to bind anyway");
|
||||
}
|
||||
cfg.listen.clone()
|
||||
}
|
||||
|
||||
fn setup_marker_path(config_path: &str) -> std::path::PathBuf {
|
||||
// marker lives next to config file: gateway.toml -> .setup_done
|
||||
std::path::Path::new(config_path)
|
||||
.parent()
|
||||
.unwrap_or_else(|| std::path::Path::new("."))
|
||||
.join(".setup_done")
|
||||
}
|
||||
|
||||
fn needs_first_run_wizard(cfg: &Config, config_path: &str) -> bool {
|
||||
let marker = setup_marker_path(config_path);
|
||||
if marker.exists() {
|
||||
return false;
|
||||
}
|
||||
// also consider "needs setup" if model pool is empty and no managed cmf
|
||||
if !cfg.models.is_empty() || !cfg.cmf.effective_servers().is_empty() {
|
||||
// still show wizard once if marker missing, but not if user already has models
|
||||
// we treat missing marker + existing models as "already configured" to avoid nagging
|
||||
// persist marker lazily
|
||||
let _ = std::fs::write(&marker, "done");
|
||||
return false;
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
let args = Args::parse();
|
||||
let config_path = args
|
||||
.config
|
||||
.clone()
|
||||
.unwrap_or_else(config::default_config_path);
|
||||
|
||||
let cfg = Config::load(&args.config)?;
|
||||
let mut cfg = Config::load_or_create(&config_path)?;
|
||||
// auto-pick a free port if the configured one is busy (unless disabled)
|
||||
let original_listen = cfg.listen.clone();
|
||||
let listen = resolve_listen(&mut cfg, args.no_port_search);
|
||||
// persist the resolved port if it changed
|
||||
if listen != original_listen {
|
||||
let _ = cfg.save(&config_path);
|
||||
}
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
tracing_subscriber::EnvFilter::try_from_default_env()
|
||||
@@ -81,11 +199,11 @@ async fn main() -> anyhow::Result<()> {
|
||||
)
|
||||
.init();
|
||||
|
||||
let listen = cfg.listen.clone();
|
||||
let admin_enabled = cfg.admin.enabled;
|
||||
let (admin_token, generated) = resolve_admin_token(&cfg, &args);
|
||||
let admin_token = resolve_admin_token(&cfg, &args);
|
||||
let first_run = needs_first_run_wizard(&cfg, &config_path);
|
||||
|
||||
let state = SharedState::build(cfg, args.config.clone())?;
|
||||
let state = SharedState::build(cfg, config_path.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
|
||||
@@ -120,11 +238,46 @@ async fn main() -> anyhow::Result<()> {
|
||||
let listener = tokio::net::TcpListener::bind(&listen).await?;
|
||||
tracing::info!("cortiq-gateway listening on {listen}");
|
||||
if admin_enabled {
|
||||
tracing::info!("admin console: http://{listen}/admin");
|
||||
if generated {
|
||||
tracing::warn!(
|
||||
"admin token (auto-generated, set [admin].token_env to persist): {admin_token}"
|
||||
);
|
||||
let host_for_url = if listen.starts_with("0.0.0.0:") {
|
||||
listen.replacen("0.0.0.0", "127.0.0.1", 1)
|
||||
} else {
|
||||
listen.clone()
|
||||
};
|
||||
// On first run open the onboarding wizard, otherwise the normal console.
|
||||
let admin_path = if first_run { "/admin#/onboarding" } else { "/admin" };
|
||||
let url = if admin_token.is_empty() {
|
||||
format!("http://{host_for_url}{admin_path}")
|
||||
} else {
|
||||
format!("http://{host_for_url}{admin_path}?token={admin_token}")
|
||||
};
|
||||
if admin_token.is_empty() {
|
||||
tracing::info!("admin console (open, no token): http://{host_for_url}/admin");
|
||||
} else {
|
||||
tracing::info!("admin console: http://{host_for_url}/admin");
|
||||
}
|
||||
if first_run {
|
||||
tracing::info!("first run — opening onboarding wizard at {url}");
|
||||
}
|
||||
if admin_enabled && should_open_browser(&args) {
|
||||
let browser_url = url.clone();
|
||||
tokio::spawn(async move {
|
||||
// give server a moment to be ready
|
||||
tokio::time::sleep(std::time::Duration::from_millis(600)).await;
|
||||
#[cfg(feature = "open-browser")]
|
||||
{
|
||||
if let Err(e) = open::that(&browser_url) {
|
||||
tracing::warn!("failed to open browser: {e}");
|
||||
} else {
|
||||
tracing::info!("opened browser at {browser_url}");
|
||||
}
|
||||
}
|
||||
#[cfg(not(feature = "open-browser"))]
|
||||
{
|
||||
tracing::info!("open browser: {browser_url} (feature disabled)");
|
||||
}
|
||||
});
|
||||
} else if first_run {
|
||||
tracing::info!("open this URL to finish setup: {url}");
|
||||
}
|
||||
}
|
||||
axum::serve(listener, app).await?;
|
||||
|
||||
Reference in New Issue
Block a user