openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-08 15:50:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

[observability] Add turn lifecycle observations\n\nIntroduce shared turn.started and turn.ended observations, including terminal\nstatus, timing, and optional provider token usage on the terminal event.\n\nProject those observations into the existing analytics reducer state and add a\nside-by-side conformance test showing the observation path produces the same\ncodex_turn_event payload as the legacy inputs for the migrated fields.\n\nUpdate the design doc so turn token accounting is modeled as part of the\nterminal turn event rather than a separate canonical event.

Browse Source
This commit is contained in:
Albin Cassirer
2026-04-19 15:54:59 -07:00
parent 10f7e100d9
commit 7418d6e364
5 changed files with 328 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
codex-rs/analytics/src/analytics_client_tests.rs
View File

@@ -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();

63
codex-rs/analytics/src/observation_projection.rs
View File

@@ -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<TurnTokenUsageFact> {
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(),

57
codex-rs/analytics/src/observation_reducer.rs
View File

@@ -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<TrackEventRequest>,
) {
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<TrackEventRequest>,
) {
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<TrackEventRequest>,
) {
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,

81
codex-rs/observability/src/events.rs
View File

@@ -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<TurnTokenUsage>,
/// 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"])]

4
docs/observability-event-stream-design.md
View File

@@ -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 |