clone/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-04-24 22:54:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

support app usage analytics (#11687)

Browse Source 
Emit app mentioned and app used events. Dedup by (turn_id, connector_id)

Example event params:
{
    "event_type": "codex_app_used",
    "connector_id": "asdk_app_xxx",
    "thread_id": "019c5527-36d4-xxx",
    "turn_id": "019c552c-cd17-xxx",
    "app_name": "Slack (OpenAI Internal)",
    "product_client_id": "codex_cli_rs",
    "invoke_type": "explicit",
    "model_slug": "gpt-5.3-codex"
}
This commit is contained in:
alexsong-oai
2026-02-13 12:00:16 -08:00
committed by GitHub
parent a02342c9e1
commit e71760fc64
3 changed files with 463 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

405
codex-rs/core/src/analytics_client.rs
View File

@@ -7,9 +7,11 @@ use codex_protocol::protocol::SkillScope;
use serde::Serialize;
use sha1::Digest;
use sha1::Sha1;
use std::collections::HashSet;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use tokio::sync::mpsc;
@@ -17,15 +19,18 @@ use tokio::sync::mpsc;
pub(crate) struct TrackEventsContext {
pub(crate) model_slug: String,
pub(crate) thread_id: String,
pub(crate) turn_id: String,
}
pub(crate) fn build_track_events_context(
model_slug: String,
thread_id: String,
turn_id: String,
) -> TrackEventsContext {
TrackEventsContext {
model_slug,
thread_id,
turn_id,
}
}
@@ -35,9 +40,16 @@ pub(crate) struct SkillInvocation {
pub(crate) skill_path: PathBuf,
}
pub(crate) struct AppInvocation {
pub(crate) connector_id: Option<String>,
pub(crate) app_name: Option<String>,
pub(crate) invoke_type: Option<String>,
}
#[derive(Clone)]
pub(crate) struct AnalyticsEventsQueue {
sender: mpsc::Sender<TrackEventsJob>,
app_used_emitted_keys: Arc<Mutex<HashSet<(String, String)>>>,
}
pub(crate) struct AnalyticsEventsClient {
@@ -50,18 +62,45 @@ impl AnalyticsEventsQueue {
let (sender, mut receiver) = mpsc::channel(ANALYTICS_EVENTS_QUEUE_SIZE);
tokio::spawn(async move {
while let Some(job) = receiver.recv().await {
send_track_skill_invocations(&auth_manager, job).await;
match job {
TrackEventsJob::SkillInvocations(job) => {
send_track_skill_invocations(&auth_manager, job).await;
}
TrackEventsJob::AppMentioned(job) => {
send_track_app_mentioned(&auth_manager, job).await;
}
TrackEventsJob::AppUsed(job) => {
send_track_app_used(&auth_manager, job).await;
}
}
}
});
Self { sender }
Self {
sender,
app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
}
}
fn try_send(&self, job: TrackEventsJob) {
if self.sender.try_send(job).is_err() {
//TODO: add a metric for this
tracing::warn!("dropping skill analytics events: queue is full");
tracing::warn!("dropping analytics events: queue is full");
}
}
fn should_enqueue_app_used(&self, tracking: &TrackEventsContext, app: &AppInvocation) -> bool {
let Some(connector_id) = app.connector_id.as_ref() else {
return true;
};
let mut emitted = self
.app_used_emitted_keys
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if emitted.len() >= ANALYTICS_APP_USED_DEDUPE_MAX_KEYS {
emitted.clear();
}
emitted.insert((tracking.turn_id.clone(), connector_id.clone()))
}
}
impl AnalyticsEventsClient {
@@ -84,32 +123,76 @@ impl AnalyticsEventsClient {
invocations,
);
}
pub(crate) fn track_app_mentioned(
&self,
tracking: TrackEventsContext,
mentions: Vec<AppInvocation>,
) {
track_app_mentioned(
&self.queue,
Arc::clone(&self.config),
Some(tracking),
mentions,
);
}
pub(crate) fn track_app_used(&self, tracking: TrackEventsContext, app: AppInvocation) {
track_app_used(&self.queue, Arc::clone(&self.config), Some(tracking), app);
}
}
struct TrackEventsJob {
enum TrackEventsJob {
SkillInvocations(TrackSkillInvocationsJob),
AppMentioned(TrackAppMentionedJob),
AppUsed(TrackAppUsedJob),
}
struct TrackSkillInvocationsJob {
config: Arc<Config>,
tracking: TrackEventsContext,
invocations: Vec<SkillInvocation>,
}
struct TrackAppMentionedJob {
config: Arc<Config>,
tracking: TrackEventsContext,
mentions: Vec<AppInvocation>,
}
struct TrackAppUsedJob {
config: Arc<Config>,
tracking: TrackEventsContext,
app: AppInvocation,
}
const ANALYTICS_EVENTS_QUEUE_SIZE: usize = 256;
const ANALYTICS_EVENTS_TIMEOUT: Duration = Duration::from_secs(10);
const ANALYTICS_APP_USED_DEDUPE_MAX_KEYS: usize = 4096;
#[derive(Serialize)]
struct TrackEventsRequest {
events: Vec<TrackEvent>,
events: Vec<TrackEventRequest>,
}
#[derive(Serialize)]
struct TrackEvent {
#[serde(untagged)]
enum TrackEventRequest {
SkillInvocation(SkillInvocationEventRequest),
AppMentioned(CodexAppMentionedEventRequest),
AppUsed(CodexAppUsedEventRequest),
}
#[derive(Serialize)]
struct SkillInvocationEventRequest {
event_type: &'static str,
skill_id: String,
skill_name: String,
event_params: TrackEventParams,
event_params: SkillInvocationEventParams,
}
#[derive(Serialize)]
struct TrackEventParams {
struct SkillInvocationEventParams {
product_client_id: Option<String>,
skill_scope: Option<String>,
repo_url: Option<String>,
@@ -118,6 +201,29 @@ struct TrackEventParams {
model_slug: Option<String>,
}
#[derive(Serialize)]
struct CodexAppMetadata {
connector_id: Option<String>,
thread_id: Option<String>,
turn_id: Option<String>,
app_name: Option<String>,
product_client_id: Option<String>,
invoke_type: Option<String>,
model_slug: Option<String>,
}
#[derive(Serialize)]
struct CodexAppMentionedEventRequest {
event_type: &'static str,
event_params: CodexAppMetadata,
}
#[derive(Serialize)]
struct CodexAppUsedEventRequest {
event_type: &'static str,
event_params: CodexAppMetadata,
}
pub(crate) fn track_skill_invocations(
queue: &AnalyticsEventsQueue,
config: Arc<Config>,
@@ -133,34 +239,66 @@ pub(crate) fn track_skill_invocations(
if invocations.is_empty() {
return;
}
let job = TrackEventsJob {
let job = TrackEventsJob::SkillInvocations(TrackSkillInvocationsJob {
config,
tracking,
invocations,
};
});
queue.try_send(job);
}
async fn send_track_skill_invocations(auth_manager: &AuthManager, job: TrackEventsJob) {
let TrackEventsJob {
pub(crate) fn track_app_mentioned(
queue: &AnalyticsEventsQueue,
config: Arc<Config>,
tracking: Option<TrackEventsContext>,
mentions: Vec<AppInvocation>,
) {
if config.analytics_enabled == Some(false) {
return;
}
let Some(tracking) = tracking else {
return;
};
if mentions.is_empty() {
return;
}
let job = TrackEventsJob::AppMentioned(TrackAppMentionedJob {
config,
tracking,
mentions,
});
queue.try_send(job);
}
pub(crate) fn track_app_used(
queue: &AnalyticsEventsQueue,
config: Arc<Config>,
tracking: Option<TrackEventsContext>,
app: AppInvocation,
) {
if config.analytics_enabled == Some(false) {
return;
}
let Some(tracking) = tracking else {
return;
};
if !queue.should_enqueue_app_used(&tracking, &app) {
return;
}
let job = TrackEventsJob::AppUsed(TrackAppUsedJob {
config,
tracking,
app,
});
queue.try_send(job);
}
async fn send_track_skill_invocations(auth_manager: &AuthManager, job: TrackSkillInvocationsJob) {
let TrackSkillInvocationsJob {
config,
tracking,
invocations,
} = job;
let Some(auth) = auth_manager.auth().await else {
return;
};
if !auth.is_chatgpt_auth() {
return;
}
let access_token = match auth.get_token() {
Ok(token) => token,
Err(_) => return,
};
let Some(account_id) = auth.get_account_id() else {
return;
};
let mut events = Vec::with_capacity(invocations.len());
for invocation in invocations {
let skill_scope = match invocation.skill_scope {
@@ -183,21 +321,95 @@ async fn send_track_skill_invocations(auth_manager: &AuthManager, job: TrackEven
invocation.skill_path.as_path(),
invocation.skill_name.as_str(),
);
events.push(TrackEvent {
event_type: "skill_invocation",
skill_id,
skill_name: invocation.skill_name.clone(),
event_params: TrackEventParams {
thread_id: Some(tracking.thread_id.clone()),
invoke_type: Some("explicit".to_string()),
model_slug: Some(tracking.model_slug.clone()),
product_client_id: Some(crate::default_client::originator().value),
repo_url,
skill_scope: Some(skill_scope.to_string()),
events.push(TrackEventRequest::SkillInvocation(
SkillInvocationEventRequest {
event_type: "skill_invocation",
skill_id,
skill_name: invocation.skill_name.clone(),
event_params: SkillInvocationEventParams {
thread_id: Some(tracking.thread_id.clone()),
invoke_type: Some("explicit".to_string()),
model_slug: Some(tracking.model_slug.clone()),
product_client_id: Some(crate::default_client::originator().value),
repo_url,
skill_scope: Some(skill_scope.to_string()),
},
},
});
));
}
send_track_events(auth_manager, config, events).await;
}
async fn send_track_app_mentioned(auth_manager: &AuthManager, job: TrackAppMentionedJob) {
let TrackAppMentionedJob {
config,
tracking,
mentions,
} = job;
let events = mentions
.into_iter()
.map(|mention| {
let event_params = codex_app_metadata(&tracking, mention);
TrackEventRequest::AppMentioned(CodexAppMentionedEventRequest {
event_type: "codex_app_mentioned",
event_params,
})
})
.collect::<Vec<_>>();
send_track_events(auth_manager, config, events).await;
}
async fn send_track_app_used(auth_manager: &AuthManager, job: TrackAppUsedJob) {
let TrackAppUsedJob {
config,
tracking,
app,
} = job;
let event_params = codex_app_metadata(&tracking, app);
let events = vec![TrackEventRequest::AppUsed(CodexAppUsedEventRequest {
event_type: "codex_app_used",
event_params,
})];
send_track_events(auth_manager, config, events).await;
}
fn codex_app_metadata(tracking: &TrackEventsContext, app: AppInvocation) -> CodexAppMetadata {
CodexAppMetadata {
connector_id: app.connector_id,
thread_id: Some(tracking.thread_id.clone()),
turn_id: Some(tracking.turn_id.clone()),
app_name: app.app_name,
product_client_id: Some(crate::default_client::originator().value),
invoke_type: app.invoke_type,
model_slug: Some(tracking.model_slug.clone()),
}
}
async fn send_track_events(
auth_manager: &AuthManager,
config: Arc<Config>,
events: Vec<TrackEventRequest>,
) {
if events.is_empty() {
return;
}
let Some(auth) = auth_manager.auth().await else {
return;
};
if !auth.is_chatgpt_auth() {
return;
}
let access_token = match auth.get_token() {
Ok(token) => token,
Err(_) => return,
};
let Some(account_id) = auth.get_account_id() else {
return;
};
let base_url = config.chatgpt_base_url.trim_end_matches('/');
let url = format!("{base_url}/codex/analytics-events/events");
let payload = TrackEventsRequest { events };
@@ -269,9 +481,21 @@ fn normalize_path_for_skill_id(
#[cfg(test)]
mod tests {
use super::AnalyticsEventsQueue;
use super::AppInvocation;
use super::CodexAppMentionedEventRequest;
use super::CodexAppUsedEventRequest;
use super::TrackEventRequest;
use super::TrackEventsContext;
use super::codex_app_metadata;
use super::normalize_path_for_skill_id;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use tokio::sync::mpsc;
fn expected_absolute_path(path: &PathBuf) -> String {
std::fs::canonicalize(path)
@@ -328,4 +552,109 @@ mod tests {
assert_eq!(path, expected);
}
#[test]
fn app_mentioned_event_serializes_expected_shape() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let event = TrackEventRequest::AppMentioned(CodexAppMentionedEventRequest {
event_type: "codex_app_mentioned",
event_params: codex_app_metadata(
&tracking,
AppInvocation {
connector_id: Some("calendar".to_string()),
app_name: Some("Calendar".to_string()),
invoke_type: Some("explicit".to_string()),
},
),
});
let payload = serde_json::to_value(&event).expect("serialize app mentioned event");
assert_eq!(
payload,
json!({
"event_type": "codex_app_mentioned",
"event_params": {
"connector_id": "calendar",
"thread_id": "thread-1",
"turn_id": "turn-1",
"app_name": "Calendar",
"product_client_id": crate::default_client::originator().value,
"invoke_type": "explicit",
"model_slug": "gpt-5"
}
})
);
}
#[test]
fn app_used_event_serializes_expected_shape() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
};
let event = TrackEventRequest::AppUsed(CodexAppUsedEventRequest {
event_type: "codex_app_used",
event_params: codex_app_metadata(
&tracking,
AppInvocation {
connector_id: Some("drive".to_string()),
app_name: Some("Google Drive".to_string()),
invoke_type: Some("implicit".to_string()),
},
),
});
let payload = serde_json::to_value(&event).expect("serialize app used event");
assert_eq!(
payload,
json!({
"event_type": "codex_app_used",
"event_params": {
"connector_id": "drive",
"thread_id": "thread-2",
"turn_id": "turn-2",
"app_name": "Google Drive",
"product_client_id": crate::default_client::originator().value,
"invoke_type": "implicit",
"model_slug": "gpt-5"
}
})
);
}
#[test]
fn app_used_dedupe_is_keyed_by_turn_and_connector() {
let (sender, _receiver) = mpsc::channel(1);
let queue = AnalyticsEventsQueue {
sender,
app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
};
let app = AppInvocation {
connector_id: Some("calendar".to_string()),
app_name: Some("Calendar".to_string()),
invoke_type: Some("implicit".to_string()),
};
let turn_1 = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let turn_2 = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-2".to_string(),
};
assert_eq!(queue.should_enqueue_app_used(&turn_1, &app), true);
assert_eq!(queue.should_enqueue_app_used(&turn_1, &app), false);
assert_eq!(queue.should_enqueue_app_used(&turn_2, &app), true);
}
}

24
codex-rs/core/src/codex.rs
View File

@@ -14,6 +14,7 @@ use crate::agent::AgentStatus;
use crate::agent::MAX_THREAD_SPAWN_DEPTH;
use crate::agent::agent_status_from_event;
use crate::analytics_client::AnalyticsEventsClient;
use crate::analytics_client::AppInvocation;
use crate::analytics_client::build_track_events_context;
use crate::apps::render_apps_section;
use crate::compact;
@@ -4233,7 +4234,11 @@ pub(crate) async fn run_turn(
let otel_manager = turn_context.otel_manager.clone();
let thread_id = sess.conversation_id.to_string();
let tracking = build_track_events_context(turn_context.model_info.slug.clone(), thread_id);
let tracking = build_track_events_context(
turn_context.model_info.slug.clone(),
thread_id,
turn_context.sub_id.clone(),
);
let SkillInjections {
items: skill_items,
warnings: skill_warnings,
@@ -4256,6 +4261,23 @@ pub(crate) async fn run_turn(
&available_connectors,
&skill_name_counts_lower,
));
let connector_names_by_id = available_connectors
.iter()
.map(|connector| (connector.id.as_str(), connector.name.as_str()))
.collect::<HashMap<&str, &str>>();
let mentioned_app_invocations = explicitly_enabled_connectors
.iter()
.map(|connector_id| AppInvocation {
connector_id: Some(connector_id.clone()),
app_name: connector_names_by_id
.get(connector_id.as_str())
.map(|name| (*name).to_string()),
invoke_type: Some("explicit".to_string()),
})
.collect::<Vec<_>>();
sess.services
.analytics_events_client
.track_app_mentioned(tracking.clone(), mentioned_app_invocations);
sess.merge_connector_selection(explicitly_enabled_connectors.clone())
.await;

73
codex-rs/core/src/mcp_tool_call.rs
View File

@@ -3,6 +3,8 @@ use std::time::Instant;
use tracing::error;
use crate::analytics_client::AppInvocation;
use crate::analytics_client::build_track_events_context;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
@@ -102,6 +104,7 @@ pub(crate) async fn handle_mcp_tool_call(
tool_call_end_event.clone(),
)
.await;
maybe_track_codex_app_used(sess.as_ref(), turn_context, &server, &tool_name).await;
result
}
McpToolApprovalDecision::Decline => {
@@ -166,6 +169,7 @@ pub(crate) async fn handle_mcp_tool_call(
});
notify_mcp_tool_call_event(sess.as_ref(), turn_context, tool_call_end_event.clone()).await;
maybe_track_codex_app_used(sess.as_ref(), turn_context, &server, &tool_name).await;
let status = if result.is_ok() { "ok" } else { "error" };
turn_context
@@ -210,6 +214,50 @@ async fn notify_mcp_tool_call_event(sess: &Session, turn_context: &TurnContext,
sess.send_event(turn_context, event).await;
}
struct McpAppUsageMetadata {
connector_id: Option<String>,
app_name: Option<String>,
}
async fn maybe_track_codex_app_used(
sess: &Session,
turn_context: &TurnContext,
server: &str,
tool_name: &str,
) {
if server != CODEX_APPS_MCP_SERVER_NAME {
return;
}
let metadata = lookup_mcp_app_usage_metadata(sess, server, tool_name).await;
let (connector_id, app_name) = metadata
.map(|metadata| (metadata.connector_id, metadata.app_name))
.unwrap_or((None, None));
let invoke_type = if let Some(connector_id) = connector_id.as_deref() {
let mentioned_connector_ids = sess.get_connector_selection().await;
if mentioned_connector_ids.contains(connector_id) {
"explicit"
} else {
"implicit"
}
} else {
"implicit"
};
let tracking = build_track_events_context(
turn_context.model_info.slug.clone(),
sess.conversation_id.to_string(),
turn_context.sub_id.clone(),
);
sess.services.analytics_events_client.track_app_used(
tracking,
AppInvocation {
connector_id,
app_name,
invoke_type: Some(invoke_type.to_string()),
},
);
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum McpToolApprovalDecision {
Accept,
@@ -332,6 +380,31 @@ async fn lookup_mcp_tool_metadata(
})
}
async fn lookup_mcp_app_usage_metadata(
sess: &Session,
server: &str,
tool_name: &str,
) -> Option<McpAppUsageMetadata> {
let tools = sess
.services
.mcp_connection_manager
.read()
.await
.list_all_tools()
.await;
tools.into_values().find_map(|tool_info| {
if tool_info.server_name == server && tool_info.tool_name == tool_name {
Some(McpAppUsageMetadata {
connector_id: tool_info.connector_id,
app_name: tool_info.connector_name,
})
} else {
None
}
})
}
fn build_mcp_tool_approval_question(
question_id: String,
tool_name: &str,