Merge remote-tracking branch 'origin/main' into tritone-2

The adapter returns the session/prompt response before the agent
finishes working.  Late-arriving notifications raced past the turn
marker and leaked into the next turn, making each response appear
one prompt behind.

The event pump now sends a settled oneshot through the done payload
and awaits it before processing further messages.
finish_queued_acp_turn fires it after persisting the turn marker and
dispatching any queued prompt.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

crates/maestro-acp/src/lib.rs

@@ -69,13 +69,28 @@ pub enum AcpUpdate {
     /// turn precede this marker on the wire, and therefore in this channel.
     /// The event pump must fire `done` after persisting every preceding
     /// notification.
+    ///
+    /// The `done` oneshot carries both the prompt result and a `settled_tx`
+    /// that the receiver fires after the turn marker is persisted.  The
+    /// event pump awaits the corresponding `settled_rx` before processing
+    /// further messages so late-arriving notifications cannot leak into the
+    /// next turn.
     TurnDone {
         turn: u64,
         result: Result<(), String>,
-        done: oneshot::Sender<Result<(), String>>,
+        done: oneshot::Sender<TurnDonePayload>,
     },
 }
 
+/// Payload sent through the `done` oneshot inside [`AcpUpdate::TurnDone`].
+#[derive(Debug)]
+pub struct TurnDonePayload {
+    pub result: Result<(), String>,
+    /// The receiver fires this after persisting the turn marker and
+    /// dispatching any queued prompt.
+    pub settled: oneshot::Sender<()>,
+}
+
 /// ACP adapter selected from the structured session's resolved command.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum AcpAdapter {
@@ -301,7 +316,7 @@ enum Command {
     Prompt {
         text: String,
         turn: u64,
-        done: oneshot::Sender<Result<(), String>>,
+        done: oneshot::Sender<TurnDonePayload>,
     },
     SetConfig {
         config_id: String,
@@ -435,7 +450,7 @@ impl AcpSession {
         &self,
         text: String,
         turn: u64,
-    ) -> Result<oneshot::Receiver<Result<(), String>>, String> {
+    ) -> Result<oneshot::Receiver<TurnDonePayload>, String> {
         let (done_tx, done_rx) = oneshot::channel();
         self.commands
             .send(Command::Prompt {
@@ -999,7 +1014,11 @@ async fn run_connection<T>(
                                     .expect("permission turn state")
                                     .finish(turn);
                                 if let Some(done) = done.lock().expect("prompt sender").take() {
-                                    let _ = done.send(Err(error.to_string()));
+                                    let (settled_tx, _) = oneshot::channel();
+                                    let _ = done.send(TurnDonePayload {
+                                        result: Err(error.to_string()),
+                                        settled: settled_tx,
+                                    });
                                 }
                             }
                         }

crates/maestro-daemon/src/session.rs

@@ -869,11 +869,24 @@ impl SessionManager {
                             stream.dispatch.queued_turn = Some(turn);
                             let manager = self.clone();
                             tokio::spawn(async move {
-                                let lifecycle = match done.await {
-                                    Ok(Ok(())) => TurnLifecycle::Finish,
-                                    Ok(Err(_)) | Err(_) => TurnLifecycle::Failure,
+                                let (lifecycle, settled) = match done.await {
+                                    Ok(payload) => {
+                                        let lifecycle = if payload.result.is_ok() {
+                                            TurnLifecycle::Finish
+                                        } else {
+                                            TurnLifecycle::Failure
+                                        };
+                                        (lifecycle, Some(payload.settled))
+                                    }
+                                    Err(_) => (TurnLifecycle::Failure, None),
                                 };
                                 manager.finish_queued_acp_turn(id, turn, lifecycle).await;
+                                // Tell the event pump it may resume processing
+                                // notifications — the turn marker is persisted
+                                // and any queued prompt has been dispatched.
+                                if let Some(settled) = settled {
+                                    let _ = settled.send(());
+                                }
                             });
                             drop(stream);
                             sync_structured_activity(&self.inner, id).await;
@@ -2443,8 +2456,17 @@ fn start_acp_event_pump(
                 } => {
                     // Every notification for this turn has been persisted and
                     // broadcast above.  Signal the done oneshot so
-                    // finish_queued_acp_turn runs with the transcript complete.
-                    let _ = done.send(result);
+                    // finish_queued_acp_turn runs with the transcript complete,
+                    // then wait for it to persist the turn marker before we
+                    // resume — otherwise late-arriving notifications race into
+                    // the next turn.
+                    let (settled_tx, settled_rx) = tokio::sync::oneshot::channel();
+                    let _ = done.send(maestro_acp::TurnDonePayload {
+                        result,
+                        settled: settled_tx,
+                    });
+                    // Block the pump until the turn is fully settled.
+                    let _ = settled_rx.await;
                 }
             }
         }