Entités BackgroundTask et Inbox, identifiants, événements et ports du domaine pour la récupération de complétion post-tour et la réception de messages concurrents. Fondations des lots infra/application suivants. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
477 lines
15 KiB
Rust
477 lines
15 KiB
Rust
//! 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<AgentId>,
|
|
/// 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<ScheduleId>,
|
|
/// Provider conversation id to resume, when any.
|
|
conversation_id: Option<String>,
|
|
},
|
|
/// 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<u64>,
|
|
/// Terminal result. Present if and only if the state is terminal.
|
|
pub result: Option<BackgroundTaskResult>,
|
|
/// 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<u64>,
|
|
) -> Result<Self, BackgroundTaskError> {
|
|
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<Self, BackgroundTaskError> {
|
|
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<Self, BackgroundTaskError> {
|
|
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<Self, BackgroundTaskError> {
|
|
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<i32>,
|
|
/// Human-readable summary.
|
|
summary: String,
|
|
/// Bounded stdout tail.
|
|
stdout_tail: Option<String>,
|
|
/// Bounded stderr tail.
|
|
stderr_tail: Option<String>,
|
|
},
|
|
/// Failed completion.
|
|
Failure {
|
|
/// Completion timestamp, epoch milliseconds.
|
|
finished_at_ms: u64,
|
|
/// Process exit code, when command-backed.
|
|
exit_code: Option<i32>,
|
|
/// Human-readable error.
|
|
error: String,
|
|
/// Bounded stdout tail.
|
|
stdout_tail: Option<String>,
|
|
/// Bounded stderr tail.
|
|
stderr_tail: Option<String>,
|
|
},
|
|
/// 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(())
|
|
}
|