fix(chat): le bouton Stop de la CLI custom arrête l'agent en cascade (#187)
Le Stop du CustomAgentChatView n'annulait que le tour courant sans libérer la session ni toucher aux tâches enfants ou appels MCP en cours. Il appelle désormais stop_live_agent sur l'agent vivant, qui: - ajoute le port StopAgentExecutionCascade (best-effort, idempotent), implémenté par OrchestratorService pour annuler les idea_run_in_background rattachés au rendez-vous actif (via BackgroundTaskRunner.cancel); - cascade l'arrêt aux wait-dépendances (stop_turn_watch + preempt) pour interrompre les appels MCP et les attentes inter-agents; - libère et supprime la session structurée. UI: nouvel état "Arrêt en cours..." (bouton disabled + loading) pendant l'arrêt, libération du sessionId, label expliquant la cascade demandée. La Tauri command cancel_agent_chat est alignée sur stop_live_agent + stop_turn_watch agent et dépendances. Tests: SpyBackgroundRunner (cascade n'annule que les tâches rattachées), workstate_actions cascade best-effort, CustomAgentChatView stop+cascade feedback et fallback NOT_FOUND.
This commit is contained in:
@ -249,7 +249,7 @@ pub use workstate::{
|
||||
ConversationTurnWorkPreview, ConversationWorkSummary, GetAppExitWorkGuardState,
|
||||
GetAppExitWorkGuardStateInput, GetLiveStateLean, GetProjectWorkState, GetProjectWorkStateInput,
|
||||
LeanLiveEntry, LeanLiveState, LiveWorkSession, ProjectWorkState, ReconcileLiveState,
|
||||
ReconcileLiveStateInput, StopLiveAgent, StopLiveAgentInput, StopLiveAgentOutput,
|
||||
TicketWorkSource, TicketWorkStatus, UpdateLiveState, UpdateLiveStateInput,
|
||||
ReconcileLiveStateInput, StopAgentExecutionCascade, StopLiveAgent, StopLiveAgentInput,
|
||||
StopLiveAgentOutput, TicketWorkSource, TicketWorkStatus, UpdateLiveState, UpdateLiveStateInput,
|
||||
LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
|
||||
};
|
||||
|
||||
@ -24,7 +24,9 @@ use domain::conversation::{ConversationParty, ConversationRegistry, SessionRef,
|
||||
use domain::conversation_log::{ConversationTurn, TurnId, TurnRole};
|
||||
use domain::input::{InputMediator, InputSource, SubmitConfig};
|
||||
use domain::mailbox::{Ticket, TicketId};
|
||||
use domain::ports::{BackgroundTaskStore, Clock, EventBus, ProfileStore, PtyHandle};
|
||||
use domain::ports::{
|
||||
BackgroundTaskRunner, BackgroundTaskStore, Clock, EventBus, ProfileStore, PtyHandle,
|
||||
};
|
||||
use domain::project::ProjectPath;
|
||||
use domain::{
|
||||
AgentId, AgentProfile, BackgroundTask, BackgroundTaskKind, BackgroundTaskRendezvousLink,
|
||||
@ -52,7 +54,9 @@ use crate::orchestrator::{
|
||||
};
|
||||
use crate::skill::{CreateSkill, CreateSkillInput, ReadSkill, ReadSkillInput};
|
||||
use crate::terminal::{CloseTerminal, CloseTerminalInput, StructuredSessions, TerminalSessions};
|
||||
use crate::workstate::{GetLiveStateLean, UpdateLiveState, UpdateLiveStateInput};
|
||||
use crate::workstate::{
|
||||
GetLiveStateLean, StopAgentExecutionCascade, UpdateLiveState, UpdateLiveStateInput,
|
||||
};
|
||||
|
||||
/// Default terminal geometry for an orchestrator-launched agent cell. The UI
|
||||
/// resizes the PTY to the real cell size on attach; these are sane starting rows
|
||||
@ -523,6 +527,9 @@ pub struct OrchestratorService {
|
||||
/// `idea_ask_agent` comme [`BackgroundTaskKind::HeadlessRendezvous`]. `None` ⇒
|
||||
/// comportement historique sans persistance de tâche (call sites non câblés).
|
||||
background_tasks: Option<Arc<dyn BackgroundTaskStore>>,
|
||||
/// Runner runtime des tâches command-backed, utilisé ici uniquement pour la cascade
|
||||
/// d'arrêt best-effort des `idea_run_in_background` attachés au rendez-vous actif.
|
||||
background_runner: Option<Arc<dyn BackgroundTaskRunner>>,
|
||||
/// Service applicatif des limites de session, injecté depuis la composition root.
|
||||
/// `None` conserve les call sites/tests legacy ; les chemins conscients des limites
|
||||
/// publient alors seulement leur erreur de tour historique.
|
||||
@ -610,6 +617,7 @@ impl OrchestratorService {
|
||||
ask_liveness_probe: None,
|
||||
ask_ceiling: ASK_AGENT_CEILING,
|
||||
background_tasks: None,
|
||||
background_runner: None,
|
||||
session_limits: None,
|
||||
spawn_background_command: None,
|
||||
}
|
||||
@ -646,6 +654,13 @@ impl OrchestratorService {
|
||||
self
|
||||
}
|
||||
|
||||
/// Branche le runner command-backed pour annuler les tâches enfants lors d'un stop.
|
||||
#[must_use]
|
||||
pub fn with_background_task_runner(mut self, runner: Arc<dyn BackgroundTaskRunner>) -> Self {
|
||||
self.background_runner = Some(runner);
|
||||
self
|
||||
}
|
||||
|
||||
/// Branche la **sonde de vivacité** (signe de vie) du rendez-vous délégué : la borne
|
||||
/// de tour devient une **fenêtre d'inactivité** réarmée à chaque progrès observé,
|
||||
/// sous le plafond [`Self::with_ask_ceiling`]. Builder additif (signature de
|
||||
@ -3101,6 +3116,40 @@ impl<'a> WaitEdgeGuard<'a> {
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl StopAgentExecutionCascade for OrchestratorService {
|
||||
async fn stop_agent_execution_children(&self, project: &Project, agent_id: AgentId) {
|
||||
let Some(runner) = &self.background_runner else {
|
||||
return;
|
||||
};
|
||||
let agent = runtime_key(project, agent_id);
|
||||
stop_active_rendezvous_background_tasks(&self.active_rendezvous, agent, runner.as_ref())
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
fn active_rendezvous_background_tasks(
|
||||
active: &StdMutex<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
|
||||
agent: RuntimeAgentKey,
|
||||
) -> Vec<TaskId> {
|
||||
active
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.get(&agent)
|
||||
.map(|active| active.background_tasks.clone())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
async fn stop_active_rendezvous_background_tasks(
|
||||
active: &StdMutex<HashMap<RuntimeAgentKey, ActiveRendezvous>>,
|
||||
agent: RuntimeAgentKey,
|
||||
runner: &dyn BackgroundTaskRunner,
|
||||
) {
|
||||
for task_id in active_rendezvous_background_tasks(active, agent) {
|
||||
let _ = runner.cancel(task_id).await;
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for WaitEdgeGuard<'_> {
|
||||
fn drop(&mut self) {
|
||||
let mut g = self
|
||||
@ -3375,6 +3424,77 @@ mod tests {
|
||||
RuntimeAgentKey::new(domain::ProjectId::from_uuid(uuid::Uuid::nil()), aid(n))
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct SpyBackgroundRunner {
|
||||
cancelled: TestMutex<Vec<TaskId>>,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl BackgroundTaskRunner for SpyBackgroundRunner {
|
||||
async fn spawn(
|
||||
&self,
|
||||
spec: domain::ports::BackgroundTaskSpec,
|
||||
) -> Result<domain::ports::BackgroundTaskHandle, domain::ports::BackgroundTaskPortError>
|
||||
{
|
||||
Ok(domain::ports::BackgroundTaskHandle {
|
||||
task_id: spec.task_id,
|
||||
})
|
||||
}
|
||||
|
||||
async fn cancel(
|
||||
&self,
|
||||
task_id: TaskId,
|
||||
) -> Result<(), domain::ports::BackgroundTaskPortError> {
|
||||
self.cancelled.lock().unwrap().push(task_id);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn subscribe_completions(&self) -> domain::ports::BackgroundCompletionStream {
|
||||
Box::new(std::iter::empty())
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stop_active_rendezvous_background_tasks_cancels_attached_tasks_only() {
|
||||
let active = StdMutex::new(HashMap::new());
|
||||
let target = rkey(2);
|
||||
let unrelated = rkey(9);
|
||||
let first = TaskId::from_uuid(uuid::Uuid::from_u128(101));
|
||||
let second = TaskId::from_uuid(uuid::Uuid::from_u128(102));
|
||||
let other = TaskId::from_uuid(uuid::Uuid::from_u128(999));
|
||||
active.lock().unwrap().insert(
|
||||
target,
|
||||
ActiveRendezvous {
|
||||
link: BackgroundTaskRendezvousLink {
|
||||
rendezvous_id: RendezvousId::from_uuid(uuid::Uuid::from_u128(201)),
|
||||
ticket_id: tid(202),
|
||||
requester_agent_id: Some(aid(1)),
|
||||
target_agent_id: target.agent_id,
|
||||
conversation_id: domain::ConversationId::from_uuid(uuid::Uuid::from_u128(203)),
|
||||
},
|
||||
background_tasks: vec![first, second],
|
||||
},
|
||||
);
|
||||
active.lock().unwrap().insert(
|
||||
unrelated,
|
||||
ActiveRendezvous {
|
||||
link: BackgroundTaskRendezvousLink {
|
||||
rendezvous_id: RendezvousId::from_uuid(uuid::Uuid::from_u128(301)),
|
||||
ticket_id: tid(302),
|
||||
requester_agent_id: Some(aid(1)),
|
||||
target_agent_id: unrelated.agent_id,
|
||||
conversation_id: domain::ConversationId::from_uuid(uuid::Uuid::from_u128(303)),
|
||||
},
|
||||
background_tasks: vec![other],
|
||||
},
|
||||
);
|
||||
let runner = SpyBackgroundRunner::default();
|
||||
|
||||
stop_active_rendezvous_background_tasks(&active, target, &runner).await;
|
||||
|
||||
assert_eq!(*runner.cancelled.lock().unwrap(), vec![first, second]);
|
||||
}
|
||||
|
||||
/// Drop d'un garde **armé** ⇒ `cancel_head` + `mark_idle` sur la cible (c'est le
|
||||
/// comportement qui débloque un agent resté `Busy` sur un futur abandonné).
|
||||
#[test]
|
||||
|
||||
@ -9,6 +9,7 @@
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use domain::input::InputMediator;
|
||||
use domain::{AgentId, NodeId, Project, RuntimeAgentKey, SessionId};
|
||||
|
||||
@ -116,6 +117,18 @@ pub struct StopLiveAgentOutput {
|
||||
pub kind: LiveSessionKind,
|
||||
}
|
||||
|
||||
/// Best-effort 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
|
||||
/// [`StopLiveAgent`], because the product contract is "stop the agent; cascade related
|
||||
/// work best-effort".
|
||||
#[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);
|
||||
}
|
||||
|
||||
/// Stops an already-running agent's live session, polymorphically.
|
||||
///
|
||||
/// Resolves the session by agent (PTY first, then structured). A PTY session is
|
||||
@ -129,6 +142,7 @@ pub struct StopLiveAgent {
|
||||
close: Arc<CloseTerminal>,
|
||||
input: Option<Arc<dyn InputMediator>>,
|
||||
waits: Option<Arc<OrchestratorService>>,
|
||||
cascade: Option<Arc<dyn StopAgentExecutionCascade>>,
|
||||
}
|
||||
|
||||
impl StopLiveAgent {
|
||||
@ -140,6 +154,7 @@ impl StopLiveAgent {
|
||||
close,
|
||||
input: None,
|
||||
waits: None,
|
||||
cascade: None,
|
||||
}
|
||||
}
|
||||
|
||||
@ -159,6 +174,13 @@ impl StopLiveAgent {
|
||||
self
|
||||
}
|
||||
|
||||
/// Wires best-effort cancellation of child resources spawned by the running turn.
|
||||
#[must_use]
|
||||
pub fn with_execution_cascade(mut self, cascade: Arc<dyn StopAgentExecutionCascade>) -> Self {
|
||||
self.cascade = Some(cascade);
|
||||
self
|
||||
}
|
||||
|
||||
/// Tears down the agent's live session.
|
||||
///
|
||||
/// # Errors
|
||||
@ -177,10 +199,18 @@ impl StopLiveAgent {
|
||||
}
|
||||
|
||||
async fn stop_dependencies(&self, project: &Project, agent_id: AgentId) {
|
||||
if let Some(cascade) = &self.cascade {
|
||||
cascade
|
||||
.stop_agent_execution_children(project, agent_id)
|
||||
.await;
|
||||
}
|
||||
let Some(waits) = &self.waits else {
|
||||
return;
|
||||
};
|
||||
for dep in waits.active_wait_dependencies(agent_id) {
|
||||
if let Some(cascade) = &self.cascade {
|
||||
cascade.stop_agent_execution_children(project, dep).await;
|
||||
}
|
||||
if let Some(mediator) = &self.input {
|
||||
mediator.preempt(RuntimeAgentKey::new(project.id, dep));
|
||||
}
|
||||
|
||||
@ -10,8 +10,8 @@ mod live;
|
||||
mod reconcile;
|
||||
|
||||
pub use actions::{
|
||||
AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput, StopLiveAgent,
|
||||
StopLiveAgentInput, StopLiveAgentOutput,
|
||||
AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput, StopAgentExecutionCascade,
|
||||
StopLiveAgent, StopLiveAgentInput, StopLiveAgentOutput,
|
||||
};
|
||||
pub use live::{
|
||||
GetLiveStateLean, LeanLiveEntry, LeanLiveState, UpdateLiveState, UpdateLiveStateInput,
|
||||
|
||||
@ -12,7 +12,8 @@ use async_trait::async_trait;
|
||||
|
||||
use application::{
|
||||
AttachLiveAgent, AttachLiveAgentInput, CloseTerminal, LiveSessionKind, LiveSessions,
|
||||
StopLiveAgent, StopLiveAgentInput, StructuredSessions, TerminalSessions,
|
||||
StopAgentExecutionCascade, StopLiveAgent, StopLiveAgentInput, StructuredSessions,
|
||||
TerminalSessions,
|
||||
};
|
||||
use domain::ports::{
|
||||
AgentSession, AgentSessionError, ExitStatus, OutputStream, PtyError, PtyHandle, PtyPort,
|
||||
@ -120,6 +121,18 @@ impl AgentSession for FakeSession {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct FakeCascade {
|
||||
calls: Mutex<Vec<(ProjectId, AgentId)>>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl StopAgentExecutionCascade for FakeCascade {
|
||||
async fn stop_agent_execution_children(&self, project: &Project, agent_id: AgentId) {
|
||||
self.calls.lock().unwrap().push((project.id, agent_id));
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// wiring helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
@ -351,6 +364,30 @@ async fn stop_structured_shuts_down_and_removes_session() {
|
||||
assert!(f.pty_port.kills().is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stop_structured_runs_execution_cascade_best_effort() {
|
||||
let f = fixture();
|
||||
let a = aid(10);
|
||||
let flag = insert_structured(&f.structured, sid(2), a, nid(100));
|
||||
let cascade = Arc::new(FakeCascade::default());
|
||||
let stop = f
|
||||
.stop
|
||||
.with_execution_cascade(Arc::clone(&cascade) as Arc<dyn StopAgentExecutionCascade>);
|
||||
|
||||
let out = stop
|
||||
.execute(StopLiveAgentInput {
|
||||
project: project(),
|
||||
agent_id: a,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(out.session_id, sid(2));
|
||||
assert_eq!(*cascade.calls.lock().unwrap(), vec![(project().id, a)]);
|
||||
assert!(flag.load(Ordering::SeqCst), "session.shutdown() was called");
|
||||
assert!(f.structured.is_empty(), "live session removed");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stop_unknown_agent_is_not_found() {
|
||||
let f = fixture();
|
||||
|
||||
Reference in New Issue
Block a user