thread event start

This commit is contained in:
Roy Han
2026-03-24 15:10:10 -07:00
parent 0f34b14b41
commit 92e529a8ec
6 changed files with 412 additions and 8 deletions
+158
View File
@@ -4,6 +4,15 @@ use crate::default_client::create_client;
use crate::git_info::collect_git_info;
use crate::git_info::get_git_repo_root;
use crate::plugins::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 serde::Serialize;
use sha1::Digest;
@@ -23,6 +32,35 @@ pub(crate) struct TrackEventsContext {
pub(crate) turn_id: String,
}
#[derive(Clone)]
pub(crate) struct CodexThreadStartedEvent {
pub(crate) thread_id: String,
pub(crate) model: String,
pub(crate) model_provider: String,
pub(crate) reasoning_effort: Option<ReasoningEffort>,
pub(crate) reasoning_summary: Option<ReasoningSummary>,
pub(crate) service_tier: Option<ServiceTier>,
pub(crate) approval_policy: AskForApproval,
pub(crate) approvals_reviewer: ApprovalsReviewer,
pub(crate) sandbox_policy: SandboxPolicy,
pub(crate) sandbox_network_access: bool,
pub(crate) collaboration_mode: ModeKind,
pub(crate) personality: Option<Personality>,
pub(crate) ephemeral: bool,
pub(crate) session_source: SessionSource,
pub(crate) parent_thread_id: Option<String>,
pub(crate) create_source: ThreadCreateSource,
pub(crate) created_at: u64,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ThreadCreateSource {
ThreadStart,
ThreadFork,
DetachedReview,
}
pub(crate) fn build_track_events_context(
model_slug: String,
thread_id: String,
@@ -78,6 +116,9 @@ impl AnalyticsEventsQueue {
TrackEventsJob::SkillInvocations(job) => {
send_track_skill_invocations(&auth_manager, job).await;
}
TrackEventsJob::ThreadStarted(job) => {
send_track_thread_started(&auth_manager, job).await;
}
TrackEventsJob::AppMentioned(job) => {
send_track_app_mentioned(&auth_manager, job).await;
}
@@ -167,6 +208,10 @@ impl AnalyticsEventsClient {
);
}
pub(crate) fn track_thread_started(&self, thread_event: CodexThreadStartedEvent) {
track_thread_started(&self.queue, Arc::clone(&self.config), thread_event);
}
pub(crate) fn track_app_mentioned(
&self,
tracking: TrackEventsContext,
@@ -236,6 +281,7 @@ impl AnalyticsEventsClient {
enum TrackEventsJob {
SkillInvocations(TrackSkillInvocationsJob),
ThreadStarted(TrackThreadStartedJob),
AppMentioned(TrackAppMentionedJob),
AppUsed(TrackAppUsedJob),
PluginUsed(TrackPluginUsedJob),
@@ -251,6 +297,11 @@ struct TrackSkillInvocationsJob {
invocations: Vec<SkillInvocation>,
}
struct TrackThreadStartedJob {
config: Arc<Config>,
thread_event: CodexThreadStartedEvent,
}
struct TrackAppMentionedJob {
config: Arc<Config>,
tracking: TrackEventsContext,
@@ -295,6 +346,7 @@ struct TrackEventsRequest {
#[serde(untagged)]
enum TrackEventRequest {
SkillInvocation(SkillInvocationEventRequest),
ThreadStarted(CodexThreadStartedEventRequest),
AppMentioned(CodexAppMentionedEventRequest),
AppUsed(CodexAppUsedEventRequest),
PluginUsed(CodexPluginUsedEventRequest),
@@ -322,6 +374,34 @@ struct SkillInvocationEventParams {
model_slug: Option<String>,
}
#[derive(Serialize)]
struct CodexThreadStartedEventParams {
thread_id: String,
product_client_id: String,
model: String,
model_provider: String,
reasoning_effort: Option<String>,
reasoning_summary: Option<String>,
service_tier: String,
approval_policy: String,
approvals_reviewer: String,
sandbox_policy: &'static str,
sandbox_network_access: bool,
collaboration_mode: &'static str,
personality: Option<String>,
ephemeral: bool,
session_source: SessionSource,
parent_thread_id: Option<String>,
create_source: ThreadCreateSource,
created_at: u64,
}
#[derive(Serialize)]
struct CodexThreadStartedEventRequest {
event_type: &'static str,
event_params: CodexThreadStartedEventParams,
}
#[derive(Serialize)]
struct CodexAppMetadata {
connector_id: Option<String>,
@@ -400,6 +480,21 @@ pub(crate) fn track_skill_invocations(
queue.try_send(job);
}
pub(crate) fn track_thread_started(
queue: &AnalyticsEventsQueue,
config: Arc<Config>,
thread_event: CodexThreadStartedEvent,
) {
if config.analytics_enabled == Some(false) {
return;
}
let job = TrackEventsJob::ThreadStarted(TrackThreadStartedJob {
config,
thread_event,
});
queue.try_send(job);
}
pub(crate) fn track_app_mentioned(
queue: &AnalyticsEventsQueue,
config: Arc<Config>,
@@ -536,6 +631,21 @@ async fn send_track_skill_invocations(auth_manager: &AuthManager, job: TrackSkil
send_track_events(auth_manager, config, events).await;
}
async fn send_track_thread_started(auth_manager: &AuthManager, job: TrackThreadStartedJob) {
let TrackThreadStartedJob {
config,
thread_event,
} = job;
let events = vec![TrackEventRequest::ThreadStarted(
CodexThreadStartedEventRequest {
event_type: "codex_thread_started",
event_params: codex_thread_started_event_params(thread_event),
},
)];
send_track_events(auth_manager, config, events).await;
}
async fn send_track_app_mentioned(auth_manager: &AuthManager, job: TrackAppMentionedJob) {
let TrackAppMentionedJob {
config,
@@ -635,6 +745,36 @@ fn codex_app_metadata(tracking: &TrackEventsContext, app: AppInvocation) -> Code
}
}
fn codex_thread_started_event_params(
thread_event: CodexThreadStartedEvent,
) -> CodexThreadStartedEventParams {
CodexThreadStartedEventParams {
thread_id: thread_event.thread_id,
product_client_id: crate::default_client::originator().value,
model: thread_event.model,
model_provider: thread_event.model_provider,
reasoning_effort: thread_event.reasoning_effort.map(|value| value.to_string()),
reasoning_summary: thread_event
.reasoning_summary
.map(|value| value.to_string()),
service_tier: thread_event
.service_tier
.map(|value| value.to_string())
.unwrap_or_else(|| "default".to_string()),
approval_policy: thread_event.approval_policy.to_string(),
approvals_reviewer: thread_event.approvals_reviewer.to_string(),
sandbox_policy: sandbox_policy_mode(&thread_event.sandbox_policy),
sandbox_network_access: thread_event.sandbox_network_access,
collaboration_mode: collaboration_mode_mode(thread_event.collaboration_mode),
personality: thread_event.personality.map(|value| value.to_string()),
ephemeral: thread_event.ephemeral,
session_source: thread_event.session_source,
parent_thread_id: thread_event.parent_thread_id,
create_source: thread_event.create_source,
created_at: thread_event.created_at,
}
}
fn codex_plugin_metadata(plugin: PluginTelemetryMetadata) -> CodexPluginMetadata {
let capability_summary = plugin.capability_summary;
CodexPluginMetadata {
@@ -670,6 +810,24 @@ fn codex_plugin_used_metadata(
}
}
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 => "default",
ModeKind::PairProgramming => "pair_programming",
ModeKind::Execute => "execute",
}
}
async fn send_track_events(
auth_manager: &AuthManager,
config: Arc<Config>,
@@ -4,17 +4,30 @@ use super::CodexAppMentionedEventRequest;
use super::CodexAppUsedEventRequest;
use super::CodexPluginEventRequest;
use super::CodexPluginUsedEventRequest;
use super::CodexThreadStartedEvent;
use super::CodexThreadStartedEventRequest;
use super::InvocationType;
use super::ThreadCreateSource;
use super::TrackEventRequest;
use super::TrackEventsContext;
use super::codex_app_metadata;
use super::codex_plugin_metadata;
use super::codex_plugin_used_metadata;
use super::codex_thread_started_event_params;
use super::normalize_path_for_skill_id;
use crate::plugins::AppConnectorId;
use crate::plugins::PluginCapabilitySummary;
use crate::plugins::PluginId;
use crate::plugins::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 pretty_assertions::assert_eq;
use serde_json::json;
use std::collections::HashSet;
@@ -185,6 +198,61 @@ fn app_used_dedupe_is_keyed_by_turn_and_connector() {
assert_eq!(queue.should_enqueue_app_used(&turn_2, &app), true);
}
#[test]
fn thread_started_event_serializes_expected_shape() {
let event = TrackEventRequest::ThreadStarted(CodexThreadStartedEventRequest {
event_type: "codex_thread_started",
event_params: codex_thread_started_event_params(CodexThreadStartedEvent {
thread_id: "thread-0".to_string(),
model: "gpt-5".to_string(),
model_provider: "openai".to_string(),
reasoning_effort: Some(ReasoningEffort::High),
reasoning_summary: Some(ReasoningSummary::Detailed),
service_tier: Some(ServiceTier::Flex),
approval_policy: AskForApproval::OnRequest,
approvals_reviewer: ApprovalsReviewer::GuardianSubagent,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
sandbox_network_access: false,
collaboration_mode: ModeKind::Plan,
personality: Some(Personality::Friendly),
ephemeral: true,
session_source: SessionSource::Exec,
parent_thread_id: None,
create_source: ThreadCreateSource::ThreadStart,
created_at: 1_716_000_000,
}),
});
let payload = serde_json::to_value(&event).expect("serialize thread started event");
assert_eq!(
payload,
json!({
"event_type": "codex_thread_started",
"event_params": {
"thread_id": "thread-0",
"product_client_id": crate::default_client::originator().value,
"model": "gpt-5",
"model_provider": "openai",
"reasoning_effort": "high",
"reasoning_summary": "detailed",
"service_tier": "flex",
"approval_policy": "on-request",
"approvals_reviewer": "guardian_subagent",
"sandbox_policy": "read_only",
"sandbox_network_access": false,
"collaboration_mode": "plan",
"personality": "friendly",
"ephemeral": true,
"session_source": "exec",
"parent_thread_id": null,
"create_source": "thread_start",
"created_at": 1716000000
}
})
);
}
#[test]
fn plugin_used_event_serializes_expected_shape() {
let tracking = TrackEventsContext {
+78
View File
@@ -5,6 +5,8 @@ use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
use crate::AuthManager;
use crate::CodexAuth;
@@ -14,7 +16,9 @@ use crate::agent::AgentStatus;
use crate::agent::agent_status_from_event;
use crate::analytics_client::AnalyticsEventsClient;
use crate::analytics_client::AppInvocation;
use crate::analytics_client::CodexThreadStartedEvent;
use crate::analytics_client::InvocationType;
use crate::analytics_client::ThreadCreateSource;
use crate::analytics_client::build_track_events_context;
use crate::apps::render_apps_section;
use crate::auth_env_telemetry::collect_auth_env_telemetry;
@@ -627,6 +631,13 @@ impl Codex {
user_shell_override,
};
let thread_create_source = classify_thread_create_source(
&conversation_history,
&session_configuration.session_source,
);
let thread_session_source = session_configuration.session_source.clone();
let thread_started_configuration = session_configuration.clone();
// Generate a unique ID for the lifetime of this Codex session.
let session_source_clone = session_configuration.session_source.clone();
let (agent_status_tx, agent_status_rx) = watch::channel(AgentStatus::PendingInit);
@@ -654,6 +665,46 @@ impl Codex {
})?;
let thread_id = session.conversation_id;
if let Some(create_source) = thread_create_source {
session
.services
.analytics_events_client
.track_thread_started(CodexThreadStartedEvent {
thread_id: thread_id.to_string(),
model: thread_started_configuration
.collaboration_mode
.model()
.to_string(),
model_provider: thread_started_configuration
.original_config_do_not_use
.model_provider_id
.clone(),
reasoning_effort: thread_started_configuration
.collaboration_mode
.reasoning_effort(),
reasoning_summary: thread_started_configuration.model_reasoning_summary,
service_tier: thread_started_configuration.service_tier,
approval_policy: thread_started_configuration.approval_policy.value(),
approvals_reviewer: thread_started_configuration.approvals_reviewer,
sandbox_policy: thread_started_configuration.sandbox_policy.get().clone(),
sandbox_network_access: thread_started_configuration
.network_sandbox_policy
.is_enabled(),
collaboration_mode: thread_started_configuration.collaboration_mode.mode,
personality: thread_started_configuration.personality,
ephemeral: thread_started_configuration
.original_config_do_not_use
.ephemeral,
parent_thread_id: session_source_parent_thread_id(&thread_session_source),
session_source: thread_session_source,
create_source,
created_at: SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs(),
});
}
// This task will run until Op::Shutdown is received.
let session_for_loop = Arc::clone(&session);
let session_loop_handle = tokio::spawn(async move {
@@ -768,6 +819,33 @@ impl Codex {
}
}
fn classify_thread_create_source(
conversation_history: &InitialHistory,
session_source: &SessionSource,
) -> Option<ThreadCreateSource> {
if matches!(
session_source,
SessionSource::SubAgent(SubAgentSource::Review)
) {
return Some(ThreadCreateSource::DetachedReview);
}
match conversation_history {
InitialHistory::New => Some(ThreadCreateSource::ThreadStart),
InitialHistory::Forked(_) => Some(ThreadCreateSource::ThreadFork),
InitialHistory::Resumed(_) => None,
}
}
fn session_source_parent_thread_id(session_source: &SessionSource) -> Option<String> {
match session_source {
SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id, ..
}) => Some(parent_thread_id.to_string()),
_ => None,
}
}
#[cfg(test)]
pub(crate) fn completed_session_loop_termination() -> SessionLoopTermination {
futures::future::ready(()).boxed().shared()
+1
View File
@@ -130,6 +130,7 @@ mod stream_error_allows_next_turn;
mod stream_no_completed;
mod subagent_notifications;
mod text_encoding_fix;
mod thread_metadata;
mod tool_harness;
mod tool_parallelism;
mod tools;
+14 -8
View File
@@ -365,13 +365,22 @@ async fn explicit_plugin_mentions_track_plugin_used_analytics() -> Result<()> {
wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let deadline = Instant::now() + Duration::from_secs(10);
let analytics_request = loop {
let plugin_event = loop {
let requests = server.received_requests().await.unwrap_or_default();
if let Some(request) = requests
if let Some(event) = requests
.into_iter()
.find(|request| request.url.path() == "/codex/analytics-events/events")
.filter(|request| request.url.path() == "/codex/analytics-events/events")
.find_map(|request| {
let payload: serde_json::Value = serde_json::from_slice(&request.body).ok()?;
payload["events"].as_array().and_then(|events| {
events
.iter()
.find(|event| event["event_type"] == "codex_plugin_used")
.cloned()
})
})
{
break request;
break event;
}
if Instant::now() >= deadline {
panic!("timed out waiting for plugin analytics request");
@@ -379,10 +388,7 @@ async fn explicit_plugin_mentions_track_plugin_used_analytics() -> Result<()> {
tokio::time::sleep(Duration::from_millis(50)).await;
};
let payload: serde_json::Value =
serde_json::from_slice(&analytics_request.body).expect("analytics payload");
let event = &payload["events"][0];
assert_eq!(event["event_type"], "codex_plugin_used");
let event = plugin_event;
assert_eq!(event["event_params"]["plugin_id"], "sample@test");
assert_eq!(event["event_params"]["plugin_name"], "sample");
assert_eq!(event["event_params"]["marketplace_name"], "test");
@@ -0,0 +1,93 @@
use anyhow::Result;
use codex_core::CodexAuth;
use codex_protocol::config_types::ApprovalsReviewer;
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 core_test_support::responses::start_mock_server;
use core_test_support::test_codex::test_codex;
use std::time::Duration;
use std::time::Instant;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn thread_start_tracks_thread_started_analytics() -> Result<()> {
let server = start_mock_server().await;
let chatgpt_base_url = server.uri();
let test = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(move |config| {
config.chatgpt_base_url = chatgpt_base_url;
config.model = Some("gpt-5".to_string());
config.model_reasoning_effort = Some(ReasoningEffort::High);
config.model_reasoning_summary = Some(ReasoningSummary::Detailed);
config.service_tier = Some(ServiceTier::Flex);
config.approvals_reviewer = ApprovalsReviewer::GuardianSubagent;
config.personality = Some(Personality::Friendly);
config.ephemeral = true;
})
.build(&server)
.await?;
let deadline = Instant::now() + Duration::from_secs(10);
let analytics_request = loop {
let requests = server.received_requests().await.unwrap_or_default();
if let Some(request) = requests
.into_iter()
.find(|request| request.url.path() == "/codex/analytics-events/events")
{
break request;
}
if Instant::now() >= deadline {
panic!("timed out waiting for thread analytics request");
}
tokio::time::sleep(Duration::from_millis(50)).await;
};
let payload: serde_json::Value =
serde_json::from_slice(&analytics_request.body).expect("analytics payload");
let event = payload["events"]
.as_array()
.expect("events array")
.iter()
.find(|event| event["event_type"] == "codex_thread_started")
.expect("codex_thread_started event should be present");
assert_eq!(
event["event_params"]["thread_id"],
serde_json::json!(test.session_configured.session_id.to_string())
);
assert_eq!(
event["event_params"]["product_client_id"],
serde_json::json!(codex_core::default_client::originator().value)
);
assert_eq!(event["event_params"]["model"], "gpt-5");
assert_eq!(event["event_params"]["model_provider"], "openai");
assert_eq!(event["event_params"]["reasoning_effort"], "high");
assert_eq!(event["event_params"]["reasoning_summary"], "detailed");
assert_eq!(event["event_params"]["service_tier"], "flex");
assert_eq!(event["event_params"]["approval_policy"], "on-request");
assert_eq!(
event["event_params"]["approvals_reviewer"],
"guardian_subagent"
);
assert_eq!(event["event_params"]["sandbox_policy"], "workspace_write");
assert_eq!(event["event_params"]["sandbox_network_access"], false);
assert_eq!(event["event_params"]["collaboration_mode"], "default");
assert_eq!(event["event_params"]["personality"], "friendly");
assert_eq!(event["event_params"]["ephemeral"], true);
assert_eq!(event["event_params"]["session_source"], "exec");
assert_eq!(
event["event_params"]["parent_thread_id"],
serde_json::Value::Null
);
assert_eq!(event["event_params"]["create_source"], "thread_start");
assert!(
event["event_params"]["created_at"]
.as_u64()
.is_some_and(|timestamp| timestamp > 0)
);
Ok(())
}