cortiq-gateway 0.2.38

//! 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"));
    }
}