fix: serialize structured prompt dispatch

crates/maestro-daemon/src/session.rs

@@ -122,6 +122,18 @@ struct AcpEventStream {
     projection: Projection,
     projected_live: broadcast::Sender<ProjectionUpdate>,
     pending_approvals: HashMap<String, PendingApproval>,
+    /// Serializes prompt dispatch with cancellation. A prompt is not merely
+    /// claimed: it is queued onto ACP before another operation can proceed.
+    operation: Arc<Mutex<()>>,
+    dispatch: AcpDispatchState,
+}
+
+#[derive(Debug, Default)]
+struct AcpDispatchState {
+    queued_turn: Option<u64>,
+    cancelled_turn: Option<u64>,
+    failed: bool,
+    settled: Arc<tokio::sync::Notify>,
 }
 
 #[derive(Debug)]
@@ -174,6 +186,15 @@ pub enum BeginAcpTurn {
     Unavailable,
 }
 
+/// Result of serializing a projected turn claim with its ACP prompt enqueue.
+#[derive(Debug)]
+pub enum QueueAcpPrompt {
+    Queued,
+    InFlight,
+    Unavailable,
+    Failed(String),
+}
+
 impl SessionError {
     pub fn to_error_info(&self) -> ErrorInfo {
         let code = match self {
@@ -540,6 +561,150 @@ impl SessionManager {
         Ok(BeginAcpTurn::Started(acp, turn))
     }
 
+    /// Claim a turn and enqueue its ACP prompt as one serialized operation.
+    ///
+    /// ACP permission callbacks do not carry a prompt id. Therefore a
+    /// cancelled prompt must settle before a later prompt is dispatched. If
+    /// it does not settle within the cancel grace, the session is failed
+    /// instead of letting a later callback become ambiguous.
+    pub async fn queue_acp_prompt(
+        &self,
+        id: Uuid,
+        text: String,
+    ) -> Result<QueueAcpPrompt, SessionError> {
+        let (acp, stream) = {
+            let sessions = self.inner.sessions.lock().await;
+            let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
+            if !record.is_live_structured() {
+                return Ok(QueueAcpPrompt::Unavailable);
+            }
+            (record.acp.clone(), record.acp_events.clone())
+        };
+        let (Some(acp), Some(stream)) = (acp, stream) else {
+            return Ok(QueueAcpPrompt::Unavailable);
+        };
+        let operation = stream.lock().await.operation.clone();
+        let _operation = operation.lock().await;
+
+        loop {
+            let waiting_for = {
+                let mut stream = stream.lock().await;
+                if stream.dispatch.failed {
+                    return Ok(QueueAcpPrompt::Unavailable);
+                }
+                if let Some(turn) = stream.dispatch.queued_turn {
+                    if stream.dispatch.cancelled_turn != Some(turn) {
+                        return Ok(QueueAcpPrompt::InFlight);
+                    }
+                    Some((turn, stream.dispatch.settled.clone()))
+                } else {
+                    if stream.projection.turn_active() {
+                        return Ok(QueueAcpPrompt::InFlight);
+                    }
+                    let (turn, update) = stream.projection.begin_turn();
+                    let _ = stream.projected_live.send(update);
+                    match acp.prompt(text.clone(), turn) {
+                        Ok(done) => {
+                            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,
+                                };
+                                manager.finish_queued_acp_turn(id, turn, lifecycle).await;
+                            });
+                            return Ok(QueueAcpPrompt::Queued);
+                        }
+                        Err(message) => {
+                            cancel_pending_approvals(&mut stream);
+                            let projected = stream
+                                .projection
+                                .turn_lifecycle(TurnLifecycle::Failure, turn)
+                                .expect(
+                                    "newly claimed turn remains current while stream is locked",
+                                );
+                            let _ = stream.projected_live.send(projected);
+                            return Ok(QueueAcpPrompt::Failed(message));
+                        }
+                    }
+                }
+            };
+            let Some((turn, settled)) = waiting_for else {
+                continue;
+            };
+            if tokio::time::timeout(maestro_acp::CANCEL_GRACE, settled.notified())
+                .await
+                .is_err()
+            {
+                self.fail_cancelled_acp_turn(id, &stream, turn).await;
+            }
+        }
+    }
+
+    async fn finish_queued_acp_turn(&self, id: Uuid, turn: u64, lifecycle: TurnLifecycle) {
+        let stream = self
+            .inner
+            .sessions
+            .lock()
+            .await
+            .get(&id)
+            .and_then(|record| record.acp_events.clone());
+        let Some(stream) = stream else {
+            return;
+        };
+        let mut stream = stream.lock().await;
+        if stream.dispatch.queued_turn != Some(turn) {
+            return;
+        }
+        stream.dispatch.queued_turn = None;
+        if stream.dispatch.cancelled_turn == Some(turn) {
+            stream.dispatch.cancelled_turn = None;
+        }
+        stream.dispatch.settled.notify_waiters();
+        if stream.projection.active_turn() != Some(turn) {
+            return;
+        }
+        cancel_pending_approvals(&mut stream);
+        if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) {
+            let _ = stream.projected_live.send(projected);
+        }
+    }
+
+    async fn fail_cancelled_acp_turn(
+        &self,
+        id: Uuid,
+        stream: &Arc<Mutex<AcpEventStream>>,
+        turn: u64,
+    ) {
+        let acp = self
+            .inner
+            .sessions
+            .lock()
+            .await
+            .get(&id)
+            .and_then(|record| record.acp.clone());
+        {
+            let mut stream = stream.lock().await;
+            if stream.dispatch.queued_turn != Some(turn)
+                || stream.dispatch.cancelled_turn != Some(turn)
+            {
+                return;
+            }
+            stream.dispatch.failed = true;
+            stream.dispatch.queued_turn = None;
+            stream.dispatch.cancelled_turn = None;
+            stream.dispatch.settled.notify_waiters();
+        }
+        mark_session_dead(&self.inner, id).await;
+        if let Some(acp) = acp {
+            tokio::spawn(async move {
+                acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
+                    .await;
+            });
+        }
+    }
+
     /// Cancel an ACP turn and immediately clear its client-facing lifecycle.
     pub async fn cancel_acp_turn(&self, id: Uuid) -> Result<bool, SessionError> {
         let (acp, stream) = {
@@ -553,10 +718,21 @@ impl SessionManager {
         let (Some(acp), Some(stream)) = (acp, stream) else {
             return Ok(false);
         };
-        let mut stream = stream.lock().await;
+        let event_stream = stream;
+        let operation = event_stream.lock().await.operation.clone();
+        let _operation = operation.lock().await;
+        let mut stream = event_stream.lock().await;
         acp.cancel();
         cancel_pending_approvals(&mut stream);
+        let cancelled_queued_turn = stream.dispatch.queued_turn;
+        if let Some(turn) = cancelled_queued_turn {
+            stream.dispatch.cancelled_turn = Some(turn);
+        }
         let Some(turn) = stream.projection.active_turn() else {
+            drop(stream);
+            if let Some(turn) = cancelled_queued_turn {
+                self.schedule_cancelled_acp_turn_failure(id, event_stream, turn);
+            }
             return Ok(true);
         };
         let projected = stream
@@ -564,9 +740,26 @@ impl SessionManager {
             .turn_lifecycle(TurnLifecycle::Cancel, turn)
             .expect("active turn remains current while stream is locked");
         let _ = stream.projected_live.send(projected);
+        drop(stream);
+        if let Some(turn) = cancelled_queued_turn {
+            self.schedule_cancelled_acp_turn_failure(id, event_stream, turn);
+        }
         Ok(true)
     }
 
+    fn schedule_cancelled_acp_turn_failure(
+        &self,
+        id: Uuid,
+        stream: Arc<Mutex<AcpEventStream>>,
+        turn: u64,
+    ) {
+        let manager = self.clone();
+        tokio::spawn(async move {
+            tokio::time::sleep(maestro_acp::CANCEL_GRACE).await;
+            manager.fail_cancelled_acp_turn(id, &stream, turn).await;
+        });
+    }
+
     /// Resolve a daemon-issued approval nonce. Missing or already-consumed
     /// nonces deliberately succeed: two clients racing to answer must not
     /// turn a harmless duplicate click into an error.
@@ -897,6 +1090,8 @@ impl SessionManager {
                         projection,
                         projected_live: projected_tx,
                         pending_approvals: HashMap::new(),
+                        operation: Arc::new(Mutex::new(())),
+                        dispatch: AcpDispatchState::default(),
                     })))
                 }
                 Err(e) => {
@@ -1341,6 +1536,8 @@ impl SessionManager {
                             projection,
                             projected_live: projected_tx,
                             pending_approvals: HashMap::new(),
+                            operation: Arc::new(Mutex::new(())),
+                            dispatch: AcpDispatchState::default(),
                         })))
                     }
                     Err(error) => {