qualia_client_core/

model_lifecycle.rs

//! Catalog LLM download → GGUF shard map → WAL → lifecycle (in-process).

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::{Arc, OnceLock};

use qualia_core_db::{
    gguf_sharder::GGufSharder,
    llm_agent::{AgentBackend, LocalLlmAgent},
    orchestrator::{ModelLifecycle, NullThermalGovernor, TaskOrchestrator},
    q_hash,
    resource_catalog::{LLMResource, ResourceCatalog},
    wal::WriteAheadLog,
};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

#[derive(Debug)]
pub enum ModelError {
    NotFound(String),
    NoDownloadUrl(String),
    Download(String),
    Shard(String),
    Wal(String),
    Activate(String),
    Io(std::io::Error),
    Json(serde_json::Error),
}

impl std::fmt::Display for ModelError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ModelError::NotFound(id) => write!(f, "LLM not found in catalog: {id}"),
            ModelError::NoDownloadUrl(id) => write!(f, "No download URL for: {id}"),
            ModelError::Download(e) => write!(f, "Download failed: {e}"),
            ModelError::Shard(e) => write!(f, "GGUF shard map failed: {e}"),
            ModelError::Wal(e) => write!(f, "WAL write failed: {e}"),
            ModelError::Activate(e) => write!(f, "Model activation failed: {e}"),
            ModelError::Io(e) => write!(f, "IO error: {e}"),
            ModelError::Json(e) => write!(f, "JSON error: {e}"),
        }
    }
}

impl From<std::io::Error> for ModelError {
    fn from(e: std::io::Error) -> Self {
        ModelError::Io(e)
    }
}

impl From<serde_json::Error> for ModelError {
    fn from(e: serde_json::Error) -> Self {
        ModelError::Json(e)
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ActiveModelRecord {
    pub model_id: String,
    pub gguf_path: String,
    pub profile_id: u64,
    pub quantization: String,
    pub lifecycle_state: String,
    #[serde(default)]
    pub modality: String,
    #[serde(default)]
    pub architecture: Option<String>,
    #[serde(default)]
    pub mmproj_path: Option<String>,
    #[serde(default)]
    pub context_window: u32,
}

#[derive(Debug, Clone, Serialize)]
pub struct ModelInstallResult {
    pub model_id: String,
    pub gguf_path: String,
    pub profile_id: u64,
    pub pointer_quin_count: usize,
    pub vision_pointer_quin_count: usize,
    pub lifecycle_state: String,
    pub wal_path: String,
    pub modality: String,
    pub mmproj_path: Option<String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct ModelStatus {
    pub active: Option<ActiveModelRecord>,
    pub lifecycle_state: String,
    pub profile_id: Option<u64>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InstallManifest {
    pub model_id: String,
    pub gguf_path: String,
    pub profile_id: u64,
    pub quantization: String,
    pub pointer_quin_count: usize,
    pub vision_pointer_quin_count: usize,
    pub installed_at: u64,
    pub wal_path: String,
    pub modality: String,
    pub architecture: Option<String>,
    pub mmproj_path: Option<String>,
    pub context_window: u32,
}

fn orchestrator() -> Arc<TaskOrchestrator> {
    task_orchestrator()
}

/// Shared orchestrator for chat inference and model lifecycle transitions.
pub fn task_orchestrator() -> Arc<TaskOrchestrator> {
    static ORCH: OnceLock<Arc<TaskOrchestrator>> = OnceLock::new();
    ORCH.get_or_init(|| Arc::new(TaskOrchestrator::new(Box::new(NullThermalGovernor))))
        .clone()
}

/// UI RAM ceiling for LLM arena pressure (matches 0.0.12 Flutter plan).
pub const MEMORY_FLOOR_MB: u32 = 512;

static LLM_MEMORY_BYTES: AtomicU64 = AtomicU64::new(0);
static KV_CACHE_USED_MB: AtomicU32 = AtomicU32::new(0);
/// Last completed chat/decode throughput × 1000 (tok/s), or 0 if never measured.
static LAST_DECODE_TOK_S_MILLI: AtomicU32 = AtomicU32::new(0);
/// Unix-ish stamp (secs since process start not required — wall via duration store is enough).
static LAST_DECODE_TOK_S_AT_SECS: AtomicU64 = AtomicU64::new(0);

pub fn record_llm_memory_bytes(bytes: u64) {
    LLM_MEMORY_BYTES.store(bytes, Ordering::Relaxed);
}

pub fn record_llm_memory_sample(bytes: u64) {
    if bytes == 0 {
        return;
    }
    let mut current = LLM_MEMORY_BYTES.load(Ordering::Relaxed);
    while bytes > current {
        match LLM_MEMORY_BYTES.compare_exchange(
            current,
            bytes,
            Ordering::Relaxed,
            Ordering::Relaxed,
        ) {
            Ok(_) => return,
            Err(observed) => current = observed,
        }
    }
}

pub fn get_llm_memory_bytes() -> u64 {
    LLM_MEMORY_BYTES.load(Ordering::Relaxed)
}

pub fn record_kv_cache_used_mb(megabytes: u32) {
    KV_CACHE_USED_MB.store(megabytes, Ordering::Relaxed);
}

pub fn get_kv_cache_used_mb() -> u32 {
    KV_CACHE_USED_MB.load(Ordering::Relaxed)
}

/// Record measured decode/chat throughput from a completed generation turn.
///
/// Call only with real token counts and wall time — never synthetic marketing numbers.
pub fn record_last_decode_tok_s(tokens: u32, duration_ms: u64) {
    if tokens == 0 || duration_ms == 0 {
        return;
    }
    let tok_s = (tokens as f64) / (duration_ms as f64 / 1000.0);
    if !tok_s.is_finite() || tok_s <= 0.0 {
        return;
    }
    let milli = (tok_s * 1000.0).round().clamp(0.0, u32::MAX as f64) as u32;
    LAST_DECODE_TOK_S_MILLI.store(milli, Ordering::Relaxed);
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    LAST_DECODE_TOK_S_AT_SECS.store(now, Ordering::Relaxed);
}

/// Last measured tok/s from a completed turn, or `None` if never measured this process.
pub fn get_last_decode_tok_s() -> Option<f64> {
    let milli = LAST_DECODE_TOK_S_MILLI.load(Ordering::Relaxed);
    if milli == 0 {
        None
    } else {
        Some(milli as f64 / 1000.0)
    }
}

pub fn get_last_decode_tok_s_at_unix() -> u64 {
    LAST_DECODE_TOK_S_AT_SECS.load(Ordering::Relaxed)
}

pub fn get_thermal_state_label() -> &'static str {
    orchestrator().thermal_state_label()
}

pub fn models_dir(storage_root: &Path) -> PathBuf {
    storage_root.join("Models")
}

pub fn lifecycle_label(state: ModelLifecycle) -> &'static str {
    match state {
        ModelLifecycle::Discovered => "Discovered",
        ModelLifecycle::MappedToDisk => "MappedToDisk",
        ModelLifecycle::StreamingVRAM => "StreamingVRAM",
        ModelLifecycle::Active => "Active",
        ModelLifecycle::Scrubbing => "Scrubbing",
    }
}

fn probe_and_activate_model(
    agent: &LocalLlmAgent,
    profile_id: u64,
    model_label: &str,
    gguf_path: &str,
) -> Result<(), ModelError> {
    let orch = orchestrator();
    if let Some(resident) = orch.resident_model_id() {
        if resident != profile_id {
            log::info!(
                "LLM_LOAD|unload-start|0.01|Evicting resident model 0x{resident:016x} before loading {}",
                model_label
            );
            orch.evict_model(resident);
            let wait_started = std::time::Instant::now();
            while orch.scrubbing_lock.load(Ordering::Acquire) {
                if wait_started.elapsed() > std::time::Duration::from_secs(5) {
                    log::error!("LLM_LOAD|failed|1.00|Timed out waiting for prior model eviction");
                    return Err(ModelError::Activate(
                        "Timed out waiting for prior model eviction".to_string(),
                    ));
                }
                std::thread::sleep(std::time::Duration::from_millis(5));
            }
            record_llm_memory_bytes(0);
            record_kv_cache_used_mb(0);
            log::info!("LLM_LOAD|unload-done|0.03|Previous resident model scrubbed from memory");
        }
    }
    let mut sys = sysinfo::System::new_all();
    sys.refresh_memory();
    let ram_total_gib = sys.total_memory() as f64 / 1024.0 / 1024.0 / 1024.0;
    let ram_used_gib = sys.used_memory() as f64 / 1024.0 / 1024.0 / 1024.0;
    let ram_free_gib = (ram_total_gib - ram_used_gib).max(0.0);
    {
        let mut state = orch.current_model_state.lock().unwrap();
        *state = ModelLifecycle::MappedToDisk;
        *state = ModelLifecycle::StreamingVRAM;
    }
    log::info!("LLM_LOAD|prepare|0.02|Preparing model {}", model_label);
    log::info!(
        "LLM_LOAD|ram-check|0.04|System RAM {:.1}/{:.1} GiB used; {:.1} GiB available",
        ram_used_gib,
        ram_total_gib,
        ram_free_gib
    );
    match qualia_core_db::resident_model::mount_resident_gguf(
        profile_id,
        gguf_path,
        orch.mlock_enabled
            .load(std::sync::atomic::Ordering::Relaxed),
    ) {
        Ok(report) => {
            let kv_cache_mb = (report.kv_cache_bytes / (1024 * 1024)).min(u32::MAX as u64) as u32;
            record_llm_memory_bytes(report.mapped_bytes);
            record_kv_cache_used_mb(kv_cache_mb);
            if let Err(err) = orch.load_model(agent, profile_id) {
                log::error!("LLM_LOAD|failed|1.00|Activation failed: {}", err);
                return Err(ModelError::Activate(err.to_string()));
            }
            orch.register_resident_model(profile_id, report.mapped_bytes + report.kv_cache_bytes);
            log::info!(
                "LLM_LOAD|placement|0.96|Model mapped in system RAM ({:.2} GiB) and KV cache reserved in VRAM/system memory ({} MiB)",
                report.mapped_bytes as f64 / (1024.0 * 1024.0 * 1024.0),
                kv_cache_mb
            );
            log::info!(
                "LLM_LOAD|active|1.00|Model ready (mapped {:.2} GiB, kv {} MiB, backend {})",
                report.mapped_bytes as f64 / (1024.0 * 1024.0 * 1024.0),
                kv_cache_mb,
                if report.directml_enabled {
                    "DirectML+wgpu"
                } else {
                    "wgpu"
                }
            );
        }
        Err(err) => {
            record_llm_memory_bytes(0);
            record_kv_cache_used_mb(0);
            let mut state = orch.current_model_state.lock().unwrap();
            *state = ModelLifecycle::MappedToDisk;
            log::error!("LLM_LOAD|failed|1.00|Activation failed: {}", err);
            return Err(ModelError::Activate(err));
        }
    }
    Ok(())
}

pub fn unload_active_model(profile_id: Option<u64>) {
    let orch = orchestrator();
    let resident = profile_id.or_else(|| orch.resident_model_id());
    if let Some(model_id) = resident {
        log::info!("LLM_LOAD|unload-start|0.00|Unloading resident model 0x{model_id:016x}");
        orch.evict_model(model_id);
        let wait_started = std::time::Instant::now();
        while orch.scrubbing_lock.load(Ordering::Acquire) {
            if wait_started.elapsed() > std::time::Duration::from_secs(5) {
                log::warn!("LLM_LOAD|failed|1.00|Timed out waiting for model scrub");
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(5));
        }
        log::info!("LLM_LOAD|unload-done|1.00|Model memory scrub complete");
    }
    record_llm_memory_bytes(0);
    record_kv_cache_used_mb(0);
}

fn unix_now() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

fn llm_filename(model: &LLMResource) -> String {
    model
        .download
        .local_filename()
        .unwrap_or_else(|| format!("{}.gguf", model.id))
}

fn install_manifest_path(models_dir: &Path, model_id: &str) -> PathBuf {
    models_dir.join(format!("{model_id}.install.json"))
}

fn projector_filename(model: &LLMResource) -> Option<String> {
    model.vision_projector.as_ref().and_then(|d| {
        d.local_filename().or_else(|| {
            d.resolved_url()
                .and_then(|u| u.rsplit('/').next().map(|s| s.to_string()))
        })
    })
}

fn write_install_manifest(
    models_dir: &Path,
    model: &LLMResource,
    gguf_path: &Path,
    mmproj_path: Option<&Path>,
    profile_id: u64,
    pointer_quin_count: usize,
    vision_pointer_quin_count: usize,
    wal_path: &Path,
) -> Result<(), ModelError> {
    let modality = if model.is_multimodal() {
        "multimodal".to_string()
    } else {
        "text".to_string()
    };
    let manifest = InstallManifest {
        model_id: model.id.clone(),
        gguf_path: gguf_path.to_string_lossy().into_owned(),
        profile_id,
        quantization: model
            .quantization
            .clone()
            .unwrap_or_else(|| "Q4_K_M".to_string()),
        pointer_quin_count,
        vision_pointer_quin_count,
        installed_at: unix_now(),
        wal_path: wal_path.to_string_lossy().into_owned(),
        modality: modality.clone(),
        architecture: model.architecture.clone(),
        mmproj_path: mmproj_path.map(|p| p.to_string_lossy().into_owned()),
        context_window: model.effective_context_window(),
    };
    let json = serde_json::to_string_pretty(&manifest)?;
    std::fs::write(install_manifest_path(models_dir, &model.id), json)?;
    Ok(())
}

pub fn finalize_llm_install(
    model: &LLMResource,
    gguf_path: &Path,
    mmproj_path: Option<&Path>,
    storage_root: &Path,
) -> Result<ModelInstallResult, ModelError> {
    let models = models_dir(storage_root);
    std::fs::create_dir_all(&models)?;

    if !gguf_path.is_file() {
        return Err(ModelError::Io(std::io::Error::new(
            std::io::ErrorKind::NotFound,
            format!("GGUF not found: {}", gguf_path.display()),
        )));
    }

    if model.is_multimodal() && mmproj_path.is_none() {
        return Err(ModelError::Shard(
            "Multimodal model requires vision projector (mmproj) GGUF".to_string(),
        ));
    }

    let path_str = gguf_path.to_string_lossy().into_owned();
    let sharder = GGufSharder::new(path_str.clone());
    let mut pointer_quins = sharder.generate_bidx_pointer_map();
    let mut vision_pointer_quin_count = 0usize;

    if let Some(mmproj) = mmproj_path {
        if !mmproj.is_file() {
            return Err(ModelError::Io(std::io::Error::new(
                std::io::ErrorKind::NotFound,
                format!("mmproj not found: {}", mmproj.display()),
            )));
        }
        let mmproj_str = mmproj.to_string_lossy().into_owned();
        let vision_sharder = GGufSharder::new(mmproj_str);
        let vision_quins = vision_sharder.generate_bidx_pointer_map();
        vision_pointer_quin_count = vision_quins.len();
        pointer_quins.extend(vision_quins);
    }

    let wal_path = models.join("models.wal");
    let mut wal = WriteAheadLog::open(&wal_path)
        .map_err(|e| ModelError::Wal(format!("Cannot open {}: {}", wal_path.display(), e)))?;

    let timestamp = unix_now();
    let prov = model.provenance_quin(timestamp, &path_str);
    wal.append_mutation(&prov)
        .map_err(|e| ModelError::Wal(e.to_string()))?;

    if let Some(src_quin) = model.source_url_quin() {
        wal.append_mutation(&src_quin)
            .map_err(|e| ModelError::Wal(e.to_string()))?;
    }

    for q in &pointer_quins {
        wal.append_mutation(q)
            .map_err(|e| ModelError::Wal(e.to_string()))?;
    }

    for q in &model.to_quins() {
        wal.append_mutation(q)
            .map_err(|e| ModelError::Wal(e.to_string()))?;
    }

    let mmproj_str = mmproj_path.map(|p| p.to_string_lossy().into_owned());
    let profile = model.to_capability_profile_with_projector(&path_str, mmproj_str.as_deref());
    write_install_manifest(
        &models,
        model,
        gguf_path,
        mmproj_path,
        profile.profile_id,
        pointer_quins
            .len()
            .saturating_sub(vision_pointer_quin_count),
        vision_pointer_quin_count,
        &wal_path,
    )?;

    {
        let orch = orchestrator();
        let mut state = orch.current_model_state.lock().unwrap();
        *state = ModelLifecycle::MappedToDisk;
    }

    let modality = if model.is_multimodal() {
        "multimodal".to_string()
    } else {
        "text".to_string()
    };

    Ok(ModelInstallResult {
        model_id: model.id.clone(),
        gguf_path: path_str,
        profile_id: profile.profile_id,
        pointer_quin_count: pointer_quins
            .len()
            .saturating_sub(vision_pointer_quin_count),
        vision_pointer_quin_count,
        lifecycle_state: lifecycle_label(ModelLifecycle::MappedToDisk).to_string(),
        wal_path: wal_path.to_string_lossy().into_owned(),
        modality,
        mmproj_path: mmproj_str,
    })
}

pub async fn install_catalog_llm(
    catalog: &ResourceCatalog,
    id: &str,
    storage_root: &Path,
) -> Result<ModelInstallResult, ModelError> {
    let model = catalog
        .find_llm(id)
        .ok_or_else(|| ModelError::NotFound(id.to_string()))?;

    let url = model
        .download
        .resolved_url()
        .ok_or_else(|| ModelError::NoDownloadUrl(id.to_string()))?;

    let models = models_dir(storage_root);
    std::fs::create_dir_all(&models)?;

    let filename = llm_filename(model);
    let local_path = models.join(&filename);

    crate::resource_import::stream_download(&url, &local_path)
        .await
        .map_err(ModelError::Download)?;

    let mmproj_path = if let (Some(ref vp), Some(vp_url)) = (
        model.vision_projector.as_ref(),
        model
            .vision_projector
            .as_ref()
            .and_then(|d| d.resolved_url()),
    ) {
        let vp_name = projector_filename(model).unwrap_or_else(|| "mmproj.gguf".to_string());
        let vp_path = models.join(&vp_name);
        crate::resource_import::stream_download(&vp_url, &vp_path)
            .await
            .map_err(ModelError::Download)?;
        let _ = vp;
        Some(vp_path)
    } else {
        None
    };

    finalize_llm_install(model, &local_path, mmproj_path.as_deref(), storage_root)
}

pub fn load_install_manifest(storage_root: &Path, model_id: &str) -> Option<InstallManifest> {
    let path = install_manifest_path(&models_dir(storage_root), model_id);
    let text = std::fs::read_to_string(path).ok()?;
    serde_json::from_str(&text).ok()
}

fn sanitize_local_model_id(stem: &str) -> String {
    let cleaned: String = stem
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect();
    if cleaned.is_empty() {
        "local-model".to_string()
    } else {
        cleaned
    }
}

/// Discovered weight container in a vault directory (CLI / lifecycle tooling).
///
/// Name retained for API stability; entries may be `.gguf` **or** preferred `.p64`.
#[derive(Debug, Clone, Serialize)]
pub struct VaultGgufEntry {
    pub name: String,
    pub path: String,
    pub profile_id: u64,
    pub size_bytes: u64,
    /// `p64` | `gguf` — engine prefers `p64` when both exist for the same stem.
    pub container: String,
}

/// Exact duplicate GGUF files, grouped by content rather than filename.
///
/// The audit is deliberately read-only: lifecycle code may report redundant files,
/// but never copies, moves, or deletes a model behind the operator's back.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct VaultDuplicateGroup {
    pub sha256: String,
    pub size_bytes: u64,
    pub canonical_path: String,
    pub duplicate_paths: Vec<String>,
}

impl VaultDuplicateGroup {
    pub fn reclaimable_bytes(&self) -> u64 {
        self.size_bytes
            .saturating_mul(self.duplicate_paths.len() as u64)
    }
}

/// Recursively scan `vault_dir` for `.p64` and `.gguf` weight containers.
///
/// When both exist for the same stem (e.g. `foo.p64` + `foo.gguf`), **only the
/// `.p64` is listed** — GGUF is import-only once converted. `.f16.p64` keeps a
/// distinct stem and is listed separately.
pub fn scan_vault_gguf(vault_dir: &Path) -> Result<Vec<VaultGgufEntry>, std::io::Error> {
    if !vault_dir.is_dir() {
        return Err(std::io::Error::new(
            std::io::ErrorKind::NotFound,
            format!("Vault directory not found: {}", vault_dir.display()),
        ));
    }
    let mut raw = Vec::new();
    collect_vault_models(vault_dir, &mut raw)?;
    // Prefer p64: drop gguf when a p64 with the same base stem exists.
    let p64_stems: std::collections::HashSet<String> = raw
        .iter()
        .filter(|e| e.container == "p64")
        .map(|e| vault_stem_key(&e.path))
        .collect();
    let mut out: Vec<VaultGgufEntry> = raw
        .into_iter()
        .filter(|e| e.container == "p64" || !p64_stems.contains(&vault_stem_key(&e.path)))
        .collect();
    out.sort_by(|a, b| a.name.cmp(&b.name));
    Ok(out)
}

/// Stem key for prefer-p64 dedup: `foo.gguf` / `foo.p64` → `foo`; `foo.f16.p64` → `foo.f16`.
fn vault_stem_key(path: &str) -> String {
    Path::new(path)
        .file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or("")
        .to_string()
}

/// Hash only same-sized GGUF candidates and report byte-identical duplicates.
///
/// Size bucketing avoids reading every multi-gigabyte model during a routine audit.
/// Within each exact-content group the shortest, then lexicographically first, path
/// is suggested as the canonical keeper. No filesystem mutation is performed.
pub fn audit_vault_duplicates(
    vault_dir: &Path,
) -> Result<Vec<VaultDuplicateGroup>, std::io::Error> {
    use std::collections::BTreeMap;
    use std::io::Read;

    let entries = scan_vault_gguf(vault_dir)?;
    let mut by_size: BTreeMap<u64, Vec<&VaultGgufEntry>> = BTreeMap::new();
    for entry in &entries {
        by_size.entry(entry.size_bytes).or_default().push(entry);
    }

    let mut groups = Vec::new();
    for (size_bytes, candidates) in by_size {
        if candidates.len() < 2 {
            continue;
        }
        let mut by_digest: BTreeMap<String, Vec<String>> = BTreeMap::new();
        for candidate in candidates {
            let mut file = std::fs::File::open(&candidate.path)?;
            let mut hasher = Sha256::new();
            // Keep the cold-path hash buffer modest enough for Windows' default
            // main-thread stack while still reading in efficient chunks.
            let mut buffer = [0u8; 64 * 1024];
            loop {
                let read = file.read(&mut buffer)?;
                if read == 0 {
                    break;
                }
                hasher.update(&buffer[..read]);
            }
            let digest = hasher.finalize();
            let mut digest_hex = String::with_capacity(digest.len() * 2);
            for byte in digest {
                use std::fmt::Write;
                let _ = write!(digest_hex, "{byte:02x}");
            }
            by_digest
                .entry(digest_hex)
                .or_default()
                .push(candidate.path.clone());
        }

        for (sha256, mut paths) in by_digest {
            if paths.len() < 2 {
                continue;
            }
            paths.sort_by(|a, b| {
                let a_depth = Path::new(a).components().count();
                let b_depth = Path::new(b).components().count();
                a_depth.cmp(&b_depth).then_with(|| a.cmp(b))
            });
            let canonical_path = paths.remove(0);
            groups.push(VaultDuplicateGroup {
                sha256,
                size_bytes,
                canonical_path,
                duplicate_paths: paths,
            });
        }
    }
    groups.sort_by(|a, b| a.canonical_path.cmp(&b.canonical_path));
    Ok(groups)
}

fn collect_vault_models(dir: &Path, out: &mut Vec<VaultGgufEntry>) -> Result<(), std::io::Error> {
    for entry in std::fs::read_dir(dir)? {
        let entry = entry?;
        let path = entry.path();
        if path.is_dir() {
            collect_vault_models(&path, out)?;
            continue;
        }
        let ext = path
            .extension()
            .and_then(|e| e.to_str())
            .unwrap_or("")
            .to_ascii_lowercase();
        let container = match ext.as_str() {
            "p64" => "p64",
            "gguf" => "gguf",
            _ => continue,
        };
        let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or("model");
        let model_id = sanitize_local_model_id(stem);
        let size_bytes = entry.metadata()?.len();
        out.push(VaultGgufEntry {
            name: path
                .file_name()
                .map(|n| n.to_string_lossy().into_owned())
                .unwrap_or_else(|| model_id.clone()),
            path: path.to_string_lossy().into_owned(),
            profile_id: q_hash(&format!("profile:local:{model_id}")),
            size_bytes,
            container: container.into(),
        });
    }
    Ok(())
}

/// Resolve a model reference (filename, stem, or path) inside `vault_dir`.
///
/// Preference order: exact path → `.p64` → `.gguf` → vault scan (p64 preferred).
pub fn resolve_vault_model(vault_dir: &Path, model_ref: &str) -> Result<PathBuf, ModelError> {
    let direct = PathBuf::from(model_ref);
    if direct.is_file() {
        // If operator points at a GGUF that has a converted sibling .p64, prefer p64.
        if let Some(p64) = prefer_p64_sibling(&direct) {
            return Ok(p64);
        }
        return Ok(direct);
    }
    let in_vault = vault_dir.join(model_ref);
    if in_vault.is_file() {
        if let Some(p64) = prefer_p64_sibling(&in_vault) {
            return Ok(p64);
        }
        return Ok(in_vault);
    }
    // Prefer p64 extension first (designed runtime format).
    if !model_ref.ends_with(".p64") && !model_ref.ends_with(".gguf") {
        let with_p64 = vault_dir.join(format!("{model_ref}.p64"));
        if with_p64.is_file() {
            return Ok(with_p64);
        }
        let with_gguf = vault_dir.join(format!("{model_ref}.gguf"));
        if with_gguf.is_file() {
            if let Some(p64) = prefer_p64_sibling(&with_gguf) {
                return Ok(p64);
            }
            return Ok(with_gguf);
        }
    }
    for entry in scan_vault_gguf(vault_dir).map_err(ModelError::Io)? {
        let stem = PathBuf::from(&entry.path)
            .file_stem()
            .and_then(|s| s.to_str())
            .unwrap_or("")
            .to_string();
        if entry.name == model_ref
            || stem == model_ref
            || format!("0x{:016x}", entry.profile_id) == model_ref.to_lowercase()
        {
            return Ok(PathBuf::from(entry.path));
        }
    }
    Err(ModelError::NotFound(format!(
        "No p64/GGUF matching `{model_ref}` under {}",
        vault_dir.display()
    )))
}

/// If `path` is a `.gguf` and `{stem}.p64` exists beside it, return the p64.
fn prefer_p64_sibling(path: &Path) -> Option<PathBuf> {
    let is_gguf = path
        .extension()
        .and_then(|e| e.to_str())
        .map(|e| e.eq_ignore_ascii_case("gguf"))
        .unwrap_or(false);
    if !is_gguf {
        return None;
    }
    let p64 = path.with_extension("p64");
    if p64.is_file() {
        Some(p64)
    } else {
        None
    }
}

/// Map and activate a vault weight file (`.p64` preferred, or `.gguf`) without catalog install.
pub async fn activate_vault_gguf(gguf_path: &Path) -> Result<ActiveModelRecord, ModelError> {
    let path = prefer_p64_sibling(gguf_path).unwrap_or_else(|| gguf_path.to_path_buf());
    if !path.is_file() {
        return Err(ModelError::Io(std::io::Error::new(
            std::io::ErrorKind::NotFound,
            format!("Model not found: {}", path.display()),
        )));
    }
    let stem = path
        .file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or("local-model");
    let model_id = sanitize_local_model_id(stem);
    let path_str = path.to_string_lossy().into_owned();
    let profile_id = q_hash(&format!("profile:local:{model_id}"));
    let quant = if path
        .extension()
        .and_then(|e| e.to_str())
        .map(|e| e.eq_ignore_ascii_case("p64"))
        .unwrap_or(false)
    {
        "p64"
    } else {
        "vault"
    };

    let agent = LocalLlmAgent::with_local_backend(
        format!("did:qualia:cli-vault:{profile_id}"),
        AgentBackend::Local {
            model_path: path_str.clone(),
            context_window: 4096,
            quantization: quant.to_string(),
            vision_projector_path: None,
            modality: "text".to_string(),
            architecture: None,
        },
    );
    probe_and_activate_model(&agent, profile_id, &model_id, &path_str)?;

    let lifecycle = *orchestrator().current_model_state.lock().unwrap();
    Ok(ActiveModelRecord {
        model_id,
        gguf_path: path_str,
        profile_id,
        quantization: quant.to_string(),
        lifecycle_state: lifecycle_label(lifecycle).to_string(),
        modality: "text".to_string(),
        architecture: None,
        mmproj_path: None,
        context_window: 4096,
    })
}

/// Block until eviction scrub completes or `timeout` elapses.
pub fn wait_for_eviction_scrub(timeout: std::time::Duration) -> bool {
    let orch = orchestrator();
    let started = std::time::Instant::now();
    while orch.scrubbing_lock.load(Ordering::Acquire) {
        if started.elapsed() > timeout {
            return false;
        }
        std::thread::sleep(std::time::Duration::from_millis(5));
    }
    true
}

/// Register and activate a user-selected `.gguf` file at an arbitrary path.
///
/// Writes `{model_id}.install.json` under `{storage_root}/Models/` (path is not copied),
/// maps GGUF tensor pointers into the WAL, and transitions lifecycle → `Active`.
pub fn finalize_local_gguf(
    gguf_path: &Path,
    storage_root: &Path,
) -> Result<ActiveModelRecord, ModelError> {
    if !gguf_path.is_file() {
        return Err(ModelError::Io(std::io::Error::new(
            std::io::ErrorKind::NotFound,
            format!("GGUF not found: {}", gguf_path.display()),
        )));
    }

    let models = models_dir(storage_root);
    std::fs::create_dir_all(&models)?;

    let stem = gguf_path
        .file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or("local-model");
    let model_id = sanitize_local_model_id(stem);
    let path_str = gguf_path.to_string_lossy().into_owned();
    let profile_id = q_hash(&format!("profile:local:{model_id}"));

    let sharder = GGufSharder::new(path_str.clone());
    let pointer_quins = sharder.generate_bidx_pointer_map();
    let pointer_quin_count = pointer_quins.len();

    let wal_path = models.join("models.wal");
    let mut wal = WriteAheadLog::open(&wal_path)
        .map_err(|e| ModelError::Wal(format!("Cannot open {}: {}", wal_path.display(), e)))?;

    let subject = q_hash(&format!("local-gguf:{model_id}"));
    let predicate = q_hash("prov:wasDerivedFrom");
    let object = q_hash(&path_str);
    let context = q_hash("ctx:local-gguf");
    let prov = qualia_core_db::NQuin {
        subject,
        predicate,
        object,
        context,
        metadata: unix_now(),
        parity: subject ^ predicate ^ object ^ context,
    };
    wal.append_mutation(&prov)
        .map_err(|e| ModelError::Wal(e.to_string()))?;

    for quin in pointer_quins {
        wal.append_mutation(&quin)
            .map_err(|e| ModelError::Wal(e.to_string()))?;
    }

    let manifest = InstallManifest {
        model_id: model_id.clone(),
        gguf_path: path_str.clone(),
        profile_id,
        quantization: "local".to_string(),
        pointer_quin_count,
        vision_pointer_quin_count: 0,
        installed_at: unix_now(),
        wal_path: wal_path.to_string_lossy().into_owned(),
        modality: "text".to_string(),
        architecture: None,
        mmproj_path: None,
        context_window: 4096,
    };
    let json = serde_json::to_string_pretty(&manifest)?;
    std::fs::write(install_manifest_path(&models, &model_id), json)?;

    let agent = LocalLlmAgent::with_local_backend(
        format!("did:qualia:local-gguf:{profile_id}"),
        AgentBackend::Local {
            model_path: path_str.clone(),
            context_window: 4096,
            quantization: "local".to_string(),
            vision_projector_path: None,
            modality: "text".to_string(),
            architecture: None,
        },
    );
    probe_and_activate_model(&agent, profile_id, &model_id, &path_str)?;

    let lifecycle = *orchestrator().current_model_state.lock().unwrap();
    Ok(ActiveModelRecord {
        model_id,
        gguf_path: path_str,
        profile_id,
        quantization: "local".to_string(),
        lifecycle_state: lifecycle_label(lifecycle).to_string(),
        modality: "text".to_string(),
        architecture: None,
        mmproj_path: None,
        context_window: 4096,
    })
}

pub fn activate_model_for_id(
    model_id: &str,
    storage_root: &Path,
) -> Result<ActiveModelRecord, ModelError> {
    let manifest = load_install_manifest(storage_root, model_id).ok_or_else(|| {
        ModelError::Activate(format!(
            "No install manifest for `{model_id}` — download the model in LLM Hub first"
        ))
    })?;

    if !Path::new(&manifest.gguf_path).is_file() {
        return Err(ModelError::Activate(format!(
            "GGUF missing at {}",
            manifest.gguf_path
        )));
    }

    if manifest.modality == "multimodal" {
        let mmproj = manifest.mmproj_path.as_deref().ok_or_else(|| {
            ModelError::Activate("Multimodal manifest missing mmproj_path".to_string())
        })?;
        if !Path::new(mmproj).is_file() {
            return Err(ModelError::Activate(format!(
                "Vision projector missing at {mmproj}"
            )));
        }
    }

    let agent = LocalLlmAgent::with_local_backend(
        format!("did:qualia:profile:{}", manifest.profile_id),
        AgentBackend::Local {
            model_path: manifest.gguf_path.clone(),
            context_window: manifest.context_window,
            quantization: manifest.quantization.clone(),
            vision_projector_path: manifest.mmproj_path.clone(),
            modality: manifest.modality.clone(),
            architecture: manifest.architecture.clone(),
        },
    );
    probe_and_activate_model(
        &agent,
        manifest.profile_id,
        &manifest.model_id,
        &manifest.gguf_path,
    )?;

    let lifecycle = *orchestrator().current_model_state.lock().unwrap();
    let record = ActiveModelRecord {
        model_id: manifest.model_id,
        gguf_path: manifest.gguf_path,
        profile_id: manifest.profile_id,
        quantization: manifest.quantization,
        lifecycle_state: lifecycle_label(lifecycle).to_string(),
        modality: manifest.modality,
        architecture: manifest.architecture,
        mmproj_path: manifest.mmproj_path,
        context_window: manifest.context_window,
    };

    Ok(record)
}

pub fn activate_model(
    profile_id: u64,
    storage_root: &Path,
) -> Result<ActiveModelRecord, ModelError> {
    let models = models_dir(storage_root);
    let entries = std::fs::read_dir(&models).map_err(ModelError::Io)?;
    for entry in entries.filter_map(Result::ok) {
        let path = entry.path();
        if path.extension().and_then(|e| e.to_str()) != Some("json") {
            continue;
        }
        if !path
            .file_name()
            .map(|n| n.to_string_lossy().contains(".install.json"))
            .unwrap_or(false)
        {
            continue;
        }
        if let Ok(text) = std::fs::read_to_string(&path) {
            if let Ok(manifest) = serde_json::from_str::<InstallManifest>(&text) {
                if manifest.profile_id == profile_id {
                    return activate_model_for_id(&manifest.model_id, storage_root);
                }
            }
        }
    }
    Err(ModelError::Activate(format!(
        "No installed model with profile_id 0x{profile_id:016x}"
    )))
}

pub fn get_model_lifecycle_state() -> ModelLifecycle {
    *orchestrator().current_model_state.lock().unwrap()
}

pub fn get_model_status(active: Option<ActiveModelRecord>) -> ModelStatus {
    let lifecycle = get_model_lifecycle_state();
    ModelStatus {
        profile_id: active.as_ref().map(|r| r.profile_id),
        active,
        lifecycle_state: lifecycle_label(lifecycle).to_string(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn duplicate_audit_groups_only_identical_gguf_files() {
        let root = std::env::temp_dir().join(format!(
            "qualia-gguf-audit-{}-{}",
            std::process::id(),
            unix_now()
        ));
        let nested = root.join("nested");
        std::fs::create_dir_all(&nested).unwrap();
        std::fs::write(root.join("keeper.gguf"), b"same model bytes").unwrap();
        std::fs::write(nested.join("copy.gguf"), b"same model bytes").unwrap();
        std::fs::write(root.join("same-size-not-copy.gguf"), b"other model byte").unwrap();

        let groups = audit_vault_duplicates(&root).unwrap();
        assert_eq!(groups.len(), 1);
        assert_eq!(groups[0].size_bytes, 16);
        assert!(groups[0].canonical_path.ends_with("keeper.gguf"));
        assert_eq!(groups[0].duplicate_paths.len(), 1);
        assert!(groups[0].duplicate_paths[0].ends_with("copy.gguf"));
        assert_eq!(groups[0].reclaimable_bytes(), 16);

        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn vault_scan_prefers_p64_over_gguf_same_stem() {
        let root = std::env::temp_dir().join(format!(
            "qualia-vault-p64-{}-{}",
            std::process::id(),
            unix_now()
        ));
        std::fs::create_dir_all(&root).unwrap();
        std::fs::write(root.join("smollm.gguf"), b"gguf-bytes").unwrap();
        std::fs::write(root.join("smollm.p64"), b"p64\0bytes").unwrap();
        std::fs::write(root.join("other.gguf"), b"only-gguf").unwrap();

        let entries = scan_vault_gguf(&root).unwrap();
        let names: Vec<_> = entries.iter().map(|e| e.name.as_str()).collect();
        assert!(
            names.iter().any(|n| *n == "smollm.p64"),
            "p64 should be listed: {names:?}"
        );
        assert!(
            !names.iter().any(|n| *n == "smollm.gguf"),
            "gguf should be hidden when p64 exists: {names:?}"
        );
        assert!(names.iter().any(|n| *n == "other.gguf"));

        let resolved = resolve_vault_model(&root, "smollm").unwrap();
        assert!(
            resolved.extension().and_then(|e| e.to_str()) == Some("p64"),
            "resolve stem should pick p64: {}",
            resolved.display()
        );
        let via_gguf = resolve_vault_model(&root, "smollm.gguf").unwrap();
        assert!(
            via_gguf.extension().and_then(|e| e.to_str()) == Some("p64"),
            "resolve gguf name should still prefer sibling p64: {}",
            via_gguf.display()
        );

        std::fs::remove_dir_all(root).unwrap();
    }
}