mirror of
https://github.com/openai/codex.git
synced 2026-04-08 14:54:47 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7b338d11db | ||
|
|
7806c42750 | ||
|
|
d02455011f |
@@ -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;
|
||||
@@ -34,12 +39,18 @@ 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::ClientResponse;
|
||||
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::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::ThreadItem;
|
||||
use codex_app_server_protocol::ThreadResumeResponse;
|
||||
use codex_app_server_protocol::ThreadStartResponse;
|
||||
use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus;
|
||||
@@ -114,6 +125,50 @@ fn sample_thread_resume_response(thread_id: &str, ephemeral: bool, model: &str)
|
||||
}
|
||||
}
|
||||
|
||||
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: "tool-call-1".to_string(),
|
||||
command: "echo hi".to_string(),
|
||||
cwd: PathBuf::from("/tmp"),
|
||||
process_id: None,
|
||||
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())
|
||||
@@ -346,6 +401,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();
|
||||
@@ -449,6 +587,75 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize
|
||||
assert_eq!(payload[0]["event_params"]["parent_thread_id"], json!(null));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn item_lifecycle_notifications_publish_basic_tool_call_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 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_tool_call_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"]["tool_call_id"], "tool-call-1");
|
||||
assert_eq!(payload[0]["event_params"]["tool_name"], "shell");
|
||||
assert_eq!(payload[0]["event_params"]["tool_kind"], "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"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn subagent_thread_started_review_serializes_expected_shape() {
|
||||
let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request(
|
||||
|
||||
@@ -16,6 +16,9 @@ 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::ServerNotification;
|
||||
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 +230,26 @@ 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),
|
||||
});
|
||||
}
|
||||
|
||||
pub fn track_notification(&self, connection_id: u64, notification: ServerNotification) {
|
||||
self.record_fact(AnalyticsFact::Notification {
|
||||
connection_id,
|
||||
notification: Box::new(notification),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async fn send_track_events(
|
||||
|
||||
@@ -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,84 @@ 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")]
|
||||
#[allow(dead_code)]
|
||||
pub(crate) enum ToolCallFinalReviewOutcome {
|
||||
NotNeeded,
|
||||
GuardianApproved,
|
||||
GuardianDenied,
|
||||
GuardianAborted,
|
||||
UserApproved,
|
||||
UserApprovedForSession,
|
||||
UserDenied,
|
||||
UserAborted,
|
||||
ConfigAllowed,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Serialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
#[allow(dead_code)]
|
||||
pub(crate) enum ToolCallTerminalStatus {
|
||||
Completed,
|
||||
Failed,
|
||||
Rejected,
|
||||
Interrupted,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Serialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
#[allow(dead_code)]
|
||||
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>,
|
||||
|
||||
@@ -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,7 +83,17 @@ pub(crate) enum AnalyticsFact {
|
||||
connection_id: u64,
|
||||
response: Box<ClientResponse>,
|
||||
},
|
||||
Notification(Box<ServerNotification>),
|
||||
ServerRequest {
|
||||
connection_id: u64,
|
||||
request: Box<ServerRequest>,
|
||||
},
|
||||
ServerResponse {
|
||||
response: Box<ServerResponse>,
|
||||
},
|
||||
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),
|
||||
|
||||
@@ -5,11 +5,17 @@ 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::SkillInvocationEventParams;
|
||||
use crate::events::SkillInvocationEventRequest;
|
||||
use crate::events::ThreadInitializationMode;
|
||||
use crate::events::ThreadInitializedEvent;
|
||||
use crate::events::ThreadInitializedEventParams;
|
||||
use crate::events::ToolCallFailureKind;
|
||||
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;
|
||||
@@ -27,7 +33,16 @@ use crate::facts::PluginUsedInput;
|
||||
use crate::facts::SkillInvokedInput;
|
||||
use crate::facts::SubAgentThreadStartedInput;
|
||||
use codex_app_server_protocol::ClientResponse;
|
||||
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::DynamicToolCallStatus;
|
||||
use codex_app_server_protocol::InitializeParams;
|
||||
use codex_app_server_protocol::McpToolCallStatus;
|
||||
use codex_app_server_protocol::PatchApplyStatus;
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
use codex_app_server_protocol::ThreadItem;
|
||||
use codex_git_utils::collect_git_info;
|
||||
use codex_git_utils::get_git_repo_root;
|
||||
use codex_login::default_client::originator;
|
||||
@@ -36,10 +51,13 @@ use codex_protocol::protocol::SkillScope;
|
||||
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 {
|
||||
connections: HashMap<u64, ConnectionState>,
|
||||
tool_calls: HashMap<String, ToolCallState>,
|
||||
}
|
||||
|
||||
struct ConnectionState {
|
||||
@@ -47,6 +65,21 @@ struct ConnectionState {
|
||||
runtime: CodexRuntimeMetadata,
|
||||
}
|
||||
|
||||
struct ToolCallState {
|
||||
connection_id: u64,
|
||||
started_at: u64,
|
||||
}
|
||||
|
||||
struct ToolCallCompletedMetadata {
|
||||
tool_call_id: String,
|
||||
tool_name: String,
|
||||
tool_kind: ToolKind,
|
||||
terminal_status: ToolCallTerminalStatus,
|
||||
failure_kind: Option<ToolCallFailureKind>,
|
||||
exit_code: Option<i32>,
|
||||
duration_ms: Option<u64>,
|
||||
}
|
||||
|
||||
impl AnalyticsReducer {
|
||||
pub(crate) async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec<TrackEventRequest>) {
|
||||
match input {
|
||||
@@ -76,7 +109,19 @@ impl AnalyticsReducer {
|
||||
} => {
|
||||
self.ingest_response(connection_id, *response, out);
|
||||
}
|
||||
AnalyticsFact::Notification(_notification) => {}
|
||||
AnalyticsFact::ServerRequest {
|
||||
connection_id: _connection_id,
|
||||
request: _request,
|
||||
} => {}
|
||||
AnalyticsFact::ServerResponse {
|
||||
response: _response,
|
||||
} => {}
|
||||
AnalyticsFact::Notification {
|
||||
connection_id,
|
||||
notification,
|
||||
} => {
|
||||
self.ingest_notification(connection_id, *notification, out);
|
||||
}
|
||||
AnalyticsFact::Custom(input) => match input {
|
||||
CustomAnalyticsFact::SubAgentThreadStarted(input) => {
|
||||
self.ingest_subagent_thread_started(input, out);
|
||||
@@ -275,6 +320,315 @@ impl AnalyticsReducer {
|
||||
},
|
||||
));
|
||||
}
|
||||
|
||||
fn ingest_notification(
|
||||
&mut self,
|
||||
connection_id: u64,
|
||||
notification: ServerNotification,
|
||||
out: &mut Vec<TrackEventRequest>,
|
||||
) {
|
||||
match notification {
|
||||
ServerNotification::ItemStarted(notification) => {
|
||||
if let Some(tool_call_id) = tool_call_id(¬ification.item) {
|
||||
self.tool_calls
|
||||
.entry(tool_call_id.to_string())
|
||||
.or_insert_with(|| ToolCallState {
|
||||
connection_id,
|
||||
started_at: now_unix_secs(),
|
||||
});
|
||||
}
|
||||
}
|
||||
ServerNotification::ItemCompleted(notification) => {
|
||||
let Some(completed_metadata) = tool_call_completed_metadata(¬ification.item)
|
||||
else {
|
||||
return;
|
||||
};
|
||||
let Some(started) = self.tool_calls.remove(&completed_metadata.tool_call_id) else {
|
||||
return;
|
||||
};
|
||||
let Some(connection_state) = self.connections.get(&started.connection_id) else {
|
||||
return;
|
||||
};
|
||||
let completed_at = now_unix_secs();
|
||||
let duration_ms = completed_metadata.duration_ms.or_else(|| {
|
||||
completed_at
|
||||
.checked_sub(started.started_at)
|
||||
.map(|duration_secs| duration_secs.saturating_mul(1000))
|
||||
});
|
||||
|
||||
out.push(TrackEventRequest::ToolCall(CodexToolCallEventRequest {
|
||||
event_type: "codex_tool_call_event",
|
||||
event_params: CodexToolCallEventParams {
|
||||
thread_id: notification.thread_id,
|
||||
turn_id: notification.turn_id,
|
||||
tool_call_id: completed_metadata.tool_call_id,
|
||||
app_server_client: connection_state.app_server_client.clone(),
|
||||
runtime: connection_state.runtime.clone(),
|
||||
tool_name: completed_metadata.tool_name,
|
||||
tool_kind: completed_metadata.tool_kind,
|
||||
started_at: started.started_at,
|
||||
completed_at: Some(completed_at),
|
||||
duration_ms,
|
||||
execution_started: true,
|
||||
review_count: 0,
|
||||
guardian_review_count: 0,
|
||||
user_review_count: 0,
|
||||
final_review_outcome: ToolCallFinalReviewOutcome::NotNeeded,
|
||||
terminal_status: completed_metadata.terminal_status,
|
||||
failure_kind: completed_metadata.failure_kind,
|
||||
exit_code: completed_metadata.exit_code,
|
||||
requested_additional_permissions: false,
|
||||
requested_network_access: false,
|
||||
retry_count: 0,
|
||||
},
|
||||
}));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn now_unix_secs() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|duration| duration.as_secs())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
fn tool_call_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_call_completed_metadata(item: &ThreadItem) -> Option<ToolCallCompletedMetadata> {
|
||||
match item {
|
||||
ThreadItem::CommandExecution {
|
||||
id,
|
||||
source,
|
||||
status,
|
||||
exit_code,
|
||||
duration_ms,
|
||||
..
|
||||
} => {
|
||||
let (terminal_status, failure_kind) = command_execution_outcome(status)?;
|
||||
let tool_kind = command_execution_tool_kind(source);
|
||||
Some(ToolCallCompletedMetadata {
|
||||
tool_call_id: id.clone(),
|
||||
tool_name: command_execution_tool_name(source).to_string(),
|
||||
tool_kind,
|
||||
terminal_status,
|
||||
failure_kind,
|
||||
exit_code: *exit_code,
|
||||
duration_ms: option_i64_to_u64(*duration_ms),
|
||||
})
|
||||
}
|
||||
ThreadItem::FileChange { id, status, .. } => {
|
||||
let (terminal_status, failure_kind) = patch_apply_outcome(status)?;
|
||||
Some(ToolCallCompletedMetadata {
|
||||
tool_call_id: id.clone(),
|
||||
tool_name: "apply_patch".to_string(),
|
||||
tool_kind: ToolKind::ApplyPatch,
|
||||
terminal_status,
|
||||
failure_kind,
|
||||
exit_code: None,
|
||||
duration_ms: None,
|
||||
})
|
||||
}
|
||||
ThreadItem::McpToolCall {
|
||||
id,
|
||||
tool,
|
||||
status,
|
||||
duration_ms,
|
||||
..
|
||||
} => {
|
||||
let (terminal_status, failure_kind) = mcp_tool_call_outcome(status)?;
|
||||
Some(ToolCallCompletedMetadata {
|
||||
tool_call_id: id.clone(),
|
||||
tool_name: tool.clone(),
|
||||
tool_kind: ToolKind::Mcp,
|
||||
terminal_status,
|
||||
failure_kind,
|
||||
exit_code: None,
|
||||
duration_ms: option_i64_to_u64(*duration_ms),
|
||||
})
|
||||
}
|
||||
ThreadItem::DynamicToolCall {
|
||||
id,
|
||||
tool,
|
||||
status,
|
||||
duration_ms,
|
||||
..
|
||||
} => {
|
||||
let (terminal_status, failure_kind) = dynamic_tool_call_outcome(status)?;
|
||||
Some(ToolCallCompletedMetadata {
|
||||
tool_call_id: id.clone(),
|
||||
tool_name: tool.clone(),
|
||||
tool_kind: ToolKind::Dynamic,
|
||||
terminal_status,
|
||||
failure_kind,
|
||||
exit_code: None,
|
||||
duration_ms: option_i64_to_u64(*duration_ms),
|
||||
})
|
||||
}
|
||||
ThreadItem::CollabAgentToolCall {
|
||||
id, tool, status, ..
|
||||
} => {
|
||||
let (terminal_status, failure_kind) = collab_tool_call_outcome(status)?;
|
||||
Some(ToolCallCompletedMetadata {
|
||||
tool_call_id: id.clone(),
|
||||
tool_name: collab_agent_tool_name(tool).to_string(),
|
||||
tool_kind: ToolKind::Other,
|
||||
terminal_status,
|
||||
failure_kind,
|
||||
exit_code: None,
|
||||
duration_ms: None,
|
||||
})
|
||||
}
|
||||
ThreadItem::WebSearch { id, .. } => Some(ToolCallCompletedMetadata {
|
||||
tool_call_id: id.clone(),
|
||||
tool_name: "web_search".to_string(),
|
||||
tool_kind: ToolKind::Other,
|
||||
terminal_status: ToolCallTerminalStatus::Completed,
|
||||
failure_kind: None,
|
||||
exit_code: None,
|
||||
duration_ms: None,
|
||||
}),
|
||||
ThreadItem::ImageGeneration { id, status, .. } => {
|
||||
let (terminal_status, failure_kind) = image_generation_outcome(status.as_str());
|
||||
Some(ToolCallCompletedMetadata {
|
||||
tool_call_id: id.clone(),
|
||||
tool_name: "image_generation".to_string(),
|
||||
tool_kind: ToolKind::Other,
|
||||
terminal_status,
|
||||
failure_kind,
|
||||
exit_code: None,
|
||||
duration_ms: None,
|
||||
})
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn command_execution_tool_kind(source: &CommandExecutionSource) -> ToolKind {
|
||||
match source {
|
||||
CommandExecutionSource::UnifiedExecStartup
|
||||
| CommandExecutionSource::UnifiedExecInteraction => ToolKind::UnifiedExec,
|
||||
CommandExecutionSource::Agent | CommandExecutionSource::UserShell => ToolKind::Shell,
|
||||
}
|
||||
}
|
||||
|
||||
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<(ToolCallTerminalStatus, Option<ToolCallFailureKind>)> {
|
||||
match status {
|
||||
CommandExecutionStatus::InProgress => None,
|
||||
CommandExecutionStatus::Completed => Some((ToolCallTerminalStatus::Completed, None)),
|
||||
CommandExecutionStatus::Failed => Some((
|
||||
ToolCallTerminalStatus::Failed,
|
||||
Some(ToolCallFailureKind::ToolError),
|
||||
)),
|
||||
CommandExecutionStatus::Declined => Some((
|
||||
ToolCallTerminalStatus::Rejected,
|
||||
Some(ToolCallFailureKind::ApprovalDenied),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn patch_apply_outcome(
|
||||
status: &PatchApplyStatus,
|
||||
) -> Option<(ToolCallTerminalStatus, Option<ToolCallFailureKind>)> {
|
||||
match status {
|
||||
PatchApplyStatus::InProgress => None,
|
||||
PatchApplyStatus::Completed => Some((ToolCallTerminalStatus::Completed, None)),
|
||||
PatchApplyStatus::Failed => Some((
|
||||
ToolCallTerminalStatus::Failed,
|
||||
Some(ToolCallFailureKind::ToolError),
|
||||
)),
|
||||
PatchApplyStatus::Declined => Some((
|
||||
ToolCallTerminalStatus::Rejected,
|
||||
Some(ToolCallFailureKind::ApprovalDenied),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn mcp_tool_call_outcome(
|
||||
status: &McpToolCallStatus,
|
||||
) -> Option<(ToolCallTerminalStatus, Option<ToolCallFailureKind>)> {
|
||||
match status {
|
||||
McpToolCallStatus::InProgress => None,
|
||||
McpToolCallStatus::Completed => Some((ToolCallTerminalStatus::Completed, None)),
|
||||
McpToolCallStatus::Failed => Some((
|
||||
ToolCallTerminalStatus::Failed,
|
||||
Some(ToolCallFailureKind::ToolError),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn dynamic_tool_call_outcome(
|
||||
status: &DynamicToolCallStatus,
|
||||
) -> Option<(ToolCallTerminalStatus, Option<ToolCallFailureKind>)> {
|
||||
match status {
|
||||
DynamicToolCallStatus::InProgress => None,
|
||||
DynamicToolCallStatus::Completed => Some((ToolCallTerminalStatus::Completed, None)),
|
||||
DynamicToolCallStatus::Failed => Some((
|
||||
ToolCallTerminalStatus::Failed,
|
||||
Some(ToolCallFailureKind::ToolError),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn collab_tool_call_outcome(
|
||||
status: &CollabAgentToolCallStatus,
|
||||
) -> Option<(ToolCallTerminalStatus, Option<ToolCallFailureKind>)> {
|
||||
match status {
|
||||
CollabAgentToolCallStatus::InProgress => None,
|
||||
CollabAgentToolCallStatus::Completed => Some((ToolCallTerminalStatus::Completed, None)),
|
||||
CollabAgentToolCallStatus::Failed => Some((
|
||||
ToolCallTerminalStatus::Failed,
|
||||
Some(ToolCallFailureKind::ToolError),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn image_generation_outcome(status: &str) -> (ToolCallTerminalStatus, Option<ToolCallFailureKind>) {
|
||||
match status {
|
||||
"failed" | "error" => (
|
||||
ToolCallTerminalStatus::Failed,
|
||||
Some(ToolCallFailureKind::ToolError),
|
||||
),
|
||||
_ => (ToolCallTerminalStatus::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 option_i64_to_u64(value: Option<i64>) -> Option<u64> {
|
||||
value.and_then(|value| u64::try_from(value).ok())
|
||||
}
|
||||
|
||||
pub(crate) fn skill_id_for_local_skill(
|
||||
|
||||
@@ -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());
|
||||
|
||||
@@ -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:?}");
|
||||
}
|
||||
@@ -518,7 +549,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
|
||||
@@ -542,6 +573,8 @@ impl OutgoingMessageSender {
|
||||
.await
|
||||
{
|
||||
warn!("failed to send server notification to client: {err:?}");
|
||||
} else if let Some(analytics_events_client) = self.analytics_events_client() {
|
||||
analytics_events_client.track_notification(connection_id.0, notification.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -552,7 +585,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
|
||||
@@ -564,6 +597,8 @@ impl OutgoingMessageSender {
|
||||
.await
|
||||
{
|
||||
warn!("failed to send server notification to client: {err:?}");
|
||||
} else if let Some(analytics_events_client) = self.analytics_events_client() {
|
||||
analytics_events_client.track_notification(connection_id.0, notification);
|
||||
}
|
||||
let _ = write_complete_rx.await;
|
||||
}
|
||||
@@ -621,6 +656,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 +697,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 +883,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);
|
||||
|
||||
Reference in New Issue
Block a user