mirrors/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-05-04 13:21:54 +03:00
Code Issues Packages Projects Releases Wiki Activity

[codex-analytics] feature plumbing and emittance

Browse Source
This commit is contained in:
rhan-oai
2026-04-10 18:31:23 -07:00
parent 58933237cd
commit 2ab4bc05df
26 changed files with 1416 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -7,7 +7,7 @@ use crate::events::CodexCompactionEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::ThreadInitializationMode;
use crate::events::CodexTurnEventRequest;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::TrackEventRequest;
@@ -34,29 +34,46 @@ use crate::facts::PluginUsedInput;
use crate::facts::SkillInvocation;
use crate::facts::SkillInvokedInput;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::ThreadInitializationMode;
use crate::facts::TrackEventsContext;
use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnStatus;
use crate::reducer::AnalyticsReducer;
use crate::reducer::normalize_path_for_skill_id;
use crate::reducer::skill_id_for_local_skill;
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::InitializeCapabilities;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::RequestId;
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::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::TurnCompletedNotification;
use codex_app_server_protocol::TurnError as AppServerTurnError;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartedNotification;
use codex_app_server_protocol::TurnStatus as AppServerTurnStatus;
use codex_app_server_protocol::UserInput;
use codex_login::default_client::DEFAULT_ORIGINATOR;
use codex_login::default_client::originator;
use codex_plugin::AppConnectorId;
use codex_plugin::PluginCapabilitySummary;
use codex_plugin::PluginId;
use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::ModeKind;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use pretty_assertions::assert_eq;
use serde_json::json;
@@ -163,6 +180,181 @@ fn sample_thread_resume_response_with_source(
}
}
fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest {
ClientRequest::TurnStart {
request_id: RequestId::Integer(request_id),
params: TurnStartParams {
thread_id: thread_id.to_string(),
input: vec![
UserInput::Text {
text: "hello".to_string(),
text_elements: vec![],
},
UserInput::Image {
url: "https://example.com/a.png".to_string(),
},
],
..Default::default()
},
}
}
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_started_notification(thread_id: &str, turn_id: &str) -> ServerNotification {
ServerNotification::TurnStarted(TurnStartedNotification {
thread_id: thread_id.to_string(),
turn: Turn {
id: turn_id.to_string(),
items: vec![],
status: AppServerTurnStatus::InProgress,
error: None,
started_at: Some(455),
completed_at: None,
duration_ms: None,
},
})
}
fn sample_turn_completed_notification(
thread_id: &str,
turn_id: &str,
status: AppServerTurnStatus,
codex_error_info: Option<codex_app_server_protocol::CodexErrorInfo>,
) -> ServerNotification {
ServerNotification::TurnCompleted(TurnCompletedNotification {
thread_id: thread_id.to_string(),
turn: Turn {
id: turn_id.to_string(),
items: vec![],
status,
error: codex_error_info.map(|codex_error_info| AppServerTurnError {
message: "turn failed".to_string(),
codex_error_info: Some(codex_error_info),
additional_details: None,
}),
started_at: None,
completed_at: Some(456),
duration_ms: Some(1234),
},
})
}
fn sample_turn_resolved_config(turn_id: &str) -> TurnResolvedConfigFact {
TurnResolvedConfigFact {
turn_id: turn_id.to_string(),
thread_id: "thread-2".to_string(),
num_input_images: 1,
submission_type: None,
ephemeral: false,
session_source: SessionSource::Exec,
model: "gpt-5".to_string(),
model_provider: "openai".to_string(),
sandbox_policy: SandboxPolicy::new_read_only_policy(),
reasoning_effort: None,
reasoning_summary: None,
service_tier: None,
approval_policy: AskForApproval::OnRequest,
approvals_reviewer: ApprovalsReviewer::GuardianSubagent,
sandbox_network_access: true,
collaboration_mode: ModeKind::Plan,
personality: None,
is_first_turn: true,
}
}
async fn ingest_turn_prerequisites(
reducer: &mut AnalyticsReducer,
out: &mut Vec<TrackEventRequest>,
include_initialize: bool,
include_resolved_config: bool,
include_started: bool,
) {
if include_initialize {
reducer
.ingest(
AnalyticsFact::Initialize {
connection_id: 7,
params: InitializeParams {
client_info: ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: None,
},
product_client_id: "codex-tui".to_string(),
runtime: CodexRuntimeMetadata {
codex_rs_version: "0.1.0".to_string(),
runtime_os: "macos".to_string(),
runtime_os_version: "15.3.1".to_string(),
runtime_arch: "aarch64".to_string(),
},
rpc_transport: AppServerRpcTransport::Stdio,
},
out,
)
.await;
}
reducer
.ingest(
AnalyticsFact::Request {
connection_id: 7,
request_id: RequestId::Integer(3),
request: Box::new(sample_turn_start_request("thread-2", /*request_id*/ 3)),
},
out,
)
.await;
reducer
.ingest(
AnalyticsFact::Response {
connection_id: 7,
response: Box::new(sample_turn_start_response("turn-2", /*request_id*/ 3)),
},
out,
)
.await;
if include_resolved_config {
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
sample_turn_resolved_config("turn-2"),
))),
out,
)
.await;
}
if include_started {
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_started_notification(
"thread-2", "turn-2",
))),
out,
)
.await;
}
}
fn expected_absolute_path(path: &PathBuf) -> String {
std::fs::canonicalize(path)
.unwrap_or_else(|_| path.to_path_buf())
@@ -563,10 +755,6 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize
payload[0]["event_params"]["runtime"]["runtime_arch"],
"x86_64"
);
assert_eq!(payload[0]["event_params"]["initialization_mode"], "resumed");
assert_eq!(payload[0]["event_params"]["thread_source"], "user");
assert_eq!(payload[0]["event_params"]["subagent_source"], json!(null));
assert_eq!(payload[0]["event_params"]["parent_thread_id"], json!(null));
}
#[tokio::test]
@@ -1087,6 +1275,305 @@ async fn reducer_ingests_plugin_state_changed_fact() {
);
}
#[test]
fn turn_event_serializes_expected_shape() {
let event = TrackEventRequest::TurnEvent(Box::new(CodexTurnEventRequest {
event_type: "codex_turn_event",
event_params: crate::events::CodexTurnEventParams {
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
app_server_client: sample_app_server_client_metadata(),
runtime: sample_runtime_metadata(),
submission_type: None,
ephemeral: false,
model: Some("gpt-5".to_string()),
model_provider: "openai".to_string(),
sandbox_policy: Some("read_only"),
reasoning_effort: Some("high".to_string()),
reasoning_summary: Some("detailed".to_string()),
service_tier: "flex".to_string(),
approval_policy: "on-request".to_string(),
approvals_reviewer: "guardian_subagent".to_string(),
sandbox_network_access: true,
collaboration_mode: Some("plan"),
personality: Some("pragmatic".to_string()),
num_input_images: 2,
is_first_turn: true,
status: Some(TurnStatus::Completed),
turn_error: None,
steer_count: None,
total_tool_call_count: None,
shell_command_count: None,
file_change_count: None,
mcp_tool_call_count: None,
dynamic_tool_call_count: None,
subagent_tool_call_count: None,
web_search_count: None,
image_generation_count: None,
duration_ms: Some(1234),
started_at: Some(455),
completed_at: Some(456),
},
}));
let payload = serde_json::to_value(&event).expect("serialize turn event");
let expected = serde_json::from_str::<serde_json::Value>(
r#"{
"event_type": "codex_turn_event",
"event_params": {
"thread_id": "thread-2",
"turn_id": "turn-2",
"submission_type": null,
"app_server_client": {
"product_client_id": "codex_cli_rs",
"client_name": "codex-tui",
"client_version": "1.0.0",
"rpc_transport": "stdio",
"experimental_api_enabled": true
},
"runtime": {
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64"
},
"ephemeral": false,
"model": "gpt-5",
"model_provider": "openai",
"sandbox_policy": "read_only",
"reasoning_effort": "high",
"reasoning_summary": "detailed",
"service_tier": "flex",
"approval_policy": "on-request",
"approvals_reviewer": "guardian_subagent",
"sandbox_network_access": true,
"collaboration_mode": "plan",
"personality": "pragmatic",
"num_input_images": 2,
"is_first_turn": true,
"status": "completed",
"turn_error": null,
"steer_count": null,
"total_tool_call_count": null,
"shell_command_count": null,
"file_change_count": null,
"mcp_tool_call_count": null,
"dynamic_tool_call_count": null,
"subagent_tool_call_count": null,
"web_search_count": null,
"image_generation_count": null,
"duration_ms": 1234,
"started_at": 455,
"completed_at": 456
}
}"#,
)
.expect("parse expected turn event");
assert_eq!(payload, expected);
}
#[tokio::test]
async fn turn_lifecycle_emits_turn_event() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
assert_eq!(out.len(), 1);
let payload = serde_json::to_value(&out[0]).expect("serialize turn event");
assert_eq!(payload["event_type"], json!("codex_turn_event"));
assert_eq!(payload["event_params"]["thread_id"], json!("thread-2"));
assert_eq!(payload["event_params"]["turn_id"], json!("turn-2"));
assert_eq!(
payload["event_params"]["app_server_client"],
json!({
"product_client_id": "codex-tui",
"client_name": "codex-tui",
"client_version": "1.0.0",
"rpc_transport": "stdio",
"experimental_api_enabled": null,
})
);
assert_eq!(
payload["event_params"]["runtime"],
json!({
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64",
})
);
assert!(payload["event_params"].get("product_client_id").is_none());
assert_eq!(payload["event_params"]["ephemeral"], json!(false));
assert_eq!(payload["event_params"]["num_input_images"], json!(1));
assert_eq!(payload["event_params"]["status"], json!("completed"));
assert_eq!(payload["event_params"]["started_at"], json!(455));
assert_eq!(payload["event_params"]["completed_at"], json!(456));
assert_eq!(payload["event_params"]["duration_ms"], json!(1234));
}
#[tokio::test]
async fn turn_does_not_emit_without_required_prerequisites() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ false,
/*include_resolved_config*/ true,
/*include_started*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
assert!(out.is_empty());
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ false,
/*include_started*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
assert!(out.is_empty());
}
#[tokio::test]
async fn turn_lifecycle_emits_failed_turn_event() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Failed,
Some(codex_app_server_protocol::CodexErrorInfo::BadRequest),
))),
&mut out,
)
.await;
assert_eq!(out.len(), 1);
let payload = serde_json::to_value(&out[0]).expect("serialize turn event");
assert_eq!(payload["event_params"]["status"], json!("failed"));
assert_eq!(payload["event_params"]["turn_error"], json!("badRequest"));
}
#[tokio::test]
async fn turn_lifecycle_emits_interrupted_turn_event_without_error() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Interrupted,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
assert_eq!(out.len(), 1);
let payload = serde_json::to_value(&out[0]).expect("serialize turn event");
assert_eq!(payload["event_params"]["status"], json!("interrupted"));
assert_eq!(payload["event_params"]["turn_error"], json!(null));
}
#[tokio::test]
async fn turn_completed_without_started_notification_emits_null_started_at() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
let payload = serde_json::to_value(&out[0]).expect("serialize turn event");
assert_eq!(payload["event_params"]["started_at"], json!(null));
assert_eq!(payload["event_params"]["duration_ms"], json!(1234));
}
fn sample_plugin_metadata() -> PluginTelemetryMetadata {
PluginTelemetryMetadata {
plugin_id: PluginId::parse("sample@test").expect("valid plugin id"),

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

@@ -14,9 +14,13 @@ use crate::facts::SkillInvocation;
use crate::facts::SkillInvokedInput;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::TrackEventsContext;
use crate::facts::TurnResolvedConfigFact;
use crate::reducer::AnalyticsReducer;
use codex_app_server_protocol::ClientRequest;
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_login::AuthManager;
use codex_login::default_client::create_client;
use codex_plugin::PluginTelemetryMetadata;
@@ -167,6 +171,14 @@ impl AnalyticsEventsClient {
)));
}
pub fn track_request(&self, connection_id: u64, request_id: RequestId, request: ClientRequest) {
self.record_fact(AnalyticsFact::Request {
connection_id,
request_id,
request: Box::new(request),
});
}
pub fn track_app_used(&self, tracking: TrackEventsContext, app: AppInvocation) {
if !self.queue.should_enqueue_app_used(&tracking, &app) {
return;
@@ -191,6 +203,12 @@ impl AnalyticsEventsClient {
)));
}
pub fn track_turn_resolved_config(&self, fact: TurnResolvedConfigFact) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::TurnResolvedConfig(Box::new(fact)),
));
}
pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
@@ -240,6 +258,10 @@ impl AnalyticsEventsClient {
response: Box::new(response),
});
}
pub fn track_notification(&self, notification: ServerNotification) {
self.record_fact(AnalyticsFact::Notification(Box::new(notification)));
}
}
async fn send_track_events(

56
codex-rs/analytics/src/events.rs
View File

@@ -3,7 +3,11 @@ use crate::facts::CodexCompactionEvent;
use crate::facts::InvocationType;
use crate::facts::PluginState;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::ThreadInitializationMode;
use crate::facts::TrackEventsContext;
use crate::facts::TurnStatus;
use crate::facts::TurnSubmissionType;
use codex_app_server_protocol::CodexErrorInfo;
use codex_login::default_client::originator;
use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::approvals::NetworkApprovalProtocol;
@@ -21,14 +25,6 @@ pub enum AppServerRpcTransport {
InProcess,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ThreadInitializationMode {
New,
Forked,
Resumed,
}
#[derive(Serialize)]
pub(crate) struct TrackEventsRequest {
pub(crate) events: Vec<TrackEventRequest>,
@@ -43,6 +39,7 @@ pub(crate) enum TrackEventRequest {
AppMentioned(CodexAppMentionedEventRequest),
AppUsed(CodexAppUsedEventRequest),
Compaction(Box<CodexCompactionEventRequest>),
TurnEvent(Box<CodexTurnEventRequest>),
PluginUsed(CodexPluginUsedEventRequest),
PluginInstalled(CodexPluginEventRequest),
PluginUninstalled(CodexPluginEventRequest),
@@ -330,6 +327,49 @@ pub(crate) struct CodexCompactionEventRequest {
pub(crate) event_params: CodexCompactionEventParams,
}
#[derive(Serialize)]
pub(crate) struct CodexTurnEventParams {
pub(crate) thread_id: String,
pub(crate) turn_id: String,
pub(crate) submission_type: Option<TurnSubmissionType>,
pub(crate) app_server_client: CodexAppServerClientMetadata,
pub(crate) runtime: CodexRuntimeMetadata,
pub(crate) ephemeral: bool,
pub(crate) model: Option<String>,
pub(crate) model_provider: String,
pub(crate) sandbox_policy: Option<&'static str>,
pub(crate) reasoning_effort: Option<String>,
pub(crate) reasoning_summary: Option<String>,
pub(crate) service_tier: String,
pub(crate) approval_policy: String,
pub(crate) approvals_reviewer: String,
pub(crate) sandbox_network_access: bool,
pub(crate) collaboration_mode: Option<&'static str>,
pub(crate) personality: Option<String>,
pub(crate) num_input_images: usize,
pub(crate) is_first_turn: bool,
pub(crate) status: Option<TurnStatus>,
pub(crate) turn_error: Option<CodexErrorInfo>,
pub(crate) steer_count: Option<usize>,
pub(crate) total_tool_call_count: Option<usize>,
pub(crate) shell_command_count: Option<usize>,
pub(crate) file_change_count: Option<usize>,
pub(crate) mcp_tool_call_count: Option<usize>,
pub(crate) dynamic_tool_call_count: Option<usize>,
pub(crate) subagent_tool_call_count: Option<usize>,
pub(crate) web_search_count: Option<usize>,
pub(crate) image_generation_count: Option<usize>,
pub(crate) duration_ms: Option<u64>,
pub(crate) started_at: Option<u64>,
pub(crate) completed_at: Option<u64>,
}
#[derive(Serialize)]
pub(crate) struct CodexTurnEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: CodexTurnEventParams,
}
#[derive(Serialize)]
pub(crate) struct CodexPluginMetadata {
pub(crate) plugin_id: Option<String>,

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

@@ -7,6 +7,15 @@ use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Personality;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::config_types::ServiceTier;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SkillScope;
use codex_protocol::protocol::SubAgentSource;
use serde::Serialize;
@@ -31,6 +40,51 @@ pub fn build_track_events_context(
}
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum TurnSubmissionType {
Default,
Queued,
}
#[derive(Clone)]
pub struct TurnResolvedConfigFact {
pub turn_id: String,
pub thread_id: String,
pub num_input_images: usize,
pub submission_type: Option<TurnSubmissionType>,
pub ephemeral: bool,
pub session_source: SessionSource,
pub model: String,
pub model_provider: String,
pub sandbox_policy: SandboxPolicy,
pub reasoning_effort: Option<ReasoningEffort>,
pub reasoning_summary: Option<ReasoningSummary>,
pub service_tier: Option<ServiceTier>,
pub approval_policy: AskForApproval,
pub approvals_reviewer: ApprovalsReviewer,
pub sandbox_network_access: bool,
pub collaboration_mode: ModeKind,
pub personality: Option<Personality>,
pub is_first_turn: bool,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ThreadInitializationMode {
New,
Forked,
Resumed,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum TurnStatus {
Completed,
Failed,
Interrupted,
}
#[derive(Clone, Debug)]
pub struct SkillInvocation {
pub skill_name: String,
@@ -156,6 +210,7 @@ pub(crate) enum CustomAnalyticsFact {
SubAgentThreadStarted(SubAgentThreadStartedInput),
Compaction(Box<CodexCompactionEvent>),
GuardianReview(Box<GuardianReviewEventParams>),
TurnResolvedConfig(Box<TurnResolvedConfigFact>),
SkillInvoked(SkillInvokedInput),
AppMentioned(AppMentionedInput),
AppUsed(AppUsedInput),

3
codex-rs/analytics/src/lib.rs
View File

@@ -27,7 +27,10 @@ pub use facts::CompactionTrigger;
pub use facts::InvocationType;
pub use facts::SkillInvocation;
pub use facts::SubAgentThreadStartedInput;
pub use facts::ThreadInitializationMode;
pub use facts::TrackEventsContext;
pub use facts::TurnResolvedConfigFact;
pub use facts::TurnStatus;
pub use facts::build_track_events_context;
#[cfg(test)]

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

@@ -6,12 +6,13 @@ use crate::events::CodexCompactionEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::CodexTurnEventParams;
use crate::events::CodexTurnEventRequest;
use crate::events::GuardianReviewEventParams;
use crate::events::GuardianReviewEventPayload;
use crate::events::GuardianReviewEventRequest;
use crate::events::SkillInvocationEventParams;
use crate::events::SkillInvocationEventRequest;
use crate::events::ThreadInitializationMode;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::TrackEventRequest;
@@ -34,11 +35,23 @@ use crate::facts::PluginStateChangedInput;
use crate::facts::PluginUsedInput;
use crate::facts::SkillInvokedInput;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::ThreadInitializationMode;
use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnStatus;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::CodexErrorInfo;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::UserInput;
use codex_git_utils::collect_git_info;
use codex_git_utils::get_git_repo_root;
use codex_login::default_client::originator;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Personality;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SkillScope;
use sha1::Digest;
@@ -47,6 +60,8 @@ use std::path::Path;
#[derive(Default)]
pub(crate) struct AnalyticsReducer {
requests: HashMap<(u64, RequestId), RequestState>,
turns: HashMap<String, TurnState>,
connections: HashMap<u64, ConnectionState>,
thread_connections: HashMap<String, u64>,
thread_metadata: HashMap<String, ThreadMetadataState>,
@@ -86,6 +101,32 @@ impl ThreadMetadataState {
}
}
enum RequestState {
TurnStart(PendingTurnStartState),
}
struct PendingTurnStartState {
thread_id: String,
num_input_images: usize,
}
#[derive(Clone)]
struct CompletedTurnState {
status: Option<TurnStatus>,
turn_error: Option<CodexErrorInfo>,
completed_at: u64,
duration_ms: Option<u64>,
}
struct TurnState {
connection_id: Option<u64>,
thread_id: Option<String>,
num_input_images: Option<usize>,
resolved_config: Option<TurnResolvedConfigFact>,
started_at: Option<u64>,
completed: Option<CompletedTurnState>,
}
impl AnalyticsReducer {
pub(crate) async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec<TrackEventRequest>) {
match input {
@@ -105,17 +146,21 @@ impl AnalyticsReducer {
);
}
AnalyticsFact::Request {
connection_id: _connection_id,
request_id: _request_id,
request: _request,
} => {}
connection_id,
request_id,
request,
} => {
self.ingest_request(connection_id, request_id, *request);
}
AnalyticsFact::Response {
connection_id,
response,
} => {
self.ingest_response(connection_id, *response, out);
}
AnalyticsFact::Notification(_notification) => {}
AnalyticsFact::Notification(notification) => {
self.ingest_notification(*notification, out);
}
AnalyticsFact::Custom(input) => match input {
CustomAnalyticsFact::SubAgentThreadStarted(input) => {
self.ingest_subagent_thread_started(input, out);
@@ -126,6 +171,9 @@ impl AnalyticsReducer {
CustomAnalyticsFact::GuardianReview(input) => {
self.ingest_guardian_review(*input, out);
}
CustomAnalyticsFact::TurnResolvedConfig(input) => {
self.ingest_turn_resolved_config(*input, out);
}
CustomAnalyticsFact::SkillInvoked(input) => {
self.ingest_skill_invoked(input, out).await;
}
@@ -216,6 +264,52 @@ impl AnalyticsReducer {
)));
}
fn ingest_request(
&mut self,
connection_id: u64,
request_id: RequestId,
request: ClientRequest,
) {
let ClientRequest::TurnStart { params, .. } = request else {
return;
};
self.requests.insert(
(connection_id, request_id),
RequestState::TurnStart(PendingTurnStartState {
thread_id: params.thread_id,
num_input_images: params
.input
.iter()
.filter(|item| {
matches!(item, UserInput::Image { .. } | UserInput::LocalImage { .. })
})
.count(),
}),
);
}
fn ingest_turn_resolved_config(
&mut self,
input: TurnResolvedConfigFact,
out: &mut Vec<TrackEventRequest>,
) {
let turn_id = input.turn_id.clone();
let thread_id = input.thread_id.clone();
let num_input_images = input.num_input_images;
let turn_state = self.turns.entry(turn_id.clone()).or_insert(TurnState {
connection_id: None,
thread_id: None,
num_input_images: None,
resolved_config: None,
started_at: None,
completed: None,
});
turn_state.thread_id = Some(thread_id);
turn_state.num_input_images = Some(num_input_images);
turn_state.resolved_config = Some(input);
self.maybe_emit_turn_event(&turn_id, out);
}
async fn ingest_skill_invoked(
&mut self,
input: SkillInvokedInput,
@@ -316,24 +410,124 @@ impl AnalyticsReducer {
response: ClientResponse,
out: &mut Vec<TrackEventRequest>,
) {
let (thread, model, initialization_mode) = match response {
ClientResponse::ThreadStart { response, .. } => (
response.thread,
response.model,
ThreadInitializationMode::New,
),
ClientResponse::ThreadResume { response, .. } => (
response.thread,
response.model,
ThreadInitializationMode::Resumed,
),
ClientResponse::ThreadFork { response, .. } => (
response.thread,
response.model,
ThreadInitializationMode::Forked,
),
_ => return,
};
match response {
ClientResponse::ThreadStart { response, .. } => {
self.emit_thread_initialized(
connection_id,
response.thread,
response.model,
ThreadInitializationMode::New,
out,
);
}
ClientResponse::ThreadResume { response, .. } => {
self.emit_thread_initialized(
connection_id,
response.thread,
response.model,
ThreadInitializationMode::Resumed,
out,
);
}
ClientResponse::ThreadFork { response, .. } => {
self.emit_thread_initialized(
connection_id,
response.thread,
response.model,
ThreadInitializationMode::Forked,
out,
);
}
ClientResponse::TurnStart {
request_id,
response,
} => {
let turn_id = response.turn.id;
let Some(RequestState::TurnStart(pending_request)) =
self.requests.remove(&(connection_id, request_id))
else {
return;
};
let turn_state = self.turns.entry(turn_id.clone()).or_insert(TurnState {
connection_id: None,
thread_id: None,
num_input_images: None,
resolved_config: None,
started_at: None,
completed: None,
});
turn_state.connection_id = Some(connection_id);
turn_state.thread_id = Some(pending_request.thread_id);
turn_state.num_input_images = Some(pending_request.num_input_images);
self.maybe_emit_turn_event(&turn_id, out);
}
_ => {}
}
}
fn ingest_notification(
&mut self,
notification: ServerNotification,
out: &mut Vec<TrackEventRequest>,
) {
match notification {
ServerNotification::TurnStarted(notification) => {
let turn_state = self.turns.entry(notification.turn.id).or_insert(TurnState {
connection_id: None,
thread_id: None,
num_input_images: None,
resolved_config: None,
started_at: None,
completed: None,
});
turn_state.started_at = notification
.turn
.started_at
.and_then(|started_at| u64::try_from(started_at).ok());
}
ServerNotification::TurnCompleted(notification) => {
let turn_state =
self.turns
.entry(notification.turn.id.clone())
.or_insert(TurnState {
connection_id: None,
thread_id: None,
num_input_images: None,
resolved_config: None,
started_at: None,
completed: None,
});
turn_state.completed = Some(CompletedTurnState {
status: analytics_turn_status(notification.turn.status),
turn_error: notification
.turn
.error
.and_then(|error| error.codex_error_info),
completed_at: notification
.turn
.completed_at
.and_then(|completed_at| u64::try_from(completed_at).ok())
.unwrap_or_default(),
duration_ms: notification
.turn
.duration_ms
.and_then(|duration_ms| u64::try_from(duration_ms).ok()),
});
let turn_id = notification.turn.id;
self.maybe_emit_turn_event(&turn_id, out);
}
_ => {}
}
}
fn emit_thread_initialized(
&mut self,
connection_id: u64,
thread: codex_app_server_protocol::Thread,
model: String,
initialization_mode: ThreadInitializationMode,
out: &mut Vec<TrackEventRequest>,
) {
let thread_source: SessionSource = thread.source.into();
let thread_id = thread.id;
let Some(connection_state) = self.connections.get(&connection_id) else {
@@ -403,6 +597,156 @@ impl AnalyticsReducer {
},
)));
}
fn maybe_emit_turn_event(&mut self, turn_id: &str, out: &mut Vec<TrackEventRequest>) {
let Some(turn_state) = self.turns.get(turn_id) else {
return;
};
if turn_state.thread_id.is_none()
|| turn_state.num_input_images.is_none()
|| turn_state.resolved_config.is_none()
|| turn_state.completed.is_none()
{
return;
}
let connection_metadata = turn_state
.connection_id
.and_then(|connection_id| self.connections.get(&connection_id))
.map(|connection_state| {
(
connection_state.app_server_client.clone(),
connection_state.runtime.clone(),
)
});
let Some((app_server_client, runtime)) = connection_metadata else {
return;
};
out.push(TrackEventRequest::TurnEvent(Box::new(
CodexTurnEventRequest {
event_type: "codex_turn_event",
event_params: codex_turn_event_params(
app_server_client,
runtime,
turn_id.to_string(),
turn_state,
),
},
)));
self.turns.remove(turn_id);
}
}
fn codex_turn_event_params(
app_server_client: CodexAppServerClientMetadata,
runtime: CodexRuntimeMetadata,
turn_id: String,
turn_state: &TurnState,
) -> CodexTurnEventParams {
let (Some(thread_id), Some(num_input_images), Some(resolved_config), Some(completed)) = (
turn_state.thread_id.clone(),
turn_state.num_input_images,
turn_state.resolved_config.clone(),
turn_state.completed.clone(),
) else {
unreachable!("turn event params require a fully populated turn state");
};
let started_at = turn_state.started_at;
let TurnResolvedConfigFact {
turn_id: _resolved_turn_id,
thread_id: _resolved_thread_id,
num_input_images: _resolved_num_input_images,
submission_type,
ephemeral,
session_source: _session_source,
model,
model_provider,
sandbox_policy,
reasoning_effort,
reasoning_summary,
service_tier,
approval_policy,
approvals_reviewer,
sandbox_network_access,
collaboration_mode,
personality,
is_first_turn,
} = resolved_config;
CodexTurnEventParams {
thread_id,
turn_id,
app_server_client,
runtime,
submission_type,
ephemeral,
model: Some(model),
model_provider,
sandbox_policy: Some(sandbox_policy_mode(&sandbox_policy)),
reasoning_effort: reasoning_effort.map(|value| value.to_string()),
reasoning_summary: reasoning_summary_mode(reasoning_summary),
service_tier: service_tier
.map(|value| value.to_string())
.unwrap_or_else(|| "default".to_string()),
approval_policy: approval_policy.to_string(),
approvals_reviewer: approvals_reviewer.to_string(),
sandbox_network_access,
collaboration_mode: Some(collaboration_mode_mode(collaboration_mode)),
personality: personality_mode(personality),
num_input_images,
is_first_turn,
status: completed.status,
turn_error: completed.turn_error,
steer_count: None,
total_tool_call_count: None,
shell_command_count: None,
file_change_count: None,
mcp_tool_call_count: None,
dynamic_tool_call_count: None,
subagent_tool_call_count: None,
web_search_count: None,
image_generation_count: None,
duration_ms: completed.duration_ms,
started_at,
completed_at: Some(completed.completed_at),
}
}
fn sandbox_policy_mode(sandbox_policy: &SandboxPolicy) -> &'static str {
match sandbox_policy {
SandboxPolicy::DangerFullAccess => "full_access",
SandboxPolicy::ReadOnly { .. } => "read_only",
SandboxPolicy::WorkspaceWrite { .. } => "workspace_write",
SandboxPolicy::ExternalSandbox { .. } => "external_sandbox",
}
}
fn collaboration_mode_mode(mode: ModeKind) -> &'static str {
match mode {
ModeKind::Plan => "plan",
ModeKind::Default | ModeKind::PairProgramming | ModeKind::Execute => "default",
}
}
fn reasoning_summary_mode(summary: Option<ReasoningSummary>) -> Option<String> {
match summary {
Some(ReasoningSummary::None) | None => None,
Some(summary) => Some(summary.to_string()),
}
}
fn personality_mode(personality: Option<Personality>) -> Option<String> {
match personality {
Some(Personality::None) | None => None,
Some(personality) => Some(personality.to_string()),
}
}
fn analytics_turn_status(status: codex_app_server_protocol::TurnStatus) -> Option<TurnStatus> {
match status {
codex_app_server_protocol::TurnStatus::Completed => Some(TurnStatus::Completed),
codex_app_server_protocol::TurnStatus::Failed => Some(TurnStatus::Failed),
codex_app_server_protocol::TurnStatus::Interrupted => Some(TurnStatus::Interrupted),
codex_app_server_protocol::TurnStatus::InProgress => None,
}
}
pub(crate) fn skill_id_for_local_skill(