IdeA/crates/application/src/background/mod.rs

//! First-class background command use cases (B8 sub-task 3).
//!
//! These use cases own the *application* side of a command-backed background
//! task: they allocate the [`TaskId`], create the persisted task
//! ([`BackgroundTaskState::Queued`] → [`Running`](BackgroundTaskState::Running))
//! and hand a [`BackgroundTaskSpec`] to the runner. They talk **only** to ports:
//! [`BackgroundTaskStore`], [`BackgroundTaskRunner`], [`Clock`] and
//! [`IdGenerator`], plus the runtime-only [`BackgroundCommandArchive`] abstraction
//! defined below (see its doc for why retry recall is not a domain port).
//!
//! The runner never writes the store; the completion sink does. So beyond the
//! initial create/transition, these use cases do not persist terminal state —
//! except on a *spawn failure*, where they mark the just-created task `Failed` so
//! it never dangles in `Running` with no worker behind it.

use std::collections::{HashSet, VecDeque};
use std::sync::Arc;

use domain::ports::{
    BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskSpec, BackgroundTaskStore, Clock,
    EventBus, IdGenerator, SpawnSpec,
};
use domain::{
    AgentId, BackgroundTask, BackgroundTaskKind, BackgroundTaskRendezvousLink,
    BackgroundTaskResult, BackgroundTaskState, BackgroundTaskWakePolicy, DomainEvent, ProjectId,
    TaskId,
};

use crate::error::AppError;

/// Maps a background-task port error into the application error shape.
fn map_port_err(err: BackgroundTaskPortError) -> AppError {
    match err {
        BackgroundTaskPortError::NotFound => AppError::NotFound("background task".to_owned()),
        BackgroundTaskPortError::AlreadyExists => {
            AppError::Invalid("background task already exists".to_owned())
        }
        BackgroundTaskPortError::Invalid(msg) => AppError::Invalid(msg),
        BackgroundTaskPortError::Runner(msg) => AppError::Process(msg),
        BackgroundTaskPortError::Store(msg) => AppError::Store(msg),
    }
}

/// Input for [`SpawnBackgroundCommand`].
#[derive(Debug, Clone)]
pub struct SpawnBackgroundCommandInput {
    /// Owning project.
    pub project_id: ProjectId,
    /// Agent that owns completion delivery.
    pub owner_agent_id: AgentId,
    /// Human-facing label.
    pub label: String,
    /// Command invocation to run.
    pub command: SpawnSpec,
    /// Completion wake policy.
    pub wake_policy: BackgroundTaskWakePolicy,
    /// Optional composite rendezvous this task participates in.
    pub rendezvous: Option<BackgroundTaskRendezvousLink>,
    /// Optional parent background task. Used for deterministic cancel cascades.
    pub parent_task_id: Option<TaskId>,
    /// Optional absolute deadline, epoch milliseconds.
    pub deadline_ms: Option<u64>,
}

/// Output of [`SpawnBackgroundCommand`] / [`RetryBackgroundTask`].
#[derive(Debug, Clone)]
pub struct SpawnBackgroundCommandOutput {
    /// The created task, in its `Running` state.
    pub task: BackgroundTask,
}

/// Creates a command-backed background task and hands it to the runner.
pub struct SpawnBackgroundCommand {
    store: Arc<dyn BackgroundTaskStore>,
    runner: Arc<dyn BackgroundTaskRunner>,
    clock: Arc<dyn Clock>,
    ids: Arc<dyn IdGenerator>,
    events: Option<Arc<dyn EventBus>>,
}

impl SpawnBackgroundCommand {
    /// Wires the use case to its ports.
    #[must_use]
    pub fn new(
        store: Arc<dyn BackgroundTaskStore>,
        runner: Arc<dyn BackgroundTaskRunner>,
        clock: Arc<dyn Clock>,
        ids: Arc<dyn IdGenerator>,
    ) -> Self {
        Self {
            store,
            runner,
            clock,
            ids,
            events: None,
        }
    }

    /// Attaches the public event stream publisher.
    #[must_use]
    pub fn with_events(mut self, events: Arc<dyn EventBus>) -> Self {
        self.events = Some(events);
        self
    }

    /// Allocates a task id, persists it (`Queued`→`Running`) and spawns it.
    ///
    /// # Errors
    /// [`AppError`] if a store/runner operation or a domain invariant fails.
    pub async fn execute(
        &self,
        input: SpawnBackgroundCommandInput,
    ) -> Result<SpawnBackgroundCommandOutput, AppError> {
        let task = self
            .create_and_run(
                input.project_id,
                input.owner_agent_id,
                input.label,
                input.command,
                input.wake_policy,
                input.rendezvous,
                input.parent_task_id,
                input.deadline_ms,
            )
            .await?;
        Ok(SpawnBackgroundCommandOutput { task })
    }

    /// Shared create/transition/spawn path, reused by retry.
    async fn create_and_run(
        &self,
        project_id: ProjectId,
        owner_agent_id: AgentId,
        label: String,
        command: SpawnSpec,
        wake_policy: BackgroundTaskWakePolicy,
        rendezvous: Option<BackgroundTaskRendezvousLink>,
        parent_task_id: Option<TaskId>,
        deadline_ms: Option<u64>,
    ) -> Result<BackgroundTask, AppError> {
        let task_id = TaskId::from_uuid(self.ids.new_uuid());
        let now = u64::try_from(self.clock.now_millis().max(0)).unwrap_or(0);

        let mut task = BackgroundTask::new(
            task_id,
            project_id,
            owner_agent_id,
            BackgroundTaskKind::Command { label },
            wake_policy,
            now,
            deadline_ms,
        )
        .map_err(|e| AppError::Invalid(e.to_string()))?;
        if let Some(link) = rendezvous {
            task = task.with_rendezvous(link);
        }
        if let Some(parent_task_id) = parent_task_id {
            task = task
                .with_parent_task(parent_task_id)
                .map_err(|e| AppError::Invalid(e.to_string()))?;
        }
        self.store.create(&task).await.map_err(map_port_err)?;

        let running = task
            .transition(BackgroundTaskState::Running, now)
            .map_err(|e| AppError::Invalid(e.to_string()))?;
        self.store.save(&running).await.map_err(map_port_err)?;
        if let Some(events) = &self.events {
            events.publish(DomainEvent::BackgroundTaskStarted {
                project_id,
                task_id,
                owner_agent_id,
            });
            events.publish(DomainEvent::BackgroundTaskStateChanged {
                project_id,
                task_id,
                owner_agent_id,
                state: BackgroundTaskState::Running,
            });
        }

        let spec = BackgroundTaskSpec {
            task_id,
            project_id,
            owner_agent_id,
            kind: running.kind.clone(),
            parent_task_id: running.parent_task_id,
            wake_policy,
            command: Some(command),
            deadline_ms,
        };

        if let Err(err) = self.runner.spawn(spec).await {
            // Never leave the task dangling in `Running` with no worker: mark it
            // Failed so reconcile/UI see a terminal state. Best-effort persist.
            if let Ok(failed) = running.complete(BackgroundTaskResult::Failure {
                finished_at_ms: now,
                exit_code: None,
                error: format!("failed to spawn background command: {err}"),
                stdout_tail: None,
                stderr_tail: None,
            }) {
                let _ = self.store.save(&failed).await;
            }
            if let Some(events) = &self.events {
                events.publish(DomainEvent::BackgroundTaskFailed {
                    project_id,
                    task_id,
                    owner_agent_id,
                    rendezvous: None,
                });
            }
            return Err(map_port_err(err));
        }

        Ok(running)
    }
}

/// Output of [`CancelBackgroundTask`].
#[derive(Debug, Clone)]
pub struct CancelBackgroundTaskOutput {
    /// The task as currently persisted after the cancel request.
    pub task: BackgroundTask,
}

/// Requests cancellation of a running background task.
pub struct CancelBackgroundTask {
    store: Arc<dyn BackgroundTaskStore>,
    runner: Arc<dyn BackgroundTaskRunner>,
    clock: Arc<dyn Clock>,
    events: Option<Arc<dyn EventBus>>,
}

impl CancelBackgroundTask {
    /// Wires the use case to its ports.
    #[must_use]
    pub fn new(
        store: Arc<dyn BackgroundTaskStore>,
        runner: Arc<dyn BackgroundTaskRunner>,
        clock: Arc<dyn Clock>,
    ) -> Self {
        Self {
            store,
            runner,
            clock,
            events: None,
        }
    }

    /// Attaches the public event stream publisher.
    #[must_use]
    pub fn with_events(mut self, events: Arc<dyn EventBus>) -> Self {
        self.events = Some(events);
        self
    }

    /// Cancels `task_id` and its open descendants.
    ///
    /// # Errors
    /// [`AppError`] if the task is missing, terminal without open descendants, or a port fails.
    pub async fn execute(&self, task_id: TaskId) -> Result<CancelBackgroundTaskOutput, AppError> {
        let root = self
            .store
            .get(task_id)
            .await
            .map_err(map_port_err)?
            .ok_or_else(|| AppError::NotFound(format!("background task {task_id}")))?;

        let tasks = self.cancel_subtree_from_task(root.clone()).await?;
        let task = tasks
            .iter()
            .find(|task| task.id == task_id)
            .cloned()
            .unwrap_or(root);
        Ok(CancelBackgroundTaskOutput { task })
    }

    /// Cancels all open tasks rooted at one agent.
    ///
    /// Terminal tasks are never modified. Errors are propagated so command
    /// handlers do not report a silent success when the runner cannot stop a child.
    pub async fn cancel_subtree_for_agent(
        &self,
        project_id: ProjectId,
        agent_id: AgentId,
    ) -> Result<Vec<BackgroundTask>, AppError> {
        let mut roots = self
            .store
            .list_open_for_agent(agent_id)
            .await
            .map_err(map_port_err)?;
        roots.retain(|task| task.project_id == project_id);
        roots.sort_by_key(|task| (task.created_at_ms, task.id));
        self.cancel_subtree(roots).await
    }

    async fn cancel_subtree_from_task(
        &self,
        root: BackgroundTask,
    ) -> Result<Vec<BackgroundTask>, AppError> {
        let mut roots = if root.is_terminal() {
            Vec::new()
        } else {
            vec![root.clone()]
        };
        roots.extend(
            self.store
                .list_children(root.id)
                .await
                .map_err(map_port_err)?,
        );
        if roots.is_empty() {
            return Err(AppError::Invalid(format!(
                "background task {} is already terminal ({:?})",
                root.id, root.state
            )));
        }
        self.cancel_subtree(roots).await
    }

    async fn cancel_subtree(
        &self,
        roots: Vec<BackgroundTask>,
    ) -> Result<Vec<BackgroundTask>, AppError> {
        let mut queue: VecDeque<BackgroundTask> = roots.into();
        let mut visited = HashSet::new();
        let mut cancelled = Vec::new();

        while let Some(task) = queue.pop_front() {
            if !visited.insert(task.id) {
                continue;
            }
            let mut children = self
                .store
                .list_children(task.id)
                .await
                .map_err(map_port_err)?;
            children.sort_by_key(|task| (task.created_at_ms, task.id));
            queue.extend(children);
            if task.is_terminal() {
                continue;
            }
            cancelled.push(self.cancel_one(task).await?);
        }

        cancelled.sort_by_key(|task| (task.created_at_ms, task.id));
        Ok(cancelled)
    }

    async fn cancel_one(&self, task: BackgroundTask) -> Result<BackgroundTask, AppError> {
        let task_id = task.id;

        let cancelling = if task.state == BackgroundTaskState::Cancelling {
            task
        } else {
            let now = u64::try_from(self.clock.now_millis().max(0)).unwrap_or(0);
            let cancelling = task
                .transition(BackgroundTaskState::Cancelling, now)
                .map_err(|e| AppError::Invalid(e.to_string()))?;
            self.store.save(&cancelling).await.map_err(map_port_err)?;
            self.publish_state_changed(&cancelling);
            cancelling
        };

        if let Err(err) = self.runner.cancel(task_id).await {
            let now = u64::try_from(self.clock.now_millis().max(0)).unwrap_or(0);
            let failed = cancelling
                .complete(BackgroundTaskResult::CancelFailed {
                    finished_at_ms: now,
                    error: format!("failed to cancel background task: {err}"),
                })
                .map_err(|e| AppError::Invalid(e.to_string()))?;
            self.store.save(&failed).await.map_err(map_port_err)?;
            self.publish_state_changed(&failed);
            return Err(map_port_err(err));
        }

        Ok(self
            .store
            .get(task_id)
            .await
            .map_err(map_port_err)?
            .unwrap_or(cancelling))
    }

    fn publish_state_changed(&self, task: &BackgroundTask) {
        if let Some(events) = &self.events {
            events.publish(DomainEvent::BackgroundTaskStateChanged {
                project_id: task.project_id,
                task_id: task.id,
                owner_agent_id: task.owner_agent_id,
                state: task.state,
            });
        }
    }
}

/// Read-only recall of the command invocation a background task was spawned with.
///
/// This is a **runtime, in-memory** abstraction, *not* a domain port: the recall
/// lives wherever the runner keeps its live invocations for the current session
/// and is deliberately **never persisted** (a [`SpawnSpec`] can carry secrets and
/// must never land in `.ideai/background-tasks/*.json`, which travels with the
/// project). Retry is therefore **session-scoped** in V1 — a task whose runner no
/// longer remembers its spec (e.g. after a restart) simply cannot be retried.
///
/// The persisted [`BackgroundTask`] carries no [`SpawnSpec`] (the command line is
/// an execution detail, not domain state). Retry, however, must re-run the *same*
/// command under a fresh task id, so it recovers the original invocation from
/// whatever component still holds it — the runner. Keeping that recall behind this
/// application-level trait lets [`RetryBackgroundTask`] stay decoupled from the
/// concrete runner without dragging the concept into the domain.
pub trait BackgroundCommandArchive: Send + Sync {
    /// Returns the [`SpawnSpec`] a command-backed task was spawned with, if the
    /// runner still remembers it. `None` when the task is unknown, was not
    /// command-backed, or is no longer in the current session's registry.
    fn spec_for(&self, task_id: TaskId) -> Option<SpawnSpec>;
}

/// Re-runs a terminal command-backed task under a brand-new task id.
///
/// The original command is recovered from the [`BackgroundCommandArchive`]; the
/// deadline is intentionally dropped (an absolute past deadline would expire the
/// retry instantly).
pub struct RetryBackgroundTask {
    store: Arc<dyn BackgroundTaskStore>,
    archive: Arc<dyn BackgroundCommandArchive>,
    spawn: Arc<SpawnBackgroundCommand>,
}

impl RetryBackgroundTask {
    /// Wires the use case to its ports and the spawn use case.
    #[must_use]
    pub fn new(
        store: Arc<dyn BackgroundTaskStore>,
        archive: Arc<dyn BackgroundCommandArchive>,
        spawn: Arc<SpawnBackgroundCommand>,
    ) -> Self {
        Self {
            store,
            archive,
            spawn,
        }
    }

    /// Retries `task_id`, returning the freshly-created task.
    ///
    /// # Errors
    /// [`AppError::NotFound`] if the task is unknown, [`AppError::Invalid`] if it
    /// is not a terminal command task or its command is no longer available.
    pub async fn execute(&self, task_id: TaskId) -> Result<SpawnBackgroundCommandOutput, AppError> {
        let old = self
            .store
            .get(task_id)
            .await
            .map_err(map_port_err)?
            .ok_or_else(|| AppError::NotFound(format!("background task {task_id}")))?;

        if !old.is_terminal() {
            return Err(AppError::Invalid(
                "cannot retry a task that is still running".to_owned(),
            ));
        }
        let label = match &old.kind {
            BackgroundTaskKind::Command { label } => label.clone(),
            _ => {
                return Err(AppError::Invalid(
                    "only command-backed tasks can be retried".to_owned(),
                ))
            }
        };
        let command = self.archive.spec_for(task_id).ok_or_else(|| {
            AppError::Invalid("original command is no longer available for retry".to_owned())
        })?;

        let task = self
            .spawn
            .create_and_run(
                old.project_id,
                old.owner_agent_id,
                label,
                command,
                old.wake_policy,
                None,
                None,
                None,
            )
            .await?;
        Ok(SpawnBackgroundCommandOutput { task })
    }
}