[codex-analytics] add protocol-native turn timestamps

This commit is contained in:
rhan-oai
2026-04-02 16:14:41 -07:00
parent 0bd31dc382
commit bd8a3ac0f9
23 changed files with 204 additions and 4 deletions
+4
View File
@@ -1061,6 +1061,8 @@ mod tests {
status: codex_app_server_protocol::TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
})
}
@@ -1835,6 +1837,8 @@ mod tests {
status: codex_app_server_protocol::TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
}
)
)
@@ -3566,6 +3566,19 @@
},
"TurnCompletedNotification": {
"properties": {
"completedAt": {
"description": "Unix timestamp (in seconds) when the turn completed.",
"format": "int64",
"type": "integer"
},
"durationMs": {
"description": "Duration between turn start and completion in milliseconds, if known.",
"format": "int64",
"type": [
"integer",
"null"
]
},
"threadId": {
"type": "string"
},
@@ -3574,6 +3587,7 @@
}
},
"required": [
"completedAt",
"threadId",
"turn"
],
@@ -3680,6 +3694,11 @@
},
"TurnStartedNotification": {
"properties": {
"createdAt": {
"description": "Unix timestamp (in seconds) when the turn started.",
"format": "int64",
"type": "integer"
},
"threadId": {
"type": "string"
},
@@ -3688,6 +3707,7 @@
}
},
"required": [
"createdAt",
"threadId",
"turn"
],
@@ -14364,6 +14364,19 @@
"TurnCompletedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"completedAt": {
"description": "Unix timestamp (in seconds) when the turn completed.",
"format": "int64",
"type": "integer"
},
"durationMs": {
"description": "Duration between turn start and completion in milliseconds, if known.",
"format": "int64",
"type": [
"integer",
"null"
]
},
"threadId": {
"type": "string"
},
@@ -14372,6 +14385,7 @@
}
},
"required": [
"completedAt",
"threadId",
"turn"
],
@@ -14640,6 +14654,11 @@
"TurnStartedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"createdAt": {
"description": "Unix timestamp (in seconds) when the turn started.",
"format": "int64",
"type": "integer"
},
"threadId": {
"type": "string"
},
@@ -14648,6 +14667,7 @@
}
},
"required": [
"createdAt",
"threadId",
"turn"
],
@@ -12219,6 +12219,19 @@
"TurnCompletedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"completedAt": {
"description": "Unix timestamp (in seconds) when the turn completed.",
"format": "int64",
"type": "integer"
},
"durationMs": {
"description": "Duration between turn start and completion in milliseconds, if known.",
"format": "int64",
"type": [
"integer",
"null"
]
},
"threadId": {
"type": "string"
},
@@ -12227,6 +12240,7 @@
}
},
"required": [
"completedAt",
"threadId",
"turn"
],
@@ -12495,6 +12509,11 @@
"TurnStartedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"createdAt": {
"description": "Unix timestamp (in seconds) when the turn started.",
"format": "int64",
"type": "integer"
},
"threadId": {
"type": "string"
},
@@ -12503,6 +12522,7 @@
}
},
"required": [
"createdAt",
"threadId",
"turn"
],
@@ -1559,6 +1559,19 @@
}
},
"properties": {
"completedAt": {
"description": "Unix timestamp (in seconds) when the turn completed.",
"format": "int64",
"type": "integer"
},
"durationMs": {
"description": "Duration between turn start and completion in milliseconds, if known.",
"format": "int64",
"type": [
"integer",
"null"
]
},
"threadId": {
"type": "string"
},
@@ -1567,6 +1580,7 @@
}
},
"required": [
"completedAt",
"threadId",
"turn"
],
@@ -1559,6 +1559,11 @@
}
},
"properties": {
"createdAt": {
"description": "Unix timestamp (in seconds) when the turn started.",
"format": "int64",
"type": "integer"
},
"threadId": {
"type": "string"
},
@@ -1567,6 +1572,7 @@
}
},
"required": [
"createdAt",
"threadId",
"turn"
],
@@ -3,4 +3,12 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { Turn } from "./Turn";
export type TurnCompletedNotification = { threadId: string, turn: Turn, };
export type TurnCompletedNotification = { threadId: string, turn: Turn,
/**
* Unix timestamp (in seconds) when the turn completed.
*/
completedAt: number,
/**
* Duration between turn start and completion in milliseconds, if known.
*/
durationMs: number | null, };
@@ -3,4 +3,8 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { Turn } from "./Turn";
export type TurnStartedNotification = { threadId: string, turn: Turn, };
export type TurnStartedNotification = { threadId: string, turn: Turn,
/**
* Unix timestamp (in seconds) when the turn started.
*/
createdAt: number, };
@@ -5043,6 +5043,9 @@ pub struct ThreadNameUpdatedNotification {
pub struct TurnStartedNotification {
pub thread_id: String,
pub turn: Turn,
/// Unix timestamp (in seconds) when the turn started.
#[ts(type = "number")]
pub created_at: i64,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
@@ -5069,6 +5072,12 @@ pub struct Usage {
pub struct TurnCompletedNotification {
pub thread_id: String,
pub turn: Turn,
/// Unix timestamp (in seconds) when the turn completed.
#[ts(type = "number")]
pub completed_at: i64,
/// Duration between turn start and completion in milliseconds, if known.
#[ts(type = "number | null")]
pub duration_ms: Option<i64>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
@@ -141,6 +141,8 @@ use std::convert::TryFrom;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
use tokio::sync::Mutex;
use tokio::sync::oneshot;
use tracing::error;
@@ -289,6 +291,7 @@ pub(crate) async fn apply_bespoke_event_handling(
let notification = TurnStartedNotification {
thread_id: conversation_id.to_string(),
turn,
created_at: set_turn_started_at(&thread_state).await,
};
outgoing
.send_server_notification(ServerNotification::TurnStarted(notification))
@@ -1947,8 +1950,10 @@ async fn emit_turn_completed_with_status(
event_turn_id: String,
status: TurnStatus,
error: Option<TurnError>,
started_at_ms: Option<i64>,
outgoing: &ThreadScopedOutgoingMessageSender,
) {
let completed_at_ms = now_unix_timestamp_millis();
let notification = TurnCompletedNotification {
thread_id: conversation_id.to_string(),
turn: Turn {
@@ -1957,6 +1962,9 @@ async fn emit_turn_completed_with_status(
error,
status,
},
completed_at: completed_at_ms / 1000,
duration_ms: started_at_ms
.map(|started_at_ms| completed_at_ms.saturating_sub(started_at_ms)),
};
outgoing
.send_server_notification(ServerNotification::TurnCompleted(notification))
@@ -2105,13 +2113,22 @@ async fn handle_turn_complete(
thread_state: &Arc<Mutex<ThreadState>>,
) {
let turn_summary = find_and_remove_turn_summary(conversation_id, thread_state).await;
let started_at_ms = turn_summary.started_at_ms;
let (status, error) = match turn_summary.last_error {
Some(error) => (TurnStatus::Failed, Some(error)),
None => (TurnStatus::Completed, None),
};
emit_turn_completed_with_status(conversation_id, event_turn_id, status, error, outgoing).await;
emit_turn_completed_with_status(
conversation_id,
event_turn_id,
status,
error,
started_at_ms,
outgoing,
)
.await;
}
async fn handle_turn_interrupted(
@@ -2120,18 +2137,33 @@ async fn handle_turn_interrupted(
outgoing: &ThreadScopedOutgoingMessageSender,
thread_state: &Arc<Mutex<ThreadState>>,
) {
find_and_remove_turn_summary(conversation_id, thread_state).await;
let turn_summary = find_and_remove_turn_summary(conversation_id, thread_state).await;
emit_turn_completed_with_status(
conversation_id,
event_turn_id,
TurnStatus::Interrupted,
/*error*/ None,
turn_summary.started_at_ms,
outgoing,
)
.await;
}
async fn set_turn_started_at(thread_state: &Arc<Mutex<ThreadState>>) -> i64 {
let started_at_ms = now_unix_timestamp_millis();
let mut state = thread_state.lock().await;
state.turn_summary.started_at_ms = Some(started_at_ms);
started_at_ms / 1000
}
fn now_unix_timestamp_millis() -> i64 {
let duration = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default();
i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)
}
async fn handle_thread_rollback_failed(
_conversation_id: ThreadId,
message: String,
+2
View File
@@ -824,6 +824,8 @@ mod tests {
status: TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
})
));
}
+1
View File
@@ -44,6 +44,7 @@ pub(crate) enum ThreadListenerCommand {
/// Per-conversation accumulation of the latest states e.g. error message while a turn runs.
#[derive(Default, Clone)]
pub(crate) struct TurnSummary {
pub(crate) started_at_ms: Option<i64>,
pub(crate) file_change_started: HashSet<String>,
pub(crate) command_execution_started: HashSet<String>,
pub(crate) last_error: Option<TurnError>,
@@ -157,6 +157,8 @@ fn turn_completed_recovers_final_message_from_turn_items() {
let status = processor.process_server_notification(ServerNotification::TurnCompleted(
codex_app_server_protocol::TurnCompletedNotification {
thread_id: "thread-1".to_string(),
completed_at: 0,
duration_ms: None,
turn: Turn {
id: "turn-1".to_string(),
items: vec![ThreadItem::AgentMessage {
@@ -201,6 +203,8 @@ fn turn_completed_overwrites_stale_final_message_from_turn_items() {
let status = processor.process_server_notification(ServerNotification::TurnCompleted(
codex_app_server_protocol::TurnCompletedNotification {
thread_id: "thread-1".to_string(),
completed_at: 0,
duration_ms: None,
turn: Turn {
id: "turn-1".to_string(),
items: vec![ThreadItem::AgentMessage {
@@ -246,6 +250,8 @@ fn turn_completed_preserves_streamed_final_message_when_turn_items_are_empty() {
let status = processor.process_server_notification(ServerNotification::TurnCompleted(
codex_app_server_protocol::TurnCompletedNotification {
thread_id: "thread-1".to_string(),
completed_at: 0,
duration_ms: None,
turn: Turn {
id: "turn-1".to_string(),
items: Vec::new(),
@@ -286,6 +292,8 @@ fn turn_failed_clears_stale_final_message() {
let status = processor.process_server_notification(ServerNotification::TurnCompleted(
codex_app_server_protocol::TurnCompletedNotification {
thread_id: "thread-1".to_string(),
completed_at: 0,
duration_ms: None,
turn: Turn {
id: "turn-1".to_string(),
items: Vec::new(),
@@ -327,6 +335,8 @@ fn turn_interrupted_clears_stale_final_message() {
let status = processor.process_server_notification(ServerNotification::TurnCompleted(
codex_app_server_protocol::TurnCompletedNotification {
thread_id: "thread-1".to_string(),
completed_at: 0,
duration_ms: None,
turn: Turn {
id: "turn-1".to_string(),
items: Vec::new(),
@@ -29,6 +29,8 @@ fn failed_turn_does_not_overwrite_output_last_message_file() {
let status = processor.process_server_notification(ServerNotification::TurnCompleted(
codex_app_server_protocol::TurnCompletedNotification {
thread_id: "thread-1".to_string(),
completed_at: 0,
duration_ms: None,
turn: codex_app_server_protocol::Turn {
id: "turn-1".to_string(),
items: Vec::new(),
+1
View File
@@ -717,6 +717,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> {
TurnStartedNotification {
thread_id: response.review_thread_id.clone(),
turn: response.turn.clone(),
created_at: 0,
},
));
let task_id = response.turn.id;
@@ -145,6 +145,7 @@ fn turn_started_emits_turn_started_event() {
status: TurnStatus::InProgress,
error: None,
},
created_at: 0,
}));
assert_eq!(
@@ -1067,6 +1068,8 @@ fn plan_update_emits_started_then_updated_then_completed() {
status: TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
},
));
assert_eq!(
@@ -1123,6 +1126,8 @@ fn plan_update_after_completion_starts_new_todo_list_with_new_id() {
status: TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
},
));
@@ -1202,6 +1207,8 @@ fn token_usage_update_is_emitted_on_turn_completion() {
status: TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
},
));
assert_eq!(
@@ -1237,6 +1244,8 @@ fn turn_completion_recovers_final_message_from_turn_items() {
status: TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
},
));
@@ -1311,6 +1320,8 @@ fn turn_completion_reconciles_started_items_from_turn_items() {
status: TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
},
));
@@ -1368,6 +1379,8 @@ fn turn_completion_overwrites_stale_final_message_from_turn_items() {
status: TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
},
));
@@ -1408,6 +1421,8 @@ fn turn_completion_preserves_streamed_final_message_when_turn_items_are_empty()
status: TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
},
));
@@ -1456,6 +1471,8 @@ fn failed_turn_clears_stale_final_message() {
codex_error_info: None,
}),
},
completed_at: 0,
duration_ms: None,
},
));
@@ -1479,6 +1496,8 @@ fn turn_completion_falls_back_to_final_plan_text() {
status: TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
},
));
@@ -1527,6 +1546,8 @@ fn turn_failure_prefers_structured_error_message() {
status: TurnStatus::Failed,
error: None,
},
completed_at: 0,
duration_ms: None,
},
));
assert_eq!(
+3
View File
@@ -9191,6 +9191,7 @@ guardian_approval = true
ServerNotification::TurnStarted(TurnStartedNotification {
thread_id: thread_id.to_string(),
turn: test_turn(turn_id, TurnStatus::InProgress, Vec::new()),
created_at: 0,
})
}
@@ -9202,6 +9203,8 @@ guardian_approval = true
ServerNotification::TurnCompleted(TurnCompletedNotification {
thread_id: thread_id.to_string(),
turn: test_turn(turn_id, status, Vec::new()),
completed_at: 0,
duration_ms: None,
})
}
@@ -1104,6 +1104,8 @@ mod tests {
status: TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
}),
)
.expect("notification should bridge");
@@ -1316,6 +1318,8 @@ mod tests {
status: TurnStatus::Interrupted,
error: None,
},
completed_at: 0,
duration_ms: None,
}),
)
.expect("notification should bridge");
@@ -1352,6 +1356,8 @@ mod tests {
additional_details: None,
}),
},
completed_at: 0,
duration_ms: None,
}),
)
.expect("notification should bridge");
@@ -677,6 +677,8 @@ mod tests {
status: TurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
})
}
+2
View File
@@ -5911,6 +5911,8 @@ impl ChatWidget {
status,
error,
},
completed_at: 0,
duration_ms: None,
},
Some(replay_kind),
);
@@ -94,6 +94,7 @@ async fn live_app_server_turn_completed_clears_working_status_after_answer_item(
status: AppServerTurnStatus::InProgress,
error: None,
},
created_at: 0,
}),
/*replay_kind*/ None,
);
@@ -133,6 +134,8 @@ async fn live_app_server_turn_completed_clears_working_status_after_answer_item(
status: AppServerTurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
}),
/*replay_kind*/ None,
);
@@ -416,6 +419,7 @@ async fn live_app_server_failed_turn_does_not_duplicate_error_history() {
status: AppServerTurnStatus::InProgress,
error: None,
},
created_at: 0,
}),
/*replay_kind*/ None,
);
@@ -451,6 +455,8 @@ async fn live_app_server_failed_turn_does_not_duplicate_error_history() {
additional_details: None,
}),
},
completed_at: 0,
duration_ms: None,
}),
/*replay_kind*/ None,
);
@@ -472,6 +478,7 @@ async fn live_app_server_stream_recovery_restores_previous_status_header() {
status: AppServerTurnStatus::InProgress,
error: None,
},
created_at: 0,
}),
/*replay_kind*/ None,
);
@@ -526,6 +533,7 @@ async fn live_app_server_server_overloaded_error_renders_warning() {
status: AppServerTurnStatus::InProgress,
error: None,
},
created_at: 0,
}),
/*replay_kind*/ None,
);
@@ -537,6 +537,7 @@ async fn replayed_retryable_app_server_error_keeps_turn_running() {
status: AppServerTurnStatus::InProgress,
error: None,
},
created_at: 0,
}),
Some(ReplayKind::ThreadSnapshot),
);
@@ -687,6 +688,7 @@ async fn live_reasoning_summary_is_not_rendered_twice_when_item_completes() {
status: AppServerTurnStatus::InProgress,
error: None,
},
created_at: 0,
}),
/*replay_kind*/ None,
);
@@ -851,6 +851,7 @@ async fn submit_user_message_queues_while_compaction_turn_is_running() {
status: AppServerTurnStatus::InProgress,
error: None,
},
created_at: 0,
}),
/*replay_kind*/ None,
);
@@ -894,6 +895,8 @@ async fn submit_user_message_queues_while_compaction_turn_is_running() {
status: AppServerTurnStatus::Completed,
error: None,
},
completed_at: 0,
duration_ms: None,
}),
/*replay_kind*/ None,
);