Add cancellation for native user-verification RPCs (#43925)

## Why

Canceling an elicitation does not stop a separate native verification RPC. Clients need a way to cancel that work and prevent a canceled proof from being delivered while it waits for outbound queue capacity.

## What changed

- Add experimental `userVerification/cancel` with `{requestId}` to signal cancellation of status, enroll, delete, or verify requests on the same connection.
- Return an empty acknowledgment without waiting for native work to exit. Unknown, finished, unrelated, and other-connection requests are no-ops.
- Retain request cancellation state until outbound capacity is available so canceled proofs are checked before enqueueing. Already enqueued responses and completed effects cannot be retracted.
- Document the API, request ID requirements, and independent lifetimes of elicitations and native verification RPCs.

## Testing

Add coverage for experimental opt-in, parameter validation, connection and request ID isolation, cancellation before native dispatch, repeated cancellation, worker-slot retention, and proof cancellation while waiting for outbound capacity.

GitOrigin-RevId: 03fcb83fcc5fab4169c68668822fc0bbb4970938

codex-rs/app-server-protocol/src/protocol/common.rs

@@ -546,6 +546,13 @@ client_request_definitions! {
         response: v2::UserVerificationVerifyResponse,
     },
 
+    #[experimental("userVerification/cancel")]
+    UserVerificationCancel => "userVerification/cancel" {
+        params: v2::UserVerificationCancelParams,
+        serialization: None,
+        response: v2::UserVerificationCancelResponse,
+    },
+
     /// NEW APIs
     // Thread lifecycle
     // Uses `inspect_params` because only some fields are experimental.

codex-rs/app-server-protocol/src/protocol/v2/user_verification.rs

@@ -2,6 +2,7 @@
 //! Native implementation and registration transport are independent of these contracts.
 
 use crate::JsonSchema;
+use crate::RequestId;
 use crate::TS;
 use serde::Deserialize;
 use serde::Serialize;
@@ -145,6 +146,22 @@ pub struct UserVerificationVerifyResponse {
     pub proof: UserVerificationProof,
 }
 
+/// Cancels a native verification RPC issued on this connection, not an elicitation.
+/// Use a fresh request ID for each operation and a distinct ID for this cancellation RPC.
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+#[ts(export_to = "v2/")]
+pub struct UserVerificationCancelParams {
+    pub request_id: RequestId,
+}
+
+/// Acknowledges the cancellation signal; native work may still be finishing.
+/// Unknown or finished requests are a no-op, and completed effects are not rolled back.
+#[derive(Serialize, Deserialize, Debug, Default, Clone, PartialEq, Eq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export_to = "v2/")]
+pub struct UserVerificationCancelResponse {}
+
 #[cfg(test)]
 #[path = "user_verification_tests.rs"]
 mod tests;

codex-rs/app-server/README.md

@@ -1,3 +1,27 @@
+# User verification cancellation (experimental)
+
+Local UI clients can cancel a native user-verification RPC by sending
+`userVerification/cancel` with `{requestId}` and the `experimentalApi` opt-in.
+The result is an empty acknowledgment (`{}`). This API does not enable desktop
+verification capability advertisement.
+
+`requestId` is the original status, enroll, delete, or verify RPC's string or
+integer ID on the same connection, not the server elicitation ID. Use fresh IDs
+for each operation and a distinct ID for the cancel RPC. Unknown, finished,
+unrelated, and other-connection requests are no-ops.
+
+The acknowledgment confirms the cancellation signal without waiting for the OS
+prompt to close. The original RPC completes independently, with
+`cancelled/interrupted` when cancellation prevents completion. Cancellation
+cannot roll back completed effects. It remains effective while a proof waits for
+outbound queue capacity, but cannot retract a response already enqueued.
+
+Canceling or resolving an elicitation does not itself stop a separate
+`userVerification/verify` RPC. Clients must cancel that RPC separately and discard
+late proofs after the approval is canceled or resolved. Only one native worker
+runs per app-server; if an OS call remains active after cancellation or timeout,
+subsequent local operations return `failed/providerError` until that worker exits.
+
 # Thread removal
 
 `thread/archive` and `thread/delete` reject attempts to remove a live internal

codex-rs/app-server/src/message_processor.rs

@@ -70,6 +70,7 @@ use codex_app_server_protocol::JSONRPCErrorError;
 use codex_app_server_protocol::JSONRPCNotification;
 use codex_app_server_protocol::JSONRPCRequest;
 use codex_app_server_protocol::JSONRPCResponse;
+use codex_app_server_protocol::UserVerificationCancelResponse;
 use codex_app_server_protocol::experimental_required_message;
 use codex_arg0::Arg0DispatchPaths;
 use codex_code_mode::CodeModeSessionProvider;
@@ -631,7 +632,12 @@ impl MessageProcessor {
             traceparent: trace.traceparent.clone(),
             tracestate: trace.tracestate.clone(),
         });
-        let request_context = RequestContext::new(request_id.clone(), request_span, request_trace);
+        let request_context = RequestContext::new(
+            request_id.clone(),
+            request_method,
+            request_span,
+            request_trace,
+        );
         Self::run_request_with_context(
             Arc::clone(&self.outgoing),
             request_context.clone(),
@@ -680,8 +686,12 @@ impl MessageProcessor {
         };
         let request_span =
             crate::app_server_tracing::typed_request_span(&request, connection_id, &session);
-        let mut request_context =
-            RequestContext::new(request_id.clone(), request_span, /*parent_trace*/ None);
+        let mut request_context = RequestContext::new(
+            request_id.clone(),
+            request.method_name(),
+            request_span,
+            /*parent_trace*/ None,
+        );
         request_context.cancellation = cancellation;
         tracing::trace!(
             ?connection_id,
@@ -995,6 +1005,15 @@ impl MessageProcessor {
             ClientRequest::Initialize { .. } => {
                 panic!("Initialize should be handled before initialized request dispatch");
             }
+            ClientRequest::UserVerificationCancel { params, .. } => {
+                self.outgoing
+                    .cancel_user_verification_request(&ConnectionRequestId {
+                        connection_id,
+                        request_id: params.request_id,
+                    })
+                    .await;
+                Ok(Some(UserVerificationCancelResponse {}.into()))
+            }
             request @ (ClientRequest::UserVerificationStatus { .. }
             | ClientRequest::UserVerificationEnroll { .. }
             | ClientRequest::UserVerificationDelete { .. }

codex-rs/app-server/src/outgoing_message.rs

@@ -56,26 +56,40 @@ pub(crate) struct ConnectionRequestId {
     pub(crate) request_id: RequestId,
 }
 
-/// Trace data we keep for an incoming request until we send its final
-/// response or error.
+/// Trace data and cancellation state retained until an incoming request's final response or error.
 #[derive(Clone)]
 pub(crate) struct RequestContext {
     request_id: ConnectionRequestId,
     pub(crate) cancellation: tokio_util::sync::CancellationToken,
+    cancellation_scope: RequestCancellationScope,
     span: Span,
     parent_trace: Option<W3cTraceContext>,
     _diagnostics_guard: Arc<GaugeGuard>,
 }
 
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum RequestCancellationScope {
+    Unavailable,
+    UserVerification,
+}
+
 impl RequestContext {
     pub(crate) fn new(
         request_id: ConnectionRequestId,
+        method: &str,
         span: Span,
         parent_trace: Option<W3cTraceContext>,
     ) -> Self {
         Self {
             request_id,
             cancellation: tokio_util::sync::CancellationToken::new(),
+            cancellation_scope: match method {
+                "userVerification/status"
+                | "userVerification/enroll"
+                | "userVerification/delete"
+                | "userVerification/verify" => RequestCancellationScope::UserVerification,
+                _ => RequestCancellationScope::Unavailable,
+            },
             span,
             parent_trace,
             _diagnostics_guard: Arc::new(IN_FLIGHT_REQUESTS.track()),
@@ -253,6 +267,15 @@ impl OutgoingMessageSender {
         }
     }
 
+    pub(crate) async fn cancel_user_verification_request(&self, request_id: &ConnectionRequestId) {
+        let contexts = self.request_contexts.lock().await;
+        if let Some(context) = contexts.get(request_id)
+            && context.cancellation_scope == RequestCancellationScope::UserVerification
+        {
+            context.cancellation.cancel();
+        }
+    }
+
     pub(crate) async fn connection_closed(&self, connection_id: ConnectionId) {
         self.disconnect_user_verification_connection(connection_id)
             .await;
@@ -648,8 +671,10 @@ impl OutgoingMessageSender {
         response: ClientResponsePayload,
         check: impl FnOnce() -> std::result::Result<(), JSONRPCErrorError>,
     ) {
+        // Remain cancellable while waiting to deliver a proof, including after native work ends.
+        let permit = self.sender.reserve().await;
         let _context = self.take_request_context(&request_id).await;
-        let Ok(permit) = self.sender.reserve().await else {
+        let Ok(permit) = permit else {
             return;
         };
         let message = match check() {
@@ -877,6 +902,10 @@ fn timestamped_server_notification(notification: ServerNotification) -> Outgoing
 #[path = "user_verification_ownership_tests.rs"]
 mod user_verification_ownership_tests;
 
+#[cfg(test)]
+#[path = "user_verification_cancel_context_tests.rs"]
+mod user_verification_cancel_context_tests;
+
 #[cfg(test)]
 mod tests {
     use std::time::Duration;
@@ -1257,6 +1286,7 @@ mod tests {
         outgoing
             .register_request_context(RequestContext::new(
                 request_id.clone(),
+                "thread/start",
                 tracing::info_span!("app_server.request", rpc.method = "thread/start"),
                 /*parent_trace*/ None,
             ))
@@ -1416,6 +1446,7 @@ mod tests {
         outgoing
             .register_request_context(RequestContext::new(
                 closed_connection_request,
+                "turn/interrupt",
                 tracing::info_span!("app_server.request", rpc.method = "turn/interrupt"),
                 /*parent_trace*/ None,
             ))
@@ -1423,6 +1454,7 @@ mod tests {
         outgoing
             .register_request_context(RequestContext::new(
                 open_connection_request,
+                "turn/start",
                 tracing::info_span!("app_server.request", rpc.method = "turn/start"),
                 /*parent_trace*/ None,
             ))

codex-rs/app-server/src/user_verification.rs

@@ -253,6 +253,10 @@ mod tests;
 #[path = "user_verification_rpc_tests.rs"]
 mod rpc_tests;
 
+#[cfg(test)]
+#[path = "user_verification_cancel_tests.rs"]
+mod cancel_tests;
+
 #[cfg(test)]
 #[path = "user_verification_test_support.rs"]
 mod test_support;

codex-rs/app-server/src/user_verification_cancel_context_tests.rs

@@ -0,0 +1,126 @@
+//! Checks cancellation ownership and the lifetime of proofs waiting for outbound capacity.
+
+use super::*;
+use codex_app_server_protocol::UserVerificationDeleteResponse;
+use codex_app_server_protocol::UserVerificationProof;
+use codex_app_server_protocol::UserVerificationVerifyResponse;
+use pretty_assertions::assert_eq;
+
+#[tokio::test]
+async fn user_verification_cancel_only_signals_matching_native_request_contexts() {
+    let (tx, _rx) = mpsc::channel(/*buffer*/ 1);
+    let outgoing = OutgoingMessageSender::new(tx, AnalyticsEventsClient::disabled());
+    for method in [
+        "userVerification/status",
+        "userVerification/enroll",
+        "userVerification/delete",
+        "userVerification/verify",
+        "userVerification/cancel",
+        "turn/start",
+    ] {
+        let target = ConnectionRequestId {
+            connection_id: ConnectionId(1),
+            request_id: RequestId::Integer(7),
+        };
+        let context = RequestContext::new(
+            target.clone(),
+            method,
+            Span::none(),
+            /*parent_trace*/ None,
+        );
+        let cancellation = context.cancellation.clone();
+        outgoing.register_request_context(context).await;
+        for wrong_target in [
+            ConnectionRequestId {
+                connection_id: ConnectionId(2),
+                request_id: RequestId::Integer(7),
+            },
+            ConnectionRequestId {
+                connection_id: ConnectionId(1),
+                request_id: RequestId::String("7".into()),
+            },
+            ConnectionRequestId {
+                connection_id: ConnectionId(1),
+                request_id: RequestId::Integer(8),
+            },
+        ] {
+            outgoing
+                .cancel_user_verification_request(&wrong_target)
+                .await;
+            assert!(!cancellation.is_cancelled());
+        }
+        outgoing.cancel_user_verification_request(&target).await;
+        outgoing.cancel_user_verification_request(&target).await;
+        assert_eq!(
+            cancellation.is_cancelled(),
+            method != "turn/start" && method != "userVerification/cancel"
+        );
+        assert_eq!(outgoing.request_context_count().await, 1);
+        outgoing.connection_closed(ConnectionId(1)).await;
+        assert_eq!(outgoing.request_context_count().await, 0);
+    }
+}
+
+#[tokio::test]
+async fn user_verification_cancel_remains_effective_until_proof_enqueue_and_cleans_up() {
+    for close_receiver in [false, true] {
+        let (tx, mut rx) = mpsc::channel(/*buffer*/ 1);
+        let outgoing = OutgoingMessageSender::new(tx, AnalyticsEventsClient::disabled());
+        let target = ConnectionRequestId {
+            connection_id: ConnectionId(1),
+            request_id: RequestId::String("verify".into()),
+        };
+        outgoing
+            .send_response(target.clone(), UserVerificationDeleteResponse {})
+            .await;
+        let context = RequestContext::new(
+            target.clone(),
+            "userVerification/verify",
+            Span::none(),
+            /*parent_trace*/ None,
+        );
+        let cancellation = context.cancellation.clone();
+        outgoing.register_request_context(context).await;
+        let response = UserVerificationVerifyResponse {
+            proof: UserVerificationProof {
+                credential_id: "AQ".into(),
+                signature: "Ag".into(),
+            },
+        };
+        let error = internal_error("canceled proof");
+        let expected_error = error.clone();
+        let sending =
+            outgoing.send_response_as_checked(target.clone(), response.into(), move || {
+                if cancellation.is_cancelled() {
+                    Err(error)
+                } else {
+                    Ok(())
+                }
+            });
+        let mut sending = std::pin::pin!(sending);
+        assert!(matches!(
+            futures::poll!(&mut sending),
+            std::task::Poll::Pending
+        ));
+        assert_eq!(outgoing.request_context_count().await, 1);
+        outgoing.cancel_user_verification_request(&target).await;
+        if close_receiver {
+            rx.close();
+        } else {
+            rx.recv().await.unwrap();
+        }
+        sending.await;
+        assert_eq!(outgoing.request_context_count().await, 0);
+        if !close_receiver {
+            let OutgoingEnvelope::ToConnection {
+                message: OutgoingMessage::Error(error),
+                ..
+            } = rx.recv().await.unwrap()
+            else {
+                panic!("a proof canceled while queued must not be sent")
+            };
+            assert_eq!(error.id, target.request_id);
+            assert_eq!(error.error, expected_error);
+        }
+    }
+}

codex-rs/app-server/src/user_verification_cancel_tests.rs

@@ -0,0 +1,172 @@
+//! Checks that client RPC cancellation reaches native work without owning an elicitation.
+
+use super::test_support::Harness;
+use super::*;
+use crate::outgoing_message::ConnectionId;
+use crate::outgoing_message::ConnectionRequestId;
+use crate::outgoing_message::OutgoingEnvelope;
+use crate::outgoing_message::OutgoingMessage;
+use anyhow::Result;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+use std::sync::atomic::Ordering;
+
+fn assert_cancelled(message: OutgoingMessage) {
+    let OutgoingMessage::Error(error) = message else {
+        panic!("canceled verification must not return a proof")
+    };
+    assert_eq!(error.id, rpc::RequestId::Integer(1));
+    assert_eq!(
+        error.error,
+        rpc::JSONRPCErrorError {
+            code: -32603,
+            message: "User verification was cancelled.".into(),
+            data: Some(json!({"type": "cancelled", "reason": "interrupted"})),
+        }
+    );
+}
+
+#[tokio::test]
+async fn user_verification_cancel_rpc_stops_native_work_and_allows_another_operation() -> Result<()>
+{
+    let mut h = Harness::new(ConnectionOrigin::Stdio, || true).await?;
+    h.initialize("Codex Desktop", /*opt_in*/ true).await;
+    h.start_verify().await?;
+    h.send(
+        /*id*/ 2,
+        "userVerification/cancel",
+        json!({"requestId": 1}),
+    )
+    .await;
+    for _ in 0..2 {
+        let message = h.response().await;
+        if let OutgoingMessage::Response(response) = message {
+            assert_eq!(response.id, rpc::RequestId::Integer(2));
+            assert_eq!(serde_json::to_value(response.result)?, json!({}));
+        } else {
+            assert_cancelled(message);
+        }
+    }
+    assert_eq!(h.service.worker.available_permits(), 1);
+    // Repeated and already-finished cancellation must not cancel the next operation.
+    h.send(
+        /*id*/ 3,
+        "userVerification/cancel",
+        json!({"requestId": 1}),
+    )
+    .await;
+    assert!(matches!(h.response().await, OutgoingMessage::Response(_)));
+    h.send(/*id*/ 4, "userVerification/status", json!({})).await;
+    let OutgoingMessage::Response(response) = h.response().await else {
+        panic!("the next native operation must remain available")
+    };
+    assert_eq!(response.id, rpc::RequestId::Integer(4));
+    h.shutdown().await;
+    Ok(())
+}
+
+#[tokio::test]
+async fn user_verification_cancel_is_registered_before_native_dispatch() -> Result<()> {
+    let mut h = Harness::new(ConnectionOrigin::Stdio, || true).await?;
+    h.initialize("Codex Desktop", /*opt_in*/ true).await;
+    h.send(
+        /*id*/ 1,
+        "userVerification/verify",
+        json!({"challenge": "AQ", "title": "Approve", "description": ""}),
+    )
+    .await;
+    // This current-thread executor has not polled the spawned RPC handler yet.
+    h.outgoing
+        .cancel_user_verification_request(&ConnectionRequestId {
+            connection_id: ConnectionId(1),
+            request_id: rpc::RequestId::Integer(1),
+        })
+        .await;
+    assert_cancelled(h.response().await);
+    assert_eq!(h.provider.calls.load(Ordering::SeqCst), 0);
+    h.shutdown().await;
+    Ok(())
+}
+
+#[tokio::test]
+async fn user_verification_cancel_rpc_cannot_target_another_connection_or_id_type() -> Result<()> {
+    let mut h = Harness::new(ConnectionOrigin::Stdio, || true).await?;
+    h.initialize("Codex Desktop", /*opt_in*/ true).await;
+    h.start_verify().await?;
+    h.send(
+        /*id*/ 2,
+        "userVerification/cancel",
+        json!({"requestId": "1"}),
+    )
+    .await;
+    assert!(matches!(h.response().await, OutgoingMessage::Response(_)));
+    h.processor
+        .process_request(
+            ConnectionId(2),
+            rpc::JSONRPCRequest {
+                id: rpc::RequestId::Integer(3),
+                method: "userVerification/cancel".into(),
+                params: Some(json!({"requestId": 1})),
+                trace: None,
+            },
+            &crate::transport::AppServerTransport::Stdio,
+            Arc::clone(&h.session),
+        )
+        .await;
+    let OutgoingEnvelope::ToConnection {
+        connection_id,
+        message: OutgoingMessage::Response(response),
+        ..
+    } = tokio::time::timeout(Duration::from_secs(/*secs*/ 5), h.messages.recv())
+        .await?
+        .unwrap()
+    else {
+        panic!("another connection must only receive its own cancel acknowledgment")
+    };
+    assert_eq!(connection_id, ConnectionId(2));
+    assert_eq!(response.id, rpc::RequestId::Integer(3));
+    assert_eq!(h.service.worker.available_permits(), 0);
+    h.outgoing
+        .cancel_user_verification_request(&ConnectionRequestId {
+            connection_id: ConnectionId(1),
+            request_id: rpc::RequestId::Integer(1),
+        })
+        .await;
+    assert_cancelled(h.response().await);
+    h.shutdown().await;
+    Ok(())
+}
+
+#[tokio::test]
+async fn user_verification_cancel_keeps_the_worker_slot_until_native_exit() -> Result<()> {
+    let mut h = Harness::new(ConnectionOrigin::Stdio, || true).await?;
+    h.initialize("Codex Desktop", /*opt_in*/ true).await;
+    h.provider
+        .ignore_cancellation
+        .store(/*val*/ true, Ordering::Release);
+    h.start_verify().await?;
+    h.send(
+        /*id*/ 2,
+        "userVerification/cancel",
+        json!({"requestId": 1}),
+    )
+    .await;
+    let OutgoingMessage::Response(response) = h.response().await else {
+        panic!("cancel acknowledgment must not wait for the native worker")
+    };
+    assert_eq!(response.id, rpc::RequestId::Integer(2));
+    assert_eq!(h.service.worker.available_permits(), 0);
+    h.send(/*id*/ 3, "userVerification/status", json!({})).await;
+    let OutgoingMessage::Error(error) = h.response().await else {
+        panic!("the native worker must remain bounded")
+    };
+    assert_eq!(
+        error.error.data,
+        Some(json!({"type": "failed", "reason": "providerError"}))
+    );
+    h.provider.released.store(/*val*/ true, Ordering::Release);
+    assert_cancelled(h.response().await);
+    assert_eq!(h.service.worker.available_permits(), 1);
+    h.shutdown().await;
+    Ok(())
+}

codex-rs/app-server/src/user_verification_test_support.rs

@@ -40,6 +40,7 @@ pub(super) struct BlockingProvider {
     pub(super) entered: Mutex<Option<oneshot::Sender<()>>>,
     pub(super) released: AtomicBool,
     pub(super) calls: AtomicUsize,
+    pub(super) ignore_cancellation: AtomicBool,
 }
 
 impl native::UserVerificationProvider for BlockingProvider {
@@ -77,7 +78,9 @@ impl native::UserVerificationProvider for BlockingProvider {
         }
         let deadline = std::time::Instant::now() + Duration::from_secs(/*secs*/ 10);
         while !self.released.load(Ordering::Acquire) {
-            guard.check()?;
+            if !self.ignore_cancellation.load(Ordering::Acquire) {
+                guard.check()?;
+            }
             assert!(
                 std::time::Instant::now() < deadline,
                 "test failed to release native worker"

codex-rs/app-server/tests/suite/v2/user_verification.rs

@@ -34,6 +34,7 @@ async fn user_verification_methods_require_experimental_opt_in() -> Result<()> {
         ("userVerification/status", json!({})),
         ("userVerification/enroll", json!({})),
         ("userVerification/delete", json!({})),
+        ("userVerification/cancel", json!({"requestId": 1})),
         (
             "userVerification/verify",
             json!({"challenge": "AQ", "title": "Approve", "description": ""}),
@@ -75,6 +76,49 @@ async fn user_verification_status_without_account_returns_local_readiness() -> R
     Ok(())
 }
 
+#[tokio::test]
+async fn user_verification_cancel_stdio_validates_and_acknowledges_unknown_requests() -> Result<()>
+{
+    let mut app_server = TestAppServer::builder().build().await?;
+    let id = app_server
+        .send_raw_request(
+            "userVerification/cancel",
+            Some(json!({"requestId": "verify"})),
+        )
+        .await?;
+    let error = app_server
+        .read_stream_until_error_message(RequestId::Integer(id))
+        .await?;
+    assert_eq!(error.error.message, "Not initialized");
+    app_server.initialize().await?;
+    for params in [
+        json!({}),
+        json!({"requestId": null}),
+        json!({"requestId": 1.5}),
+        json!({"requestId": 1, "connectionId": 2}),
+    ] {
+        let id = app_server
+            .send_raw_request("userVerification/cancel", Some(params))
+            .await?;
+        app_server
+            .read_stream_until_error_message(RequestId::Integer(id))
+            .await?;
+    }
+    for request_id in [json!(100), json!("100"), json!("100")] {
+        let id = app_server
+            .send_raw_request(
+                "userVerification/cancel",
+                Some(json!({"requestId": request_id})),
+            )
+            .await?;
+        let response = app_server
+            .read_stream_until_response_message(RequestId::Integer(id))
+            .await?;
+        assert_eq!(response.result, json!({}));
+    }
+    Ok(())
+}
+
 #[tokio::test]
 async fn user_verification_without_provider_returns_typed_unavailability() -> Result<()> {
     let mut app_server = TestAppServer::builder().build_initialized().await?;