Enable user verification for the bundled TUI on supported devices (#43715)

## Why

The app server's device-support probe always returned `false`, preventing the bundled TUI from advertising user verification even on supported hardware.

## What changed

Use `native::device_supported` to enable the existing initialization path for in-process `codex-tui` clients that opt into `experimentalApi`. Eligible clients advertise the `userVerification` MCP extension capability and can receive verification requests.

## Testing

Add tests covering initialization eligibility, rejection of responses from non-owning connections, ownership release before disconnected RPC handlers finish draining, and cancellation of pending requests when authentication changes.

GitOrigin-RevId: cdacc026e8f830b3340f2e6f273851c4a3031a7e

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

@@ -48,8 +48,7 @@ impl Service {
             auth_manager,
             provider: Arc::new(native::platform_provider),
             platform_supported: native::platform_supported(),
-            // Host activation follows the bundled TUI integration.
-            device_supported: || false,
+            device_supported: native::device_supported,
             worker: Arc::new(Semaphore::new(/*permits*/ 1)),
         }
     }
@@ -257,3 +256,11 @@ mod rpc_tests;
 #[cfg(test)]
 #[path = "user_verification_test_support.rs"]
 mod test_support;
+
+#[cfg(test)]
+#[path = "user_verification_activation_tests.rs"]
+mod activation_tests;
+
+#[cfg(test)]
+#[path = "user_verification_connection_tests.rs"]
+mod connection_tests;

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

@@ -0,0 +1,99 @@
+//! Verifies that host initialization owns capability projection and response eligibility.
+
+use super::test_support::Harness;
+use crate::outgoing_message::ConnectionId;
+use crate::outgoing_message::OutgoingMessage;
+use crate::transport::ConnectionOrigin;
+use anyhow::Result;
+use codex_app_server_protocol as rpc;
+use codex_protocol::mcp::OPENAI_ELICITATION_EXTENSION_ID;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+
+#[tokio::test]
+async fn user_verification_initialize_owns_advertisement_and_eligibility() -> Result<()> {
+    for (origin, name, opt_in, supported, expected) in [
+        (ConnectionOrigin::InProcess, "codex-tui", true, true, true),
+        (ConnectionOrigin::WebSocket, "codex-tui", true, true, false),
+        (
+            ConnectionOrigin::RemoteControl,
+            "codex-tui",
+            true,
+            true,
+            false,
+        ),
+        (ConnectionOrigin::Stdio, "codex-tui", true, true, false),
+        (ConnectionOrigin::InProcess, "other-ui", true, true, false),
+        (ConnectionOrigin::InProcess, "codex-tui", false, true, false),
+        (ConnectionOrigin::InProcess, "codex-tui", true, false, false),
+    ] {
+        let probe: fn() -> bool = if supported { || true } else { || false };
+        let mut h = Harness::new(origin, probe).await?;
+        h.initialize(name, opt_in).await;
+        let projected = h
+            .session
+            .client_mcp_extensions()
+            .iter()
+            .any(|(id, settings)| {
+                id == OPENAI_ELICITATION_EXTENSION_ID && settings.get("userVerification").is_some()
+            });
+        assert_eq!(
+            projected, expected,
+            "{origin:?}/{name}/{opt_in}/{supported}"
+        );
+        let thread_id = codex_protocol::ThreadId::new();
+        let (id, response) = h
+            .outgoing
+            .send_request_to_connections(
+                Some(&[ConnectionId(1)]),
+                rpc::ServerRequestPayload::McpServerElicitationRequest(
+                    rpc::McpServerElicitationRequestParams {
+                        thread_id: thread_id.to_string(),
+                        turn_id: None,
+                        server_name: "codex_apps".into(),
+                        request: rpc::McpServerElicitationRequest::UserVerification {
+                            challenge: "AQ".into(),
+                            title: "Approve".into(),
+                            description: String::new(),
+                        },
+                    },
+                ),
+                Some(thread_id),
+            )
+            .await;
+        if expected {
+            assert!(matches!(h.response().await, OutgoingMessage::Request(_)));
+            let proof =
+                json!({"action":"accept", "content":{"credentialId":"AQ", "signature":"Ag"}});
+            h.processor
+                .process_response(
+                    ConnectionId(2),
+                    rpc::JSONRPCResponse {
+                        id: id.clone(),
+                        result: proof.clone(),
+                    },
+                )
+                .await;
+            let mut response = std::pin::pin!(response);
+            assert!(matches!(
+                futures::poll!(&mut response),
+                std::task::Poll::Pending
+            ));
+            h.processor
+                .process_response(
+                    ConnectionId(1),
+                    rpc::JSONRPCResponse {
+                        id,
+                        result: proof.clone(),
+                    },
+                )
+                .await;
+            assert_eq!(response.await?, Ok(proof));
+        } else {
+            assert!(response.await.is_err());
+            assert!(h.messages.try_recv().is_err());
+        }
+        h.shutdown().await;
+    }
+    Ok(())
+}

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

@@ -12,7 +12,6 @@ pub(super) struct Identity {
 
 impl OutgoingMessageSender {
     /// Trusted host activation enables a UI integration after its signing path is available.
-    #[allow(dead_code)] // Called by the subsequent host activation change.
     pub(crate) async fn enable_user_verification_connection(&self, connection_id: ConnectionId) {
         self.verification_connections
             .lock()

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

@@ -0,0 +1,180 @@
+//! Exercises ownership through real initialized JSON-RPC sessions and disconnect processing.
+
+use super::test_support::Harness;
+use super::test_support::write_auth;
+use crate::message_processor::ConnectionSessionState;
+use crate::outgoing_message::ConnectionId;
+use crate::outgoing_message::OutgoingEnvelope;
+use crate::outgoing_message::OutgoingMessage;
+use crate::transport::AppServerTransport;
+use crate::transport::ConnectionOrigin;
+use anyhow::Result;
+use codex_app_server_protocol as rpc;
+use codex_protocol::ThreadId;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+use std::sync::Arc;
+use tokio::sync::oneshot;
+use tokio::time::Duration;
+use tokio::time::timeout;
+
+fn request() -> rpc::ServerRequestPayload {
+    rpc::ServerRequestPayload::McpServerElicitationRequest(rpc::McpServerElicitationRequestParams {
+        thread_id: ThreadId::new().to_string(),
+        turn_id: None,
+        server_name: "codex_apps".into(),
+        request: rpc::McpServerElicitationRequest::UserVerification {
+            title: "Approve".into(),
+            description: String::new(),
+            challenge: "AQ".into(),
+        },
+    })
+}
+
+async fn initialize_second(h: &mut Harness) -> Result<Arc<ConnectionSessionState>> {
+    let session = Arc::new(ConnectionSessionState::new(ConnectionOrigin::InProcess));
+    h.processor
+        .process_request(
+            ConnectionId(2),
+            rpc::JSONRPCRequest {
+                id: rpc::RequestId::Integer(0),
+                method: "initialize".into(),
+                params: Some(json!({"clientInfo":{"name":"codex-tui","version":"1"},"capabilities":{"experimentalApi":true}})),
+                trace: None,
+            },
+            &AppServerTransport::Stdio,
+            Arc::clone(&session),
+        )
+        .await;
+    let message = timeout(Duration::from_secs(/*secs*/ 5), h.messages.recv()).await?;
+    assert!(matches!(
+        message,
+        Some(OutgoingEnvelope::ToConnection {
+            connection_id: ConnectionId(2),
+            message: OutgoingMessage::Response(_),
+            ..
+        })
+    ));
+    Ok(session)
+}
+
+#[tokio::test]
+async fn user_verification_disconnect_releases_ownership_before_rpc_drain() -> Result<()> {
+    let mut h = Harness::new(ConnectionOrigin::InProcess, || true).await?;
+    h.initialize("codex-tui", /*opt_in*/ true).await;
+    let second = initialize_second(&mut h).await?;
+    let connections = [ConnectionId(1), ConnectionId(2)];
+    let (id, pending) = h
+        .outgoing
+        .send_request_to_connections(Some(&connections), request(), /*thread_id*/ None)
+        .await;
+    assert!(matches!(h.response().await, OutgoingMessage::Request(_)));
+    let proof = json!({"action":"accept","content":{"credentialId":"AQ","signature":"Ag"}});
+    h.processor
+        .process_response(
+            ConnectionId(2),
+            rpc::JSONRPCResponse {
+                id: id.clone(),
+                result: proof.clone(),
+            },
+        )
+        .await;
+    let mut pending = std::pin::pin!(pending);
+    assert!(matches!(
+        futures::poll!(&mut pending),
+        std::task::Poll::Pending
+    ));
+
+    // Keep an unrelated initialized handler active so production disconnect must wait at the gate.
+    let gate = Arc::clone(&h.session.rpc_gate);
+    let (entered, waiting) = oneshot::channel();
+    let (release, blocked) = oneshot::channel();
+    let running = tokio::spawn(async move {
+        gate.run(async move {
+            entered.send(()).expect("handler entry");
+            blocked.await.expect("release handler");
+        })
+        .await;
+    });
+    waiting.await?;
+    let processor = Arc::clone(&h.processor);
+    let session = Arc::clone(&h.session);
+    let mut disconnect = std::pin::pin!(processor.connection_closed(ConnectionId(1), &session));
+    assert!(matches!(
+        futures::poll!(&mut disconnect),
+        std::task::Poll::Pending
+    ));
+    assert!(
+        timeout(Duration::from_secs(/*secs*/ 5), pending)
+            .await?
+            .is_err()
+    );
+
+    // The live initialized UI must now own the request, although the disconnected RPC is draining.
+    let (id, response) = h
+        .outgoing
+        .send_request_to_connections(Some(&connections), request(), /*thread_id*/ None)
+        .await;
+    let message = timeout(Duration::from_secs(/*secs*/ 5), h.messages.recv()).await?;
+    assert!(matches!(
+        message,
+        Some(OutgoingEnvelope::ToConnection {
+            connection_id: ConnectionId(2),
+            message: OutgoingMessage::Request(_),
+            ..
+        })
+    ));
+    h.processor
+        .process_response(
+            ConnectionId(1),
+            rpc::JSONRPCResponse {
+                id: id.clone(),
+                result: proof.clone(),
+            },
+        )
+        .await;
+    let mut response = std::pin::pin!(response);
+    assert!(matches!(
+        futures::poll!(&mut response),
+        std::task::Poll::Pending
+    ));
+    h.processor
+        .process_response(
+            ConnectionId(2),
+            rpc::JSONRPCResponse {
+                id,
+                result: proof.clone(),
+            },
+        )
+        .await;
+    assert_eq!(response.await?, Ok(proof));
+    release.send(()).expect("release draining handler");
+    running.await?;
+    disconnect.await;
+    h.processor
+        .connection_closed(ConnectionId(2), &second)
+        .await;
+    h.shutdown().await;
+    Ok(())
+}
+
+#[tokio::test]
+async fn user_verification_dispatcher_auth_watcher_cancels_pending_ownership() -> Result<()> {
+    let mut h = Harness::new(ConnectionOrigin::InProcess, || true).await?;
+    h.initialize("codex-tui", /*opt_in*/ true).await;
+    let (_, pending) = h
+        .outgoing
+        .send_request_to_connections(Some(&[ConnectionId(1)]), request(), /*thread_id*/ None)
+        .await;
+    assert!(matches!(h.response().await, OutgoingMessage::Request(_)));
+    write_auth(h.home.path(), "second")?;
+    h.auth.reload().await;
+    // No response triggers a revision recheck: MessageProcessor's installed watcher must cancel it.
+    assert!(
+        timeout(Duration::from_secs(/*secs*/ 5), pending)
+            .await?
+            .is_err()
+    );
+    h.shutdown().await;
+    Ok(())
+}