From 42cb76b669a46fa7d649959cca79d660bd7d480e Mon Sep 17 00:00:00 2001 From: Roy Han Date: Tue, 28 Apr 2026 10:50:44 -0700 Subject: [PATCH] address client response review feedback --- codex-rs/analytics/src/client.rs | 4 + codex-rs/analytics/src/client_tests.rs | 92 +++++++++++++++++++ .../src/protocol/common.rs | 68 +------------- .../src/protocol/common_tests.rs | 44 +++++++++ 4 files changed, 144 insertions(+), 64 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 71a9a6a69f..e53552e1e9 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -429,3 +429,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 new file mode 100644 index 0000000000..c88a15014e --- /dev/null +++ b/codex-rs/analytics/src/client_tests.rs @@ -0,0 +1,92 @@ +use super::AnalyticsEventsClient; +use super::AnalyticsEventsQueue; +use crate::facts::AnalyticsFact; +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::ClientResponsePayload; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadArchiveParams; +use codex_app_server_protocol::ThreadArchiveResponse; +use codex_app_server_protocol::TurnSteerParams; +use codex_app_server_protocol::TurnSteerResponse; +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_steer_request() -> ClientRequest { + ClientRequest::TurnSteer { + request_id: RequestId::Integer(1), + 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(2), + params: ThreadArchiveParams { + thread_id: "thread-1".to_string(), + }, + } +} + +#[test] +fn track_request_only_enqueues_analytics_relevant_requests() { + let (client, mut receiver) = client_with_receiver(); + + client.track_request( + /*connection_id*/ 7, + RequestId::Integer(1), + &sample_turn_steer_request(), + ); + assert!(matches!( + receiver.try_recv(), + Ok(AnalyticsFact::ClientRequest { .. }) + )); + + client.track_request( + /*connection_id*/ 7, + RequestId::Integer(2), + &sample_thread_archive_request(), + ); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); +} + +#[test] +fn track_response_only_enqueues_analytics_relevant_responses() { + let (client, mut receiver) = client_with_receiver(); + + client.track_response( + /*connection_id*/ 7, + RequestId::Integer(1), + ClientResponsePayload::TurnSteer(TurnSteerResponse { + turn_id: "turn-1".to_string(), + }), + ); + assert!(matches!( + receiver.try_recv(), + Ok(AnalyticsFact::ClientResponse { .. }) + )); + + client.track_response( + /*connection_id*/ 7, + RequestId::Integer(2), + ClientResponsePayload::ThreadArchive(ThreadArchiveResponse {}), + ); + 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 069c66a75b..586eec4521 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -214,30 +214,6 @@ macro_rules! client_request_definitions { } } - pub fn into_jsonrpc_parts_and_client_response( - 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)?; - let client_response = ClientResponse::$variant { - request_id: request_id.clone(), - response, - }; - Ok((request_id, result, Some(client_response))) - } - )* - Self::InterruptConversation(response) => { - serde_json::to_value(response).map(|result| (request_id, result, None)) - } - } - } - pub fn into_jsonrpc_parts( self, request_id: RequestId, @@ -1252,7 +1228,6 @@ mod tests { use codex_protocol::protocol::RealtimeConversationVersion; use codex_protocol::protocol::RealtimeOutputModality; use codex_protocol::protocol::RealtimeVoice; - use codex_protocol::protocol::TurnAbortReason; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; @@ -1675,45 +1650,6 @@ mod tests { Ok(()) } - #[test] - fn client_response_payload_returns_jsonrpc_parts_and_client_response() -> Result<()> { - let (request_id, result, client_response) = - ClientResponsePayload::ThreadArchive(v2::ThreadArchiveResponse {}) - .into_jsonrpc_parts_and_client_response(RequestId::Integer(7))?; - - assert_eq!(request_id, RequestId::Integer(7)); - assert_eq!(result, json!({})); - - let Some(ClientResponse::ThreadArchive { - request_id, - response: _, - }) = client_response - 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, client_response) = - ClientResponsePayload::InterruptConversation(v1::InterruptConversationResponse { - abort_reason: TurnAbortReason::Interrupted, - }) - .into_jsonrpc_parts_and_client_response(RequestId::Integer(8))?; - - assert_eq!(request_id, RequestId::Integer(8)); - assert_eq!( - result, - json!({ - "abortReason": "interrupted", - }) - ); - assert!(client_response.is_none()); - Ok(()) - } - #[test] fn serialize_config_requirements_read() -> Result<()> { let request = ClientRequest::ConfigRequirementsRead { @@ -2357,3 +2293,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(()) +}