address client response review feedback

codex-rs/analytics/src/client.rs

@@ -429,3 +429,7 @@ async fn send_track_events(
         }
     }
 }
+
+#[cfg(test)]
+#[path = "client_tests.rs"]
+mod tests;

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<AnalyticsFact>) {
+    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)));
+}

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<ClientResponse>),
-                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;

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(())
+}