From c58e8e329fe70ca085ebb34baf008ca9272f45ad Mon Sep 17 00:00:00 2001 From: Albin Cassirer Date: Sun, 19 Apr 2026 17:32:36 -0700 Subject: [PATCH] Add skill invoked observation projection\n\nAdd a skill.invoked observation and map it through the existing analytics\nskill invocation reducer path. Extend the feature conformance test so the\nlegacy skill fact and shared observation produce the same analytics payload. --- .../analytics/src/analytics_client_tests.rs | 30 +++++++++++++ .../analytics/src/observation_projection.rs | 24 ++++++++++ codex-rs/analytics/src/observation_reducer.rs | 16 +++++++ codex-rs/observability/src/events.rs | 44 +++++++++++++++++++ docs/observability-event-stream-design.md | 2 +- 5 files changed, 115 insertions(+), 1 deletion(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 1ae1c60544..1c03e99819 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -1630,6 +1630,22 @@ async fn feature_observations_match_legacy_analytics_facts() { turn_id: "turn-1".to_string(), }; let connector_ids = vec!["calendar".to_string(), "drive".to_string()]; + let skill_path = PathBuf::from("/Users/abc/.codex/skills/doc/SKILL.md"); + + legacy_reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::SkillInvoked(SkillInvokedInput { + tracking: tracking.clone(), + invocations: vec![SkillInvocation { + skill_name: "doc".to_string(), + skill_scope: codex_protocol::protocol::SkillScope::User, + skill_path: skill_path.clone(), + invocation_type: InvocationType::Explicit, + }], + })), + &mut legacy_events, + ) + .await; legacy_reducer .ingest( @@ -1678,6 +1694,20 @@ async fn feature_observations_match_legacy_analytics_facts() { ) .await; + observation_reducer + .ingest_skill_invoked( + codex_observability::events::SkillInvoked { + model_slug: "gpt-5", + thread_id: "thread-1", + turn_id: "turn-1", + skill_name: "doc", + skill_scope: codex_observability::events::SkillScope::User, + skill_path: skill_path.as_path(), + invocation_type: codex_observability::events::InvocationType::Explicit, + }, + &mut observation_events, + ) + .await; observation_reducer .ingest_app_mentioned( codex_observability::events::AppMentioned { diff --git a/codex-rs/analytics/src/observation_projection.rs b/codex-rs/analytics/src/observation_projection.rs index 18c9bd284d..c010240ec9 100644 --- a/codex-rs/analytics/src/observation_projection.rs +++ b/codex-rs/analytics/src/observation_projection.rs @@ -16,6 +16,8 @@ use crate::facts; use crate::facts::AppInvocation; use crate::facts::AppMentionedInput; use crate::facts::AppUsedInput; +use crate::facts::SkillInvocation; +use crate::facts::SkillInvokedInput; use crate::facts::TrackEventsContext; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnSubmissionType as AnalyticsTurnSubmissionType; @@ -40,8 +42,30 @@ use codex_protocol::protocol::NetworkAccess; use codex_protocol::protocol::ReadOnlyAccess; use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SkillScope as ProtocolSkillScope; use codex_protocol::protocol::TokenUsage; +pub(crate) fn skill_invoked_input(observation: events::SkillInvoked<'_>) -> SkillInvokedInput { + SkillInvokedInput { + tracking: tracking_from_fields( + observation.model_slug, + observation.thread_id, + observation.turn_id, + ), + invocations: vec![SkillInvocation { + skill_name: observation.skill_name.to_string(), + skill_scope: match observation.skill_scope { + events::SkillScope::User => ProtocolSkillScope::User, + events::SkillScope::Repo => ProtocolSkillScope::Repo, + events::SkillScope::System => ProtocolSkillScope::System, + events::SkillScope::Admin => ProtocolSkillScope::Admin, + }, + skill_path: observation.skill_path.to_path_buf(), + invocation_type: map_invocation_type(observation.invocation_type), + }], + } +} + pub(crate) fn app_mentioned_input(observation: events::AppMentioned<'_>) -> AppMentionedInput { AppMentionedInput { tracking: tracking_from_fields( diff --git a/codex-rs/analytics/src/observation_reducer.rs b/codex-rs/analytics/src/observation_reducer.rs index 8423934b82..39a98169f3 100644 --- a/codex-rs/analytics/src/observation_reducer.rs +++ b/codex-rs/analytics/src/observation_reducer.rs @@ -35,6 +35,22 @@ impl AnalyticsObservationReducer { self.legacy.ingest(fact, out).await; } + /// Ingests a skill.invoked observation and emits the current analytics event. + pub(crate) async fn ingest_skill_invoked( + &mut self, + observation: events::SkillInvoked<'_>, + out: &mut Vec, + ) { + self.legacy + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::SkillInvoked( + observation_projection::skill_invoked_input(observation), + )), + out, + ) + .await; + } + /// Ingests an app.mentioned observation and emits the current analytics event. pub(crate) async fn ingest_app_mentioned( &mut self, diff --git a/codex-rs/observability/src/events.rs b/codex-rs/observability/src/events.rs index f728d05e3c..69e714ab9f 100644 --- a/codex-rs/observability/src/events.rs +++ b/codex-rs/observability/src/events.rs @@ -2,6 +2,7 @@ use crate::Observation; use serde::Serialize; +use std::path::Path; mod compaction; mod review; @@ -23,6 +24,16 @@ pub enum InvocationType { Implicit, } +/// Scope where a skill definition was found. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum SkillScope { + User, + Repo, + System, + Admin, +} + /// Status reported after a hook run reaches a terminal state. #[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] #[serde(rename_all = "snake_case")] @@ -51,6 +62,39 @@ pub enum PluginState { Disabled, } +/// Observation emitted when a skill is invoked during a turn. +#[derive(Observation)] +#[observation(name = "skill.invoked", crate = "crate", uses = ["analytics"])] +pub struct SkillInvoked<'a> { + /// Model slug active for the turn where the skill was invoked. + #[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 skill was invoked. + #[obs(level = "basic", class = "identifier")] + pub turn_id: &'a str, + + /// Skill display name. + #[obs(level = "basic", class = "operational")] + pub skill_name: &'a str, + + /// Scope where the skill was discovered. + #[obs(level = "basic", class = "operational")] + pub skill_scope: SkillScope, + + /// Local skill definition path used to derive the existing analytics skill id. + #[obs(level = "basic", class = "environment")] + pub skill_path: &'a Path, + + /// Whether the skill was explicitly requested or inferred. + #[obs(level = "basic", class = "operational")] + pub invocation_type: InvocationType, +} + /// 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 221890b55f..6a7cc4cf53 100644 --- a/docs/observability-event-stream-design.md +++ b/docs/observability-event-stream-design.md @@ -285,7 +285,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 | `app.mentioned`, `app.used`, `hook.run_completed`, `plugin.used`, `plugin.state_changed`, `review.completed` | analytics | +| Product features | `skill.invoked`, `app.mentioned`, `app.used`, `hook.run_completed`, `plugin.used`, `plugin.state_changed`, `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