From b15074d0a42e87465a39f78f033d8e3b74db6011 Mon Sep 17 00:00:00 2001 From: sayan-oai Date: Wed, 29 Apr 2026 13:47:20 -0700 Subject: [PATCH 01/25] app-server: fix outgoing sender test setup (#20258) ## Why [#17088](https://github.com/openai/codex/pull/17088) changed `OutgoingMessageSender::new` to require an `AnalyticsEventsClient`, but one `command_exec` test added earlier on `main` still called the old one-argument constructor. That leaves current `main` failing to compile in Bazel and argument-comment-lint jobs. ## What changed - Pass `AnalyticsEventsClient::disabled()` to the missed `OutgoingMessageSender::new` test call site in `command_exec.rs`. ## Verification - `cargo test -p codex-app-server timeout_or_cancellation_reports_cancellation_without_timeout_exit_code` --- codex-rs/app-server/src/command_exec.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/codex-rs/app-server/src/command_exec.rs b/codex-rs/app-server/src/command_exec.rs index 22ac92686e..699556dd5b 100644 --- a/codex-rs/app-server/src/command_exec.rs +++ b/codex-rs/app-server/src/command_exec.rs @@ -890,7 +890,10 @@ mod tests { manager .start(StartCommandExecParams { - outgoing: Arc::new(OutgoingMessageSender::new(tx)), + outgoing: Arc::new(OutgoingMessageSender::new( + tx, + codex_analytics::AnalyticsEventsClient::disabled(), + )), request_id: request_id.clone(), process_id: Some("proc-101".to_string()), exec_request: ExecRequest::new( From 973c5c823ed1b8ae76ab5a677c5864f624a2fafb Mon Sep 17 00:00:00 2001 From: rhan-oai Date: Wed, 29 Apr 2026 13:50:47 -0700 Subject: [PATCH 02/25] [app-server] type client response payloads (#20050) ## Why `pr17088` adds typed server-originated request/response plumbing, but successful client responses are still erased into bare JSON-RPC `result` values before app-server can make any typed decision about them. This precursor PR keeps successful client responses typed until the outgoing response seam. It is intentionally limited to protocol/app-server plumbing so the analytics behavior change can review separately on top. ## What changed - Add `ClientResponsePayload` as the pre-serialization client response body type. - Route app-server successful response paths through the typed payload seam while preserving existing handler-local analytics behavior. - Keep `InterruptConversation` JSON-RPC-only because it has no `ClientResponse` variant. - Move the new payload conversion tests into a dedicated protocol test module. ## Verification - `cargo check -p codex-app-server` - `cargo test -p codex-app-server-protocol` --- codex-rs/analytics/src/client.rs | 19 ++- codex-rs/analytics/src/client_tests.rs | 79 +++++++++++++ .../src/protocol/common.rs | 110 ++++++++++++++++++ .../src/protocol/common_tests.rs | 44 +++++++ .../app-server/src/codex_message_processor.rs | 11 +- codex-rs/app-server/src/message_processor.rs | 45 ++++--- codex-rs/app-server/src/outgoing_message.rs | 57 ++++++--- 7 files changed, 323 insertions(+), 42 deletions(-) create mode 100644 codex-rs/analytics/src/client_tests.rs create mode 100644 codex-rs/app-server-protocol/src/protocol/common_tests.rs diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index b149614353..2ff4c794ed 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -186,11 +186,22 @@ impl AnalyticsEventsClient { ))); } - pub fn track_request(&self, connection_id: u64, request_id: RequestId, request: ClientRequest) { + pub fn track_request( + &self, + connection_id: u64, + request_id: RequestId, + request: &ClientRequest, + ) { + if !matches!( + request, + ClientRequest::TurnStart { .. } | ClientRequest::TurnSteer { .. } + ) { + return; + } self.record_fact(AnalyticsFact::ClientRequest { connection_id, request_id, - request: Box::new(request), + request: Box::new(request.clone()), }); } @@ -324,6 +335,10 @@ impl AnalyticsEventsClient { } } +#[cfg(test)] +#[path = "client_tests.rs"] +mod tests; + async fn send_track_events( auth_manager: &AuthManager, base_url: &str, diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs new file mode 100644 index 0000000000..57679f24a1 --- /dev/null +++ b/codex-rs/analytics/src/client_tests.rs @@ -0,0 +1,79 @@ +use super::AnalyticsEventsClient; +use super::AnalyticsEventsQueue; +use crate::facts::AnalyticsFact; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnSteerParams; +use std::collections::HashSet; +use std::sync::Arc; +use std::sync::Mutex; +use tokio::sync::mpsc; +use tokio::sync::mpsc::error::TryRecvError; + +fn client_with_receiver() -> (AnalyticsEventsClient, mpsc::Receiver) { + let (sender, receiver) = mpsc::channel(4); + let queue = AnalyticsEventsQueue { + sender, + app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), + plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), + }; + (AnalyticsEventsClient { queue: Some(queue) }, receiver) +} + +fn sample_turn_start_request() -> ClientRequest { + ClientRequest::TurnStart { + request_id: RequestId::Integer(1), + params: TurnStartParams { + thread_id: "thread-1".to_string(), + input: Vec::new(), + ..Default::default() + }, + } +} + +fn sample_turn_steer_request() -> ClientRequest { + ClientRequest::TurnSteer { + request_id: RequestId::Integer(2), + params: TurnSteerParams { + thread_id: "thread-1".to_string(), + expected_turn_id: "turn-1".to_string(), + input: Vec::new(), + responsesapi_client_metadata: None, + }, + } +} + +fn sample_thread_archive_request() -> ClientRequest { + ClientRequest::ThreadArchive { + request_id: RequestId::Integer(3), + params: ThreadArchiveParams { + thread_id: "thread-1".to_string(), + }, + } +} + +#[test] +fn track_request_only_enqueues_analytics_relevant_requests() { + let (client, mut receiver) = client_with_receiver(); + + for (request_id, request) in [ + (RequestId::Integer(1), sample_turn_start_request()), + (RequestId::Integer(2), sample_turn_steer_request()), + ] { + client.track_request(/*connection_id*/ 7, request_id, &request); + assert!(matches!( + receiver.try_recv(), + Ok(AnalyticsFact::ClientRequest { .. }) + )); + } + + let ignored_request = sample_thread_archive_request(); + client.track_request( + /*connection_id*/ 7, + RequestId::Integer(3), + &ignored_request, + ); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); +} diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index e89af03d85..94659ee738 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -157,6 +157,7 @@ macro_rules! client_request_definitions { params: $(#[$params_meta:meta])* $params:ty, $(inspect_params: $inspect_params:tt,)? serialization: $serialization:ident $( ( $($serialization_args:tt)* ) )?, + $(manual_payload_conversion: $manual_payload_conversion:ident,)? response: $response:ty, } ),* $(,)? @@ -243,8 +244,100 @@ macro_rules! client_request_definitions { }) .unwrap_or_else(|| "".to_string()) } + + pub fn into_jsonrpc_parts( + self, + ) -> std::result::Result<(RequestId, crate::Result), serde_json::Error> { + match self { + $( + Self::$variant { request_id, response } => { + serde_json::to_value(response).map(|result| (request_id, result)) + } + )* + } + } } + #[derive(Debug, Clone)] + #[allow(clippy::large_enum_variant)] + pub enum ClientResponsePayload { + $( $variant($response), )* + InterruptConversation(v1::InterruptConversationResponse), + } + + impl ClientResponsePayload { + pub fn into_jsonrpc_parts_and_payload( + self, + request_id: RequestId, + ) -> std::result::Result< + (RequestId, crate::Result, Option), + serde_json::Error, + > { + match self { + $( + Self::$variant(response) => { + let result = serde_json::to_value(&response)?; + Ok((request_id, result, Some(Self::$variant(response)))) + } + )* + Self::InterruptConversation(response) => { + serde_json::to_value(response).map(|result| (request_id, result, None)) + } + } + } + + pub fn into_client_response(self, request_id: RequestId) -> Option { + match self { + $( + Self::$variant(response) => { + Some(ClientResponse::$variant { + request_id, + response, + }) + } + )* + Self::InterruptConversation(_) => None, + } + } + + pub fn into_jsonrpc_parts( + self, + request_id: RequestId, + ) -> std::result::Result<(RequestId, crate::Result), serde_json::Error> { + self.to_jsonrpc_parts(request_id) + } + + pub fn to_jsonrpc_parts( + &self, + request_id: RequestId, + ) -> std::result::Result<(RequestId, crate::Result), serde_json::Error> { + match self { + $( + Self::$variant(response) => { + serde_json::to_value(response).map(|result| (request_id, result)) + } + )* + Self::InterruptConversation(response) => { + serde_json::to_value(response).map(|result| (request_id, result)) + } + } + } + } + + impl From for ClientResponsePayload { + fn from(response: v1::InterruptConversationResponse) -> Self { + Self::InterruptConversation(response) + } + } + + $( + client_response_payload_from_impl!( + $variant, + $response + $(, $manual_payload_conversion)? + ); + )* + impl crate::experimental_api::ExperimentalApi for ClientRequest { fn experimental_reason(&self) -> Option<&'static str> { match self { @@ -317,6 +410,17 @@ macro_rules! client_request_definitions { }; } +macro_rules! client_response_payload_from_impl { + ($variant:ident, $response:ty) => { + impl From<$response> for ClientResponsePayload { + fn from(response: $response) -> Self { + Self::$variant(response) + } + } + }; + ($variant:ident, $response:ty, manual) => {}; +} + client_request_definitions! { Initialize { params: v1::InitializeParams, @@ -789,11 +893,13 @@ client_request_definitions! { ConfigValueWrite => "config/value/write" { params: v2::ConfigValueWriteParams, serialization: global("config"), + manual_payload_conversion: manual, response: v2::ConfigWriteResponse, }, ConfigBatchWrite => "config/batchWrite" { params: v2::ConfigBatchWriteParams, serialization: global("config"), + manual_payload_conversion: manual, response: v2::ConfigWriteResponse, }, @@ -2766,3 +2872,7 @@ mod tests { ); } } + +#[cfg(test)] +#[path = "common_tests.rs"] +mod common_tests; diff --git a/codex-rs/app-server-protocol/src/protocol/common_tests.rs b/codex-rs/app-server-protocol/src/protocol/common_tests.rs new file mode 100644 index 0000000000..83e5d37117 --- /dev/null +++ b/codex-rs/app-server-protocol/src/protocol/common_tests.rs @@ -0,0 +1,44 @@ +use super::*; +use anyhow::Result; +use codex_protocol::protocol::TurnAbortReason; +use pretty_assertions::assert_eq; +use serde_json::json; + +#[test] +fn client_response_payload_returns_jsonrpc_parts_and_client_response() -> Result<()> { + let (request_id, result, payload) = + ClientResponsePayload::ThreadArchive(v2::ThreadArchiveResponse {}) + .into_jsonrpc_parts_and_payload(RequestId::Integer(7))?; + + assert_eq!(request_id, RequestId::Integer(7)); + assert_eq!(result, json!({})); + + let Some(ClientResponse::ThreadArchive { + request_id, + response: _, + }) = payload.and_then(|payload| payload.into_client_response(RequestId::Integer(7))) + else { + panic!("expected thread/archive client response"); + }; + assert_eq!(request_id, RequestId::Integer(7)); + Ok(()) +} + +#[test] +fn interrupt_conversation_payload_stays_jsonrpc_only() -> Result<()> { + let (request_id, result, payload) = + ClientResponsePayload::InterruptConversation(v1::InterruptConversationResponse { + abort_reason: TurnAbortReason::Interrupted, + }) + .into_jsonrpc_parts_and_payload(RequestId::Integer(8))?; + + assert_eq!(request_id, RequestId::Integer(8)); + assert_eq!( + result, + json!({ + "abortReason": "interrupted", + }) + ); + assert!(payload.is_none()); + Ok(()) +} diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index e53b5e2c20..d250959d8d 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -43,6 +43,7 @@ use codex_app_server_protocol::CancelLoginAccountResponse; use codex_app_server_protocol::CancelLoginAccountStatus; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::CodexErrorInfo; use codex_app_server_protocol::CollaborationModeListParams; use codex_app_server_protocol::CollaborationModeListResponse; @@ -2118,7 +2119,7 @@ impl CodexMessageProcessor { let result = self .exec_one_off_command_inner(request_id.clone(), params) .await - .map(|()| None::); + .map(|()| None::); self.send_optional_result(request_id, result).await; } @@ -2864,7 +2865,6 @@ impl CodexMessageProcessor { response: response.clone(), }, ); - listener_task_context .outgoing .send_response(request_id, response) @@ -3544,7 +3544,7 @@ impl CodexMessageProcessor { let result = self .thread_rollback_start(&request_id, params) .await - .map(|()| None::); + .map(|()| None::); self.send_optional_result(request_id, result).await; } @@ -4401,6 +4401,7 @@ impl CodexMessageProcessor { permission_profile, reasoning_effort: session_configured.reasoning_effort, }; + self.analytics_events_client.track_response( request_id.connection_id.0, ClientResponse::ThreadResume { @@ -4408,7 +4409,6 @@ impl CodexMessageProcessor { response: response.clone(), }, ); - let connection_id = request_id.connection_id; let token_usage_thread = include_turns.then(|| response.thread.clone()); self.outgoing.send_response(request_id, response).await; @@ -5027,7 +5027,6 @@ impl CodexMessageProcessor { response: response.clone(), }, ); - let connection_id = request_id.connection_id; let token_usage_thread = include_turns.then(|| response.thread.clone()); self.outgoing.send_response(request_id, response).await; @@ -5811,7 +5810,7 @@ impl CodexMessageProcessor { request_id: ConnectionRequestId, result: Result, JSONRPCErrorError>, ) where - T: serde::Serialize, + T: Into, { match result { Ok(Some(response)) => self.outgoing.send_response(request_id, response).await, diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 56d5ff183a..1a28f8e278 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -37,6 +37,7 @@ use codex_app_server_protocol::ChatgptAuthTokensRefreshResponse; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientNotification; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::ConfigBatchWriteParams; use codex_app_server_protocol::ConfigValueWriteParams; use codex_app_server_protocol::ConfigWarningNotification; @@ -732,15 +733,11 @@ impl MessageProcessor { return Err(invalid_request(experimental_required_message(reason))); } let connection_id = connection_request_id.connection_id; - if let ClientRequest::TurnStart { request_id, .. } - | ClientRequest::TurnSteer { request_id, .. } = &codex_request - { - self.analytics_events_client.track_request( - connection_id.0, - request_id.clone(), - codex_request.clone(), - ); - } + self.analytics_events_client.track_request( + connection_id.0, + connection_request_id.request_id.clone(), + &codex_request, + ); let serialization_scope = codex_request.serialization_scope(); let app_server_client_name = session.app_server_client_name().map(str::to_string); @@ -992,7 +989,12 @@ impl MessageProcessor { params: ConfigValueWriteParams, ) { let result = self.config_api.write_value(params).await; - self.handle_config_mutation_result(request_id, result).await + self.handle_config_mutation_result( + request_id, + result, + ClientResponsePayload::ConfigValueWrite, + ) + .await } async fn handle_config_batch_write( @@ -1001,7 +1003,12 @@ impl MessageProcessor { params: ConfigBatchWriteParams, ) { let result = self.config_api.batch_write(params).await; - self.handle_config_mutation_result(request_id, result).await; + self.handle_config_mutation_result( + request_id, + result, + ClientResponsePayload::ConfigBatchWrite, + ) + .await; } async fn handle_experimental_feature_enablement_set( @@ -1015,7 +1022,12 @@ impl MessageProcessor { .set_experimental_feature_enablement(params) .await; let is_ok = result.is_ok(); - self.handle_config_mutation_result(request_id, result).await; + self.handle_config_mutation_result( + request_id, + result, + ClientResponsePayload::ExperimentalFeatureEnablementSet, + ) + .await; if should_refresh_apps_list && is_ok { self.refresh_apps_list_after_experimental_feature_enablement_set() .await; @@ -1091,15 +1103,18 @@ impl MessageProcessor { }); } - async fn handle_config_mutation_result( + async fn handle_config_mutation_result( &self, request_id: ConnectionRequestId, result: std::result::Result, + wrap_success: impl FnOnce(T) -> ClientResponsePayload, ) { match result { Ok(response) => { self.handle_config_mutation().await; - self.outgoing.send_response(request_id, response).await; + self.outgoing + .send_response_as(request_id, wrap_success(response)) + .await; } Err(error) => self.outgoing.send_error(request_id, error).await, } @@ -1177,7 +1192,7 @@ impl MessageProcessor { device_key_requests_allowed: bool, run_request: F, ) where - R: serde::Serialize + Send + 'static, + R: Into + Send + 'static, F: FnOnce(DeviceKeyApi) -> Fut + Send + 'static, Fut: Future> + Send + 'static, { diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 300a7044e5..588b4b53eb 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -5,6 +5,7 @@ use std::sync::atomic::AtomicI64; use std::sync::atomic::Ordering; use codex_analytics::AnalyticsEventsClient; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::Result; @@ -188,11 +189,10 @@ impl ThreadScopedOutgoingMessageSender { .await } - pub(crate) async fn send_response( - &self, - request_id: ConnectionRequestId, - response: T, - ) { + pub(crate) async fn send_response(&self, request_id: ConnectionRequestId, response: T) + where + T: Into, + { self.outgoing.send_response(request_id, response).await; } @@ -482,21 +482,28 @@ impl OutgoingMessageSender { } } - pub(crate) async fn send_response( + pub(crate) async fn send_response(&self, request_id: ConnectionRequestId, response: T) + where + T: Into, + { + self.send_response_as(request_id, response.into()).await; + } + + pub(crate) async fn send_response_as( &self, request_id: ConnectionRequestId, - response: T, + response: ClientResponsePayload, ) { + let connection_id = request_id.connection_id; + let serialized_response = response.into_jsonrpc_parts(request_id.request_id.clone()); let request_context = self.take_request_context(&request_id).await; - match serde_json::to_value(response) { - Ok(result) => { - let outgoing_message = OutgoingMessage::Response(OutgoingResponse { - id: request_id.request_id.clone(), - result, - }); + + match serialized_response { + Ok((id, result)) => { + let outgoing_message = OutgoingMessage::Response(OutgoingResponse { id, result }); self.send_outgoing_message_to_connection( request_context, - request_id.connection_id, + connection_id, outgoing_message, "response", ) @@ -592,11 +599,13 @@ impl OutgoingMessageSender { request_id: ConnectionRequestId, result: std::result::Result, ) where - T: Serialize, + T: Into, E: Into, { match result { - Ok(response) => self.send_response(request_id, response).await, + Ok(response) => { + self.send_response(request_id, response).await; + } Err(error) => self.send_error(request_id, error).await, } } @@ -966,7 +975,12 @@ mod tests { }; outgoing - .send_response(request_id.clone(), json!({ "ok": true })) + .send_response( + request_id.clone(), + ClientResponsePayload::ThreadArchive( + codex_app_server_protocol::ThreadArchiveResponse {}, + ), + ) .await; let envelope = timeout(Duration::from_secs(1), rx.recv()) @@ -985,7 +999,7 @@ mod tests { panic!("expected response message"); }; assert_eq!(response.id, request_id.request_id); - assert_eq!(response.result, json!({ "ok": true })); + assert_eq!(response.result, json!({})); } other => panic!("expected targeted response envelope, got: {other:?}"), } @@ -1011,7 +1025,12 @@ mod tests { assert_eq!(outgoing.request_context_count().await, 1); outgoing - .send_response(request_id, json!({ "ok": true })) + .send_response( + request_id, + ClientResponsePayload::ThreadArchive( + codex_app_server_protocol::ThreadArchiveResponse {}, + ), + ) .await; assert_eq!(outgoing.request_context_count().await, 0); From afbddabc8bee29e72f19e866c46db9b9f7a2f836 Mon Sep 17 00:00:00 2001 From: xli-oai Date: Wed, 29 Apr 2026 14:01:11 -0700 Subject: [PATCH 03/25] Require remote plugin detail before uninstall (#19966) ## Summary - Fetch remote plugin detail before sending the uninstall request. - Use the detail response to derive the marketplace namespace and plugin name for cache cleanup. - Stop the uninstall before the backend POST if detail lookup fails, so backend state and local cache state do not diverge. ## Testing - `just fmt` - `cargo test -p codex-app-server plugin_uninstall` - `cargo test -p codex-core-plugins` - `git diff --check` --- .../tests/suite/v2/plugin_uninstall.rs | 54 ++------ codex-rs/core-plugins/src/remote.rs | 124 ++++++------------ 2 files changed, 53 insertions(+), 125 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs b/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs index e0f71f11b5..5a05816b0d 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs @@ -196,7 +196,6 @@ async fn plugin_uninstall_writes_remote_plugin_to_cloud_when_remote_plugin_enabl )?; mount_remote_plugin_detail(&server, REMOTE_PLUGIN_ID, "1.0.0", "GLOBAL").await; - mount_empty_remote_installed_plugins(&server).await; Mock::given(method("POST")) .and(path(format!( @@ -269,7 +268,6 @@ async fn plugin_uninstall_uses_detail_scope_for_cache_namespace() -> Result<()> AuthCredentialsStoreMode::File, )?; mount_remote_plugin_detail(&server, REMOTE_PLUGIN_ID, "1.0.0", "WORKSPACE").await; - mount_empty_remote_installed_plugins(&server).await; Mock::given(method("POST")) .and(path(format!( @@ -326,7 +324,7 @@ async fn plugin_uninstall_uses_detail_scope_for_cache_namespace() -> Result<()> } #[tokio::test] -async fn plugin_uninstall_posts_even_when_remote_detail_fetch_fails() -> Result<()> { +async fn plugin_uninstall_rejects_before_post_when_remote_detail_fetch_fails() -> Result<()> { let codex_home = TempDir::new()?; let server = MockServer::start().await; write_remote_plugin_catalog_config( @@ -342,19 +340,6 @@ async fn plugin_uninstall_posts_even_when_remote_detail_fetch_fails() -> Result< AuthCredentialsStoreMode::File, )?; - Mock::given(method("POST")) - .and(path(format!( - "/backend-api/plugins/{REMOTE_PLUGIN_ID}/uninstall" - ))) - .and(header("authorization", "Bearer chatgpt-token")) - .and(header("chatgpt-account-id", "account-123")) - .respond_with( - ResponseTemplate::new(200) - .set_body_string(format!(r#"{{"id":"{REMOTE_PLUGIN_ID}","enabled":false}}"#)), - ) - .mount(&server) - .await; - let legacy_remote_plugin_cache_root = codex_home .path() .join(format!("plugins/cache/chatgpt-global/{REMOTE_PLUGIN_ID}")); @@ -368,22 +353,29 @@ async fn plugin_uninstall_posts_even_when_remote_detail_fetch_fails() -> Result< plugin_id: REMOTE_PLUGIN_ID.to_string(), }) .await?; - let response: JSONRPCResponse = timeout( + let err = timeout( DEFAULT_TIMEOUT, - mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + mcp.read_stream_until_error_message(RequestId::Integer(request_id)), ) .await??; - let response: PluginUninstallResponse = to_response(response)?; - assert_eq!(response, PluginUninstallResponse {}); + assert_eq!(err.error.code, -32600); + assert!(err.error.message.contains("remote plugin catalog request")); + wait_for_remote_plugin_request_count( + &server, + "GET", + &format!("/ps/plugins/{REMOTE_PLUGIN_ID}"), + /*expected_count*/ 1, + ) + .await?; wait_for_remote_plugin_request_count( &server, "POST", &format!("/plugins/{REMOTE_PLUGIN_ID}/uninstall"), - /*expected_count*/ 1, + /*expected_count*/ 0, ) .await?; - assert!(!legacy_remote_plugin_cache_root.exists()); + assert!(legacy_remote_plugin_cache_root.exists()); Ok(()) } @@ -557,24 +549,6 @@ async fn mount_remote_plugin_detail( .await; } -async fn mount_empty_remote_installed_plugins(server: &MockServer) { - Mock::given(method("GET")) - .and(path("/backend-api/ps/plugins/installed")) - .and(header("authorization", "Bearer chatgpt-token")) - .and(header("chatgpt-account-id", "account-123")) - .respond_with(ResponseTemplate::new(200).set_body_string( - r#"{ - "plugins": [], - "pagination": { - "limit": 50, - "next_page_token": null - } -}"#, - )) - .mount(server) - .await; -} - async fn wait_for_remote_plugin_request_count( server: &MockServer, method_name: &str, diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index 032313ca1f..e6256691f5 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -13,10 +13,8 @@ use std::collections::BTreeMap; use std::collections::BTreeSet; use std::collections::HashSet; use std::fs; -use std::path::Path; use std::path::PathBuf; use std::time::Duration; -use tracing::warn; pub const REMOTE_GLOBAL_MARKETPLACE_NAME: &str = "chatgpt-global"; pub const REMOTE_WORKSPACE_MARKETPLACE_NAME: &str = "chatgpt-workspace"; @@ -476,27 +474,6 @@ async fn fetch_remote_plugin_detail_with_download_url_option( .await } -async fn fetch_remote_plugin_detail_by_id( - config: &RemotePluginServiceConfig, - auth: &CodexAuth, - plugin_id: &str, -) -> Result { - let plugin = fetch_plugin_detail( - config, auth, plugin_id, /*include_download_urls*/ false, - ) - .await?; - let scope = plugin.scope; - build_remote_plugin_detail( - config, - auth, - scope, - scope.marketplace_name().to_string(), - plugin_id, - plugin, - ) - .await -} - async fn build_remote_plugin_detail( config: &RemotePluginServiceConfig, auth: &CodexAuth, @@ -589,6 +566,12 @@ pub async fn uninstall_remote_plugin( plugin_id: &str, ) -> Result<(), RemotePluginCatalogError> { let auth = ensure_chatgpt_auth(auth)?; + let plugin = fetch_plugin_detail( + config, auth, plugin_id, /*include_download_urls*/ false, + ) + .await?; + let marketplace_name = plugin.scope.marketplace_name().to_string(); + let plugin_name = plugin.name; let base_url = config.chatgpt_base_url.trim_end_matches('/'); let url = format!("{base_url}/plugins/{plugin_id}/uninstall"); @@ -609,19 +592,9 @@ pub async fn uninstall_remote_plugin( }); } - let remote_detail = match fetch_remote_plugin_detail_by_id(config, auth, plugin_id).await { - Ok(remote_detail) => Some(remote_detail), - Err(err) => { - warn!( - plugin_id, - "failed to read remote plugin details after uninstall; skipping named cache removal: {err}" - ); - None - } - }; let legacy_plugin_id = plugin_id.to_string(); tokio::task::spawn_blocking(move || { - remove_remote_plugin_cache(codex_home, remote_detail, legacy_plugin_id) + remove_remote_plugin_cache(codex_home, marketplace_name, plugin_name, legacy_plugin_id) }) .await .map_err(|err| { @@ -636,65 +609,46 @@ pub async fn uninstall_remote_plugin( fn remove_remote_plugin_cache( codex_home: PathBuf, - remote_detail: Option, + marketplace_name: String, + plugin_name: String, legacy_plugin_id: String, ) -> Result<(), String> { - if let Some(remote_detail) = remote_detail { - let marketplace_name = remote_detail.marketplace_name; - let plugin_name = remote_detail.summary.name; - let store = PluginStore::try_new(codex_home.clone()) - .map_err(|err| format!("failed to resolve remote plugin cache root: {err}"))?; - let plugin_id = PluginId::new(plugin_name.clone(), marketplace_name.clone()).map_err( - |err| { - format!( - "invalid remote plugin cache id for `{plugin_name}` in `{marketplace_name}`: {err}" - ) - }, - )?; - let plugin_cache_root = store.plugin_base_root(&plugin_id); - store.uninstall(&plugin_id).map_err(|err| { + let store = PluginStore::try_new(codex_home.clone()) + .map_err(|err| format!("failed to resolve remote plugin cache root: {err}"))?; + let plugin_id = + PluginId::new(plugin_name.clone(), marketplace_name.clone()).map_err(|err| { format!( - "failed to remove remote plugin cache entry {}: {err}", - plugin_cache_root.display() + "invalid remote plugin cache id for `{plugin_name}` in `{marketplace_name}`: {err}" ) })?; - - let legacy_remote_plugin_cache_root = codex_home - .join(PLUGINS_CACHE_DIR) - .join(marketplace_name) - .join(legacy_plugin_id); - if legacy_remote_plugin_cache_root != plugin_cache_root.as_path() { - remove_path_if_exists(&legacy_remote_plugin_cache_root)?; - } - return Ok(()); - } - - for scope in RemotePluginScope::all() { - let legacy_remote_plugin_cache_root = codex_home - .join(PLUGINS_CACHE_DIR) - .join(scope.marketplace_name()) - .join(&legacy_plugin_id); - remove_path_if_exists(&legacy_remote_plugin_cache_root)?; - } - Ok(()) -} - -fn remove_path_if_exists(path: &Path) -> Result<(), String> { - if !path.exists() { - return Ok(()); - } - - let result = if path.is_dir() { - fs::remove_dir_all(path) - } else { - fs::remove_file(path) - }; - result.map_err(|err| { + let plugin_cache_root = store.plugin_base_root(&plugin_id); + store.uninstall(&plugin_id).map_err(|err| { format!( "failed to remove remote plugin cache entry {}: {err}", - path.display() + plugin_cache_root.display() ) - }) + })?; + + let legacy_remote_plugin_cache_root = codex_home + .join(PLUGINS_CACHE_DIR) + .join(marketplace_name) + .join(legacy_plugin_id); + if legacy_remote_plugin_cache_root != plugin_cache_root.as_path() + && legacy_remote_plugin_cache_root.exists() + { + let result = if legacy_remote_plugin_cache_root.is_dir() { + fs::remove_dir_all(&legacy_remote_plugin_cache_root) + } else { + fs::remove_file(&legacy_remote_plugin_cache_root) + }; + result.map_err(|err| { + format!( + "failed to remove remote plugin cache entry {}: {err}", + legacy_remote_plugin_cache_root.display() + ) + })?; + } + Ok(()) } fn build_remote_plugin_summary( From 72a39e3a96e96fe8802b0f779ec82d8aceb5d24e Mon Sep 17 00:00:00 2001 From: rhan-oai Date: Wed, 29 Apr 2026 14:22:39 -0700 Subject: [PATCH 04/25] [app-server] centralize client response analytics (#20059) ## Why The precursor PR keeps successful client responses typed until app-server's outgoing response seam. This follow-up uses that seam to move successful client-response analytics out of individual handlers and into the shared sender path, while keeping filtering decisions inside `codex-analytics`. ## What changed - Emit successful client-response analytics centrally from `OutgoingMessageSender::send_response`. - Remove duplicate handler-local response tracking for the current thread/turn lifecycle responses. - Keep analytics ingestion selective inside `AnalyticsEventsClient`, so unrelated client traffic is ignored before cloning or boxing. - Collapse client-response analytics facts onto one typed path and normalize payloads in the reducer. - Add direct client-filter coverage plus sender-level coverage for the centralized forwarding path. ## Verification - `cargo test -p codex-analytics` - `cargo test -p codex-app-server outgoing_message::tests --lib` --- .../analytics/src/analytics_client_tests.rs | 191 ++++++++++++------ codex-rs/analytics/src/client.rs | 28 ++- codex-rs/analytics/src/client_tests.rs | 141 ++++++++++++- codex-rs/analytics/src/facts.rs | 5 +- codex-rs/analytics/src/reducer.rs | 5 +- .../app-server/src/codex_message_processor.rs | 38 ---- codex-rs/app-server/src/outgoing_message.rs | 14 +- 7 files changed, 311 insertions(+), 111 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 9f88ceeb94..8fd1ee7dc3 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -59,7 +59,7 @@ use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::CodexErrorInfo; use codex_app_server_protocol::InitializeCapabilities; use codex_app_server_protocol::InitializeParams; @@ -71,6 +71,8 @@ use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::SessionSource as AppServerSessionSource; use codex_app_server_protocol::Thread; +use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::ThreadArchiveResponse; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; @@ -141,23 +143,24 @@ fn sample_thread_with_source( } } -fn sample_thread_start_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse { - ClientResponse::ThreadStart { - request_id: RequestId::Integer(1), - response: ThreadStartResponse { - thread: sample_thread(thread_id, ephemeral), - model: model.to_string(), - model_provider: "openai".to_string(), - service_tier: None, - cwd: test_path_buf("/tmp").abs(), - instruction_sources: Vec::new(), - approval_policy: AppServerAskForApproval::OnFailure, - approvals_reviewer: AppServerApprovalsReviewer::User, - sandbox: AppServerSandboxPolicy::DangerFullAccess, - permission_profile: Some(sample_permission_profile()), - reasoning_effort: None, - }, - } +fn sample_thread_start_response( + thread_id: &str, + ephemeral: bool, + model: &str, +) -> ClientResponsePayload { + ClientResponsePayload::ThreadStart(ThreadStartResponse { + thread: sample_thread(thread_id, ephemeral), + model: model.to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) } fn sample_permission_profile() -> AppServerPermissionProfile { @@ -183,7 +186,11 @@ fn sample_runtime_metadata() -> CodexRuntimeMetadata { } } -fn sample_thread_resume_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse { +fn sample_thread_resume_response( + thread_id: &str, + ephemeral: bool, + model: &str, +) -> ClientResponsePayload { sample_thread_resume_response_with_source( thread_id, ephemeral, @@ -197,23 +204,20 @@ fn sample_thread_resume_response_with_source( ephemeral: bool, model: &str, source: AppServerSessionSource, -) -> ClientResponse { - ClientResponse::ThreadResume { - request_id: RequestId::Integer(2), - response: ThreadResumeResponse { - thread: sample_thread_with_source(thread_id, ephemeral, source), - model: model.to_string(), - model_provider: "openai".to_string(), - service_tier: None, - cwd: test_path_buf("/tmp").abs(), - instruction_sources: Vec::new(), - approval_policy: AppServerAskForApproval::OnFailure, - approvals_reviewer: AppServerApprovalsReviewer::User, - sandbox: AppServerSandboxPolicy::DangerFullAccess, - permission_profile: Some(sample_permission_profile()), - reasoning_effort: None, - }, - } +) -> ClientResponsePayload { + ClientResponsePayload::ThreadResume(ThreadResumeResponse { + thread: sample_thread_with_source(thread_id, ephemeral, source), + model: model.to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) } fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest { @@ -235,21 +239,18 @@ fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest } } -fn sample_turn_start_response(turn_id: &str, request_id: i64) -> ClientResponse { - ClientResponse::TurnStart { - request_id: RequestId::Integer(request_id), - response: codex_app_server_protocol::TurnStartResponse { - turn: Turn { - id: turn_id.to_string(), - items: vec![], - status: AppServerTurnStatus::InProgress, - error: None, - started_at: None, - completed_at: None, - duration_ms: None, - }, +fn sample_turn_start_response(turn_id: &str) -> ClientResponsePayload { + ClientResponsePayload::TurnStart(codex_app_server_protocol::TurnStartResponse { + turn: Turn { + id: turn_id.to_string(), + items: vec![], + status: AppServerTurnStatus::InProgress, + error: None, + started_at: None, + completed_at: None, + duration_ms: None, }, - } + }) } fn sample_turn_started_notification(thread_id: &str, turn_id: &str) -> ServerNotification { @@ -355,13 +356,10 @@ fn sample_turn_steer_request( } } -fn sample_turn_steer_response(turn_id: &str, request_id: i64) -> ClientResponse { - ClientResponse::TurnSteer { - request_id: RequestId::Integer(request_id), - response: TurnSteerResponse { - turn_id: turn_id.to_string(), - }, - } +fn sample_turn_steer_response(turn_id: &str) -> ClientResponsePayload { + ClientResponsePayload::TurnSteer(TurnSteerResponse { + turn_id: turn_id.to_string(), + }) } fn no_active_turn_steer_error() -> JSONRPCErrorError { @@ -488,6 +486,7 @@ async fn ingest_turn_prerequisites( .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(1), response: Box::new(sample_thread_start_response( "thread-2", /*ephemeral*/ false, "gpt-5", )), @@ -512,7 +511,8 @@ async fn ingest_turn_prerequisites( .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_start_response("turn-2", /*request_id*/ 3)), + request_id: RequestId::Integer(3), + response: Box::new(sample_turn_start_response("turn-2")), }, out, ) @@ -864,6 +864,7 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(1), response: Box::new(sample_thread_start_response( "thread-no-client", /*ephemeral*/ false, @@ -908,6 +909,7 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(2), response: Box::new(sample_thread_resume_response( "thread-1", /*ephemeral*/ true, "gpt-5", )), @@ -954,6 +956,65 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize ); } +#[tokio::test] +async fn unrelated_client_requests_are_ignored_by_reducer() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + + reducer + .ingest( + AnalyticsFact::ClientRequest { + connection_id: 7, + request_id: RequestId::Integer(3), + request: Box::new(ClientRequest::ThreadArchive { + request_id: RequestId::Integer(3), + params: ThreadArchiveParams { + thread_id: "thread-2".to_string(), + }, + }), + }, + &mut events, + ) + .await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(3), + response: Box::new(sample_turn_start_response("turn-2")), + }, + &mut events, + ) + .await; + + assert!( + events.is_empty(), + "unrelated requests must not create pending turn state" + ); +} + +#[tokio::test] +async fn unrelated_client_responses_are_ignored_by_reducer() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + + ingest_initialize(&mut reducer, &mut events).await; + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(9), + response: Box::new(ClientResponsePayload::ThreadArchive( + ThreadArchiveResponse {}, + )), + }, + &mut events, + ) + .await; + + assert!(events.is_empty()); +} + #[tokio::test] async fn compaction_event_ingests_custom_fact() { let mut reducer = AnalyticsReducer::default(); @@ -988,6 +1049,7 @@ async fn compaction_event_ingests_custom_fact() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(2), response: Box::new(sample_thread_resume_response_with_source( "thread-1", /*ephemeral*/ false, @@ -1099,6 +1161,7 @@ async fn guardian_review_event_ingests_custom_fact_with_optional_target_item() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, + request_id: RequestId::Integer(1), response: Box::new(sample_thread_start_response( "thread-guardian", /*ephemeral*/ false, @@ -1881,7 +1944,8 @@ async fn accepted_turn_steer_emits_expected_event() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 4)), + request_id: RequestId::Integer(4), + response: Box::new(sample_turn_steer_response("turn-2")), }, &mut out, ) @@ -2047,7 +2111,8 @@ async fn turn_start_error_response_discards_pending_start_request() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_start_response("turn-2", /*request_id*/ 3)), + request_id: RequestId::Integer(3), + response: Box::new(sample_turn_start_response("turn-2")), }, &mut out, ) @@ -2176,7 +2241,8 @@ async fn accepted_steers_increment_turn_steer_count() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 4)), + request_id: RequestId::Integer(4), + response: Box::new(sample_turn_steer_response("turn-2")), }, &mut out, ) @@ -2222,7 +2288,8 @@ async fn accepted_steers_increment_turn_steer_count() { .ingest( AnalyticsFact::ClientResponse { connection_id: 7, - response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 6)), + request_id: RequestId::Integer(6), + response: Box::new(sample_turn_steer_response("turn-2")), }, &mut out, ) diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 2ff4c794ed..d54c53ede9 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -22,7 +22,7 @@ use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnTokenUsageFact; use crate::reducer::AnalyticsReducer; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RequestId; @@ -295,9 +295,25 @@ impl AnalyticsEventsClient { } } - pub fn track_response(&self, connection_id: u64, response: ClientResponse) { + pub fn track_response( + &self, + connection_id: u64, + request_id: RequestId, + response: ClientResponsePayload, + ) { + if !matches!( + response, + ClientResponsePayload::ThreadStart(_) + | ClientResponsePayload::ThreadResume(_) + | ClientResponsePayload::ThreadFork(_) + | ClientResponsePayload::TurnStart(_) + | ClientResponsePayload::TurnSteer(_) + ) { + return; + } self.record_fact(AnalyticsFact::ClientResponse { connection_id, + request_id, response: Box::new(response), }); } @@ -335,10 +351,6 @@ impl AnalyticsEventsClient { } } -#[cfg(test)] -#[path = "client_tests.rs"] -mod tests; - async fn send_track_events( auth_manager: &AuthManager, base_url: &str, @@ -379,3 +391,7 @@ async fn send_track_events( } } } + +#[cfg(test)] +#[path = "client_tests.rs"] +mod tests; diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 57679f24a1..da18da2529 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -1,11 +1,30 @@ use super::AnalyticsEventsClient; use super::AnalyticsEventsQueue; use crate::facts::AnalyticsFact; +use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; +use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::ClientResponsePayload; +use codex_app_server_protocol::PermissionProfile as AppServerPermissionProfile; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy; +use codex_app_server_protocol::SessionSource as AppServerSessionSource; +use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::ThreadArchiveResponse; +use codex_app_server_protocol::ThreadForkResponse; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; +use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::TurnStatus as AppServerTurnStatus; use codex_app_server_protocol::TurnSteerParams; +use codex_app_server_protocol::TurnSteerResponse; +use codex_protocol::models::PermissionProfile as CorePermissionProfile; +use codex_utils_absolute_path::test_support::PathBufExt; +use codex_utils_absolute_path::test_support::test_path_buf; use std::collections::HashSet; use std::sync::Arc; use std::sync::Mutex; @@ -13,7 +32,7 @@ use tokio::sync::mpsc; use tokio::sync::mpsc::error::TryRecvError; fn client_with_receiver() -> (AnalyticsEventsClient, mpsc::Receiver) { - let (sender, receiver) = mpsc::channel(4); + let (sender, receiver) = mpsc::channel(8); let queue = AnalyticsEventsQueue { sender, app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), @@ -54,6 +73,100 @@ fn sample_thread_archive_request() -> ClientRequest { } } +fn sample_thread(thread_id: &str) -> Thread { + Thread { + id: thread_id.to_string(), + forked_from_id: None, + preview: "first prompt".to_string(), + ephemeral: false, + model_provider: "openai".to_string(), + created_at: 1, + updated_at: 2, + status: AppServerThreadStatus::Idle, + path: None, + cwd: test_path_buf("/tmp").abs(), + cli_version: "0.0.0".to_string(), + source: AppServerSessionSource::Exec, + agent_nickname: None, + agent_role: None, + git_info: None, + name: None, + turns: Vec::new(), + } +} + +fn sample_permission_profile() -> AppServerPermissionProfile { + CorePermissionProfile::Disabled.into() +} + +fn sample_thread_start_response() -> ClientResponsePayload { + ClientResponsePayload::ThreadStart(ThreadStartResponse { + thread: sample_thread("thread-1"), + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) +} + +fn sample_thread_resume_response() -> ClientResponsePayload { + ClientResponsePayload::ThreadResume(ThreadResumeResponse { + thread: sample_thread("thread-2"), + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) +} + +fn sample_thread_fork_response() -> ClientResponsePayload { + ClientResponsePayload::ThreadFork(ThreadForkResponse { + thread: sample_thread("thread-3"), + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + service_tier: None, + cwd: test_path_buf("/tmp").abs(), + instruction_sources: Vec::new(), + approval_policy: AppServerAskForApproval::OnFailure, + approvals_reviewer: AppServerApprovalsReviewer::User, + sandbox: AppServerSandboxPolicy::DangerFullAccess, + permission_profile: Some(sample_permission_profile()), + reasoning_effort: None, + }) +} + +fn sample_turn_start_response() -> ClientResponsePayload { + ClientResponsePayload::TurnStart(TurnStartResponse { + turn: Turn { + id: "turn-1".to_string(), + items: Vec::new(), + status: AppServerTurnStatus::InProgress, + error: None, + started_at: None, + completed_at: None, + duration_ms: None, + }, + }) +} + +fn sample_turn_steer_response() -> ClientResponsePayload { + ClientResponsePayload::TurnSteer(TurnSteerResponse { + turn_id: "turn-2".to_string(), + }) +} + #[test] fn track_request_only_enqueues_analytics_relevant_requests() { let (client, mut receiver) = client_with_receiver(); @@ -77,3 +190,29 @@ fn track_request_only_enqueues_analytics_relevant_requests() { ); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } + +#[test] +fn track_response_only_enqueues_analytics_relevant_responses() { + let (client, mut receiver) = client_with_receiver(); + + for (request_id, response) in [ + (RequestId::Integer(1), sample_thread_start_response()), + (RequestId::Integer(2), sample_thread_resume_response()), + (RequestId::Integer(3), sample_thread_fork_response()), + (RequestId::Integer(4), sample_turn_start_response()), + (RequestId::Integer(5), sample_turn_steer_response()), + ] { + client.track_response(/*connection_id*/ 7, request_id, response); + assert!(matches!( + receiver.try_recv(), + Ok(AnalyticsFact::ClientResponse { .. }) + )); + } + + client.track_response( + /*connection_id*/ 7, + RequestId::Integer(6), + ClientResponsePayload::ThreadArchive(ThreadArchiveResponse {}), + ); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); +} diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 86beb0b88f..424dd523b2 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -2,7 +2,7 @@ use crate::events::AppServerRpcTransport; use crate::events::CodexRuntimeMetadata; use crate::events::GuardianReviewEventParams; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; +use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::InitializeParams; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RequestId; @@ -281,7 +281,8 @@ pub(crate) enum AnalyticsFact { }, ClientResponse { connection_id: u64, - response: Box, + request_id: RequestId, + response: Box, }, ErrorResponse { connection_id: u64, diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 33e89df011..43da35c47c 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -181,9 +181,12 @@ impl AnalyticsReducer { } AnalyticsFact::ClientResponse { connection_id, + request_id, response, } => { - self.ingest_response(connection_id, *response, out); + if let Some(response) = response.into_client_response(request_id) { + self.ingest_response(connection_id, response, out); + } } AnalyticsFact::ErrorResponse { connection_id, diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index d250959d8d..a8b3beb871 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -42,7 +42,6 @@ use codex_app_server_protocol::CancelLoginAccountParams; use codex_app_server_protocol::CancelLoginAccountResponse; use codex_app_server_protocol::CancelLoginAccountStatus; use codex_app_server_protocol::ClientRequest; -use codex_app_server_protocol::ClientResponse; use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::CodexErrorInfo; use codex_app_server_protocol::CollaborationModeListParams; @@ -2856,15 +2855,6 @@ impl CodexMessageProcessor { match result { Ok((response, notif)) => { - listener_task_context - .analytics_events_client - .track_response( - request_id.connection_id.0, - ClientResponse::ThreadStart { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); listener_task_context .outgoing .send_response(request_id, response) @@ -4402,13 +4392,6 @@ impl CodexMessageProcessor { reasoning_effort: session_configured.reasoning_effort, }; - self.analytics_events_client.track_response( - request_id.connection_id.0, - ClientResponse::ThreadResume { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); let connection_id = request_id.connection_id; let token_usage_thread = include_turns.then(|| response.thread.clone()); self.outgoing.send_response(request_id, response).await; @@ -5020,13 +5003,6 @@ impl CodexMessageProcessor { return; } }; - self.analytics_events_client.track_response( - request_id.connection_id.0, - ClientResponse::ThreadFork { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); let connection_id = request_id.connection_id; let token_usage_thread = include_turns.then(|| response.thread.clone()); self.outgoing.send_response(request_id, response).await; @@ -6644,13 +6620,6 @@ impl CodexMessageProcessor { match result { Ok(response) => { - self.analytics_events_client.track_response( - request_id.connection_id.0, - ClientResponse::TurnStart { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); self.outgoing.send_response(request_id, response).await; } Err(error) => { @@ -6821,13 +6790,6 @@ impl CodexMessageProcessor { match result { Ok(response) => { - self.analytics_events_client.track_response( - request_id.connection_id.0, - ClientResponse::TurnSteer { - request_id: request_id.request_id.clone(), - response: response.clone(), - }, - ); self.outgoing.send_response(request_id, response).await; } Err(error) => { diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 588b4b53eb..34441f83a0 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -495,7 +495,19 @@ impl OutgoingMessageSender { response: ClientResponsePayload, ) { let connection_id = request_id.connection_id; - let serialized_response = response.into_jsonrpc_parts(request_id.request_id.clone()); + let request_id_for_analytics = request_id.request_id.clone(); + let serialized_response = response + .into_jsonrpc_parts_and_payload(request_id.request_id.clone()) + .map(|(id, result, response)| { + if let Some(response) = response { + self.analytics_events_client.track_response( + connection_id.0, + request_id_for_analytics, + response, + ); + } + (id, result) + }); let request_context = self.take_request_context(&request_id).await; match serialized_response { From 8d5da3ffe58cc13380226a253a7e22a3a2a2abce Mon Sep 17 00:00:00 2001 From: xli-oai Date: Wed, 29 Apr 2026 14:45:27 -0700 Subject: [PATCH 05/25] Fallback login callback port when default is busy (#19334) ## Summary - Keep the preferred ChatGPT login callback port `1455` first. - Preserve the existing `/cancel` recovery for stale Codex login servers. - Fall back to the registered localhost callback port `1457` when `1455` remains unavailable. ## Why Cursor and Codex Desktop both use the ChatGPT account login callback server. On Windows, Cursor can already be listening on `127.0.0.1:1455` / `[::1]:1455`, causing Codex Desktop sign-in to fail with: `Local callback port 1455 is already in use on this machine.` Codex already attempted to cancel a stale Codex login server on that port, but if the listener does not release the port, the old behavior was to fail. The new behavior falls back to `1457`, which matches the fixed redirect URI being registered server-side in `openai/openai#863817`. This keeps the OAuth `redirect_uri` inside Hydra's exact allow-list instead of choosing an arbitrary ephemeral port. ## Validation - `just fmt` - `cargo test -p codex-login` - `git diff --check HEAD~1..HEAD` --- codex-rs/login/src/server.rs | 25 +++++-- .../login/tests/suite/login_server_e2e.rs | 70 +++++++++++++++++++ 2 files changed, 91 insertions(+), 4 deletions(-) diff --git a/codex-rs/login/src/server.rs b/codex-rs/login/src/server.rs index 0c7b810184..f6987d46d9 100644 --- a/codex-rs/login/src/server.rs +++ b/codex-rs/login/src/server.rs @@ -50,6 +50,8 @@ use tracing::warn; const DEFAULT_ISSUER: &str = "https://auth.openai.com"; const DEFAULT_PORT: u16 = 1455; +// Keep in sync with the Codex CLI Hydra redirect URI allow-list. +const FALLBACK_PORT: u16 = 1457; static LOGIN_ERROR_PAGE_TEMPLATE: LazyLock