blomios/IdeA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(runtime): cancel en cascade des tasks filles — stop agent / cancel parent (#200, QA verte)

Browse Source 
Cancel transitif, idempotent et ownership-aware sur le graphe parent→enfants des background tasks:

- domain/background_task + ports: modèle parent/enfant (parent_task_id, rejet du self-parent), index children_by_parent, traversal des descendants en largeur.

- application/background + infrastructure/store: cascade cancel (annule les descendants running/open, laisse les terminaux), idempotente face aux double-cancel et ordres de terminaison non déterministes; respect de l'ownership (parent_task_link).

- orchestrator/service: stop agent → cascade cancel des tasks lancées par cet agent (relatesTo #196); cancel task parent → cascade de ses sous-tasks (dependsOn #199). Refactor unifiant le chemin de cancel, sans sémantique #196/#199 regreffée (tests hérités verts).

- frontend: projection parentTaskId (workstate normalization/domain/services).

Tests QA verts: domain background_task, application background_task_cancel (cascades_background_task_subtree, cascades_to_running_child, cascades_to_open_descendants_and_leaves_terminal_child, parent_task_link_roundtrips_and_rejects_self_parent), orchestrator_service, workstate_actions, workstate.test.tsx.
This commit is contained in:
Blomios
2026-08-12 16:02:17 +02:00
parent a3bda8ebc0
commit 803e9de093
22 changed files with 561 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
crates/application/src/background/mod.rs
View File

@ -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<BackgroundTaskRendezvousLink>,
/// Optional parent background task. Used for deterministic cancel cascades.
pub parent_task_id: Option<TaskId>,
/// Optional absolute deadline, epoch milliseconds.
pub deadline_ms: Option<u64>,
}
@ -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<BackgroundTaskRendezvousLink>,
parent_task_id: Option<TaskId>,
deadline_ms: Option<u64>,
) -> Result<BackgroundTask, AppError> {
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<CancelBackgroundTaskOutput, AppError> {
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<Vec<BackgroundTask>, 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<Vec<BackgroundTask>, 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<BackgroundTask>,
) -> Result<Vec<BackgroundTask>, AppError> {
let mut queue: VecDeque<BackgroundTask> = 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<BackgroundTask, AppError> {
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 })

127
crates/application/src/orchestrator/service.rs
View File

@ -574,6 +574,7 @@ pub struct OrchestratorOutcome {
#[derive(Debug, Clone)]
struct ActiveRendezvous {
link: BackgroundTaskRendezvousLink,
task_id: Option<TaskId>,
background_tasks: Vec<TaskId>,
}
@ -737,6 +738,14 @@ impl OrchestratorService {
.map(|r| r.link.clone())
}
fn active_rendezvous_task_id(&self, agent: RuntimeAgentKey) -> Option<TaskId> {
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<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
agent: RuntimeAgentKey,
) -> Vec<TaskId> {
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<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
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<TaskId>,
) -> 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<Vec<TaskId>>,
}
#[async_trait::async_trait]
impl BackgroundTaskRunner for SpyBackgroundRunner {
async fn spawn(
&self,
spec: domain::ports::BackgroundTaskSpec,
) -> Result<domain::ports::BackgroundTaskHandle, domain::ports::BackgroundTaskPortError>
{
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]

15
crates/application/src/plugin/mod.rs
View File

@ -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<Vec<BackgroundTask>, 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<Vec<BackgroundTask>, BackgroundTaskPortError> {

3
crates/application/src/workstate/mod.rs
View File

@ -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<TaskId>,
/// Kind discriminant.
pub kind: BackgroundTaskKindLabel,
/// Current lifecycle state.
@ -609,6 +611,7 @@ impl From<BackgroundTask> 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,