openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-05 20:10:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add outcome metrics for skill reads and installation listings (#59295)

Browse Source 
## 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
This commit is contained in:
Anthony Tafoya
2026-11-04 02:29:41 +00:00
committed by copyberry
parent 007f668c34
commit 75aba6768a
8 changed files with 732 additions and 367 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/session/mod.rs
+3
View File
@@ -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)
codex-rs/core/src/session/tests.rs
+38 -24
View File
@@ -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<codex_extension_api::WorldStateSectionContribution>,
> {
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::<Arc<dyn codex_extension_api::ExtensionMetrics>>()
.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::<crate::config::Config>::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");
codex-rs/ext/skills/src/tools/installations.rs
+121 -68
View File
@@ -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<Arc<dyn ExtensionMetrics>>,
}
impl<'call> ToolExecutor<ToolCall<'call>> for InstallationsTool {
@@ -72,77 +75,127 @@ impl<'call> ToolExecutor<ToolCall<'call>> 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
})
}
}
codex-rs/ext/skills/src/tools/mod.rs
+8 -1
View File
@@ -73,6 +73,9 @@ pub(crate) fn skill_tools(
.get::<SkillsSessionState>()
.and_then(|state| state.mcp_resources.clone());
let analytics = SkillAnalytics::from_stores(session_store, thread_store);
let metrics = step_store
.get::<Arc<dyn ExtensionMetrics>>()
.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
}
codex-rs/ext/skills/src/tools/read.rs
+336 -272
View File
@@ -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<Arc<dyn ExtensionMetrics>>,
}
impl<'call> ToolExecutor<ToolCall<'call>> for ReadTool {
@@ -116,289 +118,351 @@ impl<'call> ToolExecutor<ToolCall<'call>> 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::<ReadArgs>(&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::<ReadArgs>(&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
})
}
}
codex-rs/ext/skills/tests/skills_extension.rs
+27 -2
View File
@@ -3550,9 +3550,24 @@ struct RecordedHistogram {
#[derive(Default)]
struct RecordingMetrics {
samples: Mutex<Vec<RecordedHistogram>>,
counters: Mutex<Vec<RecordedCounter>>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
struct RecordedCounter {
name: String,
increment: i64,
tags: Vec<(String, String)>,
}
impl RecordingMetrics {
fn counters(&self) -> Vec<RecordedCounter> {
self.counters
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone()
}
fn samples(&self) -> Vec<RecordedHistogram> {
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)]) {
codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs
+4
View File
@@ -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<Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>>> {
step.get_or_init(|| self.catalog_metrics.clone() as Arc<dyn ExtensionMetrics>);
if let Some(snapshots) = &self.snapshots {
step.insert(snapshots.clone());
}
codex-rs/ext/skills/tests/skills_extension/tool_metrics_tests.rs
+195
View File
@@ -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(())
}