fix(application,domain,infrastructure,web-server,app-tauri,frontend): arrêt d'agent fiable avec cascade sur l'arbre de propriété et statut cohérent — #320 (QA verte)

Symptôme: stopper un agent (ex Main sur UnityPlugin) laissait des travaux
poursuivre, la cascade n'atteignait pas les agents lancés et la pastille
restait orange pendant que l'agent continuait à en appeler d'autres.

- barrière d'admission par génération d'exécution (ExecutionControl): un stop
  ferme l'admission de tout le sous-arbre propriété/attente AVANT toute
  annulation asynchrone — plus aucune nouvelle action inter-agents après stop
- arbre de propriété durable en-process (launched_by) + dépendances d'attente
  actives: la cascade couvre transitivement agents et travaux lancés
- commandes d'un agent arrêté rejetées (AppError::AgentStopped); stop
  lui-même admis pour rester idempotent; réactivation sur relance humaine
  (génération +1) et enregistrement du propriétaire au spawn
- admission exposée aux adaptateurs hors commandes cœur (tickets, templates,
  plugins) via ensure_agent_execution_active; MCP server/tools, web-server et
  commandes Tauri alignés
- cohérence live-state: statut Idle avec intent « Arrêté » ou
  « Arrêt incomplet — Certains travaux sont encore actifs » selon l'issue de
  la cascade; workstate et surfaces UI (AgentsPanel, LayoutGrid,
  ProjectWorkStatePanel) reflètent l'état réel, plus de pastille orange fantôme
- tests: orchestrator_service (cascade/générations/idempotence),
  workstate_actions, background_task_cancel, error_codes, agents, layout,
  workstate

QA verte sur fix/320-agent-stop-cascade. Aucun push.
This commit is contained in:
2026-09-11 16:42:29 +02:00
parent 1dc892d80b
commit 1316e87010
20 changed files with 1048 additions and 154 deletions

View File

@ -356,9 +356,12 @@ impl CancelBackgroundTask {
async fn cancel_one(&self, task: BackgroundTask) -> Result<BackgroundTask, AppError> {
let task_id = task.id;
let cancelling = if task.state == BackgroundTaskState::Cancelling {
task
} else {
// A repeated/cascading stop must not signal the same process twice. The
// first caller already persisted Cancelling before invoking the runner.
if task.state == BackgroundTaskState::Cancelling {
return Ok(task);
}
let cancelling = {
let now = u64::try_from(self.clock.now_millis().max(0)).unwrap_or(0);
let cancelling = task
.transition(BackgroundTaskState::Cancelling, now)

View File

@ -103,6 +103,14 @@ pub enum AppError {
agent_id: AgentId,
},
/// The agent execution generation was explicitly stopped. New work from or
/// toward that generation is rejected until an explicit launch reactivates it.
#[error("agent {agent_id} is stopped and cannot accept new work")]
AgentStopped {
/// Agent protected by the stop barrier.
agent_id: AgentId,
},
/// The delegation target finished its structured/headless turn without producing
/// a [`ReplyEvent::Final`](domain::ports::ReplyEvent::Final), so the rendezvous
/// could not capture an inline answer. Surfaced promptly instead of blocking the
@ -154,6 +162,7 @@ impl AppError {
Self::AgentNotBoundToCell { .. } => "AGENT_NOT_BOUND_TO_CELL",
Self::ConflictAgentVisibleElsewhere { .. } => "CONFLICT_AGENT_VISIBLE_ELSEWHERE",
Self::RestartStopTimeout { .. } => "STOP_TIMEOUT",
Self::AgentStopped { .. } => "AGENT_STOPPED",
Self::TargetReturnedNoReply(_) => "TARGET_RETURNED_NO_REPLY",
Self::TargetCeilingActive(_) => "RENDEZVOUS_CEILING_ACTIVE",
Self::Internal(_) => "INTERNAL",

View File

@ -13,7 +13,7 @@
//! no filesystem watching, no JSON, no process spawning here — those are the
//! infrastructure adapter's job. That keeps this fully unit-testable with fakes.
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
use std::sync::{Arc, Mutex as StdMutex};
@ -370,7 +370,12 @@ impl RestartRendezvousCoordinator {
let restart_id = match &entry.state {
RestartRendezvousState::Restarting { restart_id }
| RestartRendezvousState::Ready { restart_id } => restart_id.clone(),
_ => return false,
RestartRendezvousState::Idle if entry.active_waiters > 0 => {
format!("stop-{}", entry.activity.saturating_add(1))
}
RestartRendezvousState::Cancelled { .. }
| RestartRendezvousState::Failed { .. }
| RestartRendezvousState::Idle => return false,
};
if let Some(turn) = entry.suspended_turn.as_ref() {
crate::diag!(
@ -1117,6 +1122,9 @@ pub struct OrchestratorService {
/// attach `idea_run_in_background` tasks for observability/cancellation while the
/// synchronous answer remains the first structured `Final`.
active_rendezvous: StdMutex<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
/// Runtime-generation admission barrier and durable-in-process ownership tree.
/// A stop closes admission before any asynchronous cancellation begins.
execution_control: StdMutex<ExecutionControl>,
/// Active cross-project calls keyed by caller identity and caller-supplied
/// correlation id. Cancel uses this registry as its routing authority.
cross_project_requests: StdMutex<HashMap<CrossProjectRequestKey, ActiveCrossProjectRequest>>,
@ -1266,6 +1274,18 @@ struct ActiveRendezvous {
background_tasks: Vec<TaskId>,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct ExecutionGeneration {
generation: u64,
stopped: bool,
}
#[derive(Debug, Default)]
struct ExecutionControl {
generations: HashMap<RuntimeAgentKey, ExecutionGeneration>,
launched_by: HashMap<RuntimeAgentKey, RuntimeAgentKey>,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct CrossProjectRequestKey {
requester_project_id: domain::ProjectId,
@ -1309,6 +1329,7 @@ impl OrchestratorService {
wait_for: StdMutex::new(WaitForGraph::new()),
active_waits: StdMutex::new(Vec::new()),
active_rendezvous: StdMutex::new(HashMap::new()),
execution_control: StdMutex::new(ExecutionControl::default()),
cross_project_requests: StdMutex::new(HashMap::new()),
events: None,
ask_locks: StdMutex::new(HashMap::new()),
@ -1467,6 +1488,124 @@ impl OrchestratorService {
Arc::clone(locks.entry(runtime_key(project, agent_id)).or_default())
}
fn execution_token(&self, key: RuntimeAgentKey) -> Result<u64, AppError> {
let control = self
.execution_control
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let generation = control.generations.get(&key).copied().unwrap_or_default();
if generation.stopped {
Err(AppError::AgentStopped {
agent_id: key.agent_id,
})
} else {
Ok(generation.generation)
}
}
fn execution_token_is_current(&self, key: RuntimeAgentKey, token: u64) -> bool {
self.execution_control
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.generations
.get(&key)
.copied()
.unwrap_or_default()
== (ExecutionGeneration {
generation: token,
stopped: false,
})
}
fn reactivate_execution(
&self,
project: &Project,
agent_id: AgentId,
requester: Option<AgentId>,
) {
let key = runtime_key(project, agent_id);
let mut control = self
.execution_control
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let state = control.generations.entry(key).or_default();
if state.stopped {
state.generation = state.generation.wrapping_add(1);
state.stopped = false;
}
match requester {
Some(owner) if owner != agent_id => {
control.launched_by.insert(key, runtime_key(project, owner));
}
_ => {
control.launched_by.remove(&key);
}
}
}
/// Opens a fresh execution generation after an explicit human-driven launch.
/// Composition roots must call this alongside direct [`LaunchAgent`] entrypoints.
pub fn reactivate_agent_execution(&self, project: &Project, agent_id: AgentId) {
self.reactivate_execution(project, agent_id, None);
}
/// Admission check for driving adapters that expose tools outside the core
/// [`OrchestratorCommand`] enum (tickets, templates and plugins included).
pub fn ensure_agent_execution_active(
&self,
project: &Project,
agent_id: AgentId,
) -> Result<(), AppError> {
self.execution_token(runtime_key(project, agent_id))
.map(drop)
}
/// Closes admission for the full ownership/wait subtree before cancellation.
/// Returns only descendants; the root is always handled by the caller.
fn begin_stop_generation(&self, project: &Project, agent_id: AgentId) -> Vec<AgentId> {
let root = runtime_key(project, agent_id);
let wait_descendants = self.active_wait_dependencies_from_wait_edges(agent_id);
let mut control = self
.execution_control
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let mut keys = vec![root];
let mut seen = HashSet::from([root]);
while let Some(parent) = keys.pop() {
for child in control
.launched_by
.iter()
.filter_map(|(child, owner)| (*owner == parent).then_some(*child))
.collect::<Vec<_>>()
{
if seen.insert(child) {
keys.push(child);
}
}
}
for descendant in wait_descendants {
seen.insert(runtime_key(project, descendant));
}
for key in &seen {
let state = control.generations.entry(*key).or_default();
if !state.stopped {
state.generation = state.generation.wrapping_add(1);
state.stopped = true;
}
}
seen.remove(&root);
seen.into_iter().map(|key| key.agent_id).collect()
}
fn active_wait_dependencies_from_wait_edges(&self, agent: AgentId) -> Vec<AgentId> {
let edges = self
.active_waits
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone();
Self::active_wait_dependencies_from_edges(&edges, agent)
}
fn active_rendezvous_link(
&self,
agent: RuntimeAgentKey,
@ -1510,12 +1649,27 @@ impl OrchestratorService {
/// lock-bounded; callers perform the actual interruption/stop outside the mutex.
#[must_use]
pub fn active_wait_dependencies(&self, agent: AgentId) -> Vec<AgentId> {
let edges = self
.active_waits
let mut descendants = self.active_wait_dependencies_from_wait_edges(agent);
let control = self
.execution_control
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone();
Self::active_wait_dependencies_from_edges(&edges, agent)
.unwrap_or_else(std::sync::PoisonError::into_inner);
let roots: Vec<_> = control
.launched_by
.iter()
.filter_map(|(child, owner)| (owner.agent_id == agent).then_some(child.agent_id))
.collect();
let mut stack = roots;
while let Some(child) = stack.pop() {
if descendants.contains(&child) {
continue;
}
descendants.push(child);
stack.extend(control.launched_by.iter().filter_map(|(candidate, owner)| {
(owner.agent_id == child).then_some(candidate.agent_id)
}));
}
descendants
}
/// Returns the transitive dependencies of `agent` in an active wait-edge snapshot.
@ -1873,6 +2027,44 @@ impl OrchestratorService {
}
}
async fn mark_stopped_best_effort(&self, root: &ProjectPath, agent_id: AgentId) {
let Some(provider) = &self.live_state else {
return;
};
let Some(update) = provider.live_state_for(root) else {
return;
};
let _ = update
.execute(UpdateLiveStateInput {
agent_id,
ticket: None,
intent: "Arrêté".to_owned(),
status: domain::live_state::WorkStatus::Idle,
progress: None,
last_delegation: None,
})
.await;
}
async fn mark_stop_incomplete_best_effort(&self, root: &ProjectPath, agent_id: AgentId) {
let Some(provider) = &self.live_state else {
return;
};
let Some(update) = provider.live_state_for(root) else {
return;
};
let _ = update
.execute(UpdateLiveStateInput {
agent_id,
ticket: None,
intent: "Arrêt incomplet — Certains travaux sont encore actifs.".to_owned(),
status: domain::live_state::WorkStatus::Idle,
progress: None,
last_delegation: None,
})
.await;
}
async fn mark_cross_project_requester_best_effort(
&self,
root: &ProjectPath,
@ -2028,6 +2220,9 @@ impl OrchestratorService {
project: &Project,
command: OrchestratorCommand,
) -> Result<OrchestratorOutcome, AppError> {
if let Some(actor) = Self::command_actor(&command) {
self.execution_token(runtime_key(project, actor))?;
}
match command {
OrchestratorCommand::RestartSelf { agent, reason } => {
let scheduled = self
@ -2053,8 +2248,9 @@ impl OrchestratorService {
profile,
context,
visibility,
requester,
} => {
self.spawn_agent(project, name, profile, context, visibility)
self.spawn_agent(project, name, profile, context, visibility, requester)
.await
}
OrchestratorCommand::AskAgent {
@ -2259,6 +2455,37 @@ impl OrchestratorService {
}
}
/// Authenticated agent that owns an agent-originated command. Stop itself is
/// deliberately admitted so repeated/self stops remain idempotent.
fn command_actor(command: &OrchestratorCommand) -> Option<AgentId> {
match command {
OrchestratorCommand::RestartSelf { agent, .. }
| OrchestratorCommand::Reply { from: agent, .. }
| OrchestratorCommand::SetWorkState { agent, .. }
| OrchestratorCommand::RunInBackground { owner: agent, .. } => Some(*agent),
OrchestratorCommand::SpawnAgent { requester, .. }
| OrchestratorCommand::AskAgent { requester, .. } => *requester,
OrchestratorCommand::ReadContext { requester, .. }
| OrchestratorCommand::ProposeContext { requester, .. }
| OrchestratorCommand::UpdateProjectContext { requester, .. }
| OrchestratorCommand::ReadMemory { requester, .. }
| OrchestratorCommand::ReadSkill { requester, .. }
| OrchestratorCommand::WriteMemory { requester, .. }
| OrchestratorCommand::ReadWorkState { requester } => requester.as_agent(),
OrchestratorCommand::CreateScheduledTask { requester, .. }
| OrchestratorCommand::UpdateScheduledTask { requester, .. }
| OrchestratorCommand::DeleteScheduledTask { requester, .. }
| OrchestratorCommand::RunScheduledTaskNow { requester, .. }
| OrchestratorCommand::SetScheduledTaskEnabled { requester, .. } => Some(*requester),
OrchestratorCommand::StopAgent { .. }
| OrchestratorCommand::ListAgents
| OrchestratorCommand::CreateSkill { .. }
| OrchestratorCommand::UpdateAgentContext { .. }
| OrchestratorCommand::ListScheduledTasks
| OrchestratorCommand::ReadScheduledTask { .. } => None,
}
}
fn require_scheduled_tasks(&self) -> Result<&Arc<ScheduledProjectTaskService>, AppError> {
self.scheduled_project_tasks.as_ref().ok_or_else(|| {
AppError::Invalid("scheduled project tasks are not configured".to_owned())
@ -2370,6 +2597,11 @@ impl OrchestratorService {
requester_agent_id: AgentId,
correlation_id: String,
) -> Result<OrchestratorOutcome, AppError> {
self.execution_token(RuntimeAgentKey::new(
requester_project_id,
requester_agent_id,
))?;
self.execution_token(RuntimeAgentKey::new(target_project_id, target_agent_id))?;
let requester_project = self
.load_project_for_cross_tool(requester_project_id)
.await?;
@ -2827,6 +3059,7 @@ impl OrchestratorService {
profile: Option<String>,
context: Option<String>,
visibility: OrchestratorVisibility,
requester: Option<AgentId>,
) -> Result<OrchestratorOutcome, AppError> {
let existing = self.find_agent_id_by_name(project, &name).await?;
@ -2849,6 +3082,7 @@ impl OrchestratorService {
created.agent.id
}
};
self.reactivate_execution(project, agent_id, requester);
if let Some(session_id) = self
.sessions
@ -2990,6 +3224,8 @@ impl OrchestratorService {
) -> Result<OrchestratorOutcome, AppError> {
let target = agent.name.clone();
let agent_id = agent.id;
let agent_key = runtime_key(project, agent_id);
let execution_token = self.execution_token(agent_key)?;
// Détection de cycle (cadrage C3 §6) : si l'ask vient d'un **agent** A vers la
// cible B, refuser AVANT tout enqueue si poser l'arête A→B fermerait un cycle
// d'attente (B attend déjà …→A). Pur, sans I/O ⇒ jamais de deadlock.
@ -3049,6 +3285,9 @@ impl OrchestratorService {
return Err(AppError::from(domain::ports::AgentSessionError::Timeout));
}
};
if !self.execution_token_is_current(agent_key, execution_token) {
return Err(AppError::AgentStopped { agent_id });
}
// Poser l'arête d'attente A→B (retirée en fin de tour par le RAII `_edge`).
let _edge = requester.map(|from| WaitEdgeGuard::new(self, from, agent_id));
@ -3089,6 +3328,7 @@ impl OrchestratorService {
requester,
task,
session,
execution_token,
)
.await
}
@ -3115,6 +3355,7 @@ impl OrchestratorService {
requester: Option<AgentId>,
task: String,
session: Arc<dyn domain::ports::AgentSession>,
execution_token: u64,
) -> Result<OrchestratorOutcome, AppError> {
let (input, mailbox) = match (&self.input, &self.mailbox) {
(Some(i), Some(m)) => (i, m),
@ -3127,6 +3368,9 @@ impl OrchestratorService {
}
};
let agent_key = runtime_key(project, agent_id);
if !self.execution_token_is_current(agent_key, execution_token) {
return Err(AppError::AgentStopped { agent_id });
}
// Checkpoint Prompt (best-effort), AVANT de déplacer `task` dans le ticket.
let prompt_source = match requester {
@ -3182,6 +3426,10 @@ impl OrchestratorService {
Self::distill_intent(&task),
)
.await;
if !self.execution_token_is_current(agent_key, execution_token) {
self.mark_stopped_best_effort(&project.root, agent_id).await;
return Err(AppError::AgentStopped { agent_id });
}
let live_state_update = self
.live_state
.as_ref()
@ -3348,7 +3596,7 @@ impl OrchestratorService {
};
// Borne par la fenêtre d'inactivité (réarmée sur signe de vie) sous plafond absolu.
let result = match self
let watchdog_outcome = self
.run_ask_with_watchdog(
wait,
turn_timeout,
@ -3361,8 +3609,19 @@ impl OrchestratorService {
},
Some(agent_key),
)
.await
{
.await;
if !self.execution_token_is_current(agent_key, execution_token) {
self.complete_rendezvous_task_cancelled(
project,
agent_id,
rendezvous_task,
"agent execution was stopped".to_owned(),
)
.await?;
live_state_guard.disarm();
return Err(AppError::AgentStopped { agent_id });
}
let result = match watchdog_outcome {
WatchdogOutcome::Resolved(Ok(content)) => {
crate::diag!(
"[rendezvous] ask turn finalized (structured): target={target} \
@ -3507,6 +3766,8 @@ impl OrchestratorService {
agent_id: AgentId,
text: String,
) -> Result<OrchestratorOutcome, AppError> {
// A human submit is an explicit request for a fresh turn after a prior stop.
self.reactivate_execution(project, agent_id, None);
let input = self.input.as_ref().ok_or_else(|| {
AppError::Invalid(
"l'entrée médiée (submit_agent_input) n'est pas disponible : \
@ -4050,62 +4311,62 @@ impl OrchestratorService {
}
}
let restart_cancelled = self
.restart_rendezvous
.cancel(runtime_key(project, agent_id));
let structured = self
.structured
.clone()
.unwrap_or_else(|| Arc::new(StructuredSessions::new()));
let dependencies = self.active_wait_dependencies(agent_id);
for dep in &dependencies {
self.stop_agent_execution_children(project, *dep).await;
if let Some(input) = &self.input {
input.preempt(runtime_key(project, *dep));
}
if let Some(session) = structured.session_for_agent_in_project(project.id, dep) {
session
.cancel_current_turn()
.await
.map_err(|err| AppError::Process(err.to_string()))?;
}
}
self.stop_agent_execution_children(project, agent_id).await;
if let Some(input) = &self.input {
input.preempt(runtime_key(project, agent_id));
}
if let Some(session) = structured.session_for_agent_in_project(project.id, &agent_id) {
session
.cancel_current_turn()
.await
.map_err(|err| AppError::Process(err.to_string()))?;
return Ok(OrchestratorOutcome {
detail: format!("cancelled current work for agent {label}"),
reply: None,
});
}
let live = LiveSessions::new(Arc::clone(&self.sessions), structured);
if stop_live_agent_runtime_in_project(project, agent_id, &live, &self.close_terminal)
.await?
.is_empty()
let mut stopped_any = false;
// Admission for root and descendants closes before the first awaited
// runtime shutdown, so the owner cannot delegate during the cascade.
let mut first_error = self
.stop_agent_execution_children(project, agent_id)
.await
.err();
for dep in dependencies
.iter()
.copied()
.chain(std::iter::once(agent_id))
{
if restart_cancelled {
return Ok(OrchestratorOutcome {
detail: format!("cancelled restarting work for agent {label}"),
reply: None,
});
if let Some(input) = &self.input {
input.preempt(runtime_key(project, dep));
}
return Err(AppError::NotFound(format!(
"running session for agent {label}"
)));
match stop_live_agent_runtime_in_project(project, dep, &live, &self.close_terminal)
.await
{
Ok(stopped) => stopped_any |= !stopped.is_empty(),
Err(error) => {
first_error.get_or_insert(error);
}
}
}
if let Some(error) = first_error {
for participant in dependencies
.iter()
.copied()
.chain(std::iter::once(agent_id))
{
self.mark_stop_incomplete_best_effort(&project.root, participant)
.await;
}
return Err(error);
}
for participant in dependencies
.iter()
.copied()
.chain(std::iter::once(agent_id))
{
self.mark_stopped_best_effort(&project.root, participant)
.await;
}
Ok(OrchestratorOutcome {
detail: format!("stopped agent {label}"),
detail: if stopped_any {
format!("stopped agent {label} and its active execution subtree")
} else {
format!("agent {label} was already stopped")
},
reply: None,
})
}
@ -4488,12 +4749,67 @@ impl<'a> WaitEdgeGuard<'a> {
#[async_trait::async_trait]
impl StopAgentExecutionCascade for OrchestratorService {
async fn stop_agent_execution_children(&self, project: &Project, agent_id: AgentId) {
async fn stop_agent_execution_children(
&self,
project: &Project,
agent_id: AgentId,
) -> Result<(), AppError> {
let affected = self.begin_stop_generation(project, agent_id);
let mut participants = affected;
participants.push(agent_id);
let participant_keys: HashSet<_> = participants
.iter()
.map(|agent| runtime_key(project, *agent))
.collect();
let rendezvous_targets: Vec<_> = self
.active_rendezvous
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.iter()
.filter_map(|(target, active)| {
(participant_keys.contains(target)
|| active
.link
.requester_agent_id
.is_some_and(|requester| participants.contains(&requester)))
.then_some(*target)
})
.collect();
for target in rendezvous_targets {
self.restart_rendezvous.cancel(target);
if let Some(input) = &self.input {
input.preempt(target);
}
}
for participant in &participants {
self.restart_rendezvous
.cancel(runtime_key(project, *participant));
if let Some(session_limits) = &self.session_limits {
session_limits.cancel_resume(*participant);
}
if let Some(input) = &self.input {
input.preempt(runtime_key(project, *participant));
}
if let Some(provider) = &self.live_state {
if let Some(update) = provider.live_state_for(&project.root) {
let _ = update
.execute(UpdateLiveStateInput {
agent_id: *participant,
ticket: None,
intent: "Arrêt en cours…".to_owned(),
status: domain::live_state::WorkStatus::Idle,
progress: None,
last_delegation: None,
})
.await;
}
}
}
let (Some(store), Some(runner)) = (&self.background_tasks, &self.background_runner) else {
return;
return Ok(());
};
let Some(clock) = &self.clock else {
return;
return Ok(());
};
let mut cancel = crate::background::CancelBackgroundTask::new(
Arc::clone(store),
@ -4503,7 +4819,16 @@ impl StopAgentExecutionCascade for OrchestratorService {
if let Some(events) = &self.events {
cancel = cancel.with_events(Arc::clone(events));
}
let _ = cancel.cancel_subtree_for_agent(project.id, agent_id).await;
let mut first_error = None;
for participant in participants {
if let Err(error) = cancel
.cancel_subtree_for_agent(project.id, participant)
.await
{
first_error.get_or_insert(error);
}
}
first_error.map_or(Ok(()), Err)
}
}

View File

@ -229,16 +229,20 @@ impl RestartCellAgentRuntime {
}
}
/// Best-effort stopper for execution resources owned by one live agent turn.
/// Stopper for execution resources owned by one live agent turn.
///
/// Implementations must be idempotent: a missing, already-finished or already-stopped
/// child resource is a successful no-op. Failures are deliberately not propagated by
/// [`StopAgentRuntime`], because the product contract is "stop the agent; cascade related
/// work best-effort".
/// child resource is a successful no-op. Implementations keep cancelling remaining
/// resources after a failure and return the first error so the caller can expose an
/// `Arrêt incomplet` terminal state.
#[async_trait]
pub trait StopAgentExecutionCascade: Send + Sync {
/// Requests cancellation of child execution resources for `agent_id` in `project`.
async fn stop_agent_execution_children(&self, project: &Project, agent_id: AgentId);
async fn stop_agent_execution_children(
&self,
project: &Project,
agent_id: AgentId,
) -> Result<(), AppError>;
}
/// Stops an already-running agent's current work.
@ -316,39 +320,74 @@ impl StopAgentRuntime {
/// Stops the agent's current work.
///
/// # Errors
/// - [`AppError::NotFound`] when the agent has no live session,
/// - [`AppError::Process`] when the PTY kill or structured shutdown fails.
/// Repeated stops are successful. A cancellation or runtime failure is returned
/// only after all remaining resources have still been attempted.
pub async fn execute(
&self,
input: StopLiveAgentInput,
) -> Result<StopLiveAgentOutput, AppError> {
self.stop_dependencies(&input.project, input.agent_id)
.await?;
let (dependencies, dependency_error) =
self.stop_dependencies(&input.project, input.agent_id).await;
if let Some(coordinator) = &self.restart_rendezvous {
coordinator.cancel(RuntimeAgentKey::new(input.project.id, input.agent_id));
}
if let Some(mediator) = &self.input {
mediator.preempt(RuntimeAgentKey::new(input.project.id, input.agent_id));
}
let stopped = stop_live_agent_runtime_in_project(
let runtime_result = stop_live_agent_runtime_in_project(
&input.project,
input.agent_id,
&self.live,
&self.close,
)
.await?;
.await;
let stopped = runtime_result.clone().unwrap_or_default();
if let Some(error) = runtime_result.err().or(dependency_error) {
if let Some(provider) = &self.live_state {
if let Some(update) = provider.live_state_for(&input.project.root) {
for agent_id in dependencies
.iter()
.copied()
.chain(std::iter::once(input.agent_id))
{
let _ = update
.execute(crate::UpdateLiveStateInput {
agent_id,
ticket: None,
intent: "Arrêt incomplet — Certains travaux sont encore actifs."
.to_owned(),
status: WorkStatus::Idle,
progress: None,
last_delegation: None,
})
.await;
}
}
}
return Err(error);
}
if let Some(provider) = &self.live_state {
if let Some(update) = provider.live_state_for(&input.project.root) {
let _ = update
.execute(crate::UpdateLiveStateInput {
agent_id: input.agent_id,
ticket: None,
intent: input.reason.unwrap_or_else(|| "arrêté".to_owned()),
status: WorkStatus::Idle,
progress: None,
last_delegation: None,
})
.await;
for agent_id in dependencies
.iter()
.copied()
.chain(std::iter::once(input.agent_id))
{
let _ = update
.execute(crate::UpdateLiveStateInput {
agent_id,
ticket: None,
intent: if agent_id == input.agent_id {
input.reason.clone().unwrap_or_else(|| "Arrêté".to_owned())
} else {
"Arrêté".to_owned()
},
status: WorkStatus::Idle,
progress: None,
last_delegation: None,
})
.await;
}
}
}
Ok(StopLiveAgentOutput {
@ -362,25 +401,36 @@ impl StopAgentRuntime {
&self,
project: &Project,
agent_id: AgentId,
) -> Result<(), AppError> {
) -> (Vec<AgentId>, Option<AppError>) {
let mut first_error = None;
if let Some(cascade) = &self.cascade {
cascade
if let Err(error) = cascade
.stop_agent_execution_children(project, agent_id)
.await;
.await
{
first_error = Some(error);
}
}
let Some(waits) = &self.waits else {
return Ok(());
return (Vec::new(), first_error);
};
for dep in waits.active_wait_dependencies(agent_id) {
let dependencies = waits.active_wait_dependencies(agent_id);
for dep in dependencies.iter().copied() {
if let Some(cascade) = &self.cascade {
cascade.stop_agent_execution_children(project, dep).await;
if let Err(error) = cascade.stop_agent_execution_children(project, dep).await {
first_error.get_or_insert(error);
}
}
if let Some(mediator) = &self.input {
mediator.preempt(RuntimeAgentKey::new(project.id, dep));
}
let _ = stop_live_agent_runtime_in_project(project, dep, &self.live, &self.close).await;
if let Err(error) =
stop_live_agent_runtime_in_project(project, dep, &self.live, &self.close).await
{
first_error.get_or_insert(error);
}
}
Ok(())
(dependencies, first_error)
}
}