Record active plugin inventory in turn analytics (#46323)

## What changed

Add `active_plugin_ids_at_turn_start` to turn analytics, combining active host plugins with selected plugin packages. Prefer remote plugin IDs when present and otherwise use validated package IDs. Sort and deduplicate the inventory.

Report `null` for unknown inventories, invalid IDs, IDs longer than 128 bytes, or inventories exceeding 512 distinct IDs; preserve `[]` for an observed empty inventory. Keep the first received inventory even when later resolved configuration updates arrive.

## Testing

Add coverage for ID selection and validation, deduplication, size limits, serialization, and preservation of the first inventory. Extend app-server tests to check inventories after plugin reconciliation and with selected plugins, including turns without explicit plugin mentions.

GitOrigin-RevId: 8a0de912e2570a902cd05bfc918dc8e6ff8b8527
This commit is contained in:
marksteinbrick-oai
2026-09-17 23:47:06 +00:00
committed by copyberry
parent 1f631def3f
commit a1efb59c4a
12 changed files with 368 additions and 15 deletions
@@ -489,6 +489,7 @@ fn sample_turn_resolved_config(thread_id: &str, turn_id: &str) -> TurnResolvedCo
turn_id: turn_id.to_string(),
thread_id: thread_id.to_string(),
turn_metadata: test_turn_metadata(/*root_turn_id*/ None),
active_plugin_ids_at_turn_start: None,
num_input_images: 1,
submission_type: None,
ephemeral: false,
@@ -5290,6 +5291,10 @@ fn turn_event_serializes_expected_shape() {
thread_id: "thread-2".to_string(),
session_id: "session-thread-2".to_string(),
turn_id: "turn-2".to_string(),
active_plugin_ids_at_turn_start: Some(vec![
"plugins~Plugin_example".to_string(),
"test@marketplace".to_string(),
]),
voice_session_id: None,
root_turn_id: Some("turn-2".to_string()),
turn_trigger: Some("user".to_string()),
@@ -5368,6 +5373,7 @@ fn turn_event_serializes_expected_shape() {
"thread_id": "thread-2",
"session_id": "session-thread-2",
"turn_id": "turn-2",
"active_plugin_ids_at_turn_start": ["plugins~Plugin_example", "test@marketplace"],
"voice_session_id": null,
"root_turn_id": "turn-2",
"turn_trigger": "user",
@@ -5688,6 +5694,62 @@ async fn turn_start_error_response_discards_pending_start_request() {
assert!(out.is_empty());
}
#[tokio::test]
async fn turn_event_preserves_first_received_plugin_inventory() {
for plugin_ids in [
None,
Some(vec![]),
Some(vec!["initial@marketplace".to_string()]),
] {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ false,
/*include_started*/ true,
/*include_token_usage*/ false,
)
.await;
let config = TurnResolvedConfigFact {
active_plugin_ids_at_turn_start: plugin_ids.clone(),
..sample_turn_resolved_config("thread-2", "turn-2")
};
let later_config = TurnResolvedConfigFact {
active_plugin_ids_at_turn_start: Some(vec!["later@marketplace".to_string()]),
is_first_turn: false,
..config.clone()
};
for fact in [
AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(config))),
AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
later_config,
))),
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
] {
reducer.ingest(fact, &mut out).await;
}
let [TrackEventRequest::TurnEvent(event)] = out.as_slice() else {
panic!("expected one turn event");
};
assert_eq!(
(
&event.event_params.active_plugin_ids_at_turn_start,
event.event_params.is_first_turn,
),
(&plugin_ids, false),
);
}
}
#[tokio::test]
async fn turn_lifecycle_emits_turn_event() {
let mut reducer = AnalyticsReducer::default();
+2
View File
@@ -1068,6 +1068,8 @@ pub(crate) struct CodexTurnEventParams {
pub(crate) thread_id: String,
pub(crate) session_id: String,
pub(crate) turn_id: String,
/// First received active plugin inventory. Null is unknown; [] is observed empty.
pub(crate) active_plugin_ids_at_turn_start: Option<Vec<String>>,
pub(crate) voice_session_id: Option<String>,
pub(crate) root_turn_id: Option<String>,
pub(crate) turn_trigger: Option<String>,
+2
View File
@@ -187,6 +187,8 @@ pub struct TurnResolvedConfigFact {
pub turn_id: String,
pub thread_id: String,
pub turn_metadata: Arc<dyn TurnAnalyticsMetadata>,
/// Observed active plugin inventory. None is unknown; Some([]) is observed empty.
pub active_plugin_ids_at_turn_start: Option<Vec<String>>,
pub num_input_images: usize,
pub submission_type: Option<TurnSubmissionType>,
pub ephemeral: bool,
+9 -1
View File
@@ -1270,7 +1270,7 @@ impl AnalyticsReducer {
async fn ingest_turn_resolved_config(
&mut self,
input: TurnResolvedConfigFact,
mut input: TurnResolvedConfigFact,
out: &mut Vec<TrackEventRequest>,
) {
let turn_id = input.turn_id.clone();
@@ -1279,6 +1279,12 @@ impl AnalyticsReducer {
let turn_state = self.turns.entry(turn_id.clone()).or_default();
turn_state.thread_id = Some(thread_id);
turn_state.num_input_images = Some(num_input_images);
// Keep the first received plugin inventory, including unknown or empty,
// while the remaining resolved config continues updating.
if let Some(initial_config) = &turn_state.resolved_config {
input.active_plugin_ids_at_turn_start =
initial_config.active_plugin_ids_at_turn_start.clone();
}
turn_state.resolved_config = Some(input);
self.maybe_emit_turn_event(&turn_id, out).await;
}
@@ -3622,6 +3628,7 @@ fn codex_turn_event_params(
turn_id: _resolved_turn_id,
thread_id: _resolved_thread_id,
turn_metadata,
active_plugin_ids_at_turn_start,
num_input_images: _resolved_num_input_images,
submission_type,
ephemeral,
@@ -3658,6 +3665,7 @@ fn codex_turn_event_params(
thread_id,
session_id: thread_metadata.session_id.clone(),
turn_id,
active_plugin_ids_at_turn_start,
voice_session_id: turn_state.voice_session_id.clone(),
root_turn_id: turn_metadata.root_turn_id(),
turn_trigger: turn_metadata.turn_trigger(),
@@ -608,7 +608,10 @@ pub(crate) async fn wait_for_matching_analytics_event(
};
for request in &requests {
if request.method != "POST"
|| request.url.path() != "/codex/analytics-events/events"
|| !request
.url
.path()
.ends_with("/codex/analytics-events/events")
{
continue;
}
@@ -31,7 +31,10 @@ use wiremock::matchers::method;
use wiremock::matchers::path;
use wiremock::matchers::query_param;
use super::analytics::wait_for_matching_analytics_event;
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
const REMOTE_PLUGIN_ID: &str = "plugins~Plugin_00000000000000000000000000000000";
#[test_case("WORKSPACE", "workspace-directory"; "workspace")]
#[test_case("GLOBAL", "openai-curated-remote"; "global")]
@@ -139,6 +142,11 @@ async fn plugin_reconcile_syncs_bundles_and_reports_changes(
failed_remote_plugin_ids: failures.clone(),
failed_materialization_remote_plugin_ids: failures,
};
let active_plugin_ids = if enabled {
vec![REMOTE_PLUGIN_ID.to_string()]
} else {
Vec::new()
};
mount_installed_snapshot(&server, plugins).await;
Mock::given(method("GET"))
.and(path("/bundle"))
@@ -150,7 +158,13 @@ async fn plugin_reconcile_syncs_bundles_and_reports_changes(
// A turn exercises the loaded registry, not fresh hooks/list discovery. Hooks run
// on the app-server host even when the thread uses a remote executor.
assert_eq!(
turn_hook_runs(&mut app_server, &server, thread.id.clone()).await?,
turn_hook_runs(
&mut app_server,
&server,
thread.id.clone(),
active_plugin_ids
)
.await?,
usize::from(scope == "WORKSPACE" && enabled && capabilities.has_hooks),
);
server.verify().await;
@@ -167,7 +181,7 @@ async fn plugin_reconcile_syncs_bundles_and_reports_changes(
}
);
assert_eq!(
turn_hook_runs(&mut app_server, &server, thread.id.clone()).await?,
turn_hook_runs(&mut app_server, &server, thread.id.clone(), Vec::new()).await?,
0
);
server.verify().await;
@@ -196,7 +210,13 @@ async fn turn_hook_runs(
app_server: &mut TestAppServer,
server: &MockServer,
thread_id: String,
active_plugin_ids: Vec<String>,
) -> Result<usize> {
Mock::given(method("POST"))
.and(path("/backend-api/codex/analytics-events/events"))
.respond_with(ResponseTemplate::new(200))
.mount(server)
.await;
let _response = responses::mount_sse_once(
server,
responses::sse(vec![
@@ -209,7 +229,7 @@ async fn turn_hook_runs(
let completed = timeout(
DEFAULT_TIMEOUT,
app_server.start_turn_and_wait_for_completion(TurnStartParams {
thread_id,
thread_id: thread_id.clone(),
input: vec![UserInput::Text {
text: "run hooks".to_string(),
text_elements: Vec::new(),
@@ -219,6 +239,16 @@ async fn turn_hook_runs(
)
.await??;
assert_eq!(completed.turn.status, TurnStatus::Completed);
let event = wait_for_matching_analytics_event(server, DEFAULT_TIMEOUT, |event| {
event["event_type"] == "codex_turn_event"
&& event["event_params"]["turn_id"] == completed.turn.id
})
.await?;
assert_eq!(event["event_params"]["thread_id"], thread_id);
assert_eq!(
event["event_params"]["active_plugin_ids_at_turn_start"],
json!(active_plugin_ids)
);
Ok(app_server
.pending_notification_methods()
.iter()
@@ -284,7 +314,7 @@ async fn mount_installed_snapshot(server: &MockServer, plugins: Vec<Value>) {
fn installed_plugin(version: &str, bundle_url: &str, scope: &str) -> Value {
json!({
"id": "plugins~Plugin_00000000000000000000000000000000",
"id": REMOTE_PLUGIN_ID,
"name": "linear",
"scope": scope,
"discoverability": (scope == "WORKSPACE").then_some("LISTED"),
@@ -52,6 +52,8 @@ use tokio::process::Child;
use tokio::process::Command;
use tokio::time::timeout;
use super::analytics::mount_analytics_capture;
use super::analytics::wait_for_matching_analytics_event;
use super::app_list::connector_tool;
use super::app_list::start_apps_server_with_delays;
@@ -90,10 +92,8 @@ async fn selected_plugin_mcp_startup_respects_explicit_mentions(
) -> Result<()> {
let explicitly_mentioned = !matches!(mention, PluginMention::Unmentioned);
let responses_server = responses::start_mock_server().await;
let (apps_url, apps_server_handle) =
start_apps_server_with_delays(Vec::new(), Vec::new(), Duration::ZERO, Duration::ZERO)
.await?;
let fixture = selected_capability_fixture(&responses_server.uri(), &apps_url)?;
let fixture = selected_capability_fixture(&responses_server.uri(), &responses_server.uri())?;
mount_analytics_capture(&responses_server, fixture.codex_home.path()).await?;
let config_path = fixture.codex_home.path().join("config.toml");
let config = std::fs::read_to_string(&config_path)?.replace(
"executor_capability_discovery = true",
@@ -139,7 +139,7 @@ async fn selected_plugin_mcp_startup_respects_explicit_mentions(
};
let request_id = app_server
.send_turn_start_request(TurnStartParams {
thread_id,
thread_id: thread_id.clone(),
input: vec![input],
environments: Some(vec![TurnEnvironmentParams {
environment_id: LOCAL_ENVIRONMENT_ID.to_string(),
@@ -149,11 +149,12 @@ async fn selected_plugin_mcp_startup_respects_explicit_mentions(
..Default::default()
})
.await?;
timeout(
let response = timeout(
READ_TIMEOUT,
app_server.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let TurnStartResponse { turn } = to_response(response)?;
wait_for_pid_file(&fixture.pid_file).await?;
if explicitly_mentioned {
// An explicit mention must outwait the optional one-second grace.
@@ -180,9 +181,18 @@ async fn selected_plugin_mcp_startup_respects_explicit_mentions(
explicitly_mentioned,
);
let event = wait_for_matching_analytics_event(&responses_server, READ_TIMEOUT, |event| {
event["event_type"] == "codex_turn_event"
&& event["event_params"]["thread_id"] == thread_id
&& event["event_params"]["turn_id"] == turn.id
})
.await?;
assert_eq!(
event["event_params"]["active_plugin_ids_at_turn_start"],
json!([PLUGIN_ID])
);
exec_server.kill().await?;
apps_server_handle.abort();
let _ = apps_server_handle.await;
Ok(())
}
@@ -0,0 +1,174 @@
//! Tests source-aware plugin telemetry selection and complete-inventory bounds.
use crate::session::tests::make_session_and_context;
use codex_extension_api::SelectedPluginIdentity;
use codex_extension_api::SelectedPluginSnapshot;
use codex_utils_plugins::PluginIdentity;
use pretty_assertions::assert_eq;
#[tokio::test]
async fn active_plugin_ids_select_and_validate_by_source() {
let (_session, mut turn_context) = make_session_and_context().await;
for (name, identities, selected_ids, expected) in [
(
"remote wins even when the package fallback is invalid",
vec![
("box@local", Some("plugins~Plugin_box")),
("selected-root", Some("plugin_box-1")),
],
vec![],
Some(vec!["plugin_box-1", "plugins~Plugin_box"]),
),
(
"package fallback",
vec![("my.tool@123", None)],
vec!["executor-demo@1"],
Some(vec!["executor-demo@1", "my.tool@123"]),
),
(
"blank present remote",
vec![("box@local", Some(""))],
vec![],
None,
),
(
"remote is not trimmed",
vec![("box@local", Some(" plugin_box "))],
vec![],
None,
),
(
"package key is not a remote ID",
vec![("box@local", Some("box@local"))],
vec![],
None,
),
(
"remote-looking selected root is still a fallback",
vec![],
vec!["plugins~Plugin_box"],
None,
),
(
"one invalid identity makes the inventory unknown",
vec![("box@local", Some("plugin_box")), ("bad..name@1", None)],
vec![],
None,
),
(
"exact deduplication retains aliases and distinct remote objects",
vec![
("box@local", Some("plugins~Plugin_b")),
("box@local", None),
("box@local", Some("plugins~Plugin_a")),
("box@other", Some("plugins~Plugin_a")),
],
vec!["box@local", "box@local"],
Some(vec!["box@local", "plugins~Plugin_a", "plugins~Plugin_b"]),
),
("empty observation", vec![], vec![], Some(vec![])),
] {
let identities = identities
.into_iter()
.map(|(plugin_id, remote_plugin_id)| PluginIdentity {
plugin_id: plugin_id.to_string(),
remote_plugin_id: remote_plugin_id.map(str::to_string),
})
.collect();
let selected_plugins = selected_ids
.into_iter()
.map(|plugin_id| SelectedPluginIdentity {
selected_root_id: "capability-root".to_string(),
plugin_id: plugin_id.to_string(),
})
.collect::<Vec<_>>();
turn_context.active_host_plugin_identities = Some(identities);
turn_context.extension_data.insert(SelectedPluginSnapshot {
plugins: selected_plugins,
..Default::default()
});
assert_eq!(
turn_context.active_plugin_ids_for_telemetry(),
expected.map(|ids| ids.into_iter().map(str::to_string).collect::<Vec<_>>()),
"{name}"
);
}
}
#[tokio::test]
async fn active_plugin_ids_enforce_bounds_after_selection_and_deduplication() {
let (_session, mut turn_context) = make_session_and_context().await;
for length in [128, 129] {
for remote in [false, true] {
let id = if remote {
"r".repeat(length)
} else {
format!("{}@local", "p".repeat(length - 6))
};
// A remote ID also wins over an overlong package key.
let plugin_id = if remote {
format!("{}@local", "p".repeat(/*n*/ 128))
} else {
id.clone()
};
let identities = vec![PluginIdentity {
plugin_id,
remote_plugin_id: remote.then_some(id.clone()),
}];
turn_context.active_host_plugin_identities = Some(identities);
turn_context
.extension_data
.insert(SelectedPluginSnapshot::default());
assert_eq!(
turn_context.active_plugin_ids_for_telemetry(),
(length == 128).then_some(vec![id]),
"length={length}, remote={remote}"
);
}
let id = format!("{}@local", "p".repeat(length - 6));
let selected_plugin = SelectedPluginIdentity {
selected_root_id: "capability-root".to_string(),
plugin_id: id.clone(),
};
turn_context.active_host_plugin_identities = Some(Vec::new());
turn_context.extension_data.insert(SelectedPluginSnapshot {
plugins: vec![selected_plugin],
..Default::default()
});
assert_eq!(
turn_context.active_plugin_ids_for_telemetry(),
(length == 128).then_some(vec![id]),
"selected length={length}"
);
}
for count in [512, 513] {
let ids = (0..count)
.map(|index| format!("plugin_{index:03}@local"))
.collect::<Vec<_>>();
let identities = ids[..count / 2]
.iter()
.map(|id| PluginIdentity {
plugin_id: id.clone(),
remote_plugin_id: None,
})
.collect();
let selected_plugins = ids[count / 2 - 1..]
.iter()
.map(|id| SelectedPluginIdentity {
selected_root_id: "capability-root".to_string(),
plugin_id: id.clone(),
})
.collect::<Vec<_>>();
turn_context.active_host_plugin_identities = Some(identities);
turn_context.extension_data.insert(SelectedPluginSnapshot {
plugins: selected_plugins,
..Default::default()
});
assert_eq!(
turn_context.active_plugin_ids_for_telemetry(),
(count == 512).then_some(ids),
"{count} distinct IDs across host and selected inputs, with one overlap"
);
}
}
+1
View File
@@ -3743,6 +3743,7 @@ impl Session {
.await
}
/// Captures and retains a step with the required MCP startup dependencies.
pub(crate) async fn capture_step_context_with_required_mcp_servers(
self: &Arc<Self>,
turn_context: Arc<TurnContext>,
+1
View File
@@ -160,6 +160,7 @@ pub(super) async fn spawn_review_thread(
auth_manager: auth_manager_for_context,
initial_settings: Arc::clone(&step_settings),
disabled_plugin_ids: parent_turn_context.disabled_plugin_ids.clone(),
active_host_plugin_identities: None,
current_settings: ArcSwap::from(step_settings),
session_telemetry: session_telemetry_for_context,
provider: provider_for_context,
+1 -1
View File
@@ -1194,6 +1194,7 @@ async fn track_turn_resolved_config_analytics(
turn_id: turn_context.sub_id.clone(),
thread_id: sess.thread_id.to_string(),
turn_metadata: turn_context.turn_metadata_state.clone(),
active_plugin_ids_at_turn_start: turn_context.active_plugin_ids_for_telemetry(),
num_input_images: input
.iter()
.filter_map(|item| match item {
@@ -1675,7 +1676,6 @@ pub(crate) async fn prepare_tool_recommendations(
.services
.plugins_manager
.plugins_for_config(&turn_context.config.plugins_config_input())
.instrument(trace_span!("built_tools.load_plugins"))
.await
.without_plugins(&turn_context.disabled_plugin_ids);
let tool_suggest_is_enabled = tool_suggest_enabled(turn_context);
+60
View File
@@ -16,6 +16,7 @@ use codex_core_plugins::PluginCommandAttribution;
use codex_core_plugins::ResolvedPluginMetricsOperation;
use codex_core_plugins::TrustedPluginRoots;
use codex_exec_server::ExecutorFileSystem;
use codex_extension_api::SelectedPluginSnapshot;
use codex_file_system::FileSystemSandboxContext;
use codex_model_provider::SharedModelProvider;
use codex_protocol::SessionId;
@@ -37,6 +38,7 @@ use codex_sandboxing::policy_transforms::effective_permission_profile;
use codex_skills_extension::HostSkillsSnapshot;
use codex_skills_extension::SkillLoadOutcome;
use codex_utils_path_uri::PathUri;
use codex_utils_plugins::PluginIdentity;
use futures::FutureExt;
use futures::future::BoxFuture;
use futures::future::Shared;
@@ -318,6 +320,7 @@ pub struct TurnContext {
pub(crate) initial_settings: Arc<ResolvedStepSettings>,
/// Thread-owned plugin selection captured when this turn was admitted.
pub(crate) disabled_plugin_ids: Vec<String>,
pub(super) active_host_plugin_identities: Option<Vec<PluginIdentity>>,
/// Snapshot for the next step; request consumers use their captured StepContext.
pub(super) current_settings: ArcSwap<ResolvedStepSettings>,
/// Turn-wide telemetry; model-attributed step work should use `StepContext::session_telemetry`.
@@ -410,6 +413,46 @@ impl TurnContext {
self.initial_settings.effective_collaboration_mode()
}
/// Combines setup-time host identities with current ready selected packages.
/// Keeps the complete observation or marks it unknown; never truncates membership.
pub(super) fn active_plugin_ids_for_telemetry(&self) -> Option<Vec<String>> {
const MAX_TELEMETRY_PLUGIN_IDS: usize = 512;
const MAX_TELEMETRY_PLUGIN_ID_BYTES: usize = 128;
let mut identities = self.active_host_plugin_identities.clone()?;
// Selected roots provide a package key, not a remote identity for that object.
if let Some(selected) = self.extension_data.get::<SelectedPluginSnapshot>() {
identities.extend(selected.plugins.iter().map(|plugin| PluginIdentity {
plugin_id: plugin.plugin_id.clone(),
remote_plugin_id: None,
}));
}
let mut ids = Vec::with_capacity(identities.len());
for identity in identities {
let id = match identity.remote_plugin_id {
Some(remote_id) => {
if !codex_core_plugins::remote::is_valid_remote_plugin_id(&remote_id) {
return None;
}
remote_id
}
None => {
if codex_plugin::PluginId::parse(&identity.plugin_id).is_err() {
return None;
}
identity.plugin_id
}
};
if id.len() > MAX_TELEMETRY_PLUGIN_ID_BYTES {
return None;
}
ids.push(id);
}
ids.sort_unstable();
ids.dedup();
(ids.len() <= MAX_TELEMETRY_PLUGIN_IDS).then_some(ids)
}
pub(crate) fn plugin_attribution_for_command(
&self,
command: &[String],
@@ -618,6 +661,7 @@ impl TurnContext {
auth_manager: self.auth_manager.clone(),
initial_settings: Arc::clone(&step_settings),
disabled_plugin_ids: self.disabled_plugin_ids.clone(),
active_host_plugin_identities: self.active_host_plugin_identities.clone(),
current_settings: ArcSwap::from(step_settings),
session_telemetry,
provider: self.provider.clone(),
@@ -933,6 +977,7 @@ impl Session {
auth_manager,
initial_settings: Arc::clone(&step_settings),
disabled_plugin_ids: session_configuration.disabled_plugin_ids.clone(),
active_host_plugin_identities: None,
current_settings: ArcSwap::from(step_settings),
session_telemetry: session_telemetry_for_context,
provider,
@@ -1174,6 +1219,17 @@ impl Session {
);
turn_context.code_mode_available = self.services.code_mode_service.is_available();
turn_context.extension_data.insert(trusted_plugin_roots);
turn_context.active_host_plugin_identities = Some(
plugin_outcome
.plugins()
.iter()
.filter(|plugin| plugin.is_active())
.map(|plugin| PluginIdentity {
plugin_id: plugin.config_name.clone(),
remote_plugin_id: plugin.remote_plugin_id.clone(),
})
.collect(),
);
turn_context.realtime_active = self.conversation.running_state().await.is_some();
turn_context.final_output_json_schema = options.final_output_json_schema;
@@ -1272,3 +1328,7 @@ impl Session {
state.session_configuration.clone()
}
}
#[cfg(test)]
#[path = "active_plugin_inventory_tests.rs"]
mod tests;