mirror of
https://github.com/openai/codex.git
synced 2026-11-05 20:10:33 +00:00
codex: keep thread timing conditioning in analytics
This commit is contained in:
@@ -66,7 +66,6 @@ use crate::facts::SkillInvokedInput;
|
||||
use crate::facts::SubAgentThreadStartedInput;
|
||||
use crate::facts::ThreadInitializationMode;
|
||||
use crate::facts::ThreadStartTimingFact;
|
||||
use crate::facts::ThreadStartType;
|
||||
use crate::facts::TrackEventsContext;
|
||||
use crate::facts::TurnResolvedConfigFact;
|
||||
use crate::facts::TurnStatus;
|
||||
@@ -1332,7 +1331,6 @@ fn thread_initialized_event_serializes_expected_shape() {
|
||||
subagent_source: None,
|
||||
parent_thread_id: None,
|
||||
thread_start_timing: ThreadStartTimingEventParams {
|
||||
thread_start_type: Some(ThreadStartType::New),
|
||||
thread_start_duration_ms: Some(321),
|
||||
thread_start_prepare_duration_ms: Some(111),
|
||||
thread_start_spawn_duration_ms: Some(123),
|
||||
@@ -1369,7 +1367,6 @@ fn thread_initialized_event_serializes_expected_shape() {
|
||||
"initialization_mode": "new",
|
||||
"subagent_source": null,
|
||||
"parent_thread_id": null,
|
||||
"thread_start_type": "new",
|
||||
"thread_start_duration_ms": 321,
|
||||
"thread_start_prepare_duration_ms": 111,
|
||||
"thread_start_spawn_duration_ms": 123,
|
||||
@@ -1688,7 +1685,6 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize
|
||||
payload[0]["event_params"]["runtime"]["runtime_arch"],
|
||||
"x86_64"
|
||||
);
|
||||
assert_eq!(payload[0]["event_params"]["thread_start_type"], json!(null));
|
||||
assert_eq!(
|
||||
payload[0]["event_params"]["thread_start_duration_ms"],
|
||||
json!(null)
|
||||
@@ -1757,7 +1753,6 @@ async fn thread_start_timing_fact_enriches_thread_initialized_event() {
|
||||
AnalyticsFact::Custom(CustomAnalyticsFact::ThreadStartTiming(
|
||||
ThreadStartTimingFact {
|
||||
thread_id: "thread-1".to_string(),
|
||||
thread_start_type: ThreadStartType::Cleared,
|
||||
duration_ms: 222,
|
||||
prepare_duration_ms: 12,
|
||||
spawn_duration_ms: 123,
|
||||
@@ -1782,72 +1777,6 @@ async fn thread_start_timing_fact_enriches_thread_initialized_event() {
|
||||
|
||||
let payload = serde_json::to_value(&events).expect("serialize events");
|
||||
assert_eq!(payload[0]["event_type"], json!("codex_thread_initialized"));
|
||||
assert_eq!(
|
||||
payload[0]["event_params"]["thread_start_type"],
|
||||
json!("cleared")
|
||||
);
|
||||
assert_eq!(
|
||||
payload[0]["event_params"]["thread_start_duration_ms"],
|
||||
json!(222)
|
||||
);
|
||||
assert_eq!(
|
||||
payload[0]["event_params"]["thread_start_prepare_duration_ms"],
|
||||
json!(12)
|
||||
);
|
||||
assert_eq!(
|
||||
payload[0]["event_params"]["thread_start_spawn_duration_ms"],
|
||||
json!(123)
|
||||
);
|
||||
assert_eq!(
|
||||
payload[0]["event_params"]["thread_start_finalize_duration_ms"],
|
||||
json!(87)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_start_timing_fact_enriches_resumed_thread_initialized_event() {
|
||||
let mut reducer = AnalyticsReducer::default();
|
||||
let mut events = Vec::new();
|
||||
|
||||
ingest_initialize(&mut reducer, &mut events).await;
|
||||
reducer
|
||||
.ingest(
|
||||
AnalyticsFact::Custom(CustomAnalyticsFact::ThreadStartTiming(
|
||||
ThreadStartTimingFact {
|
||||
thread_id: "thread-1".to_string(),
|
||||
thread_start_type: ThreadStartType::Resumed,
|
||||
duration_ms: 222,
|
||||
prepare_duration_ms: 12,
|
||||
spawn_duration_ms: 123,
|
||||
finalize_duration_ms: 87,
|
||||
},
|
||||
)),
|
||||
&mut events,
|
||||
)
|
||||
.await;
|
||||
reducer
|
||||
.ingest(
|
||||
AnalyticsFact::ClientResponse {
|
||||
connection_id: 7,
|
||||
request_id: RequestId::Integer(2),
|
||||
response: Box::new(sample_thread_resume_response(
|
||||
"thread-1", /*ephemeral*/ true, "gpt-5",
|
||||
)),
|
||||
},
|
||||
&mut events,
|
||||
)
|
||||
.await;
|
||||
|
||||
let payload = serde_json::to_value(&events).expect("serialize events");
|
||||
assert_eq!(payload[0]["event_type"], json!("codex_thread_initialized"));
|
||||
assert_eq!(
|
||||
payload[0]["event_params"]["initialization_mode"],
|
||||
json!("resumed")
|
||||
);
|
||||
assert_eq!(
|
||||
payload[0]["event_params"]["thread_start_type"],
|
||||
json!("resumed")
|
||||
);
|
||||
assert_eq!(
|
||||
payload[0]["event_params"]["thread_start_duration_ms"],
|
||||
json!(222)
|
||||
|
||||
@@ -14,7 +14,6 @@ use crate::facts::InvocationType;
|
||||
use crate::facts::PluginState;
|
||||
use crate::facts::SubAgentThreadStartedInput;
|
||||
use crate::facts::ThreadInitializationMode;
|
||||
use crate::facts::ThreadStartType;
|
||||
use crate::facts::TrackEventsContext;
|
||||
use crate::facts::TurnStatus;
|
||||
use crate::facts::TurnSteerRejectionReason;
|
||||
@@ -162,7 +161,6 @@ pub(crate) struct CodexAppServerStartedEventRequest {
|
||||
|
||||
#[derive(Default, Serialize)]
|
||||
pub(crate) struct ThreadStartTimingEventParams {
|
||||
pub(crate) thread_start_type: Option<ThreadStartType>,
|
||||
pub(crate) thread_start_duration_ms: Option<u64>,
|
||||
pub(crate) thread_start_prepare_duration_ms: Option<u64>,
|
||||
pub(crate) thread_start_spawn_duration_ms: Option<u64>,
|
||||
|
||||
@@ -92,15 +92,6 @@ pub enum ThreadInitializationMode {
|
||||
Resumed,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum ThreadStartType {
|
||||
New,
|
||||
Cleared,
|
||||
Forked,
|
||||
Resumed,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct TurnTokenUsageFact {
|
||||
pub turn_id: String,
|
||||
@@ -292,7 +283,6 @@ pub(crate) struct AppServerStartedInput {
|
||||
|
||||
pub struct ThreadStartTimingFact {
|
||||
pub thread_id: String,
|
||||
pub thread_start_type: ThreadStartType,
|
||||
pub duration_ms: u64,
|
||||
pub prepare_duration_ms: u64,
|
||||
pub spawn_duration_ms: u64,
|
||||
|
||||
@@ -39,7 +39,6 @@ pub use facts::SkillInvocation;
|
||||
pub use facts::SubAgentThreadStartedInput;
|
||||
pub use facts::ThreadInitializationMode;
|
||||
pub use facts::ThreadStartTimingFact;
|
||||
pub use facts::ThreadStartType;
|
||||
pub use facts::TrackEventsContext;
|
||||
pub use facts::TurnResolvedConfigFact;
|
||||
pub use facts::TurnStatus;
|
||||
|
||||
@@ -76,7 +76,6 @@ use crate::facts::SkillInvokedInput;
|
||||
use crate::facts::SubAgentThreadStartedInput;
|
||||
use crate::facts::ThreadInitializationMode;
|
||||
use crate::facts::ThreadStartTimingFact;
|
||||
use crate::facts::ThreadStartType;
|
||||
use crate::facts::TurnResolvedConfigFact;
|
||||
use crate::facts::TurnStatus;
|
||||
use crate::facts::TurnSteerRejectionReason;
|
||||
@@ -158,7 +157,6 @@ struct ThreadAnalyticsState {
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
struct ThreadStartTimingState {
|
||||
thread_start_type: ThreadStartType,
|
||||
duration_ms: u64,
|
||||
prepare_duration_ms: u64,
|
||||
spawn_duration_ms: u64,
|
||||
@@ -580,7 +578,6 @@ impl AnalyticsReducer {
|
||||
|
||||
fn ingest_thread_start_timing(&mut self, input: ThreadStartTimingFact) {
|
||||
let thread_start_timing = ThreadStartTimingState {
|
||||
thread_start_type: input.thread_start_type,
|
||||
duration_ms: input.duration_ms,
|
||||
prepare_duration_ms: input.prepare_duration_ms,
|
||||
spawn_duration_ms: input.spawn_duration_ms,
|
||||
@@ -1297,7 +1294,9 @@ impl AnalyticsReducer {
|
||||
let thread_state = self.threads.entry(thread_id.clone()).or_default();
|
||||
thread_state.connection_id = Some(connection_id);
|
||||
thread_state.metadata = Some(thread_metadata.clone());
|
||||
let thread_start_timing = thread_state.thread_start_timing;
|
||||
let thread_start_timing = matches!(initialization_mode, ThreadInitializationMode::New)
|
||||
.then_some(thread_state.thread_start_timing)
|
||||
.flatten();
|
||||
out.push(TrackEventRequest::ThreadInitialized(
|
||||
ThreadInitializedEvent {
|
||||
event_type: "codex_thread_initialized",
|
||||
@@ -1313,7 +1312,6 @@ impl AnalyticsReducer {
|
||||
parent_thread_id: thread_metadata.parent_thread_id,
|
||||
thread_start_timing: match thread_start_timing {
|
||||
Some(timing) => ThreadStartTimingEventParams {
|
||||
thread_start_type: Some(timing.thread_start_type),
|
||||
thread_start_duration_ms: Some(timing.duration_ms),
|
||||
thread_start_prepare_duration_ms: Some(timing.prepare_duration_ms),
|
||||
thread_start_spawn_duration_ms: Some(timing.spawn_duration_ms),
|
||||
|
||||
@@ -142,7 +142,6 @@ pub(crate) fn assert_basic_thread_initialized_event(
|
||||
expected_model: &str,
|
||||
initialization_mode: &str,
|
||||
expected_thread_source: &str,
|
||||
expected_thread_start_type: &str,
|
||||
) {
|
||||
assert_eq!(event["event_params"]["thread_id"], thread_id);
|
||||
assert_eq!(
|
||||
@@ -175,9 +174,5 @@ pub(crate) fn assert_basic_thread_initialized_event(
|
||||
event["event_params"]["initialization_mode"],
|
||||
initialization_mode
|
||||
);
|
||||
assert_eq!(
|
||||
event["event_params"]["thread_start_type"],
|
||||
expected_thread_start_type
|
||||
);
|
||||
assert!(event["event_params"]["created_at"].as_u64().is_some());
|
||||
}
|
||||
|
||||
@@ -406,22 +406,7 @@ async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> {
|
||||
let event =
|
||||
wait_for_analytics_event(&server, ANALYTICS_TEST_TIMEOUT, "codex_thread_initialized")
|
||||
.await?;
|
||||
assert_basic_thread_initialized_event(
|
||||
&event,
|
||||
&thread.id,
|
||||
"mock-model",
|
||||
"forked",
|
||||
"user",
|
||||
"forked",
|
||||
);
|
||||
let thread_start_timings = [
|
||||
"thread_start_duration_ms",
|
||||
"thread_start_prepare_duration_ms",
|
||||
"thread_start_spawn_duration_ms",
|
||||
"thread_start_finalize_duration_ms",
|
||||
]
|
||||
.map(|field| event["event_params"][field].as_u64().is_some());
|
||||
assert_eq!(thread_start_timings, [true, true, true, true]);
|
||||
assert_basic_thread_initialized_event(&event, &thread.id, "mock-model", "forked", "user");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -422,23 +422,8 @@ async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> {
|
||||
let event =
|
||||
wait_for_analytics_event(&server, ANALYTICS_TEST_TIMEOUT, "codex_thread_initialized")
|
||||
.await?;
|
||||
assert_basic_thread_initialized_event(
|
||||
&event,
|
||||
&thread.id,
|
||||
"gpt-5.3-codex",
|
||||
"resumed",
|
||||
"user",
|
||||
"resumed",
|
||||
);
|
||||
assert_basic_thread_initialized_event(&event, &thread.id, "gpt-5.3-codex", "resumed", "user");
|
||||
assert_eq!(event["event_params"]["thread_source"], "user");
|
||||
let thread_start_timings = [
|
||||
"thread_start_duration_ms",
|
||||
"thread_start_prepare_duration_ms",
|
||||
"thread_start_spawn_duration_ms",
|
||||
"thread_start_finalize_duration_ms",
|
||||
]
|
||||
.map(|field| event["event_params"][field].as_u64().is_some());
|
||||
assert_eq!(thread_start_timings, [true, true, true, true]);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -18,7 +18,6 @@ use codex_app_server_protocol::ServerNotification;
|
||||
use codex_app_server_protocol::ThreadSource;
|
||||
use codex_app_server_protocol::ThreadStartParams;
|
||||
use codex_app_server_protocol::ThreadStartResponse;
|
||||
use codex_app_server_protocol::ThreadStartSource;
|
||||
use codex_app_server_protocol::ThreadStartedNotification;
|
||||
use codex_app_server_protocol::ThreadStatus;
|
||||
use codex_app_server_protocol::ThreadStatusChangedNotification;
|
||||
@@ -440,7 +439,7 @@ async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> {
|
||||
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", "user", "new");
|
||||
assert_basic_thread_initialized_event(event, &thread.id, "mock-model", "new", "user");
|
||||
let thread_start_timings = [
|
||||
"thread_start_duration_ms",
|
||||
"thread_start_prepare_duration_ms",
|
||||
@@ -452,44 +451,6 @@ async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_start_clear_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())?;
|
||||
mount_analytics_capture(&server, codex_home.path()).await?;
|
||||
|
||||
let mut mcp = McpProcess::new_without_managed_config(codex_home.path()).await?;
|
||||
timeout(ANALYTICS_TEST_TIMEOUT, mcp.initialize()).await??;
|
||||
|
||||
let req_id = mcp
|
||||
.send_thread_start_request(ThreadStartParams {
|
||||
session_start_source: Some(ThreadStartSource::Clear),
|
||||
thread_source: Some(ThreadSource::User),
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let resp: JSONRPCResponse = timeout(
|
||||
ANALYTICS_TEST_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?;
|
||||
let event = thread_initialized_event(&payload)?;
|
||||
assert_basic_thread_initialized_event(
|
||||
event,
|
||||
&thread.id,
|
||||
"mock-model",
|
||||
"new",
|
||||
"user",
|
||||
"cleared",
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_start_respects_project_config_from_cwd() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
|
||||
@@ -17,7 +17,6 @@ use crate::tasks::InterruptedTurnHistoryMarker;
|
||||
use crate::tasks::interrupted_turn_history_marker;
|
||||
use crate::thread_start_timing::ThreadStartTiming;
|
||||
use codex_analytics::AnalyticsEventsClient;
|
||||
use codex_analytics::ThreadStartType;
|
||||
use codex_app_server_protocol::ThreadHistoryBuilder;
|
||||
use codex_app_server_protocol::TurnStatus;
|
||||
use codex_core_plugins::PluginsManager;
|
||||
@@ -1190,12 +1189,7 @@ impl ThreadManagerState {
|
||||
user_shell_override: Option<crate::shell::Shell>,
|
||||
) -> CodexResult<NewThread> {
|
||||
let is_resumed_thread = matches!(&initial_history, InitialHistory::Resumed(_));
|
||||
let mut thread_start_timing = ThreadStartTiming::start(match &initial_history {
|
||||
InitialHistory::New => ThreadStartType::New,
|
||||
InitialHistory::Cleared => ThreadStartType::Cleared,
|
||||
InitialHistory::Forked(_) => ThreadStartType::Forked,
|
||||
InitialHistory::Resumed(_) => ThreadStartType::Resumed,
|
||||
});
|
||||
let mut thread_start_timing = ThreadStartTiming::start();
|
||||
if let InitialHistory::Resumed(resumed) = &initial_history {
|
||||
let mut threads = self.threads.write().await;
|
||||
if let Some(thread) = threads.get(&resumed.conversation_id).cloned() {
|
||||
|
||||
@@ -2,11 +2,9 @@ use std::time::Duration;
|
||||
use std::time::Instant;
|
||||
|
||||
use codex_analytics::ThreadStartTimingFact;
|
||||
use codex_analytics::ThreadStartType;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct ThreadStartTiming {
|
||||
thread_start_type: ThreadStartType,
|
||||
phase_started_at: Instant,
|
||||
prepare_duration: Option<Duration>,
|
||||
spawn_duration: Option<Duration>,
|
||||
@@ -14,9 +12,8 @@ pub(crate) struct ThreadStartTiming {
|
||||
}
|
||||
|
||||
impl ThreadStartTiming {
|
||||
pub(crate) fn start(thread_start_type: ThreadStartType) -> Self {
|
||||
pub(crate) fn start() -> Self {
|
||||
Self {
|
||||
thread_start_type,
|
||||
phase_started_at: Instant::now(),
|
||||
prepare_duration: None,
|
||||
spawn_duration: None,
|
||||
@@ -42,7 +39,6 @@ impl ThreadStartTiming {
|
||||
let finalize_duration = self.finalize_duration.unwrap_or_default();
|
||||
ThreadStartTimingFact {
|
||||
thread_id,
|
||||
thread_start_type: self.thread_start_type,
|
||||
duration_ms: duration_ms(prepare_duration + spawn_duration + finalize_duration),
|
||||
prepare_duration_ms: duration_ms(prepare_duration),
|
||||
spawn_duration_ms: duration_ms(spawn_duration),
|
||||
|
||||
Reference in New Issue
Block a user