From 10f7e100d96bf6c73c31af5a02479e3f18db4f50 Mon Sep 17 00:00:00 2001 From: Albin Cassirer Date: Sun, 19 Apr 2026 15:35:17 -0700 Subject: [PATCH] [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. --- .../analytics/src/analytics_client_tests.rs | 117 ++++++++++++ codex-rs/analytics/src/lib.rs | 1 + .../analytics/src/observation_projection.rs | 166 ++++++++++++++++++ codex-rs/analytics/src/observation_reducer.rs | 72 +++++--- codex-rs/observability/src/events.rs | 160 +++++++++++++++++ docs/observability-event-stream-design.md | 8 +- 6 files changed, 499 insertions(+), 25 deletions(-) create mode 100644 codex-rs/analytics/src/observation_projection.rs diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index b765bfd2d4..ebf6066384 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -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(); diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index 6b24006570..1283206986 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -1,6 +1,7 @@ mod client; mod events; mod facts; +mod observation_projection; mod observation_reducer; mod reducer; diff --git a/codex-rs/analytics/src/observation_projection.rs b/codex-rs/analytics/src/observation_projection.rs new file mode 100644 index 0000000000..4453b258ef --- /dev/null +++ b/codex-rs/analytics/src/observation_projection.rs @@ -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, +) -> 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, + mcp_server_count: Option, + 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, + } +} diff --git a/codex-rs/analytics/src/observation_reducer.rs b/codex-rs/analytics/src/observation_reducer.rs index cf660ec116..56d677ab10 100644 --- a/codex-rs/analytics/src/observation_reducer.rs +++ b/codex-rs/analytics/src/observation_reducer.rs @@ -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, + ) { + 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, ) { - 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, + ) { + 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, + ) { + 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, + ) { + out.push(observation_projection::plugin_state_changed_event( + observation, + )); } } diff --git a/codex-rs/observability/src/events.rs b/codex-rs/observability/src/events.rs index 899714d57d..61ede20de5 100644 --- a/codex-rs/observability/src/events.rs +++ b/codex-rs/observability/src/events.rs @@ -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, +} + /// 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, } + +/// 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, + + /// Number of MCP servers exposed by the plugin. + #[obs(level = "basic", class = "operational")] + pub mcp_server_count: Option, + + /// 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, + + /// Number of MCP servers exposed by the plugin. + #[obs(level = "basic", class = "operational")] + pub mcp_server_count: Option, + + /// 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, +} diff --git a/docs/observability-event-stream-design.md b/docs/observability-event-stream-design.md index 7b743134a9..6c5b4f443a 100644 --- a/docs/observability-event-stream-design.md +++ b/docs/observability-event-stream-design.md @@ -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