diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 7ff722c8..10dd4862 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -48,26 +48,27 @@ use crate::dto::{ parse_ticket_id, save_model_server_input, AgentDriftListDto, AgentDto, AgentListDto, AllocateManagedTempRequestDto, AppExitWorkGuardStateDto, AssignSkillRequestDto, AttachBackgroundTaskResultDto, AttachLiveAgentRequestDto, AttachLiveAgentResponseDto, - BackgroundTaskDto, CellKind, ChangeAgentProfileDto, ChangeAgentProfileRequestDto, - ChatAttachmentDto, ChatAttachmentInputDto, ChatDraftScopeRequestDto, ChatPromptAttachmentDto, - CloneOpenCodeProfileFromSeedRequestDto, CloneProfileFromSeedRequestDto, - ConfigureProfilesRequestDto, ConversationDetailsDto, CreateAgentFromTemplateRequestDto, - CreateAgentRequestDto, CreateLayoutRequestDto, CreateLayoutResultDto, CreateMemoryRequestDto, - CreateProjectRequestDto, CreateSkillRequestDto, CreateTemplateRequestDto, - DeleteLayoutRequestDto, DeleteLayoutResultDto, DeliveredDelegationRequestDto, - DetectProfilesRequestDto, DetectProfilesResponseDto, EmbedderEnginesDto, EmbedderProfileDto, - EmbedderProfileListDto, ErrorDto, ExecuteSlashCommandRequestDto, - ExecuteSlashCommandResponseDto, FirstRunStateDto, FrontAttachedRequestDto, GitBranchesDto, - GitCheckoutRequestDto, GitCommitDto, GitCommitListDto, GitCommitRequestDto, GitStageRequestDto, - GitStatusListDto, GraphCommitListDto, HealthRequestDto, HealthResponseDto, - ImportChatAttachmentsRequestDto, ImportChatAttachmentsResponseDto, - InspectConversationRequestDto, InterruptAgentRequestDto, LaunchAgentRequestDto, LayoutDto, - LayoutOperationDto, ListLayoutsDto, LiveAgentListDto, ManagedTempGcResponseDto, - ManagedTempHandleRequestDto, ManagedTempLeaseDto, ManagedTempLeaseResponseDto, MemoryDto, - MemoryIndexDto, MemoryLinksDto, MemoryListDto, ModelServerConfigDto, ModelServerConfigListDto, - OpenCodeProviderListDto, OpenProjectRestoreStateDto, OpenTerminalRequestDto, - PreviewModelServerCommandDto, ProfileDto, ProfileListDto, ProfileModelCatalogDto, ProjectDto, - ProjectListDto, ProjectMcpToolPermissionsDto, ProjectPermissionsDto, ProjectSettingsDto, + BackgroundTaskDto, CancelAgentChatResponseDto, CellKind, ChangeAgentProfileDto, + ChangeAgentProfileRequestDto, ChatAttachmentDto, ChatAttachmentInputDto, + ChatDraftScopeRequestDto, ChatPromptAttachmentDto, CloneOpenCodeProfileFromSeedRequestDto, + CloneProfileFromSeedRequestDto, ConfigureProfilesRequestDto, ConversationDetailsDto, + CreateAgentFromTemplateRequestDto, CreateAgentRequestDto, CreateLayoutRequestDto, + CreateLayoutResultDto, CreateMemoryRequestDto, CreateProjectRequestDto, CreateSkillRequestDto, + CreateTemplateRequestDto, DeleteLayoutRequestDto, DeleteLayoutResultDto, + DeliveredDelegationRequestDto, DetectProfilesRequestDto, DetectProfilesResponseDto, + EmbedderEnginesDto, EmbedderProfileDto, EmbedderProfileListDto, ErrorDto, + ExecuteSlashCommandRequestDto, ExecuteSlashCommandResponseDto, FirstRunStateDto, + FrontAttachedRequestDto, GitBranchesDto, GitCheckoutRequestDto, GitCommitDto, GitCommitListDto, + GitCommitRequestDto, GitStageRequestDto, GitStatusListDto, GraphCommitListDto, + HealthRequestDto, HealthResponseDto, ImportChatAttachmentsRequestDto, + ImportChatAttachmentsResponseDto, InspectConversationRequestDto, InterruptAgentRequestDto, + LaunchAgentRequestDto, LayoutDto, LayoutOperationDto, ListLayoutsDto, LiveAgentListDto, + ManagedTempGcResponseDto, ManagedTempHandleRequestDto, ManagedTempLeaseDto, + ManagedTempLeaseResponseDto, MemoryDto, MemoryIndexDto, MemoryLinksDto, MemoryListDto, + ModelServerConfigDto, ModelServerConfigListDto, OpenCodeProviderListDto, + OpenProjectRestoreStateDto, OpenTerminalRequestDto, PreviewModelServerCommandDto, ProfileDto, + ProfileListDto, ProfileModelCatalogDto, ProjectDto, ProjectListDto, + ProjectMcpToolPermissionsDto, ProjectPermissionsDto, ProjectSettingsDto, ProjectSystemPermissionsDto, ProjectWorkStateDto, ReadAgentContextResponseDto, ReadChatDraftResponseDto, ReadConversationPageRequestDto, ReattachChatDto, ReattachResultDto, RecallMemoryRequestDto, RemoveOpenProjectRestoreEntryRequestDto, RemoveProjectRequestDto, @@ -2855,18 +2856,29 @@ pub async fn agent_send( .structured_sessions .session(&sid) .ok_or_else(|| ErrorDto::from(AppError::NotFound(format!("structured session {sid}"))))?; + let turn_generation = state + .structured_sessions + .begin_turn(&sid) + .ok_or_else(|| ErrorDto::from(AppError::NotFound(format!("structured session {sid}"))))?; // (Re-)register the reply channel; bumps the generation so an earlier attach's // pump (if any) is superseded and stops delivering to its stale channel. let gen = state.chat_bridge.register(sid, on_reply); - let imported_attachments = import_chat_attachments_for_session( + let imported_attachments = match import_chat_attachments_for_session( &state, &sid, attachment_paths.unwrap_or_default(), attachments.unwrap_or_default(), ) - .await?; + .await + { + Ok(attachments) => attachments, + Err(error) => { + state.structured_sessions.finish_turn(&sid, turn_generation); + return Err(error); + } + }; let attachment_refs = chat_prompt_attachment_refs(&imported_attachments); let prompt_for_model = prompt_with_imported_attachments(&prompt, &imported_attachments, &attachment_refs); @@ -2910,10 +2922,14 @@ pub async fn agent_send( let tap_bridge = std::sync::Arc::clone(&state.chat_bridge); let tap_clock = std::sync::Arc::clone(&state.clock); let tap_sid = sid; + let tap_structured = std::sync::Arc::clone(&state.structured_sessions); let tap_turn_guard = turn_guard.as_ref().map(|guard| guard.liveness_handle()); let tap_pump = std::thread::spawn(move || { let mut settler = crate::chat::ProgressSettler::default(); for event in tap_rx { + if !tap_structured.accepts_turn_output(&tap_sid, turn_generation) { + break; + } if !matches!(event, domain::ports::ReplyEvent::Final { .. }) { if let Some(guard) = &tap_turn_guard { guard.mark_alive(); @@ -2929,11 +2945,25 @@ pub async fn agent_send( } let _ = tap_bridge.send_output_settled(&tap_sid, &mut settler, chunk); } - if let Some(completed) = settler.finish() { - let _ = tap_bridge.send_output(&tap_sid, completed); + if tap_structured.accepts_turn_output(&tap_sid, turn_generation) { + if let Some(completed) = settler.finish() { + let _ = tap_bridge.send_output(&tap_sid, completed); + } } }); + if !state + .structured_sessions + .accepts_turn_output(&sid, turn_generation) + { + drop(tap_tx); + let _ = tap_pump.join(); + if let Some(guard) = &turn_guard { + guard.finish(); + } + return Ok(()); + } + // Open the turn stream. A start failure leaves the just-registered channel in // place (the cell stays attached, ready for a retry) — mirrors the PTY pump, // which only unregisters on a hard subscribe failure; here the session is @@ -2941,6 +2971,16 @@ pub async fn agent_send( let stream_result = session.send_with_tap(&prompt_for_model, tap_tx).await; let _ = tap_pump.join(); let mut turn_guard = turn_guard; + if !state + .structured_sessions + .accepts_turn_output(&sid, turn_generation) + { + if let Some(guard) = &turn_guard { + guard.finish(); + } + state.chat_bridge.detach_if(&sid, gen); + return Ok(()); + } let stream = match stream_result { Ok(stream) => stream, Err(_error) @@ -2952,7 +2992,10 @@ pub async fn agent_send( // logical turn. The detached drain below waits for the replacement. Box::new(std::iter::empty()) as domain::ports::ReplyStream } - Err(error) => return Err(ErrorDto::from(AppError::from(error))), + Err(error) => { + state.structured_sessions.finish_turn(&sid, turn_generation); + return Err(ErrorDto::from(AppError::from(error))); + } }; // Drain the blocking reply iterator on a dedicated OS thread (the stream is a @@ -2978,6 +3021,10 @@ pub async fn agent_send( let mut resume = false; let mut resume_prompt = None; for event in current_stream { + if !structured.accepts_turn_output(&sid, turn_generation) { + terminal_visible = true; + break 'logical_turn; + } let provider_terminal = matches!( event, domain::ports::ReplyEvent::Final { .. } @@ -3126,7 +3173,7 @@ pub async fn agent_send( } } } - if !terminal_visible { + if !terminal_visible && structured.accepts_turn_output(&sid, turn_generation) { let chunk = crate::chat::with_chunk_timestamp( crate::chat::unfinished_turn_chunk(), u64::try_from(clock.now_millis()).ok(), @@ -3136,6 +3183,7 @@ pub async fn agent_send( if let Some(guard) = &turn_guard { guard.finish(); } + structured.finish_turn(&sid, turn_generation); bridge.detach_if(&sid, gen); }); @@ -3512,6 +3560,7 @@ pub async fn interrupt_agent( return session .cancel_current_turn() .await + .map(|_| ()) .map_err(|e| ErrorDto::from(AppError::from(e))); } } @@ -3524,11 +3573,10 @@ pub async fn interrupt_agent( .map_err(ErrorDto::from) } -/// `cancel_agent_chat` — stop a live structured chat session and cascade its -/// running execution children best-effort. +/// `cancel_agent_chat` — cancel only the current request of a live structured chat. /// -/// This is the session-id based twin of [`stop_live_agent`] for a custom chat -/// surface that already owns the structured `sessionId`. +/// The reusable session, conversation bridge and runtime state remain intact; +/// agent-level teardown stays exclusively on [`stop_live_agent`]. /// /// # Errors /// Returns an [`ErrorDto`] (`INVALID` for a malformed id, `NOT_FOUND` if no live @@ -3537,30 +3585,26 @@ pub async fn interrupt_agent( pub async fn cancel_agent_chat( session_id: String, state: State<'_, AppState>, -) -> Result<(), ErrorDto> { +) -> Result { let sid = parse_session_id(&session_id)?; let (project_id, agent_id, _, _) = state .structured_sessions .meta_for_session(&sid) .ok_or_else(|| ErrorDto::from(AppError::NotFound(format!("structured session {sid}"))))?; - 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, - reason: Some("fermeture de la session chat".to_owned()), - }) + let output = state + .cancel_agent_chat + .execute(sid) .await - .map(|_| ()) - .map_err(ErrorDto::from) + .map_err(ErrorDto::from)?; + if output.outcome == domain::ports::CancelCurrentTurnOutcome::Cancelled { + let project = resolve_project(&project_id.to_string(), &state).await?; + state + .orchestrator_service + .cancel_agent_turn_execution_children(&project, agent_id) + .await + .map_err(ErrorDto::from)?; + } + Ok(CancelAgentChatResponseDto::from(output)) } async fn preferred_view_for_agent( diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index 5a193e9b..293c76df 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -263,10 +263,11 @@ pub use template::{ SyncAgentWithTemplateOutput, UpdateTemplate, UpdateTemplateInput, UpdateTemplateOutput, }; pub use terminal::{ - CloseTerminal, CloseTerminalInput, CloseTerminalOutput, LiveAgentRegistry, LiveRuntimeState, - LiveRuntimeStatus, LiveSessionKind, LiveSessionSnapshot, LiveSessions, OpenTerminal, - OpenTerminalInput, OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput, StructuredSessions, - TerminalSessions, WriteToTerminal, WriteToTerminalInput, + CancelAgentChat, CancelAgentChatOutput, CloseTerminal, CloseTerminalInput, CloseTerminalOutput, + LiveAgentRegistry, LiveRuntimeState, LiveRuntimeStatus, LiveSessionKind, LiveSessionSnapshot, + LiveSessions, OpenTerminal, OpenTerminalInput, OpenTerminalOutput, ResizeTerminal, + ResizeTerminalInput, StructuredSessions, TerminalSessions, WriteToTerminal, + WriteToTerminalInput, }; pub use ticket_assistant::{ CloseTicketAssistant, CloseTicketAssistantInput, OpenTicketAssistant, OpenTicketAssistantInput, diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index b833cf02..4984c1f9 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -1672,6 +1672,105 @@ impl OrchestratorService { descendants } + /// Cancels resources owned by the current turn without closing runtime admission + /// or removing any reusable agent session. + pub async fn cancel_agent_turn_execution_children( + &self, + project: &Project, + agent_id: AgentId, + ) -> Result<(), AppError> { + let root = runtime_key(project, agent_id); + let mut keys = vec![root]; + let mut participants = HashSet::from([root]); + { + let control = self + .execution_control + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + while let Some(parent) = keys.pop() { + for child in control + .launched_by + .iter() + .filter_map(|(child, owner)| (*owner == parent).then_some(*child)) + .collect::>() + { + if participants.insert(child) { + keys.push(child); + } + } + } + } + participants.extend( + self.active_wait_dependencies_from_wait_edges(agent_id) + .into_iter() + .map(|agent| runtime_key(project, agent)), + ); + + let rendezvous_targets: Vec<_> = self + .active_rendezvous + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .iter() + .filter_map(|(target, active)| { + (participants.contains(target) + || active.link.requester_agent_id.is_some_and(|requester| { + participants + .iter() + .any(|participant| participant.agent_id == requester) + })) + .then_some(*target) + }) + .collect(); + for target in rendezvous_targets { + self.restart_rendezvous.cancel(target); + } + + let mut first_error = None; + for participant in &participants { + self.restart_rendezvous.cancel(*participant); + if let Some(session_limits) = &self.session_limits { + session_limits.cancel_resume(participant.agent_id); + } + if let Some(input) = &self.input { + input.preempt(*participant); + } + if let Some(structured) = &self.structured { + if participant.agent_id != agent_id { + if let Some(session) = + structured.session_for_agent_in_project(project.id, &participant.agent_id) + { + if let Err(error) = session.cancel_current_turn().await { + first_error.get_or_insert_with(|| AppError::from(error)); + } + } + } + } + } + + let (Some(store), Some(runner), Some(clock)) = + (&self.background_tasks, &self.background_runner, &self.clock) + else { + return first_error.map_or(Ok(()), Err); + }; + let mut cancel = crate::background::CancelBackgroundTask::new( + Arc::clone(store), + Arc::clone(runner), + Arc::clone(clock), + ); + if let Some(events) = &self.events { + cancel = cancel.with_events(Arc::clone(events)); + } + for participant in participants { + if let Err(error) = cancel + .cancel_subtree_for_agent(project.id, participant.agent_id) + .await + { + first_error.get_or_insert(error); + } + } + first_error.map_or(Ok(()), Err) + } + /// Returns the transitive dependencies of `agent` in an active wait-edge snapshot. #[must_use] fn active_wait_dependencies_from_edges( diff --git a/crates/application/src/terminal/mod.rs b/crates/application/src/terminal/mod.rs index 7dfa5e7e..a2a77c26 100644 --- a/crates/application/src/terminal/mod.rs +++ b/crates/application/src/terminal/mod.rs @@ -33,7 +33,8 @@ pub use registry::{ LiveSessions, StructuredSessions, TerminalSessions, }; pub use usecases::{ - stop_live_agent_runtime_in_project, CloseTerminal, CloseTerminalInput, CloseTerminalOutput, - OpenTerminal, OpenTerminalInput, OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput, - StoppedAgentRuntime, WriteToTerminal, WriteToTerminalInput, + stop_live_agent_runtime_in_project, CancelAgentChat, CancelAgentChatOutput, CloseTerminal, + CloseTerminalInput, CloseTerminalOutput, OpenTerminal, OpenTerminalInput, OpenTerminalOutput, + ResizeTerminal, ResizeTerminalInput, StoppedAgentRuntime, WriteToTerminal, + WriteToTerminalInput, }; diff --git a/crates/application/src/terminal/registry.rs b/crates/application/src/terminal/registry.rs index 928f4c32..d0cec2cc 100644 --- a/crates/application/src/terminal/registry.rs +++ b/crates/application/src/terminal/registry.rs @@ -537,6 +537,17 @@ struct StructuredEntry { managed_temp_handle: Option, /// Optional local-model-server usage guard retained for the live session. _model_server_guard: Option, + /// Monotonic logical-turn generation used to fence late provider events. + turn_generation: u64, + /// Current/last logical-turn state for idempotent request cancellation. + turn_state: StructuredTurnState, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum StructuredTurnState { + Completed, + Active(u64), + Cancelled(u64), } #[derive(Clone)] @@ -660,6 +671,8 @@ impl StructuredSessions { transcript_restore_expected: false, managed_temp_handle, _model_server_guard: model_server_guard, + turn_generation: 0, + turn_state: StructuredTurnState::Completed, }, ); } @@ -692,6 +705,75 @@ impl StructuredSessions { }) } + /// Starts a new logical human turn and returns its fencing generation. + pub fn begin_turn(&self, id: &SessionId) -> Option { + self.entries.lock().ok().and_then(|mut entries| { + let entry = entries.get_mut(id)?; + entry.turn_generation = entry.turn_generation.wrapping_add(1).max(1); + let generation = entry.turn_generation; + entry.turn_state = StructuredTurnState::Active(generation); + Some(generation) + }) + } + + /// Atomically records a cancellation request and returns the live session. + pub fn request_turn_cancel( + &self, + id: &SessionId, + ) -> Option<( + Arc, + domain::ports::CancelCurrentTurnOutcome, + )> { + self.entries.lock().ok().and_then(|mut entries| { + let entry = entries.get_mut(id)?; + let outcome = match entry.turn_state { + StructuredTurnState::Active(generation) => { + entry.turn_state = StructuredTurnState::Cancelled(generation); + domain::ports::CancelCurrentTurnOutcome::Cancelled + } + StructuredTurnState::Cancelled(_) => { + domain::ports::CancelCurrentTurnOutcome::AlreadyCancelled + } + StructuredTurnState::Completed => { + domain::ports::CancelCurrentTurnOutcome::AlreadyCompleted + } + }; + Some((Arc::clone(&entry.session), outcome)) + }) + } + + /// True while `generation` is still the uncancelled current turn. + #[must_use] + pub fn accepts_turn_output(&self, id: &SessionId, generation: u64) -> bool { + self.entries + .lock() + .ok() + .and_then(|entries| entries.get(id).map(|entry| entry.turn_state)) + == Some(StructuredTurnState::Active(generation)) + } + + /// Retires a logical turn without allowing an older generation to affect a newer one. + pub fn finish_turn(&self, id: &SessionId, generation: u64) { + if let Ok(mut entries) = self.entries.lock() { + if let Some(entry) = entries.get_mut(id) { + if entry.turn_state == StructuredTurnState::Active(generation) { + entry.turn_state = StructuredTurnState::Completed; + } + } + } + } + + /// Reopens the same generation when its concrete cancellation failed. + pub fn restore_turn_after_cancel_failure(&self, id: &SessionId) { + if let Ok(mut entries) = self.entries.lock() { + if let Some(entry) = entries.get_mut(id) { + if let StructuredTurnState::Cancelled(generation) = entry.turn_state { + entry.turn_state = StructuredTurnState::Active(generation); + } + } + } + } + /// Records whether this runtime was opened from an existing logical conversation. pub fn set_transcript_restore_expected(&self, id: SessionId, expected: bool) { if let Ok(mut entries) = self.entries.lock() { diff --git a/crates/application/src/terminal/usecases.rs b/crates/application/src/terminal/usecases.rs index f9367295..cfb92ee8 100644 --- a/crates/application/src/terminal/usecases.rs +++ b/crates/application/src/terminal/usecases.rs @@ -2,7 +2,7 @@ use std::sync::Arc; -use domain::ports::{EventBus, PtyPort, SpawnSpec}; +use domain::ports::{CancelCurrentTurnOutcome, EventBus, PtyPort, SpawnSpec}; use domain::{ AgentId, DomainEvent, NodeId, Project, ProjectPath, PtySize, SessionId, SessionKind, SessionStatus, TerminalSession, @@ -11,7 +11,54 @@ use domain::{ use crate::error::AppError; use crate::managed_temp::{ReleaseManagedTemp, ReleaseManagedTempInput}; -use super::registry::{LiveRuntimeState, LiveSessionKind, LiveSessions, TerminalSessions}; +use super::registry::{ + LiveRuntimeState, LiveSessionKind, LiveSessions, StructuredSessions, TerminalSessions, +}; + +/// Result of cancelling the current request while preserving its structured session. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CancelAgentChatOutput { + /// Stable session that remains registered and reusable. + pub session_id: SessionId, + /// Whether this call cancelled work or observed an already-terminal turn. + pub outcome: CancelCurrentTurnOutcome, +} + +/// Cancels one structured chat turn without stopping its reusable runtime. +pub struct CancelAgentChat { + sessions: Arc, +} + +impl CancelAgentChat { + /// Builds the use case from the structured runtime registry. + #[must_use] + pub fn new(sessions: Arc) -> Self { + Self { sessions } + } + + /// Requests cancellation from the live session and deliberately leaves its + /// registry entry, conversation identity and runtime state untouched. + /// + /// # Errors + /// Returns [`AppError::NotFound`] only when the session itself is absent, and + /// a process error when the adapter could not interrupt its current child. + pub async fn execute(&self, session_id: SessionId) -> Result { + let (session, outcome) = self + .sessions + .request_turn_cancel(&session_id) + .ok_or_else(|| AppError::NotFound(format!("structured session {session_id}")))?; + if outcome == CancelCurrentTurnOutcome::Cancelled { + if let Err(error) = session.cancel_current_turn().await { + self.sessions.restore_turn_after_cancel_failure(&session_id); + return Err(AppError::from(error)); + } + } + Ok(CancelAgentChatOutput { + session_id, + outcome, + }) + } +} /// Input for [`OpenTerminal::execute`]. #[derive(Debug, Clone, PartialEq, Eq)] @@ -281,8 +328,9 @@ pub struct StoppedAgentRuntime { /// Stops one agent's live runtime across both registries. /// /// PTY sessions are closed through [`CloseTerminal`]. Structured sessions are -/// removed from their registry before `shutdown()` so relaunch guards stop seeing -/// them as live even while shutdown runs. +/// removed from their registry, have their current turn cancelled, then receive +/// `shutdown()` so relaunch guards stop seeing them as live during teardown and +/// no per-turn child survives it. pub async fn stop_live_agent_runtime_in_project( project: &Project, agent_id: AgentId, @@ -298,6 +346,9 @@ pub async fn stop_live_agent_runtime_in_project( let removed = match snapshot.kind { LiveSessionKind::Structured => { if let Some(removed) = live.structured.remove_runtime(&snapshot.session_id) { + if let Err(error) = removed.session.cancel_current_turn().await { + first_error.get_or_insert_with(|| AppError::Process(error.to_string())); + } if let Err(error) = removed.session.shutdown().await { first_error.get_or_insert_with(|| AppError::Process(error.to_string())); } diff --git a/crates/application/src/workstate/actions.rs b/crates/application/src/workstate/actions.rs index 352da2da..b7be3603 100644 --- a/crates/application/src/workstate/actions.rs +++ b/crates/application/src/workstate/actions.rs @@ -245,13 +245,12 @@ pub trait StopAgentExecutionCascade: Send + Sync { ) -> Result<(), AppError>; } -/// Stops an already-running agent's current work. +/// Destructively stops an already-running agent runtime. /// -/// 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). +/// Both structured and PTY runtimes are removed. Structured sessions first +/// receive `cancel_current_turn()` so a per-turn child cannot survive teardown, +/// then `shutdown()` is called. Conversation-level cancellation uses the +/// separate [`crate::CancelAgentChat`] use case and never enters this path. pub struct StopAgentRuntime { live: Arc, close: Arc, @@ -317,7 +316,7 @@ impl StopAgentRuntime { self } - /// Stops the agent's current work. + /// Stops the agent's complete live runtime. /// /// # Errors /// Repeated stops are successful. A cancellation or runtime failure is returned diff --git a/crates/application/tests/cancel_agent_chat.rs b/crates/application/tests/cancel_agent_chat.rs new file mode 100644 index 00000000..d5898517 --- /dev/null +++ b/crates/application/tests/cancel_agent_chat.rs @@ -0,0 +1,142 @@ +use std::collections::VecDeque; +use std::sync::{Arc, Mutex}; + +use application::{ + CancelAgentChat, LiveRuntimeState, LiveSessions, StructuredSessions, TerminalSessions, +}; +use async_trait::async_trait; +use domain::ports::{AgentSession, AgentSessionError, CancelCurrentTurnOutcome, ReplyStream}; +use domain::{AgentId, NodeId, ProjectId, SessionId}; +use uuid::Uuid; + +struct FakeSession { + id: SessionId, + cancellations: Mutex>, + sends: Mutex>, + shutdowns: Mutex, +} + +impl FakeSession { + fn new(id: SessionId, outcomes: impl IntoIterator) -> Self { + Self { + id, + cancellations: Mutex::new(outcomes.into_iter().collect()), + sends: Mutex::new(Vec::new()), + shutdowns: Mutex::new(0), + } + } +} + +#[async_trait] +impl AgentSession for FakeSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + Some("conversation-kept".to_owned()) + } + + async fn send(&self, prompt: &str) -> Result { + self.sends.lock().unwrap().push(prompt.to_owned()); + Ok(Box::new(std::iter::empty())) + } + + async fn cancel_current_turn(&self) -> Result { + Ok(self + .cancellations + .lock() + .unwrap() + .pop_front() + .unwrap_or(CancelCurrentTurnOutcome::AlreadyCompleted)) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + *self.shutdowns.lock().unwrap() += 1; + Ok(()) + } +} + +fn id(value: u128, make: impl FnOnce(Uuid) -> T) -> T { + make(Uuid::from_u128(value)) +} + +#[tokio::test] +async fn cancel_keeps_session_and_accepts_a_second_prompt_without_shutdown() { + let session_id = id(1, SessionId::from_uuid); + let project_id = id(2, ProjectId::from_uuid); + let agent_id = id(3, AgentId::from_uuid); + let node_id = id(4, NodeId::from_uuid); + let session = Arc::new(FakeSession::new( + session_id, + [CancelCurrentTurnOutcome::Cancelled], + )); + let sessions = Arc::new(StructuredSessions::new()); + sessions.insert_in_project(project_id, session.clone(), agent_id, node_id); + let live = LiveSessions::new(Arc::new(TerminalSessions::new()), Arc::clone(&sessions)); + let cancel = CancelAgentChat::new(Arc::clone(&sessions)); + let first_generation = sessions.begin_turn(&session_id).unwrap(); + assert!(sessions.accepts_turn_output(&session_id, first_generation)); + + let output = cancel.execute(session_id).await.unwrap(); + + assert_eq!(output.outcome, CancelCurrentTurnOutcome::Cancelled); + assert!(!sessions.accepts_turn_output(&session_id, first_generation)); + let retained = sessions.session(&session_id).expect("session is retained"); + let second_generation = sessions.begin_turn(&session_id).unwrap(); + assert!(sessions.accepts_turn_output(&session_id, second_generation)); + sessions.finish_turn(&session_id, first_generation); + assert!( + sessions.accepts_turn_output(&session_id, second_generation), + "an old turn cannot retire or fence its successor" + ); + let _second_reply = retained.send("second prompt").await.unwrap(); + assert_eq!(*session.sends.lock().unwrap(), ["second prompt"]); + assert_eq!(*session.shutdowns.lock().unwrap(), 0); + assert_eq!( + live.runtime_status_for(project_id, &agent_id) + .unwrap() + .state, + LiveRuntimeState::Running + ); + assert_eq!( + sessions.meta_for_session(&session_id).unwrap().3.as_deref(), + Some("conversation-kept") + ); +} + +#[tokio::test] +async fn double_cancel_is_typed_and_missing_session_remains_distinct() { + let session_id = id(10, SessionId::from_uuid); + let session = Arc::new(FakeSession::new( + session_id, + [ + CancelCurrentTurnOutcome::Cancelled, + CancelCurrentTurnOutcome::AlreadyCancelled, + ], + )); + let sessions = Arc::new(StructuredSessions::new()); + sessions.insert_in_project( + id(11, ProjectId::from_uuid), + session, + id(12, AgentId::from_uuid), + id(13, NodeId::from_uuid), + ); + let cancel = CancelAgentChat::new(Arc::clone(&sessions)); + sessions.begin_turn(&session_id).unwrap(); + + assert_eq!( + cancel.execute(session_id).await.unwrap().outcome, + CancelCurrentTurnOutcome::Cancelled + ); + assert_eq!( + cancel.execute(session_id).await.unwrap().outcome, + CancelCurrentTurnOutcome::AlreadyCancelled + ); + let error = cancel + .execute(id(99, SessionId::from_uuid)) + .await + .unwrap_err(); + assert_eq!(error.code(), "NOT_FOUND"); + assert!(sessions.session(&session_id).is_some()); +} diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index ce310bc7..ba71443d 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -44,10 +44,10 @@ use domain::{Memory, MemoryIndexEntry, MemoryLink, MemorySlug}; use uuid::Uuid; use application::{ - CloseTerminal, CreateAgentFromScratch, CreateSkill, GetLiveStateLean, HarvestMemoryFromTurn, - LaunchAgent, ListAgents, LiveStateProvider, LiveStateReadProvider, OrchestratorService, - SessionLimitService, SpawnBackgroundCommand, TerminalSessions, UpdateAgentContext, - UpdateLiveState, + CancelAgentChat, CloseTerminal, CreateAgentFromScratch, CreateSkill, GetLiveStateLean, + HarvestMemoryFromTurn, LaunchAgent, ListAgents, LiveStateProvider, LiveStateReadProvider, + OrchestratorService, SessionLimitService, SpawnBackgroundCommand, TerminalSessions, + UpdateAgentContext, UpdateLiveState, }; use domain::live_state::{LiveEntry, LiveState, WorkStatus}; use domain::ports::LiveStateStore; @@ -1428,7 +1428,11 @@ async fn stop_agent_closes_structured_runtime_and_is_idempotent() { fx.pty.kills().is_empty(), "structured stop must not kill PTY" ); - assert_eq!(*cancel_called.lock().unwrap(), 0); + assert_eq!( + *cancel_called.lock().unwrap(), + 1, + "structured hard stop must cancel the live turn before shutdown" + ); assert_eq!( *shutdown_called.lock().unwrap(), 1, @@ -1453,6 +1457,46 @@ async fn stop_agent_closes_structured_runtime_and_is_idempotent() { assert_eq!(*shutdown_called.lock().unwrap(), 1); } +#[tokio::test] +async fn cancel_chat_keeps_structured_runtime_and_execution_admission_open() { + 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_id = sid(9343); + let session = Arc::new(CompletionSession { + id: session_id, + 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(9343)); + fx.structured.begin_turn(&session_id).unwrap(); + + let output = CancelAgentChat::new(Arc::clone(&fx.structured)) + .execute(session_id) + .await + .expect("turn cancel succeeds"); + fx.service + .cancel_agent_turn_execution_children(&project(), agent.id) + .await + .expect("child cancellation succeeds"); + + assert_eq!( + output.outcome, + domain::ports::CancelCurrentTurnOutcome::Cancelled + ); + assert_eq!(*cancel_called.lock().unwrap(), 1); + assert_eq!(*shutdown_called.lock().unwrap(), 0); + assert!(fx.structured.session(&session_id).is_some()); + fx.service + .ensure_agent_execution_active(&project(), agent.id) + .expect("turn cancellation must not close #320 admission"); +} + #[tokio::test] async fn stop_agent_from_unrelated_agent_is_rejected() { let agent = scratch_agent(aid(1), "dev-backend", "agents/dev-backend.md"); @@ -2117,11 +2161,13 @@ impl AgentSession for CompletionSession { )), } } - async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> { + async fn cancel_current_turn( + &self, + ) -> Result { if let Some(cancel_called) = &self.cancel_called { *cancel_called.lock().unwrap() += 1; } - Ok(()) + Ok(domain::ports::CancelCurrentTurnOutcome::Cancelled) } async fn shutdown(&self) -> Result<(), AgentSessionError> { if let Some(shutdown_called) = &self.shutdown_called { @@ -4999,8 +5045,10 @@ impl AgentSession for RestartingStructuredSession { Ok(()) } - async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> { - Ok(()) + async fn cancel_current_turn( + &self, + ) -> Result { + Ok(domain::ports::CancelCurrentTurnOutcome::AlreadyCompleted) } } diff --git a/crates/application/tests/workstate_actions.rs b/crates/application/tests/workstate_actions.rs index 9a9c7034..c1c953f2 100644 --- a/crates/application/tests/workstate_actions.rs +++ b/crates/application/tests/workstate_actions.rs @@ -117,11 +117,17 @@ 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> { + async fn cancel_current_turn( + &self, + ) -> Result { self.cancel_called.store(true, Ordering::SeqCst); - Ok(()) + Ok(domain::ports::CancelCurrentTurnOutcome::Cancelled) } async fn shutdown(&self) -> Result<(), AgentSessionError> { + assert!( + self.cancel_called.load(Ordering::SeqCst), + "shutdown must be preceded by cancel_current_turn" + ); self.shutdown_called.store(true, Ordering::SeqCst); Ok(()) } @@ -530,7 +536,7 @@ async fn stop_structured_shuts_down_and_removes_session() { assert_eq!(out.stopped[0].session_id, sid(2)); assert_eq!(out.stopped[0].kind, LiveSessionKind::Structured); assert_eq!(out.stopped[0].state, LiveRuntimeState::Stopped); - assert!(!cancel_called.load(Ordering::SeqCst)); + assert!(cancel_called.load(Ordering::SeqCst)); assert!( shutdown_called.load(Ordering::SeqCst), "session.shutdown() must be called" @@ -567,8 +573,8 @@ async fn stop_structured_runs_execution_cascade_before_shutdown() { assert_eq!(out.stopped[0].session_id, sid(2)); assert_eq!(*cascade.calls.lock().unwrap(), vec![(project().id, a)]); assert!( - !cancel_called.load(Ordering::SeqCst), - "session.cancel_current_turn() is not the runtime stop primitive" + cancel_called.load(Ordering::SeqCst), + "hard stop must cancel the current turn before shutdown" ); assert!( shutdown_called.load(Ordering::SeqCst), diff --git a/crates/backend/src/dto.rs b/crates/backend/src/dto.rs index b70f1999..34d0baac 100644 --- a/crates/backend/src/dto.rs +++ b/crates/backend/src/dto.rs @@ -12,16 +12,19 @@ use serde_json::Value; use application::{ AgentBackgroundTaskState, AgentTicketState, AppError, AppExitWorkGuardDetail, - AppExitWorkGuardState, AttachLiveAgentOutput, BackgroundTaskKindLabel, CellAgentPresentation, - ConversationPreviewStatus, ConversationTurnWorkPreview, ConversationWorkSummary, - CreateProjectInput, CreateProjectOutput, ExecuteSlashCommandOutput, GitGraphOutput, - HealthInput, HealthReport, LayoutKind, ListProjectsOutput, ListSlashCommandsOutput, - LiveRuntimeState, LiveSessionKind, LiveSessionSnapshot, OpenProjectOutput, ProjectWorkState, - RestartCellAgentRuntimeOutput, RestoreOpenProjectFailure, RestoreOpenProjectsOutput, - SlashCommandEffect, StopLiveAgentOutput, TicketWorkSource, TicketWorkStatus, TurnPage, - TurnSource, TurnView, + AppExitWorkGuardState, AttachLiveAgentOutput, BackgroundTaskKindLabel, CancelAgentChatOutput, + CellAgentPresentation, ConversationPreviewStatus, ConversationTurnWorkPreview, + ConversationWorkSummary, CreateProjectInput, CreateProjectOutput, ExecuteSlashCommandOutput, + GitGraphOutput, HealthInput, HealthReport, LayoutKind, ListProjectsOutput, + ListSlashCommandsOutput, LiveRuntimeState, LiveSessionKind, LiveSessionSnapshot, + OpenProjectOutput, ProjectWorkState, RestartCellAgentRuntimeOutput, RestoreOpenProjectFailure, + RestoreOpenProjectsOutput, SlashCommandEffect, StopLiveAgentOutput, TicketWorkSource, + TicketWorkStatus, TurnPage, TurnSource, TurnView, +}; +use domain::ports::{ + CancelCurrentTurnOutcome, ReplyProgress, ReplyProgressKind, ReplyProgressSource, + ReplyProgressStage, }; -use domain::ports::{ReplyProgress, ReplyProgressKind, ReplyProgressSource, ReplyProgressStage}; use domain::{ AgentBusyState, PageCursor, PageDirection, Project, ProjectId, ProjectSystemPermissions, ResolvedAgentSystemPermissions, SlashCommand, SystemPermissionSet, TurnRole, @@ -4763,6 +4766,45 @@ pub struct StopLiveAgentResponseDto { pub already_stopped: bool, } +/// Response DTO for a non-destructive `cancel_agent_chat` request. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct CancelAgentChatResponseDto { + /// Stable structured session, still live after the operation. + pub session_id: String, + /// Idempotent turn-level result. + pub outcome: CancelAgentChatOutcomeDto, +} + +/// Public cancellation outcomes, distinct from a missing-session error. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub enum CancelAgentChatOutcomeDto { + /// A running request was cancelled. + Cancelled, + /// That request had already been cancelled. + AlreadyCancelled, + /// The live session currently had no running request. + AlreadyCompleted, +} + +impl From for CancelAgentChatResponseDto { + fn from(output: CancelAgentChatOutput) -> Self { + Self { + session_id: output.session_id.to_string(), + outcome: match output.outcome { + CancelCurrentTurnOutcome::Cancelled => CancelAgentChatOutcomeDto::Cancelled, + CancelCurrentTurnOutcome::AlreadyCancelled => { + CancelAgentChatOutcomeDto::AlreadyCancelled + } + CancelCurrentTurnOutcome::AlreadyCompleted => { + CancelAgentChatOutcomeDto::AlreadyCompleted + } + }, + } + } +} + /// One stopped runtime in an agent-level stop response. #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] @@ -6298,6 +6340,22 @@ mod tests { use serde_json::json; use uuid::Uuid; + #[test] + fn cancel_agent_chat_outcomes_are_stable_camel_case_values() { + let response = CancelAgentChatResponseDto::from(CancelAgentChatOutput { + session_id: domain::SessionId::from_uuid(Uuid::from_u128(343)), + outcome: CancelCurrentTurnOutcome::AlreadyCancelled, + }); + + assert_eq!( + serde_json::to_value(response).unwrap(), + json!({ + "sessionId": Uuid::from_u128(343).to_string(), + "outcome": "alreadyCancelled" + }) + ); + } + #[test] fn settings_dtos_are_additive_and_validate_language_and_icon_contracts() { let legacy_user: UserSettingsDto = serde_json::from_value(json!({})).unwrap(); diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index 1e12e4dd..ea659ef4 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -17,21 +17,21 @@ use application::{ AgentWakeService, AllocateManagedTemp, AppError, AssignIssueAgent, AssignSkillToAgent, AssignTicketToSprint, AssignedSkillResolver, AttachLiveAgent, AuthenticateSession, BackgroundCommandArchive, BulkDeleteIssues, BulkUpdateIssuePriority, BulkUpdateIssueStatus, - CancelBackgroundTask, ChangeAgentProfile, CheckEmbedderSuggestion, ClearChatDraft, - CloneOpenCodeProfileFromSeed, CloneProfileFromSeed, CloseProject, CloseTab, CloseTerminal, - CloseTicketAssistant, ConfigureProfiles, ContextGuardUseCases, CreateAgentFromScratch, - CreateAgentFromTemplate, CreateIssue, CreateLayout, CreateMemory, CreateProject, CreateSkill, - CreateSprint, CreateTemplate, DeleteAgent, DeleteEmbedderProfile, DeleteIssue, DeleteLayout, - DeleteMemory, DeleteModelArtifact, DeleteModelServer, DeleteProfile, DeleteSkill, DeleteSprint, - DeleteTemplate, DescribeEmbedderEngines, DetectAgentDrift, DetectProfiles, - DismissEmbedderSuggestion, EnsureLocalModelServer, ExecuteSlashCommand, FirstRunState, - GcManagedTemp, GetAppExitWorkGuardState, GetLiveStateLean, GetMemory, GetProjectPermissions, - GetProjectSystemPermissions, GetProjectWorkState, GitBranches, GitCheckout, GitCommit, - GitGraph, GitInit, GitLog, GitStage, GitStatus, GitUnstage, HarvestMemoryFromTurn, - HealthUseCase, ImportChatAttachments, InspectConversation, InstallPluginFromArchive, - InstallPluginFromDirectory, JsonPluginManifestValidator, LaunchAgent, LaunchAgentInput, - LinkIssues, ListAgents, ListAgentsInput, ListClaudeModels, ListCodexModels, ListDevices, - ListEmbedderProfiles, ListIssues, ListLayouts, ListMemories, ListModelServers, + CancelAgentChat, CancelBackgroundTask, ChangeAgentProfile, CheckEmbedderSuggestion, + ClearChatDraft, CloneOpenCodeProfileFromSeed, CloneProfileFromSeed, CloseProject, CloseTab, + CloseTerminal, CloseTicketAssistant, ConfigureProfiles, ContextGuardUseCases, + CreateAgentFromScratch, CreateAgentFromTemplate, CreateIssue, CreateLayout, CreateMemory, + CreateProject, CreateSkill, CreateSprint, CreateTemplate, DeleteAgent, DeleteEmbedderProfile, + DeleteIssue, DeleteLayout, DeleteMemory, DeleteModelArtifact, DeleteModelServer, DeleteProfile, + DeleteSkill, DeleteSprint, DeleteTemplate, DescribeEmbedderEngines, DetectAgentDrift, + DetectProfiles, DismissEmbedderSuggestion, EnsureLocalModelServer, ExecuteSlashCommand, + FirstRunState, GcManagedTemp, GetAppExitWorkGuardState, GetLiveStateLean, GetMemory, + GetProjectPermissions, GetProjectSystemPermissions, GetProjectWorkState, GitBranches, + GitCheckout, GitCommit, GitGraph, GitInit, GitLog, GitStage, GitStatus, GitUnstage, + HarvestMemoryFromTurn, HealthUseCase, ImportChatAttachments, InspectConversation, + InstallPluginFromArchive, InstallPluginFromDirectory, JsonPluginManifestValidator, LaunchAgent, + LaunchAgentInput, LinkIssues, ListAgents, ListAgentsInput, ListClaudeModels, ListCodexModels, + ListDevices, ListEmbedderProfiles, ListIssues, ListLayouts, ListMemories, ListModelServers, ListOpenCodeProviders, ListPluginRuntimeContributions, ListPlugins, ListProfiles, ListProjectFiles, ListProjectPluginSettings, ListProjects, ListResumableAgents, ListSkills, ListSlashCommands, ListSprints, ListTemplates, LiveAgentRegistry, LiveSessions, @@ -2207,6 +2207,8 @@ pub struct BackendCore { /// `LaunchAgent`/`ChangeAgentProfile` ; consommé par les commandes de chat (D4) /// pour résoudre la session vivante d'un `sessionId` et l'arrêter à la fermeture. pub structured_sessions: Arc, + /// Non-destructive cancellation of the current structured chat request. + pub cancel_agent_chat: Arc, // --- Agents (L6) --- /// Create a project agent from scratch. pub create_agent: Arc, @@ -2752,6 +2754,7 @@ impl BackendCore { // `profile.structured_adapter`. Injectés dans LaunchAgent (routage §17.4) et // ChangeAgentProfile (shutdown polymorphe au hot-swap). let structured_sessions = Arc::new(StructuredSessions::new()); + let cancel_agent_chat = Arc::new(CancelAgentChat::new(Arc::clone(&structured_sessions))); let live_sessions = Arc::new(LiveSessions::new( Arc::clone(&terminal_sessions), Arc::clone(&structured_sessions), @@ -4462,6 +4465,7 @@ impl BackendCore { update_agent_mcp_tool_permissions, set_agent_mcp_tool_allowed, structured_sessions, + cancel_agent_chat, create_agent, list_agents, read_agent_context, diff --git a/crates/domain/src/ports.rs b/crates/domain/src/ports.rs index 3ea0229c..00a373a9 100644 --- a/crates/domain/src/ports.rs +++ b/crates/domain/src/ports.rs @@ -1608,8 +1608,8 @@ pub trait AgentSession: Send + Sync { /// # Errors /// [`AgentSessionError::Io`] when the adapter attempted a concrete cancel and /// it failed. - async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> { - Ok(()) + async fn cancel_current_turn(&self) -> Result { + Ok(CancelCurrentTurnOutcome::AlreadyCompleted) } /// Termine proprement la session (tue le process/SDK sous-jacent). Idempotent. @@ -1619,6 +1619,20 @@ pub trait AgentSession: Send + Sync { async fn shutdown(&self) -> Result<(), AgentSessionError>; } +/// Result of a non-destructive request to cancel one structured-agent turn. +/// +/// The session itself remains live for every outcome. A missing session is not +/// represented here: the application registry reports that separately. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CancelCurrentTurnOutcome { + /// A running turn accepted the cancellation request. + Cancelled, + /// The same turn had already accepted a cancellation request. + AlreadyCancelled, + /// The session is live but no turn is currently running. + AlreadyCompleted, +} + /// **Factory** sélectionnée par le profil (ARCHITECTURE §17.1) : crée/reprend une /// [`AgentSession`] pour un agent IA. C'est elle qui sait *quel adapter* instancier /// (Claude/Codex) selon `profile.structured_adapter` (§17.3). Open/Closed : ajouter diff --git a/crates/infrastructure/src/session/claude.rs b/crates/infrastructure/src/session/claude.rs index fc538d78..b6b76ac7 100644 --- a/crates/infrastructure/src/session/claude.rs +++ b/crates/infrastructure/src/session/claude.rs @@ -395,7 +395,9 @@ impl AgentSession for ClaudeSdkSession { Ok(()) } - async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> { + async fn cancel_current_turn( + &self, + ) -> Result { self.turn_cancel.cancel().await } } diff --git a/crates/infrastructure/src/session/codex.rs b/crates/infrastructure/src/session/codex.rs index 5c93c3d6..56a45fb6 100644 --- a/crates/infrastructure/src/session/codex.rs +++ b/crates/infrastructure/src/session/codex.rs @@ -628,7 +628,9 @@ impl AgentSession for CodexExecSession { Ok(()) } - async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> { + async fn cancel_current_turn( + &self, + ) -> Result { self.turn_cancel.cancel().await } } diff --git a/crates/infrastructure/src/session/opencode.rs b/crates/infrastructure/src/session/opencode.rs index 8758d202..62c18ef6 100644 --- a/crates/infrastructure/src/session/opencode.rs +++ b/crates/infrastructure/src/session/opencode.rs @@ -23,6 +23,8 @@ use domain::ports::{ use domain::sandbox::{SandboxEnforcer, SandboxPlan}; use domain::SessionId; +use super::process::{TurnCancel, TurnLease}; + /// Un événement OpenCode parsé depuis stdout JSONL. #[derive(Debug, Clone, PartialEq, Eq)] pub enum ParsedEvent { @@ -553,25 +555,12 @@ pub struct OpenCodeSession { project_root: String, env: Vec<(String, String)>, engine_session_id: Mutex>, - current_child: Mutex>>>, + turn_cancel: TurnCancel, sandbox: Option, sandbox_enforcer: Option>, } impl OpenCodeSession { - fn clear_current_child(&self, child: &Arc>) { - let mut current = self - .current_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; - } - } - /// Construit l'adapter. `command_prefix` peut être `opencode`, un chemin absolu, /// ou un wrapper avec arguments; IdeA ajoute ensuite `run --format json`. pub fn new( @@ -596,7 +585,7 @@ impl OpenCodeSession { project_root: project_root.into(), env, engine_session_id: Mutex::new(seed), - current_child: Mutex::new(None), + turn_cancel: TurnCancel::new(), sandbox, sandbox_enforcer, }) @@ -652,21 +641,10 @@ impl AgentSession for OpenCodeSession { Ok(()) } - async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> { - let child = self - .current_child - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .clone(); - if let Some(child) = child { - child - .lock() - .await - .kill() - .await - .map_err(|e| AgentSessionError::Io(format!("annulation OpenCode: {e}")))?; - } - Ok(()) + async fn cancel_current_turn( + &self, + ) -> Result { + self.turn_cancel.cancel().await } } @@ -676,9 +654,10 @@ impl OpenCodeSession { prompt: &str, tap: Option>, ) -> Result { + let turn = self.turn_cancel.begin(); let initial_session = self.conversation_id(); let first = self - .run_once(prompt, tap.clone(), initial_session.clone()) + .run_once(prompt, tap.clone(), initial_session.clone(), &turn) .await; let events = match first { Ok(events) => events, @@ -686,7 +665,7 @@ impl OpenCodeSession { if initial_session.is_some() && matches!(err, AgentSessionError::Decode(ref message) if opencode_missing_final(message)) => { - self.run_once(prompt, tap, None).await? + self.run_once(prompt, tap, None, &turn).await? } Err(err) => return Err(err), }; @@ -698,6 +677,7 @@ impl OpenCodeSession { prompt: &str, tap: Option>, expected_session: Option, + turn: &TurnLease, ) -> Result, AgentSessionError> { let mut cmd = Command::new(&self.command); cmd.args(self.build_args(prompt, expected_session.as_deref())) @@ -718,13 +698,7 @@ impl OpenCodeSession { .spawn() .map_err(|e| AgentSessionError::Start(format!("{}: {e}", self.command)))?; let child = Arc::new(AsyncMutex::new(child)); - { - let mut current = self - .current_child - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - *current = Some(Arc::clone(&child)); - } + turn.track_async(Arc::clone(&child)).await?; let outcome = async { let (stdout, mut stderr_pipe) = { @@ -784,7 +758,7 @@ impl OpenCodeSession { Ok((expected_session, collected, stderr_bytes, status)) } .await; - self.clear_current_child(&child); + turn.clear_async(&child); let (expected_session, collected, stderr_bytes, status) = outcome?; let stderr = String::from_utf8_lossy(&stderr_bytes); let parsed_session = expected_session.or_else(|| self.conversation_id()); diff --git a/crates/infrastructure/src/session/process.rs b/crates/infrastructure/src/session/process.rs index 4f527355..f109cee7 100644 --- a/crates/infrastructure/src/session/process.rs +++ b/crates/infrastructure/src/session/process.rs @@ -36,7 +36,7 @@ use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::process::Command; use tokio::sync::Mutex as AsyncMutex; -use domain::ports::AgentSessionError; +use domain::ports::{AgentSessionError, CancelCurrentTurnOutcome}; use domain::sandbox::{SandboxEnforcer, SandboxPlan}; /// Une invocation orientée lignes : binaire + arguments + cwd + env + prompt à @@ -64,10 +64,68 @@ pub struct SpawnLine { /// Shared handle used by structured adapters to cancel the process backing the /// current turn without shutting down the reusable session. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum LastTurnState { + Completed, + Cancelled, +} + +#[derive(Default)] +struct TurnCancelState { + next_generation: u64, + active: Option<(u64, bool)>, + last: Option, +} + +type AsyncTurnChild = Arc>; +type StdTurnChild = Arc>; +type TrackedAsyncTurnChild = Option<(u64, AsyncTurnChild)>; +type TrackedStdTurnChild = Option<(u64, StdTurnChild)>; + +/// Generation-aware handle for cancelling the process backing one current turn. #[derive(Clone, Default)] pub struct TurnCancel { - async_child: Arc>>>>, - std_child: Arc>>>>, + state: Arc>, + async_child: Arc>, + std_child: Arc>, +} + +#[derive(Clone)] +struct TurnToken { + cancel: TurnCancel, + generation: u64, +} + +pub(super) struct TurnLease { + token: TurnToken, +} + +impl TurnLease { + pub(super) async fn track_async(&self, child: AsyncTurnChild) -> Result<(), AgentSessionError> { + if !self + .token + .cancel + .set_async(self.token.generation, Arc::clone(&child)) + { + child + .lock() + .await + .kill() + .await + .map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?; + } + Ok(()) + } + + pub(super) fn clear_async(&self, child: &AsyncTurnChild) { + self.token.cancel.clear_async(self.token.generation, child); + } +} + +impl Drop for TurnLease { + fn drop(&mut self) { + self.token.cancel.finish(self.token.generation); + } } impl TurnCancel { @@ -77,76 +135,166 @@ impl TurnCancel { Self::default() } - fn set_async(&self, child: Arc>) { + pub(super) fn begin(&self) -> TurnLease { + let generation = { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.next_generation = state.next_generation.wrapping_add(1).max(1); + let generation = state.next_generation; + state.active = Some((generation, false)); + generation + }; + TurnLease { + token: TurnToken { + cancel: self.clone(), + generation, + }, + } + } + + fn finish(&self, generation: u64) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some((active_generation, cancel_requested)) = state.active { + if active_generation == generation { + state.active = None; + state.last = Some(if cancel_requested { + LastTurnState::Cancelled + } else { + LastTurnState::Completed + }); + } + } + } + + fn restore_after_cancel_failure(&self, generation: u64) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if state.active == Some((generation, true)) { + state.active = Some((generation, false)); + } + } + + fn set_async(&self, generation: u64, child: AsyncTurnChild) -> bool { + let state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !matches!(state.active, Some((active, false)) if active == generation) { + return false; + } *self .async_child .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(child); + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some((generation, child)); + true } - fn clear_async(&self, child: &Arc>) { + fn clear_async(&self, generation: u64, child: &AsyncTurnChild) { 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)) - { + if current.as_ref().is_some_and(|(stored_generation, stored)| { + *stored_generation == generation && Arc::ptr_eq(stored, child) + }) { *current = None; } } - fn set_std(&self, child: Arc>) { + fn set_std(&self, generation: u64, child: StdTurnChild) -> bool { + let state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !matches!(state.active, Some((active, false)) if active == generation) { + return false; + } *self .std_child .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(child); + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some((generation, child)); + true } - fn clear_std(&self, child: &Arc>) { + fn clear_std(&self, generation: u64, child: &StdTurnChild) { 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)) - { + if current.as_ref().is_some_and(|(stored_generation, stored)| { + *stored_generation == generation && 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> { + /// Kills the current turn process, if any, without shutting down the session. + pub async fn cancel(&self) -> Result { + let generation = { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + match state.active { + Some((_, true)) => return Ok(CancelCurrentTurnOutcome::AlreadyCancelled), + Some((generation, false)) => { + state.active = Some((generation, true)); + generation + } + None => { + return Ok(match state.last { + Some(LastTurnState::Cancelled) => { + CancelCurrentTurnOutcome::AlreadyCancelled + } + Some(LastTurnState::Completed) | None => { + CancelCurrentTurnOutcome::AlreadyCompleted + } + }); + } + } + }; let async_child = self .async_child .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .clone(); + .as_ref() + .filter(|(child_generation, _)| *child_generation == generation) + .map(|(_, child)| Arc::clone(child)); if let Some(child) = async_child { - child - .lock() - .await - .kill() - .await - .map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?; + if let Err(error) = child.lock().await.kill().await { + self.restore_after_cancel_failure(generation); + return Err(AgentSessionError::Io(format!( + "annulation du tour: {error}" + ))); + } } let std_child = self .std_child .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .clone(); + .as_ref() + .filter(|(child_generation, _)| *child_generation == generation) + .map(|(_, child)| Arc::clone(child)); 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}")))?; + if let Err(error) = child.kill() { + self.restore_after_cancel_failure(generation); + return Err(AgentSessionError::Io(format!( + "annulation du tour: {error}" + ))); + } } - Ok(()) + Ok(CancelCurrentTurnOutcome::Cancelled) } } @@ -185,7 +333,8 @@ pub async fn run_turn_cancellable( line_tap: Option>, cancel: &TurnCancel, ) -> Result, AgentSessionError> { - run_turn_impl(spec, timeout, enforcer, line_tap, Some(cancel)).await + let lease = cancel.begin(); + run_turn_impl(spec, timeout, enforcer, line_tap, Some(&lease.token)).await } async fn run_turn_impl( @@ -193,7 +342,7 @@ async fn run_turn_impl( timeout: Option, enforcer: Option<&Arc>, line_tap: Option>, - cancel: Option<&TurnCancel>, + cancel: Option<&TurnToken>, ) -> 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, @@ -220,7 +369,7 @@ async fn run_turn_impl( Ok(result) => result, Err(_elapsed) => { if let Some(cancel) = cancel { - let _ = cancel.cancel().await; + let _ = cancel.cancel.cancel().await; } Err(AgentSessionError::Timeout) } @@ -267,7 +416,7 @@ async fn run_turn_sandboxed( enforcer: Arc, timeout: Option, line_tap: Option>, - cancel: Option, + cancel: Option, ) -> Result, AgentSessionError> { // Données possédées : rien n'emprunte le thread jetable. let command = spec.command.clone(); @@ -339,7 +488,7 @@ fn drain_sandboxed( plan: &SandboxPlan, killer_tx: tokio::sync::oneshot::Sender>>, line_tap: Option>, - cancel: Option, + cancel: Option, ) -> Result, AgentSessionError> { use std::io::{BufRead, BufReader as StdBufReader, Read as _, Write as _}; use std::process::{Command as StdCommand, Stdio}; @@ -401,7 +550,13 @@ fn drain_sandboxed( }); let child = Arc::new(StdMutex::new(child)); if let Some(cancel) = &cancel { - cancel.set_std(Arc::clone(&child)); + if !cancel.cancel.set_std(cancel.generation, Arc::clone(&child)) { + child + .lock() + .map_err(|_| AgentSessionError::Io("child lock empoisonné".to_owned()))? + .kill() + .map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?; + } } // Donne au côté async de quoi tuer l'enfant en cas de timeout. let _ = killer_tx.send(Arc::clone(&child)); @@ -427,7 +582,7 @@ fn drain_sandboxed( .wait() .map_err(|e| AgentSessionError::Io(e.to_string()))?; if let Some(cancel) = &cancel { - cancel.clear_std(&child); + cancel.cancel.clear_std(cancel.generation, &child); } let stderr_bytes = stderr_handle .join() @@ -446,7 +601,7 @@ fn drain_sandboxed( async fn drain( spec: &SpawnLine, line_tap: Option>, - cancel: Option<&TurnCancel>, + cancel: Option<&TurnToken>, ) -> Result, AgentSessionError> { let mut cmd = Command::new(&spec.command); cmd.args(&spec.args) @@ -465,7 +620,17 @@ async fn drain( .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)); + if !cancel + .cancel + .set_async(cancel.generation, Arc::clone(&child)) + { + child + .lock() + .await + .kill() + .await + .map_err(|e| AgentSessionError::Io(format!("annulation du tour: {e}")))?; + } } // Pousse le prompt sur stdin puis le ferme (EOF) — sans quoi un process en mode @@ -525,7 +690,7 @@ async fn drain( // on remonte le stderr réel au lieu de laisser l'appelant conclure à un no-Final. let status = child.lock().await.wait().await.map_err(map_io)?; if let Some(cancel) = cancel { - cancel.clear_async(&child); + cancel.cancel.clear_async(cancel.generation, &child); } let stderr_bytes = stderr_task .await @@ -568,7 +733,7 @@ mod tests { command: "/bin/sh".to_owned(), args: vec![ "-c".to_owned(), - "printf 'started\\n'; sleep 10; printf 'late\\n'".to_owned(), + "printf 'started\\n'; exec sleep 10".to_owned(), ], cwd: "/".to_owned(), env: Vec::new(), @@ -583,7 +748,14 @@ mod tests { }); tokio::time::sleep(Duration::from_millis(100)).await; - cancel.cancel().await.expect("cancel current child"); + assert_eq!( + cancel.cancel().await.expect("cancel current child"), + CancelCurrentTurnOutcome::Cancelled + ); + assert_eq!( + cancel.cancel().await.expect("repeat cancel"), + CancelCurrentTurnOutcome::AlreadyCancelled + ); let lines = worker.await.expect("worker joins").expect("drain result"); assert_eq!(lines, vec!["started"]); @@ -591,5 +763,57 @@ mod tests { started.elapsed() < Duration::from_secs(3), "cancel should not wait for the child process natural exit" ); + assert_eq!( + cancel.cancel().await.expect("cancel after cancelled turn"), + CancelCurrentTurnOutcome::AlreadyCancelled + ); + } + + #[tokio::test] + async fn cancel_after_natural_completion_is_distinct_from_a_missing_session() { + let spec = SpawnLine { + command: "/bin/sh".to_owned(), + args: vec!["-c".to_owned(), "printf 'done\\n'".to_owned()], + cwd: "/".to_owned(), + env: Vec::new(), + stdin: None, + sandbox: None, + }; + let cancel = TurnCancel::new(); + + let lines = run_turn_cancellable(&spec, None, None, None, &cancel) + .await + .expect("turn completes"); + + assert_eq!(lines, ["done"]); + assert_eq!( + cancel.cancel().await.expect("terminal cancel is harmless"), + CancelCurrentTurnOutcome::AlreadyCompleted + ); + } + + #[tokio::test] + async fn cancel_before_child_registration_kills_that_same_generation_on_registration() { + let cancel = TurnCancel::new(); + let turn = cancel.begin(); + assert_eq!( + cancel.cancel().await.expect("pre-spawn cancel"), + CancelCurrentTurnOutcome::Cancelled + ); + let child = Command::new("/bin/sh") + .args(["-c", "sleep 10"]) + .spawn() + .expect("spawn delayed child"); + let child = Arc::new(AsyncMutex::new(child)); + + turn.track_async(Arc::clone(&child)) + .await + .expect("cancelled generation kills child at registration"); + let status = tokio::time::timeout(Duration::from_secs(2), child.lock().await.wait()) + .await + .expect("child exits promptly") + .expect("child wait succeeds"); + + assert!(!status.success()); } } diff --git a/frontend/src/adapters/agent.test.ts b/frontend/src/adapters/agent.test.ts index d5d2cabd..004927b3 100644 --- a/frontend/src/adapters/agent.test.ts +++ b/frontend/src/adapters/agent.test.ts @@ -226,7 +226,15 @@ describe("TauriAgentGateway invoke payloads", () => { }); it("cancelAgentChat invokes cancel_agent_chat without closing the session", async () => { - await new TauriAgentGateway().cancelAgentChat("chat-session-1"); + const response = { + sessionId: "chat-session-1", + outcome: "cancelled" as const, + }; + invoke.mockResolvedValueOnce(response); + + await expect( + new TauriAgentGateway().cancelAgentChat("chat-session-1"), + ).resolves.toEqual(response); expect(invoke).toHaveBeenCalledWith("cancel_agent_chat", { sessionId: "chat-session-1", diff --git a/frontend/src/adapters/agent.ts b/frontend/src/adapters/agent.ts index ed0a65a4..b8525b14 100644 --- a/frontend/src/adapters/agent.ts +++ b/frontend/src/adapters/agent.ts @@ -17,6 +17,7 @@ import { Channel, invoke } from "@tauri-apps/api/core"; import type { Agent, AgentContextDocument, + CancelAgentChatResult, EffortSelection, GatewayError, ExecuteSlashCommandResult, @@ -364,8 +365,8 @@ export class TauriAgentGateway implements AgentGateway { await invoke("close_agent_session", { sessionId }); } - async cancelAgentChat(sessionId: string): Promise { - await invoke("cancel_agent_chat", { sessionId }); + async cancelAgentChat(sessionId: string): Promise { + return invoke("cancel_agent_chat", { sessionId }); } async reattach( diff --git a/frontend/src/adapters/http/agentGateway.test.ts b/frontend/src/adapters/http/agentGateway.test.ts index e985c9d1..218f0651 100644 --- a/frontend/src/adapters/http/agentGateway.test.ts +++ b/frontend/src/adapters/http/agentGateway.test.ts @@ -308,11 +308,16 @@ describe("HttpAgentGateway WS round-trip (B6 frames)", () => { baseUrl: "https://host", fetchImpl: async (url, init) => { calls.push({ url, init }); + const command = JSON.parse((init as { body: string }).body).command; + const response = + command === "cancel_agent_chat" + ? { sessionId: "chat-1", outcome: "cancelled" } + : undefined; return { ok: true, status: 200, - json: async () => undefined, - text: async () => "", + json: async () => response, + text: async () => response ? JSON.stringify(response) : "", }; }, }), @@ -402,7 +407,10 @@ describe("HttpAgentGateway WS round-trip (B6 frames)", () => { }); expect(chunks).toEqual([{ kind: "final", content: "Hello back" }]); - await gw.cancelAgentChat("chat-1"); + await expect(gw.cancelAgentChat("chat-1")).resolves.toEqual({ + sessionId: "chat-1", + outcome: "cancelled", + }); await gw.closeAgentChat("chat-1"); await gw.detachAgentChat("chat-1"); diff --git a/frontend/src/adapters/http/streamGateways.ts b/frontend/src/adapters/http/streamGateways.ts index 72676004..46d0a6cc 100644 --- a/frontend/src/adapters/http/streamGateways.ts +++ b/frontend/src/adapters/http/streamGateways.ts @@ -23,6 +23,7 @@ import type { Agent, AgentContextDocument, AppExitWorkGuardState, + CancelAgentChatResult, DomainEvent, EffortSelection, ExecuteSlashCommandResult, @@ -469,8 +470,8 @@ export class HttpAgentGateway implements AgentGateway { await this.http.invoke("close_agent_session", { sessionId }); } - async cancelAgentChat(sessionId: string): Promise { - await this.http.invoke("cancel_agent_chat", { sessionId }); + async cancelAgentChat(sessionId: string): Promise { + return this.http.invoke("cancel_agent_chat", { sessionId }); } } diff --git a/frontend/src/adapters/mock/index.ts b/frontend/src/adapters/mock/index.ts index 270ea299..d8fc9115 100644 --- a/frontend/src/adapters/mock/index.ts +++ b/frontend/src/adapters/mock/index.ts @@ -9,6 +9,7 @@ import type { AgentDrift, AppExitWorkGuardState, AgentProfile, + CancelAgentChatResult, DiagnosticWarning, DomainEvent, EffortSelection, @@ -1133,13 +1134,14 @@ export class MockAgentGateway implements AgentGateway { } } - async cancelAgentChat(sessionId: string): Promise { + async cancelAgentChat(sessionId: string): Promise { if (!this.chatScrollback.has(sessionId)) { throw { code: "NOT_FOUND", message: `agent chat session ${sessionId} not found`, } as GatewayError; } + return { sessionId, outcome: "cancelled" }; } async reattach( diff --git a/frontend/src/domain/index.ts b/frontend/src/domain/index.ts index e5520794..38ec872d 100644 --- a/frontend/src/domain/index.ts +++ b/frontend/src/domain/index.ts @@ -1937,6 +1937,17 @@ export interface ExecuteSlashCommandResult { effect: SlashCommandEffect; } +/** Idempotent outcome of cancelling the current turn of a structured chat. */ +export type CancelAgentChatOutcome = + | "cancelled" + | "alreadyCancelled" + | "alreadyCompleted"; + +export interface CancelAgentChatResult { + sessionId: string; + outcome: CancelAgentChatOutcome; +} + // --------------------------------------------------------------------------- // Paired devices + pairing code (ticket #77) // --------------------------------------------------------------------------- diff --git a/frontend/src/features/agents/CustomAgentChatView.test.tsx b/frontend/src/features/agents/CustomAgentChatView.test.tsx index 98627a3f..968a6bfb 100644 --- a/frontend/src/features/agents/CustomAgentChatView.test.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.test.tsx @@ -3,7 +3,7 @@ import { useState } from "react"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { DIProvider } from "@/app/di"; -import type { AgentProfile, ReplyChunk } from "@/domain"; +import type { AgentProfile, CancelAgentChatResult, ReplyChunk } from "@/domain"; import { MockSystemGateway, MockWorkStateGateway } from "@/adapters/mock"; import { AnnouncementsProvider } from "@/features/announcements"; import { PluginRuntimeProvider } from "@/features/plugins"; @@ -1562,7 +1562,23 @@ describe("CustomAgentChatView", () => { }); it("cancels only the current request and keeps the structured session usable", async () => { - let resolveCancel: () => void = () => {}; + let resolveCancel: (result: CancelAgentChatResult) => void = () => {}; + let firstTurnChunk: (chunk: ReplyChunk) => void = () => {}; + let rejectFirstTurn: (reason: unknown) => void = () => {}; + const system = new MockSystemGateway(); + const workState = new MockWorkStateGateway(); + workState._setProjectWorkState("project-1", { + agents: [ + { + agentId: "agent-1", + name: "Worker", + profileId: "structured", + busy: { state: "idle" }, + tickets: [], + }, + ], + conversations: [], + }); const agent = { launchAgentChat: vi.fn(), reattachAgentChat: vi.fn(async (sessionId: string) => ({ @@ -1571,14 +1587,19 @@ describe("CustomAgentChatView", () => { })), sendAgentChat: vi .fn() - .mockImplementationOnce(() => new Promise(() => {})) + .mockImplementationOnce((_sessionId: string, _prompt: string, onChunk) => { + firstTurnChunk = onChunk; + return new Promise((_resolve, reject) => { + rejectFirstTurn = reject; + }); + }) .mockImplementationOnce(async (_sessionId: string, _prompt: string, onChunk) => { onChunk({ kind: "final", content: "Next answer" }); }), stopLiveAgent: vi.fn(), cancelAgentChat: vi.fn( () => - new Promise((resolve) => { + new Promise((resolve) => { resolveCancel = resolve; }), ), @@ -1590,7 +1611,8 @@ describe("CustomAgentChatView", () => { null) }, + system, + workState, } as unknown as Gateways} > { expect(hasProjectCliInteraction("project-1")).toBe(true); expect(screen.queryByRole("button", { name: "Cancel" })).toBeNull(); + act(() => { + firstTurnChunk({ + kind: "progress", + progress: { + source: "providerNative", + kind: "turn", + stage: "started", + label: "Tour actif", + nativeEvent: "item.started", + }, + }); + }); + workState._setProjectWorkState("project-1", { + agents: [ + { + agentId: "agent-1", + name: "Worker", + profileId: "structured", + busy: { state: "busy", ticket: "turn-1", sinceMs: 1 }, + tickets: [], + }, + ], + conversations: [], + }); + system.emit({ type: "agentBusyChanged", agentId: "agent-1", busy: true }); + fireEvent.click(await screen.findByRole("button", { name: "Stop" })); const stoppingLabel = await screen.findByText("Stopping..."); @@ -1639,8 +1687,29 @@ describe("CustomAgentChatView", () => { expect(agent.cancelAgentChat).toHaveBeenCalledWith("chat-session-1"), ); expect(agent.cancelAgentChat).toHaveBeenCalledTimes(1); - resolveCancel(); + resolveCancel({ sessionId: "chat-session-1", outcome: "cancelled" }); await waitFor(() => expect(screen.getByText("Demande arrêtée.")).toBeTruthy()); + expect(screen.queryByRole("button", { name: "Stop" })).toBeNull(); + expect(screen.getByRole("button", { name: "Envoyer" })).toBeTruthy(); + act(() => { + firstTurnChunk({ + kind: "progress", + progress: { + source: "providerNative", + kind: "turn", + stage: "completed", + label: "Fin tardive à ignorer", + nativeEvent: "item.completed", + }, + }); + firstTurnChunk({ kind: "final", content: "Réponse tardive à ignorer" }); + rejectFirstTurn({ code: "PROCESS", message: "cancelled turn ended" }); + }); + await act(async () => Promise.resolve()); + expect(screen.queryByText("Fin tardive à ignorer")).toBeNull(); + expect(screen.queryByText("Réponse tardive à ignorer")).toBeNull(); + expect(screen.queryByText(/cancelled turn ended/)).toBeNull(); + expect(screen.getAllByText("Demande arrêtée.")).toHaveLength(1); expect(hasProjectCliInteraction("project-1")).toBe(false); expect(onSessionId).not.toHaveBeenCalledWith(null); expect(agent.stopLiveAgent).not.toHaveBeenCalled(); @@ -1663,6 +1732,74 @@ describe("CustomAgentChatView", () => { expect(await screen.findByText("Next answer")).toBeTruthy(); }); + it.each(["alreadyCancelled", "alreadyCompleted"] as const)( + "treats backend outcome %s as one already-stopped terminal result", + async (outcome) => { + let reattachedChunk: (chunk: ReplyChunk) => void = () => {}; + const system = new MockSystemGateway(); + const workState = new MockWorkStateGateway(); + workState._setProjectWorkState("project-1", { + agents: [ + { + agentId: "agent-1", + name: "Worker", + profileId: "structured", + busy: { state: "busy", ticket: "turn-1", sinceMs: 1 }, + tickets: [], + }, + ], + conversations: [], + }); + const agent = { + launchAgentChat: vi.fn(), + reattachAgentChat: vi.fn( + async (sessionId: string, onChunk: (chunk: ReplyChunk) => void) => { + reattachedChunk = onChunk; + return { sessionId, scrollback: [] }; + }, + ), + sendAgentChat: vi.fn(async () => {}), + cancelAgentChat: vi.fn(async () => ({ + sessionId: "chat-session-1", + outcome, + })), + closeAgentChat: vi.fn(async () => {}), + }; + + render( + + + , + ); + + fireEvent.click(await screen.findByRole("button", { name: "Stop" })); + + await waitFor(() => + expect(screen.getAllByText("Demande déjà arrêtée.")).toHaveLength(1), + ); + expect(screen.queryByRole("button", { name: "Stop" })).toBeNull(); + expect(screen.getByTestId("custom-agent-chat-view")).toBeTruthy(); + expect(agent.closeAgentChat).not.toHaveBeenCalled(); + act(() => { + reattachedChunk({ kind: "final", content: "Late reattached answer" }); + }); + expect(screen.queryByText("Late reattached answer")).toBeNull(); + }, + ); + it("keeps the composer locked from canonical work-state busy even without local stream activity", async () => { const system = new MockSystemGateway(); const workState = new MockWorkStateGateway(); @@ -5625,6 +5762,9 @@ describe("CustomAgentChatView", () => { fireEvent.click(await screen.findByRole("button", { name: "Stop" })); expect(await screen.findByText("Demande déjà arrêtée.")).toBeTruthy(); + expect(screen.getAllByText("Demande déjà arrêtée.")).toHaveLength(1); + expect(screen.queryByRole("button", { name: "Stop" })).toBeNull(); + expect(screen.getByTestId("custom-agent-chat-view")).toBeTruthy(); expect(onSessionId).not.toHaveBeenCalledWith(null); expect(agent.cancelAgentChat).toHaveBeenCalledWith("gone-session"); expect(agent.stopLiveAgent).not.toHaveBeenCalled(); diff --git a/frontend/src/features/agents/CustomAgentChatView.tsx b/frontend/src/features/agents/CustomAgentChatView.tsx index 5a990867..ce2f6472 100644 --- a/frontend/src/features/agents/CustomAgentChatView.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.tsx @@ -1009,6 +1009,7 @@ export function CustomAgentChatView({ >(null); const [busy, setBusy] = useState(false); const [stopping, setStopping] = useState(false); + const [stopTerminalOverride, setStopTerminalOverride] = useState(false); const [error, setError] = useState(null); const [projectIcon, setProjectIcon] = useState(null); const scrollRef = useRef(null); @@ -1032,6 +1033,11 @@ export function CustomAgentChatView({ const draftSaveSeqRef = useRef(0); const interactionSeqRef = useRef(0); const activeInteractionIdRef = useRef(null); + const cancelledInteractionIdsRef = useRef(new Set()); + const ignoreUnscopedChunksAfterStopRef = useRef(false); + const stopInFlightRef = useRef(false); + const stopTerminalOverrideRef = useRef(false); + const requesterRefreshSeqRef = useRef(0); const draftHydratedRef = useRef(false); const draftEditedBeforeHydrationRef = useRef(false); const lastPersistedDraftRef = useRef(""); @@ -1185,18 +1191,24 @@ export function CustomAgentChatView({ busy || stopping || Boolean(restart && restart.outcome !== "failed") || - requesterDelegation.currentAgentBusy || - delegationActive; + (!stopTerminalOverride && + (requesterDelegation.currentAgentBusy || delegationActive)); const delegationFallbackText = requesterDelegation.delegationTargetName ? `${agentName} attend ${requesterDelegation.delegationTargetName}` : tr("Délégation en cours.", "Delegation in progress."); const refreshRequesterDelegation = useCallback(async () => { + const refreshSeq = ++requesterRefreshSeqRef.current; if (!workState) { + if (refreshSeq !== requesterRefreshSeqRef.current) return; setRequesterDelegation({ currentAgentBusy: false, delegationTargetName: null, }); + if (stopTerminalOverrideRef.current) { + stopTerminalOverrideRef.current = false; + setStopTerminalOverride(false); + } return; } try { @@ -1214,11 +1226,17 @@ export function CustomAgentChatView({ ticket.source.agentId === agentId, ), ); + if (refreshSeq !== requesterRefreshSeqRef.current) return; setRequesterDelegation({ currentAgentBusy, delegationTargetName: target ? workAgentDisplayName(target) : null, }); + if (stopTerminalOverrideRef.current && !currentAgentBusy && !target) { + stopTerminalOverrideRef.current = false; + setStopTerminalOverride(false); + } } catch { + if (refreshSeq !== requesterRefreshSeqRef.current) return; setRequesterDelegation({ currentAgentBusy: false, delegationTargetName: null, @@ -1446,7 +1464,20 @@ export function CustomAgentChatView({ }, []); const receive = useCallback( - (chunk: ReplyChunk) => { + (chunk: ReplyChunk, sourceInteractionId?: string) => { + if ( + sourceInteractionId && + cancelledInteractionIdsRef.current.has(sourceInteractionId) + ) { + return; + } + if (!sourceInteractionId && ignoreUnscopedChunksAfterStopRef.current) { + if (chunk.kind === "userPrompt") { + ignoreUnscopedChunksAfterStopRef.current = false; + } else { + return; + } + } setTranscriptUnavailableReason(null); scrollbackChunkCountRef.current += 1; const activeInteractionId = activeInteractionIdRef.current; @@ -2284,6 +2315,9 @@ export function CustomAgentChatView({ const attachmentLabels = outgoingAttachments.map((item) => item.label); const displayText = text || "Pièce jointe"; const interactionId = `${agentId}:${++interactionSeqRef.current}`; + ignoreUnscopedChunksAfterStopRef.current = false; + stopTerminalOverrideRef.current = false; + setStopTerminalOverride(false); activeInteractionIdRef.current = interactionId; markProjectCliInteractionStarted(projectId, interactionId); const finishInteraction = () => { @@ -2312,8 +2346,10 @@ export function CustomAgentChatView({ })); const sendTurn = (sid: string) => attachmentInputs.length > 0 - ? agent.sendAgentChat!(sid, text, receive, { attachments: attachmentInputs }) - : agent.sendAgentChat!(sid, text, receive); + ? agent.sendAgentChat!(sid, text, (chunk) => receive(chunk, interactionId), { + attachments: attachmentInputs, + }) + : agent.sendAgentChat!(sid, text, (chunk) => receive(chunk, interactionId)); try { const sid = currentSession ?? @@ -2322,8 +2358,10 @@ export function CustomAgentChatView({ retryAttachNotFound: true, })); await sendTurn(sid); + if (cancelledInteractionIdsRef.current.has(interactionId)) return; finishInteraction(); } catch (e) { + if (cancelledInteractionIdsRef.current.has(interactionId)) return; if (isNotFound(e)) { try { setCurrentSession(null); @@ -2361,11 +2399,31 @@ export function CustomAgentChatView({ async function stopCurrentRequest() { const sid = sessionRef.current; - if (!sid || !agent.cancelAgentChat || stopping) return; + if ( + !sid || + !agent.cancelAgentChat || + stopping || + stopInFlightRef.current || + stopTerminalOverrideRef.current + ) { + return; + } + stopInFlightRef.current = true; + ignoreUnscopedChunksAfterStopRef.current = true; + const interactionBeingCancelled = activeInteractionIdRef.current; + if (interactionBeingCancelled) { + cancelledInteractionIdsRef.current.add(interactionBeingCancelled); + } setStopping(true); setError(null); - try { - await agent.cancelAgentChat(sid); + const settleStoppedRequest = (alreadyStopped: boolean) => { + if (stopTerminalOverrideRef.current) return; + stopTerminalOverrideRef.current = true; + setStopTerminalOverride(true); + setRequesterDelegation({ + currentAgentBusy: false, + delegationTargetName: null, + }); if (activeInteractionIdRef.current) { markProjectCliInteractionStopped(projectId, activeInteractionIdRef.current); activeInteractionIdRef.current = null; @@ -2380,32 +2438,28 @@ export function CustomAgentChatView({ ...completed.turns, { role: "tool", - label: tr("Demande arrêtée.", "Request stopped."), + label: alreadyStopped + ? tr("Demande déjà arrêtée.", "Request already stopped.") + : tr("Demande arrêtée.", "Request stopped."), receivedAtMs: null, }, ], }; }); + void refreshRequesterDelegation(); + }; + try { + const result = await agent.cancelAgentChat(sid); + settleStoppedRequest(result.outcome !== "cancelled"); } catch (e) { if (isNotFound(e)) { - if (activeInteractionIdRef.current) { - markProjectCliInteractionStopped(projectId, activeInteractionIdRef.current); - activeInteractionIdRef.current = null; - } - setBusy(false); - setOpening(false); - updateChatViewState((prev) => { - const completed = completeChatViewProgress(prev); - return { - ...completed, - turns: [ - ...completed.turns, - { role: "tool", label: tr("Demande déjà arrêtée.", "Request already stopped."), receivedAtMs: null }, - ], - }; - }); + settleStoppedRequest(true); return; } + if (interactionBeingCancelled) { + cancelledInteractionIdsRef.current.delete(interactionBeingCancelled); + } + ignoreUnscopedChunksAfterStopRef.current = false; setError({ message: describe(e), openFailure: false }); updateChatViewState((prev) => ({ ...prev, @@ -2419,6 +2473,7 @@ export function CustomAgentChatView({ ], })); } finally { + stopInFlightRef.current = false; setStopping(false); } } diff --git a/frontend/src/ports/index.ts b/frontend/src/ports/index.ts index a711f7cd..4b8d1631 100644 --- a/frontend/src/ports/index.ts +++ b/frontend/src/ports/index.ts @@ -13,6 +13,7 @@ import type { AgentDrift, AgentProfile, AppExitWorkGuardState, + CancelAgentChatResult, CustomProviderConfig, DomainEvent, EffortSelection, @@ -398,7 +399,7 @@ export interface AgentGateway { /** Clears a persisted custom chat draft. */ clearChatDraft?(scope: ChatDraftScope): Promise; /** Interrupts only the current turn of a live structured session. */ - cancelAgentChat?(sessionId: string): Promise; + cancelAgentChat?(sessionId: string): Promise; /** Shuts a live structured session down. */ closeAgentChat?(sessionId: string): Promise; /**