diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index f77689498f..be4ec3f359 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -2182,7 +2182,7 @@ impl Session { ); // Add the tx_approve callback to the map before sending the request. let (tx_approve, rx_approve) = oneshot::channel(); - let prev_entry = { + let insert_result = { let mut active = self.active_turn.lock().await; match active.as_mut() { Some(at) => { @@ -2194,11 +2194,13 @@ impl Session { accepted_decisions, ) } - None => None, + None => return ReviewDecision::Abort, } }; - if prev_entry.is_some() { - warn!("Overwriting existing pending approval for call_id: {effective_approval_id}"); + if let Err(duplicate_approval) = insert_result { + warn!("Rejecting duplicate pending exec approval for call_id: {effective_approval_id}"); + duplicate_approval.send(ReviewDecision::Abort); + return rx_approve.await.unwrap_or(ReviewDecision::Abort); } let parsed_cmd = parse_command(&command); @@ -2237,7 +2239,7 @@ impl Session { // Add the tx_approve callback to the map before sending the request. let (tx_approve, rx_approve) = oneshot::channel(); let approval_id = call_id.clone(); - let prev_entry = { + let insert_result = { let mut active = self.active_turn.lock().await; match active.as_mut() { Some(at) => { @@ -2249,11 +2251,16 @@ impl Session { vec![ReviewDecision::Approved, ReviewDecision::ApprovedForSession], ) } - None => None, + None => { + tx_approve.send(ReviewDecision::Abort).ok(); + return rx_approve; + } } }; - if prev_entry.is_some() { - warn!("Overwriting existing pending approval for call_id: {approval_id}"); + if let Err(duplicate_approval) = insert_result { + warn!("Rejecting duplicate pending patch approval for call_id: {approval_id}"); + duplicate_approval.send(ReviewDecision::Abort); + return rx_approve; } let event = EventMsg::ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index ec4c45c05a..2b7c441f8a 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -793,6 +793,125 @@ fn accepted_command_decisions_add_only_exact_proposed_network_denials() { })); } +#[derive(Clone, Copy, Debug)] +enum DuplicateApprovalRequestKind { + Exec, + Patch, +} + +#[tokio::test] +async fn duplicate_same_kind_approval_request_aborts_without_replacing_or_emitting() { + for kind in [ + DuplicateApprovalRequestKind::Exec, + DuplicateApprovalRequestKind::Patch, + ] { + let (session, turn_context, rx_events) = make_session_and_context_with_rx().await; + *session.active_turn.lock().await = Some(ActiveTurn::default()); + let approval_id = format!("duplicate-{kind:?}"); + + match kind { + DuplicateApprovalRequestKind::Exec => { + let initial_request = tokio::spawn({ + let session = Arc::clone(&session); + let turn_context = Arc::clone(&turn_context); + let approval_id = approval_id.clone(); + async move { + session + .request_command_approval( + turn_context.as_ref(), + approval_id, + /*approval_id*/ None, + /*environment_id*/ None, + vec!["first".to_string()], + test_path_buf("/tmp").abs(), + /*reason*/ None, + /*network_approval_context*/ None, + /*proposed_execpolicy_amendment*/ None, + /*additional_permissions*/ None, + Some(vec![ReviewDecision::Approved, ReviewDecision::Abort]), + ) + .await + } + }); + let first_event = + tokio::time::timeout(std::time::Duration::from_secs(1), rx_events.recv()) + .await + .expect("initial exec approval event timed out") + .expect("initial exec approval event missing"); + assert!(matches!( + first_event.msg, + EventMsg::ExecApprovalRequest(request) + if request.effective_approval_id() == approval_id + )); + + let duplicate_decision = session + .request_command_approval( + turn_context.as_ref(), + approval_id.clone(), + /*approval_id*/ None, + /*environment_id*/ None, + vec!["second".to_string()], + test_path_buf("/tmp").abs(), + /*reason*/ None, + /*network_approval_context*/ None, + /*proposed_execpolicy_amendment*/ None, + /*additional_permissions*/ None, + Some(vec![ReviewDecision::Approved, ReviewDecision::Abort]), + ) + .await; + assert_eq!(duplicate_decision, ReviewDecision::Abort); + assert!(rx_events.try_recv().is_err()); + + session + .notify_exec_approval(&approval_id, ReviewDecision::Approved) + .await; + assert_eq!( + initial_request.await.expect("initial exec request task"), + ReviewDecision::Approved + ); + } + DuplicateApprovalRequestKind::Patch => { + let initial_rx = session + .request_patch_approval( + turn_context.as_ref(), + approval_id.clone(), + HashMap::new(), + /*reason*/ None, + /*grant_root*/ None, + ) + .await; + let first_event = + tokio::time::timeout(std::time::Duration::from_secs(1), rx_events.recv()) + .await + .expect("initial patch approval event timed out") + .expect("initial patch approval event missing"); + assert!(matches!( + first_event.msg, + EventMsg::ApplyPatchApprovalRequest(request) + if request.call_id == approval_id + )); + + let duplicate_rx = session + .request_patch_approval( + turn_context.as_ref(), + approval_id.clone(), + HashMap::new(), + /*reason*/ None, + /*grant_root*/ None, + ) + .await; + assert_eq!(duplicate_rx.await, Ok(ReviewDecision::Abort)); + assert!(rx_events.try_recv().is_err()); + + session + .notify_patch_approval(&approval_id, ReviewDecision::Approved) + .await; + assert_eq!(initial_rx.await, Ok(ReviewDecision::Approved)); + } + } + } +} + #[tokio::test] async fn start_managed_network_proxy_applies_execpolicy_network_rules() -> anyhow::Result<()> { let permission_profile = PermissionProfile::workspace_write(); diff --git a/codex-rs/core/src/state/turn.rs b/codex-rs/core/src/state/turn.rs index a036d97591..c6b23880b1 100644 --- a/codex-rs/core/src/state/turn.rs +++ b/codex-rs/core/src/state/turn.rs @@ -1,6 +1,7 @@ //! Turn-scoped state and active turn metadata scaffolding. use std::collections::HashMap; +use std::collections::hash_map::Entry; use std::sync::Arc; use tokio::sync::Mutex; use tokio::sync::Notify; @@ -139,14 +140,18 @@ impl TurnState { tx: oneshot::Sender, kind: PendingApprovalKind, accepted_decisions: Vec, - ) -> Option { - self.pending_approvals.insert( - (kind, key), - PendingApproval { - tx_approve: tx, - accepted_decisions, - }, - ) + ) -> Result<(), PendingApproval> { + let pending_approval = PendingApproval { + tx_approve: tx, + accepted_decisions, + }; + match self.pending_approvals.entry((kind, key)) { + Entry::Vacant(entry) => { + entry.insert(pending_approval); + Ok(()) + } + Entry::Occupied(_) => Err(pending_approval), + } } pub(crate) fn remove_pending_approval( diff --git a/codex-rs/core/src/state/turn_tests.rs b/codex-rs/core/src/state/turn_tests.rs index f3057018d3..8353cb256c 100644 --- a/codex-rs/core/src/state/turn_tests.rs +++ b/codex-rs/core/src/state/turn_tests.rs @@ -57,17 +57,25 @@ fn pending_approvals_with_the_same_id_are_isolated_by_request_kind() { let (tx_patch, _rx_patch) = oneshot::channel(); let (tx_exec, _rx_exec) = oneshot::channel(); let mut turn_state = TurnState::default(); - turn_state.insert_pending_approval( - "approval".to_string(), - tx_patch, - PendingApprovalKind::Patch, - vec![ReviewDecision::Approved], + assert!( + turn_state + .insert_pending_approval( + "approval".to_string(), + tx_patch, + PendingApprovalKind::Patch, + vec![ReviewDecision::Approved], + ) + .is_ok() ); - turn_state.insert_pending_approval( - "approval".to_string(), - tx_exec, - PendingApprovalKind::Exec, - vec![ReviewDecision::Approved], + assert!( + turn_state + .insert_pending_approval( + "approval".to_string(), + tx_exec, + PendingApprovalKind::Exec, + vec![ReviewDecision::Approved], + ) + .is_ok() ); assert!( @@ -83,7 +91,43 @@ fn pending_approvals_with_the_same_id_are_isolated_by_request_kind() { } #[tokio::test] -async fn consumed_approval_id_cannot_resolve_a_later_generation() { +async fn duplicate_same_kind_pending_approval_keeps_the_original_waiter() { + for kind in [PendingApprovalKind::Exec, PendingApprovalKind::Patch] { + let mut turn_state = TurnState::default(); + let (initial_tx, initial_rx) = oneshot::channel(); + assert!( + turn_state + .insert_pending_approval( + "call-id".to_string(), + initial_tx, + kind, + vec![ReviewDecision::Approved], + ) + .is_ok() + ); + + let (duplicate_tx, duplicate_rx) = oneshot::channel(); + let Err(duplicate) = turn_state.insert_pending_approval( + "call-id".to_string(), + duplicate_tx, + kind, + vec![ReviewDecision::Approved], + ) else { + panic!("duplicate {kind:?} approval id should be rejected"); + }; + duplicate.send(ReviewDecision::Abort); + assert_eq!(duplicate_rx.await, Ok(ReviewDecision::Abort)); + + turn_state + .remove_pending_approval("call-id", kind) + .expect("original waiter must remain pending") + .send(ReviewDecision::Approved); + assert_eq!(initial_rx.await, Ok(ReviewDecision::Approved)); + } +} + +#[tokio::test] +async fn consumed_approval_id_cannot_resolve_a_distinct_id_waiter() { for stale_decision in [ ReviewDecision::Approved, ReviewDecision::Abort, @@ -91,11 +135,15 @@ async fn consumed_approval_id_cannot_resolve_a_later_generation() { ] { let mut turn_state = TurnState::default(); let (initial_tx, initial_rx) = oneshot::channel(); - turn_state.insert_pending_approval( - "call-id".to_string(), - initial_tx, - PendingApprovalKind::Exec, - vec![ReviewDecision::Approved], + assert!( + turn_state + .insert_pending_approval( + "call-id".to_string(), + initial_tx, + PendingApprovalKind::Exec, + vec![ReviewDecision::Approved], + ) + .is_ok() ); turn_state .remove_pending_approval("call-id", PendingApprovalKind::Exec) @@ -104,11 +152,15 @@ async fn consumed_approval_id_cannot_resolve_a_later_generation() { assert_eq!(initial_rx.await, Ok(ReviewDecision::Approved)); let (retry_tx, retry_rx) = oneshot::channel(); - turn_state.insert_pending_approval( - "retry-id".to_string(), - retry_tx, - PendingApprovalKind::Exec, - vec![ReviewDecision::Approved], + assert!( + turn_state + .insert_pending_approval( + "retry-id".to_string(), + retry_tx, + PendingApprovalKind::Exec, + vec![ReviewDecision::Approved], + ) + .is_ok() ); assert!(