diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 668bdf69..506333c9 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -2453,7 +2453,7 @@ pub async fn launch_agent( let output = state .launch_agent .execute(LaunchAgentInput { - project, + project: project.clone(), agent_id, rows: request.rows, cols: request.cols, @@ -2473,6 +2473,9 @@ pub async fn launch_agent( }) .await .map_err(ErrorDto::from)?; + state + .orchestrator_service + .reactivate_agent_execution(&project, agent_id); if let Ok(mut contexts) = state.resume_contexts.lock() { contexts.insert( diff --git a/crates/application/src/background/mod.rs b/crates/application/src/background/mod.rs index e11b56a6..8e9c7580 100644 --- a/crates/application/src/background/mod.rs +++ b/crates/application/src/background/mod.rs @@ -356,9 +356,12 @@ impl CancelBackgroundTask { async fn cancel_one(&self, task: BackgroundTask) -> Result { 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) diff --git a/crates/application/src/error.rs b/crates/application/src/error.rs index c8a76d53..f1591ed9 100644 --- a/crates/application/src/error.rs +++ b/crates/application/src/error.rs @@ -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", diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index c432ead7..b833cf02 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -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>, + /// Runtime-generation admission barrier and durable-in-process ownership tree. + /// A stop closes admission before any asynchronous cancellation begins. + execution_control: StdMutex, /// 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>, @@ -1266,6 +1274,18 @@ struct ActiveRendezvous { background_tasks: Vec, } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +struct ExecutionGeneration { + generation: u64, + stopped: bool, +} + +#[derive(Debug, Default)] +struct ExecutionControl { + generations: HashMap, + launched_by: HashMap, +} + #[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 { + 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, + ) { + 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 { + 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::>() + { + 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 { + 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 { - 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 { + 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 { + 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, 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 { + 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, context: Option, visibility: OrchestratorVisibility, + requester: Option, ) -> Result { 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 { 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, task: String, session: Arc, + execution_token: u64, ) -> Result { 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 { + // 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) } } diff --git a/crates/application/src/workstate/actions.rs b/crates/application/src/workstate/actions.rs index 464a4d5a..352da2da 100644 --- a/crates/application/src/workstate/actions.rs +++ b/crates/application/src/workstate/actions.rs @@ -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 { - 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, Option) { + 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) } } diff --git a/crates/application/tests/background_task_cancel.rs b/crates/application/tests/background_task_cancel.rs index 9a2a5ad1..f5c7afe3 100644 --- a/crates/application/tests/background_task_cancel.rs +++ b/crates/application/tests/background_task_cancel.rs @@ -258,7 +258,7 @@ async fn cancel_terminal_task_is_explicit_error_without_runner_call() { } #[tokio::test] -async fn cancel_already_cancelling_is_idempotent_and_retries_runner_cancel() { +async fn cancel_already_cancelling_is_idempotent_without_duplicate_runner_signal() { let store = Arc::new(FakeStore::default()); let runner = Arc::new(FakeRunner::ok()); let events = Arc::new(SpyEvents::default()); @@ -274,7 +274,7 @@ async fn cancel_already_cancelling_is_idempotent_and_retries_runner_cancel() { .unwrap(); assert_eq!(out.task.state, BackgroundTaskState::Cancelling); - assert_eq!(runner.cancels(), vec![tid(10)]); + assert!(runner.cancels().is_empty()); assert!(store.saved_states().is_empty()); } diff --git a/crates/application/tests/error_codes.rs b/crates/application/tests/error_codes.rs index 32a1dcbd..aa49c828 100644 --- a/crates/application/tests/error_codes.rs +++ b/crates/application/tests/error_codes.rs @@ -16,6 +16,13 @@ fn codes_are_stable() { assert_eq!(AppError::Process("x".into()).code(), "PROCESS"); assert_eq!(AppError::Git("x".into()).code(), "GIT"); assert_eq!(AppError::Remote("x".into()).code(), "REMOTE"); + assert_eq!( + AppError::AgentStopped { + agent_id: domain::AgentId::from_uuid(uuid::Uuid::nil()), + } + .code(), + "AGENT_STOPPED" + ); assert_eq!(AppError::Internal("x".into()).code(), "INTERNAL"); } diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index bddab91f..ce310bc7 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -1065,15 +1065,22 @@ async fn cross_project_cancel_routes_by_agent_id_when_names_collide() { assert!(cancelled.detail.contains("corr-homonym-cancel")); assert_eq!(*first_cancelled.lock().unwrap(), 0); - assert_eq!(*second_cancelled.lock().unwrap(), 1); + assert_eq!(*second_cancelled.lock().unwrap(), 0); + assert!(fx + .structured + .session_for_agent_in_project(target_project.id, &first.id) + .is_some()); + assert!(fx + .structured + .session_for_agent_in_project(target_project.id, &second.id) + .is_none()); - fx.mailbox - .completions() - .push(second.id, TestCompletion::Cancelled); - let _ = timeout(TEST_GUARD, ask) + let error = timeout(TEST_GUARD, ask) .await .expect("cancelled ask must settle") - .expect("ask join"); + .expect("ask join") + .expect_err("stopped target cannot resolve the rendezvous"); + assert_eq!(error.code(), "AGENT_STOPPED"); } #[tokio::test] @@ -1389,7 +1396,7 @@ async fn stop_agent_kills_the_right_session() { } #[tokio::test] -async fn stop_agent_cancels_structured_turn_without_closing_session() { +async fn stop_agent_closes_structured_runtime_and_is_idempotent() { let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md"); let contexts = FakeContexts::with_agent(&agent, "# persona"); let fx = ask_fixture(contexts); @@ -1421,18 +1428,29 @@ async fn stop_agent_cancels_structured_turn_without_closing_session() { fx.pty.kills().is_empty(), "structured stop must not kill PTY" ); - assert_eq!(*cancel_called.lock().unwrap(), 1); + assert_eq!(*cancel_called.lock().unwrap(), 0); assert_eq!( *shutdown_called.lock().unwrap(), - 0, - "structured stop must not shutdown the session" + 1, + "structured stop must shutdown the runtime" ); assert!( fx.structured .session_for_agent_in_project(project_id(), &agent.id) - .is_some(), - "structured session is preserved for later turns" + .is_none(), + "structured session is removed at the terminal stop boundary" ); + + let second = fx + .service + .dispatch( + &project(), + cmd(r#"{ "action":"stop_agent", "name":"dev-backend" }"#), + ) + .await + .expect("repeated stop is an idempotent success"); + assert!(second.detail.contains("already stopped")); + assert_eq!(*shutdown_called.lock().unwrap(), 1); } #[tokio::test] @@ -1545,19 +1563,61 @@ async fn stop_agent_cascades_background_task_subtree() { } #[tokio::test] -async fn stop_agent_without_live_session_is_not_found() { +async fn stop_agent_without_live_session_is_idempotent_success() { let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md"); let fx = fixture(FakeContexts::with_agent(&agent, "# persona")); - let err = fx + let outcome = fx .service .dispatch( &project(), cmd(r#"{ "action":"stop_agent", "name":"dev-backend" }"#), ) .await - .unwrap_err(); - assert_eq!(err.code(), "NOT_FOUND", "got {err:?}"); + .expect("absent runtime is already stopped"); + assert!(outcome.detail.contains("already stopped")); +} + +#[tokio::test] +async fn stop_cascades_to_agent_launched_by_owner_and_closes_owner_admission() { + let owner = scratch_agent(aid(1), "main", "agents/main.md"); + let child = scratch_agent(aid(2), "dev-backend", "agents/dev-backend.md"); + let contexts = FakeContexts::with_agent(&owner, "# owner"); + contexts.add_agent(&child, "# child"); + let fx = fixture(contexts); + + fx.service + .dispatch( + &project(), + cmd(&format!( + r#"{{"action":"agent.run","targetAgent":"dev-backend","requestedBy":"{}","visibility":"background"}}"#, + owner.id + )), + ) + .await + .expect("owner launches child"); + assert!(fx.sessions.session(&sid(777)).is_some()); + + fx.service + .dispatch(&project(), cmd(r#"{"action":"stop_agent","name":"main"}"#)) + .await + .expect("owner stop cascades to launched child"); + + assert_eq!(fx.pty.kills(), vec![sid(777)]); + assert!(fx.sessions.session(&sid(777)).is_none()); + + let error = fx + .service + .dispatch( + &project(), + cmd(&format!( + r#"{{"action":"agent.run","targetAgent":"dev-backend","requestedBy":"{}","visibility":"background"}}"#, + owner.id + )), + ) + .await + .expect_err("stopped owner cannot launch more work"); + assert_eq!(error.code(), "AGENT_STOPPED"); } #[tokio::test] @@ -5664,7 +5724,7 @@ async fn cross_project_cancel_during_restart_releases_ask_and_correlation() { .await .expect("ask task joins") .expect_err("cancelled restart-aware ask must fail"); - assert!(err.to_string().to_ascii_lowercase().contains("cancel")); + assert_eq!(err.code(), "AGENT_STOPPED"); assert_eq!(fx.mailbox.pending(&aid(1)), 0); } @@ -5687,15 +5747,15 @@ async fn same_project_stop_during_restart_releases_ask_and_leaves_no_replacement .await .expect("same-project stop cancels the restart-aware logical turn"); - assert_eq!( - stopped.detail, - "cancelled restarting work for agent architect" + assert!( + stopped.detail.contains("stopped agent architect") + || stopped.detail.contains("already stopped") ); let err = ask .await .expect("ask task joins") .expect_err("cancelled restart-aware ask must fail"); - assert!(err.to_string().to_ascii_lowercase().contains("cancel")); + assert_eq!(err.code(), "AGENT_STOPPED"); assert_eq!( fx.mediator.busy_state(rkey(1)), AgentBusyState::Idle, @@ -5705,7 +5765,7 @@ async fn same_project_stop_during_restart_releases_ask_and_leaves_no_replacement await_until(|| { fx.live .entry_for(target.id) - .is_some_and(|entry| entry.status == WorkStatus::Done) + .is_some_and(|entry| entry.status == WorkStatus::Idle) }) .await; assert!( @@ -5716,6 +5776,50 @@ async fn same_project_stop_during_restart_releases_ask_and_leaves_no_replacement ); } +#[tokio::test] +async fn stop_cancels_normal_rendezvous_and_suppresses_late_final_event() { + let target = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = ask_fixture(FakeContexts::with_agent(&target, "# persona")); + fx.structured.remove(&sid(8_000)); + fx.structured.insert_in_project( + project_id(), + Arc::new(TappedDelayedFinalSession { id: sid(9_777) }), + target.id, + nid(700), + ); + + let service = Arc::clone(&fx.service); + let ask = tokio::spawn(async move { service.dispatch(&project(), cmd(ASK_JSON)).await }); + await_until(|| { + fx.live + .entry_for(target.id) + .is_some_and(|entry| entry.status == WorkStatus::Working) + }) + .await; + + fx.service + .dispatch( + &project(), + cmd(r#"{"action":"stop_agent","name":"architect"}"#), + ) + .await + .expect("stop cancels an ordinary active rendezvous"); + + let error = ask + .await + .expect("ask task joins") + .expect_err("stopped generation cannot return its delayed Final"); + assert_eq!(error.code(), "AGENT_STOPPED"); + assert!(!fx.bus.events().iter().any(|event| matches!( + event, + DomainEvent::AgentReplied { agent_id, .. } if *agent_id == target.id + ))); + assert!(fx + .live + .entry_for(target.id) + .is_some_and(|entry| { entry.status == WorkStatus::Idle && entry.intent == "Arrêté" })); +} + #[tokio::test] async fn direct_human_turn_registration_holds_restart_until_replacement_handoff() { let coordinator = Arc::new(application::RestartRendezvousCoordinator::new()); diff --git a/crates/application/tests/workstate_actions.rs b/crates/application/tests/workstate_actions.rs index cda9cebd..9a9c7034 100644 --- a/crates/application/tests/workstate_actions.rs +++ b/crates/application/tests/workstate_actions.rs @@ -130,6 +130,7 @@ impl AgentSession for FakeSession { #[derive(Default)] struct FakeCascade { calls: Mutex>, + fail: bool, } #[derive(Default)] @@ -163,8 +164,19 @@ impl AgentSelfRestartScheduler for FakeRestartScheduler { #[async_trait] impl StopAgentExecutionCascade for FakeCascade { - 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<(), application::AppError> { self.calls.lock().unwrap().push((project.id, agent_id)); + if self.fail { + Err(application::AppError::Process( + "child cancellation failed".to_owned(), + )) + } else { + Ok(()) + } } } @@ -534,7 +546,7 @@ async fn stop_structured_shuts_down_and_removes_session() { } #[tokio::test] -async fn stop_structured_runs_execution_cascade_best_effort() { +async fn stop_structured_runs_execution_cascade_before_shutdown() { let f = fixture(); let a = aid(10); let (shutdown_called, cancel_called) = insert_structured(&f.structured, sid(2), a, nid(100)); @@ -565,6 +577,33 @@ async fn stop_structured_runs_execution_cascade_best_effort() { assert!(f.structured.is_empty(), "live session is removed"); } +#[tokio::test] +async fn cascade_failure_is_reported_after_primary_runtime_is_still_stopped() { + let f = fixture(); + let a = aid(10); + let (shutdown_called, _) = insert_structured(&f.structured, sid(2), a, nid(100)); + let cascade = Arc::new(FakeCascade { + fail: true, + ..Default::default() + }); + let stop = f + .stop + .with_execution_cascade(cascade as Arc); + + let error = stop + .execute(StopLiveAgentInput { + project: project(), + agent_id: a, + reason: None, + }) + .await + .expect_err("partial cascade must be surfaced"); + + assert_eq!(error.code(), "PROCESS"); + assert!(shutdown_called.load(Ordering::SeqCst)); + assert!(f.structured.is_empty()); +} + #[tokio::test] async fn stop_removes_every_runtime_for_the_agent_across_registries() { let f = fixture(); diff --git a/crates/domain/src/orchestrator.rs b/crates/domain/src/orchestrator.rs index 3912ef78..82ea31b3 100644 --- a/crates/domain/src/orchestrator.rs +++ b/crates/domain/src/orchestrator.rs @@ -302,6 +302,8 @@ pub enum OrchestratorCommand { context: Option, /// Desired visibility for the launched session. visibility: OrchestratorVisibility, + /// Authenticated caller when the launch originates from an agent tool. + requester: Option, }, /// Stop/cancel a running agent. StopAgent { @@ -587,6 +589,7 @@ impl OrchestratorRequest { profile: Some(self.require("profile", action, self.profile.as_deref())?), context: self.context.as_ref().filter(|c| !c.is_empty()).cloned(), visibility: self.parse_visibility(action)?, + requester: self.optional_requester(action)?, }), "agent.run" => Ok(OrchestratorCommand::SpawnAgent { name: self.require_target_agent(action)?, @@ -602,6 +605,7 @@ impl OrchestratorRequest { .filter(|c| !c.is_empty()) .cloned(), visibility: self.parse_visibility(action)?, + requester: self.optional_requester(action)?, }), "stop_agent" => Ok(OrchestratorCommand::StopAgent { name: self.require_name(action)?, @@ -981,6 +985,7 @@ mod tests { profile: Some("claude-code".to_owned()), context: Some("agents/dev-backend.md".to_owned()), visibility: OrchestratorVisibility::Background, + requester: None, } ); } @@ -995,6 +1000,7 @@ mod tests { profile: Some("claude-code".to_owned()), context: None, visibility: OrchestratorVisibility::Background, + requester: None, } ); } @@ -1014,6 +1020,7 @@ mod tests { visibility: OrchestratorVisibility::Visible { node_id: NodeId::from_uuid(node) }, + requester: None, } ); @@ -1053,6 +1060,7 @@ mod tests { profile: None, context: Some("Analyse".to_owned()), visibility: OrchestratorVisibility::Background, + requester: None, } ); } @@ -1072,6 +1080,7 @@ mod tests { visibility: OrchestratorVisibility::Visible { node_id: NodeId::from_uuid(node) }, + requester: None, } ); } diff --git a/crates/infrastructure/src/orchestrator/mcp/server.rs b/crates/infrastructure/src/orchestrator/mcp/server.rs index 9c043612..8a06a180 100644 --- a/crates/infrastructure/src/orchestrator/mcp/server.rs +++ b/crates/infrastructure/src/orchestrator/mcp/server.rs @@ -472,6 +472,21 @@ impl McpServer { self.enforce_durable_tool_policy(&policy, &name)?; } + // Stop closes the entire MCP surface synchronously, including provider + // tools that do not map through OrchestratorCommand. The stop tool itself + // stays admitted so repeated/self stops remain idempotent. + if name != "idea_stop_agent" { + if let Ok(raw) = uuid::Uuid::parse_str(&self.requester) { + if let Err(error) = self + .service + .ensure_agent_execution_active(&self.project, domain::AgentId::from_uuid(raw)) + { + self.publish_processed(&name, false); + return Ok(tool_result_text(&error.to_string(), true)); + } + } + } + // Diagnostics begin beacon (best-effort, jamais le corps task/result) : trace // l'entrée d'un `tools/call`, son tool, le peer demandeur et la cible/longueurs. let started = Instant::now(); diff --git a/crates/infrastructure/src/orchestrator/mcp/tools.rs b/crates/infrastructure/src/orchestrator/mcp/tools.rs index 0c2ad0e0..02690f42 100644 --- a/crates/infrastructure/src/orchestrator/mcp/tools.rs +++ b/crates/infrastructure/src/orchestrator/mcp/tools.rs @@ -620,6 +620,7 @@ pub fn map_tool_call( }, "idea_launch_agent" => OrchestratorRequest { request_type: Some("agent.run".to_owned()), + requested_by: Some(requester.to_owned()), target_agent: s("target"), profile: s("profile"), task: s("task"), @@ -1210,10 +1211,22 @@ mod tests { profile: None, context: None, visibility: OrchestratorVisibility::Background, + requester: None, } ); } + #[test] + fn launch_agent_maps_handshake_requester_for_stop_ownership() { + assert!(matches!( + map_tool_call("idea_launch_agent", &json!({ "target": "Dev" }), REQ).unwrap(), + OrchestratorCommand::SpawnAgent { + requester: Some(agent), + .. + } if agent == domain::AgentId::from_uuid(uuid::Uuid::parse_str(REQ).unwrap()) + )); + } + #[test] fn stop_update_and_skill_map_to_their_commands() { assert_eq!( diff --git a/crates/web-server/src/lib.rs b/crates/web-server/src/lib.rs index bcb349d7..92d95258 100644 --- a/crates/web-server/src/lib.rs +++ b/crates/web-server/src/lib.rs @@ -1653,7 +1653,7 @@ async fn execute_launch_agent_for_ws( .app .launch_agent .execute(LaunchAgentInput { - project, + project: project.clone(), agent_id, rows: request.rows, cols: request.cols, @@ -1671,6 +1671,10 @@ async fn execute_launch_agent_for_ws( }) .await .map_err(ErrorDto::from)?; + state + .app + .orchestrator_service + .reactivate_agent_execution(&project, agent_id); if let Ok(mut contexts) = state.app.resume_contexts.lock() { contexts.insert( diff --git a/frontend/src/features/agents/AgentsPanel.tsx b/frontend/src/features/agents/AgentsPanel.tsx index eec05702..8a20ef99 100644 --- a/frontend/src/features/agents/AgentsPanel.tsx +++ b/frontend/src/features/agents/AgentsPanel.tsx @@ -85,6 +85,9 @@ export function AgentsPanel({ projectId, projectRoot = "" }: AgentsPanelProps) { * set by the hook once launchAgent resolves. */ const [activeAgentId, setActiveAgentId] = useState(null); + const [stopStateByAgent, setStopStateByAgent] = useState< + Record + >({}); const [panelNodeIds, setPanelNodeIds] = useState>({}); /** @@ -228,9 +231,19 @@ export function AgentsPanel({ projectId, projectRoot = "" }: AgentsPanelProps) { setActiveAgentId(agentId); } - function handleStop(agentId?: string) { - void vm.stopAgent(agentId); - setActiveAgentId(null); + async function handleStop(agentId: string) { + setStopStateByAgent((current) => ({ ...current, [agentId]: "stopping" })); + const stopped = await vm.stopAgent(agentId); + if (stopped) { + setActiveAgentId((current) => (current === agentId ? null : current)); + setStopStateByAgent((current) => { + const next = { ...current }; + delete next[agentId]; + return next; + }); + } else { + setStopStateByAgent((current) => ({ ...current, [agentId]: "incomplete" })); + } } // Determine if a template is chosen → profile selector is hidden (template imposes it). @@ -386,6 +399,7 @@ export function AgentsPanel({ projectId, projectRoot = "" }: AgentsPanelProps) { {vm.agents.map((a) => { const isSelected = a.id === vm.selectedAgentId; const isRunning = a.id === activeAgentId; + const stopState = stopStateByAgent[a.id]; const live = vm.liveAgents.find((candidate) => candidate.agentId === a.id); const agentProfile = vm.profiles.find((p) => p.id === a.profileId) ?? null; const profileName = @@ -561,11 +575,22 @@ export function AgentsPanel({ projectId, projectRoot = "" }: AgentsPanelProps) { ) : ( )} diff --git a/frontend/src/features/workstate/workstate.test.tsx b/frontend/src/features/workstate/workstate.test.tsx index 1e0bcd46..ea11f486 100644 --- a/frontend/src/features/workstate/workstate.test.tsx +++ b/frontend/src/features/workstate/workstate.test.tsx @@ -17,6 +17,16 @@ import { const PROJECT_ID = "project-work-state-test"; +function deferred() { + let resolve!: (value: T) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((done, fail) => { + resolve = done; + reject = fail; + }); + return { promise, resolve, reject }; +} + function renderPanel( workState: MockWorkStateGateway = new MockWorkStateGateway(), system: MockSystemGateway = new MockSystemGateway(), @@ -107,6 +117,43 @@ describe("ProjectWorkStatePanel", () => { expect(screen.getByText("Ticket 4c65d981")).toBeTruthy(); }); + it("projects stopping and stopped backend states for parent and descendant without orange", async () => { + const workState = new MockWorkStateGateway(); + workState._setProjectWorkState(PROJECT_ID, { + agents: [ + { + agentId: "agent-parent", + name: "Main", + profileId: "codex", + runtimeState: "stopping", + live: { nodeId: "node-main", sessionId: "session-main", kind: "pty" }, + busy: { state: "busy", ticket: "stale-parent", sinceMs: 1 }, + tickets: [], + }, + { + agentId: "agent-child", + name: "Child", + profileId: "codex", + runtimeState: "stopped", + busy: { state: "busy", ticket: "stale-child", sinceMs: 1 }, + tickets: [], + }, + ], + conversations: [], + }); + + renderPanel(workState); + + const mainRow = (await screen.findByText("Main")).closest("li")!; + const childRow = screen.getByText("Child").closest("li")!; + const stopping = within(mainRow).getByText("Arrêt en cours…"); + const stopped = within(childRow).getByText("Arrêté"); + expect(stopping.className).not.toContain("text-warning"); + expect(stopped.className).not.toContain("text-warning"); + expect(within(mainRow).queryByText("Busy")).toBeNull(); + expect(within(childRow).queryByText("Busy")).toBeNull(); + }); + it("renders an in-progress human ticket", async () => { const workState = new MockWorkStateGateway(); workState._setProjectWorkState(PROJECT_ID, { @@ -992,14 +1039,18 @@ describe("ProjectWorkStatePanel", () => { ], conversations: [], }); - const stopSpy = vi.spyOn(agent, "stopLiveAgent"); + const pendingStop = deferred>>(); + const stopSpy = vi + .spyOn(agent, "stopLiveAgent") + .mockReturnValue(pendingStop.promise); const refreshSpy = vi.spyOn(workState, "getProjectWorkState"); const confirmSpy = vi.spyOn(window, "confirm").mockReturnValue(true); renderPanel(workState, new MockSystemGateway(), { layout, agent }); await openAgentDetails("Runner"); - fireEvent.click(await screen.findByRole("button", { name: "Stop" })); + const row = await openAgentDetails("Runner"); + fireEvent.click(within(row).getByRole("button", { name: "Stop" })); await waitFor(() => expect(confirmSpy).toHaveBeenCalledWith( @@ -1013,7 +1064,39 @@ describe("ProjectWorkStatePanel", () => { "Arrêt demandé depuis Travail", ), ); + const stopping = within(row).getByRole("button", { name: "Arrêt en cours…" }); + expect((stopping as HTMLButtonElement).disabled).toBe(true); + expect(within(row).getAllByText("Arrêt en cours…")).toHaveLength(2); + expect(within(row).queryByText("Busy")).toBeNull(); + + workState._setProjectWorkState(PROJECT_ID, { + agents: [ + { + agentId: created.id, + name: "Runner", + profileId: "codex", + runtimeState: "stopped", + busy: { state: "idle" }, + tickets: [], + }, + ], + conversations: [], + }); + pendingStop.resolve({ + agentId: created.id, + alreadyStopped: false, + stopped: [ + { + sessionId: "mock-agent-session-1", + nodeId: leafId, + kind: "pty", + runtimeState: "stopped", + }, + ], + }); await waitFor(() => expect(refreshSpy).toHaveBeenCalled()); + await waitFor(() => expect(within(row).getAllByText("Arrêté")).toHaveLength(1)); + expect(within(row).queryByText("Busy")).toBeNull(); await waitFor(async () => { const leaf = leaves(await layout.loadLayout(PROJECT_ID))[0]; expect(leaf.session).toBeNull(); @@ -1021,6 +1104,52 @@ describe("ProjectWorkStatePanel", () => { confirmSpy.mockRestore(); }); + it("surfaces an incomplete stop and retries without hiding the backend state", async () => { + const workState = new MockWorkStateGateway(); + const agent = new MockAgentGateway(); + const pendingRetry = deferred>>(); + workState._setProjectWorkState(PROJECT_ID, { + agents: [ + { + agentId: "agent-partial", + name: "Main", + profileId: "codex", + live: { nodeId: "node-main", sessionId: "session-main", kind: "pty" }, + busy: { state: "busy", ticket: "ticket-main", sinceMs: 1 }, + tickets: [], + }, + ], + conversations: [], + }); + const stopSpy = vi + .spyOn(agent, "stopLiveAgent") + .mockRejectedValueOnce(new Error("cascade failed")) + .mockReturnValueOnce(pendingRetry.promise); + const confirmSpy = vi.spyOn(window, "confirm").mockReturnValue(true); + + renderPanel(workState, new MockSystemGateway(), { agent }); + + const row = await openAgentDetails("Main"); + fireEvent.click(within(row).getByRole("button", { name: "Stop" })); + + expect(await within(row).findByText("Arrêt incomplet")).toBeTruthy(); + expect(within(row).getByRole("alert").textContent).toBe( + "Certains travaux sont encore actifs.", + ); + fireEvent.click(within(row).getByRole("button", { name: "Réessayer l’arrêt" })); + + await waitFor(() => expect(stopSpy).toHaveBeenCalledTimes(2)); + expect(confirmSpy).toHaveBeenCalledTimes(1); + expect( + (within(row).getByRole("button", { name: "Arrêt en cours…" }) as HTMLButtonElement) + .disabled, + ).toBe(true); + + pendingRetry.resolve({ agentId: "agent-partial", alreadyStopped: true, stopped: [] }); + await waitFor(() => expect(within(row).getAllByText("Arrêté")).toHaveLength(2)); + confirmSpy.mockRestore(); + }); + it("View conversation exposes only Lot C preview fields", async () => { const workState = new MockWorkStateGateway(); workState._setProjectWorkState(PROJECT_ID, {