nomoreide-daemon 0.20.1

mod bundles;

use nomoreide_core::config::ConfigStore;
use nomoreide_core::log_store::LogEntry;
use nomoreide_core::port_utils::PortHolder;
use nomoreide_core::process_manager::{
    PortConflictError, ProcessManager, ServiceState, ServiceStatus,
};
use nomoreide_core::timeline::{
    TimelineEvent as CoreTimelineEvent, TimelineEventKind as CoreKind,
    TimelineSeverity as CoreSeverity,
};
use nomoreide_daemon_client::protocol::{
    InspectorRuntimeStatus, PortConflict, PortHolderIdentity, ServiceLogEntry, ServiceRuntimeState,
    ServiceRuntimeStatus, TimelineEvent, TimelineEventKind, TimelineSeverity,
};
use std::future::Future;
use std::sync::atomic::{AtomicU8, Ordering};
use std::sync::Arc;
use tokio::sync::RwLock;

const PHASE_RUNNING: u8 = 0;
const PHASE_DRAINING: u8 = 1;
const PHASE_CLEANUP_FAILED: u8 = 2;

#[derive(Debug)]
pub(crate) enum RuntimeMutationError {
    /// Carries the requested name so the refusal can echo it back, as the
    /// reference does — the caller supplied it, so repeating it leaks nothing.
    ServiceNotFound(String),
    UnsupportedServiceKind,
    PortConflict {
        message: String,
        conflict: Box<PortConflict>,
    },
    DaemonDraining,
    DaemonCleanupFailed,
    ConfigLoadFailed,
    ServiceStartFailed,
    CleanupFailed,
    BundleNotFound(String),
    DependencyCycle(String),
}

pub(crate) struct DaemonRuntime {
    config_store: ConfigStore,
    process_manager: Arc<ProcessManager>,
    mutation_gate: RwLock<()>,
    phase: AtomicU8,
}

impl DaemonRuntime {
    pub(crate) fn new(config_store: ConfigStore, process_manager: ProcessManager) -> Self {
        Self {
            config_store,
            process_manager: Arc::new(process_manager),
            mutation_gate: RwLock::new(()),
            phase: AtomicU8::new(PHASE_RUNNING),
        }
    }

    pub(crate) async fn reconcile_runtime(&self) -> anyhow::Result<()> {
        self.process_manager.reconcile_runtime().await
    }

    /// Every service this daemon is tracking, sorted by name. The process map
    /// behind it is unordered, so sorting is what makes two consecutive reads
    /// comparable.
    pub(crate) fn status(&self) -> Vec<ServiceRuntimeStatus> {
        let mut statuses = self
            .process_manager
            .status()
            .into_iter()
            .map(runtime_status)
            .collect::<Vec<_>>();
        statuses.sort_by(|left, right| left.name.cmp(&right.name));
        statuses
    }

    /// What the manager knows about one service, unmapped.
    ///
    /// Core's own status rather than the wire's, because the caller is core
    /// too — the repro bundle renders a service's run state and has no business
    /// reading a shape invented for the dashboard.
    /// Toggle a service's HTTP inspector, answering its status afterwards.
    pub(crate) async fn set_inspector_enabled(
        &self,
        name: &str,
        enabled: bool,
    ) -> Result<ServiceRuntimeStatus, String> {
        self.process_manager
            .set_inspector_enabled(name, enabled)
            .await
            .map(runtime_status)
    }

    pub(crate) fn service_status(&self, name: &str) -> Option<ServiceStatus> {
        self.process_manager
            .status()
            .into_iter()
            .find(|status| status.name == name)
    }

    /// The tail of a service's buffered output.
    ///
    /// Reading logs is not gated on the service being registered, the way a
    /// start is: a service whose definition was removed, or that this daemon
    /// never ran, still has whatever it already wrote, and that is exactly what
    /// someone debugging its disappearance needs. An unknown name has no lines
    /// rather than being an error, matching the reference.
    pub(crate) fn logs(&self, name: &str, lines: usize) -> Vec<ServiceLogEntry> {
        self.process_manager
            .logs(name, lines)
            .into_iter()
            .map(log_entry)
            .collect()
    }

    /// The most recent timeline events, oldest last.
    pub(crate) fn timeline(&self, limit: usize) -> Vec<TimelineEvent> {
        self.process_manager
            .timeline(limit)
            .into_iter()
            .map(timeline_event)
            .collect()
    }

    pub(crate) async fn start_service(
        &self,
        name: &str,
    ) -> Result<ServiceRuntimeStatus, RuntimeMutationError> {
        self.launch(name, Launch::Start).await
    }

    /// A restart ends in a start, so it takes the start gate rather than the
    /// lenient stop one, and it resolves the definition *before* anything is
    /// stopped: a restart that could never start again must not take the
    /// running process down on its way to failing.
    pub(crate) async fn restart_service(
        &self,
        name: &str,
    ) -> Result<ServiceRuntimeStatus, RuntimeMutationError> {
        self.launch(name, Launch::Restart).await
    }

    /// Both launches hold a single mutation permit for the whole operation. A
    /// restart that took one permit to stop and a second to start would let a
    /// shutdown drain in between and leave the service down. The stop and the
    /// start are handed to the process manager together so they also share its
    /// per-service operation lock.
    async fn launch(
        &self,
        name: &str,
        mode: Launch,
    ) -> Result<ServiceRuntimeStatus, RuntimeMutationError> {
        self.require_start_allowed()?;
        let _permit = self.mutation_gate.read().await;
        self.require_start_allowed()?;
        let service = self.registered_startable_service(name).await?;
        match mode {
            Launch::Start => self.process_manager.start_service(&service).await,
            Launch::Restart => self.process_manager.restart_service(&service).await,
        }
        .map_err(launch_error)?;
        self.process_manager
            .service_status(name)
            .map(runtime_status)
            .ok_or(RuntimeMutationError::ServiceStartFailed)
    }

    /// Stopping is a remediation capability, so a name this daemon is already
    /// running stays stoppable even after its definition was edited, removed,
    /// or made temporarily unreadable — otherwise config drift would strand a
    /// live process group with no way to reach it. Names this daemon does not
    /// own still have to be registered local services.
    pub(crate) async fn stop_service(
        &self,
        name: &str,
    ) -> Result<ServiceRuntimeStatus, RuntimeMutationError> {
        self.require_stop_allowed()?;
        let _permit = self.mutation_gate.read().await;
        self.require_stop_allowed()?;
        // **A registration check the reference does not have.** The reference
        // stops by name without consulting config, so a name it has never heard
        // of answers `stopped` and it records a runtime entry and a timeline
        // event for what is nearly always a typo. This is a declared divergence
        // — see the note on `error/stop-unregistered` in
        // scripts/check-mcp-runtime-parity.ts — not an oversight, and both
        // sides of it are asserted by that gate and by the service-config gate.
        if self.process_manager.service_status(name).is_none() {
            self.registered_startable_service(name).await?;
        }
        self.process_manager
            .stop_service(name)
            .await
            .map_err(|_| RuntimeMutationError::CleanupFailed)?;
        Ok(self
            .process_manager
            .service_status(name)
            .map(runtime_status)
            .unwrap_or_else(|| stopped_status(name)))
    }

    pub(crate) async fn shutdown(&self) -> Result<(), String> {
        self.shutdown_with(async {
            self.process_manager
                .shutdown_all()
                .await
                .map_err(|error| error.to_string())
        })
        .await
    }

    async fn shutdown_with<F>(&self, cleanup: F) -> Result<(), String>
    where
        F: Future<Output = Result<(), String>>,
    {
        let phase = self.phase.load(Ordering::Acquire);
        if phase == PHASE_DRAINING {
            return Err("daemon cleanup is already in progress".into());
        }
        self.phase.store(PHASE_DRAINING, Ordering::Release);
        let _permit = self.mutation_gate.write().await;
        match cleanup.await {
            Ok(()) => Ok(()),
            Err(error) => {
                self.phase.store(PHASE_CLEANUP_FAILED, Ordering::Release);
                Err(error)
            }
        }
    }

    fn require_start_allowed(&self) -> Result<(), RuntimeMutationError> {
        match self.phase.load(Ordering::Acquire) {
            PHASE_RUNNING => Ok(()),
            PHASE_DRAINING => Err(RuntimeMutationError::DaemonDraining),
            _ => Err(RuntimeMutationError::DaemonCleanupFailed),
        }
    }

    fn require_stop_allowed(&self) -> Result<(), RuntimeMutationError> {
        match self.phase.load(Ordering::Acquire) {
            PHASE_RUNNING | PHASE_CLEANUP_FAILED => Ok(()),
            _ => Err(RuntimeMutationError::DaemonDraining),
        }
    }

    async fn registered_startable_service(
        &self,
        name: &str,
    ) -> Result<nomoreide_core::config::ServiceDef, RuntimeMutationError> {
        let config = self.config().await?;
        startable_service(&config, name).cloned()
    }

    async fn config(&self) -> Result<nomoreide_core::config::Config, RuntimeMutationError> {
        self.config_store
            .load()
            .await
            .map_err(|_| RuntimeMutationError::ConfigLoadFailed)
    }
}

/// The one definition this daemon will run for `name`.
///
/// Every kind this runtime knows how to run qualifies. A local and a remote
/// service are both a child this daemon supervises, differing only in which
/// program that child is; a compose service is a container the Docker daemon
/// runs on our behalf. Anything else is a kind nothing here implements, and a
/// config can name one by hand.
fn startable_service<'a>(
    config: &'a nomoreide_core::config::Config,
    name: &str,
) -> Result<&'a nomoreide_core::config::ServiceDef, RuntimeMutationError> {
    let service = config
        .services
        .iter()
        .find(|service| service.name == name)
        .ok_or_else(|| RuntimeMutationError::ServiceNotFound(name.to_string()))?;
    if !matches!(service.effective_kind(), "local" | "ssh" | "docker-compose") {
        return Err(RuntimeMutationError::UnsupportedServiceKind);
    }
    Ok(service)
}

#[derive(Clone, Copy)]
enum Launch {
    Start,
    Restart,
}

/// A restart reports a failed stop as a failed start, because the process
/// manager returns one result for both phases. The daemon keeps the protocol's
/// existing error codes rather than inventing a restart-only one the reference
/// implementation does not have.
fn launch_error(error: anyhow::Error) -> RuntimeMutationError {
    if let Some(conflict) = error.downcast_ref::<PortConflictError>() {
        return RuntimeMutationError::PortConflict {
            message: conflict.to_string(),
            conflict: Box::new(PortConflict {
                code: nomoreide_daemon_client::protocol::PORT_IN_USE.to_string(),
                port: conflict.port,
                holder: conflict.holder.as_ref().map(holder_identity),
            }),
        };
    }
    RuntimeMutationError::ServiceStartFailed
}

fn runtime_status(status: ServiceStatus) -> ServiceRuntimeStatus {
    // `exitedAt` is the one field stamped for every ending, whatever it was, so
    // it — not the exit code — decides whether this run has an ending to
    // report. A container's end is not a process's end, though: it has no exit
    // code and was killed by no signal, so it reports neither rather than
    // reporting both as null. The same rule the MCP status surface applies.
    let ended = status.exited_at.is_some() && status.container_id.is_none();
    ServiceRuntimeStatus {
        name: status.name,
        state: match status.state {
            ServiceState::Stopped => ServiceRuntimeState::Stopped,
            ServiceState::Starting => ServiceRuntimeState::Starting,
            ServiceState::Running => ServiceRuntimeState::Running,
            ServiceState::Stopping => ServiceRuntimeState::Stopping,
            ServiceState::Exited => ServiceRuntimeState::Exited,
        },
        kind: Some(status.kind),
        host: status.host,
        container_id: status.container_id,
        pid: status.pid,
        exit_code: ended.then_some(status.exit_code),
        url: status.url,
        started_at: status.started_at.map(iso_millis),
        exited_at: status.exited_at.map(iso_millis),
        signal: ended.then_some(status.signal),
        inspector: status.inspector.map(|inspector| InspectorRuntimeStatus {
            enabled: inspector.enabled,
            port: inspector.port,
            upstream_port: inspector.upstream_port,
        }),
    }
}

/// Timestamps cross the wire the way the reference writes them — an ISO string
/// with millisecond precision — rather than in chrono's default nanosecond
/// form, which no reference client has ever seen.
fn iso_millis(at: chrono::DateTime<chrono::Utc>) -> String {
    at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
}

fn log_entry(entry: LogEntry) -> ServiceLogEntry {
    ServiceLogEntry {
        service: entry.service,
        stream: entry.stream,
        text: entry.text,
        timestamp: iso_millis(entry.timestamp),
    }
}

fn timeline_event(event: CoreTimelineEvent) -> TimelineEvent {
    TimelineEvent {
        id: event.id,
        timestamp: iso_millis(event.timestamp),
        kind: match event.kind {
            CoreKind::ServiceLifecycle => TimelineEventKind::ServiceLifecycle,
            CoreKind::ServiceLog => TimelineEventKind::ServiceLog,
            CoreKind::ServiceHealth => TimelineEventKind::ServiceHealth,
            CoreKind::ServicePort => TimelineEventKind::ServicePort,
            CoreKind::ServiceHttp => TimelineEventKind::ServiceHttp,
            CoreKind::McpTool => TimelineEventKind::McpTool,
            CoreKind::GitChange => TimelineEventKind::GitChange,
            CoreKind::UserAction => TimelineEventKind::UserAction,
        },
        service: event.service,
        severity: match event.severity {
            CoreSeverity::Info => TimelineSeverity::Info,
            CoreSeverity::Warning => TimelineSeverity::Warning,
            CoreSeverity::Error => TimelineSeverity::Error,
        },
        title: event.title,
        detail: event.detail,
        data: event.data,
    }
}

fn stopped_status(name: &str) -> ServiceRuntimeStatus {
    ServiceRuntimeStatus {
        name: name.to_string(),
        state: ServiceRuntimeState::Stopped,
        kind: None,
        host: None,
        container_id: None,
        pid: None,
        exit_code: None,
        url: None,
        started_at: None,
        exited_at: None,
        signal: None,
        // A service the daemon has never run has nothing in front of it.
        inspector: None,
    }
}

fn holder_identity(holder: &PortHolder) -> PortHolderIdentity {
    PortHolderIdentity {
        pid: holder.pid,
        pgid: holder.pgid,
        command: holder.command.clone(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use nomoreide_core::log_store::LogStore;
    use uuid::Uuid;

    fn runtime() -> DaemonRuntime {
        let root = std::env::temp_dir().join(format!("nomoreide-runtime-{}", Uuid::new_v4()));
        DaemonRuntime::new(
            ConfigStore::new(root.join("config.json")),
            ProcessManager::new(LogStore::new(root.join("logs"))),
        )
    }

    #[tokio::test]
    async fn cleanup_failure_blocks_starts_allows_stops_and_can_be_retried() {
        let runtime = runtime();

        assert_eq!(
            runtime
                .shutdown_with(async { Err("first cleanup failed".into()) })
                .await,
            Err("first cleanup failed".into())
        );
        assert!(matches!(
            runtime.start_service("missing").await,
            Err(RuntimeMutationError::DaemonCleanupFailed)
        ));
        // A restart ends in a start, so it is gated like one rather than like
        // the remediation stop beside it.
        assert!(matches!(
            runtime.restart_service("missing").await,
            Err(RuntimeMutationError::DaemonCleanupFailed)
        ));
        assert!(matches!(
            runtime.stop_service("missing").await,
            Err(RuntimeMutationError::ServiceNotFound(name)) if name == "missing"
        ));
        assert_eq!(runtime.shutdown_with(async { Ok(()) }).await, Ok(()));
    }

    #[tokio::test]
    async fn shutdown_waits_for_admitted_mutations_before_cleanup() {
        let runtime = Arc::new(runtime());
        let mutation = runtime.mutation_gate.read().await;
        let shutdown_runtime = runtime.clone();
        let shutdown =
            tokio::spawn(async move { shutdown_runtime.shutdown_with(async { Ok(()) }).await });
        tokio::task::yield_now().await;

        assert_eq!(runtime.phase.load(Ordering::Acquire), PHASE_DRAINING);
        assert!(!shutdown.is_finished());
        drop(mutation);
        assert_eq!(shutdown.await.unwrap(), Ok(()));
    }
}