From 1aabc3d702b9734fe3d6ee670e6aa01483b741b9 Mon Sep 17 00:00:00 2001 From: Blomios Date: Fri, 28 Aug 2026 22:57:32 +0200 Subject: [PATCH] =?UTF-8?q?fix(custom-cli):=20le=20bouton=20Stop=20annule?= =?UTF-8?q?=20le=20tour=20courant=20sans=20fermer=20la=20session=20et=20ca?= =?UTF-8?q?scade=20l'annulation=20aux=20agents=20d=C3=A9l=C3=A9gu=C3=A9s?= =?UTF-8?q?=20=E2=80=94=20stop=5Fagent=20suit=20l'arbre=20des=20d=C3=A9pen?= =?UTF-8?q?dances=20actives=20(cancel=5Fcurrent=5Fturn=20+=20pr=C3=A9empti?= =?UTF-8?q?on=20runtime=20+=20enfants=20d'ex=C3=A9cution),=20TurnCancel=20?= =?UTF-8?q?tue=20le=20process=20du=20tour=20en=20cours,=20autorisation=20r?= =?UTF-8?q?equester/sous-arbre=20sur=20idea=5Fstop=5Fagent,=20fixtures=20D?= =?UTF-8?q?TO=20runtime=5Fstate/node=5Fid=20align=C3=A9es=20=E2=80=94=20#2?= =?UTF-8?q?57=20(QA=20verte)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/app-tauri/tests/dto_agents.rs | 11 +- .../application/src/orchestrator/service.rs | 52 ++++- crates/application/src/workstate/actions.rs | 82 +++++-- .../application/tests/orchestrator_service.rs | 75 ++++++- crates/application/tests/workstate_actions.rs | 49 +++-- crates/domain/src/orchestrator.rs | 9 +- .../src/orchestrator/mcp/tools.rs | 17 +- crates/infrastructure/src/session/claude.rs | 18 +- crates/infrastructure/src/session/codex.rs | 18 +- crates/infrastructure/src/session/process.rs | 203 ++++++++++++++++-- .../agents/CustomAgentChatView.test.tsx | 81 +++---- .../features/agents/CustomAgentChatView.tsx | 29 ++- 12 files changed, 534 insertions(+), 110 deletions(-) diff --git a/crates/app-tauri/tests/dto_agents.rs b/crates/app-tauri/tests/dto_agents.rs index f99ee3aa..2a5a9657 100644 --- a/crates/app-tauri/tests/dto_agents.rs +++ b/crates/app-tauri/tests/dto_agents.rs @@ -11,8 +11,8 @@ use application::AppError; use application::{ AgentCapability, AgentTicketState, AgentWorkState, CreateAgentOutput, InspectConversationOutput, LaunchAgentOutput, ListAgentsOutput, ListedAgentCapabilities, - LiveSessionKind, LiveSessionSnapshot, LiveWorkSession, ProjectWorkState, TicketWorkSource, - TicketWorkStatus, + LiveRuntimeState, LiveSessionKind, LiveSessionSnapshot, LiveWorkSession, ProjectWorkState, + TicketWorkSource, TicketWorkStatus, }; use domain::ids::{AgentId, NodeId, ProfileId, SessionId}; use domain::ports::ConversationDetails; @@ -292,6 +292,7 @@ fn project_work_state_dto_serialises_live_and_busy_camelcase() { ticket, since_ms: 1_234, }, + runtime_state: None, tickets: vec![], background_tasks: vec![], }], @@ -328,6 +329,7 @@ fn project_work_state_dto_serialises_tickets_camelcase() { profile_id: profile, live: None, busy: domain::AgentBusyState::Idle, + runtime_state: None, tickets: vec![ AgentTicketState { ticket_id: head_ticket, @@ -690,18 +692,21 @@ fn stop_live_agent_response_serialises_camelcase_with_kind() { use application::{LiveSessionKind, StopLiveAgentOutput}; let agent = AgentId::from_uuid(Uuid::from_u128(11)); + let node = NodeId::from_uuid(Uuid::from_u128(12)); let session = SessionId::from_uuid(Uuid::from_u128(13)); let dto = StopLiveAgentResponseDto::from(StopLiveAgentOutput { agent_id: agent, + node_id: node, session_id: session, kind: LiveSessionKind::Structured, + runtime_state: LiveRuntimeState::Stopped, }); let v = serde_json::to_value(&dto).unwrap(); assert_eq!(v["agentId"], agent.to_string()); assert_eq!(v["sessionId"], session.to_string()); assert_eq!(v["kind"], "structured"); - // Stop response carries no node id (the session is gone). + assert_eq!(v["runtimeState"], "stopped"); assert!(v.get("nodeId").is_none()); assert!(v.get("session_id").is_none()); } diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 5e01904c..68132af6 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -6,7 +6,7 @@ //! that turns a request into the *same* calls the UI makes: //! //! - `spawn_agent` → [`CreateAgentFromScratch`] (if unknown) then [`LaunchAgent`], -//! - `stop_agent` → stop the agent's live PTY or structured runtime, +//! - `stop_agent` → stop the agent's live PTY or cancel its structured turn, //! - `update_agent_context` → [`UpdateAgentContext`]. //! //! It talks **only** to use cases and ports ([`ProfileStore`], [`TerminalSessions`]): @@ -1224,7 +1224,9 @@ impl OrchestratorService { result, } => self.reply(project, from, ticket, result), OrchestratorCommand::ListAgents => self.list_agents(project).await, - OrchestratorCommand::StopAgent { name } => self.stop_agent(project, name).await, + OrchestratorCommand::StopAgent { name, requester } => { + self.stop_agent(project, name, requester).await + } OrchestratorCommand::UpdateAgentContext { name, context } => { self.update_agent_context(project, name, context).await } @@ -2713,23 +2715,65 @@ impl OrchestratorService { }) } - /// `stop_agent`: translate the agent name → its live runtime → stop it. + /// `stop_agent`: translate the agent name → its live runtime → stop/cancel it. async fn stop_agent( &self, project: &Project, name: String, + requester: Option, ) -> Result { let agent_id = self .find_agent_id_by_name(project, &name) .await? .ok_or_else(|| AppError::NotFound(format!("agent {name}")))?; - self.stop_agent_execution_children(project, agent_id).await; + if let Some(requester) = requester { + let allowed = requester == agent_id + || self + .active_wait_dependencies(requester) + .into_iter() + .any(|dep| dep == agent_id); + if !allowed { + return Err(AppError::Invalid(format!( + "agent {requester} cannot stop agent {agent_id} outside its active subtree" + ))); + } + } 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 {name}"), + 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? diff --git a/crates/application/src/workstate/actions.rs b/crates/application/src/workstate/actions.rs index 7b130086..39033220 100644 --- a/crates/application/src/workstate/actions.rs +++ b/crates/application/src/workstate/actions.rs @@ -3,8 +3,8 @@ //! Companion to the read-only work-state model: these use cases let the UI drive //! an **already-running** agent's live session by **agent id**, never by spawning a //! new process. [`AttachLiveAgent`] rebinds the hosting cell ("a cell is a view"); -//! [`StopLiveAgent`] tears the live session down polymorphically (PTY kill vs -//! structured shutdown). Neither touches the agent entity, its tickets, the +//! [`StopLiveAgent`] tears a PTY live session down or cancels the current structured +//! turn while preserving the structured session. Neither touches the agent entity, its tickets, the //! conversation summaries, the handoff or the provider sessions. use std::sync::Arc; @@ -136,14 +136,13 @@ pub trait StopAgentExecutionCascade: Send + Sync { async fn stop_agent_execution_children(&self, project: &Project, agent_id: AgentId); } -/// Stops an already-running agent's live session, polymorphically. +/// Stops an already-running agent's current work. /// -/// Resolves the session by agent (PTY first, then structured). A PTY session is -/// torn down through the existing [`CloseTerminal`] primitive (registry removal + -/// process kill); a structured session is removed then `shutdown()`. The agent -/// entity, its tickets, conversation summaries and handoff are **not** removed. Any -/// exit event is left to the existing infrastructure (the PTY pump / session -/// backend) — no bus is invented here. +/// Resolves the session by agent (structured first, then PTY). A structured +/// session keeps its registry entry and receives `cancel_current_turn()`, so the +/// provider conversation/context survives the stop. A PTY session has no finer +/// turn boundary, so it is torn down through the existing [`CloseTerminal`] +/// primitive (registry removal + process kill). pub struct StopLiveAgent { live: Arc, close: Arc, @@ -188,7 +187,7 @@ impl StopLiveAgent { self } - /// Tears down the agent's live session. + /// Stops the agent's current work. /// /// # Errors /// - [`AppError::NotFound`] when the agent has no live session, @@ -197,22 +196,33 @@ impl StopLiveAgent { &self, input: StopLiveAgentInput, ) -> Result { - self.stop_dependencies(&input.project, input.agent_id).await; + self.stop_dependencies(&input.project, input.agent_id) + .await?; if let Some(mediator) = &self.input { mediator.preempt(RuntimeAgentKey::new(input.project.id, input.agent_id)); } + if let Some(out) = self + .cancel_structured_turn_in_project(&input.project, input.agent_id) + .await? + { + return Ok(out); + } self.stop_one_in_project(&input.project, input.agent_id) .await } - async fn stop_dependencies(&self, project: &Project, agent_id: AgentId) { + async fn stop_dependencies( + &self, + project: &Project, + agent_id: AgentId, + ) -> Result<(), AppError> { if let Some(cascade) = &self.cascade { cascade .stop_agent_execution_children(project, agent_id) .await; } let Some(waits) = &self.waits else { - return; + return Ok(()); }; for dep in waits.active_wait_dependencies(agent_id) { if let Some(cascade) = &self.cascade { @@ -221,8 +231,52 @@ impl StopLiveAgent { if let Some(mediator) = &self.input { mediator.preempt(RuntimeAgentKey::new(project.id, dep)); } - let _ = self.stop_one_in_project(project, dep).await; + if self + .cancel_structured_turn_in_project(project, dep) + .await? + .is_none() + { + let _ = self.stop_one_in_project(project, dep).await; + } } + Ok(()) + } + + async fn cancel_structured_turn_in_project( + &self, + project: &Project, + agent_id: AgentId, + ) -> Result, AppError> { + let Some(snapshot) = self + .live + .live_agent_snapshots_for(project.id, &agent_id) + .into_iter() + .find(|snapshot| snapshot.kind == LiveSessionKind::Structured) + else { + return Ok(None); + }; + let Some(session) = self + .live + .structured + .session_for_agent_in_project(project.id, &agent_id) + else { + return Ok(None); + }; + self.live + .mark_runtime_state(snapshot, LiveRuntimeState::Stopping); + session + .cancel_current_turn() + .await + .map_err(|err| AppError::Process(err.to_string()))?; + self.live + .mark_runtime_state(snapshot, LiveRuntimeState::Stopped); + Ok(Some(StopLiveAgentOutput { + agent_id: snapshot.agent_id, + session_id: snapshot.session_id, + node_id: snapshot.node_id, + kind: snapshot.kind, + runtime_state: LiveRuntimeState::Stopped, + })) } async fn stop_one_in_project( diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index f116a378..f887c30c 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -1034,14 +1034,18 @@ async fn stop_agent_kills_the_right_session() { } #[tokio::test] -async fn stop_agent_stops_structured_session() { +async fn stop_agent_cancels_structured_turn_without_closing_session() { let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md"); let contexts = FakeContexts::with_agent(&agent, "# persona"); let fx = ask_fixture(contexts); + let cancel_called = Arc::new(Mutex::new(0_u32)); + let shutdown_called = Arc::new(Mutex::new(0_u32)); let session = Arc::new(CompletionSession { id: sid(9010), agent: agent.id, completions: fx.mailbox.completions(), + cancel_called: Some(Arc::clone(&cancel_called)), + shutdown_called: Some(Arc::clone(&shutdown_called)), }) as Arc; fx.structured .insert_in_project(project_id(), session, agent.id, nid(9010)); @@ -1062,10 +1066,60 @@ async fn stop_agent_stops_structured_session() { fx.pty.kills().is_empty(), "structured stop must not kill PTY" ); + assert_eq!(*cancel_called.lock().unwrap(), 1); + assert_eq!( + *shutdown_called.lock().unwrap(), + 0, + "structured stop must not shutdown the session" + ); + assert!( + fx.structured + .session_for_agent_in_project(project_id(), &agent.id) + .is_some(), + "structured session is preserved for later turns" + ); +} + +#[tokio::test] +async fn stop_agent_from_unrelated_agent_is_rejected() { + let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md"); + let contexts = FakeContexts::with_agent(&agent, "# persona"); + let fx = ask_fixture(contexts); + let cancel_called = Arc::new(Mutex::new(0_u32)); + let shutdown_called = Arc::new(Mutex::new(0_u32)); + let session = Arc::new(CompletionSession { + id: sid(9010), + agent: agent.id, + completions: fx.mailbox.completions(), + cancel_called: Some(Arc::clone(&cancel_called)), + shutdown_called: Some(Arc::clone(&shutdown_called)), + }) as Arc; + fx.structured + .insert_in_project(project_id(), session, agent.id, nid(9010)); + + let requester = uuid::Uuid::from_u128(99); + let err = fx + .service + .dispatch( + &project(), + cmd(&format!( + r#"{{ "type":"agent.stop", "requestedBy":"{requester}", "targetAgent":"dev-backend" }}"# + )), + ) + .await + .unwrap_err(); + + assert_eq!(err.code(), "INVALID"); + assert_eq!( + *cancel_called.lock().unwrap(), + 0, + "unrelated agent must not cancel target work" + ); + assert_eq!(*shutdown_called.lock().unwrap(), 0); assert!(fx .structured .session_for_agent_in_project(project_id(), &agent.id) - .is_none()); + .is_some()); } #[tokio::test] @@ -1162,6 +1216,8 @@ async fn ask_agent_after_delete_recreate_same_name_uses_new_agent_id() { id: sid(9001), agent: old.id, completions: fx.mailbox.completions(), + cancel_called: None, + shutdown_called: None, }) as Arc; fx.structured .insert_in_project(project_id(), old_session, old.id, nid(9001)); @@ -1617,6 +1673,8 @@ struct CompletionSession { id: SessionId, agent: AgentId, completions: Arc, + cancel_called: Option>>, + shutdown_called: Option>>, } #[async_trait] @@ -1643,7 +1701,16 @@ impl AgentSession for CompletionSession { )), } } + async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> { + if let Some(cancel_called) = &self.cancel_called { + *cancel_called.lock().unwrap() += 1; + } + Ok(()) + } async fn shutdown(&self) -> Result<(), AgentSessionError> { + if let Some(shutdown_called) = &self.shutdown_called { + *shutdown_called.lock().unwrap() += 1; + } Ok(()) } } @@ -1701,6 +1768,8 @@ impl AgentSessionFactory for CompletionFactory { id, agent, completions: Arc::clone(&self.completions), + cancel_called: None, + shutdown_called: None, })) } } @@ -1728,6 +1797,8 @@ fn seed_structured_sessions( id: SessionId::from_uuid(Uuid::from_u128(8_000 + idx as u128)), agent: entry.agent_id, completions: Arc::clone(&completions), + cancel_called: None, + shutdown_called: None, }) as Arc; structured.insert(session, entry.agent_id, nid(8_000 + idx as u128)); } diff --git a/crates/application/tests/workstate_actions.rs b/crates/application/tests/workstate_actions.rs index bd482c79..38e84849 100644 --- a/crates/application/tests/workstate_actions.rs +++ b/crates/application/tests/workstate_actions.rs @@ -98,10 +98,11 @@ impl PtyPort for FakePty { } } -/// A fake structured [`AgentSession`] recording whether `shutdown` was called. +/// A fake structured [`AgentSession`] recording stop/cancel calls. struct FakeSession { id: SessionId, shutdown_called: Arc, + cancel_called: Arc, } #[async_trait] @@ -115,6 +116,10 @@ impl AgentSession for FakeSession { async fn send(&self, _prompt: &str) -> Result { Ok(Box::new(std::iter::empty())) } + async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> { + self.cancel_called.store(true, Ordering::SeqCst); + Ok(()) + } async fn shutdown(&self) -> Result<(), AgentSessionError> { self.shutdown_called.store(true, Ordering::SeqCst); Ok(()) @@ -187,18 +192,20 @@ fn insert_structured( session_id: SessionId, agent_id: AgentId, node_id: NodeId, -) -> Arc { - let flag = Arc::new(AtomicBool::new(false)); +) -> (Arc, Arc) { + let shutdown_called = Arc::new(AtomicBool::new(false)); + let cancel_called = Arc::new(AtomicBool::new(false)); sessions.insert_in_project( project().id, Arc::new(FakeSession { id: session_id, - shutdown_called: Arc::clone(&flag), + shutdown_called: Arc::clone(&shutdown_called), + cancel_called: Arc::clone(&cancel_called), }), agent_id, node_id, ); - flag + (shutdown_called, cancel_called) } // --------------------------------------------------------------------------- @@ -239,7 +246,7 @@ fn attach_pty_rebinds_node_without_changing_session() { fn attach_structured_rebinds_node() { let f = fixture(); let a = aid(10); - let _flag = insert_structured(&f.structured, sid(2), a, nid(100)); + let _flags = insert_structured(&f.structured, sid(2), a, nid(100)); let out = f .attach @@ -339,10 +346,10 @@ async fn stop_pty_kills_and_removes_session() { } #[tokio::test] -async fn stop_structured_shuts_down_and_removes_session() { +async fn stop_structured_cancels_turn_and_keeps_session() { let f = fixture(); let a = aid(10); - let flag = insert_structured(&f.structured, sid(2), a, nid(100)); + let (shutdown_called, cancel_called) = insert_structured(&f.structured, sid(2), a, nid(100)); let out = f .stop @@ -356,12 +363,19 @@ async fn stop_structured_shuts_down_and_removes_session() { assert_eq!(out.session_id, sid(2)); assert_eq!(out.kind, LiveSessionKind::Structured); assert_eq!(out.runtime_state, LiveRuntimeState::Stopped); - assert!(flag.load(Ordering::SeqCst), "session.shutdown() was called"); - assert!(f.structured.is_empty(), "live session removed"); + assert!( + cancel_called.load(Ordering::SeqCst), + "session.cancel_current_turn() was called" + ); + assert!( + !shutdown_called.load(Ordering::SeqCst), + "session.shutdown() must not be called" + ); + assert!(!f.structured.is_empty(), "live session is preserved"); assert_eq!( f.structured .session_id_for_agent_in_project(project().id, &a), - None + Some(sid(2)) ); // No PTY was touched. assert!(f.pty_port.kills().is_empty()); @@ -371,7 +385,7 @@ async fn stop_structured_shuts_down_and_removes_session() { 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 (shutdown_called, cancel_called) = insert_structured(&f.structured, sid(2), a, nid(100)); let cascade = Arc::new(FakeCascade::default()); let stop = f .stop @@ -387,8 +401,15 @@ async fn stop_structured_runs_execution_cascade_best_effort() { 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"); + assert!( + cancel_called.load(Ordering::SeqCst), + "session.cancel_current_turn() was called" + ); + assert!( + !shutdown_called.load(Ordering::SeqCst), + "session.shutdown() must not be called" + ); + assert!(!f.structured.is_empty(), "live session is preserved"); } #[tokio::test] diff --git a/crates/domain/src/orchestrator.rs b/crates/domain/src/orchestrator.rs index ea1568b7..3e800606 100644 --- a/crates/domain/src/orchestrator.rs +++ b/crates/domain/src/orchestrator.rs @@ -259,10 +259,12 @@ pub enum OrchestratorCommand { /// Desired visibility for the launched session. visibility: OrchestratorVisibility, }, - /// Stop a running agent by killing its terminal session. + /// Stop/cancel a running agent. StopAgent { /// Target agent display name. name: String, + /// Requesting agent, when the stop originates from an agent tool. + requester: Option, }, /// Overwrite an agent's `.md` context with a new body. UpdateAgentContext { @@ -482,9 +484,11 @@ impl OrchestratorRequest { }), "stop_agent" => Ok(OrchestratorCommand::StopAgent { name: self.require_name(action)?, + requester: self.optional_requester(action)?, }), "agent.stop" => Ok(OrchestratorCommand::StopAgent { name: self.require_target_agent(action)?, + requester: self.optional_requester(action)?, }), "update_agent_context" => Ok(OrchestratorCommand::UpdateAgentContext { name: self.require_name(action)?, @@ -965,7 +969,8 @@ mod tests { assert_eq!( r.validate().unwrap(), OrchestratorCommand::StopAgent { - name: "dev-backend".to_owned() + name: "dev-backend".to_owned(), + requester: None, } ); } diff --git a/crates/infrastructure/src/orchestrator/mcp/tools.rs b/crates/infrastructure/src/orchestrator/mcp/tools.rs index 5f4b2676..1a226fa3 100644 --- a/crates/infrastructure/src/orchestrator/mcp/tools.rs +++ b/crates/infrastructure/src/orchestrator/mcp/tools.rs @@ -492,6 +492,7 @@ pub fn map_tool_call( }, "idea_stop_agent" => OrchestratorRequest { request_type: Some("agent.stop".to_owned()), + requested_by: Some(requester.to_owned()), target_agent: s("target"), ..base() }, @@ -840,7 +841,8 @@ mod tests { assert_eq!( map("idea_stop_agent", &json!({ "target": "Dev" })).unwrap(), OrchestratorCommand::StopAgent { - name: "Dev".to_owned() + name: "Dev".to_owned(), + requester: None, } ); assert_eq!( @@ -868,6 +870,19 @@ mod tests { ); } + #[test] + fn stop_agent_maps_handshake_requester_for_authorization() { + assert_eq!( + map_tool_call("idea_stop_agent", &json!({ "target": "Dev" }), REQ).unwrap(), + OrchestratorCommand::StopAgent { + name: "Dev".to_owned(), + requester: Some(domain::AgentId::from_uuid( + uuid::Uuid::parse_str(REQ).unwrap() + )), + } + ); + } + #[test] fn missing_required_field_surfaces_validation_error() { let err = map("idea_update_context", &json!({ "target": "Architect" })); diff --git a/crates/infrastructure/src/session/claude.rs b/crates/infrastructure/src/session/claude.rs index ddca8bba..fc538d78 100644 --- a/crates/infrastructure/src/session/claude.rs +++ b/crates/infrastructure/src/session/claude.rs @@ -25,7 +25,7 @@ use domain::ports::{ use domain::sandbox::{SandboxEnforcer, SandboxPlan}; use domain::SessionId; -use super::process::{run_turn, SpawnLine}; +use super::process::{run_turn_cancellable, SpawnLine, TurnCancel}; /// Résultat du parsing d'une ligne : **zéro ou plusieurs** événements à émettre et/ou /// un `session_id` capté (init/result). Permet à [`parse_event`] de rester **pure** @@ -306,6 +306,8 @@ pub struct ClaudeSdkSession { /// Enforcer OS **par instance** (lot LP4-4), passé à [`run_turn`]. `None` ⇒ pas /// de sandboxing même si un plan est présent (cohérent avec le chemin PTY). sandbox_enforcer: Option>, + /// Cancel handle for the process backing the current `claude` turn. + turn_cancel: TurnCancel, } impl ClaudeSdkSession { @@ -330,6 +332,7 @@ impl ClaudeSdkSession { conversation_id: Mutex::new(seed_conversation_id), sandbox, sandbox_enforcer, + turn_cancel: TurnCancel::new(), } } @@ -391,6 +394,10 @@ impl AgentSession for ClaudeSdkSession { // tours, donc `shutdown` est intrinsèquement idempotent et sans effet. Ok(()) } + + async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> { + self.turn_cancel.cancel().await + } } impl ClaudeSdkSession { @@ -415,7 +422,14 @@ impl ClaudeSdkSession { }); (Some(line_tx), Some(parser)) }); - let raw_result = run_turn(&spec, None, self.sandbox_enforcer.as_ref(), line_tap).await; + let raw_result = run_turn_cancellable( + &spec, + None, + self.sandbox_enforcer.as_ref(), + line_tap, + &self.turn_cancel, + ) + .await; if let Some(parser) = parser_thread { let _ = parser.join(); } diff --git a/crates/infrastructure/src/session/codex.rs b/crates/infrastructure/src/session/codex.rs index 3304f529..a600dc09 100644 --- a/crates/infrastructure/src/session/codex.rs +++ b/crates/infrastructure/src/session/codex.rs @@ -22,7 +22,7 @@ use domain::ports::{ use domain::sandbox::{SandboxEnforcer, SandboxPlan}; use domain::SessionId; -use super::process::{run_turn, SpawnLine}; +use super::process::{run_turn_cancellable, SpawnLine, TurnCancel}; /// Résultat du parsing d'une ligne Codex : **zéro ou plusieurs** événements et/ou un /// id de conversation Codex capté. Miroir de `claude::ParsedLine` (vecteur d'events @@ -377,6 +377,8 @@ pub struct CodexExecSession { /// Enforcer OS **par instance** (lot LP4-4), passé à [`run_turn`]. `None` ⇒ pas /// de sandboxing même si un plan est présent (cohérent avec le chemin PTY). sandbox_enforcer: Option>, + /// Cancel handle for the process backing the current `codex exec` turn. + turn_cancel: TurnCancel, } impl CodexExecSession { @@ -473,6 +475,7 @@ impl CodexExecSession { conversation_id: Mutex::new(seed_conversation_id), sandbox, sandbox_enforcer, + turn_cancel: TurnCancel::new(), } } @@ -594,6 +597,10 @@ impl AgentSession for CodexExecSession { // « un run par tour » ⇒ pas de process long survivant : idempotent, no-op. Ok(()) } + + async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> { + self.turn_cancel.cancel().await + } } impl CodexExecSession { @@ -631,7 +638,14 @@ impl CodexExecSession { }); (Some(line_tx), Some(parser)) }); - let raw_result = run_turn(&spec, None, self.sandbox_enforcer.as_ref(), line_tap).await; + let raw_result = run_turn_cancellable( + &spec, + None, + self.sandbox_enforcer.as_ref(), + line_tap, + &self.turn_cancel, + ) + .await; if let Some(parser) = parser_thread { let _ = parser.join(); } diff --git a/crates/infrastructure/src/session/process.rs b/crates/infrastructure/src/session/process.rs index 84b21392..4f527355 100644 --- a/crates/infrastructure/src/session/process.rs +++ b/crates/infrastructure/src/session/process.rs @@ -29,11 +29,12 @@ use std::io; use std::process::Stdio; use std::sync::mpsc::Sender; -use std::sync::Arc; +use std::sync::{Arc, Mutex as StdMutex}; use std::time::Duration; use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::process::Command; +use tokio::sync::Mutex as AsyncMutex; use domain::ports::AgentSessionError; use domain::sandbox::{SandboxEnforcer, SandboxPlan}; @@ -61,6 +62,94 @@ pub struct SpawnLine { pub sandbox: Option, } +/// Shared handle used by structured adapters to cancel the process backing the +/// current turn without shutting down the reusable session. +#[derive(Clone, Default)] +pub struct TurnCancel { + async_child: Arc>>>>, + std_child: Arc>>>>, +} + +impl TurnCancel { + /// Creates an empty cancel handle with no active child. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + fn set_async(&self, child: Arc>) { + *self + .async_child + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(child); + } + + fn clear_async(&self, child: &Arc>) { + let mut current = self + .async_child + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if current + .as_ref() + .is_some_and(|stored| Arc::ptr_eq(stored, child)) + { + *current = None; + } + } + + fn set_std(&self, child: Arc>) { + *self + .std_child + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(child); + } + + fn clear_std(&self, child: &Arc>) { + let mut current = self + .std_child + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if current + .as_ref() + .is_some_and(|stored| Arc::ptr_eq(stored, child)) + { + *current = None; + } + } + + /// Kills the current turn process, if any. Missing/already-finished is a no-op. + pub async fn cancel(&self) -> Result<(), AgentSessionError> { + let async_child = self + .async_child + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone(); + if let Some(child) = async_child { + child + .lock() + .await + .kill() + .await + .map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?; + } + + let std_child = self + .std_child + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone(); + if let Some(child) = std_child { + let mut child = child + .lock() + .map_err(|_| AgentSessionError::Io("child lock empoisonné".to_owned()))?; + child + .kill() + .map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?; + } + Ok(()) + } +} + /// Lance `spec`, pousse `spec.stdin` sur l'entrée standard, **draine toutes les /// lignes de stdout jusqu'à EOF**, puis attend la fin du process. Retourne les /// lignes brutes (sans `\n`) dans l'ordre d'émission. @@ -84,14 +173,42 @@ pub async fn run_turn( timeout: Option, enforcer: Option<&Arc>, line_tap: Option>, +) -> Result, AgentSessionError> { + run_turn_impl(spec, timeout, enforcer, line_tap, None).await +} + +/// Same as [`run_turn`], but tracks the live child in `cancel` while the turn runs. +pub async fn run_turn_cancellable( + spec: &SpawnLine, + timeout: Option, + enforcer: Option<&Arc>, + line_tap: Option>, + cancel: &TurnCancel, +) -> Result, AgentSessionError> { + run_turn_impl(spec, timeout, enforcer, line_tap, Some(cancel)).await +} + +async fn run_turn_impl( + spec: &SpawnLine, + timeout: Option, + enforcer: Option<&Arc>, + line_tap: Option>, + cancel: Option<&TurnCancel>, ) -> Result, AgentSessionError> { // Chemin SANDBOXÉ (Linux + plan posé + enforcer câblé) : transpose la technique // du PTY (`spawn_command_sandboxed`) — enforce sur un thread jetable AVANT le fork, // l'enfant hérite le domaine via fork+exec. #[cfg(target_os = "linux")] if let (Some(plan), Some(enforcer)) = (spec.sandbox.as_ref(), enforcer) { - return run_turn_sandboxed(spec, plan.clone(), Arc::clone(enforcer), timeout, line_tap) - .await; + return run_turn_sandboxed( + spec, + plan.clone(), + Arc::clone(enforcer), + timeout, + line_tap, + cancel.cloned(), + ) + .await; } // Hors Linux : aucun sandboxing OS ⇒ on ignore l'enforcer (Noop de toute façon) et // on garde le drain async historique. `let _` évite l'avertissement « unused ». @@ -99,11 +216,16 @@ pub async fn run_turn( let _ = enforcer; match timeout { - Some(dur) => match tokio::time::timeout(dur, drain(spec, line_tap)).await { + Some(dur) => match tokio::time::timeout(dur, drain(spec, line_tap, cancel)).await { Ok(result) => result, - Err(_elapsed) => Err(AgentSessionError::Timeout), + Err(_elapsed) => { + if let Some(cancel) = cancel { + let _ = cancel.cancel().await; + } + Err(AgentSessionError::Timeout) + } }, - None => drain(spec, line_tap).await, + None => drain(spec, line_tap, cancel).await, } } @@ -145,9 +267,8 @@ async fn run_turn_sandboxed( enforcer: Arc, timeout: Option, line_tap: Option>, + cancel: Option, ) -> Result, AgentSessionError> { - use std::sync::Mutex as StdMutex; - // Données possédées : rien n'emprunte le thread jetable. let command = spec.command.clone(); let args = spec.args.clone(); @@ -163,7 +284,7 @@ async fn run_turn_sandboxed( // Thread JETABLE : sa restriction Landlock meurt avec lui. std::thread::spawn(move || { let result = drain_sandboxed( - command, args, cwd, env, stdin, &enforcer, &plan, killer_tx, line_tap, + command, args, cwd, env, stdin, &enforcer, &plan, killer_tx, line_tap, cancel, ); // Le récepteur peut avoir abandonné (timeout) : on ignore l'erreur d'envoi. let _ = done_tx.send(result); @@ -218,6 +339,7 @@ fn drain_sandboxed( plan: &SandboxPlan, killer_tx: tokio::sync::oneshot::Sender>>, line_tap: Option>, + cancel: Option, ) -> Result, AgentSessionError> { use std::io::{BufRead, BufReader as StdBufReader, Read as _, Write as _}; use std::process::{Command as StdCommand, Stdio}; @@ -278,6 +400,9 @@ fn drain_sandboxed( .map_err(|e| AgentSessionError::Io(e.to_string())) }); let child = Arc::new(StdMutex::new(child)); + if let Some(cancel) = &cancel { + cancel.set_std(Arc::clone(&child)); + } // Donne au côté async de quoi tuer l'enfant en cas de timeout. let _ = killer_tx.send(Arc::clone(&child)); @@ -301,6 +426,9 @@ fn drain_sandboxed( .map_err(|_| AgentSessionError::Io("child lock empoisonné".to_owned()))? .wait() .map_err(|e| AgentSessionError::Io(e.to_string()))?; + if let Some(cancel) = &cancel { + cancel.clear_std(&child); + } let stderr_bytes = stderr_handle .join() .map_err(|_| AgentSessionError::Io("stderr thread terminé sans résultat".to_owned()))??; @@ -318,6 +446,7 @@ fn drain_sandboxed( async fn drain( spec: &SpawnLine, line_tap: Option>, + cancel: Option<&TurnCancel>, ) -> Result, AgentSessionError> { let mut cmd = Command::new(&spec.command); cmd.args(&spec.args) @@ -331,14 +460,20 @@ async fn drain( cmd.env(key, value); } - let mut child = cmd + let child = cmd .spawn() .map_err(|e| AgentSessionError::Start(format!("{}: {e}", spec.command)))?; + let child = Arc::new(AsyncMutex::new(child)); + if let Some(cancel) = cancel { + cancel.set_async(Arc::clone(&child)); + } // Pousse le prompt sur stdin puis le ferme (EOF) — sans quoi un process en mode // `stream-json` attend indéfiniment. Erreur d'I/O ⇒ `Io` (pas `Start`). if let Some(input) = &spec.stdin { let mut stdin = child + .lock() + .await .stdin .take() .ok_or_else(|| AgentSessionError::Io("stdin pipe indisponible".to_owned()))?; @@ -349,14 +484,18 @@ async fn drain( // Drop explicite ⇒ EOF côté process. drop(stdin); } else { - drop(child.stdin.take()); + drop(child.lock().await.stdin.take()); } let stdout = child + .lock() + .await .stdout .take() .ok_or_else(|| AgentSessionError::Io("stdout pipe indisponible".to_owned()))?; let mut stderr = child + .lock() + .await .stderr .take() .ok_or_else(|| AgentSessionError::Io("stderr pipe indisponible".to_owned()))?; @@ -384,7 +523,10 @@ async fn drain( // n'est pas en soi une erreur de session si la CLI a imprimé une sortie // structurée, captée ensuite par le parser. Sans stdout, il n'y a rien à parser : // on remonte le stderr réel au lieu de laisser l'appelant conclure à un no-Final. - let status = child.wait().await.map_err(map_io)?; + let status = child.lock().await.wait().await.map_err(map_io)?; + if let Some(cancel) = cancel { + cancel.clear_async(&child); + } let stderr_bytes = stderr_task .await .map_err(|e| AgentSessionError::Io(format!("lecture stderr: {e}")))? @@ -414,3 +556,40 @@ fn process_exit_message(command: &str, code: Option, stderr: &[u8]) -> Stri format!("{command} exited with {status}: {stderr}") } } + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Instant; + + #[tokio::test] + async fn run_turn_cancellable_kills_current_child_without_waiting_for_natural_exit() { + let spec = SpawnLine { + command: "/bin/sh".to_owned(), + args: vec![ + "-c".to_owned(), + "printf 'started\\n'; sleep 10; printf 'late\\n'".to_owned(), + ], + cwd: "/".to_owned(), + env: Vec::new(), + stdin: None, + sandbox: None, + }; + let cancel = TurnCancel::new(); + let started = Instant::now(); + let worker = tokio::spawn({ + let cancel = cancel.clone(); + async move { run_turn_cancellable(&spec, None, None, None, &cancel).await } + }); + + tokio::time::sleep(Duration::from_millis(100)).await; + cancel.cancel().await.expect("cancel current child"); + + let lines = worker.await.expect("worker joins").expect("drain result"); + assert_eq!(lines, vec!["started"]); + assert!( + started.elapsed() < Duration::from_secs(3), + "cancel should not wait for the child process natural exit" + ); + } +} diff --git a/frontend/src/features/agents/CustomAgentChatView.test.tsx b/frontend/src/features/agents/CustomAgentChatView.test.tsx index 458e7190..f61ffbeb 100644 --- a/frontend/src/features/agents/CustomAgentChatView.test.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.test.tsx @@ -856,28 +856,27 @@ describe("CustomAgentChatView", () => { expect(screen.queryByText("Le modèle n'a renvoyé aucune réponse.")).toBeNull(); }); - it("stops the live agent with cascade feedback and releases the structured session", async () => { - let resolveStop: (value: { - agentId: string; - sessionId: string; - kind: "structured"; - }) => void = () => {}; + it("cancels only the current request and keeps the structured session usable", async () => { + let resolveCancel: () => void = () => {}; const agent = { launchAgentChat: vi.fn(), reattachAgentChat: vi.fn(async (sessionId: string) => ({ sessionId, scrollback: [], })), - sendAgentChat: vi.fn(() => new Promise(() => {})), - stopLiveAgent: vi.fn( + sendAgentChat: vi + .fn() + .mockImplementationOnce(() => new Promise(() => {})) + .mockImplementationOnce(async (_sessionId: string, _prompt: string, onChunk) => { + onChunk({ kind: "final", content: "Next answer" }); + }), + stopLiveAgent: vi.fn(), + cancelAgentChat: vi.fn( () => - new Promise<{ agentId: string; sessionId: string; kind: "structured" }>( - (resolve) => { - resolveStop = resolve; - }, - ), + new Promise((resolve) => { + resolveCancel = resolve; + }), ), - cancelAgentChat: vi.fn(async () => {}), closeAgentChat: vi.fn(async () => {}), }; const onSessionId = vi.fn(); @@ -928,28 +927,35 @@ describe("CustomAgentChatView", () => { fireEvent.click(await screen.findByRole("button", { name: "Stop" })); - const stoppingLabel = await screen.findByText("Arrêt en cours..."); + const stoppingLabel = await screen.findByText("Stopping..."); expect(stoppingLabel.closest("button")?.hasAttribute("disabled")).toBe(true); + fireEvent.click(stoppingLabel.closest("button")!); await waitFor(() => - expect(agent.stopLiveAgent).toHaveBeenCalledWith("project-1", "agent-1"), + expect(agent.cancelAgentChat).toHaveBeenCalledWith("chat-session-1"), ); - resolveStop({ - agentId: "agent-1", - sessionId: "chat-session-1", - kind: "structured", - }); - await waitFor(() => expect(onSessionId).toHaveBeenCalledWith(null)); + expect(agent.cancelAgentChat).toHaveBeenCalledTimes(1); + resolveCancel(); + await waitFor(() => expect(screen.getByText("Request stopped.")).toBeTruthy()); expect(hasProjectCliInteraction("project-1")).toBe(false); - expect(agent.cancelAgentChat).not.toHaveBeenCalled(); + expect(onSessionId).not.toHaveBeenCalledWith(null); + expect(agent.stopLiveAgent).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); + const composer = screen.getByLabelText(/message CLI custom/); + expect(composer).toHaveProperty("disabled", false); + + fireEvent.change(composer, { + target: { value: "next turn" }, + }); + fireEvent.click(screen.getByRole("button", { name: "Envoyer" })); + + await waitFor(() => expect(agent.sendAgentChat).toHaveBeenCalledTimes(2)); + expect(agent.sendAgentChat).toHaveBeenNthCalledWith( + 2, + "chat-session-1", + "next turn", + expect.any(Function), + ); + expect(await screen.findByText("Next answer")).toBeTruthy(); }); it("keeps the composer locked from canonical work-state busy even without local stream activity", async () => { @@ -4848,7 +4854,7 @@ describe("CustomAgentChatView", () => { expect(screen.queryByText(/not found: structured session/)).toBeNull(); }); - it("treats NOT_FOUND during cancel as an already-gone session", async () => { + it("treats NOT_FOUND during request cancel as an already-stopped request", async () => { const agent = { launchAgentChat: vi.fn(), reattachAgentChat: vi.fn(async (sessionId: string) => ({ @@ -4856,11 +4862,11 @@ describe("CustomAgentChatView", () => { scrollback: [], })), sendAgentChat: vi.fn(() => new Promise(() => {})), - stopLiveAgent: vi.fn().mockRejectedValueOnce({ + stopLiveAgent: vi.fn(), + cancelAgentChat: vi.fn().mockRejectedValueOnce({ code: "NOT_FOUND", message: "agent agent-1 has no live session", }), - cancelAgentChat: vi.fn(async () => {}), closeAgentChat: vi.fn(async () => {}), }; const onSessionId = vi.fn(); @@ -4902,9 +4908,10 @@ describe("CustomAgentChatView", () => { fireEvent.click(await screen.findByRole("button", { name: "Stop" })); - expect(await screen.findByText("Session déjà arrêtée.")).toBeTruthy(); - expect(onSessionId).toHaveBeenCalledWith(null); - expect(agent.cancelAgentChat).not.toHaveBeenCalled(); + expect(await screen.findByText("Request already stopped.")).toBeTruthy(); + expect(onSessionId).not.toHaveBeenCalledWith(null); + expect(agent.cancelAgentChat).toHaveBeenCalledWith("gone-session"); + expect(agent.stopLiveAgent).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 c7a0529b..d3884eab 100644 --- a/frontend/src/features/agents/CustomAgentChatView.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.tsx @@ -1070,7 +1070,7 @@ export function CustomAgentChatView({ }); const delegationActive = requesterDelegation.delegationTargetName !== null; const effectiveBusy = - busy || requesterDelegation.currentAgentBusy || delegationActive; + busy || stopping || requesterDelegation.currentAgentBusy || delegationActive; const delegationFallbackText = requesterDelegation.delegationTargetName ? `${agentName} attend ${requesterDelegation.delegationTargetName}` : "Délégation en cours."; @@ -1982,21 +1982,19 @@ export function CustomAgentChatView({ } } - async function stopAgent() { + async function stopCurrentRequest() { const sid = sessionRef.current; - if (!sid || !agent.stopLiveAgent || stopping) return; + if (!sid || !agent.cancelAgentChat || stopping) return; setStopping(true); setError(null); try { - const stopped = await agent.stopLiveAgent(projectId, agentId); + await agent.cancelAgentChat(sid); if (activeInteractionIdRef.current) { markProjectCliInteractionStopped(projectId, activeInteractionIdRef.current); activeInteractionIdRef.current = null; } setBusy(false); setOpening(false); - setCurrentSession(null); - publishSessionId(null); updateChatViewState((prev) => { const completed = completeChatViewProgress(prev); return { @@ -2005,10 +2003,7 @@ export function CustomAgentChatView({ ...completed.turns, { 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é.", + label: "Request stopped.", receivedAtMs: Date.now(), }, ], @@ -2022,15 +2017,13 @@ export function CustomAgentChatView({ } setBusy(false); setOpening(false); - setCurrentSession(null); - publishSessionId(null); updateChatViewState((prev) => { const completed = completeChatViewProgress(prev); return { ...completed, turns: [ ...completed.turns, - { role: "tool", label: "Session déjà arrêtée.", receivedAtMs: Date.now() }, + { role: "tool", label: "Request already stopped.", receivedAtMs: Date.now() }, ], }; }); @@ -2043,7 +2036,7 @@ export function CustomAgentChatView({ ...prev.turns, { role: "error", - text: `Arrêt de l'agent échoué: ${describe(e)}`, + text: `Arrêt de la demande échoué: ${describe(e)}`, receivedAtMs: Date.now(), }, ], @@ -2417,13 +2410,15 @@ export function CustomAgentChatView({ className="shrink-0 whitespace-nowrap" disabled={ effectiveBusy - ? !supported || !currentSession || !agent.stopLiveAgent || stopping + ? !supported || !currentSession || !agent.cancelAgentChat || stopping : !canSend } loading={stopping} - onClick={() => (effectiveBusy ? void stopAgent() : void send())} + title={effectiveBusy ? "Arrêter la demande en cours" : undefined} + aria-label={effectiveBusy ? "Stop" : undefined} + onClick={() => (effectiveBusy ? void stopCurrentRequest() : void send())} > - {stopping ? "Arrêt en cours..." : effectiveBusy ? "Stop" : "Envoyer"} + {stopping ? "Stopping..." : effectiveBusy ? "Stop" : "Envoyer"}