Record initialization analytics for resumed subagents (#51206)

## Why

Reloaded subagent threads did not emit initialization analytics, and subagent initialization events always reported `new`.

## What changed

Emit initialization events for resumed spawned subagents before notifying thread listeners. Inherit parent client metadata before reserving a residency slot can evict the parent, preserve the thread's original `created_at`, and report `initialization_mode` as `resumed`. Refresh stored analytics metadata when a subagent resumes; newly created subagents continue to report `new`.

## Testing

Extend the cold-resume integration test to verify a single worker initialization event with the expected session, parent, source, resume mode, and original creation timestamp. Update serialization and session tests to check initialization modes.

GitOrigin-RevId: e78b0c4d88b6ddbe993ca7e1fe7717b6f74bf3d9
This commit is contained in:
peilin-openai
2026-10-06 00:16:18 +00:00
committed by copyberry
parent 685270a56a
commit c19525e55e
11 changed files with 155 additions and 16 deletions
+1 -1
View File
@@ -1537,7 +1537,7 @@ pub(crate) fn subagent_thread_started_event_request(
ephemeral: input.ephemeral,
is_worktree: None,
thread_source: input.thread_source,
initialization_mode: ThreadInitializationMode::New,
initialization_mode: input.initialization_mode,
subagent_source: Some(subagent_source_name(&input.subagent_source)),
parent_thread_id: input.parent_thread_id,
forked_from_thread_id: input.forked_from_thread_id,
+1
View File
@@ -443,6 +443,7 @@ pub struct SubAgentThreadStartedInput {
pub ephemeral: bool,
pub thread_source: Option<ThreadSource>,
pub subagent_source: SubAgentSource,
pub initialization_mode: ThreadInitializationMode,
pub created_at: u64,
}
+6 -4
View File
@@ -1201,15 +1201,17 @@ impl AnalyticsReducer {
thread_state
.originator
.get_or_insert_with(|| input.product_client_id.clone());
thread_state
.metadata
.get_or_insert_with(|| ThreadMetadataState {
if thread_state.metadata.is_none()
|| matches!(input.initialization_mode, ThreadInitializationMode::Resumed)
{
thread_state.metadata = Some(ThreadMetadataState {
session_id: input.session_id.clone(),
thread_source: input.thread_source.clone(),
initialization_mode: ThreadInitializationMode::New,
initialization_mode: input.initialization_mode,
subagent_source: Some(subagent_source_name(&input.subagent_source)),
parent_thread_id,
});
}
if thread_state.connection_id.is_none() {
thread_state.connection_id = parent_connection_id;
}
@@ -28,6 +28,7 @@ use crate::facts::CompactionStrategy;
use crate::facts::CompactionTrigger;
use crate::facts::CustomAnalyticsFact;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::ThreadInitializationMode;
use crate::reducer::AnalyticsReducer;
use crate::tests::support::ingest_complete_child_turn;
use crate::tests::support::ingest_completed_command_execution_item;
@@ -857,6 +858,7 @@ async fn guardian_events_keep_thread_source_and_originator_with_explicit_turn_co
ephemeral: false,
thread_source: Some(ThreadSource::GuardianReview),
subagent_source: SubAgentSource::Other("guardian".to_string()),
initialization_mode: ThreadInitializationMode::New,
created_at: 130,
},
)),
@@ -555,6 +555,7 @@ fn subagent_thread_started_review_serializes_expected_shape() {
ephemeral: false,
thread_source: Some(ThreadSource::Subagent),
subagent_source: SubAgentSource::Review,
initialization_mode: ThreadInitializationMode::New,
created_at: 123,
},
));
@@ -614,6 +615,7 @@ fn subagent_thread_started_thread_spawn_serializes_thread_lineage() {
agent_nickname: None,
agent_role: None,
},
initialization_mode: ThreadInitializationMode::Resumed,
created_at: 124,
},
));
@@ -621,6 +623,8 @@ fn subagent_thread_started_thread_spawn_serializes_thread_lineage() {
let payload = serde_json::to_value(&event).expect("serialize thread spawn subagent event");
assert_eq!(payload["event_params"]["thread_id"], "thread-spawn");
assert_eq!(payload["event_params"]["thread_source"], "subagent");
assert_eq!(payload["event_params"]["initialization_mode"], "resumed");
assert_eq!(payload["event_params"]["created_at"], 124);
assert_eq!(payload["event_params"]["subagent_source"], "thread_spawn");
assert_eq!(
payload["event_params"]["parent_thread_id"],
@@ -648,6 +652,7 @@ fn subagent_thread_started_memory_consolidation_serializes_expected_shape() {
ephemeral: false,
thread_source: Some(ThreadSource::Subagent),
subagent_source: SubAgentSource::MemoryConsolidation,
initialization_mode: ThreadInitializationMode::New,
created_at: 125,
},
));
@@ -676,6 +681,7 @@ fn subagent_thread_started_other_serializes_expected_shape() {
ephemeral: false,
thread_source: Some(ThreadSource::GuardianReview),
subagent_source: SubAgentSource::Other("guardian".to_string()),
initialization_mode: ThreadInitializationMode::New,
created_at: 126,
},
));
@@ -704,6 +710,7 @@ fn subagent_thread_started_other_serializes_explicit_parent_thread_id() {
ephemeral: false,
thread_source: Some(ThreadSource::GuardianReview),
subagent_source: SubAgentSource::Other("guardian".to_string()),
initialization_mode: ThreadInitializationMode::New,
created_at: 126,
},
));
@@ -736,6 +743,7 @@ async fn subagent_thread_started_publishes_without_initialize() {
ephemeral: false,
thread_source: Some(ThreadSource::Subagent),
subagent_source: SubAgentSource::Review,
initialization_mode: ThreadInitializationMode::New,
created_at: 127,
},
)),
@@ -774,6 +782,7 @@ async fn subagent_tool_items_inherit_parent_connection_metadata() {
ephemeral: false,
thread_source: Some(ThreadSource::Subagent),
subagent_source: SubAgentSource::Review,
initialization_mode: ThreadInitializationMode::New,
created_at: 128,
},
)),
+34 -2
View File
@@ -588,8 +588,21 @@ impl LocalAgentControl {
..Default::default()
})
};
// Reserving a slot can evict an idle nested parent. Capture its instructions
// alongside its authority so the child does not depend on a later live lookup.
let subagent_analytics =
if let SessionSource::SubAgent(source @ SubAgentSource::ThreadSpawn { .. }) =
&session_source
&& let Some(parent_thread_id) = parent_thread_id
&& let Ok(parent) = state.get_thread(parent_thread_id).await
{
Some((
parent.session.app_server_client_metadata().await,
source.clone(),
))
} else {
None
};
// Reserving a slot can evict an idle nested parent. Capture its instructions and
// analytics metadata alongside its authority before the live parent can disappear.
let residency_slot = self
.reserve_v2_residency_slot(&state, &config, &membership, Some(thread_id))
.await?;
@@ -615,6 +628,24 @@ impl LocalAgentControl {
}
self.runtime.registry.clear_evicted_environments(thread_id);
residency_slot.commit(reloaded_thread.thread_id);
// Register before listeners can forward events from the resumed thread.
if let Some((client_metadata, source)) = subagent_analytics {
let thread_config = reloaded_thread.thread.config_snapshot().await;
emit_subagent_session_started(
&reloaded_thread
.thread
.session
.services
.analytics_events_client,
client_metadata,
reloaded_thread.thread.session.session_id(),
reloaded_thread.thread_id,
thread_config.parent_thread_id,
thread_config,
source,
/*resumed_created_at*/ Some(stored_thread.created_at),
);
}
state.notify_thread_created(reloaded_thread.thread_id);
Ok(())
}
@@ -831,6 +862,7 @@ impl LocalAgentControl {
parent_thread_id,
thread_config,
subagent_source.clone(),
/*resumed_created_at*/ None,
);
}
+1
View File
@@ -172,6 +172,7 @@ pub(crate) async fn run_codex_thread_interactive(
Some(parent_session.thread_id),
thread_config,
subagent_source,
/*resumed_created_at*/ None,
);
let caller_io = forward_session_io(Arc::new(io), cancel_token);
startup.release_membership();
@@ -196,6 +196,7 @@ impl PreparedGuardianContext {
Some(self.parent.thread_id()),
session.thread_config_snapshot().await,
SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_owned()),
/*resumed_created_at*/ None,
);
GuardianReviewSession {
session,
+15 -6
View File
@@ -59,12 +59,14 @@ use crate::turn_metadata::TurnMetadataState;
use crate::turn_timing::now_unix_timestamp_ms;
use async_channel::Receiver;
use async_channel::Sender;
use chrono::DateTime;
use chrono::Local;
use chrono::Utc;
use codex_analytics::AnalyticsEventsClient;
use codex_analytics::ImagePreparationFact;
use codex_analytics::ImagePreparationMetadata;
use codex_analytics::SubAgentThreadStartedInput;
use codex_analytics::ThreadInitializationMode;
use codex_analytics::TurnCodexErrorFact;
use codex_async_utils::OrCancelExt;
use codex_attachment_store::AttachmentStore;
@@ -5059,6 +5061,7 @@ fn apply_prepared_image_file_ids(
}
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn emit_subagent_session_started(
analytics_events_client: &AnalyticsEventsClient,
client_metadata: AppServerClientMetadata,
@@ -5067,7 +5070,15 @@ pub(crate) fn emit_subagent_session_started(
parent_thread_id: Option<ThreadId>,
thread_config: ThreadConfigSnapshot,
subagent_source: SubAgentSource,
resumed_created_at: Option<DateTime<Utc>>,
) {
let initialization_mode = if resumed_created_at.is_some() {
ThreadInitializationMode::Resumed
} else {
ThreadInitializationMode::New
};
let created_at =
u64::try_from(resumed_created_at.unwrap_or_else(Utc::now).timestamp()).unwrap_or_default();
let AppServerClientMetadata {
client_name,
client_version,
@@ -5078,11 +5089,7 @@ pub(crate) fn emit_subagent_session_started(
tracing::warn!("skipping subagent thread analytics: missing inherited client metadata");
return;
}
let created_at = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
analytics_events_client.track_subagent_thread_started(SubAgentThreadStartedInput {
let input = SubAgentThreadStartedInput {
session_id: session_id.to_string(),
thread_id: thread_id.to_string(),
parent_thread_id: parent_thread_id.map(|thread_id| thread_id.to_string()),
@@ -5096,8 +5103,10 @@ pub(crate) fn emit_subagent_session_started(
ephemeral: thread_config.ephemeral,
thread_source: thread_config.thread_source,
subagent_source,
initialization_mode,
created_at,
});
};
analytics_events_client.track_subagent_thread_started(input);
}
/// Builds hook configuration for one config snapshot, including any enabled plugin hooks.
+3
View File
@@ -5243,6 +5243,7 @@ async fn emit_subagent_session_started_includes_fork_lineage_and_originator() {
Some(parent_thread_id),
session_configuration.thread_config_snapshot(Vec::new()),
SubAgentSource::Other(crate::guardian::GUARDIAN_REVIEWER_NAME.to_string()),
/*resumed_created_at*/ None,
);
let event = timeout(Duration::from_secs(1), async {
@@ -5266,6 +5267,7 @@ async fn emit_subagent_session_started_includes_fork_lineage_and_originator() {
.await
.expect("subagent initialization analytics should be emitted");
assert_eq!(event["event_params"]["initialization_mode"], "new");
assert_eq!(event["event_params"]["thread_source"], "guardian_review");
assert_eq!(
event["event_params"]["parent_thread_id"],
@@ -5292,6 +5294,7 @@ async fn emit_subagent_session_started_includes_fork_lineage_and_originator() {
Some(parent_thread_id),
session_configuration.thread_config_snapshot(Vec::new()),
SubAgentSource::Other(crate::guardian::GUARDIAN_REVIEWER_NAME.to_string()),
/*resumed_created_at*/ None,
);
// Archive analytics exposes retained lineage even before a parent connection exists.
analytics_events_client.track_notification(&ServerNotification::ThreadArchived(
@@ -1,13 +1,16 @@
use anyhow::Result;
use codex_analytics::AnalyticsEventsClient;
use codex_core::TurnInputRequest;
use codex_core::config::AgentRoleConfig;
use codex_features::Feature;
use codex_login::CodexAuth;
use codex_protocol::models::PermissionProfile;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::user_input::UserInput;
use codex_thread_store::ReadThreadParams;
use core_test_support::responses::assert_parent_turn;
use core_test_support::responses::assert_root_turn;
use core_test_support::responses::ev_assistant_message;
@@ -27,6 +30,10 @@ use serde_json::json;
use std::time::Duration;
use tokio::time::Instant;
use tokio::time::sleep;
use wiremock::Mock;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
#[path = "multi_agent_restore_tests.rs"]
mod restore_tests;
@@ -272,6 +279,7 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu
});
let initial = initial_builder.build_with_auto_env(&server).await?;
let root_thread_id = initial.session_configured.thread_id;
let root_session_id = initial.session_configured.session_id;
initial
.codex
.start_or_steer_turn(
@@ -392,6 +400,16 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu
sibling_thread.flush_rollout().await?;
worker_thread.flush_rollout().await?;
initial.codex.flush_rollout().await?;
let worker_created_at = initial
.thread_store
.read_thread(ReadThreadParams {
thread_id: worker_thread_id,
include_archived: true,
include_history: false,
})
.await?
.created_at
.timestamp();
sibling_thread.shutdown_and_wait().await?;
worker_thread.shutdown_and_wait().await?;
drop(sibling_thread);
@@ -445,11 +463,34 @@ async fn cold_root_resume_restores_agent_identity_and_role_on_followup() -> Resu
)
.await;
Mock::given(method("POST"))
.and(path("/codex/analytics-events/events"))
.respond_with(ResponseTemplate::new(/* status */ 200))
.mount(&server)
.await;
let analytics_client = AnalyticsEventsClient::new(
codex_core::test_support::auth_manager_from_auth(
CodexAuth::create_dummy_chatgpt_auth_for_testing(),
),
server.uri(),
/* analytics_enabled */ Some(true),
);
let resumed_model_provider_base_url = format!("{}/v1", server.uri());
let mut resume_builder = test_codex().with_config(move |config| {
configure_multi_agent_v2_with_role(config, &resumed_model_provider_base_url);
});
let mut resume_builder = test_codex()
.with_analytics_events_client(analytics_client.clone())
.with_config(move |config| {
configure_multi_agent_v2_with_role(config, &resumed_model_provider_base_url);
config.analytics_enabled = Some(true);
});
let resumed = resume_builder.restart(&server, &initial).await?;
resumed
.codex
.set_app_server_client_info(
Some("resume-test".to_string()),
Some("1.0.0".to_string()),
/*mcp_elicitations_auto_deny*/ false,
)
.await?;
drop(initial);
assert_eq!(
resumed.thread_manager.list_thread_ids().await,
@@ -537,6 +578,44 @@ openai_base_url = "{redirected_base_url}"
matches!(event, EventMsg::TurnComplete(_))
})
.await;
tokio::time::timeout(Duration::from_secs(10), analytics_client.flush()).await?;
let mut worker_initializations = Vec::new();
for request in server.received_requests().await.unwrap_or_default() {
if request.url.path() != "/codex/analytics-events/events" {
continue;
}
let payload: Value = serde_json::from_slice(&request.body)?;
worker_initializations.extend(
payload["events"]
.as_array()
.expect("analytics events")
.iter()
.filter(|event| {
event["event_type"] == "codex_thread_initialized"
&& event["event_params"]["thread_id"] == worker_thread_id.to_string()
})
.map(|event| {
let params = &event["event_params"];
json!({
"session_id": params["session_id"],
"parent_thread_id": params["parent_thread_id"],
"subagent_source": params["subagent_source"],
"initialization_mode": params["initialization_mode"],
"created_at": params["created_at"],
})
}),
);
}
assert_eq!(
worker_initializations,
vec![json!({
"session_id": root_session_id.to_string(),
"parent_thread_id": root_thread_id.to_string(),
"subagent_source": "thread_spawn",
"initialization_mode": "resumed",
"created_at": worker_created_at,
})]
);
assert!(followup_child_request.requests().iter().any(|request| {
request.body_contains_text(FOLLOWUP_TASK)
&& request.body_contains_text(ROLE_DEVELOPER_INSTRUCTIONS)