diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index ebf6066384..2b43f88357 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -1714,6 +1714,131 @@ async fn feature_observations_match_legacy_analytics_facts() { assert_eq!(observation_payload, legacy_payload); } +#[tokio::test] +async fn turn_lifecycle_observations_match_legacy_sources() { + let mut legacy_reducer = AnalyticsReducer::default(); + let mut observation_reducer = AnalyticsObservationReducer::default(); + let mut legacy_events = Vec::new(); + let mut observation_events = Vec::new(); + + ingest_turn_prerequisites( + &mut legacy_reducer, + &mut legacy_events, + /*include_initialize*/ true, + /*include_resolved_config*/ true, + /*include_started*/ true, + /*include_token_usage*/ true, + ) + .await; + legacy_reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + /*codex_error_info*/ None, + ))), + &mut legacy_events, + ) + .await; + + observation_reducer + .ingest_existing_fact_for_test( + 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, + }, + product_client_id: "codex-tui".to_string(), + runtime: sample_runtime_metadata(), + rpc_transport: AppServerRpcTransport::Stdio, + }, + &mut observation_events, + ) + .await; + observation_reducer + .ingest_existing_fact_for_test( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_start_response( + "thread-2", /*ephemeral*/ false, "gpt-5", + )), + }, + &mut observation_events, + ) + .await; + observation_events.clear(); + + // Keep not-yet-migrated lifecycle context identical on both sides. This + // test swaps terminal turn lifecycle and token accounting to observations. + observation_reducer + .ingest_existing_fact_for_test( + AnalyticsFact::Request { + connection_id: 7, + request_id: RequestId::Integer(3), + request: Box::new(sample_turn_start_request("thread-2", /*request_id*/ 3)), + }, + &mut observation_events, + ) + .await; + observation_reducer + .ingest_existing_fact_for_test( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_turn_start_response("turn-2", /*request_id*/ 3)), + }, + &mut observation_events, + ) + .await; + observation_reducer + .ingest_existing_fact_for_test( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new( + sample_turn_resolved_config("turn-2"), + ))), + &mut observation_events, + ) + .await; + observation_reducer + .ingest_turn_started( + codex_observability::events::TurnStarted { + thread_id: "thread-2", + turn_id: "turn-2", + started_at: 455, + }, + &mut observation_events, + ) + .await; + observation_reducer + .ingest_turn_ended( + codex_observability::events::TurnEnded { + thread_id: "thread-2", + turn_id: "turn-2", + status: codex_observability::events::TurnStatus::Completed, + token_usage: Some(codex_observability::events::TurnTokenUsage { + input_tokens: 123, + cached_input_tokens: 45, + output_tokens: 140, + reasoning_output_tokens: 13, + total_tokens: 321, + }), + ended_at: 456, + duration_ms: 1234, + }, + &mut observation_events, + ) + .await; + + let legacy_payload = serde_json::to_value(&legacy_events).expect("serialize legacy events"); + let observation_payload = + serde_json::to_value(&observation_events).expect("serialize observation events"); + assert_eq!(observation_payload, legacy_payload); +} + #[tokio::test] async fn reducer_ingests_plugin_state_changed_fact() { let mut reducer = AnalyticsReducer::default(); diff --git a/codex-rs/analytics/src/observation_projection.rs b/codex-rs/analytics/src/observation_projection.rs index 4453b258ef..20277e8bd9 100644 --- a/codex-rs/analytics/src/observation_projection.rs +++ b/codex-rs/analytics/src/observation_projection.rs @@ -17,9 +17,16 @@ use crate::facts::AppInvocation; use crate::facts::AppMentionedInput; use crate::facts::AppUsedInput; use crate::facts::TrackEventsContext; +use crate::facts::TurnTokenUsageFact; +use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::Turn; +use codex_app_server_protocol::TurnCompletedNotification; +use codex_app_server_protocol::TurnStartedNotification; +use codex_app_server_protocol::TurnStatus as AppServerTurnStatus; use codex_login::default_client::originator; use codex_observability::events; use codex_protocol::protocol::HookRunStatus as ProtocolHookRunStatus; +use codex_protocol::protocol::TokenUsage; pub(crate) fn app_mentioned_input(observation: events::AppMentioned<'_>) -> AppMentionedInput { AppMentionedInput { @@ -119,6 +126,62 @@ pub(crate) fn plugin_state_changed_event( } } +pub(crate) fn turn_started_notification( + observation: events::TurnStarted<'_>, +) -> ServerNotification { + ServerNotification::TurnStarted(TurnStartedNotification { + thread_id: observation.thread_id.to_string(), + turn: Turn { + id: observation.turn_id.to_string(), + items: vec![], + status: AppServerTurnStatus::InProgress, + error: None, + started_at: Some(observation.started_at), + completed_at: None, + duration_ms: None, + }, + }) +} + +pub(crate) fn turn_token_usage_fact( + observation: &events::TurnEnded<'_>, +) -> Option { + let token_usage = observation.token_usage?; + Some(TurnTokenUsageFact { + turn_id: observation.turn_id.to_string(), + thread_id: observation.thread_id.to_string(), + token_usage: TokenUsage { + input_tokens: token_usage.input_tokens, + cached_input_tokens: token_usage.cached_input_tokens, + output_tokens: token_usage.output_tokens, + reasoning_output_tokens: token_usage.reasoning_output_tokens, + total_tokens: token_usage.total_tokens, + }, + }) +} + +pub(crate) fn turn_ended_notification(observation: events::TurnEnded<'_>) -> ServerNotification { + ServerNotification::TurnCompleted(TurnCompletedNotification { + thread_id: observation.thread_id.to_string(), + turn: Turn { + id: observation.turn_id.to_string(), + items: vec![], + status: match observation.status { + events::TurnStatus::Completed => AppServerTurnStatus::Completed, + events::TurnStatus::Failed => AppServerTurnStatus::Failed, + events::TurnStatus::Interrupted => AppServerTurnStatus::Interrupted, + }, + // Error taxonomy needs a separate design pass. Keeping it out of + // the first terminal-turn observation avoids baking app-server + // transport categories into the shared event model. + error: None, + started_at: None, + completed_at: Some(observation.ended_at), + duration_ms: Some(observation.duration_ms), + }, + }) +} + fn tracking_from_fields(model_slug: &str, thread_id: &str, turn_id: &str) -> TrackEventsContext { TrackEventsContext { model_slug: model_slug.to_string(), diff --git a/codex-rs/analytics/src/observation_reducer.rs b/codex-rs/analytics/src/observation_reducer.rs index 56d677ab10..4907911551 100644 --- a/codex-rs/analytics/src/observation_reducer.rs +++ b/codex-rs/analytics/src/observation_reducer.rs @@ -20,6 +20,20 @@ pub(crate) struct AnalyticsObservationReducer { } impl AnalyticsObservationReducer { + /// Feeds an existing analytics fact into the wrapped reducer for conformance tests. + /// + /// The observation stream is being introduced incrementally, so tests need + /// to hold the not-yet-migrated lifecycle context constant while swapping a + /// specific source from legacy facts to observations. + #[cfg(test)] + pub(crate) async fn ingest_existing_fact_for_test( + &mut self, + fact: AnalyticsFact, + out: &mut Vec, + ) { + self.legacy.ingest(fact, out).await; + } + /// Ingests an app.mentioned observation and emits the current analytics event. pub(crate) async fn ingest_app_mentioned( &mut self, @@ -52,6 +66,49 @@ impl AnalyticsObservationReducer { .await; } + /// Ingests a turn.started observation into the current turn-event reducer state. + pub(crate) async fn ingest_turn_started( + &mut self, + observation: events::TurnStarted<'_>, + out: &mut Vec, + ) { + self.legacy + .ingest( + AnalyticsFact::Notification(Box::new( + observation_projection::turn_started_notification(observation), + )), + out, + ) + .await; + } + + /// Ingests a turn.ended observation into the current turn-event reducer state. + pub(crate) async fn ingest_turn_ended( + &mut self, + observation: events::TurnEnded<'_>, + out: &mut Vec, + ) { + if let Some(token_usage) = observation_projection::turn_token_usage_fact(&observation) { + self.legacy + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnTokenUsage(Box::new( + token_usage, + ))), + out, + ) + .await; + } + + self.legacy + .ingest( + AnalyticsFact::Notification(Box::new( + observation_projection::turn_ended_notification(observation), + )), + out, + ) + .await; + } + /// Ingests a hook.run_completed observation and emits the current analytics event. pub(crate) fn ingest_hook_run_completed( &mut self, diff --git a/codex-rs/observability/src/events.rs b/codex-rs/observability/src/events.rs index 61ede20de5..5c15b99cba 100644 --- a/codex-rs/observability/src/events.rs +++ b/codex-rs/observability/src/events.rs @@ -41,6 +41,87 @@ pub enum PluginState { Disabled, } +/// Terminal turn status after Codex stops working on a turn. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum TurnStatus { + /// The turn completed successfully. + Completed, + /// The turn failed with an error. + Failed, + /// The turn was interrupted before normal completion. + Interrupted, +} + +/// Observation emitted when execution of a turn starts. +#[derive(Observation)] +#[observation(name = "turn.started", crate = "crate", uses = ["analytics"])] +pub struct TurnStarted<'a> { + /// Thread that owns the turn. + #[obs(level = "basic", class = "identifier")] + pub thread_id: &'a str, + + /// Turn that started. + #[obs(level = "basic", class = "identifier")] + pub turn_id: &'a str, + + /// Unix timestamp in seconds when the turn started. + #[obs(level = "basic", class = "operational")] + pub started_at: i64, +} + +/// Token accounting reported for a completed turn. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +pub struct TurnTokenUsage { + /// Prompt/input token count reported by the model provider. + pub input_tokens: i64, + + /// Input tokens served from provider-side cache. + pub cached_input_tokens: i64, + + /// Output token count reported by the model provider. + pub output_tokens: i64, + + /// Output tokens spent on model reasoning. + pub reasoning_output_tokens: i64, + + /// Total token count reported by the model provider. + pub total_tokens: i64, +} + +/// Observation emitted when a turn reaches a terminal state. +#[derive(Observation)] +#[observation(name = "turn.ended", crate = "crate", uses = ["analytics"])] +pub struct TurnEnded<'a> { + /// Thread that owns the turn. + #[obs(level = "basic", class = "identifier")] + pub thread_id: &'a str, + + /// Turn that reached a terminal state. + #[obs(level = "basic", class = "identifier")] + pub turn_id: &'a str, + + /// Terminal status for the turn. + #[obs(level = "basic", class = "operational")] + pub status: TurnStatus, + + /// Token usage reported by the model provider. + /// + /// This is absent when a turn ends before provider usage is available, for + /// example an early local failure, an interruption, or a provider failure + /// before a usage-bearing response is received. + #[obs(level = "basic", class = "operational")] + pub token_usage: Option, + + /// Unix timestamp in seconds when the turn ended. + #[obs(level = "basic", class = "operational")] + pub ended_at: i64, + + /// Turn duration in milliseconds. + #[obs(level = "basic", class = "operational")] + pub duration_ms: i64, +} + /// Observation emitted when an app connector is mentioned during a turn. #[derive(Observation)] #[observation(name = "app.mentioned", crate = "crate", uses = ["analytics"])] diff --git a/docs/observability-event-stream-design.md b/docs/observability-event-stream-design.md index 6c5b4f443a..9a8e3969f3 100644 --- a/docs/observability-event-stream-design.md +++ b/docs/observability-event-stream-design.md @@ -270,7 +270,7 @@ Initial workflow coverage should be chosen by conformance need: | Workflow | Canonical examples | Primary consumers | | --- | --- | --- | | Session/thread | `session.config_resolved`, `thread.started`, `thread.ended` | analytics, OTEL, rollout | -| Turn lifecycle | `turn.requested`, `turn.started`, `turn.ended`, `turn.token_usage_observed` | analytics, OTEL, rollout | +| Turn lifecycle | `turn.requested`, `turn.started`, `turn.ended` | analytics, OTEL, rollout | | Turn timing | `turn.first_token_observed`, `turn.first_message_observed` | OTEL metrics, rollout | | Model I/O | `inference.started`, `inference.sse_event_observed`, `inference.completed`, `inference.failed` | OTEL, rollout | | Tools | `tool_call.started`, `tool_call.approval_resolved`, `tool_call.ended` | OTEL, rollout | @@ -292,7 +292,7 @@ output schemas. | Observations | Analytics output | | --- | --- | | `thread.started` | `codex_thread_initialized` | -| turn config/lifecycle/token observations | `codex_turn_event` | +| `turn.started`, `turn.ended` with token usage, and future config observations | `codex_turn_event` | | `turn.steer_resolved` | `codex_turn_steer_event` | | compaction lifecycle observations | `codex_compaction_event` | | skill/app/plugin/guardian observations | existing feature events |