fix: serialize structured prompt dispatch

crates/maestro-acp/src/lib.rs

@@ -113,6 +113,15 @@ impl SpawnConfig {
     }
 }
 
+/// A blocked ACP permission request delivered to the daemon.
+#[derive(Debug)]
+pub struct PermissionRequest {
+    /// The prompt turn that was active when ACP delivered this callback.
+    pub turn: u64,
+    pub request: RequestPermissionRequest,
+    pub response: oneshot::Sender<RequestPermissionOutcome>,
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum AcpError {
     /// The adapter is not on `PATH`. The message names the install command
@@ -156,6 +165,7 @@ enum Command {
     Cancel,
     Prompt {
         text: String,
+        turn: u64,
         done: oneshot::Sender<Result<(), String>>,
     },
     Close,
@@ -166,6 +176,31 @@ enum ChildCommand {
     Reap(oneshot::Sender<bool>),
 }
 
+#[derive(Debug, Default)]
+struct PermissionTurnState {
+    active: Option<u64>,
+}
+
+impl PermissionTurnState {
+    fn begin(&mut self, turn: u64) {
+        self.active = Some(turn);
+    }
+
+    fn cancel_active(&mut self) {
+        self.active = None;
+    }
+
+    fn finish(&mut self, turn: u64) {
+        if self.active == Some(turn) {
+            self.active = None;
+        }
+    }
+
+    fn callback_turn(&self) -> Option<u64> {
+        self.active
+    }
+}
+
 /// Why the handshake did not produce a session id.
 enum ReadyError {
     /// The adapter answered with an error.
@@ -192,6 +227,7 @@ pub struct AcpSession {
     commands: mpsc::UnboundedSender<Command>,
     child_commands: mpsc::UnboundedSender<ChildCommand>,
     updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
+    permission_requests: Mutex<Option<mpsc::UnboundedReceiver<PermissionRequest>>>,
     exit: watch::Receiver<Option<ExitStatus>>,
 }
 
@@ -235,6 +271,13 @@ impl AcpSession {
         self.updates.lock().expect("updates mutex").take()
     }
 
+    pub fn take_permission_requests(&self) -> Option<mpsc::UnboundedReceiver<PermissionRequest>> {
+        self.permission_requests
+            .lock()
+            .expect("permission requests")
+            .take()
+    }
+
     /// Send `session/cancel`. Fire-and-forget: a closed connection means the
     /// child is already gone, which is the outcome cancel was after.
     pub fn cancel(&self) {
@@ -242,11 +285,16 @@ impl AcpSession {
     }
 
     /// Send a user prompt and receive completion once the agent ends the turn.
-    pub fn prompt(&self, text: String) -> Result<oneshot::Receiver<Result<(), String>>, String> {
+    pub fn prompt(
+        &self,
+        text: String,
+        turn: u64,
+    ) -> Result<oneshot::Receiver<Result<(), String>>, String> {
         let (done_tx, done_rx) = oneshot::channel();
         self.commands
             .send(Command::Prompt {
                 text,
+                turn,
                 done: done_tx,
             })
             .map_err(|_| "ACP connection is closed".to_string())?;
@@ -335,6 +383,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
     tokio::spawn(supervise_child(child, pid, child_commands_rx, exit_tx));
 
     let (updates_tx, updates_rx) = mpsc::unbounded_channel();
+    let (permission_tx, permission_rx) = mpsc::unbounded_channel();
     let (commands_tx, commands_rx) = mpsc::unbounded_channel();
     let (ready_tx, ready_rx) = oneshot::channel();
     let transport = ByteStreams::new(stdin.compat_write(), stdout.compat());
@@ -343,6 +392,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
         config.cwd.clone(),
         config.load_session.clone(),
         updates_tx,
+        permission_tx,
         commands_rx,
         ready_tx,
     ));
@@ -356,6 +406,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
                 commands: commands_tx,
                 child_commands: child_commands_tx,
                 updates: Mutex::new(Some(updates_rx)),
+                permission_requests: Mutex::new(Some(permission_rx)),
                 exit: exit_rx,
             });
         }
@@ -467,6 +518,7 @@ async fn run_connection<T>(
     cwd: PathBuf,
     load_session: Option<String>,
     updates_tx: mpsc::UnboundedSender<SessionNotification>,
+    permission_requests: mpsc::UnboundedSender<PermissionRequest>,
     mut commands_rx: mpsc::UnboundedReceiver<Command>,
     ready_tx: oneshot::Sender<Result<String, ReadyError>>,
 ) where
@@ -478,6 +530,10 @@ async fn run_connection<T>(
     // history the event log already holds (#135).
     let replaying = Arc::new(AtomicBool::new(false));
     let replaying_seen = replaying.clone();
+    // ACP callbacks have no turn id. The daemon never queues a later prompt
+    // until a cancelled prompt settles, so this active turn is unambiguous.
+    let permission_turn = Arc::new(Mutex::new(PermissionTurnState::default()));
+    let permission_turn_seen = permission_turn.clone();
     let result = Client
         .builder()
         .name("maestro")
@@ -495,15 +551,30 @@ async fn run_connection<T>(
         )
         .on_receive_request(
             async move |request: RequestPermissionRequest, responder, _cx| {
-                // Stub until #139: decline rather than grant something the
+                let turn = permission_turn_seen
-                // operator never saw.
+                    .lock()
-                tracing::warn!(
+                    .expect("permission turn state")
-                    session = %request.session_id,
+                    .callback_turn();
-                    "session/request_permission declined: not implemented yet (#139)"
+                let Some(turn) = turn else {
-                );
+                    return responder.respond(RequestPermissionResponse::new(
-                responder.respond(RequestPermissionResponse::new(
+                        RequestPermissionOutcome::Cancelled,
-                    RequestPermissionOutcome::Cancelled,
+                    ));
-                ))
+                };
+                let (response, received) = oneshot::channel();
+                if permission_requests
+                    .send(PermissionRequest {
+                        turn,
+                        request,
+                        response,
+                    })
+                    .is_err()
+                {
+                    return responder.respond(RequestPermissionResponse::new(
+                        RequestPermissionOutcome::Cancelled,
+                    ));
+                }
+                let outcome = received.await.unwrap_or(RequestPermissionOutcome::Cancelled);
+                responder.respond(RequestPermissionResponse::new(outcome))
             },
             agent_client_protocol::on_receive_request!(),
         )
@@ -575,26 +646,44 @@ async fn run_connection<T>(
                 tokio::select! {
                     cmd = commands_rx.recv() => match cmd {
                         Some(Command::Cancel) => {
+                            permission_turn
+                                .lock()
+                                .expect("permission turn state")
+                                .cancel_active();
                             let _ = cx.send_notification(CancelNotification::new(
                                 SessionId::new(session_id.as_str()),
                             ));
                         }
-                        Some(Command::Prompt { text, done }) => {
+                        Some(Command::Prompt { text, turn, done }) => {
+                            permission_turn
+                                .lock()
+                                .expect("permission turn state")
+                                .begin(turn);
                             let request = PromptRequest::new(
                                 SessionId::new(session_id.as_str()),
                                 vec![ContentBlock::from(text)],
                             );
                             let done = Arc::new(Mutex::new(Some(done)));
                             let callback_done = done.clone();
+                            let callback_turn = permission_turn.clone();
                             if let Err(error) = cx.send_request(request).on_receiving_result(move |result| {
                                 let callback_done = callback_done.clone();
+                                let callback_turn = callback_turn.clone();
                                 async move {
+                                    callback_turn
+                                        .lock()
+                                        .expect("permission turn state")
+                                        .finish(turn);
                                     if let Some(done) = callback_done.lock().expect("prompt sender").take() {
                                         let _ = done.send(result.map(|_| ()).map_err(|e| e.to_string()));
                                     }
                                     Ok(())
                                 }
                             }) {
+                                permission_turn
+                                    .lock()
+                                    .expect("permission turn state")
+                                    .finish(turn);
                                 if let Some(done) = done.lock().expect("prompt sender").take() {
                                     let _ = done.send(Err(error.to_string()));
                                 }

crates/maestro-acp/tests/fake_acp_agent.py

@@ -24,6 +24,7 @@ Environment knobs, all optional:
                              so the SIGTERM path is exercised)
   FAKE_ACP_IGNORE_EOF=1      keep running after stdin closes, until signalled
   FAKE_ACP_EMIT_ON_CANCEL=1  emit a final notification before handling cancel
+  FAKE_ACP_CANCEL_SETTLES=1  complete the outstanding prompt on cancel
   FAKE_ACP_NO_LOAD=1      advertise `loadSession: false`
   FAKE_ACP_LOAD_REPLAY=n  on `session/load`, replay n `agent_message_chunk`
                             notifications ("replayed <i>") before answering,
@@ -33,6 +34,8 @@ Environment knobs, all optional:
   FAKE_ACP_PROMPT_FAIL=1  answer every `session/prompt` with a JSON-RPC error
   FAKE_ACP_PROMPT_FAIL_ONCE=1  answer only the first `session/prompt` with a
                              JSON-RPC error
+  FAKE_ACP_PERMISSION_AFTER_CANCEL=1  defer the first prompt's permission
+                             callback until a later prompt
 
 Signals: SIGUSR1 emits one `agent_message_chunk` notification ("tick <n>",
 counting from 1) and logs `emit tick <n>`, so a test can make the agent
@@ -70,6 +73,9 @@ signal.signal(signal.SIGTERM, on_sigterm)
 session_id = f"fake-{os.getpid()}"
 ticks = 0
 prompt_failures = 0
+pending_prompt_id = None
+pending_permission_id = None
+delayed_permission = False
 
 
 def chunk(text):
@@ -102,6 +108,12 @@ for raw in sys.stdin:
     method = msg.get("method")
     if method is None:
         # A response to something we sent; nothing to do.
+        if msg.get("id") == pending_permission_id:
+            log("permission response " + json.dumps(msg.get("result"), sort_keys=True))
+            send({"jsonrpc": "2.0", "id": pending_prompt_id,
+                  "result": {"stopReason": "end_turn"}})
+            pending_prompt_id = None
+            pending_permission_id = None
         continue
     log(f"recv {method}")
     rid = msg.get("id")
@@ -140,9 +152,47 @@ for raw in sys.stdin:
     elif method == "session/cancel":
         if os.environ.get("FAKE_ACP_EMIT_ON_CANCEL") == "1":
             send(chunk("final on cancel"))
+        if (os.environ.get("FAKE_ACP_CANCEL_SETTLES") == "1"
+                and pending_prompt_id is not None):
+            send({"jsonrpc": "2.0", "id": pending_prompt_id,
+                  "result": {"stopReason": "cancelled"}})
+            pending_prompt_id = None
+            pending_permission_id = None
         if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1":
             sys.exit(0)
     elif method == "session/prompt":
+        if os.environ.get("FAKE_ACP_PERMISSION_AFTER_CANCEL") == "1":
+            if pending_prompt_id is None:
+                pending_prompt_id = rid
+                delayed_permission = True
+                continue
+            if delayed_permission:
+                delayed_permission = False
+                pending_permission_id = "permission-1"
+                send({"jsonrpc": "2.0", "id": pending_permission_id,
+                      "method": "session/request_permission", "params": {
+                          "sessionId": session_id,
+                          "toolCall": {"toolCallId": "tool-1", "fields": {}},
+                          "options": [
+                              {"optionId": "once", "name": "Allow once", "kind": "allow_once"},
+                              {"optionId": "deny", "name": "Deny", "kind": "reject_once"}
+                          ]
+                      }})
+                continue
+        if os.environ.get("FAKE_ACP_REQUEST_PERMISSION") == "1":
+            pending_prompt_id = rid
+            pending_permission_id = "permission-1"
+            send({"jsonrpc": "2.0", "id": pending_permission_id,
+                  "method": "session/request_permission", "params": {
+                      "sessionId": session_id,
+                      "toolCall": {"toolCallId": "tool-1", "fields": {}},
+                      "options": [
+                          {"optionId": "once", "name": "Allow once", "kind": "allow_once"},
+                          {"optionId": "session", "name": "Allow session", "kind": "allow_always"},
+                          {"optionId": "deny", "name": "Deny", "kind": "reject_once"}
+                      ]
+                  }})
+            continue
         time.sleep(int(os.environ.get("FAKE_ACP_DELAY_PROMPT_MS", "0")) / 1000)
         fail_once = os.environ.get("FAKE_ACP_PROMPT_FAIL_ONCE") == "1"
         if os.environ.get("FAKE_ACP_PROMPT_FAIL") == "1" or (fail_once and prompt_failures == 0):

crates/maestro-daemon/src/local_api.rs

@@ -21,13 +21,12 @@ use maestro_config::SpawnPreset;
 use maestro_protocol::{
     routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonInfo, GitRoot, NameBatch,
     RenameSessionRequest, ReportActivityRequest, SendInputRequest, Session, SetFocusRequest,
-    StructuredEvent, StructuredPromptRequest, PROTOCOL_VERSION,
+    StructuredApprovalDecision, StructuredEvent, StructuredPromptRequest, PROTOCOL_VERSION,
 };
 use serde::Deserialize;
 use uuid::Uuid;
 
-use crate::projection::TurnLifecycle;
+use crate::session::{QueueAcpPrompt, SessionManager};
-use crate::session::SessionManager;
 use crate::sidecar::EventCursor;
 
 #[derive(Debug, Clone)]
@@ -70,7 +69,11 @@ pub fn router(state: LocalApiState) -> Router {
         .route(routes::patterns::STRUCTURED, any(structured_ws))
         .route(routes::patterns::STRUCTURED_REPLAY, get(structured_replay))
         .route(routes::patterns::STRUCTURED_PROMPT, post(structured_prompt))
-        .route(routes::patterns::STRUCTURED_CANCEL, post(structured_cancel));
+        .route(routes::patterns::STRUCTURED_CANCEL, post(structured_cancel))
+        .route(
+            routes::patterns::STRUCTURED_APPROVAL,
+            post(structured_approval),
+        );
     #[cfg(feature = "haiku-autoname")]
     {
         router = router.route(routes::patterns::SESSION_PROMPT, post(session_prompt));
@@ -432,45 +435,24 @@ async fn structured_prompt(
     Path(id): Path<Uuid>,
     Json(body): Json<StructuredPromptRequest>,
 ) -> Response {
-    let (acp, turn) = match s.manager.begin_acp_turn(id).await {
+    match s.manager.queue_acp_prompt(id, body.prompt).await {
-        Ok(crate::session::BeginAcpTurn::Started(acp, turn)) => (acp, turn),
+        Ok(QueueAcpPrompt::Queued) => StatusCode::NO_CONTENT.into_response(),
-        Ok(crate::session::BeginAcpTurn::InFlight) => {
+        Ok(QueueAcpPrompt::InFlight) => error_response(
-            return error_response(
+            StatusCode::CONFLICT,
-                StatusCode::CONFLICT,
+            "turn_in_flight",
-                "turn_in_flight",
+            "the structured session already has an active turn",
-                "the structured session already has an active turn",
+        ),
-            )
+        Ok(QueueAcpPrompt::Unavailable) => error_response(
-        }
+            StatusCode::CONFLICT,
-        Ok(crate::session::BeginAcpTurn::Unavailable) => {
+            "not_structured",
-            return error_response(
+            "session is not a live structured session",
-                StatusCode::CONFLICT,
+        ),
-                "not_structured",
+        Ok(QueueAcpPrompt::Failed(message)) => {
-                "session is not a live structured session",
+            error_response(StatusCode::BAD_GATEWAY, "acp_prompt_failed", &message)
-            )
         }
         Err(e) => {
             let info = e.to_error_info();
-            return error_response(session_error_status(&info.code), &info.code, &info.message);
+            error_response(session_error_status(&info.code), &info.code, &info.message)
-        }
-    };
-    match acp.prompt(body.prompt) {
-        Ok(done) => {
-            let manager = s.manager.clone();
-            tokio::spawn(async move {
-                let lifecycle = match done.await {
-                    Ok(Ok(())) => TurnLifecycle::Finish,
-                    Ok(Err(_)) | Err(_) => TurnLifecycle::Failure,
-                };
-                let _ = manager.apply_acp_turn_lifecycle(id, lifecycle, turn).await;
-            });
-            StatusCode::NO_CONTENT.into_response()
-        }
-        Err(message) => {
-            let _ = s
-                .manager
-                .apply_acp_turn_lifecycle(id, TurnLifecycle::Failure, turn)
-                .await;
-            error_response(StatusCode::BAD_GATEWAY, "acp_prompt_failed", &message)
         }
     }
 }
@@ -490,6 +472,32 @@ async fn structured_cancel(State(s): State<LocalApiState>, Path(id): Path<Uuid>)
     }
 }
 
+async fn structured_approval(
+    State(s): State<LocalApiState>,
+    Path((id, nonce)): Path<(Uuid, String)>,
+    Json(decision): Json<StructuredApprovalDecision>,
+) -> Response {
+    let decision = match decision {
+        StructuredApprovalDecision::AllowOnce => crate::session::ApprovalDecision::AllowOnce,
+        StructuredApprovalDecision::AllowForSession => {
+            crate::session::ApprovalDecision::AllowForSession
+        }
+        StructuredApprovalDecision::Deny => crate::session::ApprovalDecision::Deny,
+    };
+    match s.manager.resolve_acp_approval(id, &nonce, decision).await {
+        Ok(Some(())) => StatusCode::NO_CONTENT.into_response(),
+        Ok(None) => error_response(
+            StatusCode::CONFLICT,
+            "not_structured",
+            "session is not a live structured session",
+        ),
+        Err(e) => {
+            let info = e.to_error_info();
+            error_response(session_error_status(&info.code), &info.code, &info.message)
+        }
+    }
+}
+
 /// Convenience constructor — pulls the bits `LocalApiState` needs from
 /// `DaemonConfig` so callers don't have to assemble it by hand. The project
 /// scanner starts with its cache cold; wire

crates/maestro-daemon/src/projection.rs

@@ -155,6 +155,61 @@ impl Projection {
         self.active_turn
     }
 
+    /// Add a daemon-originated permission row. The nonce belongs only in the
+    /// reduced state, which is the client control surface, never in ACP data.
+    pub fn request_approval(&mut self, nonce: String, request: Value) -> ProjectionUpdate {
+        let row = self.append(
+            format!("approval:{}", uuid::Uuid::new_v4()),
+            TranscriptRowKind::Approval,
+            request.clone(),
+        );
+        self.state.pending_approval = Some(serde_json::json!({
+            "nonce": nonce,
+            "request": request,
+        }));
+        ProjectionUpdate {
+            transcript: Some(row),
+            state: self.state(),
+        }
+    }
+
+    pub fn restore_approval(&mut self, row: TranscriptRow) -> ProjectionUpdate {
+        let transcript = if !self.row_index.contains_key(&row.id) {
+            self.row_index.insert(row.id.clone(), self.rows.len());
+            self.rows.push(row.clone());
+            Some(TranscriptDelta::Append { row })
+        } else {
+            None
+        };
+        ProjectionUpdate {
+            transcript,
+            state: self.state(),
+        }
+    }
+
+    /// Clear an approval only when it is still the one the client answered.
+    pub fn resolve_approval(&mut self, nonce: &str) -> Option<ProjectionUpdate> {
+        let pending = self.state.pending_approval.as_ref()?;
+        if pending.get("nonce").and_then(Value::as_str) != Some(nonce) {
+            return None;
+        }
+        self.state.pending_approval = None;
+        Some(ProjectionUpdate {
+            transcript: None,
+            state: self.state(),
+        })
+    }
+
+    /// Terminal turn paths must not leave a recovered snapshot permanently
+    /// blocked on an ACP request whose transport no longer exists.
+    pub fn clear_approval(&mut self) -> Option<ProjectionUpdate> {
+        self.state.pending_approval.take()?;
+        Some(ProjectionUpdate {
+            transcript: None,
+            state: self.state(),
+        })
+    }
+
     fn content(
         &mut self,
         seq: u64,
@@ -632,4 +687,68 @@ mod tests {
     fn failed_turn_does_not_reactivate_unfinished_tools() {
         terminal_turn_drops_unfinished_tools(TurnLifecycle::Failure, "failed-1", "next-1");
     }
+
+    #[test]
+    fn approval_history_survives_pending_cleanup() {
+        let mut projection = Projection::default();
+        let update =
+            projection.request_approval("secret".into(), serde_json::json!({"tool":"bash"}));
+        let TranscriptDelta::Append { row } = update.transcript.unwrap() else {
+            panic!("approval row")
+        };
+        assert_eq!(row.kind, TranscriptRowKind::Approval);
+        assert!(row.data.get("nonce").is_none());
+        assert!(!row.id.contains("secret"));
+        let persisted =
+            serde_json::to_string(&crate::sidecar::PersistedEvent::Approval { row: row.clone() })
+                .unwrap();
+        assert!(!persisted.contains("secret"));
+        assert_eq!(
+            projection.state().pending_approval.unwrap()["nonce"],
+            "secret"
+        );
+        projection.clear_approval();
+        assert!(projection.state().pending_approval.is_none());
+        assert_eq!(projection.snapshot(0).rows, vec![row]);
+    }
+
+    #[test]
+    fn recovered_approval_is_historical_not_pending() {
+        let mut live = Projection::default();
+        let TranscriptDelta::Append { row } = live
+            .request_approval("daemon-nonce".into(), serde_json::json!({"tool":"bash"}))
+            .transcript
+            .unwrap()
+        else {
+            panic!("approval row")
+        };
+        let mut recovered = Projection::default();
+        let update = recovered.restore_approval(row.clone());
+        assert_eq!(
+            update.transcript,
+            Some(TranscriptDelta::Append { row: row.clone() })
+        );
+        assert!(update.state.pending_approval.is_none());
+        assert_eq!(recovered.snapshot(0).rows, vec![row]);
+        let replay = serde_json::to_string(&recovered.snapshot(1).rows).unwrap();
+        assert!(!replay.contains("daemon-nonce"));
+    }
+
+    #[test]
+    fn stale_terminal_turn_cannot_clear_newer_approval() {
+        let mut projection = Projection::default();
+        let (first, _) = projection.begin_turn();
+        projection.turn_lifecycle(TurnLifecycle::Cancel, first);
+        let (second, _) = projection.begin_turn();
+        projection.request_approval("second-nonce".into(), serde_json::json!({"tool":"bash"}));
+
+        assert!(projection
+            .turn_lifecycle(TurnLifecycle::Finish, first)
+            .is_none());
+        assert_eq!(projection.active_turn(), Some(second));
+        assert_eq!(
+            projection.state().pending_approval.unwrap()["nonce"],
+            "second-nonce"
+        );
+    }
 }

crates/maestro-daemon/src/session.rs

@@ -12,18 +12,21 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use chrono::Utc;
-use maestro_acp::{AcpSession, SpawnConfig};
+use maestro_acp::{AcpSession, PermissionRequest, SpawnConfig};
 use maestro_protocol::{
     ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session,
-    SessionStatus, SessionView, StructuredReplayEvent, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
+    SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, WorktreeSpec, NAME_KEY,
+    NAME_SOURCE_KEY,
 };
-use tokio::sync::{broadcast, watch, Mutex};
+use tokio::sync::{broadcast, oneshot, watch, Mutex};
 use uuid::Uuid;
 
 use crate::panes::Panes;
 use crate::projection::{Projection, ProjectionUpdate, TurnLifecycle};
 use crate::pty::{pipe_pane_command, PtyBridge};
-use crate::sidecar::{EventCursor, EventLog, LoggedEvent, SidecarError, SidecarStore};
+use crate::sidecar::{
+    EventCursor, EventLog, LoggedEvent, PersistedEvent, SidecarError, SidecarStore,
+};
 use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord};
 use crate::tmux::TmuxError;
 use crate::viewport::{ViewportId, Viewports};
@@ -118,6 +121,33 @@ struct AcpEventStream {
     live: broadcast::Sender<LoggedEvent>,
     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)]
+struct PendingApproval {
+    response: oneshot::Sender<maestro_acp::schema::RequestPermissionOutcome>,
+    options: Vec<maestro_acp::schema::PermissionOption>,
+}
+
+/// The outcome accepted by the structured approval endpoint.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ApprovalDecision {
+    AllowOnce,
+    AllowForSession,
+    Deny,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -156,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 {
@@ -406,7 +445,7 @@ impl SessionManager {
         let mut projection = Projection::default();
         let mut replay = Vec::new();
         for event in events {
-            let update = projection.apply(event.seq, &event.notification);
+            let update = apply_persisted_event(&mut projection, &event);
             let include = match cursor {
                 EventCursor::Since(since) => event.seq > since,
                 EventCursor::Before(Some(before)) => event.seq < before,
@@ -488,6 +527,10 @@ impl SessionManager {
             return Ok(false);
         };
         let mut stream = stream.lock().await;
+        if stream.projection.active_turn() != Some(turn) {
+            return Ok(false);
+        }
+        cancel_pending_approvals(&mut stream);
         let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) else {
             return Ok(false);
         };
@@ -518,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) = {
@@ -531,19 +718,93 @@ 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);
         };
-        acp.cancel();
         let projected = stream
             .projection
             .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.
+    pub async fn resolve_acp_approval(
+        &self,
+        id: Uuid,
+        nonce: &str,
+        decision: ApprovalDecision,
+    ) -> Result<Option<()>, SessionError> {
+        let 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(None);
+            }
+            record.acp_events.clone()
+        };
+        let Some(stream) = stream else {
+            return Ok(None);
+        };
+        let mut stream = stream.lock().await;
+        let Some(pending) = stream.pending_approvals.remove(nonce) else {
+            return Ok(Some(()));
+        };
+        let outcome = permission_outcome(decision, &pending.options);
+        let _ = pending.response.send(outcome);
+        if let Some(update) = stream.projection.resolve_approval(nonce) {
+            let _ = stream.projected_live.send(update);
+        }
+        Ok(Some(()))
+    }
+
+    async fn cancel_acp_approvals(&self, id: Uuid) {
+        let stream = self
+            .inner
+            .sessions
+            .lock()
+            .await
+            .get(&id)
+            .and_then(|record| record.acp_events.clone());
+        if let Some(stream) = stream {
+            cancel_pending_approvals(&mut *stream.lock().await);
+        }
+    }
+
     /// End every adapter child before the daemon exits (#134): each gets
     /// `session/cancel`, then SIGTERM, concurrently. The sessions stay in
     /// the map, marked Dead and written through to their sidecars, so what
@@ -569,6 +830,9 @@ impl SessionManager {
         if children.is_empty() {
             return;
         }
+        for (id, _, _) in &children {
+            self.cancel_acp_approvals(*id).await;
+        }
         let reaped =
             futures_util::future::join_all(children.iter().map(|(_, acp, _)| acp.reap_if_exited()))
                 .await;
@@ -818,13 +1082,16 @@ impl SessionManager {
                     let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
                     let mut projection = Projection::default();
                     for event in &persisted {
-                        projection.apply(event.seq, &event.notification);
+                        apply_persisted_event(&mut projection, event);
                     }
                     Some(Arc::new(Mutex::new(AcpEventStream {
                         log,
                         live: tx,
                         projection,
                         projected_live: projected_tx,
+                        pending_approvals: HashMap::new(),
+                        operation: Arc::new(Mutex::new(())),
+                        dispatch: AcpDispatchState::default(),
                     })))
                 }
                 Err(e) => {
@@ -933,7 +1200,8 @@ impl SessionManager {
             );
         }
         if let Some((acp, events)) = event_pump {
-            let done = start_acp_event_pump(&self.inner, id, &acp, events);
+            let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
+            start_acp_permission_pump(&acp, events);
             if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
                 record.acp_event_pump = Some(done);
             }
@@ -1260,13 +1528,16 @@ impl SessionManager {
                         let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
                         let mut projection = Projection::default();
                         for event in &persisted {
-                            projection.apply(event.seq, &event.notification);
+                            apply_persisted_event(&mut projection, event);
                         }
                         Some(Arc::new(Mutex::new(AcpEventStream {
                             log,
                             live: tx,
                             projection,
                             projected_live: projected_tx,
+                            pending_approvals: HashMap::new(),
+                            operation: Arc::new(Mutex::new(())),
+                            dispatch: AcpDispatchState::default(),
                         })))
                     }
                     Err(error) => {
@@ -1318,7 +1589,8 @@ impl SessionManager {
                 },
             );
             if let Some((acp, events)) = event_pump {
-                let done = start_acp_event_pump(&self.inner, id, &acp, events);
+                let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
+                start_acp_permission_pump(&acp, events);
                 if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
                     record.acp_event_pump = Some(done);
                 }
@@ -1672,10 +1944,15 @@ fn start_acp_event_pump(
         while let Some(notification) = updates.recv().await {
             let append_error = {
                 let mut events = events.lock().await;
-                match events.log.append(&notification) {
+                let event = PersistedEvent::Notification(Box::new(notification));
+                match events.log.append(&event) {
                     Ok(seq) => {
-                        let projected = events.projection.apply(seq, &notification);
+                        let logged = LoggedEvent {
-                        let _ = events.live.send(LoggedEvent { seq, notification });
+                            seq,
+                            event: event.clone(),
+                        };
+                        let projected = apply_persisted_event(&mut events.projection, &logged);
+                        let _ = events.live.send(logged);
                         let _ = events.projected_live.send(projected);
                         None
                     }
@@ -1699,6 +1976,112 @@ fn start_acp_event_pump(
     done_rx
 }
 
+fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> ProjectionUpdate {
+    match &event.event {
+        PersistedEvent::Notification(notification) => projection.apply(event.seq, notification),
+        PersistedEvent::Approval { row } => projection.restore_approval(row.clone()),
+    }
+}
+
+fn start_acp_permission_pump(acp: &Arc<AcpSession>, events: Arc<Mutex<AcpEventStream>>) {
+    let Some(mut requests) = acp.take_permission_requests() else {
+        return;
+    };
+    tokio::spawn(async move {
+        while let Some(PermissionRequest {
+            turn,
+            request,
+            response,
+        }) = requests.recv().await
+        {
+            let mut events = events.lock().await;
+            if events.projection.active_turn() != Some(turn) {
+                let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
+                continue;
+            }
+            let nonce = Uuid::new_v4().to_string();
+            let data = match serde_json::to_value(&request) {
+                Ok(value) => value,
+                Err(_) => {
+                    let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
+                    continue;
+                }
+            };
+            // ACP requests serialize through its connection. Keep the newest
+            // one authoritative if an adapter nonetheless overlaps them.
+            cancel_pending_approvals(&mut events);
+            events.pending_approvals.insert(
+                nonce.clone(),
+                PendingApproval {
+                    response,
+                    options: request.options,
+                },
+            );
+            let update = events.projection.request_approval(nonce, data);
+            if let Some(TranscriptDelta::Append { row }) = &update.transcript {
+                let event = PersistedEvent::Approval { row: row.clone() };
+                if let Err(error) = events.log.append(&event) {
+                    tracing::error!(%error, "approval history append failed");
+                    cancel_pending_approvals(&mut events);
+                    continue;
+                }
+                let seq = events.log.last_seq();
+                let _ = events.live.send(LoggedEvent { seq, event });
+            }
+            let _ = events.projected_live.send(update);
+        }
+    });
+}
+
+fn cancel_pending_approvals(events: &mut AcpEventStream) {
+    for (_, pending) in events.pending_approvals.drain() {
+        let _ = pending
+            .response
+            .send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
+    }
+    if let Some(update) = events.projection.clear_approval() {
+        let _ = events.projected_live.send(update);
+    }
+}
+
+fn permission_outcome(
+    decision: ApprovalDecision,
+    options: &[maestro_acp::schema::PermissionOption],
+) -> maestro_acp::schema::RequestPermissionOutcome {
+    use maestro_acp::schema::{
+        PermissionOptionKind, RequestPermissionOutcome, SelectedPermissionOutcome,
+    };
+    let kind = match decision {
+        ApprovalDecision::AllowOnce => PermissionOptionKind::AllowOnce,
+        ApprovalDecision::AllowForSession => PermissionOptionKind::AllowAlways,
+        ApprovalDecision::Deny => {
+            // Prefer a one-shot refusal; use a permanent refusal only when
+            // that is the sole rejection option the ACP request offered.
+            return options
+                .iter()
+                .find(|option| option.kind == PermissionOptionKind::RejectOnce)
+                .or_else(|| {
+                    options
+                        .iter()
+                        .find(|option| option.kind == PermissionOptionKind::RejectAlways)
+                })
+                .map_or(RequestPermissionOutcome::Cancelled, |option| {
+                    RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(
+                        option.option_id.clone(),
+                    ))
+                });
+        }
+    };
+    options.iter().find(|option| option.kind == kind).map_or(
+        RequestPermissionOutcome::Cancelled,
+        |option| {
+            RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(
+                option.option_id.clone(),
+            ))
+        },
+    )
+}
+
 async fn wait_for_event_pump(mut done: watch::Receiver<bool>) {
     if !*done.borrow() {
         let _ = done.wait_for(|finished| *finished).await;
@@ -2024,6 +2407,15 @@ async fn mark_session_dead_with_worktree_settlement(
     let Some((mut session, worktree)) = flipped else {
         return false;
     };
+    let stream = inner
+        .sessions
+        .lock()
+        .await
+        .get(&id)
+        .and_then(|record| record.acp_events.clone());
+    if let Some(stream) = stream {
+        cancel_pending_approvals(&mut *stream.lock().await);
+    }
     tracing::info!(%id, "session process gone — marking dead");
 
     if settle_worktree {
@@ -2130,6 +2522,9 @@ async fn cleanup_record(
         let _ = tmux_blocking(move || tmux.kill_session(id)).await;
     }
     if let Some(acp) = &record.acp {
+        if let Some(stream) = &record.acp_events {
+            cancel_pending_approvals(&mut *stream.lock().await);
+        }
         // `session/cancel`, a short wait, then SIGTERM (#134).
         acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
             .await;
@@ -2163,3 +2558,58 @@ async fn cleanup_record(
     }
     let _ = inner.events.send(CatalogEvent::SessionRemoved { id });
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{permission_outcome, ApprovalDecision};
+    use maestro_acp::schema::{PermissionOption, PermissionOptionKind, RequestPermissionOutcome};
+
+    fn option(id: &str, kind: PermissionOptionKind) -> PermissionOption {
+        PermissionOption::new(id.to_owned(), id, kind)
+    }
+
+    fn selected_id(outcome: RequestPermissionOutcome) -> Option<String> {
+        match outcome {
+            RequestPermissionOutcome::Selected(selected) => Some(selected.option_id.to_string()),
+            RequestPermissionOutcome::Cancelled => None,
+            _ => None,
+        }
+    }
+
+    #[test]
+    fn permission_decisions_select_only_offered_options() {
+        let all = [
+            option("allow-once", PermissionOptionKind::AllowOnce),
+            option("allow-always", PermissionOptionKind::AllowAlways),
+            option("reject-once", PermissionOptionKind::RejectOnce),
+            option("reject-always", PermissionOptionKind::RejectAlways),
+        ];
+        assert_eq!(
+            selected_id(permission_outcome(ApprovalDecision::AllowOnce, &all)).as_deref(),
+            Some("allow-once")
+        );
+        assert_eq!(
+            selected_id(permission_outcome(ApprovalDecision::AllowForSession, &all)).as_deref(),
+            Some("allow-always")
+        );
+        assert_eq!(
+            selected_id(permission_outcome(ApprovalDecision::Deny, &all)).as_deref(),
+            Some("reject-once")
+        );
+        assert_eq!(
+            selected_id(permission_outcome(
+                ApprovalDecision::Deny,
+                &[option("reject-always", PermissionOptionKind::RejectAlways)],
+            ))
+            .as_deref(),
+            Some("reject-always")
+        );
+        assert_eq!(
+            selected_id(permission_outcome(
+                ApprovalDecision::Deny,
+                &[option("allow-once", PermissionOptionKind::AllowOnce)],
+            )),
+            None
+        );
+    }
+}

crates/maestro-daemon/src/sidecar.rs

@@ -7,8 +7,8 @@
 //! advertised but with empty metadata (handled by the caller, not here).
 //!
 //! A structured session (#135) also owns `<id>.events.jsonl` next to its
-//! sidecar: every `session/update` notification its adapter sent, one JSON
+//! sidecar: every `session/update` notification and daemon approval row is one
-//! object per line, appended before it is broadcast. The log is the source of
+//! JSON object, appended before it is broadcast. The log is the source of
 //! truth for the transcript; `seq` is the line's 1-based position, so a
 //! client that holds seq `n` asks for [`EventCursor::Since`]`(n)`. A line
 //! that does not parse — including the partial line a hard kill leaves at
@@ -21,7 +21,7 @@ use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
 
 use maestro_acp::schema::SessionNotification;
-use maestro_protocol::Session;
+use maestro_protocol::{Session, TranscriptRow};
 use uuid::Uuid;
 
 #[derive(Debug, thiserror::Error)]
@@ -150,10 +150,10 @@ impl SidecarStore {
                 }
                 continue;
             }
-            let Ok(notification) = serde_json::from_slice::<SessionNotification>(&line) else {
+            let Ok(event) = PersistedEvent::decode(&line) else {
                 continue;
             };
-            out.push_back(LoggedEvent { seq, notification });
+            out.push_back(LoggedEvent { seq, event });
             match cursor {
                 EventCursor::Since(_) if out.len() >= limit => break,
                 EventCursor::Before(_) if out.len() > limit => {
@@ -190,6 +190,7 @@ impl SidecarStore {
     /// whichever does not exist.
     pub fn delete(&self, id: Uuid) -> Result<(), SidecarError> {
         remove_if_present(&self.events_path_for(id))?;
+        remove_if_present(&self.dir.join(format!("{id}.approvals.jsonl")))?;
         remove_if_present(&self.recovery_path_for(id))?;
         remove_if_present(&self.path_for(id))
     }
@@ -247,11 +248,52 @@ pub enum EventCursor {
 }
 
 /// One line of a session's event log.
-#[derive(Debug, Clone, PartialEq, serde::Serialize)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct LoggedEvent {
     /// 1-based line position in the log; the cursor for [`EventCursor`].
     pub seq: u64,
-    pub notification: SessionNotification,
+    pub event: PersistedEvent,
+}
+
+/// A single entry in the structured event history.
+///
+/// ACP notifications retain their original JSON representation, so existing
+/// event logs remain readable. Daemon-generated rows use an envelope so they
+/// can share the same cursor sequence without pretending to be ACP traffic.
+#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
+#[serde(tag = "maestroEvent", rename_all = "snake_case")]
+pub enum PersistedEvent {
+    Approval {
+        row: TranscriptRow,
+    },
+    #[serde(skip)]
+    Notification(Box<SessionNotification>),
+}
+
+impl PersistedEvent {
+    fn decode(line: &[u8]) -> Result<Self, serde_json::Error> {
+        serde_json::from_slice::<SessionNotification>(line)
+            .map(|notification| Self::Notification(Box::new(notification)))
+            .or_else(|_| serde_json::from_slice(line))
+    }
+}
+
+impl From<SessionNotification> for PersistedEvent {
+    fn from(notification: SessionNotification) -> Self {
+        Self::Notification(Box::new(notification))
+    }
+}
+
+impl From<&SessionNotification> for PersistedEvent {
+    fn from(notification: &SessionNotification) -> Self {
+        Self::Notification(Box::new(notification.clone()))
+    }
+}
+
+impl From<&PersistedEvent> for PersistedEvent {
+    fn from(event: &PersistedEvent) -> Self {
+        event.clone()
+    }
 }
 
 /// An open, append-only handle on one session's event log. Obtained from
@@ -272,8 +314,12 @@ impl EventLog {
     /// handed to the kernel in a single write before this returns, so a
     /// caller that broadcasts afterwards never announces an event the log
     /// would not replay — short of the machine losing power.
-    pub fn append(&mut self, notification: &SessionNotification) -> Result<u64, SidecarError> {
+    pub fn append(&mut self, event: impl Into<PersistedEvent>) -> Result<u64, SidecarError> {
-        let mut line = serde_json::to_vec(notification)?;
+        let event = event.into();
+        let mut line = match &event {
+            PersistedEvent::Notification(notification) => serde_json::to_vec(notification)?,
+            _ => serde_json::to_vec(&event)?,
+        };
         line.push(b'\n');
         self.file.write_all(&line)?;
         self.file.flush()?;

crates/maestro-daemon/tests/acp.rs

@@ -83,7 +83,10 @@ impl Fake {
 }
 
 fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
-    match &event.notification.update {
+    let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
+        panic!("unexpected daemon event")
+    };
+    match &notification.update {
         SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
             ContentBlock::Text(text) => &text.text,
             other => panic!("unexpected content {other:?}"),
@@ -321,6 +324,107 @@ async fn stale_turn_completion_cannot_end_the_next_turn() {
     fx.manager.kill(session.id, false).await.unwrap();
 }
 
+#[tokio::test]
+async fn cancelled_prompt_that_never_settles_fails_the_session() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_PERMISSION_AFTER_CANCEL".into(), "1".into());
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .unwrap();
+
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "first".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+    assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
+    let second = fx
+        .manager
+        .queue_acp_prompt(session.id, "second".into())
+        .await
+        .unwrap();
+    assert!(
+        matches!(second, maestro_daemon::session::QueueAcpPrompt::Unavailable),
+        "{second:?}"
+    );
+    assert_eq!(
+        fx.manager.status(session.id).await,
+        Some(SessionStatus::Dead)
+    );
+    assert!(!fake.log().contains("permission response"));
+}
+
+#[tokio::test]
+async fn cancellation_settlement_allows_the_next_prompt_permission() {
+    let fx = DaemonFixture::new();
+    let fake = Fake::new();
+    let cwd = TempDir::new().unwrap();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_REQUEST_PERMISSION".into(), "1".into());
+    env.insert("FAKE_ACP_CANCEL_SETTLES".into(), "1".into());
+    let session = fx
+        .manager
+        .create(structured(cwd.path(), env))
+        .await
+        .unwrap();
+
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "first".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+    assert!(fx.manager.cancel_acp_turn(session.id).await.unwrap());
+    assert!(matches!(
+        fx.manager
+            .queue_acp_prompt(session.id, "second".into())
+            .await
+            .unwrap(),
+        maestro_daemon::session::QueueAcpPrompt::Queued
+    ));
+
+    let nonce = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let (_, state, _) = fx
+                .manager
+                .subscribe_acp_projection(session.id, EventCursor::Since(0))
+                .await
+                .unwrap()
+                .unwrap();
+            if let Some(pending) = state.pending_approval {
+                break pending["nonce"].as_str().unwrap().to_owned();
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    assert!(fx
+        .manager
+        .resolve_acp_approval(
+            session.id,
+            &nonce,
+            maestro_daemon::session::ApprovalDecision::AllowOnce,
+        )
+        .await
+        .unwrap()
+        .is_some());
+    wait_for_async(Duration::from_secs(2), || {
+        let log = fake.log();
+        async move { log.contains("permission response") }
+    })
+    .await;
+    fx.manager.kill(session.id, false).await.unwrap();
+}
+
 #[tokio::test]
 async fn recovery_loads_the_acp_session_without_duplicating_history() {
     let fx = DaemonFixture::new();

crates/maestro-daemon/tests/local_api.rs

@@ -142,6 +142,10 @@ impl FakeAcp {
             .next_back()
             .expect("fake ACP started")
     }
+
+    fn log(&self) -> String {
+        std::fs::read_to_string(self.dir.path().join("agent.log")).unwrap_or_default()
+    }
 }
 
 async fn structured_session(h: &Harness, fake: &FakeAcp) -> Uuid {
@@ -256,6 +260,135 @@ async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it(
         .await
         .unwrap();
     assert_eq!(next.status(), reqwest::StatusCode::NO_CONTENT);
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            if fake.log().matches("recv session/prompt").count() == 2 {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("both prompt commands reach ACP");
+    let log = fake.log();
+    let first_prompt = log.find("recv session/prompt").unwrap();
+    let cancel = log.find("recv session/cancel").unwrap();
+    let second_prompt = log.rfind("recv session/prompt").unwrap();
+    assert!(first_prompt < cancel && cancel < second_prompt, "{log}");
+}
+
+#[tokio::test]
+async fn structured_approval_is_secret_single_use_and_race_safe() {
+    let h = spawn("naps-laptop").await;
+    let fake = FakeAcp::new();
+    let mut env = fake.env();
+    env.insert("FAKE_ACP_REQUEST_PERMISSION".into(), "1".into());
+    let id = structured_session_with_env(&h, env).await;
+    let client = reqwest::Client::new();
+    let (mut ws, _) = connect_async(ws_url(h.addr, &format!("/api/structured/{id}")))
+        .await
+        .unwrap();
+    ws.next().await.unwrap().unwrap();
+    ws.next().await.unwrap().unwrap();
+    assert_eq!(
+        client
+            .post(url(
+                h.addr,
+                &format!("/api/sessions/{id}/structured/prompt")
+            ))
+            .json(&serde_json::json!({"prompt":"needs permission"}))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        reqwest::StatusCode::NO_CONTENT
+    );
+
+    let nonce = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let event: serde_json::Value =
+                serde_json::from_str(&ws.next().await.unwrap().unwrap().into_text().unwrap())
+                    .unwrap();
+            if let Some(nonce) = event["state"]["pending_approval"]["nonce"].as_str() {
+                assert_eq!(event["transcript"]["row"]["kind"], "approval");
+                assert!(event["transcript"]["row"]["data"].get("nonce").is_none());
+                break nonce.to_owned();
+            }
+        }
+    })
+    .await
+    .unwrap();
+    assert!(
+        !fake.log().contains(&nonce),
+        "daemon nonce leaked to ACP agent"
+    );
+
+    let path = format!("/api/sessions/{id}/structured/approvals/{nonce}");
+    let (a, b) = tokio::join!(
+        client
+            .post(url(h.addr, &path))
+            .json(&serde_json::json!("allow_once"))
+            .send(),
+        client
+            .post(url(h.addr, &path))
+            .json(&serde_json::json!("deny"))
+            .send(),
+    );
+    assert_eq!(a.unwrap().status(), reqwest::StatusCode::NO_CONTENT);
+    assert_eq!(b.unwrap().status(), reqwest::StatusCode::NO_CONTENT);
+    assert_eq!(
+        client
+            .post(url(h.addr, &path))
+            .json(&serde_json::json!("deny"))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        reqwest::StatusCode::NO_CONTENT
+    );
+    tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            if fake.log().contains("permission response") {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    let log = fake.log();
+    assert!(
+        log.contains("\"optionId\": \"once\"") || log.contains("\"optionId\": \"deny\""),
+        "{log}"
+    );
+    tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            let event: serde_json::Value =
+                serde_json::from_str(&ws.next().await.unwrap().unwrap().into_text().unwrap())
+                    .unwrap();
+            if event["state"]["pending_approval"].is_null() {
+                break;
+            }
+        }
+    })
+    .await
+    .expect("resolved approval clears the reduced state");
+    assert_eq!(
+        client
+            .post(url(
+                h.addr,
+                &format!(
+                    "/api/sessions/{}/structured/approvals/{nonce}",
+                    Uuid::new_v4()
+                )
+            ))
+            .json(&serde_json::json!("deny"))
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        reqwest::StatusCode::NOT_FOUND
+    );
 }
 
 #[tokio::test]

crates/maestro-daemon/tests/sidecar.rs

@@ -6,8 +6,8 @@ use std::collections::HashMap;
 use std::path::PathBuf;
 
 use chrono::{DateTime, Utc};
-use maestro_daemon::sidecar::SidecarStore;
+use maestro_daemon::sidecar::{PersistedEvent, SidecarStore};
-use maestro_protocol::{Session, SessionStatus, SessionView};
+use maestro_protocol::{Session, SessionStatus, SessionView, TranscriptRow, TranscriptRowKind};
 use serde_json::Value;
 use tempfile::TempDir;
 use uuid::Uuid;
@@ -186,6 +186,13 @@ fn text_of(n: &SessionNotification) -> &str {
     }
 }
 
+fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
+    let maestro_daemon::sidecar::PersistedEvent::Notification(notification) = &event.event else {
+        panic!("expected ACP notification")
+    };
+    text_of(notification)
+}
+
 #[test]
 fn append_assigns_line_position_as_seq_and_reads_back() {
     let dir = TempDir::new().unwrap();
@@ -194,9 +201,9 @@ fn append_assigns_line_position_as_seq_and_reads_back() {
 
     let mut log = store.open_events(id).expect("open");
     assert_eq!(log.last_seq(), 0);
-    assert_eq!(log.append(&note("a")).unwrap(), 1);
+    assert_eq!(log.append(note("a")).unwrap(), 1);
-    assert_eq!(log.append(&note("b")).unwrap(), 2);
+    assert_eq!(log.append(note("b")).unwrap(), 2);
-    assert_eq!(log.append(&note("c")).unwrap(), 3);
+    assert_eq!(log.append(note("c")).unwrap(), 3);
     assert_eq!(log.last_seq(), 3);
 
     let path = store.events_path_for(id);
@@ -211,9 +218,7 @@ fn append_assigns_line_position_as_seq_and_reads_back() {
     let all = store.read_events(id, EventCursor::Since(0), 100).unwrap();
     assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 2, 3]);
     assert_eq!(
-        all.iter()
+        all.iter().map(event_text).collect::<Vec<_>>(),
-            .map(|e| text_of(&e.notification))
-            .collect::<Vec<_>>(),
         vec!["a", "b", "c"]
     );
 }
@@ -225,7 +230,7 @@ fn since_reads_forward_and_before_reads_backward_with_limit() {
     let id = Uuid::new_v4();
     let mut log = store.open_events(id).unwrap();
     for i in 1..=6 {
-        log.append(&note(&format!("n{i}"))).unwrap();
+        log.append(note(&format!("n{i}"))).unwrap();
     }
 
     // `since` is exclusive: what a client that already holds seq 2 needs.
@@ -278,7 +283,7 @@ fn malformed_lines_are_skipped_but_keep_their_seq() {
     let store = SidecarStore::new(dir.path());
     let id = Uuid::new_v4();
     let mut log = store.open_events(id).unwrap();
-    log.append(&note("first")).unwrap();
+    log.append(note("first")).unwrap();
     drop(log);
 
     // Garbage in the middle and a line that is JSON but not a notification.
@@ -289,11 +294,11 @@ fn malformed_lines_are_skipped_but_keep_their_seq() {
     std::fs::write(&path, raw).unwrap();
     let mut log = store.open_events(id).unwrap();
     assert_eq!(log.last_seq(), 3);
-    assert_eq!(log.append(&note("fourth")).unwrap(), 4);
+    assert_eq!(log.append(note("fourth")).unwrap(), 4);
 
     let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
     assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 4]);
-    assert_eq!(text_of(&all[1].notification), "fourth");
+    assert_eq!(event_text(&all[1]), "fourth");
 }
 
 #[test]
@@ -303,7 +308,7 @@ fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() {
     let store = SidecarStore::new(dir.path());
     let id = Uuid::new_v4();
     let mut log = store.open_events(id).unwrap();
-    log.append(&note("whole")).unwrap();
+    log.append(note("whole")).unwrap();
     drop(log);
 
     let path = store.events_path_for(id);
@@ -319,10 +324,10 @@ fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() {
     // glue onto it; the partial keeps a seq of its own.
     let mut log = store.open_events(id).unwrap();
     assert_eq!(log.last_seq(), 2);
-    assert_eq!(log.append(&note("after")).unwrap(), 3);
+    assert_eq!(log.append(note("after")).unwrap(), 3);
     let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
     assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 3]);
-    assert_eq!(text_of(&all[1].notification), "after");
+    assert_eq!(event_text(&all[1]), "after");
     let raw = std::fs::read_to_string(&path).unwrap();
     assert_eq!(raw.lines().count(), 3, "{raw}");
 }
@@ -334,8 +339,10 @@ fn delete_removes_the_event_log_and_load_all_ignores_it() {
     let id = Uuid::new_v4();
     store.write(&sample(id)).unwrap();
     let mut log = store.open_events(id).unwrap();
-    log.append(&note("x")).unwrap();
+    log.append(note("x")).unwrap();
     drop(log);
+    let legacy_approvals = dir.path().join(format!("{id}.approvals.jsonl"));
+    std::fs::write(&legacy_approvals, b"obsolete").unwrap();
     assert!(store.events_path_for(id).exists());
 
     let loaded = store.load_all().unwrap();
@@ -344,6 +351,47 @@ fn delete_removes_the_event_log_and_load_all_ignores_it() {
     store.delete(id).unwrap();
     assert!(!store.path_for(id).exists());
     assert!(!store.events_path_for(id).exists());
+    assert!(!legacy_approvals.exists());
     // Idempotent with the log already gone.
     store.delete(id).unwrap();
 }
+
+#[test]
+fn approval_rows_share_event_cursors_with_legacy_acp_lines() {
+    let dir = TempDir::new().unwrap();
+    let store = SidecarStore::new(dir.path());
+    let id = Uuid::new_v4();
+    let legacy = serde_json::to_vec(&note("before")).unwrap();
+    std::fs::create_dir_all(dir.path()).unwrap();
+    std::fs::write(store.events_path_for(id), [legacy, b"\n".to_vec()].concat()).unwrap();
+
+    let mut log = store.open_events(id).unwrap();
+    assert_eq!(
+        log.append(PersistedEvent::Approval {
+            row: TranscriptRow {
+                id: "approval:secret".into(),
+                kind: TranscriptRowKind::Approval,
+                data: serde_json::json!({"tool": "bash"}),
+            },
+        })
+        .unwrap(),
+        2
+    );
+    assert_eq!(log.append(note("after")).unwrap(), 3);
+
+    let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
+    assert_eq!(
+        all.iter().map(|event| event.seq).collect::<Vec<_>>(),
+        [1, 2, 3]
+    );
+    assert!(matches!(all[1].event, PersistedEvent::Approval { .. }));
+    assert_eq!(event_text(&all[0]), "before");
+    assert_eq!(event_text(&all[2]), "after");
+
+    let since = store.read_events(id, EventCursor::Since(1), 1).unwrap();
+    assert_eq!(since[0].seq, 2);
+    let before = store
+        .read_events(id, EventCursor::Before(Some(3)), 1)
+        .unwrap();
+    assert_eq!(before[0].seq, 2);
+}

crates/maestro-protocol/src/lib.rs

@@ -872,6 +872,15 @@ pub struct StructuredPromptRequest {
     pub prompt: String,
 }
 
+/// Operator decision for a pending ACP permission request.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum StructuredApprovalDecision {
+    AllowOnce,
+    AllowForSession,
+    Deny,
+}
+
 // ---------------------------------------------------------------------------
 // Client ↔ Daemon control messages (text frames on the attach WS)
 // ---------------------------------------------------------------------------

crates/maestro-protocol/src/routes.rs

@@ -91,6 +91,11 @@ pub fn structured_cancel(id: Uuid) -> String {
     format!("{SESSIONS}/{id}/structured/cancel")
 }
 
+/// `POST /api/sessions/:id/structured/approvals/:nonce` — answer ACP permission.
+pub fn structured_approval(id: Uuid, nonce: &str) -> String {
+    format!("{SESSIONS}/{id}/structured/approvals/{nonce}")
+}
+
 /// Axum-style patterns for the two parameterised routes, so the server
 /// registers the same paths [`session`] and [`attach`] build.
 pub mod patterns {
@@ -104,6 +109,7 @@ pub mod patterns {
     pub const STRUCTURED_REPLAY: &str = "/api/sessions/:id/structured/replay";
     pub const STRUCTURED_PROMPT: &str = "/api/sessions/:id/structured/prompt";
     pub const STRUCTURED_CANCEL: &str = "/api/sessions/:id/structured/cancel";
+    pub const STRUCTURED_APPROVAL: &str = "/api/sessions/:id/structured/approvals/:nonce";
 }
 
 /// A daemon base URL that could not be turned into a WebSocket base.