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

[codex-analytics] emit tool item events from item lifecycle

Browse Source
This commit is contained in:
rhan-oai
2026-04-20 13:07:01 -07:00
parent 888f26ae4b
commit ff079effc8
5 changed files with 890 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -62,8 +62,12 @@ use codex_app_server_protocol::ClientInfo;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::CodexErrorInfo;
use codex_app_server_protocol::CommandExecutionSource;
use codex_app_server_protocol::CommandExecutionStatus;
use codex_app_server_protocol::InitializeCapabilities;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::ItemCompletedNotification;
use codex_app_server_protocol::ItemStartedNotification;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::NonSteerableTurnKind;
use codex_app_server_protocol::RequestId;
@@ -71,6 +75,7 @@ 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::ThreadItem;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus;
@@ -297,6 +302,13 @@ fn sample_turn_completed_notification(
})
}
fn notification_fact(notification: ServerNotification) -> AnalyticsFact {
AnalyticsFact::Notification {
connection_id: 7,
notification: Box::new(notification),
}
}
fn sample_turn_resolved_config(turn_id: &str) -> TurnResolvedConfigFact {
TurnResolvedConfigFact {
turn_id: turn_id.to_string(),
@@ -521,9 +533,7 @@ async fn ingest_turn_prerequisites(
if include_started {
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_started_notification(
"thread-2", "turn-2",
))),
notification_fact(sample_turn_started_notification("thread-2", "turn-2")),
out,
)
.await;
@@ -541,6 +551,50 @@ async fn ingest_turn_prerequisites(
}
}
fn sample_initialize_fact(connection_id: u64) -> AnalyticsFact {
AnalyticsFact::Initialize {
connection_id,
params: InitializeParams {
client_info: ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: Some(InitializeCapabilities {
experimental_api: false,
opt_out_notification_methods: None,
}),
},
product_client_id: DEFAULT_ORIGINATOR.to_string(),
runtime: CodexRuntimeMetadata {
codex_rs_version: "0.99.0".to_string(),
runtime_os: "linux".to_string(),
runtime_os_version: "24.04".to_string(),
runtime_arch: "x86_64".to_string(),
},
rpc_transport: AppServerRpcTransport::Websocket,
}
}
fn sample_command_execution_item(
status: CommandExecutionStatus,
exit_code: Option<i32>,
duration_ms: Option<i64>,
) -> ThreadItem {
ThreadItem::CommandExecution {
id: "item-1".to_string(),
command: "echo hi".to_string(),
cwd: test_path_buf("/tmp").abs(),
process_id: Some("pid-1".to_string()),
source: CommandExecutionSource::Agent,
status,
command_actions: Vec::new(),
aggregated_output: None,
exit_code,
duration_ms,
}
}
fn expected_absolute_path(path: &PathBuf) -> String {
std::fs::canonicalize(path)
.unwrap_or_else(|_| path.to_path_buf())
@@ -1203,6 +1257,86 @@ async fn compaction_event_ingests_custom_fact() {
assert_eq!(payload[0]["event_params"]["status"], "failed");
}
#[tokio::test]
async fn item_lifecycle_notifications_publish_command_execution_event() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
reducer
.ingest(sample_initialize_fact(/*connection_id*/ 7), &mut events)
.await;
reducer
.ingest(
AnalyticsFact::Notification {
connection_id: 7,
notification: Box::new(ServerNotification::ItemStarted(ItemStartedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
item: sample_command_execution_item(
CommandExecutionStatus::InProgress,
/*exit_code*/ None,
/*duration_ms*/ None,
),
})),
},
&mut events,
)
.await;
assert!(
events.is_empty(),
"tool item event should emit on completion"
);
reducer
.ingest(
AnalyticsFact::Notification {
connection_id: 7,
notification: Box::new(ServerNotification::ItemCompleted(
ItemCompletedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
item: sample_command_execution_item(
CommandExecutionStatus::Completed,
Some(0),
Some(42),
),
},
)),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_command_execution_event");
assert_eq!(payload[0]["event_params"]["thread_id"], "thread-1");
assert_eq!(payload[0]["event_params"]["turn_id"], "turn-1");
assert_eq!(payload[0]["event_params"]["item_id"], "item-1");
assert_eq!(payload[0]["event_params"]["tool_name"], "shell");
assert_eq!(
payload[0]["event_params"]["command_execution_source"],
"agent"
);
assert_eq!(
payload[0]["event_params"]["command_execution_family"],
"shell"
);
assert_eq!(payload[0]["event_params"]["terminal_status"], "completed");
assert_eq!(
payload[0]["event_params"]["failure_kind"],
serde_json::Value::Null
);
assert_eq!(payload[0]["event_params"]["exit_code"], 0);
assert_eq!(payload[0]["event_params"]["duration_ms"], 42);
assert_eq!(payload[0]["event_params"]["execution_started"], true);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["client_name"],
"codex-tui"
);
assert_eq!(payload[0]["event_params"]["thread_source"], json!(null));
}
#[test]
fn subagent_thread_started_review_serializes_expected_shape() {
let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request(
@@ -2070,12 +2204,12 @@ async fn turn_start_error_response_discards_pending_start_request() {
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
notification_fact(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
)),
&mut out,
)
.await;
@@ -2099,12 +2233,12 @@ async fn turn_lifecycle_emits_turn_event() {
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
notification_fact(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
)),
&mut out,
)
.await;
@@ -2236,12 +2370,12 @@ async fn accepted_steers_increment_turn_steer_count() {
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
notification_fact(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
)),
&mut out,
)
.await;
@@ -2270,12 +2404,12 @@ async fn turn_does_not_emit_without_required_prerequisites() {
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
notification_fact(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
)),
&mut out,
)
.await;
@@ -2295,12 +2429,12 @@ async fn turn_does_not_emit_without_required_prerequisites() {
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
notification_fact(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
)),
&mut out,
)
.await;
@@ -2323,12 +2457,12 @@ async fn turn_lifecycle_emits_failed_turn_event() {
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
notification_fact(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Failed,
Some(codex_app_server_protocol::CodexErrorInfo::BadRequest),
))),
)),
&mut out,
)
.await;
@@ -2355,12 +2489,12 @@ async fn turn_lifecycle_emits_interrupted_turn_event_without_error() {
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
notification_fact(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Interrupted,
/*codex_error_info*/ None,
))),
)),
&mut out,
)
.await;
@@ -2387,12 +2521,12 @@ async fn turn_completed_without_started_notification_emits_null_started_at() {
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
notification_fact(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
)),
&mut out,
)
.await;

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

@@ -301,10 +301,6 @@ impl AnalyticsEventsClient {
});
}
pub fn track_notification(&self, notification: ServerNotification) {
self.record_fact(AnalyticsFact::Notification(Box::new(notification)));
}
pub fn track_server_request(&self, connection_id: u64, request: ServerRequest) {
self.record_fact(AnalyticsFact::ServerRequest {
connection_id,
@@ -317,6 +313,24 @@ impl AnalyticsEventsClient {
response: Box::new(response),
});
}
pub fn track_notification(&self, notification: ServerNotification) {
self.record_fact(AnalyticsFact::Notification {
connection_id: 0,
notification: Box::new(notification),
});
}
pub fn track_connection_notification(
&self,
connection_id: u64,
notification: ServerNotification,
) {
self.record_fact(AnalyticsFact::Notification {
connection_id,
notification: Box::new(notification),
});
}
}
async fn send_track_events(

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

@@ -295,7 +295,10 @@ pub(crate) enum AnalyticsFact {
ServerResponse {
response: Box<ServerResponse>,
},
Notification(Box<ServerNotification>),
Notification {
connection_id: u64,
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.
Custom(CustomAnalyticsFact),

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

@@ -2,15 +2,33 @@ use crate::events::AppServerRpcTransport;
use crate::events::CodexAppMentionedEventRequest;
use crate::events::CodexAppServerClientMetadata;
use crate::events::CodexAppUsedEventRequest;
use crate::events::CodexCollabAgentToolCallEventParams;
use crate::events::CodexCollabAgentToolCallEventRequest;
use crate::events::CodexCommandExecutionEventParams;
use crate::events::CodexCommandExecutionEventRequest;
use crate::events::CodexCompactionEventRequest;
use crate::events::CodexDynamicToolCallEventParams;
use crate::events::CodexDynamicToolCallEventRequest;
use crate::events::CodexFileChangeEventParams;
use crate::events::CodexFileChangeEventRequest;
use crate::events::CodexHookRunEventRequest;
use crate::events::CodexImageGenerationEventParams;
use crate::events::CodexImageGenerationEventRequest;
use crate::events::CodexMcpToolCallEventParams;
use crate::events::CodexMcpToolCallEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::CodexToolItemEventBase;
use crate::events::CodexTurnEventParams;
use crate::events::CodexTurnEventRequest;
use crate::events::CodexTurnSteerEventParams;
use crate::events::CodexTurnSteerEventRequest;
use crate::events::CodexWebSearchEventParams;
use crate::events::CodexWebSearchEventRequest;
use crate::events::CollabAgentToolKind;
use crate::events::CommandExecutionFamily;
use crate::events::CommandExecutionSourceKind;
use crate::events::GuardianReviewEventParams;
use crate::events::GuardianReviewEventPayload;
use crate::events::GuardianReviewEventRequest;
@@ -18,7 +36,11 @@ use crate::events::SkillInvocationEventParams;
use crate::events::SkillInvocationEventRequest;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::ToolItemFailureKind;
use crate::events::ToolItemFinalApprovalOutcome;
use crate::events::ToolItemTerminalStatus;
use crate::events::TrackEventRequest;
use crate::events::WebSearchActionKind;
use crate::events::codex_app_metadata;
use crate::events::codex_compaction_event_params;
use crate::events::codex_hook_run_metadata;
@@ -50,11 +72,23 @@ use crate::now_unix_seconds;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::CodexErrorInfo;
use codex_app_server_protocol::CollabAgentStatus;
use codex_app_server_protocol::CollabAgentTool;
use codex_app_server_protocol::CollabAgentToolCallStatus;
use codex_app_server_protocol::CommandExecutionSource;
use codex_app_server_protocol::CommandExecutionStatus;
use codex_app_server_protocol::DynamicToolCallOutputContentItem;
use codex_app_server_protocol::DynamicToolCallStatus;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::McpToolCallStatus;
use codex_app_server_protocol::PatchApplyStatus;
use codex_app_server_protocol::PatchChangeKind;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::TurnSteerResponse;
use codex_app_server_protocol::UserInput;
use codex_app_server_protocol::WebSearchAction;
use codex_git_utils::collect_git_info;
use codex_git_utils::get_git_repo_root;
use codex_login::default_client::originator;
@@ -68,6 +102,8 @@ use codex_protocol::protocol::TokenUsage;
use sha1::Digest;
use std::collections::HashMap;
use std::path::Path;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
#[derive(Default)]
pub(crate) struct AnalyticsReducer {
@@ -76,6 +112,7 @@ pub(crate) struct AnalyticsReducer {
connections: HashMap<u64, ConnectionState>,
thread_connections: HashMap<String, u64>,
thread_metadata: HashMap<String, ThreadMetadataState>,
tool_items: HashMap<String, ToolItemState>,
}
struct ConnectionState {
@@ -83,6 +120,11 @@ struct ConnectionState {
runtime: CodexRuntimeMetadata,
}
struct ToolItemState {
connection_id: u64,
started_at: u64,
}
#[derive(Clone)]
struct ThreadMetadataState {
thread_source: Option<&'static str>,
@@ -91,6 +133,17 @@ struct ThreadMetadataState {
parent_thread_id: Option<String>,
}
impl Default for ThreadMetadataState {
fn default() -> Self {
Self {
thread_source: None,
initialization_mode: ThreadInitializationMode::New,
subagent_source: None,
parent_thread_id: None,
}
}
}
impl ThreadMetadataState {
fn from_thread_metadata(
session_source: &SessionSource,
@@ -192,8 +245,11 @@ impl AnalyticsReducer {
} => {
self.ingest_error_response(connection_id, request_id, error_type, out);
}
AnalyticsFact::Notification(notification) => {
self.ingest_notification(*notification, out);
AnalyticsFact::Notification {
connection_id,
notification,
} => {
self.ingest_notification(connection_id, *notification, out);
}
AnalyticsFact::ServerRequest {
connection_id: _connection_id,
@@ -610,10 +666,44 @@ impl AnalyticsReducer {
fn ingest_notification(
&mut self,
connection_id: u64,
notification: ServerNotification,
out: &mut Vec<TrackEventRequest>,
) {
match notification {
ServerNotification::ItemStarted(notification) => {
if let Some(item_id) = tool_item_id(&notification.item) {
self.tool_items
.entry(item_id.to_string())
.or_insert_with(|| ToolItemState {
connection_id,
started_at: now_unix_secs(),
});
}
}
ServerNotification::ItemCompleted(notification) => {
let Some(item_id) = tool_item_id(&notification.item) else {
return;
};
let Some(started) = self.tool_items.remove(item_id) else {
return;
};
let Some(connection_state) = self.connections.get(&started.connection_id) else {
return;
};
let completed_at = now_unix_secs();
if let Some(event) = tool_item_event(
&notification.thread_id,
&notification.turn_id,
&notification.item,
started.started_at,
completed_at,
connection_state,
self.thread_metadata.get(&notification.thread_id),
) {
out.push(event);
}
}
ServerNotification::TurnStarted(notification) => {
let turn_state = self.turns.entry(notification.turn.id).or_insert(TurnState {
connection_id: None,
@@ -866,6 +956,621 @@ impl AnalyticsReducer {
}
}
fn now_unix_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_secs())
.unwrap_or_default()
}
fn tool_item_id(item: &ThreadItem) -> Option<&str> {
match item {
ThreadItem::CommandExecution { id, .. }
| ThreadItem::FileChange { id, .. }
| ThreadItem::McpToolCall { id, .. }
| ThreadItem::DynamicToolCall { id, .. }
| ThreadItem::CollabAgentToolCall { id, .. }
| ThreadItem::WebSearch { id, .. }
| ThreadItem::ImageGeneration { id, .. } => Some(id),
_ => None,
}
}
fn tool_item_event(
thread_id: &str,
turn_id: &str,
item: &ThreadItem,
started_at: u64,
completed_at: u64,
connection_state: &ConnectionState,
thread_metadata: Option<&ThreadMetadataState>,
) -> Option<TrackEventRequest> {
match item {
ThreadItem::CommandExecution {
id,
source,
status,
command_actions,
exit_code,
duration_ms,
..
} => {
let (terminal_status, failure_kind) = command_execution_outcome(status)?;
let base = tool_item_base(
thread_id,
turn_id,
id.clone(),
command_execution_tool_name(*source).to_string(),
ToolItemOutcome {
terminal_status,
failure_kind,
duration_ms: completed_duration_ms(
option_i64_to_u64(*duration_ms),
started_at,
completed_at,
),
},
ToolItemContext {
started_at,
completed_at,
connection_state,
thread_metadata,
},
);
Some(TrackEventRequest::CommandExecution(
CodexCommandExecutionEventRequest {
event_type: "codex_command_execution_event",
event_params: CodexCommandExecutionEventParams {
base,
command_execution_source: command_execution_source_kind(*source),
command_execution_family: command_execution_family(*source),
exit_code: *exit_code,
command_action_count: Some(usize_to_u64(command_actions.len())),
},
},
))
}
ThreadItem::FileChange {
id,
changes,
status,
} => {
let (terminal_status, failure_kind) = patch_apply_outcome(status)?;
let counts = file_change_counts(changes);
let base = tool_item_base(
thread_id,
turn_id,
id.clone(),
"apply_patch".to_string(),
ToolItemOutcome {
terminal_status,
failure_kind,
duration_ms: observed_completed_duration_ms(started_at, completed_at),
},
ToolItemContext {
started_at,
completed_at,
connection_state,
thread_metadata,
},
);
Some(TrackEventRequest::FileChange(CodexFileChangeEventRequest {
event_type: "codex_file_change_event",
event_params: CodexFileChangeEventParams {
base,
file_change_count: usize_to_u64(changes.len()),
file_add_count: counts.add,
file_update_count: counts.update,
file_delete_count: counts.delete,
file_move_count: counts.move_,
},
}))
}
ThreadItem::McpToolCall {
id,
server,
tool,
status,
error,
duration_ms,
..
} => {
let (terminal_status, failure_kind) = mcp_tool_call_outcome(status)?;
let base = tool_item_base(
thread_id,
turn_id,
id.clone(),
tool.clone(),
ToolItemOutcome {
terminal_status,
failure_kind,
duration_ms: completed_duration_ms(
option_i64_to_u64(*duration_ms),
started_at,
completed_at,
),
},
ToolItemContext {
started_at,
completed_at,
connection_state,
thread_metadata,
},
);
Some(TrackEventRequest::McpToolCall(
CodexMcpToolCallEventRequest {
event_type: "codex_mcp_tool_call_event",
event_params: CodexMcpToolCallEventParams {
base,
mcp_server_name: server.clone(),
mcp_tool_name: tool.clone(),
mcp_error_present: error.is_some(),
mcp_error_code: None,
},
},
))
}
ThreadItem::DynamicToolCall {
id,
tool,
status,
content_items,
success,
duration_ms,
..
} => {
let (terminal_status, failure_kind) = dynamic_tool_call_outcome(status)?;
let counts = content_items
.as_ref()
.map(|items| dynamic_content_counts(items));
let base = tool_item_base(
thread_id,
turn_id,
id.clone(),
tool.clone(),
ToolItemOutcome {
terminal_status,
failure_kind,
duration_ms: completed_duration_ms(
option_i64_to_u64(*duration_ms),
started_at,
completed_at,
),
},
ToolItemContext {
started_at,
completed_at,
connection_state,
thread_metadata,
},
);
Some(TrackEventRequest::DynamicToolCall(
CodexDynamicToolCallEventRequest {
event_type: "codex_dynamic_tool_call_event",
event_params: CodexDynamicToolCallEventParams {
base,
dynamic_tool_name: tool.clone(),
success: *success,
output_content_item_count: counts.map(|counts| counts.total),
output_text_item_count: counts.map(|counts| counts.text),
output_image_item_count: counts.map(|counts| counts.image),
},
},
))
}
ThreadItem::CollabAgentToolCall {
id,
tool,
status,
sender_thread_id,
receiver_thread_ids,
model,
reasoning_effort,
agents_states,
..
} => {
let (terminal_status, failure_kind) = collab_tool_call_outcome(status)?;
let base = tool_item_base(
thread_id,
turn_id,
id.clone(),
collab_agent_tool_name(tool).to_string(),
ToolItemOutcome {
terminal_status,
failure_kind,
duration_ms: observed_completed_duration_ms(started_at, completed_at),
},
ToolItemContext {
started_at,
completed_at,
connection_state,
thread_metadata,
},
);
Some(TrackEventRequest::CollabAgentToolCall(
CodexCollabAgentToolCallEventRequest {
event_type: "codex_collab_agent_tool_call_event",
event_params: CodexCollabAgentToolCallEventParams {
base,
collab_agent_tool: collab_agent_tool_kind(tool),
sender_thread_id: sender_thread_id.clone(),
receiver_thread_count: usize_to_u64(receiver_thread_ids.len()),
receiver_thread_ids: Some(receiver_thread_ids.clone()),
requested_model: model.clone(),
requested_reasoning_effort: reasoning_effort
.as_ref()
.and_then(serialize_enum_as_string),
agent_state_count: Some(usize_to_u64(agents_states.len())),
completed_agent_count: Some(usize_to_u64(
agents_states
.values()
.filter(|state| state.status == CollabAgentStatus::Completed)
.count(),
)),
failed_agent_count: Some(usize_to_u64(
agents_states
.values()
.filter(|state| {
matches!(
state.status,
CollabAgentStatus::Errored
| CollabAgentStatus::Shutdown
| CollabAgentStatus::NotFound
)
})
.count(),
)),
},
},
))
}
ThreadItem::WebSearch { id, query, action } => {
let base = tool_item_base(
thread_id,
turn_id,
id.clone(),
"web_search".to_string(),
ToolItemOutcome {
terminal_status: ToolItemTerminalStatus::Completed,
failure_kind: None,
duration_ms: observed_completed_duration_ms(started_at, completed_at),
},
ToolItemContext {
started_at,
completed_at,
connection_state,
thread_metadata,
},
);
Some(TrackEventRequest::WebSearch(CodexWebSearchEventRequest {
event_type: "codex_web_search_event",
event_params: CodexWebSearchEventParams {
base,
web_search_action: action.as_ref().map(web_search_action_kind),
query_present: !query.trim().is_empty(),
query_count: web_search_query_count(query, action.as_ref()),
},
}))
}
ThreadItem::ImageGeneration {
id,
status,
revised_prompt,
saved_path,
..
} => {
let (terminal_status, failure_kind) = image_generation_outcome(status.as_str());
let base = tool_item_base(
thread_id,
turn_id,
id.clone(),
"image_generation".to_string(),
ToolItemOutcome {
terminal_status,
failure_kind,
duration_ms: observed_completed_duration_ms(started_at, completed_at),
},
ToolItemContext {
started_at,
completed_at,
connection_state,
thread_metadata,
},
);
Some(TrackEventRequest::ImageGeneration(
CodexImageGenerationEventRequest {
event_type: "codex_image_generation_event",
event_params: CodexImageGenerationEventParams {
base,
image_generation_status: status.clone(),
revised_prompt_present: revised_prompt.is_some(),
saved_path_present: saved_path.is_some(),
},
},
))
}
_ => None,
}
}
struct ToolItemOutcome {
terminal_status: ToolItemTerminalStatus,
failure_kind: Option<ToolItemFailureKind>,
duration_ms: Option<u64>,
}
struct ToolItemContext<'a> {
started_at: u64,
completed_at: u64,
connection_state: &'a ConnectionState,
thread_metadata: Option<&'a ThreadMetadataState>,
}
fn tool_item_base(
thread_id: &str,
turn_id: &str,
item_id: String,
tool_name: String,
outcome: ToolItemOutcome,
context: ToolItemContext<'_>,
) -> CodexToolItemEventBase {
let thread_metadata = context.thread_metadata.cloned().unwrap_or_default();
CodexToolItemEventBase {
thread_id: thread_id.to_string(),
turn_id: turn_id.to_string(),
item_id,
app_server_client: context.connection_state.app_server_client.clone(),
runtime: context.connection_state.runtime.clone(),
thread_source: thread_metadata.thread_source,
subagent_source: thread_metadata.subagent_source,
parent_thread_id: thread_metadata.parent_thread_id,
tool_name,
started_at: context.started_at,
completed_at: Some(context.completed_at),
duration_ms: outcome.duration_ms,
execution_started: true,
review_count: 0,
guardian_review_count: 0,
user_review_count: 0,
final_approval_outcome: ToolItemFinalApprovalOutcome::NotNeeded,
terminal_status: outcome.terminal_status,
failure_kind: outcome.failure_kind,
requested_additional_permissions: false,
requested_network_access: false,
retry_count: 0,
}
}
fn completed_duration_ms(
item_duration_ms: Option<u64>,
started_at: u64,
completed_at: u64,
) -> Option<u64> {
item_duration_ms.or_else(|| {
completed_at
.checked_sub(started_at)
.map(|duration_secs| duration_secs.saturating_mul(1000))
})
}
fn observed_completed_duration_ms(started_at: u64, completed_at: u64) -> Option<u64> {
completed_duration_ms(/*item_duration_ms*/ None, started_at, completed_at)
}
fn command_execution_source_kind(source: CommandExecutionSource) -> CommandExecutionSourceKind {
match source {
CommandExecutionSource::Agent => CommandExecutionSourceKind::Agent,
CommandExecutionSource::UserShell => CommandExecutionSourceKind::UserShell,
CommandExecutionSource::UnifiedExecStartup => {
CommandExecutionSourceKind::UnifiedExecStartup
}
CommandExecutionSource::UnifiedExecInteraction => {
CommandExecutionSourceKind::UnifiedExecInteraction
}
}
}
fn command_execution_family(source: CommandExecutionSource) -> CommandExecutionFamily {
match source {
CommandExecutionSource::Agent => CommandExecutionFamily::Shell,
CommandExecutionSource::UserShell => CommandExecutionFamily::UserShell,
CommandExecutionSource::UnifiedExecStartup
| CommandExecutionSource::UnifiedExecInteraction => CommandExecutionFamily::UnifiedExec,
}
}
fn command_execution_tool_name(source: CommandExecutionSource) -> &'static str {
match source {
CommandExecutionSource::UnifiedExecStartup
| CommandExecutionSource::UnifiedExecInteraction => "unified_exec",
CommandExecutionSource::UserShell => "user_shell",
CommandExecutionSource::Agent => "shell",
}
}
fn command_execution_outcome(
status: &CommandExecutionStatus,
) -> Option<(ToolItemTerminalStatus, Option<ToolItemFailureKind>)> {
match status {
CommandExecutionStatus::InProgress => None,
CommandExecutionStatus::Completed => Some((ToolItemTerminalStatus::Completed, None)),
CommandExecutionStatus::Failed => Some((
ToolItemTerminalStatus::Failed,
Some(ToolItemFailureKind::ToolError),
)),
CommandExecutionStatus::Declined => Some((
ToolItemTerminalStatus::Rejected,
Some(ToolItemFailureKind::ApprovalDenied),
)),
}
}
fn patch_apply_outcome(
status: &PatchApplyStatus,
) -> Option<(ToolItemTerminalStatus, Option<ToolItemFailureKind>)> {
match status {
PatchApplyStatus::InProgress => None,
PatchApplyStatus::Completed => Some((ToolItemTerminalStatus::Completed, None)),
PatchApplyStatus::Failed => Some((
ToolItemTerminalStatus::Failed,
Some(ToolItemFailureKind::ToolError),
)),
PatchApplyStatus::Declined => Some((
ToolItemTerminalStatus::Rejected,
Some(ToolItemFailureKind::ApprovalDenied),
)),
}
}
fn mcp_tool_call_outcome(
status: &McpToolCallStatus,
) -> Option<(ToolItemTerminalStatus, Option<ToolItemFailureKind>)> {
match status {
McpToolCallStatus::InProgress => None,
McpToolCallStatus::Completed => Some((ToolItemTerminalStatus::Completed, None)),
McpToolCallStatus::Failed => Some((
ToolItemTerminalStatus::Failed,
Some(ToolItemFailureKind::ToolError),
)),
}
}
fn dynamic_tool_call_outcome(
status: &DynamicToolCallStatus,
) -> Option<(ToolItemTerminalStatus, Option<ToolItemFailureKind>)> {
match status {
DynamicToolCallStatus::InProgress => None,
DynamicToolCallStatus::Completed => Some((ToolItemTerminalStatus::Completed, None)),
DynamicToolCallStatus::Failed => Some((
ToolItemTerminalStatus::Failed,
Some(ToolItemFailureKind::ToolError),
)),
}
}
fn collab_tool_call_outcome(
status: &CollabAgentToolCallStatus,
) -> Option<(ToolItemTerminalStatus, Option<ToolItemFailureKind>)> {
match status {
CollabAgentToolCallStatus::InProgress => None,
CollabAgentToolCallStatus::Completed => Some((ToolItemTerminalStatus::Completed, None)),
CollabAgentToolCallStatus::Failed => Some((
ToolItemTerminalStatus::Failed,
Some(ToolItemFailureKind::ToolError),
)),
}
}
fn image_generation_outcome(status: &str) -> (ToolItemTerminalStatus, Option<ToolItemFailureKind>) {
match status {
"failed" | "error" => (
ToolItemTerminalStatus::Failed,
Some(ToolItemFailureKind::ToolError),
),
_ => (ToolItemTerminalStatus::Completed, None),
}
}
fn collab_agent_tool_name(tool: &CollabAgentTool) -> &'static str {
match tool {
CollabAgentTool::SpawnAgent => "spawn_agent",
CollabAgentTool::SendInput => "send_input",
CollabAgentTool::ResumeAgent => "resume_agent",
CollabAgentTool::Wait => "wait_agent",
CollabAgentTool::CloseAgent => "close_agent",
}
}
fn collab_agent_tool_kind(tool: &CollabAgentTool) -> CollabAgentToolKind {
match tool {
CollabAgentTool::SpawnAgent => CollabAgentToolKind::SpawnAgent,
CollabAgentTool::SendInput => CollabAgentToolKind::SendInput,
CollabAgentTool::ResumeAgent => CollabAgentToolKind::ResumeAgent,
CollabAgentTool::Wait => CollabAgentToolKind::Wait,
CollabAgentTool::CloseAgent => CollabAgentToolKind::CloseAgent,
}
}
#[derive(Default)]
struct FileChangeCounts {
add: u64,
update: u64,
delete: u64,
move_: u64,
}
fn file_change_counts(changes: &[codex_app_server_protocol::FileUpdateChange]) -> FileChangeCounts {
let mut counts = FileChangeCounts::default();
for change in changes {
match &change.kind {
PatchChangeKind::Add => counts.add += 1,
PatchChangeKind::Delete => counts.delete += 1,
PatchChangeKind::Update { move_path: Some(_) } => counts.move_ += 1,
PatchChangeKind::Update { move_path: None } => counts.update += 1,
}
}
counts
}
#[derive(Clone, Copy)]
struct DynamicContentCounts {
total: u64,
text: u64,
image: u64,
}
fn dynamic_content_counts(items: &[DynamicToolCallOutputContentItem]) -> DynamicContentCounts {
let mut text = 0;
let mut image = 0;
for item in items {
match item {
DynamicToolCallOutputContentItem::InputText { .. } => text += 1,
DynamicToolCallOutputContentItem::InputImage { .. } => image += 1,
}
}
DynamicContentCounts {
total: usize_to_u64(items.len()),
text,
image,
}
}
fn web_search_action_kind(action: &WebSearchAction) -> WebSearchActionKind {
match action {
WebSearchAction::Search { .. } => WebSearchActionKind::Search,
WebSearchAction::OpenPage { .. } => WebSearchActionKind::OpenPage,
WebSearchAction::FindInPage { .. } => WebSearchActionKind::FindInPage,
WebSearchAction::Other => WebSearchActionKind::Other,
}
}
fn web_search_query_count(query: &str, action: Option<&WebSearchAction>) -> Option<u64> {
match action {
Some(WebSearchAction::Search { query, queries }) => queries
.as_ref()
.map(|queries| usize_to_u64(queries.len()))
.or_else(|| query.as_ref().map(|_| 1)),
Some(WebSearchAction::OpenPage { .. })
| Some(WebSearchAction::FindInPage { .. })
| Some(WebSearchAction::Other) => None,
None => (!query.trim().is_empty()).then_some(1),
}
}
fn serialize_enum_as_string<T: serde::Serialize>(value: &T) -> Option<String> {
serde_json::to_value(value)
.ok()
.and_then(|value| value.as_str().map(str::to_string))
}
fn usize_to_u64(value: usize) -> u64 {
u64::try_from(value).unwrap_or(u64::MAX)
}
fn option_i64_to_u64(value: Option<i64>) -> Option<u64> {
value.and_then(|value| u64::try_from(value).ok())
}
fn codex_turn_event_params(
app_server_client: CodexAppServerClientMetadata,
runtime: CodexRuntimeMetadata,

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

@@ -532,7 +532,7 @@ impl OutgoingMessageSender {
targeted_connections = connection_ids.len(),
"app-server event: {notification}"
);
let outgoing_message = OutgoingMessage::AppServerNotification(notification);
let outgoing_message = OutgoingMessage::AppServerNotification(notification.clone());
if connection_ids.is_empty() {
if let Err(err) = self
.sender
@@ -556,6 +556,9 @@ impl OutgoingMessageSender {
.await
{
warn!("failed to send server notification to client: {err:?}");
} else {
self.analytics_events_client
.track_connection_notification(connection_id.0, notification.clone());
}
}
}
@@ -566,7 +569,7 @@ impl OutgoingMessageSender {
notification: ServerNotification,
) {
tracing::trace!("app-server event: {notification}");
let outgoing_message = OutgoingMessage::AppServerNotification(notification);
let outgoing_message = OutgoingMessage::AppServerNotification(notification.clone());
let (write_complete_tx, write_complete_rx) = oneshot::channel();
if let Err(err) = self
.sender
@@ -578,6 +581,9 @@ impl OutgoingMessageSender {
.await
{
warn!("failed to send server notification to client: {err:?}");
} else {
self.analytics_events_client
.track_connection_notification(connection_id.0, notification);
}
let _ = write_complete_rx.await;
}