merge(application,backend,app-tauri): fix #362 — libère l'état busy après Final et draine la queue des messages agents (QA verte)

crates/app-tauri/src/chat.rs

@@ -390,6 +390,16 @@ pub fn chunk_from_event(event: ReplyEvent) -> Option<ReplyChunk> {
     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<u64>) -> Option<ReplyChunk> {

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 {

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<ReplyEvent>) {
     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<Rep
             }
             let _ = bridge.send_output(session, chunk);
         }
+        if terminal {
+            break;
+        }
     }
     if !terminal_visible {
         let _ = bridge.send_output(session, unfinished_turn_chunk());
@@ -281,6 +285,30 @@ fn pump_turn(bridge: &ChatBridge, session: &SessionId, gen: u64, events: Vec<Rep
     bridge.detach_if(session, gen);
 }
 
+#[test]
+fn pump_stops_at_final_without_waiting_for_provider_eof() {
+    let bridge = ChatBridge::new();
+    let session = sid();
+    let sink = Arc::new(Mutex::new(Vec::new()));
+    let gen = bridge.register(session, capturing_channel(Arc::clone(&sink)));
+
+    pump_turn(
+        &bridge,
+        &session,
+        gen,
+        vec![
+            ReplyEvent::Final {
+                content: "done".into(),
+            },
+            ReplyEvent::TextDelta {
+                text: "late provider output".into(),
+            },
+        ],
+    );
+
+    assert_eq!(sink.lock().unwrap().as_slice(), &[final_chunk("done")]);
+}
+
 #[test]
 fn pump_delivers_deltas_then_exactly_one_final_in_order() {
     let bridge = ChatBridge::new();
@@ -423,9 +451,14 @@ fn pump_with_existing_error_does_not_emit_generic_error() {
         &bridge,
         &session,
         gen,
-        vec![ReplyEvent::Error {
+        vec![
+            ReplyEvent::Error {
                 message: "auth failed".into(),
-        }],
+            },
+            ReplyEvent::TextDelta {
+                text: "late provider output".into(),
+            },
+        ],
     );
 
     assert_eq!(

crates/application/src/lib.rs

@@ -275,11 +275,11 @@ pub use template::{
     SyncAgentWithTemplateOutput, UpdateTemplate, UpdateTemplateInput, UpdateTemplateOutput,
 };
 pub use terminal::{
-    CancelAgentChat, CancelAgentChatOutput, CloseTerminal, CloseTerminalInput, CloseTerminalOutput,
-    LiveAgentRegistry, LiveRuntimeState, LiveRuntimeStatus, LiveSessionKind, LiveSessionSnapshot,
-    LiveSessions, OpenTerminal, OpenTerminalInput, OpenTerminalOutput, ResizeTerminal,
-    ResizeTerminalInput, StructuredSessions, TerminalSessions, WriteToTerminal,
-    WriteToTerminalInput,
+    release_turn_accounting, 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,

crates/application/src/terminal/mod.rs

@@ -33,8 +33,8 @@ pub use registry::{
     LiveSessions, StructuredSessions, TerminalSessions,
 };
 pub use usecases::{
-    stop_live_agent_runtime_in_project, CancelAgentChat, CancelAgentChatOutput, CloseTerminal,
-    CloseTerminalInput, CloseTerminalOutput, OpenTerminal, OpenTerminalInput, OpenTerminalOutput,
-    ResizeTerminal, ResizeTerminalInput, StoppedAgentRuntime, WriteToTerminal,
-    WriteToTerminalInput,
+    release_turn_accounting, stop_live_agent_runtime_in_project, CancelAgentChat,
+    CancelAgentChatOutput, CloseTerminal, CloseTerminalInput, CloseTerminalOutput, OpenTerminal,
+    OpenTerminalInput, OpenTerminalOutput, ResizeTerminal, ResizeTerminalInput,
+    StoppedAgentRuntime, WriteToTerminal, WriteToTerminalInput,
 };

crates/application/src/terminal/usecases.rs

@@ -26,6 +26,28 @@ pub struct CancelAgentChatOutput {
     pub outcome: CancelCurrentTurnOutcome,
 }
 
+/// Releases exactly one mediated turn without allowing late terminal cleanup to
+/// affect a successor that already owns the agent.
+///
+/// `cancel_head` and `mark_idle` are individually idempotent, while the ticket
+/// checks fence Final/close/timeout/cancel races across successive turns.
+pub fn release_turn_accounting(
+    input: &dyn InputMediator,
+    mailbox: &dyn AgentMailbox,
+    agent: domain::RuntimeAgentKey,
+    ticket: domain::mailbox::TicketId,
+) -> 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<StructuredSessions>,
@@ -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 {

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<dyn domain::input::InputMediator>,
+            Arc::clone(&mailbox) as Arc<dyn domain::mailbox::AgentMailbox>,
+            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<dyn domain::input::InputMediator>,
+            Arc::clone(&mailbox) as Arc<dyn domain::mailbox::AgentMailbox>,
+            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,
+            );
         }
     }
 }