diff --git a/crates/domain/src/background_task.rs b/crates/domain/src/background_task.rs new file mode 100644 index 0000000..3256491 --- /dev/null +++ b/crates/domain/src/background_task.rs @@ -0,0 +1,476 @@ +//! First-class background task model. +//! +//! This module is pure domain: it defines the persisted task state and enforces +//! local invariants only. Running commands, storing tasks and waking agents are +//! adapter/application concerns behind ports. + +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use crate::conversation::ConversationId; +use crate::ids::{AgentId, ProjectId, ScheduleId, TaskId}; +use crate::mailbox::TicketId; + +/// Maximum length for human-facing task labels. +pub const BACKGROUND_TASK_LABEL_MAX_CHARS: usize = 160; +/// Maximum size for summary/error/reason text fields. +pub const BACKGROUND_TASK_TEXT_MAX_BYTES: usize = 64 * 1024; +/// Maximum size retained for stdout/stderr tails. +pub const BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES: usize = 16 * 1024; + +/// Lifecycle state of a background task. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum BackgroundTaskState { + /// Created and waiting to be started. + Queued, + /// Currently executing. + Running, + /// Suspended while waiting for an external event. + Waiting, + /// Finished successfully. + Completed, + /// Finished with an error. + Failed, + /// Cancelled before natural completion. + Cancelled, + /// Expired after its deadline. + Expired, +} + +impl BackgroundTaskState { + /// Returns whether this state is terminal. + #[must_use] + pub const fn is_terminal(self) -> bool { + matches!( + self, + Self::Completed | Self::Failed | Self::Cancelled | Self::Expired + ) + } +} + +/// What IdeA should do with the owner when the task completes. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum BackgroundTaskWakePolicy { + /// Wake the owning agent so it can observe the completion. + WakeOwner, + /// Persist the completion only. + RecordOnly, +} + +/// Kind-specific payload of a background task. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", tag = "kind")] +pub enum BackgroundTaskKind { + /// A non-interactive command. + Command { + /// Human-facing label. + label: String, + }, + /// A headless inter-agent rendezvous. + HeadlessRendezvous { + /// Agent that requested the rendezvous, when known. + requester_agent_id: Option, + /// Target agent doing the work. + target_agent_id: AgentId, + /// Existing mailbox ticket. + ticket_id: TicketId, + /// Conversation entered by the target turn. + conversation_id: ConversationId, + }, + /// Resume of a structured agent session. + SessionResume { + /// Agent to resume. + agent_id: AgentId, + /// Scheduler handle that triggered the resume, when any. + schedule_id: Option, + /// Provider conversation id to resume, when any. + conversation_id: Option, + }, + /// Internal maintenance work. + Maintenance { + /// Human-facing label. + label: String, + }, +} + +/// Persisted first-class background task. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct BackgroundTask { + /// Stable task id. + pub id: TaskId, + /// Owning project. + pub project_id: ProjectId, + /// Agent to notify according to [`BackgroundTaskWakePolicy`]. + pub owner_agent_id: AgentId, + /// Kind-specific payload. + pub kind: BackgroundTaskKind, + /// Current lifecycle state. + pub state: BackgroundTaskState, + /// Completion wake policy. + pub wake_policy: BackgroundTaskWakePolicy, + /// Creation timestamp, epoch milliseconds. + pub created_at_ms: u64, + /// Last update timestamp, epoch milliseconds. + pub updated_at_ms: u64, + /// Optional absolute deadline, epoch milliseconds. + pub deadline_ms: Option, + /// Terminal result. Present if and only if the state is terminal. + pub result: Option, + /// Whether the completion has been delivered to the owner/inbox layer. + pub completion_delivered: bool, +} + +impl BackgroundTask { + /// Builds a queued task and validates kind/timestamps. + /// + /// # Errors + /// [`BackgroundTaskError`] when kind or timestamps are invalid. + pub fn new( + id: TaskId, + project_id: ProjectId, + owner_agent_id: AgentId, + kind: BackgroundTaskKind, + wake_policy: BackgroundTaskWakePolicy, + now_ms: u64, + deadline_ms: Option, + ) -> Result { + kind.validate()?; + let task = Self { + id, + project_id, + owner_agent_id, + kind, + state: BackgroundTaskState::Queued, + wake_policy, + created_at_ms: now_ms, + updated_at_ms: now_ms, + deadline_ms, + result: None, + completion_delivered: false, + }; + task.validate()?; + Ok(task) + } + + /// Returns a copy moved to a non-terminal state. + /// + /// Terminal transitions must use [`Self::complete`] so the result/state + /// invariant stays atomic. + /// + /// # Errors + /// [`BackgroundTaskError`] when the transition or timestamp is invalid. + pub fn transition( + &self, + to: BackgroundTaskState, + now_ms: u64, + ) -> Result { + if now_ms < self.updated_at_ms { + return Err(BackgroundTaskError::InvalidTimestamps); + } + if to.is_terminal() { + return Err(BackgroundTaskError::MissingResult); + } + self.ensure_transition_allowed(to)?; + let mut next = self.clone(); + next.state = to; + next.updated_at_ms = now_ms; + next.validate()?; + Ok(next) + } + + /// Returns a copy moved to the terminal state matching `result`. + /// + /// # Errors + /// [`BackgroundTaskError`] when the transition or result is invalid. + pub fn complete(&self, result: BackgroundTaskResult) -> Result { + result.validate()?; + if result.finished_at_ms() < self.updated_at_ms { + return Err(BackgroundTaskError::InvalidTimestamps); + } + let to = result.state(); + self.ensure_transition_allowed(to)?; + let mut next = self.clone(); + next.state = to; + next.updated_at_ms = result.finished_at_ms(); + next.result = Some(result); + next.validate()?; + Ok(next) + } + + /// Marks the terminal completion as delivered. + /// + /// # Errors + /// [`BackgroundTaskError`] if the task is not terminal or already delivered. + pub fn mark_completion_delivered(&self) -> Result { + if !self.is_terminal() { + return Err(BackgroundTaskError::MissingResult); + } + if self.completion_delivered { + return Err(BackgroundTaskError::CompletionAlreadyDelivered); + } + let mut next = self.clone(); + next.completion_delivered = true; + next.validate()?; + Ok(next) + } + + /// Returns whether the current state is terminal. + #[must_use] + pub fn is_terminal(&self) -> bool { + self.state.is_terminal() + } + + /// Returns whether a terminal completion still needs delivery. + #[must_use] + pub fn has_pending_completion_delivery(&self) -> bool { + self.is_terminal() && !self.completion_delivered + } + + fn ensure_transition_allowed( + &self, + to: BackgroundTaskState, + ) -> Result<(), BackgroundTaskError> { + if is_transition_allowed(self.state, to) { + Ok(()) + } else { + Err(BackgroundTaskError::InvalidTransition { + from: self.state, + to, + }) + } + } + + fn validate(&self) -> Result<(), BackgroundTaskError> { + self.kind.validate()?; + if self.updated_at_ms < self.created_at_ms + || self + .deadline_ms + .is_some_and(|deadline| deadline < self.created_at_ms) + { + return Err(BackgroundTaskError::InvalidTimestamps); + } + match (&self.state, &self.result) { + (state, None) if state.is_terminal() => return Err(BackgroundTaskError::MissingResult), + (state, Some(_)) if !state.is_terminal() => { + return Err(BackgroundTaskError::UnexpectedResult); + } + _ => {} + } + if self.completion_delivered && !self.state.is_terminal() { + return Err(BackgroundTaskError::MissingResult); + } + if let Some(result) = &self.result { + result.validate()?; + if result.state() != self.state { + return Err(BackgroundTaskError::InvalidTransition { + from: self.state, + to: result.state(), + }); + } + if result.finished_at_ms() < self.created_at_ms { + return Err(BackgroundTaskError::InvalidTimestamps); + } + } + Ok(()) + } +} + +/// Result of a terminal background task. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", tag = "outcome")] +pub enum BackgroundTaskResult { + /// Successful completion. + Success { + /// Completion timestamp, epoch milliseconds. + finished_at_ms: u64, + /// Process exit code, when command-backed. + exit_code: Option, + /// Human-readable summary. + summary: String, + /// Bounded stdout tail. + stdout_tail: Option, + /// Bounded stderr tail. + stderr_tail: Option, + }, + /// Failed completion. + Failure { + /// Completion timestamp, epoch milliseconds. + finished_at_ms: u64, + /// Process exit code, when command-backed. + exit_code: Option, + /// Human-readable error. + error: String, + /// Bounded stdout tail. + stdout_tail: Option, + /// Bounded stderr tail. + stderr_tail: Option, + }, + /// Cancelled completion. + Cancelled { + /// Completion timestamp, epoch milliseconds. + finished_at_ms: u64, + /// Cancellation reason. + reason: String, + }, + /// Expired completion. + Expired { + /// Completion timestamp, epoch milliseconds. + finished_at_ms: u64, + /// Expiry reason. + reason: String, + }, +} + +impl BackgroundTaskKind { + /// Validates kind-specific fields. + /// + /// # Errors + /// [`BackgroundTaskError`] when labels are empty or too large. + pub fn validate(&self) -> Result<(), BackgroundTaskError> { + match self { + Self::Command { label } | Self::Maintenance { label } => { + validate_label(label)?; + } + Self::HeadlessRendezvous { .. } | Self::SessionResume { .. } => {} + } + Ok(()) + } +} + +impl BackgroundTaskResult { + /// Returns the terminal timestamp carried by the result. + #[must_use] + pub fn finished_at_ms(&self) -> u64 { + match self { + Self::Success { finished_at_ms, .. } + | Self::Failure { finished_at_ms, .. } + | Self::Cancelled { finished_at_ms, .. } + | Self::Expired { finished_at_ms, .. } => *finished_at_ms, + } + } + + /// Validates result payload bounds. + /// + /// # Errors + /// [`BackgroundTaskError`] when a text field is too large. + pub fn validate(&self) -> Result<(), BackgroundTaskError> { + match self { + Self::Success { + summary, + stdout_tail, + stderr_tail, + .. + } => { + validate_text("summary", summary)?; + validate_tail("stdout_tail", stdout_tail.as_deref())?; + validate_tail("stderr_tail", stderr_tail.as_deref())?; + } + Self::Failure { + error, + stdout_tail, + stderr_tail, + .. + } => { + validate_text("error", error)?; + validate_tail("stdout_tail", stdout_tail.as_deref())?; + validate_tail("stderr_tail", stderr_tail.as_deref())?; + } + Self::Cancelled { reason, .. } | Self::Expired { reason, .. } => { + validate_text("reason", reason)?; + } + } + Ok(()) + } + + fn state(&self) -> BackgroundTaskState { + match self { + Self::Success { .. } => BackgroundTaskState::Completed, + Self::Failure { .. } => BackgroundTaskState::Failed, + Self::Cancelled { .. } => BackgroundTaskState::Cancelled, + Self::Expired { .. } => BackgroundTaskState::Expired, + } + } +} + +/// Errors raised by pure background task invariants. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum BackgroundTaskError { + /// A command/maintenance label is empty. + #[error("background task label is empty")] + EmptyLabel, + /// A bounded text field is too large. + #[error("background task field `{field}` exceeds {max} bytes")] + FieldTooLarge { + /// Field name. + field: &'static str, + /// Maximum allowed size in bytes. + max: usize, + }, + /// Timestamps are inconsistent. + #[error("invalid background task timestamps")] + InvalidTimestamps, + /// The requested state transition is not allowed. + #[error("invalid background task state transition from {from:?} to {to:?}")] + InvalidTransition { + /// Current state. + from: BackgroundTaskState, + /// Requested state. + to: BackgroundTaskState, + }, + /// A terminal state has no result, or delivery was attempted before terminal. + #[error("background task terminal state requires a result")] + MissingResult, + /// A non-terminal state carries a result. + #[error("background task non-terminal state cannot carry a result")] + UnexpectedResult, + /// Completion delivery was already marked. + #[error("background task completion was already delivered")] + CompletionAlreadyDelivered, +} + +fn is_transition_allowed(from: BackgroundTaskState, to: BackgroundTaskState) -> bool { + use BackgroundTaskState::{Cancelled, Completed, Expired, Failed, Queued, Running, Waiting}; + + matches!( + (from, to), + (Queued, Running | Cancelled | Expired) + | (Running, Waiting | Completed | Failed | Cancelled | Expired) + | (Waiting, Running | Completed | Failed | Cancelled | Expired) + ) +} + +fn validate_label(label: &str) -> Result<(), BackgroundTaskError> { + if label.trim().is_empty() { + return Err(BackgroundTaskError::EmptyLabel); + } + if label.chars().count() > BACKGROUND_TASK_LABEL_MAX_CHARS { + return Err(BackgroundTaskError::FieldTooLarge { + field: "label", + max: BACKGROUND_TASK_LABEL_MAX_CHARS, + }); + } + Ok(()) +} + +fn validate_text(field: &'static str, value: &str) -> Result<(), BackgroundTaskError> { + if value.len() > BACKGROUND_TASK_TEXT_MAX_BYTES { + return Err(BackgroundTaskError::FieldTooLarge { + field, + max: BACKGROUND_TASK_TEXT_MAX_BYTES, + }); + } + Ok(()) +} + +fn validate_tail(field: &'static str, value: Option<&str>) -> Result<(), BackgroundTaskError> { + if value.is_some_and(|value| value.len() > BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES) { + return Err(BackgroundTaskError::FieldTooLarge { + field, + max: BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES, + }); + } + Ok(()) +} diff --git a/crates/domain/src/events.rs b/crates/domain/src/events.rs index f1d100e..c6ba4b7 100644 --- a/crates/domain/src/events.rs +++ b/crates/domain/src/events.rs @@ -1,7 +1,7 @@ //! Domain events published on the [`crate::ports::EventBus`] and relayed to the //! presentation layer (ARCHITECTURE §3.2). -use crate::ids::{AgentId, ProfileId, ProjectId, SessionId, SkillId, TemplateId}; +use crate::ids::{AgentId, ProfileId, ProjectId, SessionId, SkillId, TaskId, TemplateId}; use crate::mailbox::TicketId; use crate::memory::MemorySlug; use crate::template::TemplateVersion; @@ -41,6 +41,108 @@ pub enum DomainEvent { /// The session it runs in. session_id: SessionId, }, + /// A first-class background task started running. + BackgroundTaskStarted { + /// The owning project. + project_id: ProjectId, + /// The task. + task_id: TaskId, + /// The agent that owns completion delivery. + owner_agent_id: AgentId, + }, + /// A first-class background task changed state. + BackgroundTaskStateChanged { + /// The owning project. + project_id: ProjectId, + /// The task. + task_id: TaskId, + /// The agent that owns completion delivery. + owner_agent_id: AgentId, + /// New state. + state: crate::background_task::BackgroundTaskState, + }, + /// A first-class background task completed successfully. + BackgroundTaskCompleted { + /// The owning project. + project_id: ProjectId, + /// The task. + task_id: TaskId, + /// The agent that owns completion delivery. + owner_agent_id: AgentId, + }, + /// A first-class background task failed. + BackgroundTaskFailed { + /// The owning project. + project_id: ProjectId, + /// The task. + task_id: TaskId, + /// The agent that owns completion delivery. + owner_agent_id: AgentId, + }, + /// A first-class background task was cancelled. + BackgroundTaskCancelled { + /// The owning project. + project_id: ProjectId, + /// The task. + task_id: TaskId, + /// The agent that owns completion delivery. + owner_agent_id: AgentId, + }, + /// A first-class background task has a terminal result not yet delivered. + BackgroundTaskCompletionDeliveryPending { + /// The owning project. + project_id: ProjectId, + /// The task. + task_id: TaskId, + /// The agent that owns completion delivery. + owner_agent_id: AgentId, + }, + /// A first-class background task completion was delivered. + BackgroundTaskCompletionDelivered { + /// The owning project. + project_id: ProjectId, + /// The task. + task_id: TaskId, + /// The agent that owns completion delivery. + owner_agent_id: AgentId, + }, + /// An item was queued in an agent inbox. + AgentInboxQueued { + /// Target agent. + agent_id: AgentId, + /// Queue depth after enqueue. + depth: usize, + }, + /// An item was drained from an agent inbox. + AgentInboxDrained { + /// Target agent. + agent_id: AgentId, + /// Queue depth after drain. + depth: usize, + }, + /// A wake was scheduled for an agent. + AgentWakeScheduled { + /// Owning project. + project_id: ProjectId, + /// Target agent. + agent_id: AgentId, + }, + /// A wake turn started for an agent. + AgentWakeStarted { + /// Owning project. + project_id: ProjectId, + /// Target agent. + agent_id: AgentId, + }, + /// A wake attempt failed. + AgentWakeFailed { + /// Owning project. + project_id: ProjectId, + /// Target agent. + agent_id: AgentId, + /// Human-readable reason. + reason: String, + }, /// A target agent produced a synchronous reply to an inter-agent `ask` /// (ARCHITECTURE §17.4): the requester sent a task via `agent.message`, IdeA /// drove the target's structured session to its turn `Final`, and this is the diff --git a/crates/domain/src/ids.rs b/crates/domain/src/ids.rs index 0e67a71..d96dfce 100644 --- a/crates/domain/src/ids.rs +++ b/crates/domain/src/ids.rs @@ -99,3 +99,7 @@ typed_id!( /// [`crate::ports::Scheduler::cancel`]. ScheduleId ); +typed_id!( + /// Identifies a first-class background task. + TaskId +); diff --git a/crates/domain/src/inbox.rs b/crates/domain/src/inbox.rs new file mode 100644 index 0000000..156197f --- /dev/null +++ b/crates/domain/src/inbox.rs @@ -0,0 +1,172 @@ +//! Agent inbox facade over the mediated input FIFO. +//! +//! The inbox is a bounded, per-agent view of incoming messages. It is a domain +//! port so infrastructure can implement it over the existing mediated mailbox +//! without creating a second competing queue. + +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use crate::ids::{AgentId, TaskId}; +use crate::mailbox::TicketId; + +/// Default bounded inbox capacity per agent. +pub const DEFAULT_AGENT_INBOX_CAPACITY: usize = 100; + +/// Source of an [`InboxItem`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", tag = "kind")] +pub enum InboxSource { + /// Human operator. + Human, + /// Another agent. + #[serde(rename_all = "camelCase")] + Agent { + /// Source agent. + agent_id: AgentId, + }, + /// A persisted background task completion. + #[serde(rename_all = "camelCase")] + BackgroundTask { + /// Source task. + task_id: TaskId, + }, + /// IdeA system message. + System, +} + +impl InboxSource { + /// Whether this source is lossless system work that must never be dropped. + #[must_use] + pub const fn is_lossless_system(self) -> bool { + matches!(self, Self::BackgroundTask { .. } | Self::System) + } +} + +/// Kind of inbox item. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum InboxItemKind { + /// Human message. + UserMessage, + /// Inter-agent delegation. + AgentDelegation, + /// Background task completion to deliver. + BackgroundCompletion, + /// Session resume notice. + ResumeNotice, +} + +impl InboxItemKind { + /// Whether this kind is lossless system work that must never be dropped. + #[must_use] + pub const fn is_lossless_system(self) -> bool { + matches!(self, Self::BackgroundCompletion | Self::ResumeNotice) + } +} + +/// Message queued for an agent. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct InboxItem { + /// Stable queue item id. Reuses `TicketId` because the underlying FIFO is the + /// mediated mailbox. + pub id: TicketId, + /// Target agent. + pub agent_id: AgentId, + /// Message source. + pub source: InboxSource, + /// Message kind. + pub kind: InboxItemKind, + /// Message body. + pub body: String, + /// Creation timestamp, epoch milliseconds. + pub created_at_ms: u64, + /// Optional correlation id, e.g. a task id. + pub correlation_id: Option, +} + +impl InboxItem { + /// Whether this item must remain persisted elsewhere rather than being lost + /// on inbox overflow. + #[must_use] + pub const fn is_lossless_system(&self) -> bool { + self.source.is_lossless_system() || self.kind.is_lossless_system() + } +} + +/// Result status of an enqueue. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum InboxReceiptStatus { + /// Item was queued in the in-memory FIFO. + Queued, + /// Item was not queued because capacity was full, but remains durable in its + /// authoritative store for later reconcile. + Deferred, +} + +/// Receipt returned by [`AgentInbox::enqueue_message`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct InboxReceipt { + /// Item id. + pub item_id: TicketId, + /// Target agent. + pub agent_id: AgentId, + /// FIFO depth after the operation. + pub depth: usize, + /// Enqueue outcome. + pub status: InboxReceiptStatus, +} + +/// Read-only inbox snapshot. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentInboxSnapshot { + /// Target agent. + pub agent_id: AgentId, + /// Number of queued inbox items. + pub depth: usize, + /// FIFO-ordered items. + pub items: Vec, +} + +/// Errors raised by [`AgentInbox`]. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum InboxError { + /// The bounded inbox is full. + #[error("agent inbox full for {agent_id} (capacity {capacity})")] + InboxFull { + /// Target agent. + agent_id: AgentId, + /// Configured capacity. + capacity: usize, + }, + /// The item target does not match the enqueue target. + #[error("inbox item target {item_agent_id} does not match enqueue target {agent_id}")] + AgentMismatch { + /// Enqueue target. + agent_id: AgentId, + /// Item target. + item_agent_id: AgentId, + }, +} + +/// Bounded per-agent inbox facade. +pub trait AgentInbox: Send + Sync { + /// Enqueues a message for `agent`. + /// + /// Normal messages fail with [`InboxError::InboxFull`] on overflow; lossless + /// system/completion items return [`InboxReceiptStatus::Deferred`] instead. + /// + /// # Errors + /// [`InboxError`] when the item is invalid or a normal message overflows. + fn enqueue_message(&self, agent: AgentId, item: InboxItem) -> Result; + + /// Pops the next queued inbox item, if any. + fn dequeue_next(&self, agent: AgentId) -> Option; + + /// Returns the FIFO snapshot for `agent`. + fn snapshot(&self, agent: AgentId) -> AgentInboxSnapshot; +} diff --git a/crates/domain/src/lib.rs b/crates/domain/src/lib.rs index 8865fe0..e41caf2 100644 --- a/crates/domain/src/lib.rs +++ b/crates/domain/src/lib.rs @@ -31,6 +31,7 @@ #![warn(missing_docs)] pub mod agent; +pub mod background_task; pub mod conversation; pub mod conversation_log; pub mod error; @@ -38,6 +39,7 @@ pub mod events; pub mod fileguard; pub mod git; pub mod ids; +pub mod inbox; pub mod input; pub mod layout; pub mod live_state; @@ -67,7 +69,7 @@ mod validation; pub use error::DomainError; pub use ids::{ - AgentId, LayoutId, NodeId, ProfileId, ProjectId, ScheduleId, SessionId, SkillId, TabId, + AgentId, LayoutId, NodeId, ProfileId, ProjectId, ScheduleId, SessionId, SkillId, TabId, TaskId, TemplateId, WindowId, }; @@ -75,6 +77,12 @@ pub use project::{Project, ProjectPath}; pub use agent::{Agent, AgentManifest, AgentOrigin, ManifestEntry}; +pub use background_task::{ + BackgroundTask, BackgroundTaskError, BackgroundTaskKind, BackgroundTaskResult, + BackgroundTaskState, BackgroundTaskWakePolicy, BACKGROUND_TASK_LABEL_MAX_CHARS, + BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES, BACKGROUND_TASK_TEXT_MAX_BYTES, +}; + pub use skill::{Skill, SkillRef, SkillScope}; pub use template::{AgentTemplate, TemplateVersion}; @@ -96,6 +104,11 @@ pub use conversation::{ pub use input::{AgentBusyState, AgentLiveness, InputMediator, InputSource}; +pub use inbox::{ + AgentInbox, AgentInboxSnapshot, InboxError, InboxItem, InboxItemKind, InboxReceipt, + InboxReceiptStatus, InboxSource, DEFAULT_AGENT_INBOX_CAPACITY, +}; + pub use live_state::{LiveEntry, LiveState, WorkStatus, FIELD_MAX_BYTES, FIELD_PREVIEW_MAX_CHARS}; pub use readiness::{ReadinessPolicy, ReadinessSignal}; @@ -154,12 +167,14 @@ pub use orchestrator::{ }; pub use ports::{ - AgentContextStore, AgentRuntime, Clock, ContextInjectionPlan, DirEntry, Embedder, - EmbedderEnvInspector, EmbedderEnvReport, EmbedderError, EmbedderProfileStore, - EmbedderPromptDismissal, EmbedderPromptStore, EventBus, EventStream, ExitStatus, FileSystem, - FsError, GitCommitInfo, GitError, GitFileStatus, GitPort, GraphCommit, IdGenerator, - LiveStateStore, MemoryError, MemoryQuery, MemoryRecall, MemoryStore, Output, OutputStream, - PermissionStore, PreparedContext, ProcessError, ProcessSpawner, ProfileStore, ProjectStore, - PtyError, PtyHandle, PtyPort, RemoteError, RemoteHost, RemotePath, RuntimeError, ScheduledTask, - Scheduler, SpawnSpec, StoreError, TemplateStore, + AgentContextStore, AgentRuntime, BackgroundCompletionStream, BackgroundTaskCompletion, + BackgroundTaskHandle, BackgroundTaskPortError, BackgroundTaskRunner, BackgroundTaskSpec, + BackgroundTaskStore, Clock, ContextInjectionPlan, DirEntry, Embedder, EmbedderEnvInspector, + EmbedderEnvReport, EmbedderError, EmbedderProfileStore, EmbedderPromptDismissal, + EmbedderPromptStore, EventBus, EventStream, ExitStatus, FileSystem, FsError, GitCommitInfo, + GitError, GitFileStatus, GitPort, GraphCommit, IdGenerator, LiveStateStore, MemoryError, + MemoryQuery, MemoryRecall, MemoryStore, Output, OutputStream, PermissionStore, PreparedContext, + ProcessError, ProcessSpawner, ProfileStore, ProjectStore, PtyError, PtyHandle, PtyPort, + RemoteError, RemoteHost, RemotePath, RuntimeError, ScheduledTask, Scheduler, SpawnSpec, + StoreError, TemplateStore, }; diff --git a/crates/domain/src/ports.rs b/crates/domain/src/ports.rs index 470740d..fbf6b45 100644 --- a/crates/domain/src/ports.rs +++ b/crates/domain/src/ports.rs @@ -28,8 +28,11 @@ use async_trait::async_trait; use thiserror::Error; use crate::agent::AgentManifest; +use crate::background_task::{ + BackgroundTask, BackgroundTaskKind, BackgroundTaskResult, BackgroundTaskWakePolicy, +}; use crate::events::DomainEvent; -use crate::ids::{AgentId, NodeId, ScheduleId, SessionId}; +use crate::ids::{AgentId, NodeId, ProjectId, ScheduleId, SessionId, TaskId}; use crate::markdown::MarkdownDoc; use crate::memory::{Memory, MemoryIndexEntry, MemoryLink, MemorySlug}; use crate::permission::ProjectPermissions; @@ -196,6 +199,59 @@ pub type OutputStream = Box> + Send>; /// A boxed stream of domain events, returned by [`EventBus::subscribe`]. pub type EventStream = Box + Send>; +/// A boxed stream of background task completions. +pub type BackgroundCompletionStream = Box + Send>; + +/// Specification passed to a background task runner. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BackgroundTaskSpec { + /// Task id allocated by the application. + pub task_id: TaskId, + /// Owning project. + pub project_id: ProjectId, + /// Agent that owns completion delivery. + pub owner_agent_id: AgentId, + /// Kind-specific task payload. + pub kind: BackgroundTaskKind, + /// Completion wake policy. + pub wake_policy: BackgroundTaskWakePolicy, + /// Optional command invocation for command-backed tasks. + pub command: Option, + /// Optional absolute deadline, epoch milliseconds. + pub deadline_ms: Option, +} + +/// Opaque runner handle for a spawned background task. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BackgroundTaskHandle { + /// Spawned task id. + pub task_id: TaskId, +} + +/// Terminal completion reported by a background runner. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BackgroundTaskCompletion { + /// Completed task id. + pub task_id: TaskId, + /// Terminal result. + pub result: BackgroundTaskResult, +} + +/// Why IdeA wakes an agent outside a human/delegation submit. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WakeReason { + /// A persisted background task completion is ready in the owner's inbox. + BackgroundCompletion { + /// Completed background task. + task_id: TaskId, + }, + /// A generic queued inbox item should be processed. + InboxItem { + /// Queued item id. + item_id: crate::mailbox::TicketId, + }, +} + /// Un événement incrémental d'un tour de réponse d'un agent IA (ARCHITECTURE §17.1). /// /// Universel : l'adapter (Claude/Codex) traduit SON format structuré documenté @@ -461,6 +517,55 @@ pub enum GitError { Operation(String), } +/// Errors from first-class background task ports. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum BackgroundTaskPortError { + /// The requested task was not found. + #[error("background task not found")] + NotFound, + /// A task with this id already exists. + #[error("background task already exists")] + AlreadyExists, + /// The task or request is invalid. + #[error("background task invalid: {0}")] + Invalid(String), + /// The runner failed. + #[error("background task runner failed: {0}")] + Runner(String), + /// The store failed. + #[error("background task store failed: {0}")] + Store(String), +} + +/// Errors from the owner wake port. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum WakeError { + /// The owner is already running a turn; the inbox item must stay queued. + #[error("agent {agent_id} is busy; wake skipped")] + AgentBusy { + /// Target agent. + agent_id: AgentId, + }, + /// There is no queued item to drain for this wake. + #[error("no pending inbox item for agent {agent_id}")] + NoPendingItem { + /// Target agent. + agent_id: AgentId, + }, + /// The drained inbox item does not match the requested wake reason. + #[error("unexpected inbox item for wake: {0}")] + UnexpectedInboxItem(String), + /// Session lookup or launch failed. + #[error("wake session failed: {0}")] + Session(String), + /// Background task store failed. + #[error("wake store failed: {0}")] + Store(String), + /// Background task payload was missing or invalid for wake delivery. + #[error("wake task invalid: {0}")] + Task(String), +} + // --------------------------------------------------------------------------- // Git port support types // --------------------------------------------------------------------------- @@ -1191,6 +1296,95 @@ pub trait PermissionStore: Send + Sync { ) -> Result<(), StoreError>; } +/// Persistence port for first-class background tasks. +#[async_trait] +pub trait BackgroundTaskStore: Send + Sync { + /// Creates a new task. + /// + /// # Errors + /// [`BackgroundTaskPortError`] if the task already exists or cannot be stored. + async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError>; + + /// Gets a task by id. + /// + /// # Errors + /// [`BackgroundTaskPortError`] on store failure. + async fn get(&self, id: TaskId) -> Result, BackgroundTaskPortError>; + + /// Saves a task. + /// + /// # Errors + /// [`BackgroundTaskPortError`] on store failure. + async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError>; + + /// Lists non-terminal tasks owned by an agent. + /// + /// # Errors + /// [`BackgroundTaskPortError`] on store failure. + async fn list_open_for_agent( + &self, + agent_id: AgentId, + ) -> Result, BackgroundTaskPortError>; + + /// Lists terminal tasks whose completion has not been delivered yet. + /// + /// # Errors + /// [`BackgroundTaskPortError`] on store failure. + async fn list_undelivered_completions( + &self, + ) -> Result, BackgroundTaskPortError>; + + /// Marks a completion as delivered. + /// + /// # Errors + /// [`BackgroundTaskPortError`] on store failure. + async fn mark_completion_delivered( + &self, + task_id: TaskId, + ) -> Result<(), BackgroundTaskPortError>; +} + +/// Execution port for first-class background tasks. +#[async_trait] +pub trait BackgroundTaskRunner: Send + Sync { + /// Spawns a background task. + /// + /// # Errors + /// [`BackgroundTaskPortError`] when the task cannot be spawned. + async fn spawn( + &self, + spec: BackgroundTaskSpec, + ) -> Result; + + /// Cancels a running task. + /// + /// # Errors + /// [`BackgroundTaskPortError`] when cancellation fails. + async fn cancel(&self, task_id: TaskId) -> Result<(), BackgroundTaskPortError>; + + /// Subscribes to terminal completions. + fn subscribe_completions(&self) -> BackgroundCompletionStream; +} + +/// Wakes an agent to process one queued system/inbox item. +#[async_trait] +pub trait AgentWakePort: Send + Sync { + /// Wakes `agent` in `project` for `reason`, if the agent is idle. + /// + /// Implementations must drain at most one inbox item and must not start a + /// concurrent turn when the agent is already busy. + /// + /// # Errors + /// [`WakeError`] when the session/store operation fails or the queued item is + /// inconsistent with the requested wake. + async fn wake_agent( + &self, + project: &Project, + agent: AgentId, + reason: WakeReason, + ) -> Result<(), WakeError>; +} + /// Git operations for a project. Named `GitPort` to avoid clashing with the /// [`crate::git::GitRepository`] *entity* (state image). #[async_trait] diff --git a/crates/domain/tests/background_task.rs b/crates/domain/tests/background_task.rs new file mode 100644 index 0000000..8b1ee49 --- /dev/null +++ b/crates/domain/tests/background_task.rs @@ -0,0 +1,302 @@ +//! Pure invariants for first-class background tasks. + +use domain::{ + BackgroundTask, BackgroundTaskError, BackgroundTaskKind, BackgroundTaskResult, + BackgroundTaskState, BackgroundTaskWakePolicy, +}; +use uuid::Uuid; + +fn task_id(n: u128) -> domain::TaskId { + domain::TaskId::from_uuid(Uuid::from_u128(n)) +} + +fn project_id(n: u128) -> domain::ProjectId { + domain::ProjectId::from_uuid(Uuid::from_u128(n)) +} + +fn agent_id(n: u128) -> domain::AgentId { + domain::AgentId::from_uuid(Uuid::from_u128(n)) +} + +fn command_task() -> BackgroundTask { + BackgroundTask::new( + task_id(1), + project_id(2), + agent_id(3), + BackgroundTaskKind::Command { + label: "build".into(), + }, + BackgroundTaskWakePolicy::WakeOwner, + 100, + Some(200), + ) + .unwrap() +} + +fn success_result(at: u64) -> BackgroundTaskResult { + BackgroundTaskResult::Success { + finished_at_ms: at, + exit_code: Some(0), + summary: "ok".into(), + stdout_tail: None, + stderr_tail: None, + } +} + +fn failure_result(at: u64) -> BackgroundTaskResult { + BackgroundTaskResult::Failure { + finished_at_ms: at, + exit_code: Some(1), + error: "failed".into(), + stdout_tail: None, + stderr_tail: None, + } +} + +#[test] +fn new_task_is_queued_without_result_or_delivery() { + let task = command_task(); + + assert_eq!(task.state, BackgroundTaskState::Queued); + assert_eq!(task.result, None); + assert!(!task.completion_delivered); + assert!(!task.is_terminal()); + assert!(!task.has_pending_completion_delivery()); +} + +#[test] +fn deadline_must_not_precede_creation() { + let err = BackgroundTask::new( + task_id(1), + project_id(2), + agent_id(3), + BackgroundTaskKind::Command { + label: "build".into(), + }, + BackgroundTaskWakePolicy::WakeOwner, + 100, + Some(99), + ) + .unwrap_err(); + + assert_eq!(err, BackgroundTaskError::InvalidTimestamps); +} + +#[test] +fn labels_must_be_non_empty_and_bounded() { + let empty = BackgroundTaskKind::Command { label: " ".into() }; + assert_eq!(empty.validate(), Err(BackgroundTaskError::EmptyLabel)); + + let oversized = BackgroundTaskKind::Maintenance { + label: "x".repeat(domain::BACKGROUND_TASK_LABEL_MAX_CHARS + 1), + }; + assert_eq!( + oversized.validate(), + Err(BackgroundTaskError::FieldTooLarge { + field: "label", + max: domain::BACKGROUND_TASK_LABEL_MAX_CHARS, + }) + ); +} + +#[test] +fn allowed_non_terminal_transitions_update_timestamp() { + let running = command_task() + .transition(BackgroundTaskState::Running, 120) + .unwrap(); + assert_eq!(running.state, BackgroundTaskState::Running); + assert_eq!(running.updated_at_ms, 120); + + let waiting = running + .transition(BackgroundTaskState::Waiting, 130) + .unwrap(); + assert_eq!(waiting.state, BackgroundTaskState::Waiting); + + let running_again = waiting + .transition(BackgroundTaskState::Running, 140) + .unwrap(); + assert_eq!(running_again.state, BackgroundTaskState::Running); +} + +#[test] +fn illegal_transitions_are_rejected() { + let task = command_task(); + let err = task + .transition(BackgroundTaskState::Waiting, 120) + .unwrap_err(); + + assert_eq!( + err, + BackgroundTaskError::InvalidTransition { + from: BackgroundTaskState::Queued, + to: BackgroundTaskState::Waiting, + } + ); +} + +#[test] +fn lifecycle_timestamps_must_not_regress() { + let running = command_task() + .transition(BackgroundTaskState::Running, 120) + .unwrap(); + + assert_eq!( + running.transition(BackgroundTaskState::Waiting, 119), + Err(BackgroundTaskError::InvalidTimestamps) + ); + assert_eq!( + running.complete(success_result(119)), + Err(BackgroundTaskError::InvalidTimestamps) + ); +} + +#[test] +fn terminal_transition_requires_complete_with_matching_result() { + let running = command_task() + .transition(BackgroundTaskState::Running, 120) + .unwrap(); + + assert_eq!( + running.transition(BackgroundTaskState::Completed, 130), + Err(BackgroundTaskError::MissingResult) + ); + + let completed = running.complete(success_result(130)).unwrap(); + assert_eq!(completed.state, BackgroundTaskState::Completed); + assert_eq!(completed.updated_at_ms, 130); + assert!(completed.is_terminal()); + assert!(completed.has_pending_completion_delivery()); +} + +#[test] +fn terminal_states_are_immutable() { + let completed = command_task() + .transition(BackgroundTaskState::Running, 120) + .unwrap() + .complete(success_result(130)) + .unwrap(); + + let err = completed + .transition(BackgroundTaskState::Running, 140) + .unwrap_err(); + assert_eq!( + err, + BackgroundTaskError::InvalidTransition { + from: BackgroundTaskState::Completed, + to: BackgroundTaskState::Running, + } + ); + + let err = completed.complete(failure_result(140)).unwrap_err(); + assert_eq!( + err, + BackgroundTaskError::InvalidTransition { + from: BackgroundTaskState::Completed, + to: BackgroundTaskState::Failed, + } + ); +} + +#[test] +fn result_variants_determine_terminal_state() { + let running = command_task() + .transition(BackgroundTaskState::Running, 120) + .unwrap(); + + assert_eq!( + running.complete(failure_result(130)).unwrap().state, + BackgroundTaskState::Failed + ); + assert_eq!( + running + .complete(BackgroundTaskResult::Cancelled { + finished_at_ms: 130, + reason: "user".into(), + }) + .unwrap() + .state, + BackgroundTaskState::Cancelled + ); + assert_eq!( + running + .complete(BackgroundTaskResult::Expired { + finished_at_ms: 130, + reason: "deadline".into(), + }) + .unwrap() + .state, + BackgroundTaskState::Expired + ); +} + +#[test] +fn completion_delivery_only_once_after_terminal_state() { + let running = command_task() + .transition(BackgroundTaskState::Running, 120) + .unwrap(); + + assert_eq!( + running.mark_completion_delivered(), + Err(BackgroundTaskError::MissingResult) + ); + + let delivered = running + .complete(success_result(130)) + .unwrap() + .mark_completion_delivered() + .unwrap(); + + assert!(delivered.completion_delivered); + assert!(!delivered.has_pending_completion_delivery()); + assert_eq!( + delivered.mark_completion_delivered(), + Err(BackgroundTaskError::CompletionAlreadyDelivered) + ); +} + +#[test] +fn result_text_fields_are_bounded() { + let summary_too_large = BackgroundTaskResult::Success { + finished_at_ms: 1, + exit_code: None, + summary: "x".repeat(domain::BACKGROUND_TASK_TEXT_MAX_BYTES + 1), + stdout_tail: None, + stderr_tail: None, + }; + assert_eq!( + summary_too_large.validate(), + Err(BackgroundTaskError::FieldTooLarge { + field: "summary", + max: domain::BACKGROUND_TASK_TEXT_MAX_BYTES, + }) + ); + + let tail_too_large = BackgroundTaskResult::Failure { + finished_at_ms: 1, + exit_code: None, + error: "failed".into(), + stdout_tail: Some("x".repeat(domain::BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES + 1)), + stderr_tail: None, + }; + assert_eq!( + tail_too_large.validate(), + Err(BackgroundTaskError::FieldTooLarge { + field: "stdout_tail", + max: domain::BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES, + }) + ); +} + +#[test] +fn background_task_serializes_with_camel_case_fields() { + let task = command_task(); + let json = serde_json::to_string(&task).unwrap(); + + assert!(json.contains("\"ownerAgentId\""), "json was {json}"); + assert!( + json.contains("\"wakePolicy\":\"wakeOwner\""), + "json was {json}" + ); + assert!(json.contains("\"completionDelivered\""), "json was {json}"); + assert_eq!(serde_json::from_str::(&json).unwrap(), task); +}