openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-27 18:48:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

[observability] Add analytics feature observations\n\nIntroduce typed observation events for app mentions, hook runs, and plugin\nusage/state changes, with analytics use markers on the event fields.\n\nAdd a private analytics projection module that maps shared observations into\nthe existing legacy analytics payloads, keeping the observation reducer focused\non ingestion orchestration. Cover the new projection path with side-by-side\ntests against the existing analytics facts.\n\nUpdate the design doc to record that backend-specific schema mapping should\nremain isolated from the shared observation taxonomy.

Browse Source
This commit is contained in:
Albin Cassirer
2026-04-19 15:35:17 -07:00
parent b615ce9291
commit 10f7e100d9
6 changed files with 499 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/analytics/src/analytics_client_tests.rs
+117
View File
@@ -1597,6 +1597,123 @@ async fn app_used_observation_matches_legacy_analytics_fact() {
assert_eq!(observation_payload, legacy_payload);
}
#[tokio::test]
async fn feature_observations_match_legacy_analytics_facts() {
let mut legacy_reducer = AnalyticsReducer::default();
let mut observation_reducer = AnalyticsObservationReducer::default();
let mut legacy_events = Vec::new();
let mut observation_events = Vec::new();
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let connector_ids = vec!["calendar".to_string(), "drive".to_string()];
legacy_reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::AppMentioned(AppMentionedInput {
tracking: tracking.clone(),
mentions: vec![AppInvocation {
connector_id: Some("calendar".to_string()),
app_name: Some("Calendar".to_string()),
invocation_type: Some(InvocationType::Explicit),
}],
})),
&mut legacy_events,
)
.await;
legacy_reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::HookRun(HookRunInput {
tracking: tracking.clone(),
hook: HookRunFact {
event_name: HookEventName::PostToolUse,
hook_source: HookSource::Unknown,
status: HookRunStatus::Failed,
},
})),
&mut legacy_events,
)
.await;
legacy_reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::PluginUsed(PluginUsedInput {
tracking: tracking.clone(),
plugin: sample_plugin_metadata(),
})),
&mut legacy_events,
)
.await;
legacy_reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::PluginStateChanged(
PluginStateChangedInput {
plugin: sample_plugin_metadata(),
state: PluginState::Disabled,
},
)),
&mut legacy_events,
)
.await;
observation_reducer
.ingest_app_mentioned(
codex_observability::events::AppMentioned {
model_slug: "gpt-5",
thread_id: "thread-1",
turn_id: "turn-1",
connector_id: Some("calendar"),
app_name: Some("Calendar"),
invocation_type: Some(codex_observability::events::InvocationType::Explicit),
},
&mut observation_events,
)
.await;
observation_reducer.ingest_hook_run_completed(
codex_observability::events::HookRunCompleted {
model_slug: "gpt-5",
thread_id: "thread-1",
turn_id: "turn-1",
hook_name: "PostToolUse",
hook_source: "unknown",
status: codex_observability::events::HookRunStatus::Failed,
},
&mut observation_events,
);
observation_reducer.ingest_plugin_used(
codex_observability::events::PluginUsed {
model_slug: "gpt-5",
thread_id: "thread-1",
turn_id: "turn-1",
plugin_id: "sample@test",
plugin_name: "sample",
marketplace_name: "test",
has_skills: Some(true),
mcp_server_count: Some(2),
connector_ids: Some(connector_ids.as_slice()),
},
&mut observation_events,
);
observation_reducer.ingest_plugin_state_changed(
codex_observability::events::PluginStateChanged {
plugin_id: "sample@test",
plugin_name: "sample",
marketplace_name: "test",
has_skills: Some(true),
mcp_server_count: Some(2),
connector_ids: Some(connector_ids.as_slice()),
state: codex_observability::events::PluginState::Disabled,
},
&mut observation_events,
);
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();
codex-rs/analytics/src/lib.rs
+1
View File
@@ -1,6 +1,7 @@
mod client;
mod events;
mod facts;
mod observation_projection;
mod observation_reducer;
mod reducer;
codex-rs/analytics/src/observation_projection.rs
+166
View File
@@ -0,0 +1,166 @@
//! Projection from shared observations into the current analytics schema.
//!
//! The observation taxonomy is intended to describe what Codex did, not the
//! shape of any particular telemetry backend. This private module is the
//! adapter boundary where typed observations are translated into the legacy
//! analytics facts and track-event payloads that already exist today.
use crate::events::CodexHookRunEventRequest;
use crate::events::CodexHookRunMetadata;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginMetadata;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexPluginUsedMetadata;
use crate::events::TrackEventRequest;
use crate::facts;
use crate::facts::AppInvocation;
use crate::facts::AppMentionedInput;
use crate::facts::AppUsedInput;
use crate::facts::TrackEventsContext;
use codex_login::default_client::originator;
use codex_observability::events;
use codex_protocol::protocol::HookRunStatus as ProtocolHookRunStatus;
pub(crate) fn app_mentioned_input(observation: events::AppMentioned<'_>) -> AppMentionedInput {
AppMentionedInput {
tracking: tracking_from_fields(
observation.model_slug,
observation.thread_id,
observation.turn_id,
),
mentions: vec![app_invocation_from_fields(
observation.connector_id,
observation.app_name,
observation.invocation_type,
)],
}
}
pub(crate) fn app_used_input(observation: events::AppUsed<'_>) -> AppUsedInput {
AppUsedInput {
tracking: tracking_from_fields(
observation.model_slug,
observation.thread_id,
observation.turn_id,
),
app: app_invocation_from_fields(
observation.connector_id,
observation.app_name,
observation.invocation_type,
),
}
}
pub(crate) fn hook_run_completed_event(
observation: events::HookRunCompleted<'_>,
) -> TrackEventRequest {
TrackEventRequest::HookRun(CodexHookRunEventRequest {
event_type: "codex_hook_run",
event_params: CodexHookRunMetadata {
thread_id: Some(observation.thread_id.to_string()),
turn_id: Some(observation.turn_id.to_string()),
model_slug: Some(observation.model_slug.to_string()),
hook_name: Some(observation.hook_name.to_string()),
hook_source: Some(observation.hook_source),
status: Some(match observation.status {
events::HookRunStatus::Completed => ProtocolHookRunStatus::Completed,
events::HookRunStatus::Failed => ProtocolHookRunStatus::Failed,
events::HookRunStatus::Blocked => ProtocolHookRunStatus::Blocked,
events::HookRunStatus::Stopped => ProtocolHookRunStatus::Stopped,
}),
},
})
}
pub(crate) fn plugin_used_event(observation: events::PluginUsed<'_>) -> TrackEventRequest {
TrackEventRequest::PluginUsed(CodexPluginUsedEventRequest {
event_type: "codex_plugin_used",
event_params: CodexPluginUsedMetadata {
plugin: plugin_metadata_from_fields(
observation.plugin_id,
observation.plugin_name,
observation.marketplace_name,
observation.has_skills,
observation.mcp_server_count,
observation.connector_ids,
),
thread_id: Some(observation.thread_id.to_string()),
turn_id: Some(observation.turn_id.to_string()),
model_slug: Some(observation.model_slug.to_string()),
},
})
}
pub(crate) fn plugin_state_changed_event(
observation: events::PluginStateChanged<'_>,
) -> TrackEventRequest {
let event = CodexPluginEventRequest {
event_type: match observation.state {
events::PluginState::Installed => "codex_plugin_installed",
events::PluginState::Uninstalled => "codex_plugin_uninstalled",
events::PluginState::Enabled => "codex_plugin_enabled",
events::PluginState::Disabled => "codex_plugin_disabled",
},
event_params: plugin_metadata_from_fields(
observation.plugin_id,
observation.plugin_name,
observation.marketplace_name,
observation.has_skills,
observation.mcp_server_count,
observation.connector_ids,
),
};
match observation.state {
events::PluginState::Installed => TrackEventRequest::PluginInstalled(event),
events::PluginState::Uninstalled => TrackEventRequest::PluginUninstalled(event),
events::PluginState::Enabled => TrackEventRequest::PluginEnabled(event),
events::PluginState::Disabled => TrackEventRequest::PluginDisabled(event),
}
}
fn tracking_from_fields(model_slug: &str, thread_id: &str, turn_id: &str) -> TrackEventsContext {
TrackEventsContext {
model_slug: model_slug.to_string(),
thread_id: thread_id.to_string(),
turn_id: turn_id.to_string(),
}
}
fn app_invocation_from_fields(
connector_id: Option<&str>,
app_name: Option<&str>,
invocation_type: Option<events::InvocationType>,
) -> AppInvocation {
AppInvocation {
connector_id: connector_id.map(str::to_string),
app_name: app_name.map(str::to_string),
invocation_type: invocation_type.map(map_invocation_type),
}
}
fn plugin_metadata_from_fields(
plugin_id: &str,
plugin_name: &str,
marketplace_name: &str,
has_skills: Option<bool>,
mcp_server_count: Option<usize>,
connector_ids: Option<&[String]>,
) -> CodexPluginMetadata {
CodexPluginMetadata {
plugin_id: Some(plugin_id.to_string()),
plugin_name: Some(plugin_name.to_string()),
marketplace_name: Some(marketplace_name.to_string()),
has_skills,
mcp_server_count,
connector_ids: connector_ids.map(<[String]>::to_vec),
product_client_id: Some(originator().value),
}
}
fn map_invocation_type(invocation_type: events::InvocationType) -> facts::InvocationType {
match invocation_type {
events::InvocationType::Explicit => facts::InvocationType::Explicit,
events::InvocationType::Implicit => facts::InvocationType::Implicit,
}
}
codex-rs/analytics/src/observation_reducer.rs
+50 -22
View File
@@ -1,4 +1,4 @@
//! Projection from shared observations into analytics track events.
//! Analytics reducer entrypoint for shared observations.
//!
//! This module deliberately reuses the existing analytics reducer while the
//! shared observation stream is being introduced. That gives conformance tests
@@ -7,12 +7,9 @@
//! requests.
use crate::events::TrackEventRequest;
use crate::facts;
use crate::facts::AnalyticsFact;
use crate::facts::AppInvocation;
use crate::facts::AppUsedInput;
use crate::facts::CustomAnalyticsFact;
use crate::facts::TrackEventsContext;
use crate::observation_projection;
use crate::reducer::AnalyticsReducer;
use codex_observability::events;
@@ -23,35 +20,66 @@ pub(crate) struct AnalyticsObservationReducer {
}
impl AnalyticsObservationReducer {
/// Ingests an app.mentioned observation and emits the current analytics event.
pub(crate) async fn ingest_app_mentioned(
&mut self,
observation: events::AppMentioned<'_>,
out: &mut Vec<TrackEventRequest>,
) {
self.legacy
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::AppMentioned(
observation_projection::app_mentioned_input(observation),
)),
out,
)
.await;
}
/// Ingests an app.used observation and emits the current analytics event.
pub(crate) async fn ingest_app_used(
&mut self,
observation: events::AppUsed<'_>,
out: &mut Vec<TrackEventRequest>,
) {
let tracking = TrackEventsContext {
model_slug: observation.model_slug.to_string(),
thread_id: observation.thread_id.to_string(),
turn_id: observation.turn_id.to_string(),
};
let app = AppInvocation {
connector_id: observation.connector_id.map(str::to_string),
app_name: observation.app_name.map(str::to_string),
invocation_type: observation.invocation_type.map(map_invocation_type),
};
self.legacy
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(AppUsedInput { tracking, app })),
AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(
observation_projection::app_used_input(observation),
)),
out,
)
.await;
}
}
fn map_invocation_type(invocation_type: events::InvocationType) -> facts::InvocationType {
match invocation_type {
events::InvocationType::Explicit => facts::InvocationType::Explicit,
events::InvocationType::Implicit => facts::InvocationType::Implicit,
/// Ingests a hook.run_completed observation and emits the current analytics event.
pub(crate) fn ingest_hook_run_completed(
&mut self,
observation: events::HookRunCompleted<'_>,
out: &mut Vec<TrackEventRequest>,
) {
out.push(observation_projection::hook_run_completed_event(
observation,
));
}
/// Ingests a plugin.used observation and emits the current analytics event.
pub(crate) fn ingest_plugin_used(
&mut self,
observation: events::PluginUsed<'_>,
out: &mut Vec<TrackEventRequest>,
) {
out.push(observation_projection::plugin_used_event(observation));
}
/// Ingests a plugin.state_changed observation and emits the current analytics event.
pub(crate) fn ingest_plugin_state_changed(
&mut self,
observation: events::PluginStateChanged<'_>,
out: &mut Vec<TrackEventRequest>,
) {
out.push(observation_projection::plugin_state_changed_event(
observation,
));
}
}
codex-rs/observability/src/events.rs
+160
View File
@@ -13,6 +13,63 @@ pub enum InvocationType {
Implicit,
}
/// Status reported after a hook run reaches a terminal state.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum HookRunStatus {
/// The hook completed successfully.
Completed,
/// The hook failed.
Failed,
/// The hook blocked the triggering action.
Blocked,
/// The hook stopped execution.
Stopped,
}
/// Plugin lifecycle state after a plugin management operation.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PluginState {
/// Plugin was installed.
Installed,
/// Plugin was uninstalled.
Uninstalled,
/// Plugin was enabled.
Enabled,
/// Plugin was disabled.
Disabled,
}
/// Observation emitted when an app connector is mentioned during a turn.
#[derive(Observation)]
#[observation(name = "app.mentioned", crate = "crate", uses = ["analytics"])]
pub struct AppMentioned<'a> {
/// Model slug active for the turn where the app was mentioned.
#[obs(level = "basic", class = "operational")]
pub model_slug: &'a str,
/// Thread that owns the turn.
#[obs(level = "basic", class = "identifier")]
pub thread_id: &'a str,
/// Turn where the app was mentioned.
#[obs(level = "basic", class = "identifier")]
pub turn_id: &'a str,
/// Stable connector identifier when the app is backed by a connector.
#[obs(level = "basic", class = "identifier")]
pub connector_id: Option<&'a str>,
/// User-facing app name.
#[obs(level = "basic", class = "operational")]
pub app_name: Option<&'a str>,
/// Whether the mention was explicit or inferred.
#[obs(level = "basic", class = "operational")]
pub invocation_type: Option<InvocationType>,
}
/// Observation emitted when Codex uses an app connector during a turn.
#[derive(Observation)]
#[observation(name = "app.used", crate = "crate", uses = ["analytics"])]
@@ -41,3 +98,106 @@ pub struct AppUsed<'a> {
#[obs(level = "basic", class = "operational")]
pub invocation_type: Option<InvocationType>,
}
/// Observation emitted after a configured hook run completes.
#[derive(Observation)]
#[observation(name = "hook.run_completed", crate = "crate", uses = ["analytics"])]
pub struct HookRunCompleted<'a> {
/// Model slug active for the turn where the hook ran.
#[obs(level = "basic", class = "operational")]
pub model_slug: &'a str,
/// Thread that owns the turn.
#[obs(level = "basic", class = "identifier")]
pub thread_id: &'a str,
/// Turn where the hook ran.
#[obs(level = "basic", class = "identifier")]
pub turn_id: &'a str,
/// Hook lifecycle point, for example PostToolUse.
#[obs(level = "basic", class = "operational")]
pub hook_name: &'a str,
/// Source that configured the hook, for example user or project.
#[obs(level = "basic", class = "operational")]
pub hook_source: &'static str,
/// Final hook run status.
#[obs(level = "basic", class = "operational")]
pub status: HookRunStatus,
}
/// Observation emitted when a plugin capability is used during a turn.
#[derive(Observation)]
#[observation(name = "plugin.used", crate = "crate", uses = ["analytics"])]
pub struct PluginUsed<'a> {
/// Model slug active for the turn where the plugin was used.
#[obs(level = "basic", class = "operational")]
pub model_slug: &'a str,
/// Thread that owns the turn.
#[obs(level = "basic", class = "identifier")]
pub thread_id: &'a str,
/// Turn where the plugin was used.
#[obs(level = "basic", class = "identifier")]
pub turn_id: &'a str,
/// Stable plugin identifier.
#[obs(level = "basic", class = "identifier")]
pub plugin_id: &'a str,
/// Plugin name component.
#[obs(level = "basic", class = "operational")]
pub plugin_name: &'a str,
/// Marketplace or namespace component.
#[obs(level = "basic", class = "operational")]
pub marketplace_name: &'a str,
/// Whether the plugin exposes skills.
#[obs(level = "basic", class = "operational")]
pub has_skills: Option<bool>,
/// Number of MCP servers exposed by the plugin.
#[obs(level = "basic", class = "operational")]
pub mcp_server_count: Option<usize>,
/// Connector identifiers exposed by the plugin.
#[obs(level = "basic", class = "identifier")]
pub connector_ids: Option<&'a [String]>,
}
/// Observation emitted after a plugin lifecycle state changes.
#[derive(Observation)]
#[observation(name = "plugin.state_changed", crate = "crate", uses = ["analytics"])]
pub struct PluginStateChanged<'a> {
/// Stable plugin identifier.
#[obs(level = "basic", class = "identifier")]
pub plugin_id: &'a str,
/// Plugin name component.
#[obs(level = "basic", class = "operational")]
pub plugin_name: &'a str,
/// Marketplace or namespace component.
#[obs(level = "basic", class = "operational")]
pub marketplace_name: &'a str,
/// Whether the plugin exposes skills.
#[obs(level = "basic", class = "operational")]
pub has_skills: Option<bool>,
/// Number of MCP servers exposed by the plugin.
#[obs(level = "basic", class = "operational")]
pub mcp_server_count: Option<usize>,
/// Connector identifiers exposed by the plugin.
#[obs(level = "basic", class = "identifier")]
pub connector_ids: Option<&'a [String]>,
/// New plugin lifecycle state.
#[obs(level = "basic", class = "operational")]
pub state: PluginState,
}
docs/observability-event-stream-design.md
+5 -3
View File
@@ -276,7 +276,7 @@ Initial workflow coverage should be chosen by conformance need:
| Tools | `tool_call.started`, `tool_call.approval_resolved`, `tool_call.ended` | OTEL, rollout |
| Compaction | `compaction.started`, `compaction.installed`, `compaction.ended` | analytics, rollout |
| Agents | `agent.task_sent`, `agent.message_sent`, `agent.result_delivered`, `agent.closed` | rollout, analytics subset |
| Product features | `skill.invoked`, `app.used`, `plugin.state_changed`, `guardian.review_completed` | analytics |
| Product features | `app.mentioned`, `app.used`, `hook.run_completed`, `plugin.used`, `plugin.state_changed`, `guardian.review_completed` | analytics |
| Transport/auth | `transport.api_request_completed`, `transport.websocket_request_completed`, `auth.recovery_step_completed` | OTEL |
This table is not a complete schema. Each event still needs a typed Rust
@@ -297,8 +297,10 @@ output schemas.
| compaction lifecycle observations | `codex_compaction_event` |
| skill/app/plugin/guardian observations | existing feature events |
The analytics reducer continues to own batching, deduplication, connection
metadata joins, and product event naming.
The analytics crate should keep the observation-to-legacy-schema translation in
a private projection module. The reducer then stays responsible for ingestion
orchestration, batching, deduplication, connection metadata joins, and product
event naming rather than becoming a mixed bag of mapping code.
### Rollout Trace