diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 67203f6e..de8100fa 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -4438,6 +4438,7 @@ pub async fn spawn_background_command( command, wake_policy, rendezvous: None, + parent_task_id: None, deadline_ms: request.deadline_ms, }) .await diff --git a/crates/application/src/background/mod.rs b/crates/application/src/background/mod.rs index 3e786bea..f3aefd5e 100644 --- a/crates/application/src/background/mod.rs +++ b/crates/application/src/background/mod.rs @@ -13,6 +13,7 @@ //! except on a *spawn failure*, where they mark the just-created task `Failed` so //! it never dangles in `Running` with no worker behind it. +use std::collections::{HashSet, VecDeque}; use std::sync::Arc; use domain::ports::{ @@ -55,6 +56,8 @@ pub struct SpawnBackgroundCommandInput { pub wake_policy: BackgroundTaskWakePolicy, /// Optional composite rendezvous this task participates in. pub rendezvous: Option, + /// Optional parent background task. Used for deterministic cancel cascades. + pub parent_task_id: Option, /// Optional absolute deadline, epoch milliseconds. pub deadline_ms: Option, } @@ -116,6 +119,7 @@ impl SpawnBackgroundCommand { input.command, input.wake_policy, input.rendezvous, + input.parent_task_id, input.deadline_ms, ) .await?; @@ -131,6 +135,7 @@ impl SpawnBackgroundCommand { command: SpawnSpec, wake_policy: BackgroundTaskWakePolicy, rendezvous: Option, + parent_task_id: Option, deadline_ms: Option, ) -> Result { let task_id = TaskId::from_uuid(self.ids.new_uuid()); @@ -149,6 +154,11 @@ impl SpawnBackgroundCommand { if let Some(link) = rendezvous { task = task.with_rendezvous(link); } + if let Some(parent_task_id) = parent_task_id { + task = task + .with_parent_task(parent_task_id) + .map_err(|e| AppError::Invalid(e.to_string()))?; + } self.store.create(&task).await.map_err(map_port_err)?; let running = task @@ -174,6 +184,7 @@ impl SpawnBackgroundCommand { project_id, owner_agent_id, kind: running.kind.clone(), + parent_task_id: running.parent_task_id, wake_policy, command: Some(command), deadline_ms, @@ -244,23 +255,101 @@ impl CancelBackgroundTask { self } - /// Projects `Cancelling` immediately, then signals the runner to cancel `task_id`. + /// Cancels `task_id` and its open descendants. /// /// # Errors - /// [`AppError`] if the task is missing, terminal, or the runner/store fails. + /// [`AppError`] if the task is missing, terminal without open descendants, or a port fails. pub async fn execute(&self, task_id: TaskId) -> Result { - let task = self + let root = self .store .get(task_id) .await .map_err(map_port_err)? .ok_or_else(|| AppError::NotFound(format!("background task {task_id}")))?; - if task.is_terminal() { + + let tasks = self.cancel_subtree_from_task(root.clone()).await?; + let task = tasks + .iter() + .find(|task| task.id == task_id) + .cloned() + .unwrap_or(root); + Ok(CancelBackgroundTaskOutput { task }) + } + + /// Cancels all open tasks rooted at one agent. + /// + /// Terminal tasks are never modified. Errors are propagated so command + /// handlers do not report a silent success when the runner cannot stop a child. + pub async fn cancel_subtree_for_agent( + &self, + project_id: ProjectId, + agent_id: AgentId, + ) -> Result, AppError> { + let mut roots = self + .store + .list_open_for_agent(agent_id) + .await + .map_err(map_port_err)?; + roots.retain(|task| task.project_id == project_id); + roots.sort_by_key(|task| (task.created_at_ms, task.id)); + self.cancel_subtree(roots).await + } + + async fn cancel_subtree_from_task( + &self, + root: BackgroundTask, + ) -> Result, AppError> { + let mut roots = if root.is_terminal() { + Vec::new() + } else { + vec![root.clone()] + }; + roots.extend( + self.store + .list_children(root.id) + .await + .map_err(map_port_err)?, + ); + if roots.is_empty() { return Err(AppError::Invalid(format!( - "background task {task_id} is already terminal ({:?})", - task.state + "background task {} is already terminal ({:?})", + root.id, root.state ))); } + self.cancel_subtree(roots).await + } + + async fn cancel_subtree( + &self, + roots: Vec, + ) -> Result, AppError> { + let mut queue: VecDeque = roots.into(); + let mut visited = HashSet::new(); + let mut cancelled = Vec::new(); + + while let Some(task) = queue.pop_front() { + if !visited.insert(task.id) { + continue; + } + let mut children = self + .store + .list_children(task.id) + .await + .map_err(map_port_err)?; + children.sort_by_key(|task| (task.created_at_ms, task.id)); + queue.extend(children); + if task.is_terminal() { + continue; + } + cancelled.push(self.cancel_one(task).await?); + } + + cancelled.sort_by_key(|task| (task.created_at_ms, task.id)); + Ok(cancelled) + } + + async fn cancel_one(&self, task: BackgroundTask) -> Result { + let task_id = task.id; let cancelling = if task.state == BackgroundTaskState::Cancelling { task @@ -287,13 +376,12 @@ impl CancelBackgroundTask { return Err(map_port_err(err)); } - let task = self + Ok(self .store .get(task_id) .await .map_err(map_port_err)? - .unwrap_or(cancelling); - Ok(CancelBackgroundTaskOutput { task }) + .unwrap_or(cancelling)) } fn publish_state_changed(&self, task: &BackgroundTask) { @@ -396,6 +484,7 @@ impl RetryBackgroundTask { old.wake_policy, None, None, + None, ) .await?; Ok(SpawnBackgroundCommandOutput { task }) diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 7b7b2c59..06357a52 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -574,6 +574,7 @@ pub struct OrchestratorOutcome { #[derive(Debug, Clone)] struct ActiveRendezvous { link: BackgroundTaskRendezvousLink, + task_id: Option, background_tasks: Vec, } @@ -737,6 +738,14 @@ impl OrchestratorService { .map(|r| r.link.clone()) } + fn active_rendezvous_task_id(&self, agent: RuntimeAgentKey) -> Option { + self.active_rendezvous + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(&agent) + .and_then(|r| r.task_id) + } + fn attach_background_to_active_rendezvous( &self, agent: RuntimeAgentKey, @@ -1306,6 +1315,7 @@ impl OrchestratorService { let cwd = resolve_background_cwd(&project.root, cwd)?; let agent_key = runtime_key(project, owner); let rendezvous = self.active_rendezvous_link(agent_key); + let parent_task_id = self.active_rendezvous_task_id(agent_key); let output = spawn .execute(SpawnBackgroundCommandInput { project_id: project.id, @@ -1321,6 +1331,7 @@ impl OrchestratorService { }, wake_policy: BackgroundTaskWakePolicy::WakeOwner, rendezvous: rendezvous.clone(), + parent_task_id, deadline_ms, }) .await?; @@ -1949,7 +1960,7 @@ impl OrchestratorService { .start_rendezvous_task(project, agent_id, rendezvous_link.clone()) .await?; let _active_rendezvous = - ActiveRendezvousGuard::new(self, agent_key, rendezvous_link.clone()); + ActiveRendezvousGuard::new(self, agent_key, rendezvous_link.clone(), rendezvous_task); // 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. @@ -2703,6 +2714,8 @@ impl OrchestratorService { .await? .ok_or_else(|| AppError::NotFound(format!("agent {name}")))?; + self.stop_agent_execution_children(project, agent_id).await; + let structured = self .structured .clone() @@ -3042,34 +3055,21 @@ impl<'a> WaitEdgeGuard<'a> { #[async_trait::async_trait] impl StopAgentExecutionCascade for OrchestratorService { async fn stop_agent_execution_children(&self, project: &Project, agent_id: AgentId) { - let Some(runner) = &self.background_runner else { + let (Some(store), Some(runner)) = (&self.background_tasks, &self.background_runner) else { return; }; - let agent = runtime_key(project, agent_id); - stop_active_rendezvous_background_tasks(&self.active_rendezvous, agent, runner.as_ref()) - .await; - } -} - -fn active_rendezvous_background_tasks( - active: &StdMutex>, - agent: RuntimeAgentKey, -) -> Vec { - active - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .get(&agent) - .map(|active| active.background_tasks.clone()) - .unwrap_or_default() -} - -async fn stop_active_rendezvous_background_tasks( - active: &StdMutex>, - agent: RuntimeAgentKey, - runner: &dyn BackgroundTaskRunner, -) { - for task_id in active_rendezvous_background_tasks(active, agent) { - let _ = runner.cancel(task_id).await; + let Some(clock) = &self.clock else { + return; + }; + let mut cancel = crate::background::CancelBackgroundTask::new( + Arc::clone(store), + Arc::clone(runner), + Arc::clone(clock), + ); + if let Some(events) = &self.events { + cancel = cancel.with_events(Arc::clone(events)); + } + let _ = cancel.cancel_subtree_for_agent(project.id, agent_id).await; } } @@ -3099,6 +3099,7 @@ impl<'a> ActiveRendezvousGuard<'a> { service: &'a OrchestratorService, agent: RuntimeAgentKey, link: BackgroundTaskRendezvousLink, + task_id: Option, ) -> Self { let rendezvous_id = link.rendezvous_id; service @@ -3109,6 +3110,7 @@ impl<'a> ActiveRendezvousGuard<'a> { agent, ActiveRendezvous { link, + task_id, background_tasks: Vec::new(), }, ); @@ -3347,77 +3349,6 @@ mod tests { RuntimeAgentKey::new(domain::ProjectId::from_uuid(uuid::Uuid::nil()), aid(n)) } - #[derive(Default)] - struct SpyBackgroundRunner { - cancelled: TestMutex>, - } - - #[async_trait::async_trait] - impl BackgroundTaskRunner for SpyBackgroundRunner { - async fn spawn( - &self, - spec: domain::ports::BackgroundTaskSpec, - ) -> Result - { - Ok(domain::ports::BackgroundTaskHandle { - task_id: spec.task_id, - }) - } - - async fn cancel( - &self, - task_id: TaskId, - ) -> Result<(), domain::ports::BackgroundTaskPortError> { - self.cancelled.lock().unwrap().push(task_id); - Ok(()) - } - - fn subscribe_completions(&self) -> domain::ports::BackgroundCompletionStream { - Box::new(std::iter::empty()) - } - } - - #[tokio::test] - async fn stop_active_rendezvous_background_tasks_cancels_attached_tasks_only() { - let active = StdMutex::new(HashMap::new()); - let target = rkey(2); - let unrelated = rkey(9); - let first = TaskId::from_uuid(uuid::Uuid::from_u128(101)); - let second = TaskId::from_uuid(uuid::Uuid::from_u128(102)); - let other = TaskId::from_uuid(uuid::Uuid::from_u128(999)); - active.lock().unwrap().insert( - target, - ActiveRendezvous { - link: BackgroundTaskRendezvousLink { - rendezvous_id: RendezvousId::from_uuid(uuid::Uuid::from_u128(201)), - ticket_id: tid(202), - requester_agent_id: Some(aid(1)), - target_agent_id: target.agent_id, - conversation_id: domain::ConversationId::from_uuid(uuid::Uuid::from_u128(203)), - }, - background_tasks: vec![first, second], - }, - ); - active.lock().unwrap().insert( - unrelated, - ActiveRendezvous { - link: BackgroundTaskRendezvousLink { - rendezvous_id: RendezvousId::from_uuid(uuid::Uuid::from_u128(301)), - ticket_id: tid(302), - requester_agent_id: Some(aid(1)), - target_agent_id: unrelated.agent_id, - conversation_id: domain::ConversationId::from_uuid(uuid::Uuid::from_u128(303)), - }, - background_tasks: vec![other], - }, - ); - let runner = SpyBackgroundRunner::default(); - - stop_active_rendezvous_background_tasks(&active, target, &runner).await; - - assert_eq!(*runner.cancelled.lock().unwrap(), vec![first, second]); - } - /// Drop d'un garde **armé** ⇒ `cancel_head` + `mark_idle` sur la cible (c'est le /// comportement qui débloque un agent resté `Busy` sur un futur abandonné). #[test] diff --git a/crates/application/src/plugin/mod.rs b/crates/application/src/plugin/mod.rs index afc932c7..1c366cf4 100644 --- a/crates/application/src/plugin/mod.rs +++ b/crates/application/src/plugin/mod.rs @@ -2011,6 +2011,7 @@ impl PluginCommandTasks { command, wake_policy, rendezvous: None, + parent_task_id: None, deadline_ms: input.deadline_ms, }) .await @@ -4384,6 +4385,20 @@ mod tests { .collect()) } + async fn list_children( + &self, + parent_task_id: TaskId, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.parent_task_id == Some(parent_task_id)) + .cloned() + .collect()) + } + async fn list_undelivered_completions( &self, ) -> Result, BackgroundTaskPortError> { diff --git a/crates/application/src/workstate/mod.rs b/crates/application/src/workstate/mod.rs index cd7fada2..13287123 100644 --- a/crates/application/src/workstate/mod.rs +++ b/crates/application/src/workstate/mod.rs @@ -221,6 +221,8 @@ pub struct AgentWorkState { pub struct AgentBackgroundTaskState { /// Stable task id. pub task_id: TaskId, + /// Optional parent background task id. + pub parent_task_id: Option, /// Kind discriminant. pub kind: BackgroundTaskKindLabel, /// Current lifecycle state. @@ -609,6 +611,7 @@ impl From for AgentBackgroundTaskState { flatten_background_rendezvous_context(&task.kind); Self { task_id: task.id, + parent_task_id: task.parent_task_id, kind: BackgroundTaskKindLabel::from(&task.kind), state: task.state, exit_code, diff --git a/crates/application/tests/agent_wake.rs b/crates/application/tests/agent_wake.rs index 2105f4c4..d42c235d 100644 --- a/crates/application/tests/agent_wake.rs +++ b/crates/application/tests/agent_wake.rs @@ -391,6 +391,20 @@ impl BackgroundTaskStore for FakeTaskStore { .collect()) } + async fn list_children( + &self, + parent_task_id: TaskId, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.parent_task_id == Some(parent_task_id)) + .cloned() + .collect()) + } + async fn list_undelivered_completions( &self, ) -> Result, BackgroundTaskPortError> { diff --git a/crates/application/tests/background_task_cancel.rs b/crates/application/tests/background_task_cancel.rs index eb1e1cf6..9a2a5ad1 100644 --- a/crates/application/tests/background_task_cancel.rs +++ b/crates/application/tests/background_task_cancel.rs @@ -43,6 +43,10 @@ fn running_task(id: TaskId) -> BackgroundTask { .unwrap() } +fn child_task(id: TaskId, parent: TaskId) -> BackgroundTask { + running_task(id).with_parent_task(parent).unwrap() +} + #[derive(Default)] struct FakeStore { tasks: Mutex>, @@ -90,6 +94,20 @@ impl BackgroundTaskStore for FakeStore { .collect()) } + async fn list_children( + &self, + parent_task_id: TaskId, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.parent_task_id == Some(parent_task_id)) + .cloned() + .collect()) + } + async fn list_undelivered_completions( &self, ) -> Result, BackgroundTaskPortError> { @@ -304,3 +322,84 @@ async fn cancel_kill_failure_persists_cancel_failed_and_returns_error() { ] ); } + +#[tokio::test] +async fn cancel_parent_cascades_to_open_descendants_and_leaves_terminal_child() { + let store = Arc::new(FakeStore::default()); + let runner = Arc::new(FakeRunner::ok()); + let events = Arc::new(SpyEvents::default()); + let parent = running_task(tid(10)); + let child = child_task(tid(11), tid(10)); + let grandchild = child_task(tid(12), tid(11)); + let terminal_child = child_task(tid(13), tid(10)) + .complete(BackgroundTaskResult::Success { + finished_at_ms: 130, + exit_code: Some(0), + summary: "done".to_owned(), + stdout_tail: None, + stderr_tail: None, + }) + .unwrap(); + store.insert(parent); + store.insert(child); + store.insert(grandchild); + store.insert(terminal_child.clone()); + + usecase(Arc::clone(&store), Arc::clone(&runner), events) + .execute(tid(10)) + .await + .unwrap(); + + assert_eq!(runner.cancels(), vec![tid(10), tid(11), tid(12)]); + assert_eq!( + store.get(tid(13)).await.unwrap().unwrap().state, + BackgroundTaskState::Completed + ); +} + +#[tokio::test] +async fn cancel_terminal_parent_still_cascades_to_running_child() { + let store = Arc::new(FakeStore::default()); + let runner = Arc::new(FakeRunner::ok()); + let events = Arc::new(SpyEvents::default()); + let parent = running_task(tid(10)) + .complete(BackgroundTaskResult::Success { + finished_at_ms: 130, + exit_code: Some(0), + summary: "done".to_owned(), + stdout_tail: None, + stderr_tail: None, + }) + .unwrap(); + store.insert(parent); + store.insert(child_task(tid(11), tid(10))); + + let out = usecase(Arc::clone(&store), Arc::clone(&runner), events) + .execute(tid(10)) + .await + .unwrap(); + + assert_eq!(out.task.state, BackgroundTaskState::Completed); + assert_eq!(runner.cancels(), vec![tid(11)]); + assert_eq!( + store.get(tid(11)).await.unwrap().unwrap().state, + BackgroundTaskState::Cancelling + ); +} + +#[tokio::test] +async fn cancel_agent_subtree_visits_each_task_once() { + let store = Arc::new(FakeStore::default()); + let runner = Arc::new(FakeRunner::ok()); + let events = Arc::new(SpyEvents::default()); + store.insert(running_task(tid(10))); + store.insert(child_task(tid(11), tid(10))); + store.insert(child_task(tid(12), tid(11))); + + usecase(Arc::clone(&store), Arc::clone(&runner), events) + .cancel_subtree_for_agent(pid(1), aid(2)) + .await + .unwrap(); + + assert_eq!(runner.cancels(), vec![tid(10), tid(11), tid(12)]); +} diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index 573a9630..4f07ea65 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -36,8 +36,8 @@ use domain::remote::RemoteRef; use domain::skill::{Skill, SkillScope}; use domain::terminal::{SessionKind, TerminalSession}; use domain::{ - BackgroundTask, BackgroundTaskState, BackgroundTaskWakePolicy, OrchestratorCommand, - OrchestratorRequest, PtySize, SessionId, TaskId, + BackgroundTask, BackgroundTaskKind, BackgroundTaskState, BackgroundTaskWakePolicy, + OrchestratorCommand, OrchestratorRequest, PtySize, SessionId, TaskId, }; use domain::{Memory, MemoryIndexEntry, MemoryLink, MemorySlug}; use uuid::Uuid; @@ -429,6 +429,20 @@ impl BackgroundTaskStore for RecordingBackgroundStore { .collect()) } + async fn list_children( + &self, + parent_task_id: TaskId, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.parent_task_id == Some(parent_task_id)) + .cloned() + .collect()) + } + async fn list_undelivered_completions( &self, ) -> Result, BackgroundTaskPortError> { @@ -458,11 +472,16 @@ impl BackgroundTaskStore for RecordingBackgroundStore { #[derive(Clone, Default)] struct RecordingBackgroundRunner { specs: Arc>>, + cancels: Arc>>, } impl RecordingBackgroundRunner { fn specs(&self) -> Vec { self.specs.lock().unwrap().clone() } + + fn cancels(&self) -> Vec { + self.cancels.lock().unwrap().clone() + } } #[async_trait] impl BackgroundTaskRunner for RecordingBackgroundRunner { @@ -475,7 +494,8 @@ impl BackgroundTaskRunner for RecordingBackgroundRunner { Ok(BackgroundTaskHandle { task_id }) } - async fn cancel(&self, _task_id: TaskId) -> Result<(), BackgroundTaskPortError> { + async fn cancel(&self, task_id: TaskId) -> Result<(), BackgroundTaskPortError> { + self.cancels.lock().unwrap().push(task_id); Ok(()) } @@ -1048,6 +1068,73 @@ async fn stop_agent_stops_structured_session() { .is_none()); } +#[tokio::test] +async fn stop_agent_cascades_background_task_subtree() { + let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md"); + let fx = structured_ask_fixture_with_observable_background(FakeContexts::with_agent( + &agent, + "# persona", + )); + fx.structured.insert_in_project( + project_id(), + Arc::new(EmptyStructuredSession { id: sid(9801) }), + aid(1), + nid(700), + ); + let parent_id = TaskId::from_uuid(Uuid::from_u128(9101)); + let child_id = TaskId::from_uuid(Uuid::from_u128(9102)); + let parent = BackgroundTask::new( + parent_id, + project_id(), + aid(1), + BackgroundTaskKind::Command { + label: "parent".to_owned(), + }, + BackgroundTaskWakePolicy::WakeOwner, + 10, + None, + ) + .unwrap() + .transition(BackgroundTaskState::Running, 11) + .unwrap(); + let child = BackgroundTask::new( + child_id, + project_id(), + aid(1), + BackgroundTaskKind::Command { + label: "child".to_owned(), + }, + BackgroundTaskWakePolicy::WakeOwner, + 12, + None, + ) + .unwrap() + .with_parent_task(parent_id) + .unwrap() + .transition(BackgroundTaskState::Running, 13) + .unwrap(); + fx.background_store.create(&parent).await.unwrap(); + fx.background_store.create(&child).await.unwrap(); + + fx.service + .dispatch( + &project(), + cmd(r#"{ "action":"stop_agent", "name":"dev-backend" }"#), + ) + .await + .expect("stop ok"); + + assert_eq!(fx.background_runner.cancels(), vec![parent_id, child_id]); + assert_eq!( + fx.background_store.task(parent_id).unwrap().state, + BackgroundTaskState::Cancelling + ); + assert_eq!( + fx.background_store.task(child_id).unwrap().state, + BackgroundTaskState::Cancelling + ); +} + #[tokio::test] async fn stop_agent_without_live_session_is_not_found() { let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md"); @@ -4715,6 +4802,11 @@ async fn ask_structured_final_wins_even_with_attached_background_task() { child_task.rendezvous.is_some(), "the child task keeps its rendezvous correlation" ); + assert_eq!( + child_task.parent_task_id, + Some(rendezvous_task.id), + "the child task is owned by the rendezvous task for subtree cancellation" + ); } /// Régression #201 : si le flux structuré se clôt sans `Final`, l'erreur publique @@ -4760,8 +4852,7 @@ async fn ask_structured_session_error_surfaces_process_message_not_no_reply() { project_id(), Arc::new(FailingStructuredSession { id: sid(9802), - message: "codex exec exited with code 1: 401 Unauthorized: Missing bearer" - .to_owned(), + message: "codex exec exited with code 1: 401 Unauthorized: Missing bearer".to_owned(), }), aid(1), nid(700), diff --git a/crates/application/tests/workstate.rs b/crates/application/tests/workstate.rs index 4032c7b0..81eacab9 100644 --- a/crates/application/tests/workstate.rs +++ b/crates/application/tests/workstate.rs @@ -220,6 +220,20 @@ impl BackgroundTaskStore for FakeBackgroundTaskStore { .collect()) } + async fn list_children( + &self, + parent_task_id: TaskId, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .iter() + .filter(|task| task.parent_task_id == Some(parent_task_id)) + .cloned() + .collect()) + } + async fn list_undelivered_completions( &self, ) -> Result, BackgroundTaskPortError> { diff --git a/crates/backend/src/dto.rs b/crates/backend/src/dto.rs index 3c55ba8d..0dedeff1 100644 --- a/crates/backend/src/dto.rs +++ b/crates/backend/src/dto.rs @@ -3638,6 +3638,9 @@ impl From for AgentTicketStateDto { pub struct AgentBackgroundTaskStateDto { /// Stable task id (UUID string). pub task_id: String, + /// Optional parent background task id (UUID string). + #[serde(skip_serializing_if = "Option::is_none")] + pub parent_task_id: Option, /// Kind discriminant. pub kind: String, /// Lifecycle state. @@ -3673,6 +3676,7 @@ impl From for AgentBackgroundTaskStateDto { fn from(task: AgentBackgroundTaskState) -> Self { Self { task_id: task.task_id.to_string(), + parent_task_id: task.parent_task_id.map(|id| id.to_string()), kind: background_kind_label_from_work_state(task.kind).to_owned(), state: background_state_label(task.state).to_owned(), exit_code: task.exit_code, @@ -5102,6 +5106,9 @@ pub struct BackgroundTaskDto { pub owner_agent_id: String, /// Owning project id (UUID string). pub project_id: String, + /// Optional parent background task id (UUID string). + #[serde(skip_serializing_if = "Option::is_none")] + pub parent_task_id: Option, /// Kind discriminant (`"command"`, `"headlessRendezvous"`, `"sessionResume"`, /// `"maintenance"`). pub kind: String, @@ -5212,6 +5219,7 @@ impl From for BackgroundTaskDto { task_id: task.id.to_string(), owner_agent_id: task.owner_agent_id.to_string(), project_id: task.project_id.to_string(), + parent_task_id: task.parent_task_id.map(|id| id.to_string()), kind: background_kind_label(&task.kind).to_owned(), state: background_state_label(task.state).to_owned(), exit_code, @@ -5814,6 +5822,7 @@ mod tests { let conversation_id = ConversationId::from_uuid(Uuid::from_u128(13)); let state = AgentBackgroundTaskState { task_id: TaskId::from_uuid(Uuid::from_u128(14)), + parent_task_id: Some(TaskId::from_uuid(Uuid::from_u128(10))), kind: BackgroundTaskKindLabel::HeadlessRendezvous, state: BackgroundTaskState::Completed, exit_code: None, @@ -5832,5 +5841,9 @@ mod tests { assert_eq!(json["requesterAgentId"], requester.to_string()); assert_eq!(json["targetAgentId"], target.to_string()); assert_eq!(json["conversationId"], conversation_id.to_string()); + assert_eq!( + json["parentTaskId"], + TaskId::from_uuid(Uuid::from_u128(10)).to_string() + ); } } diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index 809c9b53..57b46421 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -576,6 +576,34 @@ impl BackgroundTaskStore for AppBackgroundTaskStore { Ok(tasks) } + async fn list_children( + &self, + parent_task_id: TaskId, + ) -> Result, BackgroundTaskPortError> { + let project_ids = if let Some(project_id) = self.project_for_task(parent_task_id).await? { + vec![project_id] + } else { + self.projects + .list_projects() + .await + .map_err(|err| BackgroundTaskPortError::Store(err.to_string()))? + .into_iter() + .map(|project| project.id) + .collect() + }; + + let mut tasks = Vec::new(); + for project_id in project_ids { + let store = self.store_for_project(project_id).await?; + tasks.extend(store.list_children(parent_task_id).await?); + } + for task in &tasks { + self.remember(task.id, task.project_id); + } + tasks.sort_by_key(|task| (task.created_at_ms, task.id)); + Ok(tasks) + } + async fn list_undelivered_completions( &self, ) -> Result, BackgroundTaskPortError> { diff --git a/crates/domain/src/background_task.rs b/crates/domain/src/background_task.rs index 151f7a5d..49a1a485 100644 --- a/crates/domain/src/background_task.rs +++ b/crates/domain/src/background_task.rs @@ -130,6 +130,9 @@ pub struct BackgroundTask { /// Optional composite rendezvous this task participates in. #[serde(default)] pub rendezvous: Option, + /// Optional parent background task. Descendants are cancelled with their parent. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_task_id: Option, /// Current lifecycle state. pub state: BackgroundTaskState, /// Completion wake policy. @@ -167,6 +170,7 @@ impl BackgroundTask { owner_agent_id, kind, rendezvous: None, + parent_task_id: None, state: BackgroundTaskState::Queued, wake_policy, created_at_ms: now_ms, @@ -186,6 +190,16 @@ impl BackgroundTask { self } + /// Returns a copy owned by a parent background task. + /// + /// # Errors + /// [`BackgroundTaskError`] when the parent would make the task invalid. + pub fn with_parent_task(mut self, parent_task_id: TaskId) -> Result { + self.parent_task_id = Some(parent_task_id); + self.validate()?; + Ok(self) + } + /// Returns a copy moved to a non-terminal state. /// /// Terminal transitions must use [`Self::complete`] so the result/state @@ -293,6 +307,9 @@ impl BackgroundTask { if self.completion_delivered && !self.state.is_terminal() { return Err(BackgroundTaskError::MissingResult); } + if self.parent_task_id == Some(self.id) { + return Err(BackgroundTaskError::SelfParent); + } if let Some(result) = &self.result { result.validate()?; if result.state() != self.state { @@ -472,6 +489,9 @@ pub enum BackgroundTaskError { /// Completion delivery was already marked. #[error("background task completion was already delivered")] CompletionAlreadyDelivered, + /// A task cannot own itself. + #[error("background task cannot be its own parent")] + SelfParent, } fn is_transition_allowed(from: BackgroundTaskState, to: BackgroundTaskState) -> bool { diff --git a/crates/domain/src/ports.rs b/crates/domain/src/ports.rs index 9d366f4a..e0535c41 100644 --- a/crates/domain/src/ports.rs +++ b/crates/domain/src/ports.rs @@ -558,6 +558,8 @@ pub struct BackgroundTaskSpec { pub owner_agent_id: AgentId, /// Kind-specific task payload. pub kind: BackgroundTaskKind, + /// Optional parent background task. + pub parent_task_id: Option, /// Completion wake policy. pub wake_policy: BackgroundTaskWakePolicy, /// Optional command invocation for command-backed tasks. @@ -2597,6 +2599,15 @@ pub trait BackgroundTaskStore: Send + Sync { agent_id: AgentId, ) -> Result, BackgroundTaskPortError>; + /// Lists tasks owned by a parent background task. + /// + /// # Errors + /// [`BackgroundTaskPortError`] on store failure. + async fn list_children( + &self, + parent_task_id: TaskId, + ) -> Result, BackgroundTaskPortError>; + /// Lists terminal tasks whose completion has not been delivered yet. /// /// # Errors diff --git a/crates/domain/tests/background_task.rs b/crates/domain/tests/background_task.rs index 94459473..796bf447 100644 --- a/crates/domain/tests/background_task.rs +++ b/crates/domain/tests/background_task.rs @@ -61,6 +61,20 @@ fn rendezvous_link_roundtrips_on_background_task() { assert_eq!(decoded.rendezvous, Some(link)); } +#[test] +fn parent_task_link_roundtrips_and_rejects_self_parent() { + let task = command_task().with_parent_task(task_id(9)).unwrap(); + + let json = serde_json::to_string(&task).unwrap(); + let decoded: BackgroundTask = serde_json::from_str(&json).unwrap(); + + assert_eq!(decoded.parent_task_id, Some(task_id(9))); + assert_eq!( + command_task().with_parent_task(task_id(1)).unwrap_err(), + BackgroundTaskError::SelfParent + ); +} + fn failure_result(at: u64) -> BackgroundTaskResult { BackgroundTaskResult::Failure { finished_at_ms: at, diff --git a/crates/infrastructure/src/store/background_task.rs b/crates/infrastructure/src/store/background_task.rs index 6787ef85..da7fd659 100644 --- a/crates/infrastructure/src/store/background_task.rs +++ b/crates/infrastructure/src/store/background_task.rs @@ -48,6 +48,7 @@ struct RegistryIndex { loaded: bool, task_to_project: HashMap, open_by_agent: HashMap>, + children_by_parent: HashMap>, } #[derive(Debug, Clone, Default, PartialEq, Eq)] @@ -299,6 +300,16 @@ impl FsBackgroundTaskStore { for tasks in registry.open_by_agent.values_mut() { tasks.remove(&task.id); } + for tasks in registry.children_by_parent.values_mut() { + tasks.remove(&task.id); + } + if let Some(parent_task_id) = task.parent_task_id { + registry + .children_by_parent + .entry(parent_task_id) + .or_default() + .insert(task.id, task.clone()); + } if !task.is_terminal() { registry .open_by_agent @@ -315,10 +326,18 @@ impl RegistryIndex { loaded: true, task_to_project: HashMap::new(), open_by_agent: HashMap::new(), + children_by_parent: HashMap::new(), }; for doc in state.docs.values() { for task in &doc.tasks { registry.task_to_project.insert(task.id, task.project_id); + if let Some(parent_task_id) = task.parent_task_id { + registry + .children_by_parent + .entry(parent_task_id) + .or_default() + .insert(task.id, task.clone()); + } if !task.is_terminal() { registry .open_by_agent @@ -396,6 +415,21 @@ impl BackgroundTaskStore for FsBackgroundTaskStore { Ok(tasks) } + async fn list_children( + &self, + parent_task_id: TaskId, + ) -> Result, BackgroundTaskPortError> { + self.ensure_registry_loaded().await?; + let registry = self.registry.read().await; + let mut tasks: Vec = registry + .children_by_parent + .get(&parent_task_id) + .map(|tasks| tasks.values().cloned().collect()) + .unwrap_or_default(); + tasks.sort_by_key(|task| (task.created_at_ms, task.id)); + Ok(tasks) + } + async fn list_undelivered_completions( &self, ) -> Result, BackgroundTaskPortError> { diff --git a/crates/infrastructure/tests/background_completion_sink.rs b/crates/infrastructure/tests/background_completion_sink.rs index 6c8e3fcb..eada2b80 100644 --- a/crates/infrastructure/tests/background_completion_sink.rs +++ b/crates/infrastructure/tests/background_completion_sink.rs @@ -336,6 +336,20 @@ impl BackgroundTaskStore for RecordingStore { .collect()) } + async fn list_children( + &self, + parent_task_id: TaskId, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.parent_task_id == Some(parent_task_id)) + .cloned() + .collect()) + } + async fn list_undelivered_completions( &self, ) -> Result, BackgroundTaskPortError> { diff --git a/crates/infrastructure/tests/background_task_runner.rs b/crates/infrastructure/tests/background_task_runner.rs index 627e47d3..ecf92700 100644 --- a/crates/infrastructure/tests/background_task_runner.rs +++ b/crates/infrastructure/tests/background_task_runner.rs @@ -130,6 +130,7 @@ fn spawn_spec(task_id: TaskId, deadline_ms: Option) -> domain::ports::Backg kind: BackgroundTaskKind::Command { label: "test command".to_owned(), }, + parent_task_id: None, wake_policy: BackgroundTaskWakePolicy::RecordOnly, command: Some(SpawnSpec { command: "/bin/sh".to_owned(), diff --git a/crates/infrastructure/tests/background_task_store.rs b/crates/infrastructure/tests/background_task_store.rs index a09867a2..aa207e11 100644 --- a/crates/infrastructure/tests/background_task_store.rs +++ b/crates/infrastructure/tests/background_task_store.rs @@ -165,6 +165,44 @@ async fn list_open_for_agent_uses_registry_and_filters_terminal_tasks() { ); } +#[tokio::test] +async fn list_children_uses_parent_index_and_keeps_terminal_tasks_for_traversal() { + let tmp = TempDir::new(); + let store = FsBackgroundTaskStore::new(&tmp.project_path()); + let parent = running_task(1, 10, 100, 1_000); + let child = running_task(2, 10, 100, 1_000) + .with_parent_task(parent.id) + .unwrap(); + let terminal_child = running_task(3, 10, 100, 1_000) + .with_parent_task(parent.id) + .unwrap() + .complete(BackgroundTaskResult::Success { + finished_at_ms: 1_010, + exit_code: Some(0), + summary: "done".to_owned(), + stdout_tail: None, + stderr_tail: None, + }) + .unwrap(); + let unrelated = running_task(4, 10, 100, 1_000); + + store.create(&parent).await.unwrap(); + store.create(&child).await.unwrap(); + store.create(&terminal_child).await.unwrap(); + store.create(&unrelated).await.unwrap(); + + assert_eq!( + store.list_children(parent.id).await.unwrap(), + vec![child.clone(), terminal_child.clone()] + ); + + let reborn = FsBackgroundTaskStore::new(&tmp.project_path()); + assert_eq!( + reborn.list_children(parent.id).await.unwrap(), + vec![child, terminal_child] + ); +} + #[tokio::test] async fn list_undelivered_completions_and_mark_are_idempotent() { let tmp = TempDir::new(); diff --git a/crates/infrastructure/tests/orchestrator_watcher.rs b/crates/infrastructure/tests/orchestrator_watcher.rs index a5bae6e3..a28026fb 100644 --- a/crates/infrastructure/tests/orchestrator_watcher.rs +++ b/crates/infrastructure/tests/orchestrator_watcher.rs @@ -377,6 +377,20 @@ impl BackgroundTaskStore for InMemoryBackgroundTaskStore { .collect()) } + async fn list_children( + &self, + parent_task_id: TaskId, + ) -> Result, BackgroundTaskPortError> { + Ok(self + .tasks + .lock() + .unwrap() + .values() + .filter(|task| task.parent_task_id == Some(parent_task_id)) + .cloned() + .collect()) + } + async fn list_undelivered_completions( &self, ) -> Result, BackgroundTaskPortError> { diff --git a/frontend/src/adapters/workStateNormalization.ts b/frontend/src/adapters/workStateNormalization.ts index a47655a9..d4fbdae4 100644 --- a/frontend/src/adapters/workStateNormalization.ts +++ b/frontend/src/adapters/workStateNormalization.ts @@ -113,10 +113,12 @@ function normalizeBackgroundTask( const conversationId = optionalString( task.conversationId ?? task.conversation_id, ); + const parentTaskId = optionalString(task.parentTaskId ?? task.parent_task_id); return { taskId: stringValue(task.taskId, `legacy-task-${index}`), ownerAgentId: stringValue(task.ownerAgentId, fallbackAgentId), projectId: stringValue(task.projectId), + ...(parentTaskId ? { parentTaskId } : {}), kind: stringValue(task.kind, "background"), status: normalizeBackgroundStatus(task.status ?? task.state), exitCode: nullableNumber(task.exitCode), diff --git a/frontend/src/domain/index.ts b/frontend/src/domain/index.ts index a7d78f04..678128d2 100644 --- a/frontend/src/domain/index.ts +++ b/frontend/src/domain/index.ts @@ -611,6 +611,7 @@ export interface BackgroundCompletion { taskId: string; ownerAgentId: string; projectId: string; + parentTaskId?: string; kind: string; status: | "running" diff --git a/frontend/src/plugins/runtime/services.ts b/frontend/src/plugins/runtime/services.ts index 9ec77841..052a41db 100644 --- a/frontend/src/plugins/runtime/services.ts +++ b/frontend/src/plugins/runtime/services.ts @@ -141,6 +141,7 @@ export interface BackgroundTaskStatus { taskId: string; ownerAgentId: string; projectId: string; + parentTaskId?: string; kind: string; status: | "pending" @@ -447,6 +448,7 @@ function toBackgroundTaskStatus(task: BackgroundCompletion): BackgroundTaskStatu taskId: task.taskId, ownerAgentId: task.ownerAgentId, projectId: task.projectId, + ...(task.parentTaskId ? { parentTaskId: task.parentTaskId } : {}), kind: task.kind, status: task.status, exitCode: task.exitCode,