diff --git a/crates/maestro-acp/src/lib.rs b/crates/maestro-acp/src/lib.rs index 9fa88e7..d2772ff 100644 --- a/crates/maestro-acp/src/lib.rs +++ b/crates/maestro-acp/src/lib.rs @@ -33,7 +33,7 @@ #![deny(missing_debug_implementations)] -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::ffi::OsString; use std::path::{Path, PathBuf}; use std::process::{ExitStatus, Stdio}; @@ -116,6 +116,8 @@ 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, } @@ -163,6 +165,7 @@ enum Command { Cancel, Prompt { text: String, + turn: u64, done: oneshot::Sender>, }, Close, @@ -173,6 +176,38 @@ enum ChildCommand { Reap(oneshot::Sender), } +#[derive(Debug, Default)] +struct PermissionTurnState { + active: Option, + cancelled_unresolved: HashSet, +} + +impl PermissionTurnState { + fn begin(&mut self, turn: u64) { + self.active = Some(turn); + } + + fn cancel_active(&mut self) { + if let Some(turn) = self.active.take() { + self.cancelled_unresolved.insert(turn); + } + } + + fn finish(&mut self, turn: u64) { + if self.active == Some(turn) { + self.active = None; + } + self.cancelled_unresolved.remove(&turn); + } + + fn callback_turn(&self) -> Option { + self.cancelled_unresolved + .is_empty() + .then_some(self.active) + .flatten() + } +} + /// Why the handshake did not produce a session id. enum ReadyError { /// The adapter answered with an error. @@ -257,11 +292,16 @@ impl AcpSession { } /// Send a user prompt and receive completion once the agent ends the turn. - pub fn prompt(&self, text: String) -> Result>, String> { + pub fn prompt( + &self, + text: String, + turn: u64, + ) -> Result>, 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())?; @@ -497,6 +537,11 @@ async fn run_connection( // 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. A cancelled prompt remains ambiguous + // until its response arrives, so callbacks are rejected rather than + // risking attribution to a later prompt. + let permission_turn = Arc::new(Mutex::new(PermissionTurnState::default())); + let permission_turn_seen = permission_turn.clone(); let result = Client .builder() .name("maestro") @@ -514,9 +559,22 @@ async fn run_connection( ) .on_receive_request( async move |request: RequestPermissionRequest, responder, _cx| { + let turn = permission_turn_seen + .lock() + .expect("permission turn state") + .callback_turn(); + let Some(turn) = turn else { + return responder.respond(RequestPermissionResponse::new( + RequestPermissionOutcome::Cancelled, + )); + }; let (response, received) = oneshot::channel(); if permission_requests - .send(PermissionRequest { request, response }) + .send(PermissionRequest { + turn, + request, + response, + }) .is_err() { return responder.respond(RequestPermissionResponse::new( @@ -596,26 +654,44 @@ async fn run_connection( 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())); } diff --git a/crates/maestro-acp/tests/fake_acp_agent.py b/crates/maestro-acp/tests/fake_acp_agent.py index b65eab6..a377772 100755 --- a/crates/maestro-acp/tests/fake_acp_agent.py +++ b/crates/maestro-acp/tests/fake_acp_agent.py @@ -33,6 +33,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 ", counting from 1) and logs `emit tick `, so a test can make the agent @@ -72,6 +74,7 @@ ticks = 0 prompt_failures = 0 pending_prompt_id = None pending_permission_id = None +delayed_permission = False def chunk(text): @@ -151,6 +154,24 @@ 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_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" diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index 75fab8a..c11d37f 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/crates/maestro-daemon/src/local_api.rs @@ -457,7 +457,7 @@ async fn structured_prompt( return error_response(session_error_status(&info.code), &info.code, &info.message); } }; - match acp.prompt(body.prompt) { + match acp.prompt(body.prompt, turn) { Ok(done) => { let manager = s.manager.clone(); tokio::spawn(async move { diff --git a/crates/maestro-daemon/src/projection.rs b/crates/maestro-daemon/src/projection.rs index 30a0460..5b3fadd 100644 --- a/crates/maestro-daemon/src/projection.rs +++ b/crates/maestro-daemon/src/projection.rs @@ -159,7 +159,7 @@ impl Projection { /// 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}"), + format!("approval:{}", uuid::Uuid::new_v4()), TranscriptRowKind::Approval, request.clone(), ); @@ -698,6 +698,11 @@ mod tests { }; 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" @@ -725,5 +730,25 @@ mod tests { ); 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" + ); } } diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 900707f..0ef56aa 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -1791,7 +1791,17 @@ fn start_acp_permission_pump(acp: &Arc, events: Arc value, @@ -1800,7 +1810,6 @@ fn start_acp_permission_pump(acp: &Arc, events: Arc