Compare commits

...

2 Commits

Author SHA1 Message Date
rhan-oai
7806c42750 [codex-analytics] add tool call event schema 2026-04-07 22:40:06 -07:00
rhan-oai
d02455011f [codex-analytics] ingest server requests and responses 2026-04-07 22:37:09 -07:00
7 changed files with 281 additions and 1 deletions

View File

@@ -6,9 +6,14 @@ use crate::events::CodexAppUsedEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::CodexToolCallEventParams;
use crate::events::CodexToolCallEventRequest;
use crate::events::ThreadInitializationMode;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::ToolCallFinalReviewOutcome;
use crate::events::ToolCallTerminalStatus;
use crate::events::ToolKind;
use crate::events::TrackEventRequest;
use crate::events::codex_app_metadata;
use crate::events::codex_plugin_metadata;
@@ -346,6 +351,89 @@ fn thread_initialized_event_serializes_expected_shape() {
);
}
#[test]
fn tool_call_event_serializes_expected_shape() {
let event = TrackEventRequest::ToolCall(CodexToolCallEventRequest {
event_type: "codex_tool_call_event",
event_params: CodexToolCallEventParams {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
tool_call_id: "tool-call-1".to_string(),
app_server_client: CodexAppServerClientMetadata {
product_client_id: "codex_tui".to_string(),
client_name: Some("codex-tui".to_string()),
client_version: Some("1.2.3".to_string()),
rpc_transport: AppServerRpcTransport::Websocket,
experimental_api_enabled: Some(true),
},
runtime: CodexRuntimeMetadata {
codex_rs_version: "0.99.0".to_string(),
runtime_os: "macos".to_string(),
runtime_os_version: "15.3.1".to_string(),
runtime_arch: "aarch64".to_string(),
},
tool_name: "shell".to_string(),
tool_kind: ToolKind::Shell,
started_at: 123,
completed_at: Some(125),
duration_ms: Some(2000),
execution_started: true,
review_count: 0,
guardian_review_count: 0,
user_review_count: 0,
final_review_outcome: ToolCallFinalReviewOutcome::NotNeeded,
terminal_status: ToolCallTerminalStatus::Completed,
failure_kind: None,
exit_code: Some(0),
requested_additional_permissions: false,
requested_network_access: false,
retry_count: 0,
},
});
let payload = serde_json::to_value(&event).expect("serialize tool call event");
assert_eq!(
payload,
json!({
"event_type": "codex_tool_call_event",
"event_params": {
"thread_id": "thread-1",
"turn_id": "turn-1",
"tool_call_id": "tool-call-1",
"app_server_client": {
"product_client_id": "codex_tui",
"client_name": "codex-tui",
"client_version": "1.2.3",
"rpc_transport": "websocket",
"experimental_api_enabled": true
},
"runtime": {
"codex_rs_version": "0.99.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64"
},
"tool_name": "shell",
"tool_kind": "shell",
"started_at": 123,
"completed_at": 125,
"duration_ms": 2000,
"execution_started": true,
"review_count": 0,
"guardian_review_count": 0,
"user_review_count": 0,
"final_review_outcome": "not_needed",
"terminal_status": "completed",
"failure_kind": null,
"exit_code": 0,
"requested_additional_permissions": false,
"requested_network_access": false,
"retry_count": 0
}
})
);
}
#[tokio::test]
async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialized() {
let mut reducer = AnalyticsReducer::default();

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;
@@ -227,6 +229,19 @@ impl AnalyticsEventsClient {
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(

View File

@@ -37,6 +37,7 @@ pub(crate) enum TrackEventRequest {
ThreadInitialized(ThreadInitializedEvent),
AppMentioned(CodexAppMentionedEventRequest),
AppUsed(CodexAppUsedEventRequest),
ToolCall(CodexToolCallEventRequest),
PluginUsed(CodexPluginUsedEventRequest),
PluginInstalled(CodexPluginEventRequest),
PluginUninstalled(CodexPluginEventRequest),
@@ -99,6 +100,81 @@ pub(crate) struct ThreadInitializedEvent {
pub(crate) event_params: ThreadInitializedEventParams,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ToolKind {
Shell,
UnifiedExec,
ApplyPatch,
Mcp,
Dynamic,
Other,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ToolCallFinalReviewOutcome {
NotNeeded,
GuardianApproved,
GuardianDenied,
GuardianAborted,
UserApproved,
UserApprovedForSession,
UserDenied,
UserAborted,
ConfigAllowed,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ToolCallTerminalStatus {
Completed,
Failed,
Rejected,
Interrupted,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ToolCallFailureKind {
ToolError,
ApprovalDenied,
ApprovalAborted,
SandboxDenied,
PolicyForbidden,
}
#[derive(Serialize)]
pub(crate) struct CodexToolCallEventParams {
pub(crate) thread_id: String,
pub(crate) turn_id: String,
pub(crate) tool_call_id: String,
pub(crate) app_server_client: CodexAppServerClientMetadata,
pub(crate) runtime: CodexRuntimeMetadata,
pub(crate) tool_name: String,
pub(crate) tool_kind: ToolKind,
pub(crate) started_at: u64,
pub(crate) completed_at: Option<u64>,
pub(crate) duration_ms: Option<u64>,
pub(crate) execution_started: bool,
pub(crate) review_count: u64,
pub(crate) guardian_review_count: u64,
pub(crate) user_review_count: u64,
pub(crate) final_review_outcome: ToolCallFinalReviewOutcome,
pub(crate) terminal_status: ToolCallTerminalStatus,
pub(crate) failure_kind: Option<ToolCallFailureKind>,
pub(crate) exit_code: Option<i32>,
pub(crate) requested_additional_permissions: bool,
pub(crate) requested_network_access: bool,
pub(crate) retry_count: u64,
}
#[derive(Serialize)]
pub(crate) struct CodexToolCallEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: CodexToolCallEventParams,
}
#[derive(Serialize)]
pub(crate) struct CodexAppMetadata {
pub(crate) connector_id: Option<String>,

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;
@@ -81,6 +83,13 @@ pub(crate) enum AnalyticsFact {
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.

View File

@@ -76,6 +76,13 @@ impl AnalyticsReducer {
} => {
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) => {

View File

@@ -239,6 +239,7 @@ impl MessageProcessor {
config.chatgpt_base_url.trim_end_matches('/').to_string(),
config.analytics_enabled,
);
outgoing.set_analytics_events_client(analytics_events_client.clone());
thread_manager
.plugins_manager()
.set_analytics_events_client(analytics_events_client.clone());

View File

@@ -1,15 +1,18 @@
use std::collections::HashMap;
use std::fmt;
use std::sync::Arc;
use std::sync::Mutex as StdMutex;
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 +120,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: StdMutex<Option<AnalyticsEventsClient>>,
}
#[derive(Clone)]
@@ -209,9 +213,28 @@ impl OutgoingMessageSender {
sender,
request_id_to_callback: Mutex::new(HashMap::new()),
request_contexts: Mutex::new(HashMap::new()),
analytics_events_client: StdMutex::new(None),
}
}
pub(crate) fn set_analytics_events_client(
&self,
analytics_events_client: AnalyticsEventsClient,
) {
let mut client = self
.analytics_events_client
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
*client = Some(analytics_events_client);
}
fn analytics_events_client(&self) -> Option<AnalyticsEventsClient> {
self.analytics_events_client
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone()
}
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);