openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-09 15:58:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

[codex-analytics] ingest server requests and responses

Browse Source
This commit is contained in:
rhan-oai
2026-04-08 12:37:37 -07:00
parent f480b98984
commit fef43c065e
10 changed files with 164 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
codex-rs/analytics/src/analytics_client_tests.rs
View File

@@ -353,7 +353,7 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize
reducer
.ingest(
AnalyticsFact::Response {
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_thread_start_response(
"thread-no-client",
@@ -397,7 +397,7 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize
reducer
.ingest(
AnalyticsFact::Response {
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_thread_resume_response(
"thread-1", /*ephemeral*/ true, "gpt-5",

17
codex-rs/analytics/src/client.rs
View File

@@ -16,6 +16,8 @@ use crate::facts::TrackEventsContext;
use crate::reducer::AnalyticsReducer;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ServerResponse;
use codex_login::AuthManager;
use codex_login::default_client::create_client;
use codex_plugin::PluginTelemetryMetadata;
@@ -222,11 +224,24 @@ impl AnalyticsEventsClient {
}
pub fn track_response(&self, connection_id: u64, response: ClientResponse) {
self.record_fact(AnalyticsFact::Response {
self.record_fact(AnalyticsFact::ClientResponse {
connection_id,
response: Box::new(response),
});
}
pub fn track_server_request(&self, connection_id: u64, request: ServerRequest) {
self.record_fact(AnalyticsFact::ServerRequest {
connection_id,
request: Box::new(request),
});
}
pub fn track_server_response(&self, response: ServerResponse) {
self.record_fact(AnalyticsFact::ServerResponse {
response: Box::new(response),
});
}
}
async fn send_track_events(

13
codex-rs/analytics/src/facts.rs
View File

@@ -5,6 +5,8 @@ use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ServerResponse;
use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::protocol::SkillScope;
use codex_protocol::protocol::SubAgentSource;
@@ -72,15 +74,22 @@ pub(crate) enum AnalyticsFact {
runtime: CodexRuntimeMetadata,
rpc_transport: AppServerRpcTransport,
},
Request {
ClientRequest {
connection_id: u64,
request_id: RequestId,
request: Box<ClientRequest>,
},
Response {
ClientResponse {
connection_id: u64,
response: Box<ClientResponse>,
},
ServerRequest {
connection_id: u64,
request: Box<ServerRequest>,
},
ServerResponse {
response: Box<ServerResponse>,
},
Notification(Box<ServerNotification>),
// Facts that do not naturally exist on the app-server protocol surface, or
// would require non-trivial protocol reshaping on this branch.

11
codex-rs/analytics/src/reducer.rs
View File

@@ -65,17 +65,24 @@ impl AnalyticsReducer {
rpc_transport,
);
}
AnalyticsFact::Request {
AnalyticsFact::ClientRequest {
connection_id: _connection_id,
request_id: _request_id,
request: _request,
} => {}
AnalyticsFact::Response {
AnalyticsFact::ClientResponse {
connection_id,
response,
} => {
self.ingest_response(connection_id, *response, out);
}
AnalyticsFact::ServerRequest {
connection_id: _connection_id,
request: _request,
} => {}
AnalyticsFact::ServerResponse {
response: _response,
} => {}
AnalyticsFact::Notification(_notification) => {}
AnalyticsFact::Custom(input) => match input {
CustomAnalyticsFact::SubAgentThreadStarted(input) => {

16
codex-rs/app-server/src/analytics_events.rs Normal file
View File

@@ -0,0 +1,16 @@
use std::sync::Arc;
use codex_analytics::AnalyticsEventsClient;
use codex_core::config::Config;
use codex_login::AuthManager;
pub(crate) fn analytics_events_client_from_config(
auth_manager: Arc<AuthManager>,
config: &Config,
) -> AnalyticsEventsClient {
AnalyticsEventsClient::new(
auth_manager,
config.chatgpt_base_url.trim_end_matches('/').to_string(),
config.analytics_enabled,
)
}

13
codex-rs/app-server/src/in_process.rs
View File

@@ -50,6 +50,7 @@ use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Duration;
use crate::analytics_events::analytics_events_client_from_config;
use crate::error_code::INTERNAL_ERROR_CODE;
use crate::error_code::INVALID_REQUEST_ERROR_CODE;
use crate::error_code::OVERLOADED_ERROR_CODE;
@@ -355,7 +356,14 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
let runtime_handle = tokio::spawn(async move {
let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<OutgoingEnvelope>(channel_capacity);
let outgoing_message_sender = Arc::new(OutgoingMessageSender::new(outgoing_tx));
let auth_manager =
AuthManager::shared_from_config(args.config.as_ref(), args.enable_codex_api_key_env);
let analytics_events_client =
analytics_events_client_from_config(Arc::clone(&auth_manager), args.config.as_ref());
let outgoing_message_sender = Arc::new(OutgoingMessageSender::new_with_analytics(
outgoing_tx,
analytics_events_client.clone(),
));
let (writer_tx, mut writer_rx) = mpsc::channel::<QueuedOutgoingMessage>(channel_capacity);
let outbound_initialized = Arc::new(AtomicBool::new(false));
@@ -380,12 +388,11 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
});
let processor_outgoing = Arc::clone(&outgoing_message_sender);
let auth_manager =
AuthManager::shared_from_config(args.config.as_ref(), args.enable_codex_api_key_env);
let (processor_tx, mut processor_rx) = mpsc::channel::<ProcessorCommand>(channel_capacity);
let mut processor_handle = tokio::spawn(async move {
let mut processor = MessageProcessor::new(MessageProcessorArgs {
outgoing: Arc::clone(&processor_outgoing),
analytics_events_client,
arg0_paths: args.arg0_paths,
config: args.config,
environment_manager: Arc::new(EnvironmentManager::from_env()),

10
codex-rs/app-server/src/lib.rs
View File

@@ -18,6 +18,7 @@ use std::sync::Arc;
use std::sync::RwLock;
use std::sync::atomic::AtomicBool;
use crate::analytics_events::analytics_events_client_from_config;
use crate::message_processor::MessageProcessor;
use crate::message_processor::MessageProcessorArgs;
use crate::outgoing_message::ConnectionId;
@@ -63,6 +64,7 @@ use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::registry::Registry;
use tracing_subscriber::util::SubscriberInitExt;
mod analytics_events;
mod app_server_tracing;
mod bespoke_event_handling;
mod codex_message_processor;
@@ -640,14 +642,20 @@ pub async fn run_main_with_transport(
});
let processor_handle = tokio::spawn({
let outgoing_message_sender = Arc::new(OutgoingMessageSender::new(outgoing_tx));
let outbound_control_tx = outbound_control_tx;
let auth_manager =
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false);
let analytics_events_client =
analytics_events_client_from_config(Arc::clone(&auth_manager), &config);
let outgoing_message_sender = Arc::new(OutgoingMessageSender::new_with_analytics(
outgoing_tx,
analytics_events_client.clone(),
));
let cli_overrides: Vec<(String, TomlValue)> = cli_kv_overrides.clone();
let loader_overrides = loader_overrides_for_config_api;
let mut processor = MessageProcessor::new(MessageProcessorArgs {
outgoing: outgoing_message_sender,
analytics_events_client,
arg0_paths,
config: Arc::new(config),
environment_manager,

7
codex-rs/app-server/src/message_processor.rs
View File

@@ -185,6 +185,7 @@ pub(crate) struct ConnectionSessionState {
pub(crate) struct MessageProcessorArgs {
pub(crate) outgoing: Arc<OutgoingMessageSender>,
pub(crate) analytics_events_client: AnalyticsEventsClient,
pub(crate) arg0_paths: Arg0DispatchPaths,
pub(crate) config: Arc<Config>,
pub(crate) environment_manager: Arc<EnvironmentManager>,
@@ -206,6 +207,7 @@ impl MessageProcessor {
pub(crate) fn new(args: MessageProcessorArgs) -> Self {
let MessageProcessorArgs {
outgoing,
analytics_events_client,
arg0_paths,
config,
environment_manager,
@@ -234,11 +236,6 @@ impl MessageProcessor {
},
environment_manager,
));
let analytics_events_client = AnalyticsEventsClient::new(
Arc::clone(&auth_manager),
config.chatgpt_base_url.trim_end_matches('/').to_string(),
config.analytics_enabled,
);
thread_manager
.plugins_manager()
.set_analytics_events_client(analytics_events_client.clone());

4
codex-rs/app-server/src/message_processor/tracing_tests.rs
View File

@@ -1,6 +1,7 @@
use super::ConnectionSessionState;
use super::MessageProcessor;
use super::MessageProcessorArgs;
use crate::analytics_events::analytics_events_client_from_config;
use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::OutgoingMessageSender;
use crate::transport::AppServerTransport;
@@ -237,8 +238,11 @@ fn build_test_processor(
let outgoing = Arc::new(OutgoingMessageSender::new(outgoing_tx));
let auth_manager =
AuthManager::shared_from_config(config.as_ref(), /*enable_codex_api_key_env*/ false);
let analytics_events_client =
analytics_events_client_from_config(Arc::clone(&auth_manager), config.as_ref());
let processor = MessageProcessor::new(MessageProcessorArgs {
outgoing,
analytics_events_client,
arg0_paths: Arg0DispatchPaths::default(),
config,
environment_manager: Arc::new(EnvironmentManager::new(/*exec_server_url*/ None)),

86
codex-rs/app-server/src/outgoing_message.rs
View File

@@ -4,12 +4,14 @@ use std::sync::Arc;
use std::sync::atomic::AtomicI64;
use std::sync::atomic::Ordering;
use codex_analytics::AnalyticsEventsClient;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::Result;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ServerRequestPayload;
use codex_app_server_protocol::ServerResponse;
use codex_otel::span_w3c_trace_context;
use codex_protocol::ThreadId;
use codex_protocol::protocol::W3cTraceContext;
@@ -117,6 +119,7 @@ pub(crate) struct OutgoingMessageSender {
/// We keep them here because this is where responses, errors, and
/// disconnect cleanup all get handled.
request_contexts: Mutex<HashMap<ConnectionRequestId, RequestContext>>,
analytics_events_client: Option<AnalyticsEventsClient>,
}
#[derive(Clone)]
@@ -203,15 +206,35 @@ impl ThreadScopedOutgoingMessageSender {
}
impl OutgoingMessageSender {
#[cfg(test)]
pub(crate) fn new(sender: mpsc::Sender<OutgoingEnvelope>) -> Self {
Self::new_inner(sender, None)
}
pub(crate) fn new_with_analytics(
sender: mpsc::Sender<OutgoingEnvelope>,
analytics_events_client: AnalyticsEventsClient,
) -> Self {
Self::new_inner(sender, Some(analytics_events_client))
}
fn new_inner(
sender: mpsc::Sender<OutgoingEnvelope>,
analytics_events_client: Option<AnalyticsEventsClient>,
) -> Self {
Self {
next_server_request_id: AtomicI64::new(0),
sender,
request_id_to_callback: Mutex::new(HashMap::new()),
request_contexts: Mutex::new(HashMap::new()),
analytics_events_client,
}
}
fn analytics_events_client(&self) -> Option<&AnalyticsEventsClient> {
self.analytics_events_client.as_ref()
}
pub(crate) async fn register_request_context(&self, request_context: RequestContext) {
let mut request_contexts = self.request_contexts.lock().await;
if request_contexts
@@ -298,7 +321,7 @@ impl OutgoingMessageSender {
);
}
let outgoing_message = OutgoingMessage::Request(request);
let outgoing_message = OutgoingMessage::Request(request.clone());
let send_result = match connection_ids {
None => {
self.sender
@@ -321,6 +344,9 @@ impl OutgoingMessageSender {
{
send_error = Some(err);
break;
} else if let Some(analytics_events_client) = self.analytics_events_client() {
analytics_events_client
.track_server_request(connection_id.0, request.clone());
}
}
match send_error {
@@ -364,6 +390,11 @@ impl OutgoingMessageSender {
match entry {
Some((id, entry)) => {
if let Some(response) = server_response_from_result(&entry.request, result.clone())
&& let Some(analytics_events_client) = self.analytics_events_client()
{
analytics_events_client.track_server_response(response);
}
if let Err(err) = entry.callback.send(Ok(result)) {
warn!("could not notify callback for {id:?} due to: {err:?}");
}
@@ -621,6 +652,14 @@ impl OutgoingMessageSender {
}
}
fn server_response_from_result(request: &ServerRequest, result: Result) -> Option<ServerResponse> {
let mut value = serde_json::to_value(request).ok()?;
let object = value.as_object_mut()?;
object.remove("params");
object.insert("response".to_string(), result);
serde_json::from_value(value).ok()
}
/// Outgoing message from the server to the client.
#[derive(Debug, Clone, Serialize)]
#[serde(untagged)]
@@ -654,6 +693,8 @@ mod tests {
use codex_app_server_protocol::AccountUpdatedNotification;
use codex_app_server_protocol::ApplyPatchApprovalParams;
use codex_app_server_protocol::AuthMode;
use codex_app_server_protocol::CommandExecutionApprovalDecision;
use codex_app_server_protocol::CommandExecutionRequestApprovalParams;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::DynamicToolCallParams;
use codex_app_server_protocol::FileChangeRequestApprovalParams;
@@ -838,6 +879,49 @@ mod tests {
);
}
#[test]
fn server_response_from_result_decodes_typed_response_with_original_method() {
let request = ServerRequest::CommandExecutionRequestApproval {
request_id: RequestId::Integer(7),
params: CommandExecutionRequestApprovalParams {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
item_id: "item-1".to_string(),
approval_id: None,
reason: None,
network_approval_context: None,
command: Some("echo hi".to_string()),
cwd: None,
command_actions: None,
additional_permissions: None,
proposed_execpolicy_amendment: None,
proposed_network_policy_amendments: None,
available_decisions: None,
},
};
let response = server_response_from_result(
&request,
json!({
"decision": "acceptForSession",
}),
)
.expect("decode typed server response");
let ServerResponse::CommandExecutionRequestApproval {
request_id,
response,
} = response
else {
panic!("expected command execution approval response");
};
assert_eq!(request_id, RequestId::Integer(7));
assert_eq!(
response.decision,
CommandExecutionApprovalDecision::AcceptForSession
);
}
#[tokio::test]
async fn send_response_routes_to_target_connection() {
let (tx, mut rx) = mpsc::channel::<OutgoingEnvelope>(4);