merge(application,domain,infrastructure,web-server,app-tauri,frontend): fix #320 — arrêt d'agent fiable avec cascade sur travaux/agents lancés et statut cohérent (QA verte)
This commit is contained in:
@ -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(
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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());
|
||||
}
|
||||
|
||||
|
||||
@ -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");
|
||||
}
|
||||
|
||||
|
||||
@ -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());
|
||||
|
||||
@ -130,6 +130,7 @@ impl AgentSession for FakeSession {
|
||||
#[derive(Default)]
|
||||
struct FakeCascade {
|
||||
calls: Mutex<Vec<(ProjectId, AgentId)>>,
|
||||
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<dyn StopAgentExecutionCascade>);
|
||||
|
||||
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();
|
||||
|
||||
@ -302,6 +302,8 @@ pub enum OrchestratorCommand {
|
||||
context: Option<String>,
|
||||
/// Desired visibility for the launched session.
|
||||
visibility: OrchestratorVisibility,
|
||||
/// Authenticated caller when the launch originates from an agent tool.
|
||||
requester: Option<AgentId>,
|
||||
},
|
||||
/// 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,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
@ -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();
|
||||
|
||||
@ -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!(
|
||||
|
||||
@ -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(
|
||||
|
||||
@ -85,6 +85,9 @@ export function AgentsPanel({ projectId, projectRoot = "" }: AgentsPanelProps) {
|
||||
* set by the hook once launchAgent resolves.
|
||||
*/
|
||||
const [activeAgentId, setActiveAgentId] = useState<string | null>(null);
|
||||
const [stopStateByAgent, setStopStateByAgent] = useState<
|
||||
Record<string, "stopping" | "incomplete">
|
||||
>({});
|
||||
const [panelNodeIds, setPanelNodeIds] = useState<Record<string, string>>({});
|
||||
|
||||
/**
|
||||
@ -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) {
|
||||
<Button
|
||||
size="sm"
|
||||
variant="ghost"
|
||||
aria-label={`stop ${a.name}`}
|
||||
onClick={() => handleStop(a.id)}
|
||||
aria-label={
|
||||
stopState === "stopping"
|
||||
? `Arrêt en cours… — ${a.name}`
|
||||
: stopState === "incomplete"
|
||||
? `Réessayer l’arrêt — ${a.name}`
|
||||
: `stop ${a.name}`
|
||||
}
|
||||
disabled={stopState === "stopping"}
|
||||
onClick={() => void handleStop(a.id)}
|
||||
className="text-danger hover:text-danger"
|
||||
>
|
||||
Stop
|
||||
{stopState === "stopping"
|
||||
? "Arrêt en cours…"
|
||||
: stopState === "incomplete"
|
||||
? "Réessayer l’arrêt"
|
||||
: "Stop"}
|
||||
</Button>
|
||||
) : (
|
||||
<Button
|
||||
|
||||
@ -36,6 +36,14 @@ import { AgentsPanel } from "./AgentsPanel";
|
||||
|
||||
const PROJECT_ID = "proj-test-001";
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>((done) => {
|
||||
resolve = done;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
/**
|
||||
* Renders `AgentsPanel` behind a `DIProvider` with an isolated
|
||||
* `MockAgentGateway` (and optionally a seeded `MockProfileGateway`).
|
||||
@ -451,7 +459,8 @@ describe("AgentsPanel (with MockAgentGateway)", () => {
|
||||
it("stopping an agent unmounts the terminal view", async () => {
|
||||
const agent = new MockAgentGateway();
|
||||
const created = await agent.createAgent(PROJECT_ID, { name: "Echo", profileId: "p1" });
|
||||
const stop = vi.spyOn(agent, "stopLiveAgent");
|
||||
const pendingStop = deferred<Awaited<ReturnType<typeof agent.stopLiveAgent>>>();
|
||||
const stop = vi.spyOn(agent, "stopLiveAgent").mockReturnValue(pendingStop.promise);
|
||||
|
||||
renderPanel(agent);
|
||||
await waitForIdle();
|
||||
@ -464,6 +473,11 @@ describe("AgentsPanel (with MockAgentGateway)", () => {
|
||||
|
||||
// Stop
|
||||
fireEvent.click(screen.getByRole("button", { name: "stop Echo" }));
|
||||
const stopping = screen.getByRole("button", { name: "Arrêt en cours… — Echo" });
|
||||
expect((stopping as HTMLButtonElement).disabled).toBe(true);
|
||||
expect(screen.getByTestId("terminal-view")).toBeTruthy();
|
||||
|
||||
pendingStop.resolve({ agentId: created.id, alreadyStopped: true, stopped: [] });
|
||||
await waitFor(() => {
|
||||
expect(screen.queryByTestId("terminal-view")).toBeNull();
|
||||
});
|
||||
@ -474,6 +488,30 @@ describe("AgentsPanel (with MockAgentGateway)", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("keeps the agent visible after an incomplete stop and exposes retry", async () => {
|
||||
const agent = new MockAgentGateway();
|
||||
const created = await agent.createAgent(PROJECT_ID, { name: "Main", profileId: "p1" });
|
||||
const stop = vi
|
||||
.spyOn(agent, "stopLiveAgent")
|
||||
.mockRejectedValueOnce(new Error("cascade failed"))
|
||||
.mockResolvedValueOnce({ agentId: created.id, alreadyStopped: true, stopped: [] });
|
||||
|
||||
renderPanel(agent);
|
||||
await waitForIdle();
|
||||
fireEvent.click(screen.getByRole("button", { name: "launch Main" }));
|
||||
await screen.findByTestId("terminal-view");
|
||||
fireEvent.click(screen.getByRole("button", { name: "stop Main" }));
|
||||
|
||||
expect(
|
||||
await screen.findByText("Arrêt incomplet — Certains travaux sont encore actifs."),
|
||||
).toBeTruthy();
|
||||
expect(screen.getByTestId("terminal-view")).toBeTruthy();
|
||||
fireEvent.click(screen.getByRole("button", { name: "Réessayer l’arrêt — Main" }));
|
||||
|
||||
await waitFor(() => expect(stop).toHaveBeenCalledTimes(2));
|
||||
await waitFor(() => expect(screen.queryByTestId("terminal-view")).toBeNull());
|
||||
});
|
||||
|
||||
it("stops a live agent even when its TerminalView was never mounted", async () => {
|
||||
const agent = new MockAgentGateway();
|
||||
const created = await agent.createAgent(PROJECT_ID, { name: "Headless", profileId: "p1" });
|
||||
|
||||
@ -145,7 +145,7 @@ export interface AgentsViewModel {
|
||||
onData: (bytes: Uint8Array) => void,
|
||||
) => Promise<TerminalHandle>;
|
||||
/** Stops an agent PTY explicitly by closing its live session id. */
|
||||
stopAgent: (agentId?: string) => Promise<void>;
|
||||
stopAgent: (agentId?: string) => Promise<boolean>;
|
||||
/**
|
||||
* Cancels the auto-resume armed for a rate-limited agent (ARCHITECTURE §21):
|
||||
* the "Annuler la reprise" action. Optimistically drops the countdown; the
|
||||
@ -513,20 +513,23 @@ export function useAgents(projectId: string): AgentsViewModel {
|
||||
);
|
||||
|
||||
const stopAgent = useCallback(
|
||||
async (agentId?: string) => {
|
||||
async (agentId?: string): Promise<boolean> => {
|
||||
const targetAgentId = agentId ?? runningAgentId;
|
||||
if (!targetAgentId) return;
|
||||
setRunningAgentId((current) =>
|
||||
current === targetAgentId || !targetAgentId ? null : current,
|
||||
);
|
||||
if (!targetAgentId) return false;
|
||||
setError(null);
|
||||
try {
|
||||
await agent.stopLiveAgent(projectId, targetAgentId, "Arrêt demandé depuis Agents");
|
||||
setRunningAgentId((current) =>
|
||||
current === targetAgentId ? null : current,
|
||||
);
|
||||
setLiveAgents((current) =>
|
||||
current.filter((candidate) => candidate.agentId !== targetAgentId),
|
||||
);
|
||||
} catch (e) {
|
||||
setError(describe(e));
|
||||
return true;
|
||||
} catch {
|
||||
setError("Arrêt incomplet — Certains travaux sont encore actifs.");
|
||||
await refreshLiveAgents();
|
||||
return false;
|
||||
}
|
||||
},
|
||||
[agent, projectId, refreshLiveAgents, runningAgentId],
|
||||
|
||||
@ -259,6 +259,60 @@ describe("LayoutGrid (with MockLayoutGateway)", () => {
|
||||
expect(screen.queryByText("Idle")).toBeNull();
|
||||
});
|
||||
|
||||
it("removes the orange busy veil when backend live-state reports stopping", async () => {
|
||||
const layout = new MockLayoutGateway();
|
||||
const agent = new MockAgentGateway();
|
||||
const workState = new MockWorkStateGateway();
|
||||
const created = await agent.createAgent("p1", {
|
||||
name: "StoppingAgent",
|
||||
profileId: "prof-1",
|
||||
});
|
||||
const tree = await layout.loadLayout("p1");
|
||||
const leafId = leaves(tree)[0].id;
|
||||
await layout.mutateLayout("p1", {
|
||||
type: "setCellAgent",
|
||||
target: leafId,
|
||||
agent: created.id,
|
||||
});
|
||||
await layout.mutateLayout("p1", {
|
||||
type: "setSession",
|
||||
target: leafId,
|
||||
session: "stopping-session",
|
||||
});
|
||||
workState._setProjectWorkState("p1", {
|
||||
agents: [
|
||||
{
|
||||
agentId: created.id,
|
||||
name: "StoppingAgent",
|
||||
profileId: "prof-1",
|
||||
runtimeState: "stopping",
|
||||
live: {
|
||||
nodeId: leafId,
|
||||
sessionId: "stopping-session",
|
||||
kind: "pty",
|
||||
runtimeState: "stopping",
|
||||
},
|
||||
busy: { state: "busy", ticket: "stale-ticket", sinceMs: 1 },
|
||||
tickets: [],
|
||||
},
|
||||
],
|
||||
conversations: [],
|
||||
});
|
||||
|
||||
render(
|
||||
<DIProvider gateways={{ layout, agent, workState } as unknown as Gateways}>
|
||||
<LayoutGrid projectId="p1" cwd="/home/me/proj" />
|
||||
</DIProvider>,
|
||||
);
|
||||
|
||||
await waitFor(() =>
|
||||
expect(screen.queryByTestId("agent-cell-runtime-loading")).toBeNull(),
|
||||
);
|
||||
expect(screen.queryByTestId("target-announcements-overlay")).toBeNull();
|
||||
expect(screen.queryByLabelText("StoppingAgent arrêté")).toBeNull();
|
||||
expect(screen.getByTestId("terminal-view")).toBeTruthy();
|
||||
});
|
||||
|
||||
it("relaunches a stopped agent from scratch in the same cell", async () => {
|
||||
const layout = new MockLayoutGateway();
|
||||
const agent = new MockAgentGateway();
|
||||
|
||||
@ -847,6 +847,9 @@ function LeafView({
|
||||
const agentWork = agentId
|
||||
? workState?.agents.find((row) => row.agentId === agentId)
|
||||
: undefined;
|
||||
const runtimeState = agentWork?.runtimeState;
|
||||
const stoppingOrStopped =
|
||||
runtimeState === "stopping" || runtimeState === "stopped";
|
||||
const activeTicket = agentWork?.tickets.find((ticket) => ticket.status === "inProgress");
|
||||
const queuedTickets = agentWork?.tickets.filter((ticket) => ticket.status === "queued") ?? [];
|
||||
const inboxDepth = Math.max(
|
||||
@ -858,9 +861,12 @@ function LeafView({
|
||||
agentWork?.backgroundTasks?.filter((task) =>
|
||||
task.status === "completed" || task.status === "failed" || task.status === "cancelled",
|
||||
) ?? [];
|
||||
const busyByWorkState = agentWork?.busy.state === "busy" || Boolean(activeTicket);
|
||||
const busyByWorkState =
|
||||
!stoppingOrStopped &&
|
||||
(agentWork?.busy.state === "busy" || Boolean(activeTicket));
|
||||
const delegatedTicket = activeTicket?.source.kind === "agent" ? activeTicket : undefined;
|
||||
const busyActive = announcementBusyActive || busyByWorkState;
|
||||
const busyActive =
|
||||
!stoppingOrStopped && (announcementBusyActive || busyByWorkState);
|
||||
const [cellRestart, setCellRestart] = useState<CellRestartState | null>(null);
|
||||
const cellRestartRef = useRef<CellRestartState | null>(null);
|
||||
const cellRestartRequestRef = useRef(0);
|
||||
@ -938,7 +944,6 @@ function LeafView({
|
||||
const [restartError, setRestartError] = useState(false);
|
||||
const [freshRestartedSession, setFreshRestartedSession] =
|
||||
useState<FreshRestartedSession | null>(null);
|
||||
const runtimeState = agentWork?.runtimeState;
|
||||
const livePinnedAgent = agentId
|
||||
? liveAgents.find((la) => la.agentId === agentId)
|
||||
: undefined;
|
||||
@ -1762,7 +1767,7 @@ function LeafView({
|
||||
an availability veil showing its live réflexions. Native TUI cells
|
||||
need the full-cell guard; custom CLI cells keep the canonical busy
|
||||
badge and composer lock only, matching their normal chat turn flow. */}
|
||||
{!cellRestart && !otherRestartActive && !modelServerOverlay && cellMode !== "custom" && agentId != null && (
|
||||
{!cellRestart && !otherRestartActive && !modelServerOverlay && !stoppingOrStopped && cellMode !== "custom" && agentId != null && (
|
||||
<TargetAnnouncementsOverlay
|
||||
projectId={projectId}
|
||||
agentId={agentId}
|
||||
|
||||
@ -14,7 +14,7 @@ import type {
|
||||
InboxItem,
|
||||
LeafCell,
|
||||
} from "@/domain";
|
||||
import type { BackgroundTaskAttachment } from "@/ports";
|
||||
import type { BackgroundTaskAttachment, StoppedLiveAgent } from "@/ports";
|
||||
import { useGateways } from "@/app/di";
|
||||
import { leaves } from "@/features/layout/layout";
|
||||
import { useLayout } from "@/features/layout/useLayout";
|
||||
@ -138,12 +138,38 @@ function canRetryTask(status: BackgroundCompletion["status"]): boolean {
|
||||
return status === "failed" || status === "cancelFailed" || status === "cancelled";
|
||||
}
|
||||
|
||||
function agentStatusLabel(live: boolean, busy: boolean): string {
|
||||
type AgentStopUiState = "idle" | "stopping" | "stopped" | "incomplete";
|
||||
|
||||
function agentStatusLabel(
|
||||
live: boolean,
|
||||
busy: boolean,
|
||||
runtimeState: AgentWorkState["runtimeState"],
|
||||
stopState: AgentStopUiState,
|
||||
): string {
|
||||
if (stopState === "stopping" || runtimeState === "stopping") {
|
||||
return "Arrêt en cours…";
|
||||
}
|
||||
if (stopState === "incomplete") return "Arrêt incomplet";
|
||||
if (stopState === "stopped" || runtimeState === "stopped") return "Arrêté";
|
||||
if (busy) return "Busy";
|
||||
return live ? "Idle" : "Offline";
|
||||
}
|
||||
|
||||
function agentStatusClass(live: boolean, busy: boolean): string {
|
||||
function agentStatusClass(
|
||||
live: boolean,
|
||||
busy: boolean,
|
||||
runtimeState: AgentWorkState["runtimeState"],
|
||||
stopState: AgentStopUiState,
|
||||
): string {
|
||||
if (stopState === "incomplete") return "bg-danger/10 text-danger";
|
||||
if (
|
||||
stopState === "stopping" ||
|
||||
stopState === "stopped" ||
|
||||
runtimeState === "stopping" ||
|
||||
runtimeState === "stopped"
|
||||
) {
|
||||
return "bg-raised text-muted";
|
||||
}
|
||||
if (busy) return "bg-warning/15 text-warning";
|
||||
if (live) return "bg-success/10 text-success";
|
||||
return "bg-raised text-muted";
|
||||
@ -600,6 +626,7 @@ function AgentRow({
|
||||
const { agent: agentGateway } = useGateways();
|
||||
const [message, setMessage] = useState<string | null>(null);
|
||||
const [actionBusy, setActionBusy] = useState(false);
|
||||
const [stopState, setStopState] = useState<AgentStopUiState>("idle");
|
||||
const [detailsOpen, setDetailsOpen] = useState(false);
|
||||
const [backgroundOpen, setBackgroundOpen] = useState(false);
|
||||
const live = agent.live !== undefined;
|
||||
@ -617,7 +644,15 @@ function AgentRow({
|
||||
const currentTurn =
|
||||
busy || tickets.some((ticket) => ticket.status === "inProgress");
|
||||
const activeTicket = tickets.find((ticket) => ticket.status === "inProgress");
|
||||
const statusLabel = agentStatusLabel(live, busy);
|
||||
const statusLabel = agentStatusLabel(live, busy, agent.runtimeState, stopState);
|
||||
const stopLabel =
|
||||
stopState === "stopping" || agent.runtimeState === "stopping"
|
||||
? "Arrêt en cours…"
|
||||
: stopState === "incomplete"
|
||||
? "Réessayer l’arrêt"
|
||||
: stopState === "stopped" || agent.runtimeState === "stopped"
|
||||
? "Arrêté"
|
||||
: "Stop";
|
||||
const taskSummary = backgroundTaskSummary(backgroundTasks);
|
||||
const activityLine = compactIntent(
|
||||
activeTicket?.taskPreview ??
|
||||
@ -661,20 +696,33 @@ function AgentRow({
|
||||
}
|
||||
}
|
||||
|
||||
async function stop(): Promise<void> {
|
||||
async function stop(confirm = true): Promise<void> {
|
||||
if (!agent.live || !agentGateway) return;
|
||||
const warning = currentTurn
|
||||
? " The current turn will be interrupted."
|
||||
: "";
|
||||
if (!window.confirm(`Stop ${agent.name}?${warning}`)) return;
|
||||
if (confirm && !window.confirm(`Stop ${agent.name}?${warning}`)) return;
|
||||
setActionBusy(true);
|
||||
setStopState("stopping");
|
||||
setMessage(null);
|
||||
let stopped: StoppedLiveAgent;
|
||||
try {
|
||||
const stopped = await agentGateway.stopLiveAgent(
|
||||
stopped = await agentGateway.stopLiveAgent(
|
||||
projectId,
|
||||
agent.agentId,
|
||||
"Arrêt demandé depuis Travail",
|
||||
);
|
||||
} catch {
|
||||
setStopState("incomplete");
|
||||
setMessage("Certains travaux sont encore actifs.");
|
||||
setActionBusy(false);
|
||||
return;
|
||||
}
|
||||
|
||||
// The stop response is the backend confirmation. Keep the row non-busy
|
||||
// while its authoritative live-state snapshot is being refreshed.
|
||||
setStopState("stopped");
|
||||
try {
|
||||
if (visibleNodeIds.has(agent.live.nodeId)) {
|
||||
const leaf = visibleLeaves.find((candidate) => candidate.id === agent.live?.nodeId);
|
||||
const stoppedSessionIds = new Set(stopped.stopped.map((runtime) => runtime.sessionId));
|
||||
@ -684,7 +732,11 @@ function AgentRow({
|
||||
}
|
||||
await refreshAfterAction();
|
||||
} catch (e) {
|
||||
setMessage(e instanceof Error ? e.message : String(e));
|
||||
setMessage(
|
||||
`Arrêt confirmé. Impossible d’actualiser l’affichage : ${
|
||||
e instanceof Error ? e.message : String(e)
|
||||
}`,
|
||||
);
|
||||
} finally {
|
||||
setActionBusy(false);
|
||||
}
|
||||
@ -715,7 +767,7 @@ function AgentRow({
|
||||
<span
|
||||
className={cn(
|
||||
"rounded-full px-2 py-0.5 text-xs font-medium",
|
||||
agentStatusClass(live, busy),
|
||||
agentStatusClass(live, busy, agent.runtimeState, stopState),
|
||||
)}
|
||||
>
|
||||
{statusLabel}
|
||||
@ -770,11 +822,18 @@ function AgentRow({
|
||||
<Button
|
||||
size="sm"
|
||||
variant="danger"
|
||||
loading={actionBusy}
|
||||
disabled={!agentGateway}
|
||||
onClick={() => void stop()}
|
||||
aria-label={stopLabel}
|
||||
loading={stopState === "stopping"}
|
||||
disabled={
|
||||
!agentGateway ||
|
||||
actionBusy ||
|
||||
stopState === "stopped" ||
|
||||
agent.runtimeState === "stopping" ||
|
||||
agent.runtimeState === "stopped"
|
||||
}
|
||||
onClick={() => void stop(stopState !== "incomplete")}
|
||||
>
|
||||
Stop
|
||||
{stopLabel}
|
||||
</Button>
|
||||
)}
|
||||
</div>
|
||||
|
||||
@ -17,6 +17,16 @@ import {
|
||||
|
||||
const PROJECT_ID = "project-work-state-test";
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
let reject!: (reason?: unknown) => void;
|
||||
const promise = new Promise<T>((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<Awaited<ReturnType<typeof agent.stopLiveAgent>>>();
|
||||
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<Awaited<ReturnType<typeof agent.stopLiveAgent>>>();
|
||||
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, {
|
||||
|
||||
Reference in New Issue
Block a user