diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index b4089e46..a1354c39 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -2764,29 +2764,39 @@ pub async fn interrupt_agent( .map_err(ErrorDto::from) } -/// `cancel_agent_chat` — interrupt the current turn of a live structured chat -/// session without shutting the session down. +/// `cancel_agent_chat` — stop a live structured chat session and cascade its +/// running execution children best-effort. /// -/// This is the session-id based twin of [`interrupt_agent`] for a custom chat +/// This is the session-id based twin of [`stop_live_agent`] for a custom chat /// surface that already owns the structured `sessionId`. /// /// # Errors /// Returns an [`ErrorDto`] (`INVALID` for a malformed id, `NOT_FOUND` if no live -/// structured session owns the id, `PROCESS` if a concrete cancel fails). +/// structured session owns the id, `PROCESS` if a concrete stop fails). #[tauri::command] pub async fn cancel_agent_chat( session_id: String, state: State<'_, AppState>, ) -> Result<(), ErrorDto> { let sid = parse_session_id(&session_id)?; - let session = state + let (project_id, agent_id, _, _) = state .structured_sessions - .session(&sid) + .meta_for_session(&sid) .ok_or_else(|| ErrorDto::from(AppError::NotFound(format!("structured session {sid}"))))?; - session - .cancel_current_turn() + let project = resolve_project(&project_id.to_string(), &state).await?; + let dependencies = state + .orchestrator_service + .active_wait_dependencies(agent_id); + state.stop_turn_watch(project.id, agent_id); + for dependency in dependencies { + state.stop_turn_watch(project.id, dependency); + } + state + .stop_live_agent + .execute(StopLiveAgentInput { project, agent_id }) .await - .map_err(|e| ErrorDto::from(AppError::from(e))) + .map(|_| ()) + .map_err(ErrorDto::from) } async fn preferred_view_for_agent( diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index 8e78f38e..a45f61ae 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -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, }; diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 85da2e65..721de71a 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -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>, + /// 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>, /// 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) -> 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>, + agent: RuntimeAgentKey, +) -> Vec { + 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>, + 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>, + } + + #[async_trait::async_trait] + impl BackgroundTaskRunner for SpyBackgroundRunner { + async fn spawn( + &self, + spec: domain::ports::BackgroundTaskSpec, + ) -> Result + { + 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] diff --git a/crates/application/src/workstate/actions.rs b/crates/application/src/workstate/actions.rs index 5c5b8fa8..364e36a3 100644 --- a/crates/application/src/workstate/actions.rs +++ b/crates/application/src/workstate/actions.rs @@ -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, input: Option>, waits: Option>, + cascade: Option>, } 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) -> 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)); } diff --git a/crates/application/src/workstate/mod.rs b/crates/application/src/workstate/mod.rs index 184650a0..bc1a1d90 100644 --- a/crates/application/src/workstate/mod.rs +++ b/crates/application/src/workstate/mod.rs @@ -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, diff --git a/crates/application/tests/workstate_actions.rs b/crates/application/tests/workstate_actions.rs index aaf1570b..fc7d0f46 100644 --- a/crates/application/tests/workstate_actions.rs +++ b/crates/application/tests/workstate_actions.rs @@ -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>, +} + +#[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); + + 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(); diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index da3050c8..e5b0a5d2 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -2907,6 +2907,7 @@ impl BackendCore { Arc::clone(&background_tasks_port), Arc::clone(&clock) as Arc, ) + .with_background_task_runner(Arc::clone(&background_runner_port)) // Limites de session sur le chemin délégué headless : quand le drain // structuré retourne `RateLimited`, l'orchestrateur arme la reprise pour // la cible limitée via le même service que les chemins directs. @@ -2929,7 +2930,9 @@ impl BackendCore { .with_cascade( Arc::clone(&input_mediator), Arc::clone(&orchestrator_service), - ), + ) + .with_execution_cascade(Arc::clone(&orchestrator_service) + as Arc), ); // --- Windows (L10) --- diff --git a/frontend/src/features/agents/CustomAgentChatView.test.tsx b/frontend/src/features/agents/CustomAgentChatView.test.tsx index a3a200c0..216165fa 100644 --- a/frontend/src/features/agents/CustomAgentChatView.test.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.test.tsx @@ -101,7 +101,12 @@ describe("CustomAgentChatView", () => { expect(screen.getByText("Fichier: clipboard.png")).toBeTruthy(); }); - it("cancels only the current turn and keeps the structured session alive", async () => { + it("stops the live agent with cascade feedback and releases the structured session", async () => { + let resolveStop: (value: { + agentId: string; + sessionId: string; + kind: "structured"; + }) => void = () => {}; const agent = { launchAgentChat: vi.fn(), reattachAgentChat: vi.fn(async (sessionId: string) => ({ @@ -109,6 +114,14 @@ describe("CustomAgentChatView", () => { scrollback: [], })), sendAgentChat: vi.fn(() => new Promise(() => {})), + stopLiveAgent: vi.fn( + () => + new Promise<{ agentId: string; sessionId: string; kind: "structured" }>( + (resolve) => { + resolveStop = resolve; + }, + ), + ), cancelAgentChat: vi.fn(async () => {}), closeAgentChat: vi.fn(async () => {}), }; @@ -159,24 +172,27 @@ describe("CustomAgentChatView", () => { fireEvent.click(await screen.findByRole("button", { name: "Stop" })); + const stoppingLabel = await screen.findByText("Arrêt en cours..."); + expect(stoppingLabel.closest("button")?.hasAttribute("disabled")).toBe(true); await waitFor(() => - expect(agent.cancelAgentChat).toHaveBeenCalledWith("chat-session-1"), + expect(agent.stopLiveAgent).toHaveBeenCalledWith("project-1", "agent-1"), ); - expect(agent.closeAgentChat).not.toHaveBeenCalled(); - expect(onSessionId).not.toHaveBeenCalledWith(null); - expect(screen.getByText("Tour interrompu.")).toBeTruthy(); - - fireEvent.change(screen.getByLabelText(/message CLI custom/), { - target: { value: "second turn" }, + resolveStop({ + agentId: "agent-1", + sessionId: "chat-session-1", + kind: "structured", }); - fireEvent.click(screen.getByRole("button", { name: "Envoyer" })); - - await waitFor(() => expect(agent.sendAgentChat).toHaveBeenCalledTimes(2)); - expect(agent.sendAgentChat).toHaveBeenLastCalledWith( - "chat-session-1", - "second turn", - expect.any(Function), - ); + await waitFor(() => expect(onSessionId).toHaveBeenCalledWith(null)); + expect(agent.cancelAgentChat).not.toHaveBeenCalled(); + expect(agent.closeAgentChat).not.toHaveBeenCalled(); + expect( + screen.getByText( + "Agent arrêté. Cascade d'arrêt demandée pour les tâches et appels MCP liés.", + ), + ).toBeTruthy(); + expect( + screen.getByRole("button", { name: "Envoyer" }).hasAttribute("disabled"), + ).toBe(true); }); it("falls back to a fresh launch when reattach reports NOT_FOUND", async () => { @@ -2569,10 +2585,11 @@ describe("CustomAgentChatView", () => { scrollback: [], })), sendAgentChat: vi.fn(() => new Promise(() => {})), - cancelAgentChat: vi.fn().mockRejectedValueOnce({ + stopLiveAgent: vi.fn().mockRejectedValueOnce({ code: "NOT_FOUND", - message: "not found: structured session gone-session", + message: "agent agent-1 has no live session", }), + cancelAgentChat: vi.fn(async () => {}), closeAgentChat: vi.fn(async () => {}), }; const onSessionId = vi.fn(); @@ -2614,8 +2631,9 @@ describe("CustomAgentChatView", () => { fireEvent.click(await screen.findByRole("button", { name: "Stop" })); - expect(await screen.findByText("Session déjà fermée.")).toBeTruthy(); + expect(await screen.findByText("Session déjà arrêtée.")).toBeTruthy(); expect(onSessionId).toHaveBeenCalledWith(null); + expect(agent.cancelAgentChat).not.toHaveBeenCalled(); expect(agent.closeAgentChat).not.toHaveBeenCalled(); expect(screen.queryByText(/not found: structured session/)).toBeNull(); }); diff --git a/frontend/src/features/agents/CustomAgentChatView.tsx b/frontend/src/features/agents/CustomAgentChatView.tsx index 8b5ac994..2001bcc2 100644 --- a/frontend/src/features/agents/CustomAgentChatView.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.tsx @@ -534,6 +534,7 @@ export function CustomAgentChatView({ useState(null); const [opening, setOpening] = useState(false); const [busy, setBusy] = useState(false); + const [stopping, setStopping] = useState(false); const [error, setError] = useState(null); const scrollRef = useRef(null); const composerRef = useRef(null); @@ -1209,23 +1210,46 @@ export function CustomAgentChatView({ } } - async function cancel() { + async function stopAgent() { const sid = sessionRef.current; - if (!sid || !agent.cancelAgentChat) return; - setBusy(false); - setOpening(false); + if (!sid || !agent.stopLiveAgent || stopping) return; + setStopping(true); setError(null); try { - await agent.cancelAgentChat(sid); - setTurns((prev) => [...prev, { role: "tool", label: "Tour interrompu." }]); + const stopped = await agent.stopLiveAgent(projectId, agentId); + setBusy(false); + setOpening(false); + setCurrentSession(null); + publishSessionId(null); + setTurns((prev) => [ + ...completeRunningProgress(prev), + { + role: "tool", + label: + stopped.kind === "structured" + ? "Agent arrêté. Cascade d'arrêt demandée pour les tâches et appels MCP liés." + : "Agent arrêté.", + }, + ]); } catch (e) { if (isNotFound(e)) { + setBusy(false); + setOpening(false); setCurrentSession(null); publishSessionId(null); - setTurns((prev) => [...prev, { role: "tool", label: "Session déjà fermée." }]); + setTurns((prev) => [ + ...completeRunningProgress(prev), + { role: "tool", label: "Session déjà arrêtée." }, + ]); return; } setError(describe(e)); + setTurns((prev) => [ + ...prev, + { role: "error", text: `Arrêt de l'agent échoué: ${describe(e)}` }, + ]); + } finally { + setStopping(false); } } @@ -1511,10 +1535,15 @@ export function CustomAgentChatView({ size="sm" variant={busy ? "danger" : undefined} className="shrink-0 whitespace-nowrap" - disabled={busy ? !supported || !currentSession : !canSend} - onClick={() => (busy ? void cancel() : void send())} + disabled={ + busy + ? !supported || !currentSession || !agent.stopLiveAgent || stopping + : !canSend + } + loading={stopping} + onClick={() => (busy ? void stopAgent() : void send())} > - {busy ? "Stop" : "Envoyer"} + {stopping ? "Arrêt en cours..." : busy ? "Stop" : "Envoyer"}