From f94b54239bccb724ad1d70fb43a792e4bc657304 Mon Sep 17 00:00:00 2001 From: Blomios Date: Thu, 2 Jul 2026 15:47:34 +0200 Subject: [PATCH] =?UTF-8?q?feat(application):=20r=C3=A9veil=20d'agent=20et?= =?UTF-8?q?=20rendez-vous=20comme=20t=C3=A2che=20de=20fond=20(B5-B6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Orchestration du réveil (wake) d'un agent sur complétion/message et traitement du rendez-vous inter-agent comme tâche de fond de 1re classe. Couvert par agent_wake (vert). Co-Authored-By: Claude Opus 4.8 --- crates/application/src/lib.rs | 8 +- crates/application/src/orchestrator/mod.rs | 2 + .../application/src/orchestrator/service.rs | 249 ++++++++- crates/application/src/orchestrator/wake.rs | 316 +++++++++++ crates/application/tests/agent_wake.rs | 524 ++++++++++++++++++ 5 files changed, 1092 insertions(+), 7 deletions(-) create mode 100644 crates/application/src/orchestrator/wake.rs create mode 100644 crates/application/tests/agent_wake.rs diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index 0e56e1e..86a1c09 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -85,10 +85,10 @@ pub use memory::{ }; pub use orchestrator::{ resolve_rendezvous_ceiling, resolve_rendezvous_window, run_inactivity_watchdog, - AskLivenessProbe, ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, - McpRuntimeProvider, OrchestratorOutcome, OrchestratorService, ProposeContext, ReadContext, - ReadMemory, RecordTurnProvider, WatchdogOutcome, WriteMemory, DEFAULT_RENDEZVOUS_CEILING, - DEFAULT_RENDEZVOUS_WINDOW, + AgentWakeService, AskLivenessProbe, ContextGuardUseCases, LiveStateProvider, + LiveStateReadProvider, McpRuntimeProvider, OrchestratorOutcome, OrchestratorService, + ProposeContext, ReadContext, ReadMemory, RecordTurnProvider, WakeSessionProvider, + WatchdogOutcome, WriteMemory, DEFAULT_RENDEZVOUS_CEILING, DEFAULT_RENDEZVOUS_WINDOW, }; pub use permission::{ GetProjectPermissions, GetProjectPermissionsInput, GetProjectPermissionsOutput, diff --git a/crates/application/src/orchestrator/mod.rs b/crates/application/src/orchestrator/mod.rs index a204525..bc27242 100644 --- a/crates/application/src/orchestrator/mod.rs +++ b/crates/application/src/orchestrator/mod.rs @@ -7,6 +7,7 @@ mod context_guard; pub mod rendezvous; mod service; +pub mod wake; pub use context_guard::{ ProposeContext, ProposeContextInput, ProposeOutcome, ReadContext, ReadContextInput, ReadMemory, @@ -20,3 +21,4 @@ pub use service::{ AskLivenessProbe, ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider, OrchestratorOutcome, OrchestratorService, RecordTurnProvider, }; +pub use wake::{AgentWakeService, WakeSessionProvider}; diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 4fe973e..3b3a811 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -23,11 +23,12 @@ use domain::conversation::{ConversationParty, ConversationRegistry, SessionRef, use domain::conversation_log::{ConversationTurn, TurnId, TurnRole}; use domain::input::{InputMediator, InputSource, SubmitConfig}; use domain::mailbox::{Ticket, TicketId}; -use domain::ports::{Clock, EventBus, ProfileStore, PtyHandle}; +use domain::ports::{BackgroundTaskStore, Clock, EventBus, ProfileStore, PtyHandle}; use domain::project::ProjectPath; use domain::{ - AgentId, AgentProfile, DomainEvent, OrchestratorCommand, OrchestratorVisibility, ProfileId, - Project, + AgentId, AgentProfile, BackgroundTask, BackgroundTaskKind, BackgroundTaskResult, + BackgroundTaskState, BackgroundTaskWakePolicy, DomainEvent, OrchestratorCommand, + OrchestratorVisibility, ProfileId, Project, TaskId, }; use crate::conversation::RecordTurn; @@ -77,6 +78,17 @@ fn structured_no_reply_error(err: &domain::ports::AgentSessionError) -> bool { ) } +fn bound_task_text(value: &str, max_bytes: usize) -> String { + if value.len() <= max_bytes { + return value.to_owned(); + } + let mut end = max_bytes; + while !value.is_char_boundary(end) { + end -= 1; + } + value[..end].to_owned() +} + /// Bound on the synchronous inter-agent rendezvous (`agent.message` → `AskAgent`). /// /// A target agent's turn can be long (reasoning + tool use), so the cap is @@ -417,6 +429,10 @@ pub struct OrchestratorService { /// [`Self::with_ask_ceiling`] (la composition root résout l'override d'env) ; les /// tests le rétrécissent à quelques ms. Défaut = 4 h. ask_ceiling: Duration, + /// Store durable des tâches de fond de 1re classe. B6 l'utilise pour tracer un + /// `idea_ask_agent` comme [`BackgroundTaskKind::HeadlessRendezvous`]. `None` ⇒ + /// comportement historique sans persistance de tâche (call sites non câblés). + background_tasks: Option>, } /// Bundle des quatre use cases C7 sous [`domain::fileguard::FileGuard`], injectés @@ -487,9 +503,27 @@ impl OrchestratorService { live_state_read: None, ask_liveness_probe: None, ask_ceiling: ASK_AGENT_CEILING, + background_tasks: None, } } + /// Branche le store des tâches de fond pour tracer les rendez-vous inter-agents + /// comme [`BackgroundTaskKind::HeadlessRendezvous`] (B6). + /// + /// Le `clock` fournit les timestamps métier de la tâche. Il réutilise le champ + /// horloge applicatif déjà existant ; si un provider de conversation log l'avait + /// déjà posé, la même horloge reste partagée. + #[must_use] + pub fn with_background_tasks( + mut self, + store: Arc, + clock: Arc, + ) -> Self { + self.background_tasks = Some(store); + self.clock = Some(clock); + self + } + /// Branche la **sonde de vivacité** (signe de vie) du rendez-vous délégué : la borne /// de tour devient une **fenêtre d'inactivité** réarmée à chaque progrès observé, /// sous le plafond [`Self::with_ask_ceiling`]. Builder additif (signature de @@ -610,6 +644,162 @@ impl OrchestratorService { self } + fn now_ms(&self) -> u64 { + self.clock + .as_ref() + .map_or(0, |clock| clock.now_millis().max(0) as u64) + } + + #[allow(clippy::too_many_arguments)] + async fn start_rendezvous_task( + &self, + project: &Project, + owner_agent_id: AgentId, + requester_agent_id: Option, + target_agent_id: AgentId, + ticket_id: TicketId, + conversation_id: domain::conversation::ConversationId, + ) -> Result, AppError> { + let Some(store) = &self.background_tasks else { + return Ok(None); + }; + let task_id = TaskId::new_random(); + let now = self.now_ms(); + let task = BackgroundTask::new( + task_id, + project.id, + owner_agent_id, + BackgroundTaskKind::HeadlessRendezvous { + requester_agent_id, + target_agent_id, + ticket_id, + conversation_id, + }, + BackgroundTaskWakePolicy::RecordOnly, + now, + None, + ) + .map_err(|err| AppError::Internal(err.to_string()))? + .transition(BackgroundTaskState::Running, now) + .map_err(|err| AppError::Internal(err.to_string()))?; + store + .create(&task) + .await + .map_err(|err| AppError::Store(err.to_string()))?; + if let Some(events) = &self.events { + events.publish(DomainEvent::BackgroundTaskStarted { + project_id: project.id, + task_id, + owner_agent_id, + }); + } + Ok(Some(task_id)) + } + + async fn complete_rendezvous_task_success( + &self, + project: &Project, + owner_agent_id: AgentId, + task_id: Option, + reply: &str, + ) -> Result<(), AppError> { + let result = BackgroundTaskResult::Success { + finished_at_ms: self.now_ms(), + exit_code: None, + summary: "Headless rendezvous completed with Final".to_owned(), + stdout_tail: Some(bound_task_text( + reply, + domain::BACKGROUND_TASK_OUTPUT_TAIL_MAX_BYTES, + )), + stderr_tail: None, + }; + self.finish_rendezvous_task(project, owner_agent_id, task_id, result) + .await + } + + async fn complete_rendezvous_task_failure( + &self, + project: &Project, + owner_agent_id: AgentId, + task_id: Option, + error: impl Into, + ) -> Result<(), AppError> { + let result = BackgroundTaskResult::Failure { + finished_at_ms: self.now_ms(), + exit_code: None, + error: bound_task_text(&error.into(), domain::BACKGROUND_TASK_TEXT_MAX_BYTES), + stdout_tail: None, + stderr_tail: None, + }; + self.finish_rendezvous_task(project, owner_agent_id, task_id, result) + .await + } + + async fn complete_rendezvous_task_cancelled( + &self, + project: &Project, + owner_agent_id: AgentId, + task_id: Option, + reason: impl Into, + ) -> Result<(), AppError> { + let result = BackgroundTaskResult::Cancelled { + finished_at_ms: self.now_ms(), + reason: bound_task_text(&reason.into(), domain::BACKGROUND_TASK_TEXT_MAX_BYTES), + }; + self.finish_rendezvous_task(project, owner_agent_id, task_id, result) + .await + } + + async fn finish_rendezvous_task( + &self, + project: &Project, + owner_agent_id: AgentId, + task_id: Option, + result: BackgroundTaskResult, + ) -> Result<(), AppError> { + let (Some(store), Some(task_id)) = (&self.background_tasks, task_id) else { + return Ok(()); + }; + let task = store + .get(task_id) + .await + .map_err(|err| AppError::Store(err.to_string()))? + .ok_or_else(|| AppError::Store(format!("background task {task_id} not found")))?; + if task.is_terminal() { + return Ok(()); + } + let event = match &result { + BackgroundTaskResult::Success { .. } => DomainEvent::BackgroundTaskCompleted { + project_id: project.id, + task_id, + owner_agent_id, + }, + BackgroundTaskResult::Failure { .. } => DomainEvent::BackgroundTaskFailed { + project_id: project.id, + task_id, + owner_agent_id, + }, + BackgroundTaskResult::Cancelled { .. } | BackgroundTaskResult::Expired { .. } => { + DomainEvent::BackgroundTaskCancelled { + project_id: project.id, + task_id, + owner_agent_id, + } + } + }; + let completed = task + .complete(result) + .map_err(|err| AppError::Internal(err.to_string()))?; + store + .save(&completed) + .await + .map_err(|err| AppError::Store(err.to_string()))?; + if let Some(events) = &self.events { + events.publish(event); + } + Ok(()) + } + /// Branche le [`ConversationRegistry`] (cadrage C3 §5.2) pour résoudre /// paresseusement le fil `A↔B` (ou `User↔B`) d'un `ask` et séparer les contextes. /// Builder additif (signature de [`Self::new`] inchangée). @@ -1458,6 +1648,16 @@ impl OrchestratorService { // l'enqueue qui démarre le tour (le médiateur arme alors sa fenêtre de vivacité). let turn_timeout = self.turn_timeout_for(project, agent_id).await; let _pending = input.enqueue_silent(agent_id, ticket); + let rendezvous_task = self + .start_rendezvous_task( + project, + agent_id, + requester, + agent_id, + ticket_id, + conversation_id, + ) + .await?; // Auto-update live-state (lot LS3), best-effort : la cible passe `Working` sur // cette transition d'`ask` acceptée (chemin structuré). `task` est encore vivant // ici (utilisé par le drain plus bas), on le distille directement. @@ -1514,6 +1714,8 @@ impl OrchestratorService { .await { WatchdogOutcome::Resolved(Ok(content)) => { + self.complete_rendezvous_task_success(project, agent_id, rendezvous_task, &content) + .await?; crate::diag!( "[rendezvous] ask resolved (structured): target={target} \ (agent {agent_id}) ticket={ticket_id} after_ms={} reply_len={}", @@ -1526,6 +1728,33 @@ impl OrchestratorService { // `cancel_head` + `mark_idle` au Drop. On réconcilie la live-state à `Done` // pour qu'un abandon ne laisse pas un busy fantôme. WatchdogOutcome::Resolved(Err(err)) => { + let task_error = if matches!( + err, + AppError::TargetReturnedNoReply(_) | AppError::Process(_) + ) { + format!("NoReply: {err}") + } else { + err.to_string() + }; + if task_error.to_ascii_lowercase().contains("cancelled") + || task_error.to_ascii_lowercase().contains("annul") + { + self.complete_rendezvous_task_cancelled( + project, + agent_id, + rendezvous_task, + task_error, + ) + .await?; + } else { + self.complete_rendezvous_task_failure( + project, + agent_id, + rendezvous_task, + task_error, + ) + .await?; + } self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) .await; crate::diag!( @@ -1537,6 +1766,13 @@ impl OrchestratorService { } // Plafond absolu atteint alors que la cible progresse encore : verdict distinct. WatchdogOutcome::CeilingActive => { + self.complete_rendezvous_task_failure( + project, + agent_id, + rendezvous_task, + format!("Timeout: rendezvous absolute ceiling reached for target {target}"), + ) + .await?; self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) .await; crate::diag!( @@ -1549,6 +1785,13 @@ impl OrchestratorService { // Vrai silence (aucun progrès sur la fenêtre) : timeout typé, identique à // l'ancienne borne plate. Live-state réconciliée à `Done`. WatchdogOutcome::NoReply => { + self.complete_rendezvous_task_failure( + project, + agent_id, + rendezvous_task, + format!("Timeout: rendezvous inactivity window expired for target {target}"), + ) + .await?; self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) .await; crate::diag!( diff --git a/crates/application/src/orchestrator/wake.rs b/crates/application/src/orchestrator/wake.rs new file mode 100644 index 0000000..f785ca5 --- /dev/null +++ b/crates/application/src/orchestrator/wake.rs @@ -0,0 +1,316 @@ +//! Owner wake service for first-class background task completions. +//! +//! This application adapter implements the domain [`AgentWakePort`] above the +//! existing structured/headless session path. It owns no concrete process or +//! Tauri wiring: B7 injects the session provider that reuses +//! `StructuredSessions`/`LaunchAgent`. + +use std::sync::Arc; + +use async_trait::async_trait; +use domain::background_task::{BackgroundTask, BackgroundTaskResult}; +use domain::events::DomainEvent; +use domain::ids::AgentId; +use domain::inbox::{AgentInbox, InboxItem, InboxItemKind, InboxSource}; +use domain::input::InputMediator; +use domain::mailbox::{AgentMailbox, Ticket}; +use domain::ports::{ + AgentSession, AgentWakePort, BackgroundTaskStore, EventBus, WakeError, WakeReason, +}; +use domain::project::Project; + +use crate::agent::drain_with_readiness; + +/// Provides a structured/headless session for a wake turn. +/// +/// The production implementation is expected to return the live +/// `StructuredSessions` entry when present, or launch/reattach the agent using the +/// same mechanism as inter-agent delegation when absent. +#[async_trait] +pub trait WakeSessionProvider: Send + Sync { + /// Returns or starts the structured session used to send the wake prompt. + /// + /// # Errors + /// [`WakeError::Session`] when the agent cannot be driven headlessly. + async fn session_for_wake( + &self, + project: &Project, + agent: AgentId, + ) -> Result, WakeError>; +} + +/// Drain-then-wake implementation of [`AgentWakePort`]. +pub struct AgentWakeService { + inbox: Arc, + input: Arc, + mailbox: Arc, + tasks: Arc, + sessions: Arc, + events: Option>, +} + +impl AgentWakeService { + /// Builds a wake service from its ports. + #[must_use] + pub fn new( + inbox: Arc, + input: Arc, + mailbox: Arc, + tasks: Arc, + sessions: Arc, + events: Option>, + ) -> Self { + Self { + inbox, + input, + mailbox, + tasks, + sessions, + events, + } + } + + fn publish(&self, event: DomainEvent) { + if let Some(events) = &self.events { + events.publish(event); + } + } + + fn publish_failed(&self, project: &Project, agent: AgentId, reason: impl ToString) { + self.publish(DomainEvent::AgentWakeFailed { + project_id: project.id, + agent_id: agent, + reason: reason.to_string(), + }); + } + + async fn wake_one( + &self, + project: &Project, + agent: AgentId, + reason: WakeReason, + ) -> Result<(), WakeError> { + self.publish(DomainEvent::AgentWakeScheduled { + project_id: project.id, + agent_id: agent, + }); + + if self.input.busy_state(agent).is_busy() { + return Ok(()); + } + + let Some(item) = self.inbox.dequeue_next(agent) else { + return Ok(()); + }; + let delivery = self.delivery_from_item(item, &reason).await?; + let ticket = Ticket::new(delivery.ticket_id, "IdeA", delivery.prompt.clone()); + let _pending = self.input.enqueue_silent(agent, ticket); + let guard = WakeTurnGuard::new( + Arc::clone(&self.input), + Arc::clone(&self.mailbox), + agent, + delivery.ticket_id, + ); + + self.publish(DomainEvent::AgentWakeStarted { + project_id: project.id, + agent_id: agent, + }); + let session = self.sessions.session_for_wake(project, agent).await?; + drain_with_readiness( + session.as_ref(), + &delivery.prompt, + None, + self.input.as_ref(), + agent, + ) + .await + .map_err(|err| WakeError::Session(err.to_string()))?; + + self.mailbox.cancel_head(agent, delivery.ticket_id); + self.input.mark_idle(agent); + guard.disarm(); + + if let Some(task_id) = delivery.delivered_task_id { + self.tasks + .mark_completion_delivered(task_id) + .await + .map_err(|err| WakeError::Store(err.to_string()))?; + } + Ok(()) + } + + async fn delivery_from_item( + &self, + item: InboxItem, + reason: &WakeReason, + ) -> Result { + match (reason, item.kind, item.source) { + ( + WakeReason::BackgroundCompletion { task_id }, + InboxItemKind::BackgroundCompletion, + InboxSource::BackgroundTask { + task_id: item_task_id, + }, + ) if *task_id == item_task_id => { + let task = self.load_task(item_task_id).await?; + Ok(WakeDelivery { + ticket_id: item.id, + prompt: background_completion_prompt(&task), + delivered_task_id: Some(item_task_id), + }) + } + (WakeReason::InboxItem { item_id }, _, _) if *item_id == item.id => Ok(WakeDelivery { + ticket_id: item.id, + prompt: system_prompt_from_item(&item), + delivered_task_id: None, + }), + _ => Err(WakeError::UnexpectedInboxItem(format!( + "item={} kind={:?} source={:?} reason={:?}", + item.id, item.kind, item.source, reason + ))), + } + } + + async fn load_task(&self, task_id: domain::ids::TaskId) -> Result { + self.tasks + .get(task_id) + .await + .map_err(|err| WakeError::Store(err.to_string()))? + .ok_or_else(|| WakeError::Task(format!("background task {task_id} not found"))) + .and_then(|task| { + if task.result.is_some() { + Ok(task) + } else { + Err(WakeError::Task(format!( + "background task {task_id} has no terminal result" + ))) + } + }) + } +} + +#[async_trait] +impl AgentWakePort for AgentWakeService { + async fn wake_agent( + &self, + project: &Project, + agent: AgentId, + reason: WakeReason, + ) -> Result<(), WakeError> { + match self.wake_one(project, agent, reason).await { + Ok(()) => Ok(()), + Err(err) => { + self.publish_failed(project, agent, &err); + Err(err) + } + } + } +} + +struct WakeDelivery { + ticket_id: domain::mailbox::TicketId, + prompt: String, + delivered_task_id: Option, +} + +struct WakeTurnGuard { + input: Arc, + mailbox: Arc, + agent: AgentId, + ticket: domain::mailbox::TicketId, + armed: bool, +} + +impl WakeTurnGuard { + fn new( + input: Arc, + mailbox: Arc, + agent: AgentId, + ticket: domain::mailbox::TicketId, + ) -> Self { + Self { + input, + mailbox, + agent, + ticket, + armed: true, + } + } + + fn disarm(mut self) { + self.armed = false; + } +} + +impl Drop for WakeTurnGuard { + fn drop(&mut self) { + if self.armed { + self.mailbox.cancel_head(self.agent, self.ticket); + self.input.mark_idle(self.agent); + } + } +} + +fn system_prompt_from_item(item: &InboxItem) -> String { + format!("Message système pour l'agent :\n\n{}", item.body) +} + +fn background_completion_prompt(task: &BackgroundTask) -> String { + match task.result.as_ref() { + Some(BackgroundTaskResult::Success { + exit_code, + summary, + stdout_tail, + stderr_tail, + .. + }) => format!( + "La tâche de fond {} est terminée : succès{}.\nRésumé : {}\n{}{}", + task.id, + exit_code_text(*exit_code), + summary, + tail_text("stdout", stdout_tail.as_deref()), + tail_text("stderr", stderr_tail.as_deref()) + ), + Some(BackgroundTaskResult::Failure { + exit_code, + error, + stdout_tail, + stderr_tail, + .. + }) => format!( + "La tâche de fond {} est terminée : échec{}.\nErreur : {}\n{}{}", + task.id, + exit_code_text(*exit_code), + error, + tail_text("stdout", stdout_tail.as_deref()), + tail_text("stderr", stderr_tail.as_deref()) + ), + Some(BackgroundTaskResult::Cancelled { reason, .. }) => format!( + "La tâche de fond {} est terminée : annulée.\nRaison : {}", + task.id, reason + ), + Some(BackgroundTaskResult::Expired { reason, .. }) => format!( + "La tâche de fond {} est terminée : expirée.\nRaison : {}", + task.id, reason + ), + None => format!( + "La tâche de fond {} est terminée, mais son résultat est indisponible.", + task.id + ), + } +} + +fn exit_code_text(exit_code: Option) -> String { + exit_code.map_or_else(String::new, |code| format!(" (exit {code})")) +} + +fn tail_text(label: &str, value: Option<&str>) -> String { + value.map_or_else(String::new, |text| { + if text.is_empty() { + String::new() + } else { + format!("{label} :\n{text}\n") + } + }) +} diff --git a/crates/application/tests/agent_wake.rs b/crates/application/tests/agent_wake.rs new file mode 100644 index 0000000..d5cd519 --- /dev/null +++ b/crates/application/tests/agent_wake.rs @@ -0,0 +1,524 @@ +use std::collections::{HashMap, VecDeque}; +use std::sync::{Arc, Mutex}; + +use application::{AgentWakeService, WakeSessionProvider}; +use async_trait::async_trait; +use domain::background_task::{ + BackgroundTask, BackgroundTaskKind, BackgroundTaskResult, BackgroundTaskState, + BackgroundTaskWakePolicy, +}; +use domain::ids::{AgentId, ProjectId, SessionId, TaskId}; +use domain::inbox::{ + AgentInbox, AgentInboxSnapshot, InboxError, InboxItem, InboxItemKind, InboxReceipt, + InboxReceiptStatus, InboxSource, +}; +use domain::input::{AgentBusyState, InputMediator}; +use domain::mailbox::{AgentMailbox, MailboxError, PendingReply, Ticket, TicketId}; +use domain::ports::{ + AgentSession, AgentSessionError, AgentWakePort, BackgroundTaskPortError, BackgroundTaskStore, + ReplyEvent, ReplyStream, WakeError, WakeReason, +}; +use domain::project::{Project, ProjectPath}; +use domain::remote::RemoteRef; +use uuid::Uuid; + +fn id(n: u128) -> Uuid { + Uuid::from_u128(n) +} + +fn agent(n: u128) -> AgentId { + AgentId::from_uuid(id(n)) +} + +fn task_id(n: u128) -> TaskId { + TaskId::from_uuid(id(n)) +} + +fn ticket(n: u128) -> TicketId { + TicketId::from_uuid(id(n)) +} + +fn project() -> Project { + Project::new( + ProjectId::from_uuid(id(100)), + "p", + ProjectPath::new("/tmp/idea-agent-wake-test").unwrap(), + RemoteRef::local(), + 0, + ) + .unwrap() +} + +fn completion_item(agent_id: AgentId, task_id: TaskId, ticket_id: TicketId) -> InboxItem { + InboxItem { + id: ticket_id, + agent_id, + source: InboxSource::BackgroundTask { task_id }, + kind: InboxItemKind::BackgroundCompletion, + body: format!("Background task {task_id} completed."), + created_at_ms: 10, + correlation_id: Some(task_id.to_string()), + } +} + +fn completed_task( + project: &Project, + owner: AgentId, + task_id: TaskId, + summary: &str, +) -> BackgroundTask { + BackgroundTask::new( + task_id, + project.id, + owner, + BackgroundTaskKind::Command { + label: "build".to_owned(), + }, + BackgroundTaskWakePolicy::WakeOwner, + 1, + None, + ) + .unwrap() + .transition(BackgroundTaskState::Running, 10) + .unwrap() + .complete(BackgroundTaskResult::Success { + finished_at_ms: 20, + exit_code: Some(0), + summary: summary.to_owned(), + stdout_tail: Some("ok".to_owned()), + stderr_tail: None, + }) + .unwrap() +} + +#[derive(Default)] +struct FakeInbox { + queues: Mutex>>, +} + +impl AgentInbox for FakeInbox { + fn enqueue_message( + &self, + agent_id: AgentId, + item: InboxItem, + ) -> Result { + let mut queues = self.queues.lock().unwrap(); + let queue = queues.entry(agent_id).or_default(); + let item_id = item.id; + queue.push_back(item); + Ok(InboxReceipt { + item_id, + agent_id, + depth: queue.len(), + status: InboxReceiptStatus::Queued, + }) + } + + fn dequeue_next(&self, agent_id: AgentId) -> Option { + self.queues + .lock() + .unwrap() + .entry(agent_id) + .or_default() + .pop_front() + } + + fn snapshot(&self, agent_id: AgentId) -> AgentInboxSnapshot { + let queues = self.queues.lock().unwrap(); + let items = queues + .get(&agent_id) + .map(|queue| queue.iter().cloned().collect::>()) + .unwrap_or_default(); + AgentInboxSnapshot { + agent_id, + depth: items.len(), + items, + } + } +} + +#[derive(Default)] +struct SharedTurnState { + busy: Mutex>, + tickets: Mutex>>, +} + +impl SharedTurnState { + fn force_busy(&self, agent_id: AgentId, ticket_id: TicketId) { + self.busy.lock().unwrap().insert( + agent_id, + AgentBusyState::Busy { + ticket: ticket_id, + since_ms: 1, + }, + ); + } + + fn ticket_depth(&self, agent_id: AgentId) -> usize { + self.tickets + .lock() + .unwrap() + .get(&agent_id) + .map(VecDeque::len) + .unwrap_or_default() + } +} + +impl InputMediator for SharedTurnState { + fn enqueue(&self, agent_id: AgentId, ticket: Ticket) -> PendingReply { + self.enqueue_silent(agent_id, ticket) + } + + fn enqueue_silent(&self, agent_id: AgentId, ticket: Ticket) -> PendingReply { + self.busy.lock().unwrap().insert( + agent_id, + AgentBusyState::Busy { + ticket: ticket.id, + since_ms: 1, + }, + ); + self.tickets + .lock() + .unwrap() + .entry(agent_id) + .or_default() + .push_back(ticket); + PendingReply::new(Box::pin(async { Err(MailboxError::Cancelled) })) + } + + fn preempt(&self, _agent: AgentId) {} + + fn mark_idle(&self, agent_id: AgentId) { + self.busy + .lock() + .unwrap() + .insert(agent_id, AgentBusyState::Idle); + } + + fn busy_state(&self, agent_id: AgentId) -> AgentBusyState { + self.busy + .lock() + .unwrap() + .get(&agent_id) + .copied() + .unwrap_or(AgentBusyState::Idle) + } +} + +impl AgentMailbox for SharedTurnState { + fn enqueue(&self, agent_id: AgentId, ticket: Ticket) -> PendingReply { + ::enqueue(self, agent_id, ticket) + } + + fn resolve(&self, _agent: AgentId, _result: String) -> Result<(), MailboxError> { + Ok(()) + } + + fn cancel_head(&self, agent_id: AgentId, ticket_id: TicketId) { + let mut tickets = self.tickets.lock().unwrap(); + if let Some(queue) = tickets.get_mut(&agent_id) { + if queue.front().is_some_and(|ticket| ticket.id == ticket_id) { + queue.pop_front(); + } + } + } +} + +#[derive(Default)] +struct FakeTaskStore { + tasks: Mutex>, + delivered: Mutex>, +} + +impl FakeTaskStore { + fn insert(&self, task: BackgroundTask) { + self.tasks.lock().unwrap().insert(task.id, task); + } + + fn delivered(&self) -> Vec { + self.delivered.lock().unwrap().clone() + } +} + +#[async_trait] +impl BackgroundTaskStore for FakeTaskStore { + async fn create(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> { + self.insert(task.clone()); + Ok(()) + } + + async fn get(&self, id: TaskId) -> Result, BackgroundTaskPortError> { + Ok(self.tasks.lock().unwrap().get(&id).cloned()) + } + + async fn save(&self, task: &BackgroundTask) -> Result<(), BackgroundTaskPortError> { + self.insert(task.clone()); + Ok(()) + } + + async fn list_open_for_agent( + &self, + agent_id: AgentId, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.owner_agent_id == agent_id && !task.is_terminal()) + .cloned() + .collect()) + } + + async fn list_undelivered_completions( + &self, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.is_terminal() && !task.completion_delivered) + .cloned() + .collect()) + } + + async fn mark_completion_delivered( + &self, + task_id: TaskId, + ) -> Result<(), BackgroundTaskPortError> { + self.delivered.lock().unwrap().push(task_id); + if let Some(task) = self.tasks.lock().unwrap().get_mut(&task_id) { + task.completion_delivered = true; + } + Ok(()) + } +} + +#[derive(Default)] +struct FakeSession { + prompts: Mutex>, +} + +#[async_trait] +impl AgentSession for FakeSession { + fn id(&self) -> SessionId { + SessionId::from_uuid(id(500)) + } + + fn conversation_id(&self) -> Option { + Some("conversation".to_owned()) + } + + async fn send(&self, prompt: &str) -> Result { + self.prompts.lock().unwrap().push(prompt.to_owned()); + Ok(Box::new( + vec![ReplyEvent::Final { + content: "ack".to_owned(), + }] + .into_iter(), + )) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + +#[derive(Default)] +struct FakeSessionProvider { + session: Mutex>>, + launches: Mutex, +} + +impl FakeSessionProvider { + fn with_session(session: Arc) -> Self { + Self { + session: Mutex::new(Some(session)), + launches: Mutex::new(0), + } + } + + fn launches(&self) -> usize { + *self.launches.lock().unwrap() + } + + fn session(&self) -> Arc { + self.session.lock().unwrap().as_ref().unwrap().clone() + } +} + +#[async_trait] +impl WakeSessionProvider for FakeSessionProvider { + async fn session_for_wake( + &self, + _project: &Project, + _agent: AgentId, + ) -> Result, WakeError> { + let mut session = self.session.lock().unwrap(); + if session.is_none() { + *self.launches.lock().unwrap() += 1; + *session = Some(Arc::new(FakeSession::default())); + } + Ok(session.as_ref().unwrap().clone()) + } +} + +fn service( + inbox: Arc, + turns: Arc, + tasks: Arc, + sessions: Arc, +) -> AgentWakeService { + AgentWakeService::new(inbox, turns.clone(), turns, tasks, sessions, None) +} + +#[tokio::test] +async fn wake_if_idle_starts_turn_with_background_completion_prompt() { + let project = project(); + let owner = agent(1); + let task_id = task_id(10); + let inbox = Arc::new(FakeInbox::default()); + let turns = Arc::new(SharedTurnState::default()); + let tasks = Arc::new(FakeTaskStore::default()); + let session = Arc::new(FakeSession::default()); + let sessions = Arc::new(FakeSessionProvider::with_session(session.clone())); + tasks.insert(completed_task(&project, owner, task_id, "build finished")); + inbox + .enqueue_message(owner, completion_item(owner, task_id, ticket(20))) + .unwrap(); + + service(inbox, turns, tasks, sessions) + .wake_agent( + &project, + owner, + WakeReason::BackgroundCompletion { task_id }, + ) + .await + .unwrap(); + + let prompts = session.prompts.lock().unwrap(); + assert_eq!(prompts.len(), 1); + assert!(prompts[0].contains(&task_id.to_string())); + assert!(prompts[0].contains("exit 0")); + assert!(prompts[0].contains("build finished")); + assert!(prompts[0].contains("stdout")); +} + +#[tokio::test] +async fn owner_busy_does_not_start_concurrent_wake_and_keeps_item_queued() { + let project = project(); + let owner = agent(1); + let task_id = task_id(10); + let inbox = Arc::new(FakeInbox::default()); + let turns = Arc::new(SharedTurnState::default()); + let tasks = Arc::new(FakeTaskStore::default()); + let session = Arc::new(FakeSession::default()); + let sessions = Arc::new(FakeSessionProvider::with_session(session.clone())); + tasks.insert(completed_task(&project, owner, task_id, "done")); + inbox + .enqueue_message(owner, completion_item(owner, task_id, ticket(20))) + .unwrap(); + turns.force_busy(owner, ticket(99)); + + service(inbox.clone(), turns, tasks, sessions) + .wake_agent( + &project, + owner, + WakeReason::BackgroundCompletion { task_id }, + ) + .await + .unwrap(); + + assert!(session.prompts.lock().unwrap().is_empty()); + assert_eq!(inbox.snapshot(owner).depth, 1); +} + +#[tokio::test] +async fn absent_session_is_launched_or_reattached_by_provider() { + let project = project(); + let owner = agent(1); + let task_id = task_id(10); + let inbox = Arc::new(FakeInbox::default()); + let turns = Arc::new(SharedTurnState::default()); + let tasks = Arc::new(FakeTaskStore::default()); + let sessions = Arc::new(FakeSessionProvider::default()); + tasks.insert(completed_task(&project, owner, task_id, "done")); + inbox + .enqueue_message(owner, completion_item(owner, task_id, ticket(20))) + .unwrap(); + + service(inbox, turns, tasks, sessions.clone()) + .wake_agent( + &project, + owner, + WakeReason::BackgroundCompletion { task_id }, + ) + .await + .unwrap(); + + assert_eq!(sessions.launches(), 1); + assert_eq!(sessions.session().prompts.lock().unwrap().len(), 1); +} + +#[tokio::test] +async fn completion_is_marked_delivered_after_successful_wake() { + let project = project(); + let owner = agent(1); + let task_id = task_id(10); + let inbox = Arc::new(FakeInbox::default()); + let turns = Arc::new(SharedTurnState::default()); + let tasks = Arc::new(FakeTaskStore::default()); + let sessions = Arc::new(FakeSessionProvider::with_session(Arc::new( + FakeSession::default(), + ))); + tasks.insert(completed_task(&project, owner, task_id, "done")); + inbox + .enqueue_message(owner, completion_item(owner, task_id, ticket(20))) + .unwrap(); + + service(inbox, turns, tasks.clone(), sessions) + .wake_agent( + &project, + owner, + WakeReason::BackgroundCompletion { task_id }, + ) + .await + .unwrap(); + + assert_eq!(tasks.delivered(), vec![task_id]); +} + +#[tokio::test] +async fn wake_drains_exactly_one_item_per_turn() { + let project = project(); + let owner = agent(1); + let first = task_id(10); + let second = task_id(11); + let inbox = Arc::new(FakeInbox::default()); + let turns = Arc::new(SharedTurnState::default()); + let tasks = Arc::new(FakeTaskStore::default()); + let session = Arc::new(FakeSession::default()); + let sessions = Arc::new(FakeSessionProvider::with_session(session.clone())); + tasks.insert(completed_task(&project, owner, first, "first")); + tasks.insert(completed_task(&project, owner, second, "second")); + inbox + .enqueue_message(owner, completion_item(owner, first, ticket(20))) + .unwrap(); + inbox + .enqueue_message(owner, completion_item(owner, second, ticket(21))) + .unwrap(); + + service(inbox.clone(), turns.clone(), tasks, sessions) + .wake_agent( + &project, + owner, + WakeReason::BackgroundCompletion { task_id: first }, + ) + .await + .unwrap(); + + assert_eq!(session.prompts.lock().unwrap().len(), 1); + assert_eq!(inbox.snapshot(owner).depth, 1); + assert_eq!(turns.ticket_depth(owner), 0); +}