feat(infrastructure): store, sink de complétion et mailbox des tâches de fond (B2-B4)
Adapters de persistance des BackgroundTask (store), sink de récupération de complétion post-tour et boîte de réception (mailbox) pour les messages concurrents, câblés sur l'entrée. Couvert par background_task_store, background_completion_sink, agent_inbox et orchestrator_watcher (verts). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
233
crates/infrastructure/src/background_task.rs
Normal file
233
crates/infrastructure/src/background_task.rs
Normal file
@ -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<dyn BackgroundTaskStore>,
|
||||
ready: UnboundedSender<BackgroundTaskReadyToDeliver>,
|
||||
processed: Mutex<HashSet<TaskId>>,
|
||||
}
|
||||
|
||||
/// 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<dyn BackgroundTaskStore>,
|
||||
ready: UnboundedSender<BackgroundTaskReadyToDeliver>,
|
||||
) -> 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<Self>,
|
||||
runner: Arc<dyn BackgroundTaskRunner>,
|
||||
) -> 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<BackgroundCompletionSinkOutcome, BackgroundCompletionSinkError> {
|
||||
{
|
||||
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<BackgroundCompletionSinkOutcome, BackgroundCompletionSinkError> {
|
||||
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<BackgroundTaskReadyToDeliver>,
|
||||
inbox: Arc<dyn AgentInbox>,
|
||||
) -> 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)
|
||||
}
|
||||
@ -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<HashMap<AgentId, Option<u32>>>,
|
||||
/// Rich inbox payloads keyed by the mailbox ticket id. The mailbox remains the
|
||||
/// only FIFO; this map is metadata only.
|
||||
inbox_items: Mutex<HashMap<TicketId, InboxItem>>,
|
||||
/// 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<AgentId, PtyHandle>> {
|
||||
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<TicketId, InboxItem>> {
|
||||
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<InboxReceipt, InboxError> {
|
||||
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<InboxItem> {
|
||||
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<InboxItem> = 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::<usize>().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::*;
|
||||
|
||||
@ -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,
|
||||
};
|
||||
|
||||
390
crates/infrastructure/src/store/background_task.rs
Normal file
390
crates/infrastructure/src/store/background_task.rs
Normal file
@ -0,0 +1,390 @@
|
||||
//! [`FsBackgroundTaskStore`] — durable store for first-class background tasks.
|
||||
//!
|
||||
//! Persistence is segmented by project id under the project root:
|
||||
//!
|
||||
//! ```text
|
||||
//! <project_root>/.ideai/background-tasks/<projectId>.json
|
||||
//! ```
|
||||
//!
|
||||
//! Each file is a small JSON document `{ version, projectId, tasks }`. Writes are
|
||||
//! atomic: serialize to `<projectId>.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<RegistryIndex>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
struct RegistryIndex {
|
||||
loaded: bool,
|
||||
task_to_project: HashMap<TaskId, ProjectId>,
|
||||
open_by_agent: HashMap<AgentId, BTreeMap<TaskId, BackgroundTask>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
struct StoreState {
|
||||
docs: BTreeMap<ProjectId, TaskDoc>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct TaskDoc {
|
||||
version: u32,
|
||||
project_id: ProjectId,
|
||||
tasks: Vec<BackgroundTask>,
|
||||
}
|
||||
|
||||
/// 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<TaskId>,
|
||||
/// Terminal tasks whose completion remains to be delivered.
|
||||
pub delivery_pending_task_ids: Vec<TaskId>,
|
||||
}
|
||||
|
||||
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()),
|
||||
}
|
||||
}
|
||||
|
||||
/// `<root>/.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<TaskId> {
|
||||
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<BackgroundTaskReconcileReport, BackgroundTaskPortError> {
|
||||
let live: HashSet<TaskId> = 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<StoreState, BackgroundTaskPortError> {
|
||||
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<TaskDoc, BackgroundTaskPortError> {
|
||||
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<TaskDoc, BackgroundTaskPortError> {
|
||||
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<Option<ProjectId>, 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<Option<BackgroundTask>, 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<Vec<BackgroundTask>, BackgroundTaskPortError> {
|
||||
self.ensure_registry_loaded().await?;
|
||||
let registry = self.registry.read().await;
|
||||
let mut tasks: Vec<BackgroundTask> = 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<Vec<BackgroundTask>, BackgroundTaskPortError> {
|
||||
let mut tasks: Vec<BackgroundTask> = 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())
|
||||
}
|
||||
@ -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;
|
||||
|
||||
212
crates/infrastructure/tests/agent_inbox.rs
Normal file
212
crates/infrastructure/tests/agent_inbox.rs
Normal file
@ -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<_>>(),
|
||||
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<dyn AgentInbox>);
|
||||
|
||||
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();
|
||||
}
|
||||
368
crates/infrastructure/tests/background_completion_sink.rs
Normal file
368
crates/infrastructure/tests/background_completion_sink.rs
Normal file
@ -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<Option<mpsc::Receiver<BackgroundTaskCompletion>>>,
|
||||
}
|
||||
|
||||
impl FakeRunner {
|
||||
fn new(rx: mpsc::Receiver<BackgroundTaskCompletion>) -> Self {
|
||||
Self {
|
||||
rx: Mutex::new(Some(rx)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl BackgroundTaskRunner for FakeRunner {
|
||||
async fn spawn(
|
||||
&self,
|
||||
_spec: BackgroundTaskSpec,
|
||||
) -> Result<BackgroundTaskHandle, BackgroundTaskPortError> {
|
||||
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<HashMap<TaskId, BackgroundTask>>,
|
||||
order: Arc<Mutex<Vec<&'static str>>>,
|
||||
fail_saves_remaining: Mutex<usize>,
|
||||
}
|
||||
|
||||
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<Option<BackgroundTask>, 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<Vec<BackgroundTask>, 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<Vec<BackgroundTask>, 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(())
|
||||
}
|
||||
}
|
||||
258
crates/infrastructure/tests/background_task_store.rs
Normal file
258
crates/infrastructure/tests/background_task_store.rs
Normal file
@ -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,
|
||||
..
|
||||
})
|
||||
));
|
||||
}
|
||||
@ -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<HashMap<TaskId, BackgroundTask>>,
|
||||
}
|
||||
|
||||
impl InMemoryBackgroundTaskStore {
|
||||
fn all(&self) -> Vec<BackgroundTask> {
|
||||
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<Option<BackgroundTask>, 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<Vec<BackgroundTask>, 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<Vec<BackgroundTask>, 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<Mutex<BlockingReplyOutcome>>,
|
||||
notify: Arc<tokio::sync::Notify>,
|
||||
}
|
||||
|
||||
impl BlockingReplyFactory {
|
||||
fn new(reply: impl Into<String>) -> 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<Mutex<BlockingReplyOutcome>>,
|
||||
notify: Arc<tokio::sync::Notify>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl AgentSession for BlockingReplySession {
|
||||
fn id(&self) -> SessionId {
|
||||
self.id
|
||||
}
|
||||
|
||||
fn conversation_id(&self) -> Option<String> {
|
||||
None
|
||||
}
|
||||
|
||||
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
|
||||
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<Arc<dyn AgentSession>, 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<OrchestratorService>,
|
||||
Arc<InMemoryMailbox>,
|
||||
Arc<TerminalSessions>,
|
||||
BlockingReplyFactory,
|
||||
Arc<InMemoryBackgroundTaskStore>,
|
||||
) {
|
||||
// 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,7 +596,8 @@ fn build_service_with_mailbox(
|
||||
Arc::new(SeqIds(Mutex::new(1))),
|
||||
bus.clone(),
|
||||
));
|
||||
let launch = Arc::new(LaunchAgent::new(
|
||||
let launch = Arc::new(
|
||||
LaunchAgent::new(
|
||||
Arc::new(contexts.clone()),
|
||||
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
|
||||
Arc::new(FakeRuntime),
|
||||
@ -436,7 +609,12 @@ fn build_service_with_mailbox(
|
||||
Arc::new(SeqIds(Mutex::new(1))),
|
||||
Arc::new(FakeRecall),
|
||||
None,
|
||||
));
|
||||
)
|
||||
.with_structured(
|
||||
Arc::new(structured_factory.clone()) as Arc<dyn AgentSessionFactory>,
|
||||
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<dyn domain::conversation::ConversationRegistry>,
|
||||
)
|
||||
.with_structured(Arc::clone(&structured))
|
||||
.with_background_tasks(
|
||||
Arc::clone(&background_tasks) as Arc<dyn BackgroundTaskStore>,
|
||||
Arc::new(FixedClock(1_700_000_000_000)) as Arc<dyn Clock>,
|
||||
),
|
||||
);
|
||||
(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`
|
||||
|
||||
Reference in New Issue
Block a user