infosave2007
15de36be3b
Fifteen changes aimed at the first day of real use: - Import and media jobs persist to <data>/jobs-*.json and reload on start (running ones settle as errors); the Register button no longer vanishes because the gateway restarted. - Docker ships the cortiq binary (built in the builder stage) — local .cmf chat models and image/video generation now work in containers, where auto-install never could (no cargo in the runtime image). libvulkan1 added for optional GPU offload. - Open admin console on 0.0.0.0 without a token now warns in the log and shows a dismissible banner — the VPS footgun is at least labeled. - "Restart gateway" button in Settings: stops managed model servers, spawns a fresh copy (bind retried while the port frees), Docker relies on the container restart policy instead. - Dashboard "System" panel: CPU sparkline + load, RAM (with the gateway's own RSS), free disk at the models dir — sampled by sysinfo every 2 s. - Featured model cards warn "not enough disk" / "may not fit in RAM" before the download, from the same live snapshot. - Log viewer in the sidebar: a 500-line in-memory ring fed by a second ANSI-free tracing layer; filter, level, pause, copy. - Update check: crates.io polled twice a day, the console shows an "update available" notice. - Media generations queue up (with position shown) instead of being rejected; progress gained an honest ETA from measured throughput. - Generation gallery under the image/video tabs: lazy PPM thumbnails, click to reopen with seed/steps, delete. - Playground chat is a real dialogue now: history is resent, bubbles, Clear, Cmd/Ctrl+Enter. - "Connect your app" snippets (curl / Python / JS) on the keys page, with the real key inlined right after creation. - One-click benchmark on the Models page via `cortiq bench --json`. - Config export now downloads the actual gateway.toml (comments intact); import accepts TOML (validated, hot-reloaded) and the legacy JSON. - Streaming token counts are exact when the model server provides usage (cortiq-server patch, shipped separately); the gateway fallback now counts content chunks (1/token on cortiq) instead of chars/4. Wizard's Finish lands on the Playground. 37 new strings in all 7 languages.
1318 lines
48 KiB
Rust
1318 lines
48 KiB
Rust
//! CMF model factory: search HuggingFace, then convert a chosen repo to a
|
|
//! local quantized `.cmf` via the Python converter — as a tracked, streamed
|
|
//! background job so the admin UI can show live progress.
|
|
|
|
use crate::config::CmfCfg;
|
|
use serde::{Deserialize, Serialize};
|
|
use std::collections::HashMap;
|
|
use std::process::Stdio;
|
|
use std::sync::{Arc, Mutex};
|
|
use std::time::{SystemTime, UNIX_EPOCH};
|
|
use tokio::io::{AsyncBufReadExt, BufReader};
|
|
use tokio::sync::Notify;
|
|
|
|
/// Parse a `@PROGRESS <0..1> <phase>` marker emitted by the converter.
|
|
fn parse_progress(line: &str) -> Option<(f32, String)> {
|
|
let rest = line.strip_prefix("@PROGRESS ")?;
|
|
let mut it = rest.splitn(2, ' ');
|
|
let frac: f32 = it.next()?.trim().parse().ok()?;
|
|
let phase = it.next().unwrap_or("").trim().to_string();
|
|
Some((frac.clamp(0.0, 1.0), phase))
|
|
}
|
|
|
|
/// Remove a partial/aborted output — `.cmf` may be a file or a tensor-dir.
|
|
fn cleanup_output(path: &str) {
|
|
let p = std::path::Path::new(path);
|
|
if p.is_dir() {
|
|
let _ = std::fs::remove_dir_all(p);
|
|
} else if p.exists() {
|
|
let _ = std::fs::remove_file(p);
|
|
}
|
|
}
|
|
|
|
/// 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)
|
|
.unwrap_or_default()
|
|
.as_secs()
|
|
}
|
|
|
|
/// 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 mut h = std::collections::hash_map::DefaultHasher::new();
|
|
repo.hash(&mut h);
|
|
format!("{:016x}", h.finish())
|
|
}
|
|
|
|
/// Parameters chosen in the import wizard.
|
|
#[derive(Clone, Debug, Deserialize)]
|
|
pub struct ImportParams {
|
|
pub repo: String,
|
|
#[serde(default = "default_quant")]
|
|
pub quant: String,
|
|
#[serde(default)]
|
|
pub name: String, // output basename; empty → derived from repo
|
|
/// 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
|
|
#[serde(default)]
|
|
pub nphase: Option<u32>,
|
|
#[serde(default)]
|
|
pub vbit_shape: Option<String>, // log2 | cubic
|
|
#[serde(default)]
|
|
pub mean_bits: Option<f32>,
|
|
#[serde(default)]
|
|
pub shard_max_gb: Option<f32>,
|
|
#[serde(default)]
|
|
pub skip_mtp: bool,
|
|
/// O(1) Nyström attention hint (native converter, cortiq ≥ 0.2.0):
|
|
/// `all` | `deepN` | `i,j,k`. Weights pass through unchanged — the runtime
|
|
/// reads the hint at load. Empty/None/"off" = exact attention.
|
|
#[serde(default)]
|
|
pub o1: Option<String>,
|
|
/// Landmark budget for the o1 hint (validated default 32).
|
|
#[serde(default)]
|
|
pub o1_m: Option<usize>,
|
|
/// Exact-window width for the o1 hint (validated default 128).
|
|
#[serde(default)]
|
|
pub o1_window: Option<usize>,
|
|
/// Permanent exact sink keys for the o1 hint (validated default 4).
|
|
#[serde(default)]
|
|
pub o1_sink: Option<usize>,
|
|
}
|
|
fn default_quant() -> String {
|
|
"Q8_2F".into()
|
|
}
|
|
|
|
/// Live state of one conversion job.
|
|
#[derive(Clone, Debug, Serialize, Deserialize)]
|
|
pub struct Job {
|
|
pub id: String,
|
|
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,
|
|
pub finished: Option<u64>,
|
|
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)]
|
|
pub struct JobStore {
|
|
jobs: Mutex<HashMap<String, Job>>,
|
|
/// Cancel handles, keyed by job id (not serialized). Notifying aborts the
|
|
/// converter process and cleans up its partial output.
|
|
cancels: Mutex<HashMap<String, Arc<Notify>>>,
|
|
/// When set, every state change lands in this JSON file, so the job list —
|
|
/// and its Register buttons — survives a gateway restart.
|
|
persist: Mutex<Option<std::path::PathBuf>>,
|
|
last_save: Mutex<Option<std::time::Instant>>,
|
|
}
|
|
|
|
impl JobStore {
|
|
pub fn new() -> Arc<Self> {
|
|
Arc::new(Self::default())
|
|
}
|
|
/// Load previously saved jobs and start persisting to `path`. A job that
|
|
/// was `running` when the gateway died is settled as an error — its process
|
|
/// is gone; the file it produced (if complete) is still on disk.
|
|
pub fn attach_persistence(&self, path: std::path::PathBuf) {
|
|
if let Ok(text) = std::fs::read_to_string(&path) {
|
|
if let Ok(mut jobs) = serde_json::from_str::<Vec<Job>>(&text) {
|
|
let mut g = self.jobs.lock().unwrap();
|
|
for j in jobs.drain(..) {
|
|
let mut j = j;
|
|
if j.state == "running" {
|
|
j.state = "error".into();
|
|
j.log.push("✗ interrupted by a gateway restart".into());
|
|
j.finished = Some(now());
|
|
}
|
|
g.insert(j.id.clone(), j);
|
|
}
|
|
}
|
|
}
|
|
*self.persist.lock().unwrap() = Some(path);
|
|
}
|
|
fn save(&self, force: bool) {
|
|
let Some(path) = self.persist.lock().unwrap().clone() else {
|
|
return;
|
|
};
|
|
if !force {
|
|
let mut last = self.last_save.lock().unwrap();
|
|
if let Some(t) = *last {
|
|
if t.elapsed() < std::time::Duration::from_secs(2) {
|
|
return;
|
|
}
|
|
}
|
|
*last = Some(std::time::Instant::now());
|
|
}
|
|
let jobs: Vec<Job> = self.jobs.lock().unwrap().values().cloned().collect();
|
|
if let Ok(json) = serde_json::to_vec(&jobs) {
|
|
let tmp = path.with_extension("json.tmp");
|
|
if std::fs::write(&tmp, json).is_ok() {
|
|
let _ = std::fs::rename(&tmp, &path);
|
|
}
|
|
}
|
|
}
|
|
fn insert(&self, job: Job) {
|
|
self.jobs.lock().unwrap().insert(job.id.clone(), job);
|
|
self.save(true);
|
|
}
|
|
fn set_cancel(&self, id: &str, n: Arc<Notify>) {
|
|
self.cancels.lock().unwrap().insert(id.to_string(), n);
|
|
}
|
|
fn drop_cancel(&self, id: &str) {
|
|
self.cancels.lock().unwrap().remove(id);
|
|
}
|
|
/// Signal a running job to abort. Returns false if the job isn't cancellable.
|
|
pub fn cancel(&self, id: &str) -> bool {
|
|
let running = matches!(
|
|
self.jobs.lock().unwrap().get(id).map(|j| j.state.as_str()),
|
|
Some("running")
|
|
);
|
|
if !running {
|
|
return false;
|
|
}
|
|
if let Some(n) = self.cancels.lock().unwrap().get(id) {
|
|
n.notify_one();
|
|
true
|
|
} else {
|
|
false
|
|
}
|
|
}
|
|
fn push_line(&self, id: &str, line: String) {
|
|
if let Some(j) = self.jobs.lock().unwrap().get_mut(id) {
|
|
j.log.push(line);
|
|
let n = j.log.len();
|
|
if n > 200 {
|
|
j.log.drain(0..n - 200); // keep the tail
|
|
}
|
|
}
|
|
}
|
|
fn set_progress(&self, id: &str, frac: f32, phase: String) {
|
|
if let Some(j) = self.jobs.lock().unwrap().get_mut(id) {
|
|
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),
|
|
// but an unset bar takes whatever the first reading is
|
|
if j.progress.is_none() || f > cur + 0.001 {
|
|
j.progress = Some(f);
|
|
}
|
|
if !phase.is_empty() {
|
|
j.phase = Some(phase);
|
|
}
|
|
}
|
|
self.save(false);
|
|
}
|
|
/// 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();
|
|
if !j.log.is_empty() && j.log[0].starts_with("→ downloading ready .cmf model") {
|
|
j.log[0] = format!("→ downloading ready .cmf model {} ({})", j.repo, quant);
|
|
}
|
|
}
|
|
}
|
|
fn set_state(&self, id: &str, state: &str) {
|
|
if let Some(j) = self.jobs.lock().unwrap().get_mut(id) {
|
|
j.state = state.into();
|
|
j.finished = Some(now());
|
|
}
|
|
self.save(true);
|
|
}
|
|
fn finish(&self, id: &str, ok: bool) {
|
|
if let Some(j) = self.jobs.lock().unwrap().get_mut(id) {
|
|
if j.state == "cancelled" {
|
|
return; // a concurrent cancel already settled this job
|
|
}
|
|
j.finished = Some(now());
|
|
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));
|
|
}
|
|
}
|
|
self.save(true);
|
|
}
|
|
pub fn get(&self, id: &str) -> Option<Job> {
|
|
self.jobs.lock().unwrap().get(id).cloned()
|
|
}
|
|
/// Delete a finished job together with its converted output file(s).
|
|
/// Refuses a running job (cancel it first). `Ok(false)` if unknown.
|
|
pub fn delete(&self, id: &str) -> Result<bool, String> {
|
|
let job = self.jobs.lock().unwrap().get(id).cloned();
|
|
let Some(job) = job else { return Ok(false) };
|
|
if job.state == "running" {
|
|
return Err("cancel the running job before deleting".into());
|
|
}
|
|
cleanup_output(&job.output);
|
|
self.cancels.lock().unwrap().remove(id);
|
|
self.jobs.lock().unwrap().remove(id);
|
|
self.save(true);
|
|
Ok(true)
|
|
}
|
|
pub fn list(&self) -> Vec<Job> {
|
|
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
|
|
}
|
|
}
|
|
|
|
/// Proxy HuggingFace model search (avoids browser CORS + adds our token).
|
|
pub async fn hf_search(
|
|
query: &str,
|
|
limit: usize,
|
|
token: Option<&str>,
|
|
) -> Result<serde_json::Value, String> {
|
|
let q: String = url_escape(query);
|
|
let sort = if query.trim().is_empty() {
|
|
"trendingScore"
|
|
} else {
|
|
"downloads"
|
|
};
|
|
let url = format!(
|
|
"https://huggingface.co/api/models?search={q}&sort={sort}&direction=-1&limit={limit}&full=false"
|
|
);
|
|
let client = reqwest::Client::builder()
|
|
.timeout(std::time::Duration::from_secs(15))
|
|
.build()
|
|
.map_err(|e| e.to_string())?;
|
|
let mut req = client.get(&url).header("User-Agent", "cortiq-gateway");
|
|
if let Some(t) = token {
|
|
req = req.bearer_auth(t);
|
|
}
|
|
let resp = req.send().await.map_err(|e| e.to_string())?;
|
|
if !resp.status().is_success() {
|
|
return Err(format!("HuggingFace API {}", resp.status()));
|
|
}
|
|
resp.json().await.map_err(|e| e.to_string())
|
|
}
|
|
|
|
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 {
|
|
'a'..='z' | 'A'..='Z' | '0'..='9' | '-' | '_' | '.' | '/' => c.to_string(),
|
|
' ' => "+".to_string(),
|
|
_ => format!("%{:02X}", c as u32),
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
fn sanitize(s: &str) -> String {
|
|
s.chars()
|
|
.map(|c| {
|
|
if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
|
|
c
|
|
} else {
|
|
'-'
|
|
}
|
|
})
|
|
.collect()
|
|
}
|
|
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.
|
|
fn map_quant(q: &str) -> &'static str {
|
|
match q.to_ascii_uppercase().as_str() {
|
|
s if s.starts_with("Q4T") || s == "Q4_TILED" => "q4t",
|
|
s if s.starts_with("Q4") => "q4",
|
|
"Q8_2F" | "Q82F" => "q8_2f",
|
|
"F16" | "FP16" => "f16",
|
|
"VBIT" => "vbit",
|
|
"Q1" => "q1",
|
|
"Q1P" | "Q1_PTQ" => "q1p",
|
|
"Q1S" | "Q1_MASK" => "q1s",
|
|
"Q1T" | "Q1_TERNARY" => "q1t",
|
|
_ => "q8",
|
|
}
|
|
}
|
|
|
|
fn is_cmf_repo(repo: &str) -> bool {
|
|
let lower = repo.to_ascii_lowercase();
|
|
lower.contains("cmf") || lower.ends_with(".cmf")
|
|
}
|
|
|
|
async fn download_cmf_repo(
|
|
repo: String,
|
|
output_abs: String,
|
|
hf_token: Option<String>,
|
|
store: Arc<JobStore>,
|
|
id: String,
|
|
cancel: Arc<Notify>,
|
|
variant: Option<String>,
|
|
) {
|
|
// 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::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_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_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;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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();
|
|
let detected_quant =
|
|
if path_lower.contains("-q1t.") || repo.to_ascii_lowercase().contains("2bit") {
|
|
"Q1T".to_string()
|
|
} else if path_lower.contains("-q1.") || repo.to_ascii_lowercase().contains("cmf") {
|
|
"Q1".to_string()
|
|
} else if path_lower.contains("-q4.") {
|
|
"Q4".to_string()
|
|
} else if path_lower.contains("-q8.") {
|
|
"Q8".to_string()
|
|
} else {
|
|
"READY_CMF".to_string()
|
|
};
|
|
store.update_quant(&id, detected_quant);
|
|
|
|
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("-s")
|
|
.arg("-S")
|
|
.arg("-o")
|
|
.arg(&output_abs);
|
|
|
|
if let Some(t) = &hf_token {
|
|
cmd.arg("-H").arg(format!("Authorization: Bearer {t}"));
|
|
}
|
|
|
|
cmd.arg(&download_url);
|
|
cmd.stderr(std::process::Stdio::piped());
|
|
cmd.stdout(std::process::Stdio::null());
|
|
|
|
let mut child = match cmd.spawn() {
|
|
Ok(c) => c,
|
|
Err(e) => {
|
|
store.push_line(&id, format!("✗ failed to start curl download: {e}"));
|
|
store.finish(&id, false);
|
|
store.drop_cancel(&id);
|
|
return;
|
|
}
|
|
};
|
|
|
|
// -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 {
|
|
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);
|
|
}
|
|
}
|
|
}));
|
|
}
|
|
|
|
tokio::select! {
|
|
status_res = child.wait() => {
|
|
if let Some(t) = t_err { t.abort(); }
|
|
match status_res {
|
|
Ok(status) if status.success() => {
|
|
store.push_line(&id, "✓ done".into());
|
|
store.finish(&id, true);
|
|
}
|
|
Ok(status) => {
|
|
cleanup_output(&output_abs);
|
|
// curl 23 = write error: almost always a non-writable models dir
|
|
let hint = match status.code() {
|
|
Some(23) => " (write error — is the models dir writable?)",
|
|
Some(22) => " (HTTP error — repo/file not found or auth required)",
|
|
_ => "",
|
|
};
|
|
store.push_line(&id, format!("✗ download failed: exit code {:?}{hint}", status.code()));
|
|
store.finish(&id, false);
|
|
}
|
|
Err(e) => {
|
|
cleanup_output(&output_abs);
|
|
store.push_line(&id, format!("✗ download process error: {e}"));
|
|
store.finish(&id, false);
|
|
}
|
|
}
|
|
}
|
|
_ = 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);
|
|
}
|
|
|
|
/// into the job log. Fails fast if the converter script is missing.
|
|
pub fn start_import(store: Arc<JobStore>, cfg: &CmfCfg, p: ImportParams) -> Result<String, String> {
|
|
if p.repo.trim().is_empty() {
|
|
return Err("empty repo id".into());
|
|
}
|
|
// Native `cortiq convert` (from crates.io, no Python) handles standard + MoE
|
|
// models; the bundled Python converter is used only for advanced options it
|
|
// doesn't support yet (linear-attention folding, v-bit shaping).
|
|
let use_python = p.linear_core.is_some() || p.vbit_shape.is_some() || p.mean_bits.is_some();
|
|
if use_python && !std::path::Path::new(&cfg.converter).exists() {
|
|
return Err(format!(
|
|
"advanced options need the Python converter, not found: {}",
|
|
cfg.converter
|
|
));
|
|
}
|
|
// O(1) attention hint — native-converter feature (cortiq ≥ 0.2.0). Fail fast
|
|
// with a clear message instead of a mid-job "unexpected argument" error.
|
|
let o1 =
|
|
p.o1.as_deref()
|
|
.map(str::trim)
|
|
.filter(|s| !s.is_empty() && *s != "off")
|
|
.map(str::to_string);
|
|
if o1.is_some() {
|
|
if use_python {
|
|
return Err(
|
|
"O(1) attention is supported by the native converter only — remove the \
|
|
linear-core / v-bit shape / mean-bits options to use it"
|
|
.into(),
|
|
);
|
|
}
|
|
if p.repo.to_ascii_lowercase().contains("gguf") {
|
|
return Err("O(1) attention is not supported for GGUF imports yet".into());
|
|
}
|
|
let ver = crate::cmf_runtime::installed_version(&cfg.cortiq_bin);
|
|
if let Some(v) = &ver {
|
|
if crate::cmf_runtime::version_lt(v, "0.2.0") {
|
|
return Err(format!(
|
|
"O(1) attention needs cortiq ≥ 0.2.0 (installed: {v}) — update the runtime \
|
|
in Settings → Local models"
|
|
));
|
|
}
|
|
}
|
|
}
|
|
std::fs::create_dir_all(&cfg.models_dir).map_err(|e| {
|
|
format!(
|
|
"cannot create models dir {}: {e} — check that the gateway has write \
|
|
access (in Docker, mount a writable volume at the models path)",
|
|
cfg.models_dir
|
|
)
|
|
})?;
|
|
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()
|
|
.to_string();
|
|
let output_abs = std::fs::canonicalize(&cfg.models_dir)
|
|
.map(|d| d.join(format!("{base}.cmf")).to_string_lossy().to_string())
|
|
.unwrap_or_else(|_| output.clone());
|
|
|
|
let id = gen_id(&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();
|
|
if r.contains("2bit") || r.contains("q1t") {
|
|
"Q1T".to_string()
|
|
} else if r.contains("1.7bcmf")
|
|
|| r.contains("27bcmf")
|
|
|| r.contains("cmf")
|
|
|| p.quant.is_empty()
|
|
|| p.quant == "auto"
|
|
|| p.quant == "Q8_2F"
|
|
{
|
|
"Q1".to_string()
|
|
} else {
|
|
p.quant.clone()
|
|
}
|
|
} else if p.quant.is_empty() || p.quant == "auto" {
|
|
"q8".to_string()
|
|
} else {
|
|
p.quant.clone()
|
|
};
|
|
store.insert(Job {
|
|
id: id.clone(),
|
|
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!(
|
|
"→ downloading ready .cmf model {} ({})",
|
|
p.repo, quant_display
|
|
)]
|
|
} else {
|
|
vec![format!("→ converting {} to {} ({})", p.repo, base, p.quant)]
|
|
},
|
|
started: now(),
|
|
finished: None,
|
|
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());
|
|
|
|
let hf_token = if cfg.hf_token_env.is_empty() {
|
|
None
|
|
} else {
|
|
std::env::var(&cfg.hf_token_env).ok()
|
|
};
|
|
|
|
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, variant).await;
|
|
});
|
|
return Ok(ret_id);
|
|
}
|
|
|
|
// Program + args: `cortiq convert` by default, or the Python converter for
|
|
// advanced options. Both stream `@PROGRESS <frac>` markers into the log.
|
|
let (program, args, workdir): (String, Vec<String>, std::path::PathBuf) = if use_python {
|
|
let conv = std::path::Path::new(&cfg.converter);
|
|
let mut a = vec![
|
|
std::fs::canonicalize(conv)
|
|
.map(|p| p.to_string_lossy().to_string())
|
|
.unwrap_or_else(|_| cfg.converter.clone()),
|
|
"--model".into(),
|
|
p.repo.clone(),
|
|
"--quant".into(),
|
|
p.quant.clone(),
|
|
"--output".into(),
|
|
output_abs.clone(),
|
|
];
|
|
if let Some(lc) = &p.linear_core {
|
|
a.push("--linear-core".into());
|
|
a.push(lc.clone());
|
|
}
|
|
if let Some(n) = p.nphase {
|
|
a.push("--nphase".into());
|
|
a.push(n.to_string());
|
|
}
|
|
if let Some(vs) = &p.vbit_shape {
|
|
a.push("--vbit-shape".into());
|
|
a.push(vs.clone());
|
|
}
|
|
if let Some(mb) = p.mean_bits {
|
|
a.push("--mean-bits".into());
|
|
a.push(mb.to_string());
|
|
}
|
|
if let Some(g) = p.shard_max_gb {
|
|
a.push("--shard-max-gb".into());
|
|
a.push(g.to_string());
|
|
}
|
|
if p.skip_mtp {
|
|
a.push("--skip-mtp".into());
|
|
}
|
|
let wd = conv
|
|
.parent()
|
|
.map(|d| d.to_path_buf())
|
|
.unwrap_or_else(|| std::path::PathBuf::from("."));
|
|
(cfg.python_bin.clone(), a, wd)
|
|
} else if p.repo.to_ascii_lowercase().contains("gguf") {
|
|
// GGUF repo → native `cortiq import-gguf` (downloads + dequantizes any
|
|
// common ggml quant type: Q4_0/1, Q5_0/1, Q8_0, Q2_K..Q6_K, IQ4_NL/XS).
|
|
let mut a = vec![
|
|
"import-gguf".into(),
|
|
p.repo.clone(),
|
|
"--quant".into(),
|
|
map_quant(&p.quant).into(),
|
|
"--output".into(),
|
|
output_abs.clone(),
|
|
];
|
|
if let Some(t) = &hf_token {
|
|
a.push("--hf-token".into());
|
|
a.push(t.clone());
|
|
}
|
|
(
|
|
cfg.cortiq_bin.clone(),
|
|
a,
|
|
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
|
|
)
|
|
} else {
|
|
// safetensors → native `cortiq convert` (dense / MoE / GatedDeltaNet).
|
|
let mut a = vec![
|
|
"convert".into(),
|
|
"--model".into(),
|
|
p.repo.clone(),
|
|
"--quant".into(),
|
|
map_quant(&p.quant).into(),
|
|
"--output".into(),
|
|
output_abs.clone(),
|
|
];
|
|
if let Some(t) = &hf_token {
|
|
a.push("--hf-token".into());
|
|
a.push(t.clone());
|
|
}
|
|
// O(1) attention hint (weights unchanged; the runtime reads it at load)
|
|
if let Some(spec) = &o1 {
|
|
a.push("--o1".into());
|
|
a.push(spec.clone());
|
|
if let Some(m) = p.o1_m {
|
|
a.push("--o1-m".into());
|
|
a.push(m.to_string());
|
|
}
|
|
if let Some(w) = p.o1_window {
|
|
a.push("--o1-window".into());
|
|
a.push(w.to_string());
|
|
}
|
|
if let Some(s) = p.o1_sink {
|
|
a.push("--o1-sink".into());
|
|
a.push(s.to_string());
|
|
}
|
|
}
|
|
(
|
|
cfg.cortiq_bin.clone(),
|
|
a,
|
|
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
|
|
)
|
|
};
|
|
|
|
let ret_id = id.clone();
|
|
let output_for_task = output_abs.clone();
|
|
tokio::spawn(async move {
|
|
let mut cmd = tokio::process::Command::new(&program);
|
|
cmd.args(&args)
|
|
.current_dir(&workdir)
|
|
.env("PYTHONUNBUFFERED", "1")
|
|
.stdout(Stdio::piped())
|
|
.stderr(Stdio::piped());
|
|
// HF token for the Python converter (native cortiq gets --hf-token).
|
|
if use_python {
|
|
if let Some(tok) = hf_token {
|
|
cmd.env("HF_TOKEN", &tok).env("HUGGING_FACE_HUB_TOKEN", tok);
|
|
}
|
|
}
|
|
let mut child = match cmd.spawn() {
|
|
Ok(c) => c,
|
|
Err(e) => {
|
|
store.push_line(&id, format!("✗ spawn failed: {e}"));
|
|
store.finish(&id, false);
|
|
store.drop_cancel(&id);
|
|
return;
|
|
}
|
|
};
|
|
let stdout = child.stdout.take();
|
|
let stderr = child.stderr.take();
|
|
// stdout: parse @PROGRESS markers into progress/phase; everything else logs.
|
|
let s1 = store.clone();
|
|
let id1 = id.clone();
|
|
let t_out = tokio::spawn(async move {
|
|
if let Some(o) = stdout {
|
|
let mut lines = BufReader::new(o).lines();
|
|
while let Ok(Some(l)) = lines.next_line().await {
|
|
if let Some((frac, phase)) = parse_progress(&l) {
|
|
s1.set_progress(&id1, frac, phase);
|
|
} else {
|
|
s1.push_line(&id1, l);
|
|
}
|
|
}
|
|
}
|
|
});
|
|
let s2 = store.clone();
|
|
let id2 = id.clone();
|
|
let t_err = tokio::spawn(async move {
|
|
if let Some(e) = stderr {
|
|
let mut lines = BufReader::new(e).lines();
|
|
while let Ok(Some(l)) = lines.next_line().await {
|
|
s2.push_line(&id2, l);
|
|
}
|
|
}
|
|
});
|
|
|
|
tokio::select! {
|
|
status = child.wait() => {
|
|
let _ = tokio::join!(t_out, t_err);
|
|
let ok = status.map(|s| s.success()).unwrap_or(false);
|
|
if !ok {
|
|
cleanup_output(&output_for_task); // drop partial/corrupt output
|
|
}
|
|
store.push_line(
|
|
&id,
|
|
if ok { "✓ done".into() } else { "✗ converter exited with error".into() },
|
|
);
|
|
store.finish(&id, ok);
|
|
}
|
|
_ = cancel.notified() => {
|
|
let _ = child.start_kill();
|
|
let _ = child.wait().await;
|
|
t_out.abort();
|
|
t_err.abort();
|
|
cleanup_output(&output_for_task);
|
|
store.push_line(&id, "✗ cancelled by user — partial output removed".into());
|
|
store.set_state(&id, "cancelled");
|
|
}
|
|
}
|
|
store.drop_cancel(&id);
|
|
});
|
|
|
|
Ok(ret_id)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn test_map_quant_variants() {
|
|
assert_eq!(map_quant("Q8_2F"), "q8_2f");
|
|
assert_eq!(map_quant("q82f"), "q8_2f");
|
|
assert_eq!(map_quant("Q8_ROW"), "q8");
|
|
assert_eq!(map_quant("Q4_BLOCK"), "q4");
|
|
assert_eq!(map_quant("Q4_TILED"), "q4t");
|
|
assert_eq!(map_quant("q4t"), "q4t");
|
|
assert_eq!(map_quant("vbit"), "vbit");
|
|
assert_eq!(map_quant("q1"), "q1");
|
|
assert_eq!(map_quant("Q1P"), "q1p");
|
|
assert_eq!(map_quant("q1s"), "q1s");
|
|
assert_eq!(map_quant("Q1T"), "q1t");
|
|
assert_eq!(map_quant("F16"), "f16");
|
|
}
|
|
|
|
#[test]
|
|
fn test_is_cmf_repo_check() {
|
|
assert!(is_cmf_repo("infosave/Bonsai-8B_2bit_cmf"));
|
|
assert!(is_cmf_repo("infosave/Bonsai-1.7Bcmf"));
|
|
assert!(is_cmf_repo("user/model.cmf"));
|
|
assert!(!is_cmf_repo("Qwen/Qwen2.5-0.5B-Instruct"));
|
|
}
|
|
}
|