diff --git a/crates/maestro-acp/src/lib.rs b/crates/maestro-acp/src/lib.rs index 72c3783..9fa88e7 100644 --- a/crates/maestro-acp/src/lib.rs +++ b/crates/maestro-acp/src/lib.rs @@ -113,6 +113,13 @@ impl SpawnConfig { } } +/// A blocked ACP permission request delivered to the daemon. +#[derive(Debug)] +pub struct PermissionRequest { + pub request: RequestPermissionRequest, + pub response: oneshot::Sender, +} + #[derive(Debug, thiserror::Error)] pub enum AcpError { /// The adapter is not on `PATH`. The message names the install command @@ -192,6 +199,7 @@ pub struct AcpSession { commands: mpsc::UnboundedSender, child_commands: mpsc::UnboundedSender, updates: Mutex>>, + permission_requests: Mutex>>, exit: watch::Receiver>, } @@ -235,6 +243,13 @@ impl AcpSession { self.updates.lock().expect("updates mutex").take() } + pub fn take_permission_requests(&self) -> Option> { + 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) { @@ -335,6 +350,7 @@ pub async fn spawn(config: SpawnConfig) -> Result { 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 +359,7 @@ pub async fn spawn(config: SpawnConfig) -> Result { config.cwd.clone(), config.load_session.clone(), updates_tx, + permission_tx, commands_rx, ready_tx, )); @@ -356,6 +373,7 @@ pub async fn spawn(config: SpawnConfig) -> Result { 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 +485,7 @@ async fn run_connection( cwd: PathBuf, load_session: Option, updates_tx: mpsc::UnboundedSender, + permission_requests: mpsc::UnboundedSender, mut commands_rx: mpsc::UnboundedReceiver, ready_tx: oneshot::Sender>, ) where @@ -495,15 +514,17 @@ async fn run_connection( ) .on_receive_request( async move |request: RequestPermissionRequest, responder, _cx| { - // Stub until #139: decline rather than grant something the - // operator never saw. - tracing::warn!( - session = %request.session_id, - "session/request_permission declined: not implemented yet (#139)" - ); - responder.respond(RequestPermissionResponse::new( - RequestPermissionOutcome::Cancelled, - )) + let (response, received) = oneshot::channel(); + if permission_requests + .send(PermissionRequest { 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!(), ) diff --git a/crates/maestro-acp/tests/fake_acp_agent.py b/crates/maestro-acp/tests/fake_acp_agent.py index f05aacf..b65eab6 100755 --- a/crates/maestro-acp/tests/fake_acp_agent.py +++ b/crates/maestro-acp/tests/fake_acp_agent.py @@ -70,6 +70,8 @@ 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 def chunk(text): @@ -102,6 +104,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") @@ -143,6 +151,20 @@ for raw in sys.stdin: if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1": sys.exit(0) elif method == "session/prompt": + 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): diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index da83481..75fab8a 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/crates/maestro-daemon/src/local_api.rs @@ -21,7 +21,7 @@ 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; @@ -70,7 +70,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)); @@ -490,6 +494,32 @@ async fn structured_cancel(State(s): State, Path(id): Path) } } +async fn structured_approval( + State(s): State, + Path((id, nonce)): Path<(Uuid, String)>, + Json(decision): Json, +) -> 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 diff --git a/crates/maestro-daemon/src/projection.rs b/crates/maestro-daemon/src/projection.rs index 0494110..6b984e1 100644 --- a/crates/maestro-daemon/src/projection.rs +++ b/crates/maestro-daemon/src/projection.rs @@ -155,6 +155,54 @@ 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:{nonce}"), + 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) { + if !self.row_index.contains_key(&row.id) { + self.row_index.insert(row.id.clone(), self.rows.len()); + self.rows.push(row); + } + } + + /// Clear an approval only when it is still the one the client answered. + pub fn resolve_approval(&mut self, nonce: &str) -> Option { + 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 { + self.state.pending_approval.take()?; + Some(ProjectionUpdate { + transcript: None, + state: self.state(), + }) + } + fn content( &mut self, seq: u64, @@ -632,4 +680,23 @@ 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_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]); + } } diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 4ae618d..9dfdb0a 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -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, projection: Projection, projected_live: broadcast::Sender, + pending_approvals: HashMap, +} + +#[derive(Debug)] +struct PendingApproval { + response: oneshot::Sender, + options: Vec, +} + +/// 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, 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, events: Arc>) { + 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) { 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; diff --git a/crates/maestro-daemon/src/sidecar.rs b/crates/maestro-daemon/src/sidecar.rs index ab09816..05aef06 100644 --- a/crates/maestro-daemon/src/sidecar.rs +++ b/crates/maestro-daemon/src/sidecar.rs @@ -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)] @@ -62,6 +62,38 @@ impl SidecarStore { self.dir.join(format!("{id}.recover")) } + fn approvals_path_for(&self, id: Uuid) -> PathBuf { + self.dir.join(format!("{id}.approvals.jsonl")) + } + + /// Approval rows are a daemon projection, not ACP notifications. Keep + /// only their display payload so a restart retains history but never + /// resurrects the daemon-local pending nonce. + pub fn append_approval(&self, id: Uuid, row: &TranscriptRow) -> Result<(), SidecarError> { + std::fs::create_dir_all(&self.dir)?; + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(self.approvals_path_for(id))?; + serde_json::to_writer(&mut file, row)?; + file.write_all(b"\n")?; + file.flush()?; + Ok(()) + } + + pub fn read_approvals(&self, id: Uuid) -> Result, SidecarError> { + let file = match File::open(self.approvals_path_for(id)) { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => return Err(error.into()), + }; + Ok(BufReader::new(file) + .lines() + .map_while(Result::ok) + .filter_map(|line| serde_json::from_str(&line).ok()) + .collect()) + } + /// Mark `id` as eligible for one structured-session recovery attempt. pub fn mark_for_recovery(&self, id: Uuid) -> Result<(), SidecarError> { std::fs::create_dir_all(&self.dir)?; @@ -190,6 +222,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.approvals_path_for(id))?; remove_if_present(&self.recovery_path_for(id))?; remove_if_present(&self.path_for(id)) } diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs index f7262b6..67229a7 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/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 { @@ -258,6 +262,120 @@ async fn structured_prompt_rejects_a_second_in_flight_turn_and_cancel_clears_it( assert_eq!(next.status(), reqwest::StatusCode::NO_CONTENT); } +#[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] async fn structured_prompt_failure_clears_the_turn() { let h = spawn("naps-laptop").await; diff --git a/crates/maestro-protocol/src/lib.rs b/crates/maestro-protocol/src/lib.rs index d7b62d2..3151597 100644 --- a/crates/maestro-protocol/src/lib.rs +++ b/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) // --------------------------------------------------------------------------- diff --git a/crates/maestro-protocol/src/routes.rs b/crates/maestro-protocol/src/routes.rs index b402eb4..6717d0c 100644 --- a/crates/maestro-protocol/src/routes.rs +++ b/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.