diff --git a/crates/app-tauri/src/chat.rs b/crates/app-tauri/src/chat.rs index 0b1549d1..2ccf7e17 100644 --- a/crates/app-tauri/src/chat.rs +++ b/crates/app-tauri/src/chat.rs @@ -390,6 +390,16 @@ pub fn chunk_from_event(event: ReplyEvent) -> Option { chunk_from_event_at(event, None) } +/// Returns whether one provider event closes the concrete structured turn. +/// +/// This is the control-plane boundary used by the `agent_send` pump. In +/// particular, the provider is not required to close its iterator after a +/// terminal event: keeping the iterator alive must not keep the agent Busy. +#[must_use] +pub fn reply_event_ends_turn(event: &ReplyEvent) -> bool { + matches!(event, ReplyEvent::Final { .. } | ReplyEvent::Error { .. }) +} + /// Maps a domain reply event while preserving the backend acceptance time. #[must_use] pub fn chunk_from_event_at(event: ReplyEvent, at_ms: Option) -> Option { diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index 9f32604a..d829e4b8 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -3122,11 +3122,7 @@ pub async fn agent_send( terminal_visible = true; break 'logical_turn; } - let provider_terminal = matches!( - event, - domain::ports::ReplyEvent::Final { .. } - | domain::ports::ReplyEvent::Error { .. } - ); + let provider_terminal = crate::chat::reply_event_ends_turn(&event); if provider_terminal { concrete_terminal = true; if let Some(guard) = turn_guard.as_mut() { @@ -3199,6 +3195,13 @@ pub async fn agent_send( // (no channel at this generation) we keep draining so the turn still // completes and the scrollback stays whole for the next re-attach. let _ = bridge.send_output_settled(&sid, &mut settler, chunk); + if provider_terminal { + // Final/Error is the authoritative structured boundary. Some + // providers keep their iterator/channel alive after publishing it; + // waiting for EOF would leave the mediated ticket Busy forever even + // though the UI already displayed the terminal chunk. + break 'logical_turn; + } } if !resume && !concrete_terminal { diff --git a/crates/app-tauri/tests/chat_bridge.rs b/crates/app-tauri/tests/chat_bridge.rs index 5a00e0dc..f872da31 100644 --- a/crates/app-tauri/tests/chat_bridge.rs +++ b/crates/app-tauri/tests/chat_bridge.rs @@ -14,7 +14,7 @@ use std::sync::{Arc, Mutex}; use tauri::ipc::{Channel, InvokeResponseBody}; use app_tauri_lib::chat::{ - chunk_from_event, unfinished_turn_chunk, ChatBridge, ProgressSettler, + chunk_from_event, reply_event_ends_turn, unfinished_turn_chunk, ChatBridge, ProgressSettler, MAX_CHAT_SCROLLBACK_BYTES, MAX_CHAT_SCROLLBACK_CHUNKS, MODEL_RESPONSE_NOT_FINALIZED_MESSAGE, }; use app_tauri_lib::dto::{ @@ -264,6 +264,7 @@ fn chunk_from_event_drops_heartbeat() { fn pump_turn(bridge: &ChatBridge, session: &SessionId, gen: u64, events: Vec) { let mut terminal_visible = false; for event in events { + let terminal = reply_event_ends_turn(&event); // Mirror the real pump: heartbeats map to no chunk and are skipped. if let Some(chunk) = chunk_from_event(event) { if matches!( @@ -274,6 +275,9 @@ fn pump_turn(bridge: &ChatBridge, session: &SessionId, gen: u64, events: Vec bool { + if input.busy_state(agent).ticket() != Some(ticket) { + return false; + } + mailbox.cancel_head(agent, ticket); + if input.busy_state(agent).ticket() != Some(ticket) { + return false; + } + input.mark_idle(agent); + true +} + /// Cancels one structured chat turn without stopping its reusable runtime. pub struct CancelAgentChat { sessions: Arc, @@ -72,14 +94,7 @@ impl CancelAgentChat { if let (Some((agent, ticket)), Some((input, mailbox))) = (active_ticket, self.turn_accounting.as_ref()) { - // The provider kill can race the old pump's normal guard cleanup and - // a caller may already have started the next turn. Retire only the - // exact ticket observed before cancellation; never mark a successor - // idle when it has already replaced that ticket. - if input.busy_state(agent).ticket() == Some(ticket) { - mailbox.cancel_head(agent, ticket); - input.mark_idle(agent); - } + release_turn_accounting(input.as_ref(), mailbox.as_ref(), agent, ticket); } } Ok(CancelAgentChatOutput { diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index b11cabe8..2f3bf787 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -1165,6 +1165,7 @@ fn automatic_resume_launch_input( mod self_restart_contract_tests { use super::*; use domain::input::InputMediator; + use domain::mailbox::AgentMailbox; #[test] fn relaunch_preserves_conversation_and_never_requests_fresh_start() { @@ -1342,6 +1343,70 @@ mod self_restart_contract_tests { assert_eq!(mailbox.pending(&key), 0); } + #[tokio::test] + async fn human_turn_finish_is_idempotent() { + let mailbox = Arc::new(InMemoryMailbox::new()); + let input = Arc::new(MediatedInbox::new( + Arc::clone(&mailbox), + Arc::new(SystemMillisClock), + )); + let project_id = ProjectId::from_uuid(Uuid::from_u128(362)); + let agent_id = AgentId::from_uuid(Uuid::from_u128(1)); + let key = domain::RuntimeAgentKey::new(project_id, agent_id); + let turn = StructuredHumanTurnGuard::start( + Arc::clone(&input) as Arc, + Arc::clone(&mailbox) as Arc, + Arc::new(application::RestartRendezvousCoordinator::new()), + project_id, + agent_id, + "first", + ); + + turn.finish(); + turn.finish(); + + assert_eq!(input.busy_state(key), domain::input::AgentBusyState::Idle); + assert_eq!(mailbox.pending(&key), 0); + } + + #[tokio::test] + async fn late_human_turn_finish_cannot_release_a_successor() { + let mailbox = Arc::new(InMemoryMailbox::new()); + let input = Arc::new(MediatedInbox::new( + Arc::clone(&mailbox), + Arc::new(SystemMillisClock), + )); + let project_id = ProjectId::from_uuid(Uuid::from_u128(363)); + let agent_id = AgentId::from_uuid(Uuid::from_u128(1)); + let key = domain::RuntimeAgentKey::new(project_id, agent_id); + let turn = StructuredHumanTurnGuard::start( + Arc::clone(&input) as Arc, + Arc::clone(&mailbox) as Arc, + Arc::new(application::RestartRendezvousCoordinator::new()), + project_id, + agent_id, + "first", + ); + let (_, first_ticket) = turn.accounting_identity().expect("first turn owns Busy"); + + // Stop owns the first release, then the FIFO starts its successor before a + // late Final/close/timeout path reaches the original guard. + mailbox.cancel_head(key, first_ticket); + input.mark_idle(key); + let successor = domain::mailbox::Ticket::new( + domain::mailbox::TicketId::new_random(), + "user", + "successor", + ); + let successor_ticket = successor.id; + let _pending = input.enqueue_silent(key, successor); + + turn.finish(); + + assert_eq!(input.busy_state(key).ticket(), Some(successor_ticket)); + assert_eq!(mailbox.pending(&key), 1); + } + #[tokio::test] async fn cell_restart_stop_timeout_is_typed_and_never_reaches_the_next_step() { let agent = AgentId::from_uuid(Uuid::from_u128(317)); @@ -1846,11 +1911,16 @@ impl StructuredHumanTurnGuard { } } - /// Retires the FIFO head and marks the agent idle. Idempotent across clones. + /// Retires the FIFO head and marks the agent idle. Idempotent across clones and + /// fenced against a late cleanup from an older turn. pub fn finish(&self) { if self.armed.swap(false, std::sync::atomic::Ordering::AcqRel) { - self.mailbox.cancel_head(self.agent, self.ticket_id); - self.input.mark_idle(self.agent); + application::release_turn_accounting( + self.input.as_ref(), + self.mailbox.as_ref(), + self.agent, + self.ticket_id, + ); } } }