From 307ce6cda900be1669c271a7124d002870fcb435 Mon Sep 17 00:00:00 2001 From: marksteinbrick-oai Date: Thu, 27 Aug 2026 02:13:31 +0000 Subject: [PATCH] Track Guardian reviewer turn and tool analytics (#41023) ## Why Trusted Guardian reviewer sessions can emit events without an app-server listener, leaving their turns and tool usage absent from analytics. ## What changed - Translate Guardian session lifecycle and canonical tool item events into analytics notifications while excluding messages, reasoning, and review content. - Preserve Guardian lineage during prewarming even when inherited client metadata is not yet available, without emitting an incomplete thread initialization event. - Limit direct session tracking to trusted Guardian children so resumed threads that have an app-server listener do not report duplicate events. ## Testing - Cover completed, failed, and interrupted Guardian turns, tool counts and timing, lineage, content exclusion, prewarmed lineage, and resumed-thread deduplication. GitOrigin-RevId: 70ace5f7ed30a7e3b6a9d6097c8b2a6a326dc4f7 --- .../analytics/src/analytics_client_tests.rs | 32 +-- codex-rs/analytics/src/client.rs | 106 +++++++++ codex-rs/analytics/src/events.rs | 4 +- codex-rs/analytics/src/facts.rs | 4 +- codex-rs/analytics/src/reducer.rs | 12 +- .../app-server/tests/suite/v2/analytics.rs | 211 ++++++++++++++++++ .../tests/suite/v2/client_metadata.rs | 35 +++ codex-rs/core/src/session/mod.rs | 24 +- codex-rs/core/src/session/tests.rs | 55 +++++ codex-rs/core/tests/suite/guardian_review.rs | 12 +- 10 files changed, 467 insertions(+), 28 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index fa85905bd7..f713623400 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -3323,8 +3323,8 @@ fn subagent_thread_started_review_serializes_expected_shape() { parent_thread_id: None, forked_from_thread_id: None, product_client_id: "codex-tui".to_string(), - client_name: "codex-tui".to_string(), - client_version: "1.0.0".to_string(), + client_name: Some("codex-tui".to_string()), + client_version: Some("1.0.0".to_string()), model: "gpt-5".to_string(), ephemeral: false, thread_source: Some(ThreadSource::Subagent), @@ -3376,8 +3376,8 @@ fn subagent_thread_started_thread_spawn_serializes_thread_lineage() { parent_thread_id: Some(parent_thread_id.to_string()), forked_from_thread_id: Some(forked_from_thread_id.to_string()), product_client_id: "codex-tui".to_string(), - client_name: "codex-tui".to_string(), - client_version: "1.0.0".to_string(), + client_name: Some("codex-tui".to_string()), + client_version: Some("1.0.0".to_string()), model: "gpt-5".to_string(), ephemeral: true, thread_source: Some(ThreadSource::Subagent), @@ -3416,8 +3416,8 @@ fn subagent_thread_started_memory_consolidation_serializes_expected_shape() { parent_thread_id: None, forked_from_thread_id: None, product_client_id: "codex-tui".to_string(), - client_name: "codex-tui".to_string(), - client_version: "1.0.0".to_string(), + client_name: Some("codex-tui".to_string()), + client_version: Some("1.0.0".to_string()), model: "gpt-5".to_string(), ephemeral: false, thread_source: Some(ThreadSource::Subagent), @@ -3444,8 +3444,8 @@ fn subagent_thread_started_other_serializes_expected_shape() { parent_thread_id: None, forked_from_thread_id: None, product_client_id: "codex-tui".to_string(), - client_name: "codex-tui".to_string(), - client_version: "1.0.0".to_string(), + client_name: Some("codex-tui".to_string()), + client_version: Some("1.0.0".to_string()), model: "gpt-5".to_string(), ephemeral: false, thread_source: Some(ThreadSource::GuardianReview), @@ -3472,8 +3472,8 @@ fn subagent_thread_started_other_serializes_explicit_parent_thread_id() { parent_thread_id: Some(parent_thread_id.to_string()), forked_from_thread_id: None, product_client_id: "codex-tui".to_string(), - client_name: "codex-tui".to_string(), - client_version: "1.0.0".to_string(), + client_name: Some("codex-tui".to_string()), + client_version: Some("1.0.0".to_string()), model: "gpt-5".to_string(), ephemeral: false, thread_source: Some(ThreadSource::GuardianReview), @@ -3504,8 +3504,8 @@ async fn subagent_thread_started_publishes_without_initialize() { parent_thread_id: None, forked_from_thread_id: None, product_client_id: "codex-tui".to_string(), - client_name: "codex-tui".to_string(), - client_version: "1.0.0".to_string(), + client_name: Some("codex-tui".to_string()), + client_version: Some("1.0.0".to_string()), model: "gpt-5".to_string(), ephemeral: false, thread_source: Some(ThreadSource::Subagent), @@ -3581,8 +3581,8 @@ async fn guardian_events_keep_thread_source_and_originator_with_explicit_turn_co parent_thread_id: Some(parent_thread_id.to_string()), forked_from_thread_id: None, product_client_id: "parent-client".to_string(), - client_name: "parent-client".to_string(), - client_version: "1.0.0".to_string(), + client_name: Some("parent-client".to_string()), + client_version: Some("1.0.0".to_string()), model: "gpt-5".to_string(), ephemeral: false, thread_source: Some(ThreadSource::GuardianReview), @@ -3718,8 +3718,8 @@ async fn subagent_tool_items_inherit_parent_connection_metadata() { parent_thread_id: Some("thread-1".to_string()), forked_from_thread_id: None, product_client_id: "codex-tui".to_string(), - client_name: "codex-tui".to_string(), - client_version: "1.0.0".to_string(), + client_name: Some("codex-tui".to_string()), + client_version: Some("1.0.0".to_string()), model: "gpt-5".to_string(), ephemeral: false, thread_source: Some(ThreadSource::Subagent), diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 42d8f0e7c3..7b5056acc8 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -36,6 +36,7 @@ use crate::facts::TurnTokenUsageFact; use crate::now_unix_millis; use crate::reducer::AnalyticsReducer; use crate::reducer::MAX_PLUGIN_MEASUREMENTS_PER_BATCH; +use crate::reducer::tracked_tool_item_id; use crate::reducer::valid_plugin_measurement_identifier; use crate::reducer::valid_plugin_measurement_row; use codex_app_server_protocol::ClientRequest; @@ -48,14 +49,24 @@ use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerResponse; +use codex_app_server_protocol::Turn; +use codex_app_server_protocol::TurnCompletedNotification; +use codex_app_server_protocol::TurnError; +use codex_app_server_protocol::TurnItemsView; +use codex_app_server_protocol::TurnStartedNotification; +use codex_app_server_protocol::TurnStatus; +use codex_app_server_protocol::item_event_to_server_notification; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_login::default_client::create_client; use codex_plugin::PluginId; use codex_plugin::PluginTelemetryMetadata; +use codex_protocol::ThreadId; use codex_protocol::items::CollabAgentToolCallItem; use codex_protocol::items::CollabAgentToolCallStatus; use codex_protocol::items::TurnItem; +use codex_protocol::protocol::Event; +use codex_protocol::protocol::EventMsg; use codex_protocol::request_permissions::RequestPermissionsResponse; use std::collections::HashSet; use std::path::PathBuf; @@ -335,6 +346,13 @@ impl AnalyticsEventsClient { )); } + /// Logs Guardian turn and tool events that bypass the app-server listener. + pub fn track_guardian_session_event(&self, thread_id: ThreadId, event: &Event) { + if let Some(notification) = session_event_to_analytics_notification(thread_id, event) { + self.track_notification(¬ification); + } + } + pub fn track_collab_tool_call( &self, turn_id: String, @@ -710,6 +728,94 @@ impl AnalyticsEventsClient { } } +fn session_event_to_analytics_notification( + thread_id: ThreadId, + event: &Event, +) -> Option { + let notification = match &event.msg { + EventMsg::ItemStarted(started) => item_event_to_server_notification( + event.msg.clone(), + &started.thread_id.to_string(), + &started.turn_id, + ), + EventMsg::ItemCompleted(completed) => item_event_to_server_notification( + event.msg.clone(), + &completed.thread_id.to_string(), + &completed.turn_id, + ), + EventMsg::TurnStarted(started) => { + ServerNotification::TurnStarted(TurnStartedNotification { + thread_id: thread_id.to_string(), + turn: Turn { + started_at: started.started_at, + ..analytics_turn(&started.turn_id, TurnStatus::InProgress) + }, + }) + } + EventMsg::TurnComplete(completed) => { + let error = completed.error.as_ref().map(|error| TurnError { + message: String::new(), + codex_error_info: error.codex_error_info.clone().map(Into::into), + additional_details: None, + misalignment: None, + }); + let status = if error.is_some() { + TurnStatus::Failed + } else { + TurnStatus::Completed + }; + ServerNotification::TurnCompleted(TurnCompletedNotification { + thread_id: thread_id.to_string(), + turn: Turn { + error, + started_at: completed.started_at, + completed_at: completed.completed_at, + duration_ms: completed.duration_ms, + ..analytics_turn(&completed.turn_id, status) + }, + }) + } + EventMsg::TurnAborted(aborted) => { + ServerNotification::TurnCompleted(TurnCompletedNotification { + thread_id: thread_id.to_string(), + turn: Turn { + started_at: aborted.started_at, + completed_at: aborted.completed_at, + duration_ms: aborted.duration_ms, + ..analytics_turn( + aborted.turn_id.as_deref().unwrap_or(&event.id), + TurnStatus::Interrupted, + ) + }, + }) + } + // Legacy tool events accompany canonical items. Messages, reasoning, and review + // content must not enter the analytics queue. + _ => return None, + }; + match ¬ification { + ServerNotification::ItemStarted(ItemStartedNotification { item, .. }) + | ServerNotification::ItemCompleted(ItemCompletedNotification { item, .. }) => { + tracked_tool_item_id(item)?; + } + _ => {} + } + Some(notification) +} + +fn analytics_turn(turn_id: &str, status: TurnStatus) -> Turn { + Turn { + id: turn_id.to_string(), + items: Vec::new(), + items_view: TurnItemsView::NotLoaded, + status, + error: None, + started_at: None, + completed_at: None, + duration_ms: None, + } +} + async fn send_track_events( auth_manager: &AuthManager, destination: &AnalyticsEventsDestination, diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 98aae450ee..dcfed40304 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -1450,8 +1450,8 @@ pub(crate) fn subagent_thread_started_event_request( session_id: input.session_id, app_server_client: CodexAppServerClientMetadata { product_client_id: input.product_client_id, - client_name: Some(input.client_name), - client_version: Some(input.client_version), + client_name: input.client_name, + client_version: input.client_version, rpc_transport: AppServerRpcTransport::InProcess, experimental_api_enabled: None, }, diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 4616732383..8f8b4ed68a 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -398,8 +398,8 @@ pub struct SubAgentThreadStartedInput { pub parent_thread_id: Option, pub forked_from_thread_id: Option, pub product_client_id: String, - pub client_name: String, - pub client_version: String, + pub client_name: Option, + pub client_version: Option, pub model: String, pub ephemeral: bool, pub thread_source: Option, diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 36a1855007..3e3b56af7c 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -1091,9 +1091,13 @@ impl AnalyticsReducer { if thread_state.connection_id.is_none() { thread_state.connection_id = parent_connection_id; } - out.push(TrackEventRequest::ThreadInitialized( - subagent_thread_started_event_request(input), - )); + // Guardian prewarm can register lineage before parent client metadata is set. + // Keep the existing completeness requirement for the initialization event. + if input.client_name.is_some() && input.client_version.is_some() { + out.push(TrackEventRequest::ThreadInitialized( + subagent_thread_started_event_request(input), + )); + } } fn ingest_guardian_review( @@ -2427,7 +2431,7 @@ fn warn_missing_analytics_context( ); } -fn tracked_tool_item_id(item: &ThreadItem) -> Option<&str> { +pub(crate) fn tracked_tool_item_id(item: &ThreadItem) -> Option<&str> { match item { ThreadItem::CommandExecution { id, .. } | ThreadItem::FileChange { id, .. } diff --git a/codex-rs/app-server/tests/suite/v2/analytics.rs b/codex-rs/app-server/tests/suite/v2/analytics.rs index edffe2c2aa..12de353176 100644 --- a/codex-rs/app-server/tests/suite/v2/analytics.rs +++ b/codex-rs/app-server/tests/suite/v2/analytics.rs @@ -8,6 +8,8 @@ use app_test_support::create_mock_responses_server_sequence; use app_test_support::to_response; use app_test_support::write_chatgpt_auth; use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url; +use codex_app_server_protocol::ApprovalsReviewer; +use codex_app_server_protocol::AskForApproval; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; @@ -18,6 +20,7 @@ use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::UserInput; use codex_config::types::AuthCredentialsStoreMode; use codex_config::types::OtelExporterKind; @@ -49,6 +52,197 @@ use wiremock::matchers::path_regex; const SERVICE_VERSION: &str = "0.0.0-test"; +#[tokio::test] +async fn guardian_review_turns_and_tools_reach_analytics() -> Result<()> { + skip_if_no_network!(Ok(())); + const PRIVATE: &str = "guardian-private-content"; + const READ_TIMEOUT: Duration = Duration::from_secs(60); + let server = responses::start_mock_server().await; + let codex_home = TempDir::new()?; + mount_analytics_capture(&server, codex_home.path()).await?; + MockResponsesConfig::new(&server.uri()) + .with_root_config(&format!("chatgpt_base_url = \"{}\"", server.uri())) + .with_provider_config("supports_websockets = false") + .enable_feature(Feature::GuardianApproval) + .disable_feature(Feature::Apps) + .write(codex_home.path())?; + let parent_tool = |id: &str| { + responses::sse_response(responses::sse(vec![ + responses::ev_response_created(id), + responses::ev_function_call( + id, + "exec_command", + &json!({ + "cmd": format!("echo {id} > {id}.txt"), + "sandbox_permissions": "require_escalated", + "justification": "Exercise Guardian analytics", + }) + .to_string(), + ), + responses::ev_completed(id), + ])) + }; + let requests = responses::mount_response_sequence( + &server, + vec![ + parent_tool("parent-first"), + responses::sse_response(responses::sse(vec![ + responses::ev_response_created("guardian-first"), + responses::ev_web_search_call_added_partial("guardian-search", "in_progress"), + responses::ev_web_search_call_done("guardian-search", "completed", PRIVATE), + responses::ev_reasoning_item("guardian-reasoning", &[PRIVATE], &[PRIVATE]), + responses::ev_assistant_message( + "guardian-answer", + &json!({"outcome": "deny", "rationale": PRIVATE}).to_string(), + ), + responses::ev_completed("guardian-first"), + ])), + parent_tool("parent-second"), + responses::sse_response(responses::sse_failed( + "guardian-failed", + "invalid_prompt", + PRIVATE, + )), + parent_tool("parent-third"), + responses::sse_response(responses::sse(vec![responses::ev_response_created( + "guardian-pending", + )])) + .set_delay(READ_TIMEOUT), + ], + ) + .await; + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_managed_config() + // Basic Guardian command approvals require host-native paths. + .without_auto_env() + .build_initialized_with_timeout(READ_TIMEOUT) + .await?; + let ThreadStartResponse { thread, .. } = app_server + .request(|request_id| ClientRequest::ThreadStart { + request_id, + params: ThreadStartParams { + approval_policy: Some(AskForApproval::OnRequest), + approvals_reviewer: Some(ApprovalsReviewer::AutoReview), + ..Default::default() + }, + }) + .await?; + let request_id = app_server + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "Review three commands".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let TurnStartResponse { turn } = + timeout(READ_TIMEOUT, app_server.read_response(request_id)).await??; + let reviews = timeout(READ_TIMEOUT, async { + loop { + let reviews = requests + .requests() + .into_iter() + .map(|request| request.body_json()["client_metadata"].clone()) + .filter(|metadata| metadata["x-openai-subagent"] == "guardian") + .collect::>(); + if reviews.len() == 3 { + break reviews; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await?; + app_server + .interrupt_turn_and_wait_for_aborted(thread.id.clone(), turn.id, READ_TIMEOUT) + .await?; + wait_for_matching_analytics_event(&server, READ_TIMEOUT, |event| { + event["event_type"] == "codex_turn_event" + && event["event_params"]["turn_id"] == reviews[2]["turn_id"] + }) + .await?; + timeout(READ_TIMEOUT, app_server.shutdown_gracefully()).await??; + let events = captured_analytics_events(&server).await; + assert!(!serde_json::to_string(&events)?.contains(PRIVATE)); + let children = events + .iter() + .filter(|event| event["event_params"]["subagent_source"] == "guardian") + .collect::>(); + for event in &children { + let params = &event["event_params"]; + assert_eq!( + json!([ + params["thread_id"], + params["session_id"], + params["parent_thread_id"], + params["thread_source"] + ]), + json!([ + reviews[0]["thread_id"], + thread.session_id, + thread.id, + "guardian_review" + ]) + ); + } + let turns = children + .iter() + .filter(|event| event["event_type"] == "codex_turn_event") + .map(|event| { + let params = &event["event_params"]; + let started_at = params["started_at"].as_u64().expect("turn start time"); + let completed_at = params["completed_at"] + .as_u64() + .expect("turn completion time"); + assert!(started_at > 0 && completed_at >= started_at); + assert!(params["duration_ms"].is_u64()); + json!([ + params["turn_id"], + params["status"], + params["turn_error"], + params["total_tool_call_count"], + params["web_search_count"] + ]) + }) + .collect::>(); + assert_eq!( + turns, + vec![ + json!([reviews[0]["turn_id"], "completed", null, 1, 1]), + json!([reviews[1]["turn_id"], "failed", "other", 0, 0]), + json!([reviews[2]["turn_id"], "interrupted", null, 0, 0]), + ] + ); + let tools = children + .iter() + .filter(|event| event["event_type"] == "codex_web_search_event") + .collect::>(); + let [tool] = tools.as_slice() else { + anyhow::bail!( + "expected one Guardian web search event, got {}", + tools.len() + ); + }; + let params = &tool["event_params"]; + let started_at = params["started_at_ms"].as_u64().expect("tool start time"); + let completed_at = params["completed_at_ms"] + .as_u64() + .expect("tool completion time"); + assert!(started_at > 0 && completed_at >= started_at); + assert_eq!(params["duration_ms"], completed_at - started_at); + assert_eq!( + json!([ + params["turn_id"], + params["item_id"], + params["terminal_status"] + ]), + json!([reviews[0]["turn_id"], "guardian-search", "completed"]) + ); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn multi_agent_v2_tools_emit_collaborator_analytics() -> Result<()> { const PARENT_PROMPT: &str = "exercise collaborator analytics"; @@ -314,6 +508,23 @@ pub(crate) async fn mount_analytics_capture(server: &MockServer, codex_home: &Pa Ok(()) } +pub(crate) async fn captured_analytics_events(server: &MockServer) -> Vec { + server + .received_requests() + .await + .unwrap_or_default() + .into_iter() + .filter(|request| request.url.path() == "/codex/analytics-events/events") + .flat_map(|request| { + let payload: Value = serde_json::from_slice(&request.body).expect("analytics payload"); + payload["events"] + .as_array() + .expect("analytics events") + .clone() + }) + .collect() +} + pub(crate) async fn wait_for_analytics_payload( server: &MockServer, read_timeout: Duration, diff --git a/codex-rs/app-server/tests/suite/v2/client_metadata.rs b/codex-rs/app-server/tests/suite/v2/client_metadata.rs index 28f441fd4c..ebefde6e49 100644 --- a/codex-rs/app-server/tests/suite/v2/client_metadata.rs +++ b/codex-rs/app-server/tests/suite/v2/client_metadata.rs @@ -30,6 +30,10 @@ use std::collections::HashMap; use tempfile::TempDir; use tokio::time::timeout; +use super::analytics::captured_analytics_events; +use super::analytics::mount_analytics_capture; +use super::analytics::wait_for_analytics_event; + // Bazel CI can spend tens of seconds starting app-server subprocesses or // processing turn RPCs under load. const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60); @@ -306,6 +310,8 @@ async fn turn_start_sends_nested_subagent_lineage_after_cold_thread_resume_v2() &server, responses::sse(vec![ responses::ev_response_created("resp-1"), + responses::ev_web_search_call_added_partial("resumed-search", "in_progress"), + responses::ev_web_search_call_done("resumed-search", "completed", "test query"), responses::ev_assistant_message("msg-1", "Done"), responses::ev_completed("resp-1"), ]), @@ -314,8 +320,10 @@ async fn turn_start_sends_nested_subagent_lineage_after_cold_thread_resume_v2() let codex_home = TempDir::new()?; MockResponsesConfig::new(&server.uri()) + .with_root_config(&format!("chatgpt_base_url = \"{}\"", server.uri())) .with_provider_config("supports_websockets = false") .write(codex_home.path())?; + mount_analytics_capture(&server, codex_home.path()).await?; let root_thread_id = CoreThreadId::new(); let root_thread_id_str = root_thread_id.to_string(); @@ -393,6 +401,33 @@ async fn turn_start_sends_nested_subagent_lineage_after_cold_thread_resume_v2() assert_eq!(metadata["turn_id"].as_str(), Some(turn.id.as_str())); assert!(metadata.get("forked_from_thread_id").is_none()); + let turn_event = + wait_for_analytics_event(&server, DEFAULT_READ_TIMEOUT, "codex_turn_event").await?; + let params = &turn_event["event_params"]; + assert_eq!( + ( + params["total_tool_call_count"].as_u64(), + params["web_search_count"].as_u64() + ), + (Some(1), Some(1)) + ); + timeout(DEFAULT_READ_TIMEOUT, mcp.shutdown_gracefully()).await??; + let events = captured_analytics_events(&server).await; + let count = |event_type: &str| { + events + .iter() + .filter(|event| { + event["event_type"] == event_type + && event["event_params"]["thread_id"] == thread.id + && event["event_params"]["turn_id"] == turn.id + }) + .count() + }; + assert_eq!( + (count("codex_turn_event"), count("codex_web_search_event")), + (1, 1) + ); + Ok(()) } diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 04d47879e0..1d7fe51c1d 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1993,6 +1993,24 @@ impl Session { id: turn_context.sub_id.clone(), msg, }; + // Private reviewers have no app-server listener; publicly resumed Guardian threads do. + if matches!( + &turn_context.session_source, + SessionSource::SubAgent(SubAgentSource::Other(name)) + if name == crate::guardian::GUARDIAN_REVIEWER_NAME + ) && self.services.analytics_events_client.is_enabled() + && turn_context.parent_thread_id.is_some() + && self + .state + .lock() + .await + .session_configuration + .trusted_guardian_reviewer + { + self.services + .analytics_events_client + .track_guardian_session_event(self.thread_id, &event); + } self.send_event_raw(event).await; self.maybe_notify_parent_of_terminal_turn(turn_context, &legacy_source) .await; @@ -4329,10 +4347,12 @@ pub(crate) fn emit_subagent_session_started( client_name, client_version, } = client_metadata; - let (Some(client_name), Some(client_version)) = (client_name, client_version) else { + if (client_name.is_none() || client_version.is_none()) + && subagent_source.kind() != crate::guardian::GUARDIAN_REVIEWER_NAME + { tracing::warn!("skipping subagent thread analytics: missing inherited client metadata"); return; - }; + } let created_at = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 3889978a0a..9aa49c1a19 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -5151,6 +5151,8 @@ pub(crate) async fn make_session_configuration_for_tests() -> SessionConfigurati #[tokio::test] async fn emit_subagent_session_started_includes_fork_lineage_and_originator() { + use codex_app_server_protocol::ServerNotification; + use codex_app_server_protocol::ThreadArchivedNotification; use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; @@ -5226,6 +5228,59 @@ async fn emit_subagent_session_started_includes_fork_lineage_and_originator() { event["event_params"]["app_server_client"]["product_client_id"], "test_originator" ); + + let prewarmed_thread_id = ThreadId::new(); + emit_subagent_session_started( + &analytics_events_client, + AppServerClientMetadata { + client_name: None, + client_version: None, + }, + SessionId::from(parent_thread_id), + prewarmed_thread_id, + Some(parent_thread_id), + session_configuration.thread_config_snapshot(Vec::new()), + SubAgentSource::Other(crate::guardian::GUARDIAN_REVIEWER_NAME.to_string()), + ); + // Archive analytics exposes retained lineage even before a parent connection exists. + analytics_events_client.track_notification(&ServerNotification::ThreadArchived( + ThreadArchivedNotification { + thread_id: prewarmed_thread_id.to_string(), + }, + )); + analytics_events_client.flush().await; + let events = server + .received_requests() + .await + .expect("analytics requests") + .into_iter() + .flat_map(|request| { + let payload: serde_json::Value = + serde_json::from_slice(&request.body).expect("valid analytics payload"); + payload["events"] + .as_array() + .expect("analytics events") + .clone() + }) + .collect::>(); + let [initialization, archive] = events.as_slice() else { + panic!("expected one complete initialization and one archive: {events:?}"); + }; + assert_eq!(initialization, &event); + assert_eq!( + json!([ + archive["event_type"], + archive["event_params"]["thread_id"], + archive["event_params"]["thread_source"], + archive["event_params"]["parent_thread_id"], + ]), + json!([ + "codex_thread_archive_event", + prewarmed_thread_id.to_string(), + "guardian_review", + parent_thread_id.to_string(), + ]) + ); } async fn resolved_environments_for_configuration( diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index 6d06a74ec5..4a963e802e 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/codex-rs/core/tests/suite/guardian_review.rs @@ -145,10 +145,11 @@ async fn guardian_session_inherits_parent_http_fallback( ); let server = start_mock_server().await; - Mock::given(method("GET")) + let websocket_fallback = Mock::given(method("GET")) .and(path_regex(".*/responses$")) .respond_with(ResponseTemplate::new(426)) - .mount(&server) + .expect(1) + .mount_as_scoped(&server) .await; let responses = mount_sse_sequence( @@ -189,6 +190,13 @@ async fn guardian_session_inherits_parent_http_fallback( }); let test = builder.build_with_auto_env(&server).await?; + // Let startup prewarm observe the initial reviewer before enabling Guardian. + tokio::time::timeout( + Duration::from_secs(5), + websocket_fallback.wait_until_satisfied(), + ) + .await?; + test.codex .start_or_steer_turn( TurnInputRequest::user_input(vec![UserInput::Text {