From 75aba6768a32622f145f97c6ce6c46fd92f4271a Mon Sep 17 00:00:00 2001 From: Anthony Tafoya Date: Wed, 4 Nov 2026 02:15:29 +0000 Subject: [PATCH] Add outcome metrics for skill reads and installation listings (#59295) ## What changed - Record `codex.skills.read.result` and `codex.skills.installations.result` counters with bounded outcome labels for successes and failures. - Tag both counters with `discovery_version` and installation listings with `source_filter`. - Make session telemetry available through the step store so skill tools emit metrics with the current step's model metadata. ## Testing Add coverage for read metrics in both discovery versions, installation listing outcomes and source filters, and extension metrics retaining the current model metadata after a model switch. GitOrigin-RevId: 0b5418e2b94a89149a20877f32ca2b2e196884e7 --- codex-rs/core/src/session/mod.rs | 3 + codex-rs/core/src/session/tests.rs | 62 +- .../ext/skills/src/tools/installations.rs | 189 ++++-- codex-rs/ext/skills/src/tools/mod.rs | 9 +- codex-rs/ext/skills/src/tools/read.rs | 608 ++++++++++-------- codex-rs/ext/skills/tests/skills_extension.rs | 29 +- .../skills_extension/merge_catalog_tests.rs | 4 + .../skills_extension/tool_metrics_tests.rs | 195 ++++++ 8 files changed, 732 insertions(+), 367 deletions(-) create mode 100644 codex-rs/ext/skills/tests/skills_extension/tool_metrics_tests.rs diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 729db3180d..60f8fc46ec 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -4316,6 +4316,9 @@ impl Session { } let extension_data = codex_extension_api::ExtensionData::new(turn_context.sub_id.clone()); + extension_data.insert(extension_metrics::from_session_telemetry( + session_telemetry.clone(), + )); extension_data.insert_arc(Arc::clone(&settings.model_info.model)); extension_data.insert_arc( self.host_skills_for_step(&turn_context, &environments) diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index c8494dd0b0..98da3f4737 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -465,9 +465,8 @@ fn single_histogram_attributes( .collect() } -#[test] -fn extension_metrics_preserve_session_metadata_tags() { - let metrics = MetricsClient::new( +fn test_metrics_client() -> MetricsClient { + MetricsClient::new( MetricsConfig::in_memory( "test", "codex-core", @@ -476,7 +475,12 @@ fn extension_metrics_preserve_session_metadata_tags() { ) .with_runtime_reader(), ) - .expect("in-memory metrics client"); + .expect("in-memory metrics client") +} + +#[test] +fn extension_metrics_preserve_session_metadata_tags() { + let metrics = test_metrics_client(); let session_telemetry = SessionTelemetry::new( ThreadId::new(), "gpt-5.4", @@ -565,25 +569,24 @@ async fn world_state_extension_metrics_follow_turn_model_switch() { Vec, > { Box::pin(async move { - input - .extension_metrics - .expect("turn metrics should be available") - .histogram("codex.test.extension.turn", /*value*/ 1, &[]); + for metrics in [ + input.extension_metrics.expect("turn metrics"), + Arc::clone( + input + .step_store + .get::>() + .expect("step metrics") + .as_ref(), + ), + ] { + metrics.histogram("codex.test.extension.turn", /*value*/ 1, &[]); + } Vec::new() }) } } - let metrics = MetricsClient::new( - MetricsConfig::in_memory( - "test", - "codex-core", - env!("CARGO_PKG_VERSION"), - InMemoryMetricExporter::default(), - ) - .with_runtime_reader(), - ) - .expect("in-memory metrics client"); + let metrics = test_metrics_client(); let (mut session, mut turn_context) = make_session_and_context().await; turn_context.session_telemetry = turn_context .session_telemetry @@ -594,16 +597,27 @@ async fn world_state_extension_metrics_follow_turn_model_switch() { } else { "gpt-5.4" }; - let turn_context = Arc::new( - turn_context - .with_model(next_model.to_string(), &session.services.models_manager) - .await, - ); + let mut turn_context = turn_context + .with_model(next_model.to_string(), &session.services.models_manager) + .await; + turn_context.session_telemetry = turn_context + .session_telemetry + .clone() + .with_model("startup-model", "startup-model"); + let turn_context = Arc::new(turn_context); let mut builder = codex_extension_api::ExtensionRegistryBuilder::::new(); builder.prompt_contributor(Arc::new(WorldStateMetricsRecorder)); session.services.extensions = Arc::new(builder.build()); - let _world_state = build_world_state_from_turn_context(&session, &turn_context).await; + let session = Arc::new(session); + let step = session + .capture_step_context(turn_context, &CancellationToken::new()) + .await + .expect("step capture should succeed after switching to a supported model"); + session + .build_world_state_for_step(&step, /*new_window*/ false) + .await + .expect("world state preparation should succeed and emit extension metrics"); let snapshot = metrics.snapshot().expect("metrics snapshot"); let attributes = single_histogram_attributes(&snapshot, "codex.test.extension.turn"); diff --git a/codex-rs/ext/skills/src/tools/installations.rs b/codex-rs/ext/skills/src/tools/installations.rs index 5391f87e98..e4f97e193c 100644 --- a/codex-rs/ext/skills/src/tools/installations.rs +++ b/codex-rs/ext/skills/src/tools/installations.rs @@ -1,7 +1,9 @@ //! Retrieves installation locations from the prepared catalog without reading skill contents. use std::collections::HashSet; +use std::sync::Arc; +use codex_extension_api::ExtensionMetrics; use codex_extension_api::FunctionCallError; use codex_extension_api::ToolCall; use codex_extension_api::ToolExecutor; @@ -49,6 +51,7 @@ struct InstallationsResponse { pub(super) struct InstallationsTool { pub(super) context: SkillToolContext, + pub(super) metrics: Option>, } impl<'call> ToolExecutor> for InstallationsTool { @@ -72,77 +75,127 @@ impl<'call> ToolExecutor> for InstallationsTool { 'call: 'a, { Box::pin(async move { - let args: InstallationsArgs = parse_args(&call)?; - validate_handle("package", args.package.as_str(), MAX_HANDLE_BYTES)?; - if let Some(source) = &args.source { - source.validate()?; - } - let matches = self.context.matching_packages(&args.package); - let (entry, _) = matches.first().ok_or_else(|| { - FunctionCallError::RespondToModel("skill package is not available".to_string()) - })?; - let mut installations = Vec::new(); - let mut seen = HashSet::new(); - for installation in matches - .iter() - .flat_map(|(entry, matched_source)| { - std::iter::once(matched_source.clone()).chain(entry.source_locations()) - }) - .filter(|source| { - args.source - .as_ref() - .is_none_or(|filter| filter.matches(source)) - }) - .filter_map(|source| InstallationLocation::from_source(&source)) - { - if seen.insert(installation.clone()) { - installations.push(installation); + let mut outcome = "internal_error"; + let mut source_filter = "unparsed"; + let result = (|| { + let args: InstallationsArgs = + parse_args(&call).inspect_err(|_| outcome = "invalid_arguments")?; + source_filter = match &args.source { + None => "all", + Some(InstallationSource::Cloud) => "cloud", + Some(InstallationSource::Executor { .. }) => "executor", + }; + validate_handle("package", args.package.as_str(), MAX_HANDLE_BYTES) + .inspect_err(|_| outcome = "invalid_package")?; + if let Some(source) = &args.source { + source + .validate() + .inspect_err(|_| outcome = "invalid_source")?; } - } - let identity = (&entry.id.0, &args.source, &installations); - let start = - parse_pagination_cursor(args.cursor.as_deref(), &identity, "skills.installations")?; - if start > installations.len() { - return Err(FunctionCallError::RespondToModel( - "skills.installations cursor is invalid".to_string(), - )); - } - let fingerprint = value_fingerprint(&identity); - let cursor_at = |offset| format!("{fingerprint:016x}:{offset}"); - let budget = call.response_byte_budget(MAX_INSTALLATIONS_RESPONSE_BYTES); - let mut response = InstallationsResponse { - installations: Vec::new(), - next_cursor: None, - }; - let installation_count = installations.len(); - for (index, installation) in installations.into_iter().enumerate().skip(start) { - response.installations.push(installation); - response.next_cursor = - (index + 1 < installation_count).then(|| cursor_at(index + 1)); - if serialized_len(&response)? > budget { - response.installations.pop(); - response.next_cursor = Some(cursor_at(index)); - break; + let matches = self.context.matching_packages(&args.package); + let (entry, _) = matches.first().ok_or_else(|| { + outcome = if args.package.alias_source().is_some() + && self + .context + .step_state + .catalog + .aliases + .resolve(&args.package) + .is_none() + { + "unresolved_alias" + } else { + "package_unavailable" + }; + FunctionCallError::RespondToModel("skill package is not available".to_string()) + })?; + let mut installations = Vec::new(); + let mut seen = HashSet::new(); + for installation in matches + .iter() + .flat_map(|(entry, matched_source)| { + std::iter::once(matched_source.clone()).chain(entry.source_locations()) + }) + .filter(|source| { + args.source + .as_ref() + .is_none_or(|filter| filter.matches(source)) + }) + .filter_map(|source| InstallationLocation::from_source(&source)) + { + if seen.insert(installation.clone()) { + installations.push(installation); + } } + let identity = (&entry.id.0, &args.source, &installations); + let start = parse_pagination_cursor( + args.cursor.as_deref(), + &identity, + "skills.installations", + ) + .inspect_err(|_| outcome = "invalid_cursor")?; + if start > installations.len() { + outcome = "invalid_cursor"; + return Err(FunctionCallError::RespondToModel( + "skills.installations cursor is invalid".to_string(), + )); + } + let fingerprint = value_fingerprint(&identity); + let cursor_at = |offset| format!("{fingerprint:016x}:{offset}"); + let budget = call.response_byte_budget(MAX_INSTALLATIONS_RESPONSE_BYTES); + let mut response = InstallationsResponse { + installations: Vec::new(), + next_cursor: None, + }; + let installation_count = installations.len(); + for (index, installation) in installations.into_iter().enumerate().skip(start) { + response.installations.push(installation); + response.next_cursor = + (index + 1 < installation_count).then(|| cursor_at(index + 1)); + if serialized_len(&response)? > budget { + response.installations.pop(); + response.next_cursor = Some(cursor_at(index)); + break; + } + } + if (start < installation_count && response.installations.is_empty()) + || serialized_len(&response)? > budget + { + outcome = "response_budget"; + return Err(FunctionCallError::RespondToModel( + "skills.installations response budget leaves no room for an installation" + .to_string(), + )); + } + let authority = if response + .installations + .iter() + .any(|location| matches!(location, InstallationLocation::Cloud { .. })) + { + SkillToolAuthoritySelector::Cloud + } else { + SkillToolAuthoritySelector::Executor + }; + let output = skill_json_output(&response, authority)?; + outcome = if response.installations.is_empty() { + "empty" + } else { + "success" + }; + Ok(output) + })(); + if let Some(metrics) = &self.metrics { + metrics.counter( + "codex.skills.installations.result", + /*inc*/ 1, + &[ + ("outcome", outcome), + ("source_filter", source_filter), + ("discovery_version", "v2"), + ], + ); } - if (start < installation_count && response.installations.is_empty()) - || serialized_len(&response)? > budget - { - return Err(FunctionCallError::RespondToModel( - "skills.installations response budget leaves no room for an installation" - .to_string(), - )); - } - let authority = if response - .installations - .iter() - .any(|location| matches!(location, InstallationLocation::Cloud { .. })) - { - SkillToolAuthoritySelector::Cloud - } else { - SkillToolAuthoritySelector::Executor - }; - skill_json_output(&response, authority) + result }) } } diff --git a/codex-rs/ext/skills/src/tools/mod.rs b/codex-rs/ext/skills/src/tools/mod.rs index ea89558116..580cbfbc2f 100644 --- a/codex-rs/ext/skills/src/tools/mod.rs +++ b/codex-rs/ext/skills/src/tools/mod.rs @@ -73,6 +73,9 @@ pub(crate) fn skill_tools( .get::() .and_then(|state| state.mcp_resources.clone()); let analytics = SkillAnalytics::from_stores(session_store, thread_store); + let metrics = step_store + .get::>() + .map(|metrics| Arc::clone(metrics.as_ref())); let context = SkillToolContext { thread_id: thread_store.level_id().to_string(), providers, @@ -87,10 +90,14 @@ pub(crate) fn skill_tools( }), Arc::new(read::ReadTool { context: context.clone(), + metrics: metrics.clone(), }), ]; if context.step_state.discovery_v2 { - tools.push(Arc::new(installations::InstallationsTool { context })); + tools.push(Arc::new(installations::InstallationsTool { + context, + metrics, + })); } tools } diff --git a/codex-rs/ext/skills/src/tools/read.rs b/codex-rs/ext/skills/src/tools/read.rs index a8dcaa19e6..6390c1745c 100644 --- a/codex-rs/ext/skills/src/tools/read.rs +++ b/codex-rs/ext/skills/src/tools/read.rs @@ -1,6 +1,7 @@ use std::sync::Arc; use codex_analytics::InvocationType; +use codex_extension_api::ExtensionMetrics; use codex_extension_api::FunctionCallError; use codex_extension_api::ToolCall; use codex_extension_api::ToolExecutor; @@ -87,6 +88,7 @@ enum ReadMetadata { #[derive(Clone)] pub(super) struct ReadTool { pub(super) context: SkillToolContext, + pub(super) metrics: Option>, } impl<'call> ToolExecutor> for ReadTool { @@ -116,289 +118,351 @@ impl<'call> ToolExecutor> for ReadTool { 'call: 'a, { Box::pin(async move { - let (args, source) = if self.context.step_state.discovery_v2 { - let args: ReadInstallationArgs = parse_args(&call)?; - (args.read, args.source) - } else { - (parse_args::(&call)?, None) - }; - if let Some(source) = &source { - source.validate()?; - } - let response_byte_budget = call.response_byte_budget(MAX_SKILL_RESPONSE_BYTES); - validate_handle("package", args.package.as_str(), MAX_HANDLE_BYTES)?; - if let Some(resource) = args.resource.as_deref() { - validate_handle("resource", resource, MAX_HANDLE_BYTES)?; - } - - let matches = self.context.matching_packages(&args.package); - let (skill_entry, selected_source, output_authority) = matches - .iter() - .find_map(|(entry, matched_source)| { - let selected_source = match &source { - None => matched_source.clone(), - Some(filter) => std::iter::once(matched_source.clone()) - .chain(entry.source_locations()) - .find(|candidate| filter.matches(candidate))?, - }; - let authority = - SkillToolAuthority::from_authority(&selected_source.authority)?.selector(); - Some((*entry, selected_source, authority)) - }) - .ok_or_else(|| { - FunctionCallError::RespondToModel(if source.is_some() { - "skill package is not available from the requested source".to_string() - } else { - "skill package is not available".to_string() - }) - })?; - let authority = selected_source.authority.clone(); - let package = selected_source.package.clone(); - let main_prompt = selected_source.main_prompt.clone(); - let requested_resource = match args.resource { - None => main_prompt.clone(), - Some(resource) if resource == main_prompt.as_str() => main_prompt.clone(), - Some(resource) => main_prompt - .bind_environment_package_resource(&package, resource.clone()) - .unwrap_or_else(|| SkillResourceId::new(resource)), - }; - // Record the selected route before either cache or provider can serve the read. - // This is an attempted read, not a successful invocation or a backend RPC. - tracing::event!( - name: "codex.skill_read.source_selected", - target: "codex_otel.trace_safe", - tracing::Level::INFO, - event.name = "codex.skill_read.source_selected", - conversation.id = %self.context.thread_id, - turn_id = %call.turn_id, - call_id = %call.call_id, - tool_namespace = SKILLS_NAMESPACE, - tool_name = TOOL_NAME, - skill_source = match output_authority { - SkillToolAuthoritySelector::Cloud => "cloud", - SkillToolAuthoritySelector::Executor => "executor", - }, - skill_name = %skill_entry.name, - skill_plugin_id = selected_source.plugin_id.as_deref(), - skill_resource_kind = if requested_resource == main_prompt { - "main_prompt" + let mut outcome = "internal_error"; + let result = async { + let (args, source) = if self.context.step_state.discovery_v2 { + let args: ReadInstallationArgs = + parse_args(&call).inspect_err(|_| outcome = "invalid_arguments")?; + (args.read, args.source) } else { - "supporting_resource" - }, - is_continuation = args.cursor.is_some(), - "Selected skill read source" - ); - let (context, access_key) = match output_authority { - SkillToolAuthoritySelector::Cloud => ( - SkillReadContext::Cloud { - mcp_resources: self.context.mcp_resources.clone(), - }, - None, - ), - SkillToolAuthoritySelector::Executor => { - let fs = requested_resource - .environment_path() - .and_then(|(id, _)| { - call.environments - .iter() - .find(|env| env.environment_id == id) - }) - .map(codex_extension_api::ToolEnvironment::fs) - .ok_or_else(|| { - FunctionCallError::RespondToModel( - "skill environment is not available for this callback".to_string(), - ) - })?; - (SkillReadContext::Executor { fs }, Some(fs.cache_key())) - } - }; - let cursor_start = |contents: &str| { - if self.context.step_state.discovery_v2 { - parse_pagination_cursor( - args.cursor.as_deref(), - &(&source, &package, contents), - "skills.read", + ( + parse_args::(&call) + .inspect_err(|_| outcome = "invalid_arguments")?, + None, ) - } else { - parse_pagination_cursor(args.cursor.as_deref(), contents, "skills.read") + }; + if let Some(source) = &source { + source + .validate() + .inspect_err(|_| outcome = "invalid_arguments")?; } - }; - // Reuse the snapshot to avoid reading the whole file for each page. File edits - // are not detected on cache hits, so a continuation can return old contents. - // Snapshots have no expiry and are not cleared after the final page. - let cached = args.cursor.as_deref().and_then(|_| { - let access = access_key.as_ref()?; - let snapshot = self - .context - .thread_state - .executor_read_snapshot - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let snapshot = snapshot.as_ref().filter(|snapshot| { - snapshot.authority == authority - && snapshot.package == package - && snapshot.result.resource == requested_resource - && snapshot.access == *access - })?; - let start = cursor_start(&snapshot.result.contents).ok()?; - Some((Arc::clone(&snapshot.result), start)) - }); - let (result, start) = match cached { - Some(cached) => cached, - None => { - let result = self + let response_byte_budget = call.response_byte_budget(MAX_SKILL_RESPONSE_BYTES); + validate_handle("package", args.package.as_str(), MAX_HANDLE_BYTES) + .inspect_err(|_| outcome = "invalid_arguments")?; + if let Some(resource) = args.resource.as_deref() { + validate_handle("resource", resource, MAX_HANDLE_BYTES) + .inspect_err(|_| outcome = "invalid_arguments")?; + } + + let matches = self.context.matching_packages(&args.package); + let (skill_entry, selected_source, output_authority) = matches + .iter() + .find_map(|(entry, matched_source)| { + let selected_source = match &source { + None => matched_source.clone(), + Some(filter) => std::iter::once(matched_source.clone()) + .chain(entry.source_locations()) + .find(|candidate| filter.matches(candidate))?, + }; + let authority = + SkillToolAuthority::from_authority(&selected_source.authority)? + .selector(); + Some((*entry, selected_source, authority)) + }) + .ok_or_else(|| { + outcome = if !matches.is_empty() { + "source_unavailable" + } else if args.package.alias_source().is_some() + && self + .context + .step_state + .catalog + .aliases + .resolve(&args.package) + .is_none() + { + "unresolved_alias" + } else { + "package_unavailable" + }; + FunctionCallError::RespondToModel(if source.is_some() { + "skill package is not available from the requested source".to_string() + } else { + "skill package is not available".to_string() + }) + })?; + let authority = selected_source.authority.clone(); + let package = selected_source.package.clone(); + let main_prompt = selected_source.main_prompt.clone(); + let requested_resource = match args.resource { + None => main_prompt.clone(), + Some(resource) if resource == main_prompt.as_str() => main_prompt.clone(), + Some(resource) => main_prompt + .bind_environment_package_resource(&package, resource.clone()) + .unwrap_or_else(|| SkillResourceId::new(resource)), + }; + // Record the selected route before either cache or provider can serve the read. + // This is an attempted read, not a successful invocation or a backend RPC. + tracing::event!( + name: "codex.skill_read.source_selected", + target: "codex_otel.trace_safe", + tracing::Level::INFO, + event.name = "codex.skill_read.source_selected", + conversation.id = %self.context.thread_id, + turn_id = %call.turn_id, + call_id = %call.call_id, + tool_namespace = SKILLS_NAMESPACE, + tool_name = TOOL_NAME, + skill_source = match output_authority { + SkillToolAuthoritySelector::Cloud => "cloud", + SkillToolAuthoritySelector::Executor => "executor", + }, + skill_name = %skill_entry.name, + skill_plugin_id = selected_source.plugin_id.as_deref(), + skill_resource_kind = if requested_resource == main_prompt { + "main_prompt" + } else { + "supporting_resource" + }, + is_continuation = args.cursor.is_some(), + "Selected skill read source" + ); + let (context, access_key) = match output_authority { + SkillToolAuthoritySelector::Cloud => ( + SkillReadContext::Cloud { + mcp_resources: self.context.mcp_resources.clone(), + }, + None, + ), + SkillToolAuthoritySelector::Executor => { + let fs = requested_resource + .environment_path() + .and_then(|(id, _)| { + call.environments + .iter() + .find(|env| env.environment_id == id) + }) + .map(codex_extension_api::ToolEnvironment::fs) + .ok_or_else(|| { + outcome = "environment_unavailable"; + FunctionCallError::RespondToModel( + "skill environment is not available for this callback" + .to_string(), + ) + })?; + (SkillReadContext::Executor { fs }, Some(fs.cache_key())) + } + }; + let cursor_start = |contents: &str| { + if self.context.step_state.discovery_v2 { + parse_pagination_cursor( + args.cursor.as_deref(), + &(&source, &package, contents), + "skills.read", + ) + } else { + parse_pagination_cursor(args.cursor.as_deref(), contents, "skills.read") + } + }; + // Reuse the snapshot to avoid reading the whole file for each page. File edits + // are not detected on cache hits, so a continuation can return old contents. + // Snapshots have no expiry and are not cleared after the final page. + let cached = args.cursor.as_deref().and_then(|_| { + let access = access_key.as_ref()?; + let snapshot = self .context .thread_state - .read_skill( - &self.context.providers, - SkillReadRequest { - authority: authority.clone(), - package: package.clone(), - resource: requested_resource.clone(), - context, - }, - ) - .await - .map_err(|err| { - tracing::warn!( - error = %err, - turn_id = %call.turn_id, - call_id = %call.call_id, - resource = requested_resource.as_str(), - "skills.read provider request failed" - ); - FunctionCallError::RespondToModel( - "failed to read skill resource".to_string(), - ) - })?; - if result.resource != requested_resource { - return Err(FunctionCallError::Fatal( - "skill provider returned a different resource".to_string(), - )); - } - let start = cursor_start(&result.contents)?; - (Arc::new(result), start) - } - }; - if start > result.contents.len() || !result.contents.is_char_boundary(start) { - return Err(FunctionCallError::RespondToModel( - "skills.read cursor is invalid".to_string(), - )); - } - // Shadow candidates use the merged identity, not the selected read route. - if skill_entry.authority.kind == SkillSourceKind::Cloud - && let Some(state) = self - .context - .thread_state - .shadow_selection_turn(&call.turn_id) - { - state.record_invocation(skill_entry.main_prompt.as_str()); - } - let metadata = if self.context.step_state.discovery_v2 { - let mut installations = Vec::new(); - let mut installations_truncated = false; - let mut installation_bytes = 2; // JSON array brackets. - let installation_budget = - MAX_INSTALLATION_METADATA_BYTES.min(response_byte_budget / 4); - // Prefer the installation used by this read before other sources fill the budget. - for source in - std::iter::once(selected_source.clone()).chain(skill_entry.source_locations()) - { - let Some(installation) = InstallationLocation::from_source(&source) else { - continue; - }; - if installations.contains(&installation) { - continue; - } - let size = - serialized_len(&installation)? + usize::from(!installations.is_empty()); - if installation_bytes + size > installation_budget { - installations_truncated = true; - continue; - } - installation_bytes += size; - installations.push(installation); - } - installations_truncated |= matches.iter().any(|(entry, _)| { - entry - .source_locations() - .filter_map(|source| InstallationLocation::from_source(&source)) - .any(|location| !installations.contains(&location)) + .executor_read_snapshot + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let snapshot = snapshot.as_ref().filter(|snapshot| { + snapshot.authority == authority + && snapshot.package == package + && snapshot.result.resource == requested_resource + && snapshot.access == *access + })?; + let start = cursor_start(&snapshot.result.contents).ok()?; + Some((Arc::clone(&snapshot.result), start)) }); - ReadMetadata::Installations(InstallationReadMetadata { - installations, - installations_truncated, - }) - } else { - let skill_root = if output_authority == SkillToolAuthoritySelector::Executor { - main_prompt - .environment_path() - .and_then(|(_, path)| path.parent()) - .map(|path| path.inferred_native_path_string()) - } else { - None - }; - ReadMetadata::Legacy(LegacyReadMetadata { skill_root }) - }; - let response = page_response( - result.resource.as_str(), - &result.contents, - &metadata, - start, - response_byte_budget, - |offset| { - if self.context.step_state.discovery_v2 { - pagination_cursor(&(&source, &package, result.contents.as_str()), offset) - } else { - pagination_cursor(result.contents.as_str(), offset) + let (result, start) = match cached { + Some(cached) => cached, + None => { + let result = self + .context + .thread_state + .read_skill( + &self.context.providers, + SkillReadRequest { + authority: authority.clone(), + package: package.clone(), + resource: requested_resource.clone(), + context, + }, + ) + .await + .map_err(|err| { + outcome = "read_failed"; + tracing::warn!( + error = %err, + turn_id = %call.turn_id, + call_id = %call.call_id, + resource = requested_resource.as_str(), + "skills.read provider request failed" + ); + FunctionCallError::RespondToModel( + "failed to read skill resource".to_string(), + ) + })?; + if result.resource != requested_resource { + return Err(FunctionCallError::Fatal( + "skill provider returned a different resource".to_string(), + )); + } + let start = cursor_start(&result.contents) + .inspect_err(|_| outcome = "invalid_cursor")?; + (Arc::new(result), start) } - }, - )?; - let response_authority = if matches!( - &response.metadata, - ReadMetadata::Installations(metadata) - if metadata.installations.iter().any(|location| { - matches!(location, InstallationLocation::Cloud { .. }) - }) - ) { - SkillToolAuthoritySelector::Cloud - } else { - output_authority - }; - let output = skill_json_output(&response, response_authority)?; - if output_authority == super::SkillToolAuthoritySelector::Executor - && response.next_cursor.is_some() - && result.contents.len() <= MAX_SKILL_RESOURCE_CONTENT_BYTES - && let Some(access) = access_key - { - *self - .context - .thread_state - .executor_read_snapshot - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) = - Some(ExecutorReadSnapshot { - authority, - package, - access, - result, + }; + if start > result.contents.len() || !result.contents.is_char_boundary(start) { + outcome = "invalid_cursor"; + return Err(FunctionCallError::RespondToModel( + "skills.read cursor is invalid".to_string(), + )); + } + // Shadow candidates use the merged identity, not the selected read route. + if skill_entry.authority.kind == SkillSourceKind::Cloud + && let Some(state) = self + .context + .thread_state + .shadow_selection_turn(&call.turn_id) + { + state.record_invocation(skill_entry.main_prompt.as_str()); + } + let metadata = if self.context.step_state.discovery_v2 { + let mut installations = Vec::new(); + let mut installations_truncated = false; + let mut installation_bytes = 2; // JSON array brackets. + let installation_budget = + MAX_INSTALLATION_METADATA_BYTES.min(response_byte_budget / 4); + // Prefer the installation used by this read before other sources fill the budget. + for source in std::iter::once(selected_source.clone()) + .chain(skill_entry.source_locations()) + { + let Some(installation) = InstallationLocation::from_source(&source) else { + continue; + }; + if installations.contains(&installation) { + continue; + } + let size = + serialized_len(&installation)? + usize::from(!installations.is_empty()); + if installation_bytes + size > installation_budget { + installations_truncated = true; + continue; + } + installation_bytes += size; + installations.push(installation); + } + installations_truncated |= matches.iter().any(|(entry, _)| { + entry + .source_locations() + .filter_map(|source| InstallationLocation::from_source(&source)) + .any(|location| !installations.contains(&location)) }); - } + ReadMetadata::Installations(InstallationReadMetadata { + installations, + installations_truncated, + }) + } else { + let skill_root = if output_authority == SkillToolAuthoritySelector::Executor { + main_prompt + .environment_path() + .and_then(|(_, path)| path.parent()) + .map(|path| path.inferred_native_path_string()) + } else { + None + }; + ReadMetadata::Legacy(LegacyReadMetadata { skill_root }) + }; + let response = page_response( + result.resource.as_str(), + &result.contents, + &metadata, + start, + response_byte_budget, + |offset| { + if self.context.step_state.discovery_v2 { + pagination_cursor( + &(&source, &package, result.contents.as_str()), + offset, + ) + } else { + pagination_cursor(result.contents.as_str(), offset) + } + }, + ) + .inspect_err(|error| { + outcome = match error { + FunctionCallError::RespondToModel(_) => "response_budget", + FunctionCallError::Fatal(_) => "internal_error", + }; + })?; + let response_authority = if matches!( + &response.metadata, + ReadMetadata::Installations(metadata) + if metadata.installations.iter().any(|location| { + matches!(location, InstallationLocation::Cloud { .. }) + }) + ) { + SkillToolAuthoritySelector::Cloud + } else { + output_authority + }; + let output = skill_json_output(&response, response_authority)?; + if output_authority == super::SkillToolAuthoritySelector::Executor + && response.next_cursor.is_some() + && result.contents.len() <= MAX_SKILL_RESOURCE_CONTENT_BYTES + && let Some(access) = access_key + { + *self + .context + .thread_state + .executor_read_snapshot + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = + Some(ExecutorReadSnapshot { + authority, + package, + access, + result, + }); + } - if requested_resource == main_prompt - && args.cursor.is_none() - && let Some(analytics) = self.context.analytics.as_ref() - { - analytics.track_skill_invocation( - skill_entry, - &selected_source, - call.model.clone(), - call.turn_id.clone(), - InvocationType::Implicit, + if requested_resource == main_prompt + && args.cursor.is_none() + && let Some(analytics) = self.context.analytics.as_ref() + { + analytics.track_skill_invocation( + skill_entry, + &selected_source, + call.model.clone(), + call.turn_id.clone(), + InvocationType::Implicit, + ); + } + + outcome = "success"; + Ok(output) + } + .await; + if let Some(metrics) = &self.metrics { + metrics.counter( + "codex.skills.read.result", + /*inc*/ 1, + &[ + ("outcome", outcome), + ( + "discovery_version", + if self.context.step_state.discovery_v2 { + "v2" + } else { + "v1" + }, + ), + ], ); } - - Ok(output) + result }) } } diff --git a/codex-rs/ext/skills/tests/skills_extension.rs b/codex-rs/ext/skills/tests/skills_extension.rs index 24d4b9b94f..7c7dbbd9e4 100644 --- a/codex-rs/ext/skills/tests/skills_extension.rs +++ b/codex-rs/ext/skills/tests/skills_extension.rs @@ -3550,9 +3550,24 @@ struct RecordedHistogram { #[derive(Default)] struct RecordingMetrics { samples: Mutex>, + counters: Mutex>, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct RecordedCounter { + name: String, + increment: i64, + tags: Vec<(String, String)>, } impl RecordingMetrics { + fn counters(&self) -> Vec { + self.counters + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } + fn samples(&self) -> Vec { self.samples .lock() @@ -3572,8 +3587,18 @@ impl ExtensionMetrics for RecordingMetrics { self.histogram(name, value, tags); } - fn counter(&self, name: &str, _inc: i64, _tags: &[(&str, &str)]) { - panic!("unexpected counter: {name}"); + fn counter(&self, name: &str, inc: i64, tags: &[(&str, &str)]) { + self.counters + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(RecordedCounter { + name: name.to_string(), + increment: inc, + tags: tags + .iter() + .map(|(key, value)| ((*key).to_string(), (*value).to_string())) + .collect(), + }); } fn histogram(&self, name: &str, value: i64, tags: &[(&str, &str)]) { diff --git a/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs b/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs index 1997379e3b..cb3d62b906 100644 --- a/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs +++ b/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs @@ -16,6 +16,9 @@ use serde_json::json; #[path = "source_telemetry_tests.rs"] mod source_telemetry_tests; +#[path = "tool_metrics_tests.rs"] +mod tool_metrics_tests; + fn cloud_skill(name: &str) -> SkillCatalogEntry { let package = format!("skill://Plugin_123/{name}"); SkillCatalogEntry::new( @@ -145,6 +148,7 @@ impl MergeHarness { &self, step: &ExtensionData, ) -> Vec ToolExecutor>>> { + step.get_or_init(|| self.catalog_metrics.clone() as Arc); if let Some(snapshots) = &self.snapshots { step.insert(snapshots.clone()); } diff --git a/codex-rs/ext/skills/tests/skills_extension/tool_metrics_tests.rs b/codex-rs/ext/skills/tests/skills_extension/tool_metrics_tests.rs new file mode 100644 index 0000000000..2cd8b4b41b --- /dev/null +++ b/codex-rs/ext/skills/tests/skills_extension/tool_metrics_tests.rs @@ -0,0 +1,195 @@ +//! Checks bounded outcome metrics for skill reads and installation listings. + +use super::*; +use pretty_assertions::assert_eq; + +fn read_result(outcome: &str, version: &str) -> RecordedCounter { + RecordedCounter { + name: "codex.skills.read.result".to_string(), + increment: 1, + tags: vec![ + ("outcome".to_string(), outcome.to_string()), + ("discovery_version".to_string(), version.to_string()), + ], + } +} + +fn installations_result(outcome: &str, source_filter: &str) -> RecordedCounter { + RecordedCounter { + name: "codex.skills.installations.result".to_string(), + increment: 1, + tags: vec![ + ("outcome".to_string(), outcome.to_string()), + ("source_filter".to_string(), source_filter.to_string()), + ("discovery_version".to_string(), "v2".to_string()), + ], + } +} + +#[tokio::test] +async fn reads_report_outcomes_for_each_discovery_version() -> TestResult { + for discovery_version in ["v1", "v2"] { + let mut executor = installed_skill("executor", "local")? + .with_remote_plugin_id(/*remote_plugin_id*/ None) + .with_plugin_id(/*plugin_id*/ None); + executor.main_prompt = SkillResourceId::environment( + executor.main_prompt.as_str(), + "missing-environment", + PathUri::parse("file:///skills/local/SKILL.md")?, + ); + let mut harness = MergeHarness::new( + provider(vec![executor]), + provider(vec![ + cloud_skill("work").with_alias_root("skill://Plugin_123"), + ]), + default_config(), + ) + .await?; + if discovery_version == "v1" { + harness.snapshots = None; + } + harness.turn_store.insert(selected_plugins(&[None], &[])); + let tools = harness.tools(&harness.step_store).await; + let mut expected = Vec::new(); + for (arguments, budget, outcome) in [ + ( + json!({"package": "skill://Plugin_123/work"}), + 10_000, + "success", + ), + (json!({}), 10_000, "invalid_arguments"), + (json!({"package": "missing"}), 10_000, "package_unavailable"), + (json!({"package": "c99/work"}), 10_000, "unresolved_alias"), + ( + json!({"package": "skill://executor/local"}), + 10_000, + "environment_unavailable", + ), + ( + json!({"package": "skill://Plugin_123/work", "source": {"kind": "executor", "environment_id": "missing"}}), + 10_000, + "source_unavailable", + ), + ( + json!({"package": "skill://Plugin_123/work", "cursor": "invalid"}), + 10_000, + "invalid_cursor", + ), + ( + json!({"package": "skill://Plugin_123/work"}), + 1, + "response_budget", + ), + ( + json!({"package": "skill://Plugin_123/work"}), + 10_000, + "read_failed", + ), + ] { + if discovery_version == "v1" && outcome != "success" { + continue; + } + if outcome == "read_failed" { + let config = default_config(); + harness.registry.config_contributors()[0].on_config_changed( + &harness.session_store, + &harness.thread_store, + &config, + &TestConfig { + cloud_skill_enabled: false, + ..config.clone() + }, + ); + } + let result = + call_tool_with_budget(&tools, "read", arguments, TruncationPolicy::Bytes(budget)) + .await; + assert_eq!( + result.is_ok(), + outcome == "success", + "{outcome}: {result:?}" + ); + expected.push(read_result(outcome, discovery_version)); + assert_eq!(harness.catalog_metrics.counters(), expected); + } + } + Ok(()) +} + +#[tokio::test] +async fn installations_report_outcomes_for_each_discovery_version() -> TestResult { + let harness = MergeHarness::new( + provider(Vec::new()), + provider(vec![ + cloud_skill("work").with_alias_root("skill://Plugin_123"), + ]), + default_config(), + ) + .await?; + harness.turn_store.insert(selected_plugins(&[None], &[])); + let tools = harness.tools(&harness.step_store).await; + let mut expected = Vec::new(); + for (arguments, budget, outcome, source_filter) in [ + ( + json!({"package": "skill://Plugin_123/work"}), + 10_000, + "success", + "all", + ), + ( + json!({"package": "c0/work", "source": {"kind": "cloud"}}), + 10_000, + "success", + "cloud", + ), + ( + json!({"package": "c0/work", "source": {"kind": "executor", "environment_id": "missing"}}), + 10_000, + "empty", + "executor", + ), + (json!({}), 10_000, "invalid_arguments", "unparsed"), + (json!({"package": ""}), 10_000, "invalid_package", "all"), + ( + json!({"package": "c0/work", "source": {"kind": "executor", "environment_id": ""}}), + 10_000, + "invalid_source", + "executor", + ), + ( + json!({"package": "c0/missing"}), + 10_000, + "package_unavailable", + "all", + ), + ( + json!({"package": "c99/work"}), + 10_000, + "unresolved_alias", + "all", + ), + ( + json!({"package": "c0/work", "cursor": "invalid"}), + 10_000, + "invalid_cursor", + "all", + ), + (json!({"package": "c0/work"}), 1, "response_budget", "all"), + ] { + let result = call_tool_with_budget( + &tools, + "installations", + arguments, + TruncationPolicy::Bytes(budget), + ) + .await; + assert_eq!( + result.is_ok(), + matches!(outcome, "success" | "empty"), + "{outcome}: {result:?}" + ); + expected.push(installations_result(outcome, source_filter)); + assert_eq!(harness.catalog_metrics.counters(), expected); + } + Ok(()) +}