From 12f7272fbe577d1d7bb16ffcf2a506ef97b059be Mon Sep 17 00:00:00 2001 From: Blomios Date: Wed, 16 Sep 2026 20:46:38 +0200 Subject: [PATCH] =?UTF-8?q?fix(application,backend,app-tauri):=20lib=C3=A8?= =?UTF-8?q?re=20l'=C3=A9tat=20busy=20apr=C3=A8s=20Final=20et=20draine=20la?= =?UTF-8?q?=20queue=20des=20messages=20agents=20=E2=80=94=20Final/Error=20?= =?UTF-8?q?terminent=20imm=C3=A9diatement=20le=20pump=20structur=C3=A9=20v?= =?UTF-8?q?ia=20reply=5Fevent=5Fends=5Fturn=20(le=20provider=20n'est=20pas?= =?UTF-8?q?=20tenu=20de=20fermer=20son=20it=C3=A9rateur),=20release=20busy?= =?UTF-8?q?/FIFO=20idempotente=20born=C3=A9e=20au=20ticket=20observ=C3=A9?= =?UTF-8?q?=20(release=5Fturn=5Faccounting:=20cancel=5Fhead=20+=20mark=5Fi?= =?UTF-8?q?dle=20seulement=20si=20le=20ticket=20appartient=20toujours=20au?= =?UTF-8?q?=20tour),=20cleanup=20tardif=20Final/close/timeout/cancel=20san?= =?UTF-8?q?s=20lib=C3=A9rer=20un=20tour=20successeur,=20factoring=20Cancel?= =?UTF-8?q?AgentChat=20sur=20la=20m=C3=AAme=20primitive,=20r=C3=A9gression?= =?UTF-8?q?=20#357=20pr=C3=A9serv=C3=A9e=20=E2=80=94=20#362=20(QA=20verte:?= =?UTF-8?q?=20cargo=20fmt=20--check,=20chat=5Fbridge=2036=20tests,=20cance?= =?UTF-8?q?l=5Fagent=5Fchat=204=20tests,=20backend=20106=20tests,=20drain?= =?UTF-8?q?=5Fwith=5Freadiness=5Flot1=207=20tests)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/app-tauri/src/chat.rs | 10 +++ crates/app-tauri/src/commands.rs | 13 ++-- crates/app-tauri/tests/chat_bridge.rs | 41 +++++++++-- crates/application/src/lib.rs | 10 +-- crates/application/src/terminal/mod.rs | 8 +-- crates/application/src/terminal/usecases.rs | 31 ++++++--- crates/backend/src/lib.rs | 76 ++++++++++++++++++++- 7 files changed, 160 insertions(+), 29 deletions(-) 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, + ); } } }