feat(daemon): resolve ACP approvals

2026-09-05 02:32:01 +01:00
parent 13009d0f0e
commit d9815d86c6
9 changed files with 490 additions and 18 deletions
@@ -12,12 +12,13 @@ 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;
@@ -114,10 +115,27 @@ impl SessionRecord {

 #[derive(Debug)]
 struct AcpEventStream {
+    id: Uuid,
+    sidecar: SidecarStore,
    log: EventLog,
    live: broadcast::Sender<LoggedEvent>,
    projection: Projection,
    projected_live: broadcast::Sender<ProjectionUpdate>,
+    pending_approvals: HashMap<String, PendingApproval>,
+}
+
+#[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)]
@@ -488,6 +506,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);
        };
@@ -532,10 +554,11 @@ impl SessionManager {
            return Ok(false);
        };
        let mut stream = stream.lock().await;
+        acp.cancel();
+        cancel_pending_approvals(&mut stream);
        let Some(turn) = stream.projection.active_turn() else {
            return Ok(true);
        };
-        acp.cancel();
        let projected = stream
            .projection
            .turn_lifecycle(TurnLifecycle::Cancel, turn)
@@ -544,6 +567,51 @@ impl SessionManager {
        Ok(true)
    }

+    /// 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 +637,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;
@@ -820,11 +891,17 @@ impl SessionManager {
                    for event in &persisted {
                        projection.apply(event.seq, &event.notification);
                    }
+                    for row in self.inner.sidecar.read_approvals(id)? {
+                        projection.restore_approval(row);
+                    }
                    Some(Arc::new(Mutex::new(AcpEventStream {
+                        id,
+                        sidecar: self.inner.sidecar.clone(),
                        log,
                        live: tx,
                        projection,
                        projected_live: projected_tx,
+                        pending_approvals: HashMap::new(),
                    })))
                }
                Err(e) => {
@@ -933,7 +1010,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);
            }
@@ -1262,11 +1340,17 @@ impl SessionManager {
                        for event in &persisted {
                            projection.apply(event.seq, &event.notification);
                        }
+                        for row in self.inner.sidecar.read_approvals(id)? {
+                            projection.restore_approval(row);
+                        }
                        Some(Arc::new(Mutex::new(AcpEventStream {
+                            id,
+                            sidecar: self.inner.sidecar.clone(),
                            log,
                            live: tx,
                            projection,
                            projected_live: projected_tx,
+                            pending_approvals: HashMap::new(),
                        })))
                    }
                    Err(error) => {
@@ -1318,7 +1402,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);
                }
@@ -1699,6 +1784,75 @@ fn start_acp_event_pump(
    done_rx
 }

+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 { request, response }) = requests.recv().await {
+            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;
+                }
+            };
+            let mut events = events.lock().await;
+            // 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 {
+                if let Err(error) = events.sidecar.append_approval(events.id, row) {
+                    tracing::error!(%error, "approval history append failed");
+                }
+            }
+            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 => PermissionOptionKind::RejectOnce,
+    };
+    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 +2178,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 +2293,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;