IdeA/crates/application/src/model_server.rs

//! Use cases for local model servers.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use std::time::Instant as StdInstant;

use domain::events::DomainEvent;
use domain::model_server::{
    LocalModelServerConfig, ModelServerLifecycleStatus, ModelServerReady, ModelServerStatus,
    ModelSource,
};
use domain::ports::{
    AgentContextStore, EventBus, FileSystem, ManagedProcess, ManagedProcessHandle,
    ModelArtifactCancel, ModelArtifactDownloader, ModelArtifactProgress, ModelArtifactState,
    ModelServerError, ModelServerProbe, ModelServerRegistry, ModelServerRuntime, ProcessStatus,
    ProfileStore, ProjectStore, RemotePath,
};
use domain::{LocalModelServerId, StopPolicy};
use tokio::sync::{Mutex as AsyncMutex, Notify};
use tokio::time::Instant;

use crate::error::AppError;
use crate::terminal::LiveAgentRegistry;

/// Artifact cache state exposed by model-server list use cases.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ModelArtifactView {
    /// The configured source is not managed by IdeA's downloader.
    NotManaged,
    /// The configured source is managed but not present in cache.
    Missing,
    /// A download/prepare operation is currently running for this server.
    Downloading,
    /// The configured source is present in cache.
    Downloaded {
        /// Local artifact path used by llama.cpp.
        path: String,
        /// Total on-disk size when known.
        size_bytes: Option<u64>,
    },
}

impl From<ModelArtifactState> for ModelArtifactView {
    fn from(state: ModelArtifactState) -> Self {
        match state {
            ModelArtifactState::NotManaged => Self::NotManaged,
            ModelArtifactState::Missing => Self::Missing,
            ModelArtifactState::Downloaded { path, size_bytes } => Self::Downloaded {
                path: path.as_str().to_owned(),
                size_bytes,
            },
        }
    }
}

/// A configured local model server plus derived artifact state.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ModelServerListItem {
    /// Persisted local model-server config.
    pub config: LocalModelServerConfig,
    /// Derived artifact cache state.
    pub artifact: ModelArtifactView,
}

/// Output of [`ListModelServers::execute`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ListModelServersOutput {
    /// Persisted local model-server configs enriched with artifact state.
    pub servers: Vec<ModelServerListItem>,
}

/// Lists local model-server configurations.
pub struct ListModelServers {
    registry: Arc<dyn ModelServerRegistry>,
    downloader: Option<Arc<dyn ModelArtifactDownloader>>,
    downloads: Option<Arc<dyn ModelArtifactDownloadTracker>>,
}

impl ListModelServers {
    /// Builds the use case.
    #[must_use]
    pub fn new(registry: Arc<dyn ModelServerRegistry>) -> Self {
        Self {
            registry,
            downloader: None,
            downloads: None,
        }
    }

    /// Enables artifact state enrichment for Hugging Face-backed servers.
    #[must_use]
    pub fn with_model_artifact_downloader(
        mut self,
        downloader: Arc<dyn ModelArtifactDownloader>,
    ) -> Self {
        self.downloader = Some(downloader);
        self
    }

    /// Enables in-flight download state enrichment.
    #[must_use]
    pub fn with_download_tracker(
        mut self,
        downloads: Arc<dyn ModelArtifactDownloadTracker>,
    ) -> Self {
        self.downloads = Some(downloads);
        self
    }

    /// Lists configs.
    ///
    /// # Errors
    /// [`AppError::ModelServer`] on registry failure.
    pub async fn execute(&self) -> Result<ListModelServersOutput, AppError> {
        let configs = self.registry.list().await?;
        let mut servers = Vec::with_capacity(configs.len());
        for config in configs {
            let artifact = self.artifact_view(&config).await?;
            servers.push(ModelServerListItem { config, artifact });
        }
        Ok(ListModelServersOutput { servers })
    }

    async fn artifact_view(
        &self,
        config: &LocalModelServerConfig,
    ) -> Result<ModelArtifactView, AppError> {
        if self
            .downloads
            .as_ref()
            .is_some_and(|downloads| downloads.is_model_artifact_download_in_progress(config.id))
        {
            return Ok(ModelArtifactView::Downloading);
        }
        let Some(ModelSource::HuggingFace { repo }) = config.model.source.as_ref() else {
            return Ok(ModelArtifactView::NotManaged);
        };
        let Some(downloader) = self.downloader.as_ref() else {
            return Ok(ModelArtifactView::Missing);
        };
        downloader
            .hf_model_state(repo)
            .await
            .map(ModelArtifactView::from)
            .map_err(AppError::from)
    }
}

/// Read-only in-flight download state shared by list/delete use cases.
pub trait ModelArtifactDownloadTracker: Send + Sync {
    /// Whether the model artifact for `server_id` is currently being resolved/downloaded.
    fn is_model_artifact_download_in_progress(&self, server_id: LocalModelServerId) -> bool;
}

/// Input for [`SaveModelServer::execute`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SaveModelServerInput {
    /// Config to upsert by id.
    pub config: LocalModelServerConfig,
}

/// Output of [`SaveModelServer::execute`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SaveModelServerOutput {
    /// Saved config.
    pub config: LocalModelServerConfig,
}

/// Saves a local model-server configuration.
pub struct SaveModelServer {
    registry: Arc<dyn ModelServerRegistry>,
}

impl SaveModelServer {
    /// Builds the use case.
    #[must_use]
    pub fn new(registry: Arc<dyn ModelServerRegistry>) -> Self {
        Self { registry }
    }

    /// Saves a config.
    ///
    /// # Errors
    /// [`AppError::ModelServer`] on registry failure.
    pub async fn execute(
        &self,
        input: SaveModelServerInput,
    ) -> Result<SaveModelServerOutput, AppError> {
        self.registry.save(input.config.clone()).await?;
        Ok(SaveModelServerOutput {
            config: input.config,
        })
    }
}

/// Input for [`DeleteModelServer::execute`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeleteModelServerInput {
    /// Config id to delete.
    pub server_id: LocalModelServerId,
}

/// Deletes a local model-server config when no OpenCode profile still references it.
pub struct DeleteModelServer {
    registry: Arc<dyn ModelServerRegistry>,
    profiles: Arc<dyn ProfileStore>,
}

impl DeleteModelServer {
    /// Builds the use case.
    #[must_use]
    pub fn new(registry: Arc<dyn ModelServerRegistry>, profiles: Arc<dyn ProfileStore>) -> Self {
        Self { registry, profiles }
    }

    /// Deletes a config after checking profile references.
    ///
    /// # Errors
    /// [`AppError::ModelServer`] with `code=model_server_in_use` when referenced.
    pub async fn execute(&self, input: DeleteModelServerInput) -> Result<(), AppError> {
        let profiles = self.profiles.list().await?;
        if profiles.iter().any(|profile| {
            profile
                .opencode
                .as_ref()
                .and_then(|config| config.local_model_server_id)
                == Some(input.server_id)
        }) {
            return Err(ModelServerError::InUse(input.server_id.to_string()).into());
        }
        self.registry.delete(input.server_id).await?;
        Ok(())
    }
}

/// Input for [`DeleteModelArtifact::execute`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeleteModelArtifactInput {
    /// Config id whose managed artifact cache should be deleted.
    pub server_id: LocalModelServerId,
}

/// Deletes a downloaded model artifact without deleting the server config.
pub struct DeleteModelArtifact {
    registry: Arc<dyn ModelServerRegistry>,
    probe: Arc<dyn ModelServerProbe>,
    downloader: Arc<dyn ModelArtifactDownloader>,
    downloads: Arc<dyn ModelArtifactDownloadTracker>,
    profiles: Arc<dyn ProfileStore>,
    projects: Arc<dyn ProjectStore>,
    contexts: Arc<dyn AgentContextStore>,
    live: Arc<dyn LiveAgentRegistry>,
}

impl DeleteModelArtifact {
    /// Builds the use case.
    #[allow(clippy::too_many_arguments)]
    #[must_use]
    pub fn new(
        registry: Arc<dyn ModelServerRegistry>,
        probe: Arc<dyn ModelServerProbe>,
        downloader: Arc<dyn ModelArtifactDownloader>,
        downloads: Arc<dyn ModelArtifactDownloadTracker>,
        profiles: Arc<dyn ProfileStore>,
        projects: Arc<dyn ProjectStore>,
        contexts: Arc<dyn AgentContextStore>,
        live: Arc<dyn LiveAgentRegistry>,
    ) -> Self {
        Self {
            registry,
            probe,
            downloader,
            downloads,
            profiles,
            projects,
            contexts,
            live,
        }
    }

    /// Deletes a managed Hugging Face artifact after safety checks.
    ///
    /// # Errors
    /// [`AppError::ModelServer`] when the server is missing, the source is not
    /// deletable, a download is active, or a live agent uses the server.
    pub async fn execute(&self, input: DeleteModelArtifactInput) -> Result<(), AppError> {
        let config = self
            .registry
            .get(&input.server_id)
            .await?
            .ok_or(ModelServerError::NotConfigured)?;
        let Some(ModelSource::HuggingFace { repo }) = config.model.source.as_ref() else {
            return Err(ModelServerError::Invalid(
                "only managed Hugging Face model artifacts can be deleted".to_owned(),
            )
            .into());
        };
        if self
            .downloads
            .is_model_artifact_download_in_progress(input.server_id)
        {
            return Err(ModelServerError::InUse(format!(
                "model artifact download in progress for {}",
                input.server_id
            ))
            .into());
        }
        self.ensure_server_not_reachable(&config).await?;
        self.ensure_not_used_by_live_agent(input.server_id).await?;
        self.downloader.delete_hf_model(repo).await?;
        Ok(())
    }

    async fn ensure_server_not_reachable(
        &self,
        config: &LocalModelServerConfig,
    ) -> Result<(), AppError> {
        match self.probe.probe(&config.endpoint).await? {
            ModelServerStatus::Unreachable => Ok(()),
            ModelServerStatus::ReadyReused | ModelServerStatus::ReadyStarted => {
                Err(ModelServerError::InUse(format!(
                    "model server {} is currently reachable",
                    config.id
                ))
                .into())
            }
        }
    }

    async fn ensure_not_used_by_live_agent(
        &self,
        server_id: LocalModelServerId,
    ) -> Result<(), AppError> {
        let profiles = self.profiles.list().await?;
        let profile_server: HashMap<_, _> = profiles
            .iter()
            .filter_map(|profile| {
                profile
                    .opencode
                    .as_ref()
                    .and_then(|opencode| opencode.local_model_server_id)
                    .map(|id| (profile.id, id))
            })
            .collect();

        let mut agents_by_project = HashMap::new();
        for snapshot in self.live.live_agent_snapshots() {
            let agents = if let Some(agents) = agents_by_project.get(&snapshot.project_id) {
                agents
            } else {
                let project = self.projects.load_project(snapshot.project_id).await?;
                let manifest = self.contexts.load_manifest(&project).await?;
                agents_by_project.insert(
                    snapshot.project_id,
                    manifest
                        .entries
                        .iter()
                        .map(|entry| {
                            entry
                                .to_agent()
                                .map_err(|err| AppError::Invalid(err.to_string()))
                        })
                        .collect::<Result<Vec<_>, _>>()?,
                );
                agents_by_project
                    .get(&snapshot.project_id)
                    .expect("project agents inserted")
            };
            let Some(agent) = agents.iter().find(|agent| agent.id == snapshot.agent_id) else {
                continue;
            };
            if profile_server.get(&agent.profile_id) == Some(&server_id) {
                return Err(ModelServerError::InUse(format!(
                    "model server {server_id} is used by live agent {}",
                    agent.id
                ))
                .into());
            }
        }
        Ok(())
    }
}

/// Input for [`EnsureLocalModelServer::execute`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EnsureLocalModelServerInput {
    /// Referenced local model server.
    pub server_id: LocalModelServerId,
}

/// Output of [`EnsureLocalModelServer::execute`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EnsureLocalModelServerOutput {
    /// Ready server data to inject into OpenCode config.
    pub ready: ModelServerReady,
}

#[derive(Debug, Default)]
struct ServerUse {
    refs: usize,
}

/// RAII guard held by live OpenCode sessions while they use a local model server.
///
/// Local llama.cpp servers expose a stateless OpenAI-compatible endpoint, so they
/// may be shared by legitimate OpenCode launches across projects. The guard only
/// tracks liveness so session removal releases the usage reference.
#[derive(Debug)]
pub struct ModelServerUseGuard {
    server_id: LocalModelServerId,
    usages: Arc<Mutex<HashMap<LocalModelServerId, ServerUse>>>,
}

impl Drop for ModelServerUseGuard {
    fn drop(&mut self) {
        let Ok(mut usages) = self.usages.lock() else {
            return;
        };
        let Some(active) = usages.get_mut(&self.server_id) else {
            return;
        };
        active.refs = active.refs.saturating_sub(1);
        if active.refs == 0 {
            usages.remove(&self.server_id);
        }
    }
}

/// Readiness retry policy.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ReadinessPolicy {
    /// Number of probes after spawning.
    pub attempts: usize,
    /// Delay between attempts.
    pub backoff: Duration,
    /// Maximum time to wait for an auto-started process to expose its endpoint.
    pub warmup_deadline: Duration,
}

impl Default for ReadinessPolicy {
    fn default() -> Self {
        Self {
            attempts: 20,
            backoff: Duration::from_millis(250),
            warmup_deadline: Duration::from_secs(600),
        }
    }
}

/// Default upper bound for Hugging Face model download/preparation.
///
/// llama.cpp may download multi-GB models before binding the OpenAI-compatible
/// endpoint. The short readiness window remains for local files; HF sources use
/// this separate deadline while the process is still alive.
pub const DEFAULT_HF_DOWNLOAD_DEADLINE: Duration = Duration::from_secs(30 * 60);

#[derive(Debug, Clone)]
struct ActiveServer {
    handle: ManagedProcessHandle,
    port: u16,
    stop_policy: StopPolicy,
}

#[derive(Debug)]
struct InflightEnsure {
    result: Mutex<Option<Result<EnsureLocalModelServerOutput, AppError>>>,
    notify: Notify,
}

impl InflightEnsure {
    fn new() -> Self {
        Self {
            result: Mutex::new(None),
            notify: Notify::new(),
        }
    }

    async fn wait(&self) -> Result<EnsureLocalModelServerOutput, AppError> {
        loop {
            if let Some(result) = self.result.lock().unwrap().clone() {
                return result;
            }
            self.notify.notified().await;
        }
    }

    fn complete(&self, result: Result<EnsureLocalModelServerOutput, AppError>) {
        *self.result.lock().unwrap() = Some(result);
        self.notify.notify_waiters();
    }
}

/// Ensures a configured local model server is reachable, starting it when allowed.
pub struct EnsureLocalModelServer {
    registry: Arc<dyn ModelServerRegistry>,
    probe: Arc<dyn ModelServerProbe>,
    process: Arc<dyn ManagedProcess>,
    runtime: Arc<dyn ModelServerRuntime>,
    fs: Arc<dyn FileSystem>,
    events: Arc<dyn EventBus>,
    active: Mutex<HashMap<LocalModelServerId, ActiveServer>>,
    inflight: AsyncMutex<HashMap<LocalModelServerId, Arc<InflightEnsure>>>,
    usages: Arc<Mutex<HashMap<LocalModelServerId, ServerUse>>>,
    download_cancels: Mutex<HashMap<LocalModelServerId, ModelArtifactCancel>>,
    readiness: ReadinessPolicy,
    hf_download_deadline: Duration,
    model_artifact_downloader: Option<Arc<dyn ModelArtifactDownloader>>,
}

impl EnsureLocalModelServer {
    /// Builds the use case.
    #[allow(clippy::too_many_arguments)]
    #[must_use]
    pub fn new(
        registry: Arc<dyn ModelServerRegistry>,
        probe: Arc<dyn ModelServerProbe>,
        process: Arc<dyn ManagedProcess>,
        runtime: Arc<dyn ModelServerRuntime>,
        fs: Arc<dyn FileSystem>,
        events: Arc<dyn EventBus>,
    ) -> Self {
        Self {
            registry,
            probe,
            process,
            runtime,
            fs,
            events,
            active: Mutex::new(HashMap::new()),
            inflight: AsyncMutex::new(HashMap::new()),
            usages: Arc::new(Mutex::new(HashMap::new())),
            download_cancels: Mutex::new(HashMap::new()),
            readiness: ReadinessPolicy::default(),
            hf_download_deadline: DEFAULT_HF_DOWNLOAD_DEADLINE,
            model_artifact_downloader: None,
        }
    }

    /// Overrides readiness policy, mainly for tests.
    #[must_use]
    pub fn with_readiness_policy(mut self, readiness: ReadinessPolicy) -> Self {
        self.readiness = readiness;
        self
    }

    /// Returns the readiness policy effective for a persisted server config.
    ///
    /// The config may override only the warmup deadline; probe cadence remains
    /// owned by the application policy.
    #[must_use]
    pub fn effective_readiness_policy(&self, config: &LocalModelServerConfig) -> ReadinessPolicy {
        self.readiness_for(config)
    }

    /// Overrides the long Hugging Face download/preparation deadline.
    #[must_use]
    pub fn with_hf_download_deadline(mut self, deadline: Duration) -> Self {
        self.hf_download_deadline = deadline;
        self
    }

    /// Enables pre-resolution/download of Hugging Face artifacts.
    #[must_use]
    pub fn with_model_artifact_downloader(
        mut self,
        downloader: Arc<dyn ModelArtifactDownloader>,
    ) -> Self {
        self.model_artifact_downloader = Some(downloader);
        self
    }

    /// Acquires a live-use reference of `server_id`.
    ///
    /// # Errors
    /// [`AppError::ModelServer`] with `code=model_server_in_use` if the usage
    /// tracker lock is poisoned.
    pub fn acquire_use(
        &self,
        server_id: LocalModelServerId,
    ) -> Result<ModelServerUseGuard, AppError> {
        let mut usages = self
            .usages
            .lock()
            .map_err(|_| ModelServerError::InUse(server_id.to_string()))?;
        let active = usages.entry(server_id).or_default();
        active.refs = active.refs.saturating_add(1);
        Ok(ModelServerUseGuard {
            server_id,
            usages: Arc::clone(&self.usages),
        })
    }

    /// Ensures the server is reachable.
    ///
    /// # Errors
    /// [`AppError::ModelServer`] if the server cannot be prepared.
    pub async fn execute(
        &self,
        input: EnsureLocalModelServerInput,
    ) -> Result<EnsureLocalModelServerOutput, AppError> {
        let server_id = input.server_id;
        let (flight, is_leader) = {
            let mut inflight = self.inflight.lock().await;
            if let Some(flight) = inflight.get(&server_id) {
                (Arc::clone(flight), false)
            } else {
                let flight = Arc::new(InflightEnsure::new());
                inflight.insert(server_id, Arc::clone(&flight));
                (flight, true)
            }
        };

        if !is_leader {
            return flight.wait().await;
        }

        let result = self.execute_inner(input).await;
        flight.complete(result.clone());
        let mut inflight = self.inflight.lock().await;
        if inflight
            .get(&server_id)
            .is_some_and(|current| Arc::ptr_eq(current, &flight))
        {
            inflight.remove(&server_id);
        }
        result
    }

    async fn execute_inner(
        &self,
        input: EnsureLocalModelServerInput,
    ) -> Result<EnsureLocalModelServerOutput, AppError> {
        let config = self
            .registry
            .get(&input.server_id)
            .await?
            .ok_or(ModelServerError::NotConfigured)?;

        self.publish(config.id, ModelServerLifecycleStatus::Probing);
        let initial_probe = match self.probe.probe(&config.endpoint).await {
            Ok(status) => status,
            Err(err) => return self.fail(config.id, err),
        };
        match initial_probe {
            ModelServerStatus::ReadyReused | ModelServerStatus::ReadyStarted => {
                self.publish(
                    config.id,
                    ModelServerLifecycleStatus::Ready { reused: true },
                );
                return Ok(EnsureLocalModelServerOutput {
                    ready: ready(&config, ModelServerStatus::ReadyReused),
                });
            }
            ModelServerStatus::Unreachable => {}
        }

        if !config.auto_start {
            let err = ModelServerError::Probe("server unreachable and autoStart=false".to_owned());
            self.publish_failure(config.id, &err);
            return Err(err.into());
        }

        self.ensure_model_path_accessible(&config).await?;
        self.ensure_no_active_port_collision(&config).await?;

        let spawn_config = match self.prepare_spawn_config(config.clone()).await {
            Ok(config) => config,
            Err(err) => return self.fail(config.id, err),
        };

        self.publish(config.id, ModelServerLifecycleStatus::Starting);
        let spec = match self.runtime.build_spawn_spec(&spawn_config) {
            Ok(spec) => spec,
            Err(err) => return self.fail(config.id, err),
        };
        let handle = match self.process.spawn(spec).await {
            Ok(handle) => handle,
            Err(err) => return self.fail(config.id, err),
        };
        self.active.lock().unwrap().insert(
            config.id,
            ActiveServer {
                handle: handle.clone(),
                port: spawn_config.endpoint.port,
                stop_policy: spawn_config.stop_policy,
            },
        );

        let hf_source = hf_source(&spawn_config);
        self.wait_for_started_server(&spawn_config, &handle, hf_source)
            .await
    }

    async fn prepare_spawn_config(
        &self,
        config: LocalModelServerConfig,
    ) -> Result<LocalModelServerConfig, ModelServerError> {
        let Some(ModelSource::HuggingFace { repo }) = config.model.source.clone() else {
            return Ok(config);
        };
        let Some(downloader) = self.model_artifact_downloader.as_ref() else {
            return Ok(config);
        };

        let cancel = ModelArtifactCancel::new();
        self.download_cancels
            .lock()
            .unwrap()
            .insert(config.id, cancel.clone());

        let events = Arc::clone(&self.events);
        let source = repo.as_str().to_owned();
        let server_id = config.id;
        let debouncer = Arc::new(Mutex::new(DownloadProgressDebouncer::new(
            server_id,
            source.clone(),
            events,
        )));
        let progress = {
            let debouncer = Arc::clone(&debouncer);
            Arc::new(move |progress: ModelArtifactProgress| {
                debouncer.lock().unwrap().on_progress(progress);
            }) as Arc<dyn Fn(ModelArtifactProgress) + Send + Sync>
        };

        let result = tokio::time::timeout(
            self.hf_download_deadline,
            downloader.resolve_hf_model(&repo, progress, cancel.clone()),
        )
        .await;
        self.download_cancels.lock().unwrap().remove(&config.id);

        let resolution = match result {
            Ok(Ok(resolution)) => resolution,
            Ok(Err(err)) => return Err(err),
            Err(_) => {
                cancel.cancel();
                return Err(ModelServerError::Timeout);
            }
        };

        let mut resolved = config;
        resolved.model.source = Some(ModelSource::LocalPath {
            path: resolution.path,
        });
        Ok(resolved)
    }

    async fn wait_for_started_server(
        &self,
        config: &LocalModelServerConfig,
        handle: &ManagedProcessHandle,
        hf_source: Option<String>,
    ) -> Result<EnsureLocalModelServerOutput, AppError> {
        let readiness = self.readiness_for(config);
        let mut attempts = 0usize;
        let deadline = Instant::now() + readiness.warmup_deadline;
        loop {
            match self.probe.probe(&config.endpoint).await {
                Err(err) => {
                    self.stop_started_server(config.id, handle).await;
                    return self.fail(config.id, err);
                }
                Ok(ModelServerStatus::ReadyReused | ModelServerStatus::ReadyStarted) => {
                    self.publish(
                        config.id,
                        ModelServerLifecycleStatus::Ready { reused: false },
                    );
                    return Ok(EnsureLocalModelServerOutput {
                        ready: ready(config, ModelServerStatus::ReadyStarted),
                    });
                }
                Ok(ModelServerStatus::Unreachable) => {}
            }

            match self.process.status(handle).await {
                Ok(ProcessStatus::Running) => {
                    if attempts.saturating_add(1) == readiness.attempts {
                        if let Some(source) = hf_source.as_ref() {
                            self.publish(
                                config.id,
                                ModelServerLifecycleStatus::Downloading {
                                    downloaded_bytes: None,
                                    total_bytes: None,
                                    percent: None,
                                    source: Some(source.clone()),
                                },
                            );
                        }
                    }
                }
                Ok(ProcessStatus::Exited { code }) => {
                    self.active.lock().unwrap().remove(&config.id);
                    return self.fail(config.id, premature_exit_error(code));
                }
                Ok(ProcessStatus::Unknown) => {
                    self.active.lock().unwrap().remove(&config.id);
                    return self.fail(
                        config.id,
                        ModelServerError::Process("process status unknown".to_owned()),
                    );
                }
                Err(err) => return self.fail(config.id, err),
            }

            attempts = attempts.saturating_add(1);
            if Instant::now() >= deadline {
                let err = ModelServerError::Timeout;
                self.stop_started_server(config.id, handle).await;
                return self.fail(config.id, err);
            }
            if !readiness.backoff.is_zero() {
                tokio::time::sleep(readiness.backoff).await;
            } else {
                tokio::task::yield_now().await;
            }
        }
    }

    fn readiness_for(&self, config: &LocalModelServerConfig) -> ReadinessPolicy {
        ReadinessPolicy {
            warmup_deadline: config
                .warmup_deadline_secs
                .map(Duration::from_secs)
                .unwrap_or(self.readiness.warmup_deadline),
            ..self.readiness
        }
    }

    /// Stops active servers whose policy is [`StopPolicy::StopOnAppExit`].
    ///
    /// # Errors
    /// Returns the first process error after attempting every eligible stop.
    pub async fn stop_on_app_exit(&self) -> Result<(), AppError> {
        for cancel in self.download_cancels.lock().unwrap().values() {
            cancel.cancel();
        }
        let entries: Vec<(LocalModelServerId, ActiveServer)> = self
            .active
            .lock()
            .unwrap()
            .iter()
            .map(|(id, active)| (*id, active.clone()))
            .collect();
        let mut first_error: Option<ModelServerError> = None;
        for (id, active) in entries {
            if active.stop_policy != StopPolicy::StopOnAppExit {
                continue;
            }
            if let Err(err) = self.process.kill(&active.handle).await {
                if first_error.is_none() {
                    first_error = Some(err);
                }
            } else {
                self.active.lock().unwrap().remove(&id);
            }
        }
        if let Some(err) = first_error {
            Err(err.into())
        } else {
            Ok(())
        }
    }

    async fn ensure_model_path_accessible(
        &self,
        config: &LocalModelServerConfig,
    ) -> Result<(), AppError> {
        let Some(ModelSource::LocalPath { path }) = config.model.source.as_ref() else {
            return Ok(());
        };
        if path.as_str().is_empty() {
            let err = ModelServerError::PathNotAccessible("model.source.path missing".to_owned());
            self.publish_failure(config.id, &err);
            return Err(err.into());
        }
        match self
            .fs
            .exists(&RemotePath::new(path.as_str().to_owned()))
            .await
        {
            Ok(true) => Ok(()),
            Ok(false) => {
                let err = ModelServerError::PathNotAccessible(path.as_str().to_owned());
                self.publish_failure(config.id, &err);
                Err(err.into())
            }
            Err(domain::ports::FsError::PermissionDenied(p)) => {
                let err = ModelServerError::PermissionDenied(p);
                self.publish_failure(config.id, &err);
                Err(err.into())
            }
            Err(err) => {
                let err = ModelServerError::PathNotAccessible(err.to_string());
                self.publish_failure(config.id, &err);
                Err(err.into())
            }
        }
    }

    async fn ensure_no_active_port_collision(
        &self,
        config: &LocalModelServerConfig,
    ) -> Result<(), AppError> {
        let active: Vec<(LocalModelServerId, ActiveServer)> = self
            .active
            .lock()
            .unwrap()
            .iter()
            .map(|(id, active)| (*id, active.clone()))
            .collect();
        for (id, active) in active {
            if active.port != config.endpoint.port {
                continue;
            }
            match self.process.status(&active.handle).await {
                Ok(ProcessStatus::Running) => {
                    let err = ModelServerError::PortOccupied(config.endpoint.port);
                    self.publish_failure(config.id, &err);
                    return Err(err.into());
                }
                Ok(ProcessStatus::Exited { .. } | ProcessStatus::Unknown) => {
                    self.active.lock().unwrap().remove(&id);
                }
                Err(err) => return self.fail(config.id, err),
            }
        }
        Ok(())
    }

    async fn stop_started_server(
        &self,
        server_id: LocalModelServerId,
        handle: &ManagedProcessHandle,
    ) {
        let _ = self.process.kill(handle).await;
        self.active.lock().unwrap().remove(&server_id);
    }

    fn publish(&self, server_id: LocalModelServerId, status: ModelServerLifecycleStatus) {
        self.events
            .publish(DomainEvent::ModelServerStatusChanged { server_id, status });
    }

    fn fail<T>(&self, server_id: LocalModelServerId, err: ModelServerError) -> Result<T, AppError> {
        self.publish_failure(server_id, &err);
        Err(err.into())
    }

    fn publish_failure(&self, server_id: LocalModelServerId, err: &ModelServerError) {
        self.publish(
            server_id,
            ModelServerLifecycleStatus::Failed {
                code: model_server_error_code(err).to_owned(),
                message: err.to_string(),
            },
        );
    }
}

impl ModelArtifactDownloadTracker for EnsureLocalModelServer {
    fn is_model_artifact_download_in_progress(&self, server_id: LocalModelServerId) -> bool {
        self.download_cancels
            .lock()
            .unwrap()
            .contains_key(&server_id)
    }
}

fn ready(config: &LocalModelServerConfig, status: ModelServerStatus) -> ModelServerReady {
    ModelServerReady {
        base_url: config.endpoint.base_url.clone(),
        model: config.model.served_name.clone(),
        status,
    }
}

fn hf_source(config: &LocalModelServerConfig) -> Option<String> {
    match config.model.source.as_ref()? {
        ModelSource::HuggingFace { repo } => Some(repo.as_str().to_owned()),
        ModelSource::LocalPath { .. } => None,
    }
}

fn premature_exit_error(code: Option<i32>) -> ModelServerError {
    ModelServerError::Process(match code {
        Some(code) => format!("model server exited before readiness with code {code}"),
        None => "model server exited before readiness".to_owned(),
    })
}

/// Stable model-server error code for event/DTO mapping.
#[must_use]
pub fn model_server_error_code(err: &ModelServerError) -> &'static str {
    match err {
        ModelServerError::NotConfigured => "not_configured",
        ModelServerError::Invalid(_) => "invalid",
        ModelServerError::PermissionDenied(_) => "permission_denied",
        ModelServerError::PathNotAccessible(_) => "path_not_accessible",
        ModelServerError::PortOccupied(_) => "port_occupied",
        ModelServerError::InUse(_) => "model_server_in_use",
        ModelServerError::Probe(_) => "probe",
        ModelServerError::Process(_) => "process",
        ModelServerError::Store(_) => "store",
        ModelServerError::Cancelled => "cancelled",
        ModelServerError::Timeout => "timeout",
    }
}

struct DownloadProgressDebouncer {
    server_id: LocalModelServerId,
    fallback_source: String,
    events: Arc<dyn EventBus>,
    last_emit: Option<StdInstant>,
    last_percent: Option<f32>,
    window_start: StdInstant,
    window_count: u32,
}

impl DownloadProgressDebouncer {
    fn new(
        server_id: LocalModelServerId,
        fallback_source: String,
        events: Arc<dyn EventBus>,
    ) -> Self {
        let now = StdInstant::now();
        Self {
            server_id,
            fallback_source,
            events,
            last_emit: None,
            last_percent: None,
            window_start: now,
            window_count: 0,
        }
    }

    fn on_progress(&mut self, progress: ModelArtifactProgress) {
        let percent = progress_percent(&progress);
        let final_progress = is_final_progress(&progress);
        if !self.should_publish(percent, final_progress) {
            return;
        }
        self.mark_published(percent);
        self.events.publish(DomainEvent::ModelServerStatusChanged {
            server_id: self.server_id,
            status: ModelServerLifecycleStatus::Downloading {
                downloaded_bytes: progress.downloaded_bytes,
                total_bytes: progress.total_bytes,
                percent,
                source: progress
                    .source
                    .or_else(|| Some(self.fallback_source.clone())),
            },
        });
    }

    fn should_publish(&mut self, percent: Option<f32>, final_progress: bool) -> bool {
        let now = StdInstant::now();
        if now.duration_since(self.window_start) >= Duration::from_secs(1) {
            self.window_start = now;
            self.window_count = 0;
        }
        if self.last_emit.is_none() || final_progress {
            return true;
        }
        if self.window_count >= 10 {
            return false;
        }
        if self
            .last_emit
            .is_some_and(|last| now.duration_since(last) < Duration::from_millis(100))
        {
            return false;
        }
        if self
            .last_emit
            .is_some_and(|last| now.duration_since(last) >= Duration::from_millis(500))
        {
            return true;
        }
        match (self.last_percent, percent) {
            (Some(last), Some(current)) => current - last >= 1.0,
            (None, Some(_)) => true,
            _ => false,
        }
    }

    fn mark_published(&mut self, percent: Option<f32>) {
        self.last_emit = Some(StdInstant::now());
        self.last_percent = percent.or(self.last_percent);
        self.window_count += 1;
    }
}

fn progress_percent(progress: &ModelArtifactProgress) -> Option<f32> {
    let downloaded = progress.downloaded_bytes?;
    let total = progress.total_bytes?;
    if total == 0 {
        return None;
    }
    Some((downloaded as f32 / total as f32) * 100.0)
}

fn is_final_progress(progress: &ModelArtifactProgress) -> bool {
    matches!(
        (progress.downloaded_bytes, progress.total_bytes),
        (Some(downloaded), Some(total)) if total > 0 && downloaded == total
    )
}