diff --git a/crates/infrastructure/src/background_task.rs b/crates/infrastructure/src/background_task.rs new file mode 100644 index 0000000..66183c0 --- /dev/null +++ b/crates/infrastructure/src/background_task.rs @@ -0,0 +1,233 @@ +//! Background task completion sink. +//! +//! The sink is the durable boundary between a runner completion and later +//! mailbox/wake work: it writes the terminal task state to the +//! [`BackgroundTaskStore`](domain::ports::BackgroundTaskStore) first, then emits +//! a lightweight "ready to deliver" signal. It never wakes an agent and never +//! enqueues mailbox items; B4/B5 consume the ready signal. + +use std::collections::HashSet; +use std::sync::Arc; + +use thiserror::Error; +use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender}; +use tokio::sync::Mutex; +use tokio::task::JoinHandle; + +use domain::ports::{BackgroundTaskCompletion, BackgroundTaskRunner, BackgroundTaskStore}; +use domain::{ + AgentInbox, BackgroundTaskPortError, InboxError, InboxItem, InboxItemKind, InboxReceiptStatus, + InboxSource, ProjectId, TaskId, +}; + +/// Signal emitted only after a completion has been persisted. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BackgroundTaskReadyToDeliver { + /// Persisted task id. + pub task_id: TaskId, + /// Owning project, copied from the persisted task. + pub project_id: ProjectId, + /// Agent that owns the completion delivery. + pub owner_agent_id: domain::AgentId, +} + +/// Outcome of processing one completion event. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum BackgroundCompletionSinkOutcome { + /// Completion was persisted and signalled as ready to deliver. + PersistedAndSignalled(BackgroundTaskReadyToDeliver), + /// Task was already terminal, so the duplicate completion was ignored. + IgnoredAlreadyTerminal { + /// Ignored task id. + task_id: TaskId, + }, + /// Another completion for this task was already processed by this sink. + IgnoredDuplicate { + /// Ignored task id. + task_id: TaskId, + }, + /// The task no longer exists in the store. + IgnoredMissingTask { + /// Ignored task id. + task_id: TaskId, + }, +} + +/// Errors raised by the completion sink. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum BackgroundCompletionSinkError { + /// Store operation failed. + #[error("background completion store failed: {0}")] + Store(#[from] BackgroundTaskPortError), + /// The ready-to-deliver channel is closed. + #[error("background completion ready signal channel is closed")] + ReadySignalClosed, +} + +/// Consumes runner completions, persists terminal state, then signals delivery. +pub struct BackgroundCompletionSink { + store: Arc, + ready: UnboundedSender, + processed: Mutex>, +} + +/// Handle for the ready-to-inbox bridge task. +pub type BackgroundReadyInboxBridgeHandle = JoinHandle<()>; + +impl BackgroundCompletionSink { + /// Builds a sink from a task store and ready-to-deliver channel. + #[must_use] + pub fn new( + store: Arc, + ready: UnboundedSender, + ) -> Self { + Self { + store, + ready, + processed: Mutex::new(HashSet::new()), + } + } + + /// Starts consuming [`BackgroundTaskRunner::subscribe_completions`]. + /// + /// The domain completion stream is a blocking iterator, so consumption runs on + /// a blocking task and re-enters the current Tokio runtime for persistence. + /// + /// # Panics + /// Panics if called outside a Tokio runtime. + #[must_use] + pub fn start_from_runner( + self: Arc, + runner: Arc, + ) -> JoinHandle<()> { + let mut stream = runner.subscribe_completions(); + let runtime = tokio::runtime::Handle::current(); + tokio::task::spawn_blocking(move || { + while let Some(completion) = stream.next() { + let _ = runtime.block_on(self.process_completion(completion)); + } + }) + } + + /// Processes a single completion event. + /// + /// Order is strict: `store.save(terminal_task)` must succeed before the ready + /// signal is sent. If persistence fails, no ready signal is emitted and the + /// task id is released so a later retry can persist it. + /// + /// # Errors + /// [`BackgroundCompletionSinkError`] if the store fails, the task transition + /// is invalid, or the ready channel is closed after persistence. + pub async fn process_completion( + &self, + completion: BackgroundTaskCompletion, + ) -> Result { + { + let mut processed = self.processed.lock().await; + if processed.contains(&completion.task_id) { + return Ok(BackgroundCompletionSinkOutcome::IgnoredDuplicate { + task_id: completion.task_id, + }); + } + processed.insert(completion.task_id); + } + + let outcome = self.persist_and_signal(completion.clone()).await; + if outcome.is_err() { + self.processed.lock().await.remove(&completion.task_id); + } + outcome + } + + async fn persist_and_signal( + &self, + completion: BackgroundTaskCompletion, + ) -> Result { + let Some(task) = self.store.get(completion.task_id).await? else { + return Ok(BackgroundCompletionSinkOutcome::IgnoredMissingTask { + task_id: completion.task_id, + }); + }; + if task.is_terminal() { + return Ok(BackgroundCompletionSinkOutcome::IgnoredAlreadyTerminal { + task_id: completion.task_id, + }); + } + + let terminal = task.complete(completion.result).map_err(|e| { + BackgroundCompletionSinkError::Store(BackgroundTaskPortError::Invalid(e.to_string())) + })?; + self.store.save(&terminal).await?; + + let ready = BackgroundTaskReadyToDeliver { + task_id: terminal.id, + project_id: terminal.project_id, + owner_agent_id: terminal.owner_agent_id, + }; + self.ready + .send(ready.clone()) + .map_err(|_| BackgroundCompletionSinkError::ReadySignalClosed)?; + Ok(BackgroundCompletionSinkOutcome::PersistedAndSignalled( + ready, + )) + } +} + +/// Starts the B3→B4 bridge: persisted completions become inbox items. +/// +/// Overflow of completion/system items is intentionally non-fatal: the completion +/// is already durable and remains delivery-pending for boot reconcile. +#[must_use] +pub fn start_background_ready_inbox_bridge( + mut ready: UnboundedReceiver, + inbox: Arc, +) -> BackgroundReadyInboxBridgeHandle { + tokio::spawn(async move { + while let Some(ready) = ready.recv().await { + let item = InboxItem { + id: domain::TicketId::new_random(), + agent_id: ready.owner_agent_id, + source: InboxSource::BackgroundTask { + task_id: ready.task_id, + }, + kind: InboxItemKind::BackgroundCompletion, + body: format!("Background task {} completed.", ready.task_id), + created_at_ms: now_ms(), + correlation_id: Some(ready.task_id.to_string()), + }; + match inbox.enqueue_message(ready.owner_agent_id, item) { + Ok(receipt) if receipt.status == InboxReceiptStatus::Deferred => { + application::diag!( + "[background-task] completion deferred: task={} owner={} queue_depth={}", + ready.task_id, + ready.owner_agent_id, + receipt.depth + ); + } + Ok(_) => {} + Err(InboxError::InboxFull { .. }) => { + application::diag!( + "[background-task] completion inbox full but durable: task={} owner={}", + ready.task_id, + ready.owner_agent_id + ); + } + Err(err) => { + application::diag!( + "[background-task] completion inbox enqueue failed: task={} owner={} err={err}", + ready.task_id, + ready.owner_agent_id + ); + } + } + } + }) +} + +fn now_ms() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX)) + .unwrap_or(0) +} diff --git a/crates/infrastructure/src/input/mod.rs b/crates/infrastructure/src/input/mod.rs index 1638c5c..ab1b44b 100644 --- a/crates/infrastructure/src/input/mod.rs +++ b/crates/infrastructure/src/input/mod.rs @@ -26,8 +26,12 @@ use std::time::Duration; use domain::events::DomainEvent; use domain::ids::AgentId; +use domain::inbox::{ + AgentInbox, AgentInboxSnapshot, InboxError, InboxItem, InboxReceipt, InboxReceiptStatus, + InboxSource, DEFAULT_AGENT_INBOX_CAPACITY, +}; use domain::input::{AgentBusyState, AgentLiveness, InputMediator, SubmitConfig}; -use domain::mailbox::{AgentMailbox, PendingReply, Ticket, TicketId}; +use domain::mailbox::{AgentMailbox, AgentQueueSnapshot, PendingReply, Ticket, TicketId}; use domain::ports::{EventBus, PtyHandle, PtyPort}; use crate::mailbox::InMemoryMailbox; @@ -496,6 +500,11 @@ pub struct MediatedInbox { /// `set_stall_threshold` and consumed to arm a fresh liveness window on the enqueue /// that starts a turn. Absent ⇒ `None` (no stall detection — legacy behaviour). stall: Mutex>>, + /// Rich inbox payloads keyed by the mailbox ticket id. The mailbox remains the + /// only FIFO; this map is metadata only. + inbox_items: Mutex>, + /// Bounded inbox capacity per agent. + inbox_capacity: usize, /// Fenêtre de grâce (G) propagée au [`BusyTracker`] à chaque (re)construction. /// Prod = [`PROMPT_READY_GRACE`] (2 s) ; surchargée par [`Self::with_grace`] en test. grace: Duration, @@ -525,6 +534,8 @@ impl MediatedInbox { front_owned, submit: Mutex::new(HashMap::new()), stall: Mutex::new(HashMap::new()), + inbox_items: Mutex::new(HashMap::new()), + inbox_capacity: configured_inbox_capacity(), grace: PROMPT_READY_GRACE, } } @@ -716,6 +727,8 @@ impl MediatedInbox { front_owned, submit: Mutex::new(HashMap::new()), stall: Mutex::new(HashMap::new()), + inbox_items: Mutex::new(HashMap::new()), + inbox_capacity: configured_inbox_capacity(), grace: PROMPT_READY_GRACE, } } @@ -729,6 +742,13 @@ impl MediatedInbox { ) } + /// Overrides bounded inbox capacity. Intended for tests and explicit wiring. + #[must_use] + pub fn with_inbox_capacity(mut self, capacity: usize) -> Self { + self.inbox_capacity = capacity; + self + } + fn handles(&self) -> std::sync::MutexGuard<'_, HashMap> { self.handles .lock() @@ -747,6 +767,12 @@ impl MediatedInbox { .unwrap_or_else(std::sync::PoisonError::into_inner) } + fn inbox_items(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.inbox_items + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + /// **Détection de stagnation** (lot 2) : balaie les agents `Busy` et bascule /// `Alive→Stalled` ceux dont le dernier battement remonte à plus de /// `stall_after_ms`. Émet `AgentLivenessChanged` **une fois par transition**. La @@ -1023,6 +1049,107 @@ impl InputMediator for MediatedInbox { } } +impl AgentInbox for MediatedInbox { + fn enqueue_message(&self, agent: AgentId, item: InboxItem) -> Result { + if item.agent_id != agent { + return Err(InboxError::AgentMismatch { + agent_id: agent, + item_agent_id: item.agent_id, + }); + } + + let depth_before = self.mailbox.pending(&agent); + if depth_before >= self.inbox_capacity { + if item.is_lossless_system() { + return Ok(InboxReceipt { + item_id: item.id, + agent_id: agent, + depth: depth_before, + status: InboxReceiptStatus::Deferred, + }); + } + return Err(InboxError::InboxFull { + agent_id: agent, + capacity: self.inbox_capacity, + }); + } + + let item_id = item.id; + let ticket = inbox_item_to_ticket(&item); + self.inbox_items().insert(item_id, item); + let _pending = match ticket.source { + domain::input::InputSource::Human => self.enqueue(agent, ticket), + domain::input::InputSource::Agent { .. } => self.enqueue_silent(agent, ticket), + }; + let depth = self.mailbox.pending(&agent); + if let Some(events) = &self.tracker.events { + events.publish(DomainEvent::AgentInboxQueued { + agent_id: agent, + depth, + }); + } + Ok(InboxReceipt { + item_id, + agent_id: agent, + depth, + status: InboxReceiptStatus::Queued, + }) + } + + fn dequeue_next(&self, agent: AgentId) -> Option { + let head = self.mailbox.head_ticket(&agent)?; + let item = self.inbox_items().remove(&head)?; + self.mailbox.cancel_head(agent, head); + if let Some(events) = &self.tracker.events { + events.publish(DomainEvent::AgentInboxDrained { + agent_id: agent, + depth: self.snapshot(agent).depth, + }); + } + Some(item) + } + + fn snapshot(&self, agent: AgentId) -> AgentInboxSnapshot { + let items_by_id = self.inbox_items(); + let items: Vec = self + .mailbox + .queue_for(agent) + .into_iter() + .filter_map(|ticket| items_by_id.get(&ticket.id).cloned()) + .collect(); + AgentInboxSnapshot { + agent_id: agent, + depth: self.mailbox.pending(&agent), + items, + } + } +} + +fn configured_inbox_capacity() -> usize { + std::env::var("IDEA_AGENT_INBOX_CAPACITY") + .ok() + .and_then(|raw| raw.parse::().ok()) + .filter(|capacity| *capacity > 0) + .unwrap_or(DEFAULT_AGENT_INBOX_CAPACITY) +} + +fn inbox_item_to_ticket(item: &InboxItem) -> Ticket { + let conversation = domain::conversation::ConversationId::from_uuid(uuid::Uuid::nil()); + match item.source { + InboxSource::Agent { agent_id } => Ticket::from_agent( + item.id, + agent_id, + conversation, + agent_id.to_string(), + item.body.clone(), + ), + InboxSource::Human => Ticket::from_human(item.id, conversation, "vous", item.body.clone()), + InboxSource::BackgroundTask { .. } | InboxSource::System => { + Ticket::from_human(item.id, conversation, "IdeA", item.body.clone()) + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/infrastructure/src/lib.rs b/crates/infrastructure/src/lib.rs index e19f9e2..5445934 100644 --- a/crates/infrastructure/src/lib.rs +++ b/crates/infrastructure/src/lib.rs @@ -12,6 +12,7 @@ #![forbid(unsafe_code)] #![warn(missing_docs)] +pub mod background_task; pub mod clock; pub mod conversation; pub mod conversation_log; @@ -36,6 +37,11 @@ pub mod session; pub mod store; pub mod timeparse; +pub use background_task::{ + start_background_ready_inbox_bridge, BackgroundCompletionSink, BackgroundCompletionSinkError, + BackgroundCompletionSinkOutcome, BackgroundReadyInboxBridgeHandle, + BackgroundTaskReadyToDeliver, +}; pub use clock::SystemClock; pub use conversation::InMemoryConversationRegistry; pub use conversation_log::{ @@ -74,9 +80,10 @@ pub use store::OnnxEmbedder; pub use store::{detect_ollama, HttpEmbedder, DEFAULT_LOCAL_EMBED_ENDPOINT}; pub use store::{ embedder_from_profile, index_token_size, onnx_model_is_cached, should_use_vector, - AdaptiveMemoryRecall, EmbedderEnvProbe, FsEmbedderProfileStore, FsEmbedderPromptStore, - FsLiveStateStore, FsMemoryStore, FsPermissionStore, FsProfileStore, FsProjectStore, - FsSkillStore, FsTemplateStore, HashEmbedder, IdeaiContextStore, NaiveMemoryRecall, - OnnxModelInfo, StubEmbedder, VectorMemoryRecall, DEFAULT_OLLAMA_BASE_URL, ONNX_CACHE_SUBDIR, - RECOMMENDED_ONNX_MODELS, VECTOR_HTTP_ENABLED, VECTOR_ONNX_ENABLED, + AdaptiveMemoryRecall, BackgroundTaskReconcileReport, EmbedderEnvProbe, FsBackgroundTaskStore, + FsEmbedderProfileStore, FsEmbedderPromptStore, FsLiveStateStore, FsMemoryStore, + FsPermissionStore, FsProfileStore, FsProjectStore, FsSkillStore, FsTemplateStore, HashEmbedder, + IdeaiContextStore, NaiveMemoryRecall, OnnxModelInfo, StubEmbedder, VectorMemoryRecall, + DEFAULT_OLLAMA_BASE_URL, ONNX_CACHE_SUBDIR, RECOMMENDED_ONNX_MODELS, VECTOR_HTTP_ENABLED, + VECTOR_ONNX_ENABLED, }; diff --git a/crates/infrastructure/src/store/background_task.rs b/crates/infrastructure/src/store/background_task.rs new file mode 100644 index 0000000..a691114 --- /dev/null +++ b/crates/infrastructure/src/store/background_task.rs @@ -0,0 +1,390 @@ +//! [`FsBackgroundTaskStore`] — durable store for first-class background tasks. +//! +//! Persistence is segmented by project id under the project root: +//! +//! ```text +//! /.ideai/background-tasks/.json +//! ``` +//! +//! Each file is a small JSON document `{ version, projectId, tasks }`. Writes are +//! atomic: serialize to `.json.tmp`, then rename over the target. The +//! in-memory registry indexes open tasks by owner agent and task id; it is +//! rebuilt lazily from disk on first use and updated on every successful mutation. +//! +//! Boot reconcile rule for B2: because the runner registry is not wired yet, a +//! task in `Running` or `Waiting` whose id is not present in the caller-provided +//! live handle list is marked `Failed` with a synthetic restart-loss error and +//! `completion_delivered = false`. Existing terminal tasks that are not delivered +//! are only reported as delivery-pending. `Queued` tasks are left unchanged. + +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; +use std::path::{Path, PathBuf}; + +use async_trait::async_trait; +use serde::{Deserialize, Serialize}; +use tokio::sync::RwLock; + +use domain::{ + AgentId, BackgroundTask, BackgroundTaskPortError, BackgroundTaskResult, BackgroundTaskState, + BackgroundTaskStore, ProjectId, ProjectPath, TaskId, +}; + +const IDEAI_DIR: &str = ".ideai"; +const BACKGROUND_TASKS_DIR: &str = "background-tasks"; +const TASK_DOC_VERSION: u32 = 1; + +/// File-backed implementation of [`BackgroundTaskStore`]. +pub struct FsBackgroundTaskStore { + dir: PathBuf, + registry: RwLock, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +struct RegistryIndex { + loaded: bool, + task_to_project: HashMap, + open_by_agent: HashMap>, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +struct StoreState { + docs: BTreeMap, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct TaskDoc { + version: u32, + project_id: ProjectId, + tasks: Vec, +} + +/// Summary returned by [`FsBackgroundTaskStore::reconcile_boot`]. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct BackgroundTaskReconcileReport { + /// Open tasks converted to `Failed` because no live runtime handle exists. + pub failed_task_ids: Vec, + /// Terminal tasks whose completion remains to be delivered. + pub delivery_pending_task_ids: Vec, +} + +impl FsBackgroundTaskStore { + /// Builds the store for a project root. + #[must_use] + pub fn new(root: &ProjectPath) -> Self { + let dir = PathBuf::from(root.as_str()) + .join(IDEAI_DIR) + .join(BACKGROUND_TASKS_DIR); + Self { + dir, + registry: RwLock::new(RegistryIndex::default()), + } + } + + /// `/.ideai/background-tasks`. + #[must_use] + pub fn dir(&self) -> &Path { + &self.dir + } + + /// Returns the open task ids known by the in-memory registry for `agent_id`. + /// + /// This is intentionally a registry view: it does not read the filesystem. + pub async fn registry_open_task_ids_for_agent(&self, agent_id: AgentId) -> Vec { + self.registry + .read() + .await + .open_by_agent + .get(&agent_id) + .map(|tasks| tasks.keys().copied().collect()) + .unwrap_or_default() + } + + /// Reconciles persisted tasks against live runtime handles at boot. + /// + /// B2 rule: `Running`/`Waiting` tasks missing from `live_task_ids` are marked + /// `Failed` with a synthetic restart-loss failure and become delivery-pending. + /// Already-terminal undelivered tasks are reported as delivery-pending. `Queued` + /// tasks are left open. + /// + /// # Errors + /// [`BackgroundTaskPortError`] on I/O, serialization or invalid transition. + pub async fn reconcile_boot( + &self, + live_task_ids: &[TaskId], + now_ms: u64, + ) -> Result { + let live: HashSet = live_task_ids.iter().copied().collect(); + let mut state = self.read_all_docs().await?; + let mut report = BackgroundTaskReconcileReport::default(); + let mut changed_projects = BTreeSet::new(); + + for (project_id, doc) in &mut state.docs { + for task in &mut doc.tasks { + match task.state { + BackgroundTaskState::Running | BackgroundTaskState::Waiting + if !live.contains(&task.id) => + { + let finished_at_ms = now_ms.max(task.updated_at_ms); + let result = BackgroundTaskResult::Failure { + finished_at_ms, + exit_code: None, + error: "background task lost its runtime handle during IdeA restart" + .into(), + stdout_tail: None, + stderr_tail: None, + }; + *task = task.complete(result).map_err(invalid_task)?; + report.failed_task_ids.push(task.id); + report.delivery_pending_task_ids.push(task.id); + changed_projects.insert(*project_id); + } + _ if task.has_pending_completion_delivery() => { + report.delivery_pending_task_ids.push(task.id); + } + _ => {} + } + } + } + + for project_id in changed_projects { + if let Some(doc) = state.docs.get(&project_id) { + self.write_doc(doc).await?; + } + } + self.replace_registry_from_state(&state).await; + Ok(report) + } + + async fn ensure_registry_loaded(&self) -> Result<(), BackgroundTaskPortError> { + if self.registry.read().await.loaded { + return Ok(()); + } + let state = self.read_all_docs().await?; + self.replace_registry_from_state(&state).await; + Ok(()) + } + + async fn replace_registry_from_state(&self, state: &StoreState) { + let mut registry = self.registry.write().await; + *registry = RegistryIndex::from_state(state); + } + + fn path_for_project(&self, project_id: ProjectId) -> PathBuf { + self.dir.join(format!("{project_id}.json")) + } + + fn tmp_path_for_project(&self, project_id: ProjectId) -> PathBuf { + self.dir.join(format!("{project_id}.json.tmp")) + } + + async fn read_all_docs(&self) -> Result { + let mut state = StoreState::default(); + let mut entries = match tokio::fs::read_dir(&self.dir).await { + Ok(entries) => entries, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(state), + Err(e) => return Err(io_error(e)), + }; + + while let Some(entry) = entries.next_entry().await.map_err(io_error)? { + let path = entry.path(); + if path.extension().and_then(|ext| ext.to_str()) != Some("json") { + continue; + } + let doc = self.read_doc_path(&path).await?; + state.docs.insert(doc.project_id, doc); + } + Ok(state) + } + + async fn read_doc(&self, project_id: ProjectId) -> Result { + let path = self.path_for_project(project_id); + match self.read_doc_path(&path).await { + Ok(doc) => Ok(doc), + Err(BackgroundTaskPortError::NotFound) => Ok(TaskDoc { + version: TASK_DOC_VERSION, + project_id, + tasks: Vec::new(), + }), + Err(e) => Err(e), + } + } + + async fn read_doc_path(&self, path: &Path) -> Result { + match tokio::fs::read(path).await { + Ok(bytes) => serde_json::from_slice(&bytes) + .map_err(|e| BackgroundTaskPortError::Store(e.to_string())), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + Err(BackgroundTaskPortError::NotFound) + } + Err(e) => Err(io_error(e)), + } + } + + async fn write_doc(&self, doc: &TaskDoc) -> Result<(), BackgroundTaskPortError> { + tokio::fs::create_dir_all(&self.dir) + .await + .map_err(io_error)?; + let bytes = serde_json::to_vec_pretty(doc) + .map_err(|e| BackgroundTaskPortError::Store(e.to_string()))?; + let tmp = self.tmp_path_for_project(doc.project_id); + tokio::fs::write(&tmp, &bytes).await.map_err(io_error)?; + tokio::fs::rename(&tmp, self.path_for_project(doc.project_id)) + .await + .map_err(io_error)?; + Ok(()) + } + + async fn find_task_project( + &self, + task_id: TaskId, + ) -> Result, BackgroundTaskPortError> { + self.ensure_registry_loaded().await?; + Ok(self + .registry + .read() + .await + .task_to_project + .get(&task_id) + .copied()) + } + + async fn update_registry_for_task(&self, task: &BackgroundTask) { + let mut registry = self.registry.write().await; + registry.loaded = true; + registry.task_to_project.insert(task.id, task.project_id); + for tasks in registry.open_by_agent.values_mut() { + tasks.remove(&task.id); + } + if !task.is_terminal() { + registry + .open_by_agent + .entry(task.owner_agent_id) + .or_default() + .insert(task.id, task.clone()); + } + } +} + +impl RegistryIndex { + fn from_state(state: &StoreState) -> Self { + let mut registry = Self { + loaded: true, + task_to_project: HashMap::new(), + open_by_agent: HashMap::new(), + }; + for doc in state.docs.values() { + for task in &doc.tasks { + registry.task_to_project.insert(task.id, task.project_id); + if !task.is_terminal() { + registry + .open_by_agent + .entry(task.owner_agent_id) + .or_default() + .insert(task.id, task.clone()); + } + } + } + registry + } +} + +#[async_trait] +impl BackgroundTaskStore for FsBackgroundTaskStore { + async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> { + self.ensure_registry_loaded().await?; + if self + .registry + .read() + .await + .task_to_project + .contains_key(&task.id) + { + return Err(BackgroundTaskPortError::AlreadyExists); + } + + let mut doc = self.read_doc(task.project_id).await?; + if doc.tasks.iter().any(|existing| existing.id == task.id) { + return Err(BackgroundTaskPortError::AlreadyExists); + } + doc.tasks.push(task.clone()); + self.write_doc(&doc).await?; + self.update_registry_for_task(task).await; + Ok(()) + } + + async fn get(&self, id: TaskId) -> Result, BackgroundTaskPortError> { + let Some(project_id) = self.find_task_project(id).await? else { + return Ok(None); + }; + let doc = self.read_doc(project_id).await?; + Ok(doc.tasks.into_iter().find(|task| task.id == id)) + } + + async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> { + self.ensure_registry_loaded().await?; + let mut doc = self.read_doc(task.project_id).await?; + if let Some(slot) = doc.tasks.iter_mut().find(|existing| existing.id == task.id) { + *slot = task.clone(); + } else { + return Err(BackgroundTaskPortError::NotFound); + } + self.write_doc(&doc).await?; + self.update_registry_for_task(task).await; + Ok(()) + } + + async fn list_open_for_agent( + &self, + agent_id: AgentId, + ) -> Result, BackgroundTaskPortError> { + self.ensure_registry_loaded().await?; + let registry = self.registry.read().await; + let mut tasks: Vec = registry + .open_by_agent + .get(&agent_id) + .map(|tasks| tasks.values().cloned().collect()) + .unwrap_or_default(); + tasks.sort_by_key(|task| (task.created_at_ms, task.id)); + Ok(tasks) + } + + async fn list_undelivered_completions( + &self, + ) -> Result, BackgroundTaskPortError> { + let mut tasks: Vec = self + .read_all_docs() + .await? + .docs + .into_values() + .flat_map(|doc| doc.tasks) + .filter(BackgroundTask::has_pending_completion_delivery) + .collect(); + tasks.sort_by_key(|task| (task.updated_at_ms, task.id)); + Ok(tasks) + } + + async fn mark_completion_delivered( + &self, + task_id: TaskId, + ) -> Result<(), BackgroundTaskPortError> { + let task = self + .get(task_id) + .await? + .ok_or(BackgroundTaskPortError::NotFound)?; + let delivered = match task.mark_completion_delivered() { + Ok(delivered) => delivered, + Err(domain::BackgroundTaskError::CompletionAlreadyDelivered) => return Ok(()), + Err(e) => return Err(invalid_task(e)), + }; + self.save(&delivered).await + } +} + +fn io_error(error: std::io::Error) -> BackgroundTaskPortError { + BackgroundTaskPortError::Store(error.to_string()) +} + +fn invalid_task(error: domain::BackgroundTaskError) -> BackgroundTaskPortError { + BackgroundTaskPortError::Invalid(error.to_string()) +} diff --git a/crates/infrastructure/src/store/mod.rs b/crates/infrastructure/src/store/mod.rs index 3d9204f..24300e3 100644 --- a/crates/infrastructure/src/store/mod.rs +++ b/crates/infrastructure/src/store/mod.rs @@ -4,6 +4,7 @@ //! the known-projects **registry** and the **workspace** are stored as plain //! JSON files in the app data directory (machine-local, outside any project). +mod background_task; mod context; mod embedder; mod live_state; @@ -15,6 +16,7 @@ mod skill; mod template; mod vector; +pub use background_task::{BackgroundTaskReconcileReport, FsBackgroundTaskStore}; pub use context::IdeaiContextStore; #[cfg(feature = "vector-onnx")] pub use embedder::OnnxEmbedder; diff --git a/crates/infrastructure/tests/agent_inbox.rs b/crates/infrastructure/tests/agent_inbox.rs new file mode 100644 index 0000000..a0bd73f --- /dev/null +++ b/crates/infrastructure/tests/agent_inbox.rs @@ -0,0 +1,212 @@ +//! B4 tests for the bounded AgentInbox facade over MediatedInbox. + +use std::sync::Arc; +use std::time::Duration; + +use domain::{ + AgentId, AgentInbox, InboxError, InboxItem, InboxItemKind, InboxReceiptStatus, InboxSource, + InputMediator, Ticket, TicketId, +}; +use infrastructure::{ + start_background_ready_inbox_bridge, BackgroundTaskReadyToDeliver, InMemoryMailbox, + MediatedInbox, MillisClock, +}; +use tokio::sync::mpsc::unbounded_channel; +use uuid::Uuid; + +struct FixedClock(u64); + +impl MillisClock for FixedClock { + fn now_ms(&self) -> u64 { + self.0 + } +} + +fn agent(n: u128) -> AgentId { + AgentId::from_uuid(Uuid::from_u128(n)) +} + +fn ticket_id(n: u128) -> TicketId { + TicketId::from_uuid(Uuid::from_u128(n)) +} + +fn task_id(n: u128) -> domain::TaskId { + domain::TaskId::from_uuid(Uuid::from_u128(n)) +} + +fn inbox(capacity: usize) -> MediatedInbox { + MediatedInbox::new( + Arc::new(InMemoryMailbox::new()), + Arc::new(FixedClock(1_000)), + ) + .with_inbox_capacity(capacity) +} + +fn user_item(id: u128, target: AgentId, body: &str) -> InboxItem { + InboxItem { + id: ticket_id(id), + agent_id: target, + source: InboxSource::Human, + kind: InboxItemKind::UserMessage, + body: body.into(), + created_at_ms: 1_000, + correlation_id: None, + } +} + +fn completion_item(id: u128, target: AgentId, task: u128) -> InboxItem { + InboxItem { + id: ticket_id(id), + agent_id: target, + source: InboxSource::BackgroundTask { + task_id: task_id(task), + }, + kind: InboxItemKind::BackgroundCompletion, + body: "done".into(), + created_at_ms: 1_000, + correlation_id: Some(task_id(task).to_string()), + } +} + +fn make_busy(inbox: &MediatedInbox, target: AgentId) { + let ticket = Ticket::new(ticket_id(900), "active", "active turn"); + let _pending = inbox.enqueue(target, ticket); +} + +#[test] +fn enqueue_message_while_busy_is_queued_not_rejected() { + let inbox = inbox(10); + let target = agent(1); + make_busy(&inbox, target); + + let receipt = inbox + .enqueue_message(target, user_item(1, target, "queued")) + .unwrap(); + + assert_eq!(receipt.status, InboxReceiptStatus::Queued); + assert_eq!(receipt.depth, 2); + let snapshot = inbox.snapshot(target); + assert_eq!(snapshot.depth, 2); + assert_eq!(snapshot.items.len(), 1); + assert_eq!(snapshot.items[0].body, "queued"); +} + +#[test] +fn inbox_fifo_drains_two_entrants_in_order() { + let inbox = inbox(10); + let target = agent(1); + let first = user_item(1, target, "first"); + let second = user_item(2, target, "second"); + + inbox.enqueue_message(target, first.clone()).unwrap(); + inbox.enqueue_message(target, second.clone()).unwrap(); + + assert_eq!(inbox.dequeue_next(target), Some(first)); + assert_eq!(inbox.dequeue_next(target), Some(second)); + assert_eq!(inbox.dequeue_next(target), None); +} + +#[test] +fn overflow_normal_message_returns_inbox_full() { + let inbox = inbox(1); + let target = agent(1); + make_busy(&inbox, target); + + let err = inbox + .enqueue_message(target, user_item(1, target, "overflow")) + .unwrap_err(); + + assert_eq!( + err, + InboxError::InboxFull { + agent_id: target, + capacity: 1, + } + ); +} + +#[test] +fn overflow_completion_is_deferred_not_lost_or_enqueued() { + let inbox = inbox(1); + let target = agent(1); + make_busy(&inbox, target); + + let receipt = inbox + .enqueue_message(target, completion_item(1, target, 42)) + .unwrap(); + + assert_eq!(receipt.status, InboxReceiptStatus::Deferred); + assert_eq!(receipt.depth, 1); + assert!(inbox.snapshot(target).items.is_empty()); +} + +#[test] +fn snapshot_exposes_queue_depth_and_items() { + let inbox = inbox(10); + let target = agent(1); + + inbox + .enqueue_message(target, user_item(1, target, "first")) + .unwrap(); + inbox + .enqueue_message(target, user_item(2, target, "second")) + .unwrap(); + + let snapshot = inbox.snapshot(target); + assert_eq!(snapshot.agent_id, target); + assert_eq!(snapshot.depth, 2); + assert_eq!( + snapshot + .items + .iter() + .map(|item| item.body.as_str()) + .collect::>(), + vec!["first", "second"] + ); +} + +#[test] +fn drain_removes_only_one_item_at_a_time() { + let inbox = inbox(10); + let target = agent(1); + inbox + .enqueue_message(target, user_item(1, target, "first")) + .unwrap(); + inbox + .enqueue_message(target, user_item(2, target, "second")) + .unwrap(); + + assert_eq!(inbox.dequeue_next(target).unwrap().body, "first"); + + let snapshot = inbox.snapshot(target); + assert_eq!(snapshot.depth, 1); + assert_eq!(snapshot.items[0].body, "second"); +} + +#[tokio::test] +async fn ready_bridge_enqueues_background_completion_item() { + let inbox = Arc::new(inbox(10)); + let target = agent(1); + let (tx, rx) = unbounded_channel(); + let handle = start_background_ready_inbox_bridge(rx, inbox.clone() as Arc); + + tx.send(BackgroundTaskReadyToDeliver { + task_id: task_id(42), + project_id: domain::ProjectId::from_uuid(Uuid::from_u128(7)), + owner_agent_id: target, + }) + .unwrap(); + + tokio::time::sleep(Duration::from_millis(20)).await; + let snapshot = inbox.snapshot(target); + assert_eq!(snapshot.depth, 1); + assert_eq!(snapshot.items[0].kind, InboxItemKind::BackgroundCompletion); + assert_eq!( + snapshot.items[0].source, + InboxSource::BackgroundTask { + task_id: task_id(42) + } + ); + drop(tx); + handle.await.unwrap(); +} diff --git a/crates/infrastructure/tests/background_completion_sink.rs b/crates/infrastructure/tests/background_completion_sink.rs new file mode 100644 index 0000000..6c8e3fc --- /dev/null +++ b/crates/infrastructure/tests/background_completion_sink.rs @@ -0,0 +1,368 @@ +//! B3 tests for the background completion sink. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::{mpsc, Arc, Mutex}; + +use async_trait::async_trait; +use domain::ports::{ + BackgroundCompletionStream, BackgroundTaskCompletion, BackgroundTaskHandle, + BackgroundTaskRunner, BackgroundTaskSpec, +}; +use domain::{ + AgentId, BackgroundTask, BackgroundTaskKind, BackgroundTaskPortError, BackgroundTaskResult, + BackgroundTaskState, BackgroundTaskStore, BackgroundTaskWakePolicy, ProjectId, ProjectPath, + TaskId, +}; +use infrastructure::{ + BackgroundCompletionSink, BackgroundCompletionSinkError, BackgroundCompletionSinkOutcome, + FsBackgroundTaskStore, +}; +use tokio::sync::mpsc::unbounded_channel; +use uuid::Uuid; + +struct TempDir(PathBuf); + +impl TempDir { + fn new() -> Self { + let path = std::env::temp_dir().join(format!("idea-b3-completion-sink-{}", Uuid::new_v4())); + std::fs::create_dir_all(&path).unwrap(); + Self(path) + } + + fn project_path(&self) -> ProjectPath { + ProjectPath::new(self.0.to_string_lossy().into_owned()).unwrap() + } +} + +impl Drop for TempDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +fn task_id(n: u128) -> TaskId { + TaskId::from_uuid(Uuid::from_u128(n)) +} + +fn project_id(n: u128) -> ProjectId { + ProjectId::from_uuid(Uuid::from_u128(n)) +} + +fn agent_id(n: u128) -> AgentId { + AgentId::from_uuid(Uuid::from_u128(n)) +} + +fn running_task(id: u128) -> BackgroundTask { + BackgroundTask::new( + task_id(id), + project_id(10), + agent_id(100), + BackgroundTaskKind::Command { + label: format!("task-{id}"), + }, + BackgroundTaskWakePolicy::WakeOwner, + 1_000, + None, + ) + .unwrap() + .transition(BackgroundTaskState::Running, 1_010) + .unwrap() +} + +fn success_completion(id: u128) -> BackgroundTaskCompletion { + BackgroundTaskCompletion { + task_id: task_id(id), + result: BackgroundTaskResult::Success { + finished_at_ms: 1_100, + exit_code: Some(0), + summary: "ok".into(), + stdout_tail: Some("done".into()), + stderr_tail: None, + }, + } +} + +#[tokio::test] +async fn runner_stream_completion_is_persisted_then_signalled() { + let tmp = TempDir::new(); + let store = Arc::new(FsBackgroundTaskStore::new(&tmp.project_path())); + let task = running_task(1); + store.create(&task).await.unwrap(); + + let (ready_tx, mut ready_rx) = unbounded_channel(); + let sink = Arc::new(BackgroundCompletionSink::new(store.clone(), ready_tx)); + let (completion_tx, completion_rx) = mpsc::channel(); + let runner = Arc::new(FakeRunner::new(completion_rx)); + let handle = sink.start_from_runner(runner); + + completion_tx.send(success_completion(1)).unwrap(); + drop(completion_tx); + + let ready = tokio::time::timeout(std::time::Duration::from_secs(2), ready_rx.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(ready.task_id, task.id); + + let persisted = store.get(task.id).await.unwrap().unwrap(); + assert_eq!(persisted.state, BackgroundTaskState::Completed); + assert!(persisted.has_pending_completion_delivery()); + handle.await.unwrap(); +} + +#[tokio::test] +async fn persistence_happens_before_ready_signal() { + let task = running_task(1); + let store = Arc::new(RecordingStore::with_task(task.clone())); + let order = store.order.clone(); + let (ready_tx, mut ready_rx) = unbounded_channel(); + let sink = BackgroundCompletionSink::new(store, ready_tx); + + let receiver_order = order.clone(); + let receiver = tokio::spawn(async move { + let _ = ready_rx.recv().await; + receiver_order.lock().unwrap().push("ready"); + }); + + let outcome = sink + .process_completion(success_completion(1)) + .await + .unwrap(); + assert!(matches!( + outcome, + BackgroundCompletionSinkOutcome::PersistedAndSignalled(_) + )); + receiver.await.unwrap(); + + assert_eq!(*order.lock().unwrap(), vec!["save", "ready"]); +} + +#[tokio::test] +async fn duplicate_completion_same_task_signals_once() { + let tmp = TempDir::new(); + let store = Arc::new(FsBackgroundTaskStore::new(&tmp.project_path())); + let task = running_task(1); + store.create(&task).await.unwrap(); + let (ready_tx, mut ready_rx) = unbounded_channel(); + let sink = BackgroundCompletionSink::new(store.clone(), ready_tx); + + let first = sink + .process_completion(success_completion(1)) + .await + .unwrap(); + let second = sink + .process_completion(success_completion(1)) + .await + .unwrap(); + + assert!(matches!( + first, + BackgroundCompletionSinkOutcome::PersistedAndSignalled(_) + )); + assert_eq!( + second, + BackgroundCompletionSinkOutcome::IgnoredDuplicate { task_id: task.id } + ); + assert!(ready_rx.recv().await.is_some()); + assert!(ready_rx.try_recv().is_err()); + + let persisted = store.get(task.id).await.unwrap().unwrap(); + assert_eq!(persisted.state, BackgroundTaskState::Completed); + assert!(persisted.has_pending_completion_delivery()); +} + +#[tokio::test] +async fn completion_for_already_terminal_task_is_ignored() { + let tmp = TempDir::new(); + let store = Arc::new(FsBackgroundTaskStore::new(&tmp.project_path())); + let completed = running_task(1) + .complete(success_completion(1).result) + .unwrap(); + store.create(&completed).await.unwrap(); + let (ready_tx, mut ready_rx) = unbounded_channel(); + let sink = BackgroundCompletionSink::new(store.clone(), ready_tx); + + let outcome = sink + .process_completion(success_completion(1)) + .await + .unwrap(); + + assert_eq!( + outcome, + BackgroundCompletionSinkOutcome::IgnoredAlreadyTerminal { + task_id: completed.id + } + ); + assert!(ready_rx.try_recv().is_err()); + assert_eq!(store.get(completed.id).await.unwrap(), Some(completed)); +} + +#[tokio::test] +async fn failed_persistence_does_not_signal_ready() { + let task = running_task(1); + let store = Arc::new(RecordingStore::with_task(task.clone()).failing_save()); + let (ready_tx, mut ready_rx) = unbounded_channel(); + let sink = BackgroundCompletionSink::new(store, ready_tx); + + let err = sink + .process_completion(success_completion(1)) + .await + .unwrap_err(); + + assert_eq!( + err, + BackgroundCompletionSinkError::Store(BackgroundTaskPortError::Store("boom".into())) + ); + assert!(ready_rx.try_recv().is_err()); +} + +#[tokio::test] +async fn failed_persistence_releases_task_for_later_retry() { + let task = running_task(1); + let store = Arc::new(RecordingStore::with_task(task.clone()).fail_first_save()); + let (ready_tx, mut ready_rx) = unbounded_channel(); + let sink = BackgroundCompletionSink::new(store, ready_tx); + + assert!(sink + .process_completion(success_completion(1)) + .await + .is_err()); + let second = sink + .process_completion(success_completion(1)) + .await + .unwrap(); + + assert!(matches!( + second, + BackgroundCompletionSinkOutcome::PersistedAndSignalled(_) + )); + assert!(ready_rx.recv().await.is_some()); +} + +struct FakeRunner { + rx: Mutex>>, +} + +impl FakeRunner { + fn new(rx: mpsc::Receiver) -> Self { + Self { + rx: Mutex::new(Some(rx)), + } + } +} + +#[async_trait] +impl BackgroundTaskRunner for FakeRunner { + async fn spawn( + &self, + _spec: BackgroundTaskSpec, + ) -> Result { + Err(BackgroundTaskPortError::Runner("not implemented".into())) + } + + async fn cancel(&self, _task_id: TaskId) -> Result<(), BackgroundTaskPortError> { + Err(BackgroundTaskPortError::Runner("not implemented".into())) + } + + fn subscribe_completions(&self) -> BackgroundCompletionStream { + let rx = self.rx.lock().unwrap().take().unwrap(); + Box::new(rx.into_iter()) + } +} + +struct RecordingStore { + tasks: Mutex>, + order: Arc>>, + fail_saves_remaining: Mutex, +} + +impl RecordingStore { + fn with_task(task: BackgroundTask) -> Self { + Self { + tasks: Mutex::new(HashMap::from([(task.id, task)])), + order: Arc::new(Mutex::new(Vec::new())), + fail_saves_remaining: Mutex::new(0), + } + } + + fn failing_save(self) -> Self { + *self.fail_saves_remaining.lock().unwrap() = usize::MAX; + self + } + + fn fail_first_save(self) -> Self { + *self.fail_saves_remaining.lock().unwrap() = 1; + self + } +} + +#[async_trait] +impl BackgroundTaskStore for RecordingStore { + async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> { + self.tasks.lock().unwrap().insert(task.id, task.clone()); + Ok(()) + } + + async fn get(&self, id: TaskId) -> Result, BackgroundTaskPortError> { + Ok(self.tasks.lock().unwrap().get(&id).cloned()) + } + + async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> { + let mut failures = self.fail_saves_remaining.lock().unwrap(); + if *failures > 0 { + if *failures != usize::MAX { + *failures -= 1; + } + return Err(BackgroundTaskPortError::Store("boom".into())); + } + drop(failures); + self.tasks.lock().unwrap().insert(task.id, task.clone()); + self.order.lock().unwrap().push("save"); + Ok(()) + } + + async fn list_open_for_agent( + &self, + agent_id: AgentId, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.owner_agent_id == agent_id && !task.is_terminal()) + .cloned() + .collect()) + } + + async fn list_undelivered_completions( + &self, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.has_pending_completion_delivery()) + .cloned() + .collect()) + } + + async fn mark_completion_delivered( + &self, + task_id: TaskId, + ) -> Result<(), BackgroundTaskPortError> { + let mut tasks = self.tasks.lock().unwrap(); + let task = tasks + .get(&task_id) + .cloned() + .ok_or(BackgroundTaskPortError::NotFound)?; + tasks.insert( + task_id, + task.mark_completion_delivered() + .map_err(|e| BackgroundTaskPortError::Invalid(e.to_string()))?, + ); + Ok(()) + } +} diff --git a/crates/infrastructure/tests/background_task_store.rs b/crates/infrastructure/tests/background_task_store.rs new file mode 100644 index 0000000..595fb95 --- /dev/null +++ b/crates/infrastructure/tests/background_task_store.rs @@ -0,0 +1,258 @@ +//! B2 integration tests for [`FsBackgroundTaskStore`]. + +use std::path::PathBuf; + +use domain::{ + AgentId, BackgroundTask, BackgroundTaskKind, BackgroundTaskResult, BackgroundTaskState, + BackgroundTaskStore, BackgroundTaskWakePolicy, ProjectId, ProjectPath, TaskId, +}; +use infrastructure::FsBackgroundTaskStore; +use uuid::Uuid; + +struct TempDir(PathBuf); + +impl TempDir { + fn new() -> Self { + let path = std::env::temp_dir().join(format!("idea-b2-bg-tasks-{}", Uuid::new_v4())); + std::fs::create_dir_all(&path).unwrap(); + Self(path) + } + + fn project_path(&self) -> ProjectPath { + ProjectPath::new(self.0.to_string_lossy().into_owned()).unwrap() + } + + fn task_dir(&self) -> PathBuf { + self.0.join(".ideai").join("background-tasks") + } + + fn task_file(&self, project_id: ProjectId) -> PathBuf { + self.task_dir().join(format!("{project_id}.json")) + } + + fn task_tmp_file(&self, project_id: ProjectId) -> PathBuf { + self.task_dir().join(format!("{project_id}.json.tmp")) + } +} + +impl Drop for TempDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +fn task_id(n: u128) -> TaskId { + TaskId::from_uuid(Uuid::from_u128(n)) +} + +fn project_id(n: u128) -> ProjectId { + ProjectId::from_uuid(Uuid::from_u128(n)) +} + +fn agent_id(n: u128) -> AgentId { + AgentId::from_uuid(Uuid::from_u128(n)) +} + +fn queued_task(id: u128, project: u128, owner: u128, now_ms: u64) -> BackgroundTask { + BackgroundTask::new( + task_id(id), + project_id(project), + agent_id(owner), + BackgroundTaskKind::Command { + label: format!("task-{id}"), + }, + BackgroundTaskWakePolicy::WakeOwner, + now_ms, + None, + ) + .unwrap() +} + +fn running_task(id: u128, project: u128, owner: u128, now_ms: u64) -> BackgroundTask { + queued_task(id, project, owner, now_ms) + .transition(BackgroundTaskState::Running, now_ms + 1) + .unwrap() +} + +fn completed_task(id: u128, project: u128, owner: u128, now_ms: u64) -> BackgroundTask { + running_task(id, project, owner, now_ms) + .complete(BackgroundTaskResult::Success { + finished_at_ms: now_ms + 2, + exit_code: Some(0), + summary: "ok".into(), + stdout_tail: Some("done".into()), + stderr_tail: None, + }) + .unwrap() +} + +#[tokio::test] +async fn create_get_save_round_trips_across_fresh_store() { + let tmp = TempDir::new(); + let store = FsBackgroundTaskStore::new(&tmp.project_path()); + let task = queued_task(1, 10, 100, 1_000); + + store.create(&task).await.unwrap(); + assert_eq!(store.get(task.id).await.unwrap(), Some(task.clone())); + + let running = task + .transition(BackgroundTaskState::Running, 1_010) + .unwrap(); + store.save(&running).await.unwrap(); + + let reborn = FsBackgroundTaskStore::new(&tmp.project_path()); + assert_eq!(reborn.get(task.id).await.unwrap(), Some(running)); + assert!(tmp.task_file(project_id(10)).exists()); + assert!(!tmp.task_tmp_file(project_id(10)).exists()); + + let raw = std::fs::read_to_string(tmp.task_file(project_id(10))).unwrap(); + assert!(raw.contains("\"projectId\""), "camelCase doc: {raw}"); + assert!(raw.contains("\"ownerAgentId\""), "camelCase task: {raw}"); +} + +#[tokio::test] +async fn create_rejects_duplicate_task_id() { + let tmp = TempDir::new(); + let store = FsBackgroundTaskStore::new(&tmp.project_path()); + let task = queued_task(1, 10, 100, 1_000); + + store.create(&task).await.unwrap(); + let err = store.create(&task).await.unwrap_err(); + + assert_eq!(err, domain::BackgroundTaskPortError::AlreadyExists); +} + +#[tokio::test] +async fn list_open_for_agent_uses_registry_and_filters_terminal_tasks() { + let tmp = TempDir::new(); + let store = FsBackgroundTaskStore::new(&tmp.project_path()); + let open_a = queued_task(1, 10, 100, 1_000); + let open_b = running_task(2, 10, 200, 1_000); + let done_a = completed_task(3, 10, 100, 1_000); + + store.create(&open_a).await.unwrap(); + store.create(&open_b).await.unwrap(); + store.create(&done_a).await.unwrap(); + + assert_eq!( + store.registry_open_task_ids_for_agent(agent_id(100)).await, + vec![open_a.id] + ); + + let listed = store.list_open_for_agent(agent_id(100)).await.unwrap(); + assert_eq!(listed, vec![open_a]); + assert_eq!( + store.list_open_for_agent(agent_id(200)).await.unwrap(), + vec![open_b] + ); +} + +#[tokio::test] +async fn list_undelivered_completions_and_mark_are_idempotent() { + let tmp = TempDir::new(); + let store = FsBackgroundTaskStore::new(&tmp.project_path()); + let task = completed_task(1, 10, 100, 1_000); + + store.create(&task).await.unwrap(); + assert_eq!( + store.list_undelivered_completions().await.unwrap(), + vec![task.clone()] + ); + + store.mark_completion_delivered(task.id).await.unwrap(); + store.mark_completion_delivered(task.id).await.unwrap(); + + let delivered = store.get(task.id).await.unwrap().unwrap(); + assert!(delivered.completion_delivered); + assert!(store + .list_undelivered_completions() + .await + .unwrap() + .is_empty()); + assert!(store + .list_open_for_agent(agent_id(100)) + .await + .unwrap() + .is_empty()); +} + +#[tokio::test] +async fn persistence_is_segmented_by_project_id() { + let tmp = TempDir::new(); + let store = FsBackgroundTaskStore::new(&tmp.project_path()); + let p1 = queued_task(1, 10, 100, 1_000); + let p2 = queued_task(2, 20, 100, 1_000); + + store.create(&p1).await.unwrap(); + store.create(&p2).await.unwrap(); + + assert!(tmp.task_file(project_id(10)).exists()); + assert!(tmp.task_file(project_id(20)).exists()); + assert_ne!(tmp.task_file(project_id(10)), tmp.task_file(project_id(20))); + + let raw_p1 = std::fs::read_to_string(tmp.task_file(project_id(10))).unwrap(); + let raw_p2 = std::fs::read_to_string(tmp.task_file(project_id(20))).unwrap(); + assert!(raw_p1.contains(&p1.id.to_string())); + assert!(!raw_p1.contains(&p2.id.to_string())); + assert!(raw_p2.contains(&p2.id.to_string())); + assert!(!raw_p2.contains(&p1.id.to_string())); +} + +#[tokio::test] +async fn reconcile_marks_running_without_live_handle_failed_and_pending() { + let tmp = TempDir::new(); + let store = FsBackgroundTaskStore::new(&tmp.project_path()); + let lost = running_task(1, 10, 100, 1_000); + let still_live = running_task(2, 10, 100, 1_000); + let already_pending = completed_task(3, 10, 100, 1_000); + + store.create(&lost).await.unwrap(); + store.create(&still_live).await.unwrap(); + store.create(&already_pending).await.unwrap(); + + let report = store.reconcile_boot(&[still_live.id], 2_000).await.unwrap(); + + assert_eq!(report.failed_task_ids, vec![lost.id]); + assert_eq!( + report.delivery_pending_task_ids, + vec![lost.id, already_pending.id] + ); + + let lost_after = store.get(lost.id).await.unwrap().unwrap(); + assert_eq!(lost_after.state, BackgroundTaskState::Failed); + assert!(lost_after.has_pending_completion_delivery()); + assert!(matches!( + lost_after.result, + Some(BackgroundTaskResult::Failure { + finished_at_ms: 2_000, + exit_code: None, + .. + }) + )); + + let live_after = store.get(still_live.id).await.unwrap().unwrap(); + assert_eq!(live_after.state, BackgroundTaskState::Running); + assert_eq!( + store.registry_open_task_ids_for_agent(agent_id(100)).await, + vec![still_live.id] + ); +} + +#[tokio::test] +async fn reconcile_uses_monotonic_timestamp_when_clock_is_behind() { + let tmp = TempDir::new(); + let store = FsBackgroundTaskStore::new(&tmp.project_path()); + let lost = running_task(1, 10, 100, 1_000); + + store.create(&lost).await.unwrap(); + store.reconcile_boot(&[], 900).await.unwrap(); + + let after = store.get(lost.id).await.unwrap().unwrap(); + assert!(matches!( + after.result, + Some(BackgroundTaskResult::Failure { + finished_at_ms: 1_001, + .. + }) + )); +} diff --git a/crates/infrastructure/tests/orchestrator_watcher.rs b/crates/infrastructure/tests/orchestrator_watcher.rs index 3100357..208e909 100644 --- a/crates/infrastructure/tests/orchestrator_watcher.rs +++ b/crates/infrastructure/tests/orchestrator_watcher.rs @@ -16,15 +16,15 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use domain::agent::{AgentManifest, ManifestEntry}; use domain::events::{DomainEvent, OrchestrationSource}; -use domain::ids::NodeId; use domain::ids::SkillId; use domain::ids::{AgentId, ProfileId, ProjectId}; use domain::markdown::MarkdownDoc; use domain::ports::{ - AgentContextStore, AgentRuntime, ContextInjectionPlan, DirEntry, EventBus, EventStream, - ExitStatus, FileSystem, FsError, IdGenerator, OutputStream, PreparedContext, ProfileStore, - PtyError, PtyHandle, PtyPort, RemotePath, RuntimeError, SessionPlan, SkillStore, SpawnSpec, - StoreError, + AgentContextStore, AgentRuntime, AgentSession, AgentSessionError, AgentSessionFactory, + BackgroundTaskPortError, BackgroundTaskStore, Clock, ContextInjectionPlan, DirEntry, EventBus, + EventStream, ExitStatus, FileSystem, FsError, IdGenerator, OutputStream, PreparedContext, + ProfileStore, PtyError, PtyHandle, PtyPort, RemotePath, ReplyEvent, ReplyStream, RuntimeError, + SessionPlan, SkillStore, SpawnSpec, StoreError, }; use domain::profile::{ AgentProfile, ContextInjection, McpCapability, McpConfigStrategy, McpTransport, @@ -33,13 +33,14 @@ use domain::profile::{ use domain::project::{Project, ProjectPath}; use domain::remote::RemoteRef; use domain::skill::{Skill, SkillScope}; -use domain::terminal::{SessionKind, TerminalSession}; -use domain::{PtySize, SessionId}; +use domain::{ + BackgroundTask, BackgroundTaskResult, BackgroundTaskState, PtySize, SessionId, TaskId, +}; use uuid::Uuid; use application::{ CloseTerminal, CreateAgentFromScratch, CreateSkill, LaunchAgent, ListAgents, - OrchestratorService, TerminalSessions, UpdateAgentContext, + OrchestratorService, StructuredSessions, TerminalSessions, UpdateAgentContext, }; use infrastructure::{ process_request_file, FsOrchestratorWatcher, InMemoryMailbox, OrchestratorResponse, @@ -316,6 +317,172 @@ impl IdGenerator for SeqIds { } } +struct FixedClock(i64); +impl Clock for FixedClock { + fn now_millis(&self) -> i64 { + self.0 + } +} + +#[derive(Default)] +struct InMemoryBackgroundTaskStore { + tasks: Mutex>, +} + +impl InMemoryBackgroundTaskStore { + fn all(&self) -> Vec { + self.tasks.lock().unwrap().values().cloned().collect() + } +} + +#[async_trait] +impl BackgroundTaskStore for InMemoryBackgroundTaskStore { + async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> { + let mut tasks = self.tasks.lock().unwrap(); + if tasks.contains_key(&task.id) { + return Err(BackgroundTaskPortError::AlreadyExists); + } + tasks.insert(task.id, task.clone()); + Ok(()) + } + + async fn get(&self, id: TaskId) -> Result, BackgroundTaskPortError> { + Ok(self.tasks.lock().unwrap().get(&id).cloned()) + } + + async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> { + self.tasks.lock().unwrap().insert(task.id, task.clone()); + Ok(()) + } + + async fn list_open_for_agent( + &self, + agent_id: AgentId, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.owner_agent_id == agent_id && !task.is_terminal()) + .cloned() + .collect()) + } + + async fn list_undelivered_completions( + &self, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.has_pending_completion_delivery()) + .cloned() + .collect()) + } + + async fn mark_completion_delivered( + &self, + task_id: TaskId, + ) -> Result<(), BackgroundTaskPortError> { + let mut tasks = self.tasks.lock().unwrap(); + let task = tasks + .get(&task_id) + .cloned() + .ok_or(BackgroundTaskPortError::NotFound)?; + let delivered = task + .mark_completion_delivered() + .map_err(|err| BackgroundTaskPortError::Invalid(err.to_string()))?; + tasks.insert(task_id, delivered); + Ok(()) + } +} + +#[derive(Clone)] +struct BlockingReplyFactory { + outcome: Arc>, + notify: Arc, +} + +impl BlockingReplyFactory { + fn new(reply: impl Into) -> Self { + Self { + outcome: Arc::new(Mutex::new(BlockingReplyOutcome::Final(reply.into()))), + notify: Arc::new(tokio::sync::Notify::new()), + } + } + + fn no_reply(&self) { + *self.outcome.lock().unwrap() = BlockingReplyOutcome::NoReply; + } + + fn release(&self) { + self.notify.notify_waiters(); + } +} + +#[derive(Clone)] +enum BlockingReplyOutcome { + Final(String), + NoReply, +} + +struct BlockingReplySession { + id: SessionId, + outcome: Arc>, + notify: Arc, +} + +#[async_trait] +impl AgentSession for BlockingReplySession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + None + } + + async fn send(&self, _prompt: &str) -> Result { + self.notify.notified().await; + match self.outcome.lock().unwrap().clone() { + BlockingReplyOutcome::Final(content) => { + Ok(Box::new(vec![ReplyEvent::Final { content }].into_iter())) + } + BlockingReplyOutcome::NoReply => Err(AgentSessionError::Io( + "target returned without a structured final".to_owned(), + )), + } + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + +#[async_trait] +impl AgentSessionFactory for BlockingReplyFactory { + fn supports(&self, profile: &AgentProfile) -> bool { + profile.structured_adapter.is_some() + } + + async fn start( + &self, + _profile: &AgentProfile, + _ctx: &PreparedContext, + _cwd: &ProjectPath, + _session: &SessionPlan, + _sandbox: Option<&domain::sandbox::SandboxPlan>, + ) -> Result, AgentSessionError> { + Ok(Arc::new(BlockingReplySession { + id: SessionId::from_uuid(Uuid::new_v4()), + outcome: Arc::clone(&self.outcome), + notify: Arc::clone(&self.notify), + })) + } +} + fn project() -> Project { Project::new( ProjectId::from_uuid(Uuid::from_u128(1000)), @@ -396,6 +563,8 @@ fn build_service_with_mailbox( Arc, Arc, Arc, + BlockingReplyFactory, + Arc, ) { // Profil Claude **complet** (adaptateur structuré + capacité MCP `.mcp.json`) : // seul profil que la garde F2 (`guard_mcp_bridge_supported`) accepte comme cible @@ -417,6 +586,9 @@ fn build_service_with_mailbox( McpTransport::Stdio, ))]))); let sessions = Arc::new(TerminalSessions::new()); + let structured = Arc::new(StructuredSessions::new()); + let structured_factory = BlockingReplyFactory::new("the answer is 42"); + let background_tasks = Arc::new(InMemoryBackgroundTaskStore::default()); let mailbox = Arc::new(InMemoryMailbox::new()); let bus = Arc::new(NoopBus); let create = Arc::new(CreateAgentFromScratch::new( @@ -424,19 +596,25 @@ fn build_service_with_mailbox( Arc::new(SeqIds(Mutex::new(1))), bus.clone(), )); - let launch = Arc::new(LaunchAgent::new( - Arc::new(contexts.clone()), - Arc::clone(&profiles) as Arc, - Arc::new(FakeRuntime), - Arc::new(FakeFs), - Arc::new(FakePty), - Arc::new(FakeSkills), - Arc::clone(&sessions), - bus.clone(), - Arc::new(SeqIds(Mutex::new(1))), - Arc::new(FakeRecall), - None, - )); + let launch = Arc::new( + LaunchAgent::new( + Arc::new(contexts.clone()), + Arc::clone(&profiles) as Arc, + Arc::new(FakeRuntime), + Arc::new(FakeFs), + Arc::new(FakePty), + Arc::new(FakeSkills), + Arc::clone(&sessions), + bus.clone(), + Arc::new(SeqIds(Mutex::new(1))), + Arc::new(FakeRecall), + None, + ) + .with_structured( + Arc::new(structured_factory.clone()) as Arc, + Arc::clone(&structured), + ), + ); let list = Arc::new(ListAgents::new(Arc::new(contexts.clone()))); let close = Arc::new(CloseTerminal::new(Arc::new(FakePty), Arc::clone(&sessions))); let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts))); @@ -466,23 +644,20 @@ fn build_service_with_mailbox( .with_conversations( Arc::new(infrastructure::InMemoryConversationRegistry::new()) as Arc, + ) + .with_structured(Arc::clone(&structured)) + .with_background_tasks( + Arc::clone(&background_tasks) as Arc, + Arc::new(FixedClock(1_700_000_000_000)) as Arc, ), ); - (service, mailbox, sessions) -} - -/// Pre-seeds a live PTY terminal session for `agent_id` so `ask_agent` reuses it. -fn seed_live_pty(sessions: &TerminalSessions, agent_id: AgentId, session_id: SessionId) { - sessions.insert( - PtyHandle { session_id }, - TerminalSession::starting( - session_id, - NodeId::from_uuid(Uuid::from_u128(7)), - ProjectPath::new("/home/me/proj").unwrap(), - SessionKind::Agent { agent_id }, - PtySize { rows: 24, cols: 80 }, - ), - ); + ( + service, + mailbox, + sessions, + structured_factory, + background_tasks, + ) } fn read_response(request_path: &std::path::Path) -> OrchestratorResponse { @@ -637,21 +812,15 @@ async fn unknown_action_yields_error_response() { /// alongside `ok: true`. #[tokio::test] async fn ask_request_surfaces_reply_alongside_detail() { - // Option 1 (B-3/B-4): an `agent.message` request blocks awaiting the target's - // `idea_reply`. Over the file protocol we drive the ask request on a task, wait - // for the mailbox to hold a ticket, then process a separate `agent.reply` request - // (carrying the target's id as `requestedBy`, the handshake identity) which - // resolves it. The ask's `*.response.json` then carries reply + detail. + // B6: an `agent.message` request is still synchronous for the caller, but the + // target turn is driven through the structured/headless session. The mailbox still + // receives exactly one silent ticket for turn accounting; the response comes from + // `ReplyEvent::Final`, not from a separate `agent.reply` request. let tmp = TempDir::new(); let contexts = FakeContexts::new(); let agent_id = contexts.seed_agent("architect"); - let (service, mailbox, sessions) = build_service_with_mailbox(contexts.clone()); - // Target already live in the PTY registry, so the ask reuses its terminal. - seed_live_pty( - &sessions, - agent_id, - SessionId::from_uuid(Uuid::from_u128(4242)), - ); + let (service, mailbox, _sessions, structured_reply, tasks) = + build_service_with_mailbox(contexts.clone()); let req = tmp.0.join("ask-req.json"); std::fs::write( @@ -674,19 +843,7 @@ async fn ask_request_surfaces_reply_alongside_detail() { .await .expect("ask must enqueue a ticket"); - // The target renders its result via an `agent.reply` request whose `requestedBy` - // is its own id (the handshake identity the MCP server would inject as `from`). - let reply_req = tmp.0.join("reply-req.json"); - std::fs::write( - &reply_req, - format!( - r#"{{ "type": "agent.reply", "requestedBy": "{agent_id}", "result": "the answer is 42" }}"# - ) - .as_bytes(), - ) - .unwrap(); - let reply_resp = process_request_file(&reply_req, &project(), &service).await; - assert!(reply_resp.ok, "reply request ok: {reply_resp:?}"); + structured_reply.release(); let response = tokio::time::timeout(std::time::Duration::from_secs(10), ask) .await @@ -718,6 +875,74 @@ async fn ask_request_surfaces_reply_alongside_detail() { assert_eq!(json["reply"], serde_json::json!("the answer is 42")); assert!(json.get("detail").is_some()); assert!(!req.exists(), "request file must be removed"); + + let stored = tasks.all(); + assert_eq!(stored.len(), 1, "one rendezvous task persisted: {stored:?}"); + let task = &stored[0]; + assert_eq!(task.owner_agent_id, agent_id); + assert_eq!(task.state, BackgroundTaskState::Completed); + assert!(matches!( + task.kind, + domain::BackgroundTaskKind::HeadlessRendezvous { .. } + )); + assert!(matches!( + task.result, + Some(BackgroundTaskResult::Success { .. }) + )); +} + +#[tokio::test] +async fn ask_request_no_reply_persists_failed_rendezvous_task() { + let tmp = TempDir::new(); + let contexts = FakeContexts::new(); + let agent_id = contexts.seed_agent("architect"); + let (service, mailbox, _sessions, structured_reply, tasks) = + build_service_with_mailbox(contexts.clone()); + structured_reply.no_reply(); + + let req = tmp.0.join("ask-no-reply.json"); + std::fs::write( + &req, + br#"{ "type": "agent.message", "requestedBy": "Main", "targetAgent": "architect", "task": "Please forget to finalise" }"#, + ) + .unwrap(); + + let svc = Arc::clone(&service); + let proj = project(); + let ask_req = req.clone(); + let ask = tokio::spawn(async move { process_request_file(&ask_req, &proj, &svc).await }); + + tokio::time::timeout(std::time::Duration::from_secs(10), async { + while mailbox.pending(&agent_id) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("ask must enqueue a ticket before no-reply"); + + structured_reply.release(); + + let response = tokio::time::timeout(std::time::Duration::from_secs(10), ask) + .await + .expect("ask completes after no-reply") + .expect("join ok"); + assert!(!response.ok, "expected no-reply error, got {response:?}"); + + let stored = tasks.all(); + assert_eq!(stored.len(), 1, "one rendezvous task persisted: {stored:?}"); + let task = &stored[0]; + assert_eq!(task.owner_agent_id, agent_id); + assert_eq!(task.state, BackgroundTaskState::Failed); + match task.result.as_ref() { + Some(BackgroundTaskResult::Failure { error, .. }) => { + assert!( + error.contains("NoReply"), + "error should classify no-reply: {error}" + ); + } + other => panic!("expected failure result, got {other:?}"), + } + assert_eq!(mailbox.pending(&agent_id), 0, "turn-lock mailbox is freed"); } /// Point 2 — a non-`ask` command (here `spawn_agent`, reply `None`) ⇒ the `reply`