mirror of
https://github.com/openai/codex.git
synced 2026-10-27 18:48:37 +00:00
rebase
This commit is contained in:
Generated
+1
@@ -1388,6 +1388,7 @@ checksum = "e9b18233253483ce2f65329a24072ec414db782531bdbb7d0bbc4bd2ce6b7e21"
|
||||
name = "codex-analytics"
|
||||
version = "0.0.0"
|
||||
dependencies = [
|
||||
"codex-app-server-protocol",
|
||||
"codex-git-utils",
|
||||
"codex-login",
|
||||
"codex-plugin",
|
||||
|
||||
@@ -13,6 +13,7 @@ path = "src/lib.rs"
|
||||
workspace = true
|
||||
|
||||
[dependencies]
|
||||
codex-app-server-protocol = { workspace = true }
|
||||
codex-git-utils = { workspace = true }
|
||||
codex-login = { workspace = true }
|
||||
codex-plugin = { workspace = true }
|
||||
|
||||
@@ -1,3 +1,7 @@
|
||||
use codex_app_server_protocol::ClientRequest;
|
||||
use codex_app_server_protocol::InitializeParams;
|
||||
use codex_app_server_protocol::RequestId;
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
use codex_git_utils::collect_git_info;
|
||||
use codex_git_utils::get_git_repo_root;
|
||||
use codex_login::AuthManager;
|
||||
@@ -15,7 +19,6 @@ use codex_protocol::protocol::CodexErrorInfo;
|
||||
use codex_protocol::protocol::SandboxPolicy;
|
||||
use codex_protocol::protocol::SessionSource;
|
||||
use codex_protocol::protocol::SkillScope;
|
||||
use codex_protocol::protocol::SubAgentSource;
|
||||
use serde::Serialize;
|
||||
use sha1::Digest;
|
||||
use sha1::Sha1;
|
||||
@@ -87,21 +90,13 @@ pub enum TurnStatus {
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct CodexThreadInitializedInput {
|
||||
pub struct ThreadInitializedInput {
|
||||
pub connection_id: u64,
|
||||
pub thread_id: String,
|
||||
pub model: String,
|
||||
pub product_client_id: String,
|
||||
pub created_at: u64,
|
||||
pub thread_context: CodexThreadContext,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct CodexThreadContext {
|
||||
pub ephemeral: bool,
|
||||
pub session_source: SessionSource,
|
||||
pub initialization_mode: InitializationMode,
|
||||
pub subagent_source: Option<SubAgentSource>,
|
||||
pub parent_thread_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Serialize)]
|
||||
@@ -145,9 +140,30 @@ pub struct AppInvocation {
|
||||
pub invocation_type: Option<InvocationType>,
|
||||
}
|
||||
|
||||
pub enum AnalyticsInput {
|
||||
CodexThreadInitialized(CodexThreadInitializedInput),
|
||||
TurnStarted(TurnStartedInput),
|
||||
pub enum AnalyticsFact {
|
||||
Initialize {
|
||||
connection_id: u64,
|
||||
params: InitializeParams,
|
||||
},
|
||||
Request {
|
||||
connection_id: u64,
|
||||
request_id: RequestId,
|
||||
request: ClientRequest,
|
||||
},
|
||||
Notification {
|
||||
connection_id: u64,
|
||||
notification: 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),
|
||||
}
|
||||
|
||||
pub enum CustomAnalyticsFact {
|
||||
// This remains custom on this branch because app-server-protocol does not
|
||||
// yet expose a generic client response enum we can reduce over directly.
|
||||
ThreadInitialized(ThreadInitializedInput),
|
||||
TurnStarted(Box<TurnStartedInput>),
|
||||
TurnCompleted(TurnCompletedInput),
|
||||
SkillInvoked(SkillInvokedInput),
|
||||
AppMentioned(AppMentionedInput),
|
||||
@@ -200,12 +216,12 @@ pub enum PluginState {
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct AnalyticsReducer {
|
||||
threads: HashMap<String, ThreadState>,
|
||||
turns: HashMap<String, TurnState>,
|
||||
clients: HashMap<u64, ClientState>,
|
||||
}
|
||||
|
||||
struct ThreadState {
|
||||
_initialized_input: CodexThreadInitializedInput,
|
||||
struct ClientState {
|
||||
product_client_id: String,
|
||||
}
|
||||
|
||||
struct TurnState {
|
||||
@@ -214,7 +230,7 @@ struct TurnState {
|
||||
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct AnalyticsEventsQueue {
|
||||
sender: mpsc::Sender<AnalyticsInput>,
|
||||
sender: mpsc::Sender<AnalyticsFact>,
|
||||
app_used_emitted_keys: Arc<Mutex<HashSet<(String, String)>>>,
|
||||
plugin_used_emitted_keys: Arc<Mutex<HashSet<(String, String)>>>,
|
||||
}
|
||||
@@ -243,7 +259,7 @@ impl AnalyticsEventsQueue {
|
||||
}
|
||||
}
|
||||
|
||||
fn try_send(&self, input: AnalyticsInput) {
|
||||
fn try_send(&self, input: AnalyticsFact) {
|
||||
if self.sender.try_send(input).is_err() {
|
||||
//TODO: add a metric for this
|
||||
tracing::warn!("dropping analytics events: queue is full");
|
||||
@@ -300,93 +316,106 @@ impl AnalyticsEventsClient {
|
||||
if invocations.is_empty() {
|
||||
return;
|
||||
}
|
||||
self.record(AnalyticsInput::SkillInvoked(SkillInvokedInput {
|
||||
tracking,
|
||||
invocations,
|
||||
}));
|
||||
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::SkillInvoked(
|
||||
SkillInvokedInput {
|
||||
tracking,
|
||||
invocations,
|
||||
},
|
||||
)));
|
||||
}
|
||||
|
||||
pub fn track_thread_initialized(&self, input: CodexThreadInitializedInput) {
|
||||
self.record(AnalyticsInput::CodexThreadInitialized(input));
|
||||
pub fn track_initialize(&self, connection_id: u64, params: InitializeParams) {
|
||||
self.record_fact(AnalyticsFact::Initialize {
|
||||
connection_id,
|
||||
params,
|
||||
});
|
||||
}
|
||||
|
||||
pub fn track_thread_initialized(&self, input: ThreadInitializedInput) {
|
||||
self.record_fact(AnalyticsFact::Custom(
|
||||
CustomAnalyticsFact::ThreadInitialized(input),
|
||||
));
|
||||
}
|
||||
|
||||
pub fn track_app_mentioned(&self, tracking: TrackEventsContext, mentions: Vec<AppInvocation>) {
|
||||
if mentions.is_empty() {
|
||||
return;
|
||||
}
|
||||
self.record(AnalyticsInput::AppMentioned(AppMentionedInput {
|
||||
tracking,
|
||||
mentions,
|
||||
}));
|
||||
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::AppMentioned(
|
||||
AppMentionedInput { tracking, mentions },
|
||||
)));
|
||||
}
|
||||
|
||||
pub fn track_app_used(&self, tracking: TrackEventsContext, app: AppInvocation) {
|
||||
if !self.queue.should_enqueue_app_used(&tracking, &app) {
|
||||
return;
|
||||
}
|
||||
self.record(AnalyticsInput::AppUsed(AppUsedInput { tracking, app }));
|
||||
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(
|
||||
AppUsedInput { tracking, app },
|
||||
)));
|
||||
}
|
||||
|
||||
pub fn track_plugin_used(&self, tracking: TrackEventsContext, plugin: PluginTelemetryMetadata) {
|
||||
if !self.queue.should_enqueue_plugin_used(&tracking, &plugin) {
|
||||
return;
|
||||
}
|
||||
self.record(AnalyticsInput::PluginUsed(PluginUsedInput {
|
||||
tracking,
|
||||
plugin,
|
||||
}));
|
||||
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::PluginUsed(
|
||||
PluginUsedInput { tracking, plugin },
|
||||
)));
|
||||
}
|
||||
|
||||
pub fn track_turn_started(&self, tracking: TrackEventsContext, turn_event: CodexTurnEvent) {
|
||||
self.record(AnalyticsInput::TurnStarted(TurnStartedInput {
|
||||
tracking,
|
||||
turn_event,
|
||||
}));
|
||||
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::TurnStarted(
|
||||
Box::new(TurnStartedInput {
|
||||
tracking,
|
||||
turn_event,
|
||||
}),
|
||||
)));
|
||||
}
|
||||
|
||||
pub fn track_turn_completed(&self, turn_id: String) {
|
||||
self.record(AnalyticsInput::TurnCompleted(TurnCompletedInput {
|
||||
turn_id,
|
||||
}));
|
||||
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::TurnCompleted(
|
||||
TurnCompletedInput { turn_id },
|
||||
)));
|
||||
}
|
||||
|
||||
pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) {
|
||||
self.record(AnalyticsInput::PluginStateChanged(
|
||||
PluginStateChangedInput {
|
||||
self.record_fact(AnalyticsFact::Custom(
|
||||
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
|
||||
plugin,
|
||||
state: PluginState::Installed,
|
||||
},
|
||||
}),
|
||||
));
|
||||
}
|
||||
|
||||
pub fn track_plugin_uninstalled(&self, plugin: PluginTelemetryMetadata) {
|
||||
self.record(AnalyticsInput::PluginStateChanged(
|
||||
PluginStateChangedInput {
|
||||
self.record_fact(AnalyticsFact::Custom(
|
||||
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
|
||||
plugin,
|
||||
state: PluginState::Uninstalled,
|
||||
},
|
||||
}),
|
||||
));
|
||||
}
|
||||
|
||||
pub fn track_plugin_enabled(&self, plugin: PluginTelemetryMetadata) {
|
||||
self.record(AnalyticsInput::PluginStateChanged(
|
||||
PluginStateChangedInput {
|
||||
self.record_fact(AnalyticsFact::Custom(
|
||||
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
|
||||
plugin,
|
||||
state: PluginState::Enabled,
|
||||
},
|
||||
}),
|
||||
));
|
||||
}
|
||||
|
||||
pub fn track_plugin_disabled(&self, plugin: PluginTelemetryMetadata) {
|
||||
self.record(AnalyticsInput::PluginStateChanged(
|
||||
PluginStateChangedInput {
|
||||
self.record_fact(AnalyticsFact::Custom(
|
||||
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
|
||||
plugin,
|
||||
state: PluginState::Disabled,
|
||||
},
|
||||
}),
|
||||
));
|
||||
}
|
||||
|
||||
pub fn record(&self, input: AnalyticsInput) {
|
||||
pub fn record_fact(&self, input: AnalyticsFact) {
|
||||
if self.analytics_enabled == Some(false) {
|
||||
return;
|
||||
}
|
||||
@@ -557,48 +586,71 @@ struct CodexPluginUsedEventRequest {
|
||||
}
|
||||
|
||||
impl AnalyticsReducer {
|
||||
async fn ingest(&mut self, input: AnalyticsInput, out: &mut Vec<TrackEventRequest>) {
|
||||
async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec<TrackEventRequest>) {
|
||||
match input {
|
||||
AnalyticsInput::CodexThreadInitialized(input) => {
|
||||
self.ingest_thread_initialized(input, out);
|
||||
}
|
||||
AnalyticsInput::TurnStarted(input) => {
|
||||
self.ingest_turn_started(input);
|
||||
}
|
||||
AnalyticsInput::TurnCompleted(input) => {
|
||||
self.ingest_turn_completed(input, out);
|
||||
}
|
||||
AnalyticsInput::SkillInvoked(input) => {
|
||||
self.ingest_skill_invoked(input, out).await;
|
||||
}
|
||||
AnalyticsInput::AppMentioned(input) => {
|
||||
self.ingest_app_mentioned(input, out);
|
||||
}
|
||||
AnalyticsInput::AppUsed(input) => {
|
||||
self.ingest_app_used(input, out);
|
||||
}
|
||||
AnalyticsInput::PluginUsed(input) => {
|
||||
self.ingest_plugin_used(input, out);
|
||||
}
|
||||
AnalyticsInput::PluginStateChanged(input) => {
|
||||
self.ingest_plugin_state_changed(input, out);
|
||||
AnalyticsFact::Initialize {
|
||||
connection_id,
|
||||
params,
|
||||
} => {
|
||||
self.ingest_initialize(connection_id, params);
|
||||
}
|
||||
AnalyticsFact::Request {
|
||||
connection_id: _connection_id,
|
||||
request_id: _request_id,
|
||||
request: _request,
|
||||
} => {}
|
||||
AnalyticsFact::Notification {
|
||||
connection_id: _connection_id,
|
||||
notification: _notification,
|
||||
} => {}
|
||||
AnalyticsFact::Custom(input) => match input {
|
||||
CustomAnalyticsFact::ThreadInitialized(input) => {
|
||||
self.ingest_thread_initialized(input, out);
|
||||
}
|
||||
CustomAnalyticsFact::TurnStarted(input) => {
|
||||
self.ingest_turn_started(*input);
|
||||
}
|
||||
CustomAnalyticsFact::TurnCompleted(input) => {
|
||||
self.ingest_turn_completed(input, out);
|
||||
}
|
||||
CustomAnalyticsFact::SkillInvoked(input) => {
|
||||
self.ingest_skill_invoked(input, out).await;
|
||||
}
|
||||
CustomAnalyticsFact::AppMentioned(input) => {
|
||||
self.ingest_app_mentioned(input, out);
|
||||
}
|
||||
CustomAnalyticsFact::AppUsed(input) => {
|
||||
self.ingest_app_used(input, out);
|
||||
}
|
||||
CustomAnalyticsFact::PluginUsed(input) => {
|
||||
self.ingest_plugin_used(input, out);
|
||||
}
|
||||
CustomAnalyticsFact::PluginStateChanged(input) => {
|
||||
self.ingest_plugin_state_changed(input, out);
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn ingest_initialize(&mut self, connection_id: u64, params: InitializeParams) {
|
||||
self.clients.insert(
|
||||
connection_id,
|
||||
ClientState {
|
||||
product_client_id: params.client_info.name,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
fn ingest_thread_initialized(
|
||||
&mut self,
|
||||
input: CodexThreadInitializedInput,
|
||||
input: ThreadInitializedInput,
|
||||
out: &mut Vec<TrackEventRequest>,
|
||||
) {
|
||||
self.threads.insert(
|
||||
input.thread_id.clone(),
|
||||
ThreadState {
|
||||
_initialized_input: input.clone(),
|
||||
},
|
||||
);
|
||||
let Some(client_state) = self.clients.get(&input.connection_id) else {
|
||||
return;
|
||||
};
|
||||
out.push(TrackEventRequest::CodexThreadInitialized(
|
||||
codex_thread_initialized_event_request(input),
|
||||
codex_thread_initialized_event_request(client_state.product_client_id.clone(), input),
|
||||
));
|
||||
}
|
||||
|
||||
@@ -824,30 +876,29 @@ fn personality_mode(personality: Option<Personality>) -> Option<String> {
|
||||
}
|
||||
|
||||
fn codex_thread_initialized_event_request(
|
||||
input: CodexThreadInitializedInput,
|
||||
product_client_id: String,
|
||||
input: ThreadInitializedInput,
|
||||
) -> CodexThreadInitializedEvent {
|
||||
CodexThreadInitializedEvent {
|
||||
event_type: "codex_thread_initialized",
|
||||
event_params: codex_thread_initialized_event_params(input),
|
||||
event_params: codex_thread_initialized_event_params(product_client_id, input),
|
||||
}
|
||||
}
|
||||
|
||||
fn codex_thread_initialized_event_params(
|
||||
input: CodexThreadInitializedInput,
|
||||
product_client_id: String,
|
||||
input: ThreadInitializedInput,
|
||||
) -> CodexThreadInitializedEventParams {
|
||||
CodexThreadInitializedEventParams {
|
||||
thread_id: input.thread_id,
|
||||
product_client_id: input.product_client_id,
|
||||
product_client_id,
|
||||
model: input.model,
|
||||
ephemeral: input.thread_context.ephemeral,
|
||||
session_source: session_source_name(&input.thread_context.session_source),
|
||||
initialization_mode: input.thread_context.initialization_mode,
|
||||
subagent_source: input
|
||||
.thread_context
|
||||
.subagent_source
|
||||
.map(subagent_source_name),
|
||||
parent_thread_id: input.thread_context.parent_thread_id,
|
||||
created_at: input.created_at,
|
||||
ephemeral: input.ephemeral,
|
||||
session_source: session_source_name(&input.session_source),
|
||||
initialization_mode: input.initialization_mode,
|
||||
subagent_source: None,
|
||||
parent_thread_id: None,
|
||||
created_at: now_unix_timestamp_secs(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -889,19 +940,18 @@ fn codex_plugin_used_metadata(
|
||||
fn session_source_name(session_source: &SessionSource) -> Option<&'static str> {
|
||||
match session_source {
|
||||
SessionSource::Cli | SessionSource::VSCode | SessionSource::Exec => Some("user"),
|
||||
SessionSource::SubAgent(_) => Some("subagent"),
|
||||
SessionSource::Mcp | SessionSource::Custom(_) | SessionSource::Unknown => None,
|
||||
SessionSource::SubAgent(_)
|
||||
| SessionSource::Mcp
|
||||
| SessionSource::Custom(_)
|
||||
| SessionSource::Unknown => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn subagent_source_name(subagent_source: SubAgentSource) -> String {
|
||||
match subagent_source {
|
||||
SubAgentSource::Review => "review".to_string(),
|
||||
SubAgentSource::Compact => "compact".to_string(),
|
||||
SubAgentSource::ThreadSpawn { .. } => "thread_spawn".to_string(),
|
||||
SubAgentSource::MemoryConsolidation => "memory_consolidation".to_string(),
|
||||
SubAgentSource::Other(other) => other,
|
||||
}
|
||||
fn now_unix_timestamp_secs() -> u64 {
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs()
|
||||
}
|
||||
|
||||
async fn send_track_events(
|
||||
|
||||
@@ -1,17 +1,17 @@
|
||||
use super::AnalyticsEventsQueue;
|
||||
use super::AnalyticsInput;
|
||||
use super::AnalyticsFact;
|
||||
use super::AnalyticsReducer;
|
||||
use super::AppInvocation;
|
||||
use super::CodexAppMentionedEventRequest;
|
||||
use super::CodexAppUsedEventRequest;
|
||||
use super::CodexPluginEventRequest;
|
||||
use super::CodexPluginUsedEventRequest;
|
||||
use super::CodexThreadContext;
|
||||
use super::CodexThreadInitializedInput;
|
||||
use super::CodexTurnEvent;
|
||||
use super::CodexTurnEventRequest;
|
||||
use super::CustomAnalyticsFact;
|
||||
use super::InitializationMode;
|
||||
use super::InvocationType;
|
||||
use super::ThreadInitializedInput;
|
||||
use super::TrackEventRequest;
|
||||
use super::TrackEventsContext;
|
||||
use super::TurnCompletedInput;
|
||||
@@ -23,6 +23,8 @@ use super::codex_plugin_used_metadata;
|
||||
use super::codex_thread_initialized_event_request;
|
||||
use super::codex_turn_event_params;
|
||||
use super::normalize_path_for_skill_id;
|
||||
use codex_app_server_protocol::ClientInfo;
|
||||
use codex_app_server_protocol::InitializeParams;
|
||||
use codex_login::default_client::originator;
|
||||
use codex_plugin::AppConnectorId;
|
||||
use codex_plugin::PluginCapabilitySummary;
|
||||
@@ -37,7 +39,6 @@ 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::SubAgentSource;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::json;
|
||||
use std::collections::HashSet;
|
||||
@@ -316,43 +317,45 @@ async fn turn_started_then_completed_emits_turn_event() {
|
||||
|
||||
reducer
|
||||
.ingest(
|
||||
AnalyticsInput::TurnStarted(TurnStartedInput {
|
||||
tracking: tracking.clone(),
|
||||
turn_event: CodexTurnEvent {
|
||||
submission_type: None,
|
||||
model_provider: "openai".to_string(),
|
||||
sandbox_policy: SandboxPolicy::new_read_only_policy(),
|
||||
reasoning_effort: Some(ReasoningEffort::High),
|
||||
reasoning_summary: Some(ReasoningSummary::Detailed),
|
||||
service_tier: Some(ServiceTier::Flex),
|
||||
approval_policy: AskForApproval::OnRequest,
|
||||
approvals_reviewer: ApprovalsReviewer::GuardianSubagent,
|
||||
sandbox_network_access: true,
|
||||
collaboration_mode: ModeKind::Plan,
|
||||
personality: Some(Personality::Pragmatic),
|
||||
num_input_images: 2,
|
||||
is_first_turn: true,
|
||||
status: None,
|
||||
turn_error: None,
|
||||
steer_count: None,
|
||||
total_tool_call_count: None,
|
||||
shell_command_count: None,
|
||||
file_change_count: None,
|
||||
mcp_tool_call_count: None,
|
||||
dynamic_tool_call_count: None,
|
||||
subagent_tool_call_count: None,
|
||||
web_search_count: None,
|
||||
image_generation_count: None,
|
||||
input_tokens: None,
|
||||
cached_input_tokens: None,
|
||||
output_tokens: None,
|
||||
reasoning_output_tokens: None,
|
||||
total_tokens: None,
|
||||
duration_ms: None,
|
||||
started_at: None,
|
||||
completed_at: None,
|
||||
AnalyticsFact::Custom(CustomAnalyticsFact::TurnStarted(Box::new(
|
||||
TurnStartedInput {
|
||||
tracking: tracking.clone(),
|
||||
turn_event: CodexTurnEvent {
|
||||
submission_type: None,
|
||||
model_provider: "openai".to_string(),
|
||||
sandbox_policy: SandboxPolicy::new_read_only_policy(),
|
||||
reasoning_effort: Some(ReasoningEffort::High),
|
||||
reasoning_summary: Some(ReasoningSummary::Detailed),
|
||||
service_tier: Some(ServiceTier::Flex),
|
||||
approval_policy: AskForApproval::OnRequest,
|
||||
approvals_reviewer: ApprovalsReviewer::GuardianSubagent,
|
||||
sandbox_network_access: true,
|
||||
collaboration_mode: ModeKind::Plan,
|
||||
personality: Some(Personality::Pragmatic),
|
||||
num_input_images: 2,
|
||||
is_first_turn: true,
|
||||
status: None,
|
||||
turn_error: None,
|
||||
steer_count: None,
|
||||
total_tool_call_count: None,
|
||||
shell_command_count: None,
|
||||
file_change_count: None,
|
||||
mcp_tool_call_count: None,
|
||||
dynamic_tool_call_count: None,
|
||||
subagent_tool_call_count: None,
|
||||
web_search_count: None,
|
||||
image_generation_count: None,
|
||||
input_tokens: None,
|
||||
cached_input_tokens: None,
|
||||
output_tokens: None,
|
||||
reasoning_output_tokens: None,
|
||||
total_tokens: None,
|
||||
duration_ms: None,
|
||||
started_at: None,
|
||||
completed_at: None,
|
||||
},
|
||||
},
|
||||
}),
|
||||
))),
|
||||
&mut out,
|
||||
)
|
||||
.await;
|
||||
@@ -361,9 +364,9 @@ async fn turn_started_then_completed_emits_turn_event() {
|
||||
|
||||
reducer
|
||||
.ingest(
|
||||
AnalyticsInput::TurnCompleted(TurnCompletedInput {
|
||||
AnalyticsFact::Custom(CustomAnalyticsFact::TurnCompleted(TurnCompletedInput {
|
||||
turn_id: tracking.turn_id.clone(),
|
||||
}),
|
||||
})),
|
||||
&mut out,
|
||||
)
|
||||
.await;
|
||||
@@ -378,18 +381,14 @@ async fn turn_started_then_completed_emits_turn_event() {
|
||||
#[test]
|
||||
fn thread_initialized_event_serializes_expected_shape() {
|
||||
let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request(
|
||||
CodexThreadInitializedInput {
|
||||
"codex-tui".to_string(),
|
||||
ThreadInitializedInput {
|
||||
connection_id: 1,
|
||||
thread_id: "thread-0".to_string(),
|
||||
model: "gpt-5".to_string(),
|
||||
product_client_id: originator().value,
|
||||
created_at: 1_716_000_000,
|
||||
thread_context: CodexThreadContext {
|
||||
ephemeral: true,
|
||||
session_source: SessionSource::Exec,
|
||||
initialization_mode: InitializationMode::New,
|
||||
subagent_source: None,
|
||||
parent_thread_id: None,
|
||||
},
|
||||
ephemeral: true,
|
||||
session_source: SessionSource::Exec,
|
||||
initialization_mode: InitializationMode::New,
|
||||
},
|
||||
));
|
||||
|
||||
@@ -401,66 +400,84 @@ fn thread_initialized_event_serializes_expected_shape() {
|
||||
"event_type": "codex_thread_initialized",
|
||||
"event_params": {
|
||||
"thread_id": "thread-0",
|
||||
"product_client_id": originator().value,
|
||||
"product_client_id": "codex-tui",
|
||||
"model": "gpt-5",
|
||||
"ephemeral": true,
|
||||
"session_source": "user",
|
||||
"initialization_mode": "new",
|
||||
"subagent_source": null,
|
||||
"parent_thread_id": null,
|
||||
"created_at": 1716000000
|
||||
"created_at": payload["event_params"]["created_at"]
|
||||
}
|
||||
})
|
||||
);
|
||||
assert!(payload["event_params"]["created_at"].as_u64().is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn thread_initialized_event_serializes_subagent_source() {
|
||||
let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request(
|
||||
CodexThreadInitializedInput {
|
||||
thread_id: "thread-1".to_string(),
|
||||
model: "gpt-5".to_string(),
|
||||
product_client_id: originator().value,
|
||||
created_at: 1,
|
||||
thread_context: CodexThreadContext {
|
||||
ephemeral: false,
|
||||
session_source: SessionSource::SubAgent(SubAgentSource::Review),
|
||||
initialization_mode: InitializationMode::New,
|
||||
subagent_source: Some(SubAgentSource::Review),
|
||||
parent_thread_id: None,
|
||||
#[tokio::test]
|
||||
async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialized() {
|
||||
let mut reducer = AnalyticsReducer::default();
|
||||
let mut events = Vec::new();
|
||||
|
||||
reducer
|
||||
.ingest(
|
||||
AnalyticsFact::Custom(CustomAnalyticsFact::ThreadInitialized(
|
||||
ThreadInitializedInput {
|
||||
connection_id: 7,
|
||||
thread_id: "thread-no-client".to_string(),
|
||||
model: "gpt-5".to_string(),
|
||||
ephemeral: false,
|
||||
session_source: SessionSource::Exec,
|
||||
initialization_mode: InitializationMode::New,
|
||||
},
|
||||
)),
|
||||
&mut events,
|
||||
)
|
||||
.await;
|
||||
assert!(events.is_empty(), "thread events should require initialize");
|
||||
|
||||
reducer
|
||||
.ingest(
|
||||
AnalyticsFact::Initialize {
|
||||
connection_id: 7,
|
||||
params: InitializeParams {
|
||||
client_info: ClientInfo {
|
||||
name: "codex-tui".to_string(),
|
||||
title: None,
|
||||
version: "1.0.0".to_string(),
|
||||
},
|
||||
capabilities: None,
|
||||
},
|
||||
},
|
||||
},
|
||||
));
|
||||
&mut events,
|
||||
)
|
||||
.await;
|
||||
assert!(events.is_empty(), "initialize should not publish by itself");
|
||||
|
||||
let payload =
|
||||
serde_json::to_value(&event).expect("serialize subagent thread initialized event");
|
||||
assert_eq!(payload["event_params"]["session_source"], "subagent");
|
||||
assert_eq!(payload["event_params"]["subagent_source"], "review");
|
||||
}
|
||||
reducer
|
||||
.ingest(
|
||||
AnalyticsFact::Custom(CustomAnalyticsFact::ThreadInitialized(
|
||||
ThreadInitializedInput {
|
||||
connection_id: 7,
|
||||
thread_id: "thread-1".to_string(),
|
||||
model: "gpt-5".to_string(),
|
||||
ephemeral: true,
|
||||
session_source: SessionSource::Exec,
|
||||
initialization_mode: InitializationMode::Resumed,
|
||||
},
|
||||
)),
|
||||
&mut events,
|
||||
)
|
||||
.await;
|
||||
|
||||
#[test]
|
||||
fn thread_initialized_event_omits_non_user_non_subagent_session_source() {
|
||||
let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request(
|
||||
CodexThreadInitializedInput {
|
||||
thread_id: "thread-2".to_string(),
|
||||
model: "gpt-5".to_string(),
|
||||
product_client_id: originator().value,
|
||||
created_at: 1,
|
||||
thread_context: CodexThreadContext {
|
||||
ephemeral: false,
|
||||
session_source: SessionSource::Mcp,
|
||||
initialization_mode: InitializationMode::New,
|
||||
subagent_source: None,
|
||||
parent_thread_id: None,
|
||||
},
|
||||
},
|
||||
));
|
||||
|
||||
let payload = serde_json::to_value(&event).expect("serialize mcp thread initialized event");
|
||||
assert_eq!(
|
||||
payload["event_params"]["session_source"],
|
||||
serde_json::Value::Null
|
||||
);
|
||||
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_thread_initialized");
|
||||
assert_eq!(payload[0]["event_params"]["product_client_id"], "codex-tui");
|
||||
assert_eq!(payload[0]["event_params"]["initialization_mode"], "resumed");
|
||||
assert_eq!(payload[0]["event_params"]["session_source"], "user");
|
||||
assert_eq!(payload[0]["event_params"]["subagent_source"], json!(null));
|
||||
assert_eq!(payload[0]["event_params"]["parent_thread_id"], json!(null));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -1,14 +1,13 @@
|
||||
mod analytics_client;
|
||||
|
||||
pub use analytics_client::AnalyticsEventsClient;
|
||||
pub use analytics_client::AnalyticsInput;
|
||||
pub use analytics_client::AnalyticsFact;
|
||||
pub use analytics_client::AnalyticsReducer;
|
||||
pub use analytics_client::AppInvocation;
|
||||
pub use analytics_client::AppMentionedInput;
|
||||
pub use analytics_client::AppUsedInput;
|
||||
pub use analytics_client::CodexThreadContext;
|
||||
pub use analytics_client::CodexThreadInitializedInput;
|
||||
pub use analytics_client::CodexTurnEvent;
|
||||
pub use analytics_client::CustomAnalyticsFact;
|
||||
pub use analytics_client::InitializationMode;
|
||||
pub use analytics_client::InvocationType;
|
||||
pub use analytics_client::PluginState;
|
||||
@@ -16,6 +15,7 @@ pub use analytics_client::PluginStateChangedInput;
|
||||
pub use analytics_client::PluginUsedInput;
|
||||
pub use analytics_client::SkillInvocation;
|
||||
pub use analytics_client::SkillInvokedInput;
|
||||
pub use analytics_client::ThreadInitializedInput;
|
||||
pub use analytics_client::TrackEventsContext;
|
||||
pub use analytics_client::TurnCompletedInput;
|
||||
pub use analytics_client::TurnStartedInput;
|
||||
|
||||
@@ -179,16 +179,19 @@ use codex_arg0::Arg0DispatchPaths;
|
||||
use codex_backend_client::Client as BackendClient;
|
||||
use codex_chatgpt::connectors;
|
||||
use codex_cloud_requirements::cloud_requirements_loader;
|
||||
use codex_core::AnalyticsEventsClient;
|
||||
use codex_core::AuthManager;
|
||||
use codex_core::CodexAuth;
|
||||
use codex_core::CodexThread;
|
||||
use codex_core::Cursor as RolloutCursor;
|
||||
use codex_core::ForkSnapshot;
|
||||
use codex_core::InitializationMode;
|
||||
use codex_core::NewThread;
|
||||
use codex_core::RolloutRecorder;
|
||||
use codex_core::SessionMeta;
|
||||
use codex_core::SteerInputError;
|
||||
use codex_core::ThreadConfigSnapshot;
|
||||
use codex_core::ThreadInitializedInput;
|
||||
use codex_core::ThreadManager;
|
||||
use codex_core::ThreadSortKey as CoreThreadSortKey;
|
||||
use codex_core::auth::AuthMode as CoreAuthMode;
|
||||
@@ -374,6 +377,7 @@ pub(crate) struct CodexMessageProcessor {
|
||||
auth_manager: Arc<AuthManager>,
|
||||
thread_manager: Arc<ThreadManager>,
|
||||
outgoing: Arc<OutgoingMessageSender>,
|
||||
analytics_events_client: AnalyticsEventsClient,
|
||||
arg0_paths: Arg0DispatchPaths,
|
||||
config: Arc<Config>,
|
||||
cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
|
||||
@@ -404,6 +408,7 @@ struct ListenerTaskContext {
|
||||
thread_manager: Arc<ThreadManager>,
|
||||
thread_state_manager: ThreadStateManager,
|
||||
outgoing: Arc<OutgoingMessageSender>,
|
||||
analytics_events_client: AnalyticsEventsClient,
|
||||
thread_watch_manager: ThreadWatchManager,
|
||||
fallback_model_provider: String,
|
||||
codex_home: PathBuf,
|
||||
@@ -426,6 +431,7 @@ pub(crate) struct CodexMessageProcessorArgs {
|
||||
pub(crate) auth_manager: Arc<AuthManager>,
|
||||
pub(crate) thread_manager: Arc<ThreadManager>,
|
||||
pub(crate) outgoing: Arc<OutgoingMessageSender>,
|
||||
pub(crate) analytics_events_client: AnalyticsEventsClient,
|
||||
pub(crate) arg0_paths: Arg0DispatchPaths,
|
||||
pub(crate) config: Arc<Config>,
|
||||
pub(crate) cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
|
||||
@@ -490,6 +496,7 @@ impl CodexMessageProcessor {
|
||||
auth_manager,
|
||||
thread_manager,
|
||||
outgoing,
|
||||
analytics_events_client,
|
||||
arg0_paths,
|
||||
config,
|
||||
cli_overrides,
|
||||
@@ -502,6 +509,7 @@ impl CodexMessageProcessor {
|
||||
auth_manager,
|
||||
thread_manager,
|
||||
outgoing: outgoing.clone(),
|
||||
analytics_events_client,
|
||||
arg0_paths,
|
||||
config,
|
||||
cli_overrides,
|
||||
@@ -1921,6 +1929,7 @@ impl CodexMessageProcessor {
|
||||
thread_manager: Arc::clone(&self.thread_manager),
|
||||
thread_state_manager: self.thread_state_manager.clone(),
|
||||
outgoing: Arc::clone(&self.outgoing),
|
||||
analytics_events_client: self.analytics_events_client.clone(),
|
||||
thread_watch_manager: self.thread_watch_manager.clone(),
|
||||
fallback_model_provider: self.config.model_provider_id.clone(),
|
||||
codex_home: self.config.codex_home.clone(),
|
||||
@@ -2153,6 +2162,14 @@ impl CodexMessageProcessor {
|
||||
sandbox: config_snapshot.sandbox_policy.into(),
|
||||
reasoning_effort: config_snapshot.reasoning_effort,
|
||||
};
|
||||
listener_task_context
|
||||
.analytics_events_client
|
||||
.track_thread_initialized(thread_initialize_input(
|
||||
request_id.connection_id,
|
||||
&thread,
|
||||
response.model.clone(),
|
||||
InitializationMode::New,
|
||||
));
|
||||
|
||||
listener_task_context
|
||||
.outgoing
|
||||
@@ -3631,6 +3648,13 @@ impl CodexMessageProcessor {
|
||||
sandbox: session_configured.sandbox_policy.into(),
|
||||
reasoning_effort: session_configured.reasoning_effort,
|
||||
};
|
||||
self.analytics_events_client
|
||||
.track_thread_initialized(thread_initialize_input(
|
||||
request_id.connection_id,
|
||||
&response.thread,
|
||||
response.model.clone(),
|
||||
InitializationMode::Resumed,
|
||||
));
|
||||
|
||||
self.outgoing.send_response(request_id, response).await;
|
||||
}
|
||||
@@ -4238,6 +4262,13 @@ impl CodexMessageProcessor {
|
||||
sandbox: session_configured.sandbox_policy.into(),
|
||||
reasoning_effort: session_configured.reasoning_effort,
|
||||
};
|
||||
self.analytics_events_client
|
||||
.track_thread_initialized(thread_initialize_input(
|
||||
request_id.connection_id,
|
||||
&thread,
|
||||
response.model.clone(),
|
||||
InitializationMode::Forked,
|
||||
));
|
||||
|
||||
self.outgoing.send_response(request_id, response).await;
|
||||
|
||||
@@ -6820,6 +6851,7 @@ impl CodexMessageProcessor {
|
||||
thread_manager: Arc::clone(&self.thread_manager),
|
||||
thread_state_manager: self.thread_state_manager.clone(),
|
||||
outgoing: Arc::clone(&self.outgoing),
|
||||
analytics_events_client: self.analytics_events_client.clone(),
|
||||
thread_watch_manager: self.thread_watch_manager.clone(),
|
||||
fallback_model_provider: self.config.model_provider_id.clone(),
|
||||
codex_home: self.config.codex_home.clone(),
|
||||
@@ -6907,6 +6939,7 @@ impl CodexMessageProcessor {
|
||||
thread_manager: Arc::clone(&self.thread_manager),
|
||||
thread_state_manager: self.thread_state_manager.clone(),
|
||||
outgoing: Arc::clone(&self.outgoing),
|
||||
analytics_events_client: self.analytics_events_client.clone(),
|
||||
thread_watch_manager: self.thread_watch_manager.clone(),
|
||||
fallback_model_provider: self.config.model_provider_id.clone(),
|
||||
codex_home: self.config.codex_home.clone(),
|
||||
@@ -6938,6 +6971,7 @@ impl CodexMessageProcessor {
|
||||
outgoing,
|
||||
thread_manager,
|
||||
thread_state_manager,
|
||||
analytics_events_client: _,
|
||||
thread_watch_manager,
|
||||
fallback_model_provider,
|
||||
codex_home,
|
||||
@@ -7855,6 +7889,22 @@ fn cloud_requirements_load_error(err: &std::io::Error) -> Option<&CloudRequireme
|
||||
None
|
||||
}
|
||||
|
||||
fn thread_initialize_input(
|
||||
connection_id: ConnectionId,
|
||||
thread: &Thread,
|
||||
model: String,
|
||||
initialization_mode: InitializationMode,
|
||||
) -> ThreadInitializedInput {
|
||||
ThreadInitializedInput {
|
||||
connection_id: connection_id.0,
|
||||
thread_id: thread.id.clone(),
|
||||
model,
|
||||
ephemeral: thread.ephemeral,
|
||||
session_source: thread.source.clone().into(),
|
||||
initialization_mode,
|
||||
}
|
||||
}
|
||||
|
||||
fn config_load_error(err: &std::io::Error) -> JSONRPCErrorError {
|
||||
let data = cloud_requirements_load_error(err).map(|cloud_error| {
|
||||
let mut data = serde_json::json!({
|
||||
|
||||
@@ -158,6 +158,7 @@ pub(crate) struct MessageProcessor {
|
||||
external_agent_config_api: ExternalAgentConfigApi,
|
||||
fs_api: FsApi,
|
||||
auth_manager: Arc<AuthManager>,
|
||||
analytics_events_client: AnalyticsEventsClient,
|
||||
fs_watch_manager: FsWatchManager,
|
||||
config: Arc<Config>,
|
||||
config_warnings: Arc<Vec<ConfigWarningNotification>>,
|
||||
@@ -238,6 +239,7 @@ impl MessageProcessor {
|
||||
auth_manager: auth_manager.clone(),
|
||||
thread_manager: Arc::clone(&thread_manager),
|
||||
outgoing: outgoing.clone(),
|
||||
analytics_events_client: analytics_events_client.clone(),
|
||||
arg0_paths,
|
||||
config: Arc::clone(&config),
|
||||
cli_overrides: cli_overrides.clone(),
|
||||
@@ -258,7 +260,7 @@ impl MessageProcessor {
|
||||
loader_overrides,
|
||||
cloud_requirements,
|
||||
thread_manager,
|
||||
analytics_events_client,
|
||||
analytics_events_client.clone(),
|
||||
);
|
||||
let external_agent_config_api = ExternalAgentConfigApi::new(config.codex_home.clone());
|
||||
let fs_api = FsApi::default();
|
||||
@@ -271,6 +273,7 @@ impl MessageProcessor {
|
||||
external_agent_config_api,
|
||||
fs_api,
|
||||
auth_manager,
|
||||
analytics_events_client,
|
||||
fs_watch_manager,
|
||||
config,
|
||||
config_warnings: Arc::new(config_warnings),
|
||||
@@ -543,6 +546,7 @@ impl MessageProcessor {
|
||||
// shared thread when another connected client did not opt into
|
||||
// experimental API). Proposed direction is instance-global first-write-wins
|
||||
// with initialize-time mismatch rejection.
|
||||
let analytics_initialize_params = params.clone();
|
||||
let (experimental_api_enabled, opt_out_notification_methods) =
|
||||
match params.capabilities {
|
||||
Some(capabilities) => (
|
||||
@@ -570,7 +574,7 @@ impl MessageProcessor {
|
||||
} else {
|
||||
name.clone()
|
||||
};
|
||||
if let Err(error) = set_default_originator(originator) {
|
||||
if let Err(error) = set_default_originator(originator.clone()) {
|
||||
match error {
|
||||
SetOriginatorError::InvalidHeaderValue => {
|
||||
let error = JSONRPCErrorError {
|
||||
@@ -593,6 +597,8 @@ impl MessageProcessor {
|
||||
}
|
||||
}
|
||||
}
|
||||
self.analytics_events_client
|
||||
.track_initialize(connection_id.0, analytics_initialize_params);
|
||||
set_default_client_residency_requirement(self.config.enforce_residency.value());
|
||||
let user_agent_suffix = format!("{name}; {version}");
|
||||
if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
|
||||
|
||||
@@ -1,10 +1,23 @@
|
||||
use anyhow::Result;
|
||||
use app_test_support::ChatGptAuthFixture;
|
||||
use app_test_support::DEFAULT_CLIENT_NAME;
|
||||
use app_test_support::write_chatgpt_auth;
|
||||
use codex_core::auth::AuthCredentialsStoreMode;
|
||||
use codex_core::config::ConfigBuilder;
|
||||
use codex_core::config::types::OtelExporterKind;
|
||||
use codex_core::config::types::OtelHttpProtocol;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::Value;
|
||||
use std::collections::HashMap;
|
||||
use std::path::Path;
|
||||
use std::time::Duration;
|
||||
use tempfile::TempDir;
|
||||
use tokio::time::timeout;
|
||||
use wiremock::Mock;
|
||||
use wiremock::MockServer;
|
||||
use wiremock::ResponseTemplate;
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::matchers::path;
|
||||
|
||||
const SERVICE_VERSION: &str = "0.0.0-test";
|
||||
|
||||
@@ -65,3 +78,83 @@ async fn app_server_default_analytics_enabled_with_flag() -> Result<()> {
|
||||
assert_eq!(has_metrics, true);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) async fn enable_analytics_capture(server: &MockServer, codex_home: &Path) -> Result<()> {
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/codex/analytics-events/events"))
|
||||
.respond_with(ResponseTemplate::new(200))
|
||||
.mount(server)
|
||||
.await;
|
||||
|
||||
write_chatgpt_auth(
|
||||
codex_home,
|
||||
ChatGptAuthFixture::new("chatgpt-token")
|
||||
.account_id("account-123")
|
||||
.chatgpt_user_id("user-123")
|
||||
.chatgpt_account_id("account-123"),
|
||||
AuthCredentialsStoreMode::File,
|
||||
)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) async fn wait_for_analytics_payload(
|
||||
server: &MockServer,
|
||||
read_timeout: Duration,
|
||||
) -> Result<Value> {
|
||||
let body = timeout(read_timeout, async {
|
||||
loop {
|
||||
let Some(requests) = server.received_requests().await else {
|
||||
tokio::time::sleep(Duration::from_millis(25)).await;
|
||||
continue;
|
||||
};
|
||||
if let Some(request) = requests.iter().find(|request| {
|
||||
request.method == "POST" && request.url.path() == "/codex/analytics-events/events"
|
||||
}) {
|
||||
break request.body.clone();
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(25)).await;
|
||||
}
|
||||
})
|
||||
.await?;
|
||||
serde_json::from_slice(&body).map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}"))
|
||||
}
|
||||
|
||||
pub(crate) fn thread_initialized_event(payload: &Value) -> Result<&Value> {
|
||||
let events = payload["events"]
|
||||
.as_array()
|
||||
.ok_or_else(|| anyhow::anyhow!("analytics payload missing events array"))?;
|
||||
events
|
||||
.iter()
|
||||
.find(|event| event["event_type"] == "codex_thread_initialized")
|
||||
.ok_or_else(|| anyhow::anyhow!("codex_thread_initialized event should be present"))
|
||||
}
|
||||
|
||||
pub(crate) fn assert_basic_thread_initialized_event(
|
||||
event: &Value,
|
||||
thread_id: &str,
|
||||
expected_model: &str,
|
||||
initialization_mode: &str,
|
||||
) {
|
||||
assert_eq!(event["event_params"]["thread_id"], thread_id);
|
||||
assert_eq!(
|
||||
event["event_params"]["product_client_id"],
|
||||
DEFAULT_CLIENT_NAME
|
||||
);
|
||||
assert_eq!(event["event_params"]["model"], expected_model);
|
||||
assert_eq!(event["event_params"]["ephemeral"], false);
|
||||
assert_eq!(event["event_params"]["session_source"], "user");
|
||||
assert_eq!(
|
||||
event["event_params"]["subagent_source"],
|
||||
serde_json::Value::Null
|
||||
);
|
||||
assert_eq!(
|
||||
event["event_params"]["parent_thread_id"],
|
||||
serde_json::Value::Null
|
||||
);
|
||||
assert_eq!(
|
||||
event["event_params"]["initialization_mode"],
|
||||
initialization_mode
|
||||
);
|
||||
assert!(event["event_params"]["created_at"].as_u64().is_some());
|
||||
}
|
||||
|
||||
@@ -38,6 +38,11 @@ use wiremock::ResponseTemplate;
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::matchers::path;
|
||||
|
||||
use super::analytics::assert_basic_thread_initialized_event;
|
||||
use super::analytics::enable_analytics_capture;
|
||||
use super::analytics::thread_initialized_event;
|
||||
use super::analytics::wait_for_analytics_payload;
|
||||
|
||||
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
|
||||
|
||||
#[tokio::test]
|
||||
@@ -177,6 +182,45 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
|
||||
let codex_home = TempDir::new()?;
|
||||
create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?;
|
||||
enable_analytics_capture(&server, codex_home.path()).await?;
|
||||
|
||||
let conversation_id = create_fake_rollout(
|
||||
codex_home.path(),
|
||||
"2025-01-05T12-00-00",
|
||||
"2025-01-05T12:00:00Z",
|
||||
"Saved user message",
|
||||
Some("mock_provider"),
|
||||
None,
|
||||
)?;
|
||||
|
||||
let mut mcp = McpProcess::new(codex_home.path()).await?;
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||
|
||||
let fork_id = mcp
|
||||
.send_thread_fork_request(ThreadForkParams {
|
||||
thread_id: conversation_id,
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let fork_resp: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
|
||||
)
|
||||
.await??;
|
||||
let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
|
||||
|
||||
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
|
||||
let event = thread_initialized_event(&payload)?;
|
||||
assert_basic_thread_initialized_event(event, &thread.id, "mock-model", "forked");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
|
||||
@@ -70,6 +70,11 @@ use wiremock::ResponseTemplate;
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::matchers::path;
|
||||
|
||||
use super::analytics::assert_basic_thread_initialized_event;
|
||||
use super::analytics::enable_analytics_capture;
|
||||
use super::analytics::thread_initialized_event;
|
||||
use super::analytics::wait_for_analytics_payload;
|
||||
|
||||
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
|
||||
const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals.";
|
||||
|
||||
@@ -150,6 +155,46 @@ async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
|
||||
let codex_home = TempDir::new()?;
|
||||
create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?;
|
||||
enable_analytics_capture(&server, codex_home.path()).await?;
|
||||
|
||||
let conversation_id = create_fake_rollout_with_text_elements(
|
||||
codex_home.path(),
|
||||
"2025-01-05T12-00-00",
|
||||
"2025-01-05T12:00:00Z",
|
||||
"Saved user message",
|
||||
Vec::new(),
|
||||
Some("mock_provider"),
|
||||
None,
|
||||
)?;
|
||||
|
||||
let mut mcp = McpProcess::new(codex_home.path()).await?;
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||
|
||||
let resume_id = mcp
|
||||
.send_thread_resume_request(ThreadResumeParams {
|
||||
thread_id: conversation_id,
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let resume_resp: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
|
||||
)
|
||||
.await??;
|
||||
let ThreadResumeResponse { thread, .. } = to_response::<ThreadResumeResponse>(resume_resp)?;
|
||||
|
||||
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
|
||||
let event = thread_initialized_event(&payload)?;
|
||||
assert_basic_thread_initialized_event(event, &thread.id, "gpt-5.2-codex", "resumed");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_resume_returns_rollout_history() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
|
||||
@@ -34,6 +34,11 @@ use wiremock::ResponseTemplate;
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::matchers::path;
|
||||
|
||||
use super::analytics::assert_basic_thread_initialized_event;
|
||||
use super::analytics::enable_analytics_capture;
|
||||
use super::analytics::thread_initialized_event;
|
||||
use super::analytics::wait_for_analytics_payload;
|
||||
|
||||
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
|
||||
|
||||
#[tokio::test]
|
||||
@@ -153,6 +158,34 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
|
||||
let codex_home = TempDir::new()?;
|
||||
create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?;
|
||||
enable_analytics_capture(&server, codex_home.path()).await?;
|
||||
|
||||
let mut mcp = McpProcess::new(codex_home.path()).await?;
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||
|
||||
let req_id = mcp
|
||||
.send_thread_start_request(ThreadStartParams::default())
|
||||
.await?;
|
||||
let resp: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
|
||||
)
|
||||
.await??;
|
||||
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(resp)?;
|
||||
|
||||
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
|
||||
assert_eq!(payload["events"].as_array().expect("events array").len(), 1);
|
||||
let event = thread_initialized_event(&payload)?;
|
||||
assert_basic_thread_initialized_event(event, &thread.id, "mock-model", "new");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_start_respects_project_config_from_cwd() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
|
||||
@@ -5,8 +5,6 @@ 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;
|
||||
@@ -352,10 +350,7 @@ use crate::util::backoff;
|
||||
use crate::windows_sandbox::WindowsSandboxLevelExt;
|
||||
use codex_analytics::AnalyticsEventsClient;
|
||||
use codex_analytics::AppInvocation;
|
||||
use codex_analytics::CodexThreadContext;
|
||||
use codex_analytics::CodexThreadInitializedInput;
|
||||
use codex_analytics::CodexTurnEvent;
|
||||
use codex_analytics::InitializationMode;
|
||||
use codex_analytics::InvocationType;
|
||||
use codex_analytics::TurnSubmissionType;
|
||||
use codex_analytics::build_track_events_context;
|
||||
@@ -641,12 +636,6 @@ impl Codex {
|
||||
user_shell_override,
|
||||
};
|
||||
|
||||
let initialization_mode = initialization_mode(&conversation_history);
|
||||
let thread_session_source = session_configuration.session_source.clone();
|
||||
let subagent_source = session_source_subagent_source(&thread_session_source);
|
||||
let parent_thread_id = session_source_parent_thread_id(&thread_session_source);
|
||||
let thread_initialized_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);
|
||||
@@ -674,31 +663,6 @@ impl Codex {
|
||||
})?;
|
||||
let thread_id = session.conversation_id;
|
||||
|
||||
session
|
||||
.services
|
||||
.analytics_events_client
|
||||
.track_thread_initialized(CodexThreadInitializedInput {
|
||||
thread_id: thread_id.to_string(),
|
||||
model: thread_initialized_configuration
|
||||
.collaboration_mode
|
||||
.model()
|
||||
.to_string(),
|
||||
product_client_id: crate::default_client::originator().value,
|
||||
created_at: SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs(),
|
||||
thread_context: CodexThreadContext {
|
||||
ephemeral: thread_initialized_configuration
|
||||
.original_config_do_not_use
|
||||
.ephemeral,
|
||||
session_source: thread_session_source,
|
||||
initialization_mode,
|
||||
subagent_source,
|
||||
parent_thread_id,
|
||||
},
|
||||
});
|
||||
|
||||
// This task will run until Op::Shutdown is received.
|
||||
let session_for_loop = Arc::clone(&session);
|
||||
let session_loop_handle = tokio::spawn(async move {
|
||||
@@ -843,7 +807,6 @@ fn turn_submission_type(submission_type: SubmissionType) -> TurnSubmissionType {
|
||||
SubmissionType::PromptQueued => TurnSubmissionType::Queued,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn completed_session_loop_termination() -> SessionLoopTermination {
|
||||
futures::future::ready(()).boxed().shared()
|
||||
|
||||
@@ -136,6 +136,10 @@ pub type CodexConversation = CodexThread;
|
||||
pub use auth::AuthManager;
|
||||
pub use auth::CodexAuth;
|
||||
pub use codex_analytics::AnalyticsEventsClient;
|
||||
pub use codex_analytics::AnalyticsFact;
|
||||
pub use codex_analytics::CustomAnalyticsFact;
|
||||
pub use codex_analytics::InitializationMode;
|
||||
pub use codex_analytics::ThreadInitializedInput;
|
||||
mod default_client_forwarding;
|
||||
|
||||
/// Default Codex HTTP client headers and reqwest construction.
|
||||
|
||||
@@ -1,78 +0,0 @@
|
||||
use anyhow::Result;
|
||||
use codex_core::CodexAuth;
|
||||
use codex_core::config::Constrained;
|
||||
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_initialization_tracks_thread_initialized_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.permissions.sandbox_policy = Constrained::allow_any(
|
||||
codex_protocol::protocol::SandboxPolicy::new_workspace_write_policy(),
|
||||
);
|
||||
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_initialized")
|
||||
.expect("codex_thread_initialized 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"]["ephemeral"], true);
|
||||
assert_eq!(event["event_params"]["session_source"], "user");
|
||||
assert_eq!(event["event_params"]["initialization_mode"], "new");
|
||||
assert_eq!(
|
||||
event["event_params"]["subagent_source"],
|
||||
serde_json::Value::Null
|
||||
);
|
||||
assert_eq!(
|
||||
event["event_params"]["parent_thread_id"],
|
||||
serde_json::Value::Null
|
||||
);
|
||||
assert!(
|
||||
event["event_params"]["created_at"]
|
||||
.as_u64()
|
||||
.is_some_and(|timestamp| timestamp > 0)
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -58,7 +58,6 @@ pub static CODEX_ALIASES_TEMP_DIR: TestCodexAliasesGuard = unsafe {
|
||||
mod abort_tasks;
|
||||
mod agent_jobs;
|
||||
mod agent_websocket;
|
||||
mod analytics;
|
||||
mod apply_patch_cli;
|
||||
#[cfg(not(target_os = "windows"))]
|
||||
mod approvals;
|
||||
|
||||
@@ -427,7 +427,7 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result
|
||||
config_warnings,
|
||||
session_source: SessionSource::Exec,
|
||||
enable_codex_api_key_env: true,
|
||||
client_name: "codex-exec".to_string(),
|
||||
client_name: "codex_exec".to_string(),
|
||||
client_version: env!("CARGO_PKG_VERSION").to_string(),
|
||||
experimental_api: true,
|
||||
opt_out_notification_methods: Vec::new(),
|
||||
|
||||
Reference in New Issue
Block a user