diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index ca8c828..2b982dc 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -30,7 +30,8 @@ use domain::ports::PtyHandle; use crate::dto::{ parse_agent_id, parse_close_terminal, parse_delete_profile, parse_layout_id, parse_memory_slug, parse_node_id, parse_profile_id, parse_project_id, parse_session_id, parse_skill_id, - parse_template_id, parse_ticket_id, AgentDriftListDto, AgentDto, AgentListDto, + parse_task_id, parse_template_id, parse_ticket_id, AgentDriftListDto, AgentDto, AgentListDto, + BackgroundTaskDto, AssignSkillRequestDto, AttachLiveAgentRequestDto, AttachLiveAgentResponseDto, ChangeAgentProfileDto, ChangeAgentProfileRequestDto, ConfigureProfilesRequestDto, ConversationDetailsDto, CreateAgentFromTemplateRequestDto, CreateAgentRequestDto, @@ -2555,3 +2556,160 @@ pub async fn resolve_memory_links( .map(MemoryLinksDto::from) .map_err(ErrorDto::from) } + +// --------------------------------------------------------------------------- +// Background tasks (B8 — first-class background command) +// --------------------------------------------------------------------------- + +/// Maps a background-task port error to the frontend [`ErrorDto`] shape. +fn background_error(err: domain::ports::BackgroundTaskPortError) -> ErrorDto { + use domain::ports::BackgroundTaskPortError as E; + let code = match &err { + E::NotFound => "NOT_FOUND", + E::AlreadyExists | E::Invalid(_) => "INVALID", + E::Runner(_) => "PROCESS", + E::Store(_) => "STORE", + }; + ErrorDto { + code: code.to_owned(), + message: err.to_string(), + } +} + +/// `spawn_background_command` — start a command-backed first-class background +/// task whose completion is delivered to the owning agent's inbox. +/// +/// # Errors +/// Returns an [`ErrorDto`] (`INVALID` for a malformed id/path, `PROCESS` if the +/// command cannot be spawned, `STORE` on persistence failure). +#[tauri::command] +pub async fn spawn_background_command( + request: crate::dto::SpawnBackgroundCommandRequestDto, + state: State<'_, AppState>, +) -> Result { + let project_id = parse_project_id(&request.project_id)?; + let owner_agent_id = parse_agent_id(&request.owner_agent_id)?; + let cwd = domain::project::ProjectPath::new(request.cwd.clone()).map_err(|_| ErrorDto { + code: "INVALID".to_owned(), + message: format!("invalid working directory: {}", request.cwd), + })?; + let command = domain::ports::SpawnSpec { + command: request.command, + args: request.args, + cwd, + env: request.env, + context_plan: None, + sandbox: None, + }; + let wake_policy = if request.record_only { + domain::BackgroundTaskWakePolicy::RecordOnly + } else { + domain::BackgroundTaskWakePolicy::WakeOwner + }; + state + .spawn_background_command + .execute(application::SpawnBackgroundCommandInput { + project_id, + owner_agent_id, + label: request.label, + command, + wake_policy, + deadline_ms: request.deadline_ms, + }) + .await + .map(|out| BackgroundTaskDto::from(out.task)) + .map_err(ErrorDto::from) +} + +/// `cancel_background_task` — request cancellation of a running background task. +/// +/// The terminal `Cancelled` state is written by the completion sink; this returns +/// the task as currently persisted (absent if it no longer exists). +/// +/// # Errors +/// Returns an [`ErrorDto`] (`INVALID` for a malformed id, `PROCESS`/`STORE` on +/// failure). +#[tauri::command] +pub async fn cancel_background_task( + task_id: String, + state: State<'_, AppState>, +) -> Result, ErrorDto> { + let id = parse_task_id(&task_id)?; + state + .cancel_background_task + .execute(id) + .await + .map(|out| out.task.map(BackgroundTaskDto::from)) + .map_err(ErrorDto::from) +} + +/// `retry_background_task` — re-run a terminal command task under a **new** task +/// id (the original id is never reused). +/// +/// # Errors +/// Returns an [`ErrorDto`] (`INVALID` for a malformed id or a non-retryable task, +/// `NOT_FOUND` if the task is unknown, `PROCESS`/`STORE` on failure). +#[tauri::command] +pub async fn retry_background_task( + task_id: String, + state: State<'_, AppState>, +) -> Result { + let id = parse_task_id(&task_id)?; + state + .retry_background_task + .execute(id) + .await + .map(|out| BackgroundTaskDto::from(out.task)) + .map_err(ErrorDto::from) +} + +/// `list_background_tasks` — read the background-task read-model for a project, +/// optionally narrowed to one owning agent. +/// +/// Returns the union of the agent's open tasks and undelivered completions, +/// filtered to `projectId`. Without `agentId`, only the project's undelivered +/// completions are returned (the store cannot enumerate open tasks project-wide). +/// +/// # Errors +/// Returns an [`ErrorDto`] (`INVALID` for a malformed id, `STORE` on failure). +#[tauri::command] +pub async fn list_background_tasks( + project_id: String, + agent_id: Option, + state: State<'_, AppState>, +) -> Result, ErrorDto> { + let project_id = parse_project_id(&project_id)?; + let agent_id = match &agent_id { + Some(raw) => Some(parse_agent_id(raw)?), + None => None, + }; + + let store = &state.background_task_store; + let mut by_id: std::collections::HashMap = + std::collections::HashMap::new(); + + if let Some(agent_id) = agent_id { + for task in store + .list_open_for_agent(agent_id) + .await + .map_err(background_error)? + { + by_id.insert(task.id, task); + } + } + for task in store + .list_undelivered_completions() + .await + .map_err(background_error)? + { + by_id.entry(task.id).or_insert(task); + } + + let mut tasks: Vec = by_id + .into_values() + .filter(|task| task.project_id == project_id) + .filter(|task| agent_id.map_or(true, |a| task.owner_agent_id == a)) + .collect(); + tasks.sort_by_key(|task| (task.created_at_ms, task.id)); + Ok(tasks.into_iter().map(BackgroundTaskDto::from).collect()) +} diff --git a/crates/app-tauri/src/dto.rs b/crates/app-tauri/src/dto.rs index 664c0cb..55f4849 100644 --- a/crates/app-tauri/src/dto.rs +++ b/crates/app-tauri/src/dto.rs @@ -2866,3 +2866,167 @@ mod ls6_pagination_tests { assert!(dto.turns.is_empty() && !dto.has_more && dto.next_anchor.is_none()); } } + +// --------------------------------------------------------------------------- +// Background tasks (B8 — first-class background command) +// --------------------------------------------------------------------------- + +use domain::{ + BackgroundTask, BackgroundTaskKind, BackgroundTaskResult, BackgroundTaskState, TaskId, +}; + +/// One first-class background task as seen by the frontend (camelCase wire shape). +/// +/// Mirrors the persisted [`BackgroundTask`], flattening the terminal +/// [`BackgroundTaskResult`] into optional `exitCode` / `summary` / tails so the +/// UI has a single shape for every lifecycle state. Optional fields are omitted +/// from the wire when absent (`skip_serializing_if`). +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct BackgroundTaskDto { + /// Stable task id (UUID string). + pub task_id: String, + /// Owning agent id (UUID string). + pub owner_agent_id: String, + /// Owning project id (UUID string). + pub project_id: String, + /// Kind discriminant (`"command"`, `"headlessRendezvous"`, `"sessionResume"`, + /// `"maintenance"`). + pub kind: String, + /// Lifecycle state (`"queued"`, `"running"`, `"waiting"`, `"completed"`, + /// `"failed"`, `"cancelled"`, `"expired"`). + pub state: String, + /// Process exit code, when the terminal result carries one. + #[serde(skip_serializing_if = "Option::is_none")] + pub exit_code: Option, + /// Human-readable summary / error / reason of the terminal result. + #[serde(skip_serializing_if = "Option::is_none")] + pub summary: Option, + /// Bounded stdout tail (merged output for PTY-backed commands). + #[serde(skip_serializing_if = "Option::is_none")] + pub stdout_tail: Option, + /// Bounded stderr tail (unset for PTY-backed commands, which merge streams). + #[serde(skip_serializing_if = "Option::is_none")] + pub stderr_tail: Option, + /// Creation timestamp, epoch milliseconds. + pub created_at_ms: u64, + /// Last update timestamp, epoch milliseconds. + pub updated_at_ms: u64, +} + +/// Kind discriminant string for a [`BackgroundTaskKind`]. +fn background_kind_label(kind: &BackgroundTaskKind) -> &'static str { + match kind { + BackgroundTaskKind::Command { .. } => "command", + BackgroundTaskKind::HeadlessRendezvous { .. } => "headlessRendezvous", + BackgroundTaskKind::SessionResume { .. } => "sessionResume", + BackgroundTaskKind::Maintenance { .. } => "maintenance", + } +} + +/// Lifecycle state string for a [`BackgroundTaskState`]. +fn background_state_label(state: BackgroundTaskState) -> &'static str { + match state { + BackgroundTaskState::Queued => "queued", + BackgroundTaskState::Running => "running", + BackgroundTaskState::Waiting => "waiting", + BackgroundTaskState::Completed => "completed", + BackgroundTaskState::Failed => "failed", + BackgroundTaskState::Cancelled => "cancelled", + BackgroundTaskState::Expired => "expired", + } +} + +impl From for BackgroundTaskDto { + fn from(task: BackgroundTask) -> Self { + let (exit_code, summary, stdout_tail, stderr_tail) = match &task.result { + Some(BackgroundTaskResult::Success { + exit_code, + summary, + stdout_tail, + stderr_tail, + .. + }) => ( + *exit_code, + Some(summary.clone()), + stdout_tail.clone(), + stderr_tail.clone(), + ), + Some(BackgroundTaskResult::Failure { + exit_code, + error, + stdout_tail, + stderr_tail, + .. + }) => ( + *exit_code, + Some(error.clone()), + stdout_tail.clone(), + stderr_tail.clone(), + ), + Some(BackgroundTaskResult::Cancelled { reason, .. }) + | Some(BackgroundTaskResult::Expired { reason, .. }) => { + (None, Some(reason.clone()), None, None) + } + None => (None, None, None, None), + }; + Self { + task_id: task.id.to_string(), + owner_agent_id: task.owner_agent_id.to_string(), + project_id: task.project_id.to_string(), + kind: background_kind_label(&task.kind).to_owned(), + state: background_state_label(task.state).to_owned(), + exit_code, + summary, + stdout_tail, + stderr_tail, + created_at_ms: task.created_at_ms, + updated_at_ms: task.updated_at_ms, + } + } +} + +/// Parses a task-id string (UUID) coming from the frontend. +/// +/// # Errors +/// Returns an [`ErrorDto`] with code `INVALID` if the string is not a UUID. +pub fn parse_task_id(raw: &str) -> Result { + uuid::Uuid::parse_str(raw) + .map(TaskId::from_uuid) + .map_err(|_| ErrorDto { + code: "INVALID".to_owned(), + message: format!("invalid task id: {raw}"), + }) +} + +/// Request DTO for `spawn_background_command`. +/// +/// Builds a command-backed [`BackgroundTask`]: the command line runs under the +/// project host's PTY and its completion is delivered to `ownerAgentId`. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SpawnBackgroundCommandRequestDto { + /// Owning project id (UUID string). + pub project_id: String, + /// Agent that owns completion delivery (UUID string). + pub owner_agent_id: String, + /// Human-facing label. + pub label: String, + /// Executable to run. + pub command: String, + /// Arguments. + #[serde(default)] + pub args: Vec, + /// Working directory (absolute path). + pub cwd: String, + /// Extra environment variables. + #[serde(default)] + pub env: Vec<(String, String)>, + /// When `true` (the default), wake the owner on completion; otherwise only + /// record it. + #[serde(default)] + pub record_only: bool, + /// Optional absolute deadline, epoch milliseconds. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub deadline_ms: Option, +} diff --git a/crates/app-tauri/src/lib.rs b/crates/app-tauri/src/lib.rs index 371fcfe..b149f91 100644 --- a/crates/app-tauri/src/lib.rs +++ b/crates/app-tauri/src/lib.rs @@ -214,6 +214,10 @@ pub fn run() { commands::recall_memory, commands::resolve_memory_links, commands::move_tab_to_new_window, + commands::spawn_background_command, + commands::cancel_background_task, + commands::retry_background_task, + commands::list_background_tasks, ]) .run(tauri::generate_context!()) .expect("error while running IdeA Tauri application"); diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index e9004aa..acd28ce 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -13,7 +13,8 @@ use std::sync::{Arc, Mutex}; use application::{ AgentResumer, AgentWakeService, AppError, AssignSkillToAgent, AttachLiveAgent, - ChangeAgentProfile, CheckEmbedderSuggestion, CloseProject, CloseTab, CloseTerminal, + BackgroundCommandArchive, CancelBackgroundTask, ChangeAgentProfile, CheckEmbedderSuggestion, + CloseProject, CloseTab, CloseTerminal, RetryBackgroundTask, SpawnBackgroundCommand, ConfigureProfiles, ContextGuardUseCases, CreateAgentFromScratch, CreateAgentFromTemplate, CreateLayout, CreateMemory, CreateProject, CreateSkill, CreateTemplate, DeleteAgent, DeleteEmbedderProfile, DeleteLayout, DeleteMemory, DeleteProfile, DeleteSkill, DeleteTemplate, @@ -39,7 +40,8 @@ use application::{ use async_trait::async_trait; use domain::ports::{ AgentContextStore, AgentRuntime, AgentSession, AgentSessionFactory, AgentWakePort, - BackgroundTaskPortError, BackgroundTaskStore, Clock, Embedder, EmbedderEnvInspector, + BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskStore, + Clock, Embedder, EmbedderEnvInspector, EmbedderProfileStore, EmbedderPromptStore, EventBus, FileSystem, GitPort, IdGenerator, MemoryRecall, MemoryStore, PermissionStore, ProcessSpawner, ProfileStore, ProjectStore, PtyPort, ScheduledTask, Scheduler, SkillStore, TemplateStore, WakeError, WakeReason, @@ -57,8 +59,9 @@ use serde_json::{json, Map, Value}; use uuid::Uuid; use infrastructure::{ - embedder_from_profile, AdaptiveMemoryRecall, BackgroundTaskReadyToDeliver, - ClaudePermissionProjector, ClaudeTranscriptInspector, CliAgentRuntime, + embedder_from_profile, AdaptiveMemoryRecall, BackgroundCompletionSink, + BackgroundTaskReadyToDeliver, ClaudePermissionProjector, ClaudeTranscriptInspector, + CliAgentRuntime, CommandBackgroundRunner, CodexPermissionProjector, EmbedderEnvProbe, FsBackgroundTaskStore, FsConversationLog, FsEmbedderProfileStore, FsEmbedderPromptStore, FsHandoffStore, FsLiveStateStore, FsMemoryStore, FsOrchestratorWatcher, FsPermissionStore, FsProfileStore, FsProjectStore, @@ -795,6 +798,15 @@ pub struct AppState { // --- Windows (L10) --- /// Detach a tab into a new OS window (persists the workspace topology). pub move_tab: Arc, + // --- Background tasks (B8) --- + /// Spawn a command-backed first-class background task. + pub spawn_background_command: Arc, + /// Cancel a running background task. + pub cancel_background_task: Arc, + /// Retry a terminal command task under a fresh task id. + pub retry_background_task: Arc, + /// Store handle used by `list_background_tasks` to read the task read-model. + pub background_task_store: Arc, // --- Templates & sync (L7) --- /// Create a template in the global store. pub create_template: Arc, @@ -1541,6 +1553,49 @@ impl AppState { let background_tasks_port = Arc::clone(&background_tasks) as Arc; let (background_ready_tx, mut background_ready_rx) = tokio::sync::mpsc::unbounded_channel::(); + + // --- B8 : runner de commandes + fermeture de la boucle du sink --- + // Le runner concret exécute les tâches command-backed sur le `PtyPort` composé + // (local ici ; SSH/WSL via `RemoteHost` quand ils atterriront — Liskov). Il + // émet UNE complétion par tâche ; le sink (single-writer) persiste l'état + // terminal AVANT de signaler la livraison (persist-avant-signal), puis le pont + // ready→inbox ci-dessous réveille l'agent propriétaire. + let background_runner = Arc::new(CommandBackgroundRunner::new( + Arc::clone(&pty_port), + Arc::clone(&clock) as Arc, + )); + let background_runner_port = + Arc::clone(&background_runner) as Arc; + { + let sink = Arc::new(BackgroundCompletionSink::new( + Arc::clone(&background_tasks_port), + background_ready_tx.clone(), + )); + let runner = Arc::clone(&background_runner_port); + // `start_from_runner` exige un runtime Tokio ambiant (`Handle::current`) : + // on l'appelle donc DANS la tâche async. Le `JoinHandle` du drain est gardé + // vivant en l'attendant (il ne se termine qu'à la fermeture du flux de + // complétions), ce qui maintient sink + runner en vie pour la session. + tauri::async_runtime::spawn(async move { + let drain = sink.start_from_runner(runner); + let _ = drain.await; + }); + } + let spawn_background_command = Arc::new(SpawnBackgroundCommand::new( + Arc::clone(&background_tasks_port), + Arc::clone(&background_runner_port), + Arc::clone(&clock) as Arc, + Arc::clone(&ids) as Arc, + )); + let cancel_background_task = Arc::new(CancelBackgroundTask::new( + Arc::clone(&background_tasks_port), + Arc::clone(&background_runner_port), + )); + let retry_background_task = Arc::new(RetryBackgroundTask::new( + Arc::clone(&background_tasks_port), + Arc::clone(&background_runner) as Arc, + Arc::clone(&spawn_background_command), + )); let background_wake = Arc::new(AgentWakeService::new( Arc::clone(&mediated_inbox) as Arc, Arc::clone(&input_mediator), @@ -1977,6 +2032,10 @@ impl AppState { fs_port: Arc::clone(&fs_port), home_dir, move_tab, + spawn_background_command, + cancel_background_task, + retry_background_task, + background_task_store: Arc::clone(&background_tasks_port), } } diff --git a/crates/application/src/background/mod.rs b/crates/application/src/background/mod.rs new file mode 100644 index 0000000..32f6a97 --- /dev/null +++ b/crates/application/src/background/mod.rs @@ -0,0 +1,300 @@ +//! 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::sync::Arc; + +use domain::ports::{ + BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskSpec, BackgroundTaskStore, Clock, + IdGenerator, SpawnSpec, +}; +use domain::{ + AgentId, BackgroundTask, BackgroundTaskKind, BackgroundTaskResult, BackgroundTaskState, + BackgroundTaskWakePolicy, 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 absolute deadline, epoch milliseconds. + pub deadline_ms: Option, +} + +/// 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, + runner: Arc, + clock: Arc, + ids: Arc, +} + +impl SpawnBackgroundCommand { + /// Wires the use case to its ports. + #[must_use] + pub fn new( + store: Arc, + runner: Arc, + clock: Arc, + ids: Arc, + ) -> Self { + Self { + store, + runner, + clock, + ids, + } + } + + /// 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 { + let task = self + .create_and_run( + input.project_id, + input.owner_agent_id, + input.label, + input.command, + input.wake_policy, + 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, + deadline_ms: Option, + ) -> Result { + 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 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()))?; + 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)?; + + let spec = BackgroundTaskSpec { + task_id, + project_id, + owner_agent_id, + kind: running.kind.clone(), + 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; + } + return Err(map_port_err(err)); + } + + Ok(running) + } +} + +/// Output of [`CancelBackgroundTask`]. +#[derive(Debug, Clone)] +pub struct CancelBackgroundTaskOutput { + /// The task as currently persisted, if it still exists. The terminal + /// `Cancelled` state is written by the completion sink, not here. + pub task: Option, +} + +/// Requests cancellation of a running background task. +pub struct CancelBackgroundTask { + store: Arc, + runner: Arc, +} + +impl CancelBackgroundTask { + /// Wires the use case to its ports. + #[must_use] + pub fn new(store: Arc, runner: Arc) -> Self { + Self { store, runner } + } + + /// Signals the runner to cancel `task_id` (idempotent). + /// + /// # Errors + /// [`AppError`] if the runner or store fails. + pub async fn execute( + &self, + task_id: TaskId, + ) -> Result { + self.runner.cancel(task_id).await.map_err(map_port_err)?; + let task = self.store.get(task_id).await.map_err(map_port_err)?; + Ok(CancelBackgroundTaskOutput { task }) + } +} + +/// 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; +} + +/// 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, + archive: Arc, + spawn: Arc, +} + +impl RetryBackgroundTask { + /// Wires the use case to its ports and the spawn use case. + #[must_use] + pub fn new( + store: Arc, + archive: Arc, + spawn: Arc, + ) -> 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 { + 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, + ) + .await?; + Ok(SpawnBackgroundCommandOutput { task }) + } +} diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index 86a1c09..e0133b9 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -12,6 +12,7 @@ #![warn(missing_docs)] pub mod agent; +pub mod background; pub mod conversation; pub mod diag; pub mod embedder; @@ -47,6 +48,10 @@ pub use agent::{ StructuredSessionDescriptor, TurnOutcome, UpdateAgentContext, UpdateAgentContextInput, AGENT_MEMORY_RECALL_BUDGET, CODEX_SUBMIT_DELAY_MS, LIVE_STATE_INJECT_MAX, RESUME_PROMPT, }; +pub use background::{ + BackgroundCommandArchive, CancelBackgroundTask, CancelBackgroundTaskOutput, RetryBackgroundTask, + SpawnBackgroundCommand, SpawnBackgroundCommandInput, SpawnBackgroundCommandOutput, +}; pub use conversation::{ ConversationArchiveProvider, ReadConversationPage, ReadConversationPageInput, RecordTurn, RotateConversationLog, RotateConversationLogInput, TurnPage, TurnSource, TurnView, diff --git a/crates/domain/src/lib.rs b/crates/domain/src/lib.rs index e41caf2..17266e4 100644 --- a/crates/domain/src/lib.rs +++ b/crates/domain/src/lib.rs @@ -167,9 +167,9 @@ pub use orchestrator::{ }; pub use ports::{ - AgentContextStore, AgentRuntime, BackgroundCompletionStream, BackgroundTaskCompletion, - BackgroundTaskHandle, BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskSpec, - BackgroundTaskStore, Clock, ContextInjectionPlan, DirEntry, Embedder, EmbedderEnvInspector, + AgentContextStore, AgentRuntime, BackgroundCompletionStream, + BackgroundTaskCompletion, BackgroundTaskHandle, BackgroundTaskPortError, BackgroundTaskRunner, + BackgroundTaskSpec, BackgroundTaskStore, Clock, ContextInjectionPlan, DirEntry, Embedder, EmbedderEnvInspector, EmbedderEnvReport, EmbedderError, EmbedderProfileStore, EmbedderPromptDismissal, EmbedderPromptStore, EventBus, EventStream, ExitStatus, FileSystem, FsError, GitCommitInfo, GitError, GitFileStatus, GitPort, GraphCommit, IdGenerator, LiveStateStore, MemoryError, diff --git a/crates/infrastructure/src/background_task/mod.rs b/crates/infrastructure/src/background_task/mod.rs new file mode 100644 index 0000000..a95c3c0 --- /dev/null +++ b/crates/infrastructure/src/background_task/mod.rs @@ -0,0 +1,22 @@ +//! First-class background task infrastructure. +//! +//! Three cohesive pieces close the B2→B8 loop: +//! +//! - [`sink`] — the durable [`BackgroundCompletionSink`], single writer between a +//! runner completion and the ready-to-deliver signal (persist-before-signal). +//! - [`tail`] — a UTF-8-safe bounded ring buffer for stdout/stderr tails. +//! - [`runner`] — [`CommandBackgroundRunner`], the concrete +//! [`BackgroundTaskRunner`](domain::ports::BackgroundTaskRunner) that spawns a +//! command-backed task over a composed [`PtyPort`](domain::ports::PtyPort) and +//! emits exactly one completion per task. + +mod runner; +mod sink; +mod tail; + +pub use runner::CommandBackgroundRunner; +pub use sink::{ + start_background_ready_inbox_bridge, BackgroundCompletionSink, BackgroundCompletionSinkError, + BackgroundCompletionSinkOutcome, BackgroundReadyInboxBridgeHandle, BackgroundTaskReadyToDeliver, +}; +pub use tail::{bounded_tail, tail_cap_bytes, BoundedTail}; diff --git a/crates/infrastructure/src/background_task/runner.rs b/crates/infrastructure/src/background_task/runner.rs new file mode 100644 index 0000000..0c7c4b9 --- /dev/null +++ b/crates/infrastructure/src/background_task/runner.rs @@ -0,0 +1,308 @@ +//! [`CommandBackgroundRunner`] — the concrete command-backed background task +//! runner (B8 sub-task 2). +//! +//! # Role +//! +//! This is the `impl` of the frozen [`BackgroundTaskRunner`] port that the +//! composition root wires under the [`BackgroundCompletionSink`]. It spawns a +//! command-backed task over a **composed** [`PtyPort`] (resolved by the project's +//! `RemoteHost`, so local/SSH/WSL are transparent — Liskov) and, for each task, +//! emits **exactly one** [`BackgroundTaskCompletion`] on the stream returned by +//! [`subscribe_completions`](BackgroundTaskRunner::subscribe_completions). +//! +//! It writes **neither** the store **nor** the inbox: the sink is the single +//! writer that persists the terminal state before signalling delivery +//! (persist-before-signal). The runner's only outward effect is the completion +//! event. +//! +//! # Lifecycle of one task +//! +//! `spawn` opens the PTY and starts a detached worker. The worker watches the +//! PTY's output stream for EOF (natural process exit), a cancel notification, or +//! the deadline, whichever comes first. It then snapshots the bounded output tail +//! from the PTY scrollback, reaps the real exit status via +//! [`PtyPort::kill`](domain::ports::PtyPort::kill), maps that to a terminal +//! [`BackgroundTaskResult`], and sends the completion. +//! +//! # Output tail & the live tee +//! +//! A PTY **merges** stdout and stderr onto one stream, so the completion carries +//! the merged output in `stdout_tail` and leaves `stderr_tail` empty. Exit +//! detection reads the PTY's single-consumer output subscription; a *live tee* +//! that re-subscribes the same PTY (to render it in an xterm cell) would supersede +//! the runner's subscription and break EOF detection — teeing a running command +//! therefore requires a multi-consumer PTY broadcast, which is out of scope here. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::{self, Receiver, Sender}; +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; + +use application::BackgroundCommandArchive; +use domain::ports::{ + BackgroundCompletionStream, BackgroundTaskCompletion, BackgroundTaskHandle, + BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskSpec, Clock, PtyHandle, PtyPort, + SpawnSpec, +}; +use domain::terminal::PtySize; +use domain::{BackgroundTaskResult, TaskId}; + +use super::tail::{bounded_tail, tail_cap_bytes}; + +/// Default PTY geometry for a headless background command (no visible cell). +const DEFAULT_SIZE: PtySize = PtySize { rows: 24, cols: 80 }; + +/// Why the worker's terminal wait resolved. +enum WaitOutcome { + /// The process exited on its own (stream hit EOF). + Exited, + /// A [`cancel`](BackgroundTaskRunner::cancel) was requested. + Cancelled, + /// The deadline elapsed before exit. + Expired, +} + +/// Live control handles for one running task (kept in the registry). +#[derive(Clone)] +struct TaskControl { + pty_handle: PtyHandle, + cancel: Arc, + cancelled: Arc, +} + +/// Command-backed [`BackgroundTaskRunner`] over a composed [`PtyPort`]. +pub struct CommandBackgroundRunner { + pty: Arc, + clock: Arc, + completion_tx: Sender, + completion_rx: Mutex>>, + running: Arc>>, + /// Retained command invocations, so [`BackgroundCommandArchive::spec_for`] + /// can recover a task's command for retry. + specs: Arc>>, +} + +impl CommandBackgroundRunner { + /// Builds a runner over a composed PTY port and clock. + #[must_use] + pub fn new(pty: Arc, clock: Arc) -> Self { + let (completion_tx, completion_rx) = mpsc::channel(); + Self { + pty, + clock, + completion_tx, + completion_rx: Mutex::new(Some(completion_rx)), + running: Arc::new(Mutex::new(HashMap::new())), + specs: Arc::new(Mutex::new(HashMap::new())), + } + } + + fn now_ms(&self) -> u64 { + u64::try_from(self.clock.now_millis().max(0)).unwrap_or(0) + } + + /// Detached worker driving one command to its single completion. + #[allow(clippy::too_many_arguments)] + async fn run_to_completion( + pty: Arc, + clock: Arc, + completion_tx: Sender, + running: Arc>>, + task_id: TaskId, + control: TaskControl, + deadline_ms: Option, + ) { + let outcome = Self::await_terminal(&pty, &control, deadline_ms, &clock).await; + // A cancel that raced a natural EOF still wins: the user asked to cancel. + let outcome = if control.cancelled.load(Ordering::SeqCst) { + WaitOutcome::Cancelled + } else { + outcome + }; + + // Snapshot the merged output tail *before* kill tears the session down. + let tail = pty + .scrollback(&control.pty_handle) + .ok() + .map(|bytes| bounded_tail(&bytes, tail_cap_bytes())) + .filter(|s| !s.is_empty()); + + // Reap the real exit status (works for a natural exit and a forced kill). + let exit_code = pty + .kill(&control.pty_handle) + .await + .ok() + .and_then(|status| status.code); + + let finished_at_ms = u64::try_from(clock.now_millis().max(0)).unwrap_or(0); + let result = match outcome { + WaitOutcome::Cancelled => BackgroundTaskResult::Cancelled { + finished_at_ms, + reason: "cancelled by request".to_owned(), + }, + WaitOutcome::Expired => BackgroundTaskResult::Expired { + finished_at_ms, + reason: "deadline elapsed before completion".to_owned(), + }, + WaitOutcome::Exited => { + if exit_code == Some(0) { + BackgroundTaskResult::Success { + finished_at_ms, + exit_code, + summary: "command completed successfully".to_owned(), + stdout_tail: tail, + stderr_tail: None, + } + } else { + BackgroundTaskResult::Failure { + finished_at_ms, + exit_code, + error: match exit_code { + Some(code) => format!("command exited with code {code}"), + None => "command terminated without an exit code".to_owned(), + }, + stdout_tail: tail, + stderr_tail: None, + } + } + } + }; + + running.lock().expect("runner registry poisoned").remove(&task_id); + let _ = completion_tx.send(BackgroundTaskCompletion { task_id, result }); + } + + /// Resolves when the process exits, is cancelled, or hits its deadline. + async fn await_terminal( + pty: &Arc, + control: &TaskControl, + deadline_ms: Option, + clock: &Arc, + ) -> WaitOutcome { + let eof = Self::wait_for_eof(Arc::clone(pty), control.pty_handle.clone()); + tokio::pin!(eof); + + let sleep_ms = deadline_ms.map(|deadline| { + let now = u64::try_from(clock.now_millis().max(0)).unwrap_or(0); + deadline.saturating_sub(now) + }); + + match sleep_ms { + Some(ms) => tokio::select! { + () = &mut eof => WaitOutcome::Exited, + () = control.cancel.notified() => WaitOutcome::Cancelled, + () = tokio::time::sleep(std::time::Duration::from_millis(ms)) => WaitOutcome::Expired, + }, + None => tokio::select! { + () = &mut eof => WaitOutcome::Exited, + () = control.cancel.notified() => WaitOutcome::Cancelled, + }, + } + } + + /// Resolves when the PTY output stream ends (the process has exited). + /// + /// The domain `OutputStream` is a blocking iterator, so it is drained on a + /// blocking task and its bytes discarded — the persisted tail is read from the + /// PTY scrollback instead, which the reader thread fills regardless of + /// subscribers. If the handle is already unknown, resolves immediately so the + /// worker proceeds to reap. + async fn wait_for_eof(pty: Arc, handle: PtyHandle) { + let Ok(stream) = pty.subscribe_output(&handle) else { + return; + }; + let _ = tokio::task::spawn_blocking(move || { + for _chunk in stream {} + }) + .await; + } +} + +#[async_trait] +impl BackgroundTaskRunner for CommandBackgroundRunner { + async fn spawn( + &self, + spec: BackgroundTaskSpec, + ) -> Result { + let command = spec.command.clone().ok_or_else(|| { + BackgroundTaskPortError::Runner( + "command-backed runner requires a command spec".to_owned(), + ) + })?; + + let pty_handle = self + .pty + .spawn(command.clone(), DEFAULT_SIZE) + .await + .map_err(|e| BackgroundTaskPortError::Runner(format!("pty spawn failed: {e}")))?; + + self.specs + .lock() + .expect("runner spec registry poisoned") + .insert(spec.task_id, command); + + let control = TaskControl { + pty_handle, + cancel: Arc::new(tokio::sync::Notify::new()), + cancelled: Arc::new(AtomicBool::new(false)), + }; + self.running + .lock() + .expect("runner registry poisoned") + .insert(spec.task_id, control.clone()); + + tokio::spawn(Self::run_to_completion( + Arc::clone(&self.pty), + Arc::clone(&self.clock), + self.completion_tx.clone(), + Arc::clone(&self.running), + spec.task_id, + control, + spec.deadline_ms, + )); + + // Touch `now_ms` only to keep the clock wired for future scheduling; the + // authoritative timestamps come from the worker at completion. + let _ = self.now_ms(); + Ok(BackgroundTaskHandle { + task_id: spec.task_id, + }) + } + + async fn cancel(&self, task_id: TaskId) -> Result<(), BackgroundTaskPortError> { + let control = self + .running + .lock() + .expect("runner registry poisoned") + .get(&task_id) + .cloned(); + // Unknown task ⇒ already terminal or never ours: cancellation is idempotent. + if let Some(control) = control { + control.cancelled.store(true, Ordering::SeqCst); + control.cancel.notify_one(); + } + Ok(()) + } + + fn subscribe_completions(&self) -> BackgroundCompletionStream { + let rx = self + .completion_rx + .lock() + .expect("runner completion receiver poisoned") + .take() + .expect("subscribe_completions must be called exactly once"); + Box::new(rx.into_iter()) + } +} + +impl BackgroundCommandArchive for CommandBackgroundRunner { + fn spec_for(&self, task_id: TaskId) -> Option { + self.specs + .lock() + .expect("runner spec registry poisoned") + .get(&task_id) + .cloned() + } +} diff --git a/crates/infrastructure/src/background_task.rs b/crates/infrastructure/src/background_task/sink.rs similarity index 100% rename from crates/infrastructure/src/background_task.rs rename to crates/infrastructure/src/background_task/sink.rs diff --git a/crates/infrastructure/src/background_task/tail.rs b/crates/infrastructure/src/background_task/tail.rs new file mode 100644 index 0000000..cb62822 --- /dev/null +++ b/crates/infrastructure/src/background_task/tail.rs @@ -0,0 +1,149 @@ +//! UTF-8-safe bounded output tail (B8 sub-task 1). +//! +//! A background command can print megabytes; the completion contract only keeps a +//! bounded *tail* (the most recent bytes). [`BoundedTail`] is a byte ring buffer +//! capped at construction; [`bounded_tail`] renders the retained bytes as a +//! `String`, trimming any partial UTF-8 sequence at the front so a multi-byte +//! character never gets sliced by the cap. + +use std::collections::VecDeque; + +use domain::background_task::BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES; + +/// Default tail cap when `IDEA_BG_TAIL_BYTES` is unset (8 KiB). +const DEFAULT_TAIL_BYTES: usize = 8 * 1024; +/// Floor so a hostile/typo'd env value can't disable the tail entirely. +const MIN_TAIL_BYTES: usize = 1024; + +/// Resolves the tail cap in bytes from `IDEA_BG_TAIL_BYTES`. +/// +/// Defaults to 8 KiB, floored at 1 KiB, and clamped to the domain maximum +/// ([`BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES`], 16 KiB) so the persisted result +/// always validates. +#[must_use] +pub fn tail_cap_bytes() -> usize { + std::env::var("IDEA_BG_TAIL_BYTES") + .ok() + .and_then(|v| v.trim().parse::().ok()) + .unwrap_or(DEFAULT_TAIL_BYTES) + .clamp(MIN_TAIL_BYTES, BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES) +} + +/// A byte ring buffer that retains only the last `cap` bytes pushed into it. +#[derive(Debug)] +pub struct BoundedTail { + buf: VecDeque, + cap: usize, +} + +impl BoundedTail { + /// Builds a tail retaining at most `cap` bytes (a `cap` of `0` retains + /// nothing but never panics). + #[must_use] + pub fn with_cap(cap: usize) -> Self { + Self { + buf: VecDeque::new(), + cap, + } + } + + /// Appends a chunk, dropping the oldest bytes past the cap. + pub fn push(&mut self, chunk: &[u8]) { + if self.cap == 0 { + return; + } + // If the incoming chunk alone exceeds the cap, only its tail can survive. + let start = chunk.len().saturating_sub(self.cap); + self.buf.extend(&chunk[start..]); + let overflow = self.buf.len().saturating_sub(self.cap); + if overflow > 0 { + self.buf.drain(0..overflow); + } + } + + /// Returns whether any byte is retained. + #[must_use] + pub fn is_empty(&self) -> bool { + self.buf.is_empty() + } + + /// Renders the retained bytes as a UTF-8 string, dropping a leading partial + /// multi-byte sequence left dangling by the cap. Returns `None` when empty. + #[must_use] + pub fn to_tail_string(&self) -> Option { + if self.buf.is_empty() { + return None; + } + let bytes: Vec = self.buf.iter().copied().collect(); + Some(bounded_tail(&bytes, self.cap)) + } +} + +/// Renders the last `cap` bytes of `bytes` as a lossless-at-the-boundary +/// `String`. +/// +/// The slice is taken from the end, then any partial UTF-8 sequence at the *front* +/// (a continuation byte with no lead byte, because the cap cut mid-character) is +/// skipped so the result is valid UTF-8 without replacement characters at the +/// seam. Invalid bytes in the interior are still replaced lossily. +#[must_use] +pub fn bounded_tail(bytes: &[u8], cap: usize) -> String { + let start = bytes.len().saturating_sub(cap); + let mut slice = &bytes[start..]; + // Only realign when we actually truncated the front (start > 0): otherwise a + // legitimately leading continuation byte is just invalid input, handled below. + if start > 0 { + let mut skip = 0; + while skip < slice.len() && is_utf8_continuation(slice[skip]) { + skip += 1; + } + slice = &slice[skip..]; + } + String::from_utf8_lossy(slice).into_owned() +} + +/// Whether `b` is a UTF-8 continuation byte (`10xxxxxx`). +const fn is_utf8_continuation(b: u8) -> bool { + b & 0b1100_0000 == 0b1000_0000 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn keeps_only_the_last_cap_bytes() { + let mut tail = BoundedTail::with_cap(4); + tail.push(b"abcdef"); + assert_eq!(tail.to_tail_string().as_deref(), Some("cdef")); + } + + #[test] + fn push_bigger_than_cap_keeps_tail() { + let mut tail = BoundedTail::with_cap(3); + tail.push(b"0123456789"); + assert_eq!(tail.to_tail_string().as_deref(), Some("789")); + } + + #[test] + fn empty_tail_is_none() { + let tail = BoundedTail::with_cap(8); + assert!(tail.is_empty()); + assert_eq!(tail.to_tail_string(), None); + } + + #[test] + fn cut_multibyte_char_is_trimmed_at_front() { + // "é" is 0xC3 0xA9. Cap of 1 would keep only the trailing 0xA9 (a lone + // continuation byte); it must be dropped, yielding an empty string. + let bytes = "é".as_bytes(); + assert_eq!(bounded_tail(bytes, 1), ""); + // Cap of 2 keeps the whole character. + assert_eq!(bounded_tail(bytes, 2), "é"); + } + + #[test] + fn interior_is_preserved() { + assert_eq!(bounded_tail("héllo".as_bytes(), 100), "héllo"); + } +} diff --git a/crates/infrastructure/src/lib.rs b/crates/infrastructure/src/lib.rs index 5445934..7cc6d3b 100644 --- a/crates/infrastructure/src/lib.rs +++ b/crates/infrastructure/src/lib.rs @@ -38,9 +38,9 @@ pub mod store; pub mod timeparse; pub use background_task::{ - start_background_ready_inbox_bridge, BackgroundCompletionSink, BackgroundCompletionSinkError, - BackgroundCompletionSinkOutcome, BackgroundReadyInboxBridgeHandle, - BackgroundTaskReadyToDeliver, + bounded_tail, start_background_ready_inbox_bridge, tail_cap_bytes, BackgroundCompletionSink, + BackgroundCompletionSinkError, BackgroundCompletionSinkOutcome, BackgroundReadyInboxBridgeHandle, + BackgroundTaskReadyToDeliver, BoundedTail, CommandBackgroundRunner, }; pub use clock::SystemClock; pub use conversation::InMemoryConversationRegistry;