fix(custom-cli): busy/progress/tools/chatbox bloqués après le premier appel — #245 (QA verte)

Backend : StructuredHumanTurnGuard pour partager la comptabilité busy/FIFO entre chat direct humain (agent_send, web chat.send) et asks inter-agents structurés, avec liveness mark_alive et retrait idempotent du tour. Web-server : tap d'événements sur chat.send pour le scrollback temps réel. Frontend : scoping des clés de progression par tour pour éviter la déduplication/ghosting des progress et outils entre appels.

crates/app-tauri/src/commands.rs

@@ -2564,6 +2564,16 @@ pub async fn agent_send(
         .iter()
         .map(|attachment| attachment.filename.clone())
         .collect();
+    let session_meta = state.structured_sessions.meta_for_session(&sid);
+    let turn_guard = session_meta.as_ref().map(|(project_id, agent_id, _, _)| {
+        backend::StructuredHumanTurnGuard::start(
+            std::sync::Arc::clone(&state.turn_watch_input),
+            std::sync::Arc::clone(&state.input_mailbox),
+            *project_id,
+            *agent_id,
+            &prompt_for_model,
+        )
+    });
 
     // Retain the human submit in the same live scrollback as model chunks so
     // `reattach_agent_chat` can repaint the whole in-flight conversation.
@@ -2583,9 +2593,15 @@ pub async fn agent_send(
     let (tap_tx, tap_rx) = std::sync::mpsc::channel();
     let tap_bridge = std::sync::Arc::clone(&state.chat_bridge);
     let tap_sid = sid;
+    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 !matches!(event, domain::ports::ReplyEvent::Final { .. }) {
+                if let Some(guard) = &tap_turn_guard {
+                    guard.mark_alive();
+                }
+            }
             let Some(chunk) = crate::chat::chunk_from_event(event) else {
                 continue;
             };
@@ -2617,11 +2633,16 @@ pub async fn agent_send(
     // event then feeds the service (détecter→planifier). Without a conversation id the
     // service must not pretend an automatic resume is armed.
     let service = std::sync::Arc::clone(&state.session_limit_service);
-    let meta = state.structured_sessions.meta_for_session(&sid);
+    let meta = session_meta;
     std::thread::spawn(move || {
         let mut terminal_visible = false;
         let mut settler = crate::chat::ProgressSettler::default();
         for event in stream {
+            if !matches!(event, domain::ports::ReplyEvent::Final { .. }) {
+                if let Some(guard) = &turn_guard {
+                    guard.mark_alive();
+                }
+            }
             // Non-terminal, sans contenu chat : un signal de limite alimente le service
             // (le badge UI vient du bus `AgentRateLimited`, pas du flux chat) puis on
             // continue à drainer comme pour un battement.
@@ -2663,6 +2684,9 @@ pub async fn agent_send(
             let chunk = crate::chat::unfinished_turn_chunk();
             let _ = bridge.send_output_settled(&sid, &mut settler, chunk);
         }
+        if let Some(guard) = &turn_guard {
+            guard.finish();
+        }
         bridge.detach_if(&sid, gen);
     });
 

crates/backend/src/lib.rs

@@ -836,6 +836,87 @@ struct AppAgentResumer {
     contexts: ResumeContexts,
 }
 
+/// Guard for a direct human turn sent to a structured agent session.
+///
+/// Direct chat adapters (`agent_send`, web `chat.send`) still drive the provider
+/// through `AgentSession::send`, but they must share the same busy/FIFO accounting
+/// as structured inter-agent asks. This guard starts a silent human turn and
+/// retires it on terminal outcome or drop.
+pub struct StructuredHumanTurnGuard {
+    input: Arc<dyn domain::input::InputMediator>,
+    mailbox: Arc<dyn domain::mailbox::AgentMailbox>,
+    agent: domain::RuntimeAgentKey,
+    ticket_id: domain::mailbox::TicketId,
+    armed: Arc<std::sync::atomic::AtomicBool>,
+    finish_on_drop: bool,
+}
+
+impl StructuredHumanTurnGuard {
+    /// Starts the mediated busy turn for a direct human structured chat message.
+    #[must_use]
+    pub fn start(
+        input: Arc<dyn domain::input::InputMediator>,
+        mailbox: Arc<dyn domain::mailbox::AgentMailbox>,
+        project_id: ProjectId,
+        agent_id: AgentId,
+        prompt: &str,
+    ) -> Self {
+        let agent = domain::RuntimeAgentKey::new(project_id, agent_id);
+        let ticket_id = domain::mailbox::TicketId::new_random();
+        let owns_turn = if input.busy_state(agent).is_busy() {
+            false
+        } else {
+            let ticket = domain::mailbox::Ticket::new(ticket_id, "user", prompt);
+            let _pending = input.enqueue_silent(agent, ticket);
+            input.busy_state(agent).ticket() == Some(ticket_id)
+        };
+        Self {
+            input,
+            mailbox,
+            agent,
+            ticket_id,
+            armed: Arc::new(std::sync::atomic::AtomicBool::new(owns_turn)),
+            finish_on_drop: true,
+        }
+    }
+
+    /// Returns a non-owning handle that can keep the turn alive from relay code.
+    #[must_use]
+    pub fn liveness_handle(&self) -> Self {
+        Self {
+            input: Arc::clone(&self.input),
+            mailbox: Arc::clone(&self.mailbox),
+            agent: self.agent,
+            ticket_id: self.ticket_id,
+            armed: Arc::clone(&self.armed),
+            finish_on_drop: false,
+        }
+    }
+
+    /// Records proof of liveness for the running turn.
+    pub fn mark_alive(&self) {
+        if self.armed.load(std::sync::atomic::Ordering::Acquire) {
+            self.input.mark_alive(self.agent);
+        }
+    }
+
+    /// Retires the FIFO head and marks the agent idle. Idempotent across clones.
+    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);
+        }
+    }
+}
+
+impl Drop for StructuredHumanTurnGuard {
+    fn drop(&mut self) {
+        if self.finish_on_drop {
+            self.finish();
+        }
+    }
+}
+
 #[async_trait::async_trait]
 impl AgentResumer for AppAgentResumer {
     async fn resume(
@@ -1384,6 +1465,9 @@ pub struct BackendCore {
     /// Médiateur d'entrée partagé, capturé pour câbler le callback `turn_ended` du
     /// [`turn_watcher`](Self::turn_watcher) à l'armement.
     pub turn_watch_input: Arc<dyn domain::input::InputMediator>,
+    /// Mailbox paired with [`Self::turn_watch_input`], used by direct structured
+    /// human turns to retire the synthetic busy/FIFO ticket on terminal outcome.
+    pub input_mailbox: Arc<dyn domain::mailbox::AgentMailbox>,
     /// Handles des watches de fin-de-tour vivants, par clé runtime. (Re)lancer un agent
     /// dans un projet **remplace** son handle (l'ancien est droppé ⇒ polling arrêté) ;
     /// fermer/arrêter l'agent le retire. `Mutex` car launch/stop y accèdent concurremment.
@@ -3259,6 +3343,7 @@ impl BackendCore {
             resume_contexts,
             turn_watcher,
             turn_watch_input: Arc::clone(&input_mediator),
+            input_mailbox: Arc::clone(&mailbox),
             turn_watch_handles: Mutex::new(HashMap::new()),
             fs_port: Arc::clone(&fs_port),
             home_dir,

crates/web-server/src/lib.rs

@@ -1870,13 +1870,50 @@ async fn ws_send_chat(
             code: "NOT_FOUND".to_owned(),
             message: "structured chat session not found".to_owned(),
         })?;
-    let stream = session.send(&prompt).await.map_err(|err| ErrorDto {
+    let session_meta = state.app.structured_sessions.meta_for_session(&sid);
+    let turn_guard = session_meta.as_ref().map(|(project_id, agent_id, _, _)| {
+        backend::StructuredHumanTurnGuard::start(
+            Arc::clone(&state.app.turn_watch_input),
+            Arc::clone(&state.app.input_mailbox),
+            *project_id,
+            *agent_id,
+            &prompt,
+        )
+    });
+
+    let (tap_tx, tap_rx) = std::sync::mpsc::channel();
+    let tap_out = tx.clone();
+    let tap_turn_guard = turn_guard.as_ref().map(|guard| guard.liveness_handle());
+    let tap_pump = std::thread::spawn(move || {
+        for event in tap_rx {
+            if !matches!(event, ReplyEvent::Final { .. }) {
+                if let Some(guard) = &tap_turn_guard {
+                    guard.mark_alive();
+                }
+            }
+            let chunk = reply_event_chunk(event);
+            if tap_out
+                .blocking_send(ServerFrame::chat_output(None, sid, chunk))
+                .is_err()
+            {
+                break;
+            }
+        }
+    });
+    let stream_result = session.send_with_tap(&prompt, tap_tx).await;
+    let _ = tap_pump.join();
+    let stream = stream_result.map_err(|err| ErrorDto {
         code: "CHAT_SEND_FAILED".to_owned(),
         message: err.to_string(),
     })?;
 
     let mut sent_ack = false;
     for event in stream {
+        if !matches!(event, ReplyEvent::Final { .. }) {
+            if let Some(guard) = &turn_guard {
+                guard.mark_alive();
+            }
+        }
         let chunk = reply_event_chunk(event);
         if !sent_ack {
             tx.send(ServerFrame::chat_output(Some(&frame.id), sid, chunk))
@@ -1909,6 +1946,10 @@ async fn ws_send_chat(
         })?;
     }
 
+    if let Some(guard) = &turn_guard {
+        guard.finish();
+    }
+
     Ok(())
 }
 
@@ -6820,6 +6861,7 @@ mod tests {
         id: SessionId,
         conversation_id: Option<String>,
         events: Vec<ReplyEvent>,
+        tap_events: Vec<ReplyEvent>,
         prompts: Mutex<Vec<String>>,
     }
 
@@ -6829,6 +6871,21 @@ mod tests {
                 id,
                 conversation_id: Some("engine-chat-session".to_owned()),
                 events,
+                tap_events: Vec::new(),
+                prompts: Mutex::new(Vec::new()),
+            }
+        }
+
+        fn with_tap_events(
+            id: SessionId,
+            events: Vec<ReplyEvent>,
+            tap_events: Vec<ReplyEvent>,
+        ) -> Self {
+            Self {
+                id,
+                conversation_id: Some("engine-chat-session".to_owned()),
+                events,
+                tap_events,
                 prompts: Mutex::new(Vec::new()),
             }
         }
@@ -6852,6 +6909,92 @@ mod tests {
             Ok(Box::new(self.events.clone().into_iter()))
         }
 
+        async fn send_with_tap(
+            &self,
+            prompt: &str,
+            tap: std::sync::mpsc::Sender<ReplyEvent>,
+        ) -> Result<ReplyStream, AgentSessionError> {
+            for event in self.tap_events.clone() {
+                let _ = tap.send(event);
+            }
+            self.send(prompt).await
+        }
+
+        async fn shutdown(&self) -> Result<(), AgentSessionError> {
+            Ok(())
+        }
+    }
+
+    struct GatedChatSession {
+        id: SessionId,
+        prompts: Mutex<Vec<String>>,
+        started: Mutex<Option<oneshot::Sender<()>>>,
+        release: Mutex<Option<oneshot::Receiver<()>>>,
+    }
+
+    impl GatedChatSession {
+        fn new(
+            id: SessionId,
+            started: oneshot::Sender<()>,
+            release: oneshot::Receiver<()>,
+        ) -> Self {
+            Self {
+                id,
+                prompts: Mutex::new(Vec::new()),
+                started: Mutex::new(Some(started)),
+                release: Mutex::new(Some(release)),
+            }
+        }
+    }
+
+    #[async_trait]
+    impl AgentSession for GatedChatSession {
+        fn id(&self) -> SessionId {
+            self.id
+        }
+
+        fn conversation_id(&self) -> Option<String> {
+            Some("engine-gated-chat-session".to_owned())
+        }
+
+        async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
+            self.send_with_tap(prompt, std::sync::mpsc::channel().0)
+                .await
+        }
+
+        async fn send_with_tap(
+            &self,
+            prompt: &str,
+            _tap: std::sync::mpsc::Sender<ReplyEvent>,
+        ) -> Result<ReplyStream, AgentSessionError> {
+            self.prompts
+                .lock()
+                .expect("prompt trace mutex poisoned")
+                .push(prompt.to_owned());
+            if let Some(started) = self
+                .started
+                .lock()
+                .expect("started gate mutex poisoned")
+                .take()
+            {
+                let _ = started.send(());
+            }
+            let release = self
+                .release
+                .lock()
+                .expect("release gate mutex poisoned")
+                .take();
+            if let Some(release) = release {
+                let _ = release.await;
+            }
+            Ok(Box::new(
+                vec![ReplyEvent::Final {
+                    content: "released".to_owned(),
+                }]
+                .into_iter(),
+            ))
+        }
+
         async fn shutdown(&self) -> Result<(), AgentSessionError> {
             Ok(())
         }
@@ -7533,6 +7676,130 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn websocket_chat_send_marks_direct_structured_turn_busy_until_final() {
+        let state = state();
+        let session_id = SessionId::new_random();
+        let agent_id = AgentId::from_uuid(Uuid::new_v4());
+        let node_id = NodeId::from_uuid(Uuid::new_v4());
+        let (started_tx, started_rx) = oneshot::channel();
+        let (release_tx, release_rx) = oneshot::channel();
+        let session = Arc::new(GatedChatSession::new(session_id, started_tx, release_rx));
+        state.app.structured_sessions.insert(
+            Arc::clone(&session) as Arc<dyn AgentSession>,
+            agent_id,
+            node_id,
+        );
+        let runtime_key = domain::RuntimeAgentKey::new(ProjectId::from_uuid(Uuid::nil()), agent_id);
+        let (tx, mut rx) = mpsc::channel(16);
+        let owned = Arc::new(Mutex::new(Vec::new()));
+        let frame = ws_frame(
+            "chat-send-busy",
+            "chat.send",
+            json!({ "sessionId": session_id.to_string(), "prompt": "Work" }),
+        );
+
+        let send_task = tokio::spawn({
+            let state = Arc::clone(&state);
+            let tx = tx.clone();
+            let owned = Arc::clone(&owned);
+            async move {
+                handle_client_frame(frame, &state, &tx, &owned).await;
+            }
+        });
+        started_rx.await.expect("gated session started");
+
+        assert!(
+            state.app.turn_watch_input.busy_state(runtime_key).is_busy(),
+            "direct structured human turn should be busy while send is in flight"
+        );
+
+        release_tx.send(()).expect("release gated session");
+        let final_chunk = recv_server_frame(&mut rx).await;
+        send_task.await.expect("chat send task finished");
+
+        assert_eq!(final_chunk.kind, "chat.output");
+        assert_eq!(final_chunk.reply_to.as_deref(), Some("chat-send-busy"));
+        assert_eq!(
+            final_chunk.payload["chunk"],
+            json!({ "kind": "final", "content": "released" })
+        );
+        assert!(
+            !state.app.turn_watch_input.busy_state(runtime_key).is_busy(),
+            "direct structured human turn should be idle after final"
+        );
+    }
+
+    #[tokio::test]
+    async fn websocket_chat_send_relays_tap_progress_live() {
+        let state = state();
+        let session_id = SessionId::new_random();
+        let agent_id = AgentId::from_uuid(Uuid::new_v4());
+        let node_id = NodeId::from_uuid(Uuid::new_v4());
+        let session = Arc::new(FakeChatSession::with_tap_events(
+            session_id,
+            vec![ReplyEvent::Final {
+                content: "done".to_owned(),
+            }],
+            vec![ReplyEvent::Progress {
+                progress: ReplyProgress::new(
+                    ReplyProgressSource::IdeaLocal,
+                    ReplyProgressKind::Mcp,
+                    ReplyProgressStage::Started,
+                    "idea_ask_agent",
+                )
+                .with_tool_name("idea_ask_agent"),
+            }],
+        ));
+        state.app.structured_sessions.insert(
+            Arc::clone(&session) as Arc<dyn AgentSession>,
+            agent_id,
+            node_id,
+        );
+        let (tx, mut rx) = mpsc::channel(16);
+        let owned = Arc::new(Mutex::new(Vec::new()));
+
+        handle_client_frame(
+            ws_frame(
+                "chat-send-progress",
+                "chat.send",
+                json!({ "sessionId": session_id.to_string(), "prompt": "Run tool" }),
+            ),
+            &state,
+            &tx,
+            &owned,
+        )
+        .await;
+
+        let first = recv_server_frame(&mut rx).await;
+        let second = recv_server_frame(&mut rx).await;
+        let frames = [first, second];
+        let progress = frames
+            .iter()
+            .find(|frame| frame.payload["chunk"]["kind"] == "progress")
+            .expect("tap progress frame was relayed");
+        let final_chunk = frames
+            .iter()
+            .find(|frame| frame.payload["chunk"]["kind"] == "final")
+            .expect("final frame was relayed");
+
+        assert_eq!(progress.kind, "chat.output");
+        assert!(progress.reply_to.is_none());
+        assert_eq!(progress.payload["sessionId"], session_id.to_string());
+        assert_eq!(progress.payload["chunk"]["progress"]["source"], "ideaLocal");
+        assert_eq!(progress.payload["chunk"]["progress"]["kind"], "mcp");
+        assert_eq!(progress.payload["chunk"]["progress"]["stage"], "started");
+        assert_eq!(
+            progress.payload["chunk"]["progress"]["toolName"],
+            "idea_ask_agent"
+        );
+        assert_eq!(final_chunk.reply_to.as_deref(), Some("chat-send-progress"));
+        assert_eq!(
+            final_chunk.payload["chunk"],
+            json!({ "kind": "final", "content": "done" })
+        );
+    }
+
     #[tokio::test]
     async fn websocket_chat_detach_unregisters_sink_without_killing_session() {
         let state = state();