codex-rs/codex-mcp/src/elicitation.rs

//! MCP elicitation request tracking and policy handling.
//!
//! RMCP clients call into this module when a server asks Codex to elicit data
//! from the user. It decides whether the request can be automatically accepted,
//! must be declined by policy, or should be surfaced as a Codex protocol event
//! and later resolved through the stored responder.

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::Mutex as StdMutex;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;

use crate::McpConfig;
use crate::mcp::McpPermissionPromptAutoApproveContext;
use crate::mcp::mcp_permission_prompt_is_auto_approved;
use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use async_channel::Sender;
use codex_protocol::approvals::ElicitationRequest;
use codex_protocol::approvals::ElicitationRequestEvent;
use codex_protocol::mcp::ClientMcpExtensions;
use codex_protocol::mcp::OPENAI_ELICITATION_EXTENSION_ID;
use codex_protocol::mcp::RequestId as ProtocolRequestId;
use codex_protocol::mcp_approval_meta::APPROVAL_KIND_KEY;
use codex_protocol::mcp_approval_meta::APPROVAL_KIND_TOOL_SUGGESTION;
use codex_protocol::mcp_approval_meta::APPROVALS_REVIEWER_KEY;
use codex_protocol::mcp_approval_meta::STRICT_AUTO_REVIEW_KEY;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_rmcp_client::Elicitation;
use codex_rmcp_client::ElicitationResponse;
use codex_rmcp_client::SendElicitation;
use futures::future::BoxFuture;
use futures::future::FutureExt;
use rmcp::model::ElicitationAction;
use rmcp::model::RequestId;
use serde_json::Value;
use tokio::sync::oneshot;

static NEXT_ELICITATION_REQUEST_ID: AtomicU64 = AtomicU64::new(0);

const STRICT_AUTO_REVIEW_DECLINE_MESSAGE: &str = "Automated review of this operation failed. Do not proceed without asking the user for explicit approval.";

#[path = "user_verification_elicitation.rs"]
mod user_verification_elicitation;

#[derive(Debug, Clone)]
pub struct ElicitationReviewRequest {
    pub server_name: String,
    pub request_id: RequestId,
    pub elicitation: Elicitation,
}

pub trait ElicitationReviewer: Send + Sync {
    fn review(
        &self,
        request: ElicitationReviewRequest,
    ) -> BoxFuture<'static, Result<Option<ElicitationResponse>>>;
}

pub type ElicitationReviewerHandle = Arc<dyn ElicitationReviewer>;

/// Holds an owner-provided registration while an MCP elicitation is waiting for a response.
#[derive(Clone)]
pub struct ElicitationLifecycle {
    register: Arc<dyn Fn() -> Box<dyn Send + Sync> + Send + Sync>,
}

impl ElicitationLifecycle {
    pub fn new<T>(register: impl Fn() -> T + Send + Sync + 'static) -> Self
    where
        T: Send + Sync + 'static,
    {
        Self {
            register: Arc::new(move || Box::new(register())),
        }
    }

    fn start(&self) -> ActiveElicitation {
        ActiveElicitation {
            _registration: (self.register)(),
        }
    }
}

struct ActiveElicitation {
    _registration: Box<dyn Send + Sync>,
}

/// Routes model-visible elicitation response tokens to their exact pending responders.
///
/// One router is shared by every MCP runtime created for a thread. The public response token is
/// generated by Codex rather than copied from the MCP connection, so separate runtimes may reuse
/// the same server request ID without colliding.
#[derive(Clone, Default)]
pub(crate) struct ElicitationRequestRouter {
    requests: Arc<StdMutex<ResponderMap>>,
    auto_deny: Arc<AtomicBool>,
    full_access_form_input_enabled: Arc<AtomicBool>,
}

struct PendingElicitationRequest {
    router: ElicitationRequestRouter,
    key: (String, RequestId),
}

impl Drop for PendingElicitationRequest {
    fn drop(&mut self) {
        let responder = self
            .router
            .requests
            .lock()
            .ok()
            .and_then(|mut requests| requests.remove(&self.key));
        drop(responder);
    }
}

impl ElicitationRequestRouter {
    pub(crate) fn auto_deny(&self) -> bool {
        self.auto_deny.load(Ordering::Relaxed)
    }

    pub(crate) fn set_auto_deny(&self, auto_deny: bool) {
        self.auto_deny.store(auto_deny, Ordering::Relaxed);
    }

    pub(crate) fn full_access_form_input_enabled(&self) -> bool {
        self.full_access_form_input_enabled.load(Ordering::Acquire)
    }

    pub(crate) fn enable_full_access_form_input(&self) {
        self.full_access_form_input_enabled
            .store(true, Ordering::Release);
    }

    pub(crate) async fn resolve(
        &self,
        server_name: String,
        id: RequestId,
        response: ElicitationResponse,
    ) -> Result<()> {
        let responder = self
            .requests
            .lock()
            .map_err(|_| anyhow!("elicitation request router unavailable"))?
            .remove(&(server_name, id))
            .ok_or_else(|| anyhow!("elicitation request not found"))?;
        responder
            .send(response)
            .map_err(|_| anyhow!("elicitation response receiver closed"))
    }
}

#[derive(Clone)]
pub(crate) struct ElicitationAuthority {
    pub(crate) config: Arc<McpConfig>,
    reviewer: Option<ElicitationReviewerHandle>,
    lifecycle: Option<ElicitationLifecycle>,
}

#[derive(Clone, Default)]
pub(crate) struct ElicitationRequestManager {
    router: ElicitationRequestRouter,
    pub(crate) authority: Arc<StdMutex<Option<ElicitationAuthority>>>,
}

impl ElicitationRequestManager {
    pub(crate) fn new(
        config: Arc<McpConfig>,
        reviewer: Option<ElicitationReviewerHandle>,
        lifecycle: Option<ElicitationLifecycle>,
        router: ElicitationRequestRouter,
    ) -> Self {
        Self {
            router,
            authority: Arc::new(StdMutex::new(Some(ElicitationAuthority {
                config,
                reviewer,
                lifecycle,
            }))),
        }
    }

    pub(crate) fn update(
        &self,
        config: Arc<McpConfig>,
        reviewer: Option<ElicitationReviewerHandle>,
        lifecycle: Option<ElicitationLifecycle>,
    ) -> bool {
        let Ok(mut authority) = self.authority.lock() else {
            return false;
        };
        *authority = Some(ElicitationAuthority {
            config,
            reviewer,
            lifecycle,
        });
        true
    }

    pub(crate) fn make_sender(
        &self,
        server_name: String,
        tx_event: Option<Sender<Event>>,
        client_mcp_extensions: &ClientMcpExtensions,
    ) -> SendElicitation {
        // Event receivers such as codex mcp-server do not necessarily handle verification.
        // Only wait for a response when trusted host activation enabled this exact route.
        let user_verification_enabled = client_mcp_extensions
            .get(OPENAI_ELICITATION_EXTENSION_ID)
            .and_then(|settings| settings.get("userVerification"))
            .is_some_and(Value::is_object);
        let router = self.router.clone();
        let authority = self.authority.clone();
        Box::new(move |id, elicitation| {
            let router = router.clone();
            let tx_event = tx_event.clone();
            let server_name = server_name.clone();
            let authority = authority.clone();
            async move {
                if let Elicitation::UserVerification {
                    title,
                    description,
                    challenge,
                } = elicitation
                {
                    if !user_verification_enabled {
                        return Ok(ElicitationResponse {
                            action: ElicitationAction::Cancel,
                            content: None,
                            meta: None,
                        });
                    }
                    return user_verification_elicitation::route(
                        router,
                        tx_event,
                        authority,
                        server_name,
                        ElicitationRequest::UserVerification {
                            title,
                            description,
                            challenge,
                        },
                    )
                    .await;
                }
                if router.auto_deny() {
                    return Ok(ElicitationResponse {
                        action: ElicitationAction::Decline,
                        content: None,
                        meta: None,
                    });
                }

                if elicitation
                    .meta()
                    .and_then(|meta| meta.get(APPROVAL_KIND_KEY))
                    .and_then(serde_json::Value::as_str)
                    == Some(APPROVAL_KIND_TOOL_SUGGESTION)
                {
                    return Ok(ElicitationResponse {
                        action: ElicitationAction::Decline,
                        content: None,
                        meta: None,
                    });
                }

                let Ok(Some(authority)) = authority.lock().map(|authority| authority.clone())
                else {
                    return Ok(ElicitationResponse {
                        action: ElicitationAction::Decline,
                        content: None,
                        meta: None,
                    });
                };
                let ElicitationAuthority {
                    config,
                    reviewer,
                    lifecycle,
                } = authority;
                let approval_policy = config.approval_policy.value();
                let Some(permission_profile) = config.permission_profile_for_server(&server_name)
                else {
                    return Ok(ElicitationResponse {
                        action: ElicitationAction::Decline,
                        content: None,
                        meta: None,
                    });
                };

                match elicitation
                    .meta()
                    .and_then(|meta| meta.get(STRICT_AUTO_REVIEW_KEY))
                {
                    Some(Value::Bool(true)) => {
                        if matches!(
                            approval_policy,
                            AskForApproval::Granular(config) if !config.allows_mcp_elicitations()
                        ) {
                            return Ok(strict_auto_review_decline());
                        }
                        let Some(reviewer) = reviewer.as_ref() else {
                            return Ok(strict_auto_review_decline());
                        };
                        let _active_elicitation =
                            lifecycle.as_ref().map(ElicitationLifecycle::start);
                        return Ok(
                            match reviewer
                                .review(ElicitationReviewRequest {
                                    server_name,
                                    request_id: id,
                                    elicitation,
                                })
                                .await
                            {
                                Ok(Some(response))
                                    if (response.action == ElicitationAction::Accept
                                        && response.content == Some(serde_json::json!({}))
                                        || matches!(
                                            response.action,
                                            ElicitationAction::Decline | ElicitationAction::Cancel
                                        ) && response.content.is_none())
                                        && response
                                            .meta
                                            .as_ref()
                                            .and_then(Value::as_object)
                                            .and_then(|meta| meta.get(APPROVALS_REVIEWER_KEY))
                                            .and_then(Value::as_str)
                                            == Some("auto_review") =>
                                {
                                    response
                                }
                                Ok(Some(_)) | Ok(None) | Err(_) => strict_auto_review_decline(),
                            },
                        );
                    }
                    None | Some(Value::Bool(false)) => {}
                    Some(_) => return Ok(strict_auto_review_decline()),
                }

                let permission_prompt_is_auto_approved = mcp_permission_prompt_is_auto_approved(
                    approval_policy,
                    permission_profile,
                    McpPermissionPromptAutoApproveContext::default(),
                );
                if permission_prompt_is_auto_approved && can_auto_accept_elicitation(&elicitation) {
                    return Ok(ElicitationResponse {
                        action: ElicitationAction::Accept,
                        content: Some(serde_json::json!({})),
                        meta: None,
                    });
                }

                let should_surface_form_in_full_access = router.full_access_form_input_enabled()
                    && permission_prompt_is_auto_approved
                    && !elicitation
                        .meta()
                        .is_some_and(|meta| meta.contains_key(APPROVAL_KIND_KEY))
                    && match &elicitation {
                        Elicitation::Mcp(
                            rmcp::model::ElicitRequestParams::FormElicitationParams {
                                requested_schema,
                                ..
                            },
                        ) => !requested_schema.properties.is_empty(),
                        Elicitation::OpenAiElicitationForm {
                            requested_schema, ..
                        } => requested_schema
                            .get("properties")
                            .and_then(Value::as_object)
                            .is_some_and(|properties| !properties.is_empty()),
                        Elicitation::Mcp(_)
                        | Elicitation::OpenAiForm { .. }
                        | Elicitation::UserVerification { .. } => false,
                    };

                if !should_surface_form_in_full_access {
                    if elicitation_is_rejected_by_policy(approval_policy) {
                        return Ok(ElicitationResponse {
                            action: ElicitationAction::Decline,
                            content: None,
                            meta: None,
                        });
                    }

                    if let Some(reviewer) = reviewer {
                        let request = ElicitationReviewRequest {
                            server_name: server_name.clone(),
                            request_id: id.clone(),
                            elicitation: elicitation.clone(),
                        };
                        if let Some(response) = reviewer.review(request).await? {
                            return Ok(response);
                        }
                    }
                }

                let Some(tx_event) = tx_event else {
                    return Ok(ElicitationResponse {
                        action: ElicitationAction::Decline,
                        content: None,
                        meta: None,
                    });
                };

                let public_request_id = format!(
                    "codex-mcp-elicitation-{}",
                    NEXT_ELICITATION_REQUEST_ID.fetch_add(1, Ordering::Relaxed)
                );
                let routed_request_id = RequestId::String(public_request_id.clone().into());
                let request = match elicitation {
                    Elicitation::UserVerification { .. } => {
                        return Ok(ElicitationResponse {
                            action: ElicitationAction::Cancel,
                            content: None,
                            meta: None,
                        });
                    }
                    Elicitation::Mcp(rmcp::model::ElicitRequestParams::FormElicitationParams {
                        meta,
                        message,
                        requested_schema,
                    }) => ElicitationRequest::Form {
                        meta: meta
                            .map(serde_json::to_value)
                            .transpose()
                            .context("failed to serialize MCP elicitation metadata")?,
                        message,
                        requested_schema: serde_json::to_value(requested_schema)
                            .context("failed to serialize MCP elicitation schema")?,
                    },
                    Elicitation::Mcp(rmcp::model::ElicitRequestParams::UrlElicitationParams {
                        meta,
                        message,
                        url,
                        elicitation_id,
                    }) => ElicitationRequest::Url {
                        meta: meta
                            .map(serde_json::to_value)
                            .transpose()
                            .context("failed to serialize MCP elicitation metadata")?,
                        message,
                        url,
                        elicitation_id,
                    },
                    Elicitation::Mcp(_) => {
                        return Ok(ElicitationResponse {
                            action: ElicitationAction::Decline,
                            content: None,
                            meta: None,
                        });
                    }
                    Elicitation::OpenAiForm {
                        meta,
                        message,
                        requested_schema,
                    } => ElicitationRequest::OpenAiForm {
                        meta,
                        message,
                        requested_schema,
                    },
                    Elicitation::OpenAiElicitationForm {
                        meta,
                        message,
                        requested_schema,
                    } => ElicitationRequest::OpenAiElicitationForm {
                        meta,
                        message,
                        requested_schema,
                    },
                };
                let (tx, rx) = oneshot::channel();
                let _active_elicitation = lifecycle.as_ref().map(ElicitationLifecycle::start);
                let request_key = (server_name.clone(), routed_request_id);
                router
                    .requests
                    .lock()
                    .map_err(|_| anyhow!("elicitation request router unavailable"))?
                    .insert(request_key.clone(), tx);
                let _pending_request = PendingElicitationRequest {
                    router: router.clone(),
                    key: request_key,
                };
                tx_event
                    .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("failed to deliver MCP elicitation request")?;
                rx.await
                    .context("elicitation request channel closed unexpectedly")
            }
            .boxed()
        })
    }
}

fn strict_auto_review_decline() -> ElicitationResponse {
    ElicitationResponse {
        action: ElicitationAction::Decline,
        content: None,
        meta: Some(serde_json::json!({
            "message": STRICT_AUTO_REVIEW_DECLINE_MESSAGE,
        })),
    }
}

pub(crate) fn elicitation_is_rejected_by_policy(approval_policy: AskForApproval) -> bool {
    match approval_policy {
        AskForApproval::Never => true,
        AskForApproval::OnRequest => false,
        AskForApproval::UnlessTrusted => false,
        AskForApproval::Granular(granular_config) => !granular_config.allows_mcp_elicitations(),
    }
}

type ResponderMap = HashMap<(String, RequestId), oneshot::Sender<ElicitationResponse>>;

fn can_auto_accept_elicitation(elicitation: &Elicitation) -> bool {
    match elicitation {
        Elicitation::Mcp(rmcp::model::ElicitRequestParams::FormElicitationParams {
            requested_schema,
            ..
        }) => {
            // Auto-accept confirm/approval elicitations without schema requirements.
            requested_schema.properties.is_empty()
        }
        Elicitation::Mcp(_)
        | Elicitation::OpenAiForm { .. }
        | Elicitation::OpenAiElicitationForm { .. }
        | Elicitation::UserVerification { .. } => false,
    }
}

#[cfg(test)]
#[path = "elicitation_tests.rs"]
mod tests;