mirror of
https://github.com/openai/codex.git
synced 2026-10-20 00:18:49 +00:00
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
This commit is contained in:
committed by
copyberry
parent
81e180044d
commit
307ce6cda9
@@ -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),
|
||||
|
||||
@@ -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<ServerNotification> {
|
||||
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,
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
|
||||
@@ -398,8 +398,8 @@ pub struct SubAgentThreadStartedInput {
|
||||
pub parent_thread_id: Option<String>,
|
||||
pub forked_from_thread_id: Option<String>,
|
||||
pub product_client_id: String,
|
||||
pub client_name: String,
|
||||
pub client_version: String,
|
||||
pub client_name: Option<String>,
|
||||
pub client_version: Option<String>,
|
||||
pub model: String,
|
||||
pub ephemeral: bool,
|
||||
pub thread_source: Option<ThreadSource>,
|
||||
|
||||
@@ -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, .. }
|
||||
|
||||
@@ -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::<Vec<_>>();
|
||||
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::<Vec<_>>();
|
||||
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::<Vec<_>>();
|
||||
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::<Vec<_>>();
|
||||
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<Value> {
|
||||
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,
|
||||
|
||||
@@ -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(())
|
||||
}
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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::<Vec<_>>();
|
||||
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(
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user