diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 397ba2f016..400ef070be 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 005e14a3ca..947f2bfb0b 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/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. diff --git a/codex-rs/app-server-protocol/src/protocol/v2/user_verification.rs b/codex-rs/app-server-protocol/src/protocol/v2/user_verification.rs index 93ad4b8d7c..99321969d4 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/user_verification.rs +++ b/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; diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 2b9d476b67..80aede118c 100644 --- a/codex-rs/app-server/README.md +++ b/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 diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index c462bb738f..aff6d9ad6c 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/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 { .. } diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 5c20dad5d7..ec84a83262 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/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, _diagnostics_guard: Arc, } +#[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, ) -> 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, )) diff --git a/codex-rs/app-server/src/user_verification.rs b/codex-rs/app-server/src/user_verification.rs index 2345ffe5c4..95c36e96b5 100644 --- a/codex-rs/app-server/src/user_verification.rs +++ b/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; diff --git a/codex-rs/app-server/src/user_verification_cancel_context_tests.rs b/codex-rs/app-server/src/user_verification_cancel_context_tests.rs new file mode 100644 index 0000000000..0b571e2f0d --- /dev/null +++ b/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); + } + } +} diff --git a/codex-rs/app-server/src/user_verification_cancel_tests.rs b/codex-rs/app-server/src/user_verification_cancel_tests.rs new file mode 100644 index 0000000000..a284e1a0fb --- /dev/null +++ b/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(()) +} diff --git a/codex-rs/app-server/src/user_verification_test_support.rs b/codex-rs/app-server/src/user_verification_test_support.rs index 413be31d79..d382e9972b 100644 --- a/codex-rs/app-server/src/user_verification_test_support.rs +++ b/codex-rs/app-server/src/user_verification_test_support.rs @@ -40,6 +40,7 @@ pub(super) struct BlockingProvider { pub(super) entered: Mutex>>, 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" diff --git a/codex-rs/app-server/tests/suite/v2/user_verification.rs b/codex-rs/app-server/tests/suite/v2/user_verification.rs index 20a54a7fd8..a388cda5f4 100644 --- a/codex-rs/app-server/tests/suite/v2/user_verification.rs +++ b/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?;