mirror of
https://github.com/openai/codex.git
synced 2026-10-19 16:17:49 +00:00
Signal abandonment of unanswered MCP elicitations (#51611)
## Why Published MCP elicitation requests could lose their response waiter without notifying the event consumer that the request was no longer answerable. ## What changed - Add `ElicitationAbandoned` with the original server name and response token for requests published through unbounded event channels. - Emit abandonment exactly once when a waiter is dropped, response delivery fails, or the MCP runtime shuts down. - Close the elicitation router during shutdown, reject new requests on unbounded sources, and wait for terminal deliveries before shutdown completes. Preserve bounded sources' request-only behavior and backpressure. ## Testing Add router tests for ordered, exactly-once abandonment, closed receivers, failed response delivery, shutdown coordination, and bounded-channel backpressure. Add a core integration scenario verifying that a pending form's abandonment precedes `ShutdownComplete`. GitOrigin-RevId: 53fe9f4a8b32f7324be180288aa492f31f4d2057
This commit is contained in:
BIN
Binary file not shown.
@@ -20,6 +20,7 @@ use anyhow::Context;
|
||||
use anyhow::Result;
|
||||
use anyhow::anyhow;
|
||||
use async_channel::Sender;
|
||||
use codex_protocol::approvals::ElicitationAbandonedEvent;
|
||||
use codex_protocol::approvals::ElicitationRequest;
|
||||
use codex_protocol::approvals::ElicitationRequestEvent;
|
||||
use codex_protocol::mcp::ClientMcpExtensions;
|
||||
@@ -99,29 +100,148 @@ struct ActiveElicitation {
|
||||
/// the same server request ID without colliding.
|
||||
#[derive(Clone, Default)]
|
||||
pub(crate) struct ElicitationRequestRouter {
|
||||
requests: Arc<StdMutex<ResponderMap>>,
|
||||
requests: Arc<StdMutex<PendingResponders>>,
|
||||
terminal_actions_finished: Arc<tokio::sync::Notify>,
|
||||
auto_deny: Arc<AtomicBool>,
|
||||
full_access_form_input_enabled: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
struct PendingElicitationRequest {
|
||||
// Removes and abandons a published request when its native waiter is dropped.
|
||||
struct ElicitationRequestGuard {
|
||||
router: ElicitationRequestRouter,
|
||||
key: (String, RequestId),
|
||||
}
|
||||
|
||||
impl Drop for PendingElicitationRequest {
|
||||
#[derive(Default, PartialEq)]
|
||||
enum RouterPhase {
|
||||
#[default]
|
||||
Open,
|
||||
Closing,
|
||||
Closed,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct PendingResponders {
|
||||
pending: ResponderMap,
|
||||
phase: RouterPhase,
|
||||
terminal_actions: usize,
|
||||
}
|
||||
|
||||
// Owns response delivery and the original source for abandonment notifications.
|
||||
struct ElicitationResponder {
|
||||
response: oneshot::Sender<ElicitationResponse>,
|
||||
abandonment_events: Option<Sender<Event>>,
|
||||
}
|
||||
|
||||
// Removal elects a winner, but shutdown must also wait for its outside-lock delivery.
|
||||
struct ElicitationDeliveryGuard(ElicitationRequestRouter);
|
||||
|
||||
impl Drop for ElicitationDeliveryGuard {
|
||||
fn drop(&mut self) {
|
||||
let responder = self
|
||||
.router
|
||||
let mut state = self
|
||||
.0
|
||||
.requests
|
||||
.lock()
|
||||
.ok()
|
||||
.and_then(|mut requests| requests.remove(&self.key));
|
||||
drop(responder);
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
state.terminal_actions -= 1;
|
||||
if state.phase == RouterPhase::Closing && state.terminal_actions == 0 {
|
||||
state.phase = RouterPhase::Closed;
|
||||
}
|
||||
drop(state);
|
||||
self.0.terminal_actions_finished.notify_waiters();
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for ElicitationRequestGuard {
|
||||
fn drop(&mut self) {
|
||||
if let Ok(Some((responder, _terminal_action))) = self.router.take_responder(&self.key) {
|
||||
emit_abandonment(&self.key, responder.abandonment_events.as_ref());
|
||||
drop(responder);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn emit_abandonment(key: &(String, RequestId), events: Option<&Sender<Event>>) {
|
||||
if let Some(events) = events {
|
||||
let event = Event {
|
||||
id: "mcp_elicitation_abandoned".to_string(),
|
||||
msg: EventMsg::ElicitationAbandoned(ElicitationAbandonedEvent {
|
||||
server_name: key.0.clone(),
|
||||
id: match &key.1 {
|
||||
RequestId::String(id) => ProtocolRequestId::String(id.to_string()),
|
||||
RequestId::Number(id) => ProtocolRequestId::Integer(*id),
|
||||
},
|
||||
}),
|
||||
};
|
||||
match events.try_send(event) {
|
||||
Ok(()) | Err(async_channel::TrySendError::Closed(_)) => {}
|
||||
Err(async_channel::TrySendError::Full(_)) => {
|
||||
unreachable!("validated unbounded elicitation event channel became full")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl ElicitationRequestRouter {
|
||||
fn take_responder(
|
||||
&self,
|
||||
key: &(String, RequestId),
|
||||
) -> Result<Option<(ElicitationResponder, ElicitationDeliveryGuard)>> {
|
||||
let mut state = self
|
||||
.requests
|
||||
.lock()
|
||||
.map_err(|_| anyhow!("elicitation request router unavailable"))?;
|
||||
Ok(state.pending.remove(key).map(|responder| {
|
||||
state.terminal_actions += 1;
|
||||
(responder, ElicitationDeliveryGuard(self.clone()))
|
||||
}))
|
||||
}
|
||||
|
||||
/// Fence unbounded source publication and terminal deliveries before reporting shutdown.
|
||||
/// Bounded sources retain their existing request-only behavior and backpressure.
|
||||
pub(crate) async fn close(&self) {
|
||||
{
|
||||
let (pending, _terminal_action) = {
|
||||
let mut state = self
|
||||
.requests
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
if state.phase == RouterPhase::Closed {
|
||||
return;
|
||||
}
|
||||
if state.phase == RouterPhase::Open {
|
||||
state.phase = RouterPhase::Closing;
|
||||
}
|
||||
let pending = state
|
||||
.pending
|
||||
.extract_if(|_, responder| responder.abandonment_events.is_some())
|
||||
.collect::<ResponderMap>();
|
||||
state.terminal_actions += 1;
|
||||
(pending, ElicitationDeliveryGuard(self.clone()))
|
||||
};
|
||||
// This synchronous batch completes before the first cancellation point.
|
||||
for (key, responder) in pending {
|
||||
emit_abandonment(&key, responder.abandonment_events.as_ref());
|
||||
drop(responder);
|
||||
}
|
||||
}
|
||||
loop {
|
||||
let notified = self.terminal_actions_finished.notified();
|
||||
tokio::pin!(notified);
|
||||
notified.as_mut().enable();
|
||||
if self
|
||||
.requests
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.phase
|
||||
== RouterPhase::Closed
|
||||
{
|
||||
return;
|
||||
}
|
||||
notified.await;
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn auto_deny(&self) -> bool {
|
||||
self.auto_deny.load(Ordering::Relaxed)
|
||||
}
|
||||
@@ -145,15 +265,15 @@ impl ElicitationRequestRouter {
|
||||
id: RequestId,
|
||||
response: ElicitationResponse,
|
||||
) -> Result<()> {
|
||||
let responder = self
|
||||
.requests
|
||||
.lock()
|
||||
.map_err(|_| anyhow!("elicitation request router unavailable"))?
|
||||
.remove(&(server_name, id))
|
||||
let key = (server_name, id);
|
||||
let (responder, _terminal_action) = self
|
||||
.take_responder(&key)?
|
||||
.ok_or_else(|| anyhow!("elicitation request not found"))?;
|
||||
responder
|
||||
.send(response)
|
||||
.map_err(|_| anyhow!("elicitation response receiver closed"))
|
||||
if responder.response.send(response).is_err() {
|
||||
emit_abandonment(&key, responder.abandonment_events.as_ref());
|
||||
return Err(anyhow!("elicitation response receiver closed"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn request_user_interaction(
|
||||
@@ -170,6 +290,10 @@ impl ElicitationRequestRouter {
|
||||
meta: None,
|
||||
});
|
||||
};
|
||||
// Core uses an unbounded source channel, which can atomically publish a request
|
||||
// and synchronously enqueue abandonment from Drop. Preserve bounded callers'
|
||||
// existing request-only behavior because they require asynchronous backpressure.
|
||||
let abandonment_events = events.capacity().is_none().then(|| events.clone());
|
||||
let (delivery_context, response_context) =
|
||||
if matches!(&request, ElicitationRequest::UserVerification { .. }) {
|
||||
(
|
||||
@@ -195,26 +319,61 @@ impl ElicitationRequestRouter {
|
||||
.lifecycle
|
||||
.as_ref()
|
||||
.map(ElicitationLifecycle::start);
|
||||
self.requests
|
||||
.lock()
|
||||
.map_err(|_| anyhow!("elicitation request router unavailable"))?
|
||||
.insert(request_key.clone(), tx);
|
||||
let _pending_request = PendingElicitationRequest {
|
||||
let event = Event {
|
||||
id: "mcp_elicitation_request".to_string(),
|
||||
msg: EventMsg::ElicitationRequest(ElicitationRequestEvent {
|
||||
turn_id: None,
|
||||
server_name,
|
||||
id: ProtocolRequestId::String(public_request_id),
|
||||
request,
|
||||
}),
|
||||
};
|
||||
let legacy_event = if abandonment_events.is_some() {
|
||||
let mut state = self
|
||||
.requests
|
||||
.lock()
|
||||
.map_err(|_| anyhow!("elicitation request router unavailable"))?;
|
||||
anyhow::ensure!(
|
||||
state.phase == RouterPhase::Open,
|
||||
"elicitation request router closed"
|
||||
);
|
||||
let mut responder =
|
||||
state
|
||||
.pending
|
||||
.entry(request_key.clone())
|
||||
.insert_entry(ElicitationResponder {
|
||||
response: tx,
|
||||
abandonment_events: None,
|
||||
});
|
||||
// Resolve and close cannot observe a published request without its responder.
|
||||
// The validated unbounded send cannot wait or yield while admission is locked.
|
||||
if let Err(error) = events.try_send(event) {
|
||||
responder.remove();
|
||||
return Err(error).context(delivery_context);
|
||||
}
|
||||
responder.get_mut().abandonment_events = abandonment_events;
|
||||
None
|
||||
} else {
|
||||
self.requests
|
||||
.lock()
|
||||
.map_err(|_| anyhow!("elicitation request router unavailable"))?
|
||||
.pending
|
||||
.insert(
|
||||
request_key.clone(),
|
||||
ElicitationResponder {
|
||||
response: tx,
|
||||
abandonment_events: None,
|
||||
},
|
||||
);
|
||||
Some(event)
|
||||
};
|
||||
let _pending_request = ElicitationRequestGuard {
|
||||
router: self.clone(),
|
||||
key: request_key,
|
||||
};
|
||||
events
|
||||
.send(Event {
|
||||
id: "mcp_elicitation_request".to_string(),
|
||||
msg: EventMsg::ElicitationRequest(ElicitationRequestEvent {
|
||||
turn_id: None,
|
||||
server_name,
|
||||
id: ProtocolRequestId::String(public_request_id),
|
||||
request,
|
||||
}),
|
||||
})
|
||||
.await
|
||||
.context(delivery_context)?;
|
||||
if let Some(event) = legacy_event {
|
||||
events.send(event).await.context(delivery_context)?;
|
||||
}
|
||||
rx.await.context(response_context)
|
||||
}
|
||||
}
|
||||
@@ -558,7 +717,7 @@ pub(crate) fn elicitation_is_rejected_by_policy(approval_policy: AskForApproval)
|
||||
}
|
||||
}
|
||||
|
||||
type ResponderMap = HashMap<(String, RequestId), oneshot::Sender<ElicitationResponse>>;
|
||||
type ResponderMap = HashMap<(String, RequestId), ElicitationResponder>;
|
||||
|
||||
fn can_auto_accept_elicitation(elicitation: &Elicitation) -> bool {
|
||||
match elicitation {
|
||||
|
||||
@@ -14,6 +14,309 @@ use std::sync::atomic::Ordering::Relaxed;
|
||||
|
||||
type ReviewerResponse = std::result::Result<Option<ElicitationResponse>, &'static str>;
|
||||
|
||||
#[tokio::test]
|
||||
async fn abandonment_is_ordered_and_exactly_once_when_the_native_waiter_is_dropped() {
|
||||
let (manager, _, _) = verification_fixture(AskForApproval::OnRequest, /*reviewer*/ None);
|
||||
let router = manager.router.clone();
|
||||
let authority = manager.authority.lock().unwrap().clone().unwrap();
|
||||
let (tx, events) = async_channel::unbounded();
|
||||
let (replacement_tx, replacement_events) = async_channel::unbounded();
|
||||
let request = ElicitationRequest::Form {
|
||||
meta: None,
|
||||
message: "Continue".into(),
|
||||
requested_schema: json!({"type": "object"}),
|
||||
};
|
||||
let mut first = Box::pin(router.request_user_interaction(
|
||||
Some(tx),
|
||||
&authority,
|
||||
"server".into(),
|
||||
request.clone(),
|
||||
));
|
||||
let mut sibling = Box::pin(router.request_user_interaction(
|
||||
Some(replacement_tx),
|
||||
&authority,
|
||||
"server".into(),
|
||||
request,
|
||||
));
|
||||
assert!(futures::poll!(&mut first).is_pending());
|
||||
let EventMsg::ElicitationRequest(first_request) = events.recv().await.unwrap().msg else {
|
||||
panic!("request must precede abandonment");
|
||||
};
|
||||
assert!(futures::poll!(&mut sibling).is_pending());
|
||||
let EventMsg::ElicitationRequest(sibling_request) =
|
||||
replacement_events.recv().await.unwrap().msg
|
||||
else {
|
||||
panic!("expected sibling request");
|
||||
};
|
||||
drop(first);
|
||||
let EventMsg::ElicitationAbandoned(abandoned) = events.recv().await.unwrap().msg else {
|
||||
panic!("expected abandonment");
|
||||
};
|
||||
assert_eq!(
|
||||
abandoned,
|
||||
ElicitationAbandonedEvent {
|
||||
server_name: first_request.server_name,
|
||||
id: first_request.id.clone()
|
||||
}
|
||||
);
|
||||
assert_eq!(router.requests.lock().unwrap().pending.len(), 1);
|
||||
let ProtocolRequestId::String(id) = first_request.id else {
|
||||
panic!("generated token")
|
||||
};
|
||||
assert!(
|
||||
router
|
||||
.resolve(
|
||||
"server".into(),
|
||||
RequestId::String(id.into()),
|
||||
approved_response()
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
let ProtocolRequestId::String(id) = sibling_request.id else {
|
||||
panic!("generated token")
|
||||
};
|
||||
router
|
||||
.resolve(
|
||||
"server".into(),
|
||||
RequestId::String(id.into()),
|
||||
approved_response(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(sibling.await.unwrap(), approved_response());
|
||||
assert!(events.try_recv().is_err());
|
||||
assert!(replacement_events.try_recv().is_err());
|
||||
assert!(router.requests.lock().unwrap().pending.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn bounded_sources_preserve_backpressure_across_close() {
|
||||
let (manager, _, _) = verification_fixture(AskForApproval::OnRequest, /*reviewer*/ None);
|
||||
let authority = manager.authority.lock().unwrap().clone().unwrap();
|
||||
let router = manager.router.clone();
|
||||
let (tx, events) = async_channel::bounded(1);
|
||||
tx.send(Event {
|
||||
id: "queued".into(),
|
||||
msg: EventMsg::ShutdownComplete,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
let mut pending = Box::pin(router.request_user_interaction(
|
||||
Some(tx),
|
||||
&authority,
|
||||
"server".into(),
|
||||
ElicitationRequest::Form {
|
||||
meta: None,
|
||||
message: "Continue".into(),
|
||||
requested_schema: json!({"type": "object"}),
|
||||
},
|
||||
));
|
||||
assert!(futures::poll!(&mut pending).is_pending());
|
||||
router.close().await;
|
||||
assert_eq!(router.requests.lock().unwrap().pending.len(), 1);
|
||||
assert_eq!(events.recv().await.unwrap().id, "queued");
|
||||
assert!(futures::poll!(&mut pending).is_pending());
|
||||
let EventMsg::ElicitationRequest(request) = events.recv().await.unwrap().msg else {
|
||||
panic!("bounded request must publish after capacity becomes available");
|
||||
};
|
||||
let ProtocolRequestId::String(id) = request.id else {
|
||||
panic!("expected generated token");
|
||||
};
|
||||
router
|
||||
.resolve(
|
||||
"server".into(),
|
||||
RequestId::String(id.into()),
|
||||
approved_response(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(pending.await.unwrap(), approved_response());
|
||||
assert!(events.try_recv().is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn abandonment_closed_receiver_cleans_up_without_delivery_or_replay() {
|
||||
let (manager, _, _) = verification_fixture(AskForApproval::OnRequest, /*reviewer*/ None);
|
||||
let authority = manager.authority.lock().unwrap().clone().unwrap();
|
||||
let router = manager.router.clone();
|
||||
for close_before_publication in [true, false] {
|
||||
let (tx, events) = async_channel::unbounded();
|
||||
if close_before_publication {
|
||||
events.close();
|
||||
}
|
||||
let mut pending = Box::pin(router.request_user_interaction(
|
||||
Some(tx),
|
||||
&authority,
|
||||
"server".into(),
|
||||
ElicitationRequest::Form {
|
||||
meta: None,
|
||||
message: "Continue".into(),
|
||||
requested_schema: json!({"type": "object"}),
|
||||
},
|
||||
));
|
||||
if close_before_publication {
|
||||
assert!(pending.await.is_err());
|
||||
} else {
|
||||
assert!(futures::poll!(&mut pending).is_pending());
|
||||
assert!(matches!(
|
||||
events.recv().await.unwrap().msg,
|
||||
EventMsg::ElicitationRequest(_)
|
||||
));
|
||||
events.close();
|
||||
drop(pending);
|
||||
}
|
||||
assert!(router.requests.lock().unwrap().pending.is_empty());
|
||||
assert!(events.try_recv().is_err());
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn abandonment_close_joins_elected_actions_and_rejects_retained_router_admission() {
|
||||
let (manager, _, _) = verification_fixture(AskForApproval::OnRequest, /*reviewer*/ None);
|
||||
let authority = manager.authority.lock().unwrap().clone().unwrap();
|
||||
let router = manager.router.clone();
|
||||
let retained_router = router.clone();
|
||||
let (tx, events) = async_channel::unbounded();
|
||||
let request = ElicitationRequest::Form {
|
||||
meta: None,
|
||||
message: "Continue".into(),
|
||||
requested_schema: json!({"type": "object"}),
|
||||
};
|
||||
let mut pending = Box::pin(router.request_user_interaction(
|
||||
Some(tx.clone()),
|
||||
&authority,
|
||||
"server".into(),
|
||||
request.clone(),
|
||||
));
|
||||
assert!(futures::poll!(&mut pending).is_pending());
|
||||
let EventMsg::ElicitationRequest(published) = events.recv().await.unwrap().msg else {
|
||||
panic!("expected published request");
|
||||
};
|
||||
let ProtocolRequestId::String(id) = &published.id else {
|
||||
panic!("expected generated token");
|
||||
};
|
||||
let key = (
|
||||
published.server_name.clone(),
|
||||
RequestId::String(id.clone().into()),
|
||||
);
|
||||
let (responder, terminal_action) = router.take_responder(&key).unwrap().unwrap();
|
||||
let mut close = Box::pin(router.close());
|
||||
assert!(futures::poll!(&mut close).is_pending());
|
||||
drop(close);
|
||||
let mut repeated_close = Box::pin(retained_router.close());
|
||||
assert!(futures::poll!(&mut repeated_close).is_pending());
|
||||
assert!(
|
||||
retained_router
|
||||
.request_user_interaction(
|
||||
Some(tx.clone()),
|
||||
&authority,
|
||||
"server".into(),
|
||||
request.clone(),
|
||||
)
|
||||
.await
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("router closed")
|
||||
);
|
||||
assert!(events.try_recv().is_err());
|
||||
emit_abandonment(&key, responder.abandonment_events.as_ref());
|
||||
drop(responder);
|
||||
drop(terminal_action);
|
||||
repeated_close.await;
|
||||
assert!(pending.await.is_err());
|
||||
let EventMsg::ElicitationAbandoned(abandoned) = events.recv().await.unwrap().msg else {
|
||||
panic!("close must wait for the winning remover to enqueue abandonment");
|
||||
};
|
||||
assert_eq!(
|
||||
abandoned,
|
||||
ElicitationAbandonedEvent {
|
||||
server_name: published.server_name,
|
||||
id: published.id,
|
||||
}
|
||||
);
|
||||
router.close().await;
|
||||
assert!(
|
||||
retained_router
|
||||
.request_user_interaction(Some(tx), &authority, "server".into(), request,)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
assert!(events.try_recv().is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn abandonment_close_invalidates_a_published_responder_before_returning() {
|
||||
let (manager, _, _) = verification_fixture(AskForApproval::OnRequest, /*reviewer*/ None);
|
||||
let authority = manager.authority.lock().unwrap().clone().unwrap();
|
||||
let router = manager.router.clone();
|
||||
let (tx, events) = async_channel::unbounded();
|
||||
let mut pending = Box::pin(router.request_user_interaction(
|
||||
Some(tx),
|
||||
&authority,
|
||||
"server".into(),
|
||||
ElicitationRequest::Form {
|
||||
meta: None,
|
||||
message: "Continue".into(),
|
||||
requested_schema: json!({"type": "object"}),
|
||||
},
|
||||
));
|
||||
assert!(futures::poll!(&mut pending).is_pending());
|
||||
let EventMsg::ElicitationRequest(published) = events.recv().await.unwrap().msg else {
|
||||
panic!("expected request");
|
||||
};
|
||||
router.close().await;
|
||||
assert!(pending.await.is_err());
|
||||
let EventMsg::ElicitationAbandoned(abandoned) = events.try_recv().unwrap().msg else {
|
||||
panic!("close must enqueue abandonment before returning");
|
||||
};
|
||||
assert_eq!(
|
||||
abandoned,
|
||||
ElicitationAbandonedEvent {
|
||||
server_name: published.server_name,
|
||||
id: published.id,
|
||||
}
|
||||
);
|
||||
assert!(events.try_recv().is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn abandonment_resolve_reports_failed_response_delivery_once() {
|
||||
let router = ElicitationRequestRouter::default();
|
||||
let (tx, events) = async_channel::unbounded();
|
||||
let (response, receiver) = tokio::sync::oneshot::channel();
|
||||
let key = (
|
||||
"server".to_string(),
|
||||
RequestId::String("published-token".into()),
|
||||
);
|
||||
router.requests.lock().unwrap().pending.insert(
|
||||
key.clone(),
|
||||
ElicitationResponder {
|
||||
response,
|
||||
abandonment_events: Some(tx),
|
||||
},
|
||||
);
|
||||
drop(receiver);
|
||||
assert!(
|
||||
router
|
||||
.resolve(key.0, key.1, approved_response())
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
router.close().await;
|
||||
let EventMsg::ElicitationAbandoned(abandoned) = events.try_recv().unwrap().msg else {
|
||||
panic!("failed oneshot delivery must abandon the published request");
|
||||
};
|
||||
assert_eq!(
|
||||
abandoned,
|
||||
ElicitationAbandonedEvent {
|
||||
server_name: "server".into(),
|
||||
id: ProtocolRequestId::String("published-token".into()),
|
||||
}
|
||||
);
|
||||
assert!(events.try_recv().is_err());
|
||||
}
|
||||
|
||||
struct RecordingReviewer {
|
||||
calls: AtomicUsize,
|
||||
active_elicitations: Arc<AtomicUsize>,
|
||||
@@ -189,7 +492,7 @@ async fn permission_elicitation_preserves_automatic_review_decisions() {
|
||||
assert_eq!(reviewer.calls.load(Relaxed), expected_reviews);
|
||||
assert_eq!(reviewer.active_elicitations.load(Relaxed), 0);
|
||||
assert!(events.is_empty());
|
||||
assert!(manager.router.requests.lock().unwrap().is_empty());
|
||||
assert!(manager.router.requests.lock().unwrap().pending.is_empty());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -254,7 +557,7 @@ async fn elicitation_without_an_automatic_decision_prompts() {
|
||||
assert_eq!(reviewer.active_elicitations.load(Relaxed), 0);
|
||||
}
|
||||
assert!(events.is_empty());
|
||||
assert!(manager.router.requests.lock().unwrap().is_empty());
|
||||
assert!(manager.router.requests.lock().unwrap().pending.is_empty());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -320,6 +623,7 @@ fn closed_event_channel_immediately_cleans_up_pending_elicitation() {
|
||||
.requests
|
||||
.lock()
|
||||
.expect("pending request router should be available")
|
||||
.pending
|
||||
.is_empty()
|
||||
);
|
||||
assert_eq!(active_elicitations.load(Relaxed), 0);
|
||||
@@ -716,7 +1020,7 @@ async fn user_verification_cancels_for_an_event_receiver_without_host_activation
|
||||
},
|
||||
);
|
||||
assert!(events.try_recv().is_err());
|
||||
assert!(manager.router.requests.lock().unwrap().is_empty());
|
||||
assert!(manager.router.requests.lock().unwrap().pending.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -748,7 +1052,7 @@ async fn user_verification_drops_pending_response_when_the_request_is_cancelled(
|
||||
assert_eq!(active_elicitations.load(Relaxed), 1);
|
||||
pending.abort();
|
||||
assert!(pending.await.unwrap_err().is_cancelled());
|
||||
assert!(manager.router.requests.lock().unwrap().is_empty());
|
||||
assert!(manager.router.requests.lock().unwrap().pending.is_empty());
|
||||
assert_eq!(active_elicitations.load(Relaxed), 0);
|
||||
}
|
||||
|
||||
|
||||
@@ -83,6 +83,7 @@ pub struct McpRuntimeInput {
|
||||
pub ready_selected_capability_roots: Vec<SelectedCapabilityRoot>,
|
||||
pub mcp_servers: HashMap<String, EffectiveMcpServer>,
|
||||
pub submit_id: String,
|
||||
/// Unbounded sources receive abandonment notifications; bounded sources retain request-only behavior.
|
||||
pub tx_event: Option<Sender<Event>>,
|
||||
pub startup_cancellation_token: CancellationToken,
|
||||
pub runtime_context: McpRuntimeContext,
|
||||
@@ -778,6 +779,7 @@ impl McpRuntime {
|
||||
}
|
||||
|
||||
pub async fn shutdown(&self) {
|
||||
self.elicitation_router.close().await;
|
||||
self.latest_connections().shutdown().await;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2128,6 +2128,7 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<RealtimeEventTex
|
||||
| EventMsg::McpStartupComplete(_)
|
||||
| EventMsg::McpToolCallBegin(_)
|
||||
| EventMsg::McpToolCallEnd(_)
|
||||
| EventMsg::ElicitationAbandoned(_)
|
||||
| EventMsg::WebSearchBegin(_)
|
||||
| EventMsg::WebSearchEnd(_)
|
||||
| EventMsg::ExecCommandBegin(_)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
//! MCP server elicitations reach subagents and preserve automatic approval and review.
|
||||
//! MCP server elicitations preserve approval policy and signal abandonment before shutdown.
|
||||
|
||||
use anyhow::Result;
|
||||
use codex_core::StartThreadOptions;
|
||||
@@ -75,12 +75,14 @@ pub(super) enum Caller {
|
||||
pub(super) enum RequestKind {
|
||||
BrowserAuth,
|
||||
FormInput,
|
||||
ShutdownFormInput,
|
||||
Permission,
|
||||
StrictReview,
|
||||
}
|
||||
|
||||
#[test_case(Caller::FullAccessSubagent, RequestKind::BrowserAuth; "full_access_subagent_preserves_automatic_approval")]
|
||||
#[test_case(Caller::Root, RequestKind::BrowserAuth; "root_browser_auth_remains_interactive")]
|
||||
#[test_case(Caller::Root, RequestKind::ShutdownFormInput; "pending_root_form_is_abandoned_before_shutdown")]
|
||||
#[test_case(Caller::Subagent, RequestKind::FormInput; "subagent_receives_form_input")]
|
||||
#[test_case(Caller::Subagent, RequestKind::Permission; "subagent_permission_remains_interactive")]
|
||||
#[test_case(Caller::FullAccessSubagent, RequestKind::Permission; "full_access_subagent_automatically_approves_permission")]
|
||||
@@ -118,7 +120,7 @@ pub(super) async fn mcp_server_elicitation_scenario(
|
||||
"fields": [{"id": "password", "label": "Password", "type": "password", "required": true}]
|
||||
});
|
||||
}
|
||||
RequestKind::FormInput => {
|
||||
RequestKind::FormInput | RequestKind::ShutdownFormInput => {
|
||||
elicitation["requestedSchema"]["properties"] = json!({"answer": {"type": "string"}});
|
||||
}
|
||||
RequestKind::Permission | RequestKind::StrictReview => {
|
||||
@@ -255,6 +257,39 @@ pub(super) async fn mcp_server_elicitation_scenario(
|
||||
follow_up.requests().is_empty(),
|
||||
"the tool must wait for input"
|
||||
);
|
||||
if matches!(request_kind, RequestKind::ShutdownFormInput) {
|
||||
thread.submit(Op::Shutdown).await?;
|
||||
let EventMsg::ElicitationAbandoned(abandoned) = wait_for_event(&thread, |event| {
|
||||
matches!(
|
||||
event,
|
||||
EventMsg::ElicitationAbandoned(_) | EventMsg::ShutdownComplete
|
||||
)
|
||||
})
|
||||
.await
|
||||
else {
|
||||
panic!("abandonment must precede shutdown completion");
|
||||
};
|
||||
assert_eq!(
|
||||
abandoned,
|
||||
codex_protocol::approvals::ElicitationAbandonedEvent {
|
||||
server_name: request.server_name,
|
||||
id: request.id,
|
||||
}
|
||||
);
|
||||
assert!(matches!(
|
||||
wait_for_event(&thread, |event| {
|
||||
matches!(
|
||||
event,
|
||||
EventMsg::ElicitationAbandoned(_) | EventMsg::ShutdownComplete
|
||||
)
|
||||
})
|
||||
.await,
|
||||
EventMsg::ShutdownComplete
|
||||
));
|
||||
assert!(follow_up.requests().is_empty());
|
||||
test.codex.shutdown_and_wait().await?;
|
||||
return Ok(tool_call.requests());
|
||||
}
|
||||
thread
|
||||
.submit(Op::ResolveElicitation {
|
||||
server_name: request.server_name,
|
||||
|
||||
@@ -447,6 +447,14 @@ pub enum ElicitationRequest {
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema, TS)]
|
||||
pub struct ElicitationAbandonedEvent {
|
||||
pub server_name: String,
|
||||
/// Original Core-generated response token of the unanswered request.
|
||||
#[ts(type = "string | number")]
|
||||
pub id: RequestId,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema, TS)]
|
||||
pub struct ElicitationRequestEvent {
|
||||
/// Turn ID that this elicitation belongs to, when known.
|
||||
|
||||
@@ -19,6 +19,7 @@ use crate::ResponseItemId;
|
||||
use crate::SanitizedGitUrl;
|
||||
use crate::SessionId;
|
||||
use crate::ThreadId;
|
||||
use crate::approvals::ElicitationAbandonedEvent;
|
||||
use crate::approvals::ElicitationRequestEvent;
|
||||
use crate::capabilities::SelectedCapabilityRoot;
|
||||
use crate::config_types::ApprovalsReviewer;
|
||||
@@ -1536,6 +1537,9 @@ pub enum EventMsg {
|
||||
|
||||
ElicitationRequest(ElicitationRequestEvent),
|
||||
|
||||
/// An unanswered MCP responder was abandoned at its original unbounded source.
|
||||
ElicitationAbandoned(ElicitationAbandonedEvent),
|
||||
|
||||
ApplyPatchApprovalRequest(ApplyPatchApprovalRequestEvent),
|
||||
|
||||
/// Structured lifecycle event for a guardian-reviewed approval request.
|
||||
|
||||
@@ -398,6 +398,7 @@ pub(crate) fn tool_runtime_trace_event(event: &EventMsg) -> Option<ToolRuntimeTr
|
||||
| EventMsg::DynamicToolCallRequest(_)
|
||||
| EventMsg::DynamicToolCallResponse(_)
|
||||
| EventMsg::ElicitationRequest(_)
|
||||
| EventMsg::ElicitationAbandoned(_)
|
||||
| EventMsg::ApplyPatchApprovalRequest(_)
|
||||
| EventMsg::GuardianAssessment(_)
|
||||
| EventMsg::DeprecationNotice(_)
|
||||
@@ -477,6 +478,7 @@ pub(crate) fn wrapped_protocol_event_type(event: &EventMsg) -> Option<&'static s
|
||||
| EventMsg::DynamicToolCallRequest(_)
|
||||
| EventMsg::DynamicToolCallResponse(_)
|
||||
| EventMsg::ElicitationRequest(_)
|
||||
| EventMsg::ElicitationAbandoned(_)
|
||||
| EventMsg::ApplyPatchApprovalRequest(_)
|
||||
| EventMsg::GuardianAssessment(_)
|
||||
| EventMsg::DeprecationNotice(_)
|
||||
|
||||
@@ -196,6 +196,7 @@ fn persisted_event_msg(
|
||||
| EventMsg::RequestPermissions(_)
|
||||
| EventMsg::RequestUserInput(_)
|
||||
| EventMsg::ElicitationRequest(_)
|
||||
| EventMsg::ElicitationAbandoned(_)
|
||||
| EventMsg::ApplyPatchApprovalRequest(_)
|
||||
| EventMsg::StreamError(_)
|
||||
| EventMsg::PatchApplyBegin(_)
|
||||
|
||||
Reference in New Issue
Block a user