[codex-analytics] emit internally started turn events (#27392)

## Why
Currently, the analytics reducer omits `codex_turn_event` for internally
started subagent turns
- It uses `TurnState.connection_id` to select app-server client and
runtime metadata
- `turn/start` sets this field for client-started turns, while internal
subagent turns bypass that path
- Spawned child threads inherit the correct connection, but turn
emission does not use thread state

## What Changed
- Keeps explicit `TurnState.connection_id` authoritative for
client-started turns
- Falls back to the matching thread’s inherited connection when the turn
connection is absent
- Preserves completeness gates, event schema, and post-emission state
removal
- Extends subagent lifecycle test coverage

## Verification
- `just test -p codex-analytics` (71 tests passed)
- `just fix -p codex-analytics`
- `just fmt`
This commit is contained in:
marksteinbrick-oai
2026-06-10 15:35:41 -07:00
committed by GitHub
Unverified
parent c72205239f
commit b39f943a63
2 changed files with 99 additions and 4 deletions
@@ -778,6 +778,33 @@ fn sample_initialize_fact(connection_id: u64) -> AnalyticsFact {
}
}
async fn ingest_complete_child_turn(
reducer: &mut AnalyticsReducer,
events: &mut Vec<TrackEventRequest>,
thread_id: &str,
turn_id: &str,
) {
for fact in [
AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
sample_turn_resolved_config(thread_id, turn_id),
))),
AnalyticsFact::Custom(CustomAnalyticsFact::TurnProfile(Box::new(
TurnProfileFact {
turn_id: turn_id.to_string(),
profile: sample_turn_profile(),
},
))),
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
thread_id,
turn_id,
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
] {
reducer.ingest(fact, events).await;
}
}
fn sample_command_execution_item(
status: CommandExecutionStatus,
exit_code: Option<i32>,
@@ -2667,7 +2694,7 @@ async fn subagent_thread_started_publishes_without_initialize() {
}
#[tokio::test]
async fn subagent_thread_started_inherits_parent_connection_for_new_thread() {
async fn subagent_events_use_inherited_connection_unless_turn_connection_is_explicit() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
let parent_thread_id =
@@ -2774,6 +2801,68 @@ async fn subagent_thread_started_inherits_parent_connection_for_new_thread() {
payload[0]["event_params"]["parent_thread_id"],
"44444444-4444-4444-4444-444444444444"
);
events.clear();
ingest_complete_child_turn(&mut reducer, &mut events, "thread-review", "turn-inherited").await;
let [TrackEventRequest::TurnEvent(event)] = events.as_slice() else {
panic!("expected one turn event");
};
let params = &event.event_params;
assert_eq!(params.session_id, "session-root");
assert_eq!(params.thread_source, Some(ThreadSource::Subagent));
assert_eq!(params.subagent_source.as_deref(), Some("thread_spawn"));
assert_eq!(
params.parent_thread_id.as_deref(),
Some("44444444-4444-4444-4444-444444444444")
);
assert_eq!(params.app_server_client.product_client_id, "parent-client");
assert_eq!(params.runtime.codex_rs_version, "0.1.0");
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnTokenUsage(Box::new(
sample_turn_token_usage_fact("thread-review", "turn-inherited"),
))),
&mut events,
)
.await;
assert_eq!(events.len(), 1);
events.clear();
reducer
.ingest(sample_initialize_fact(/*connection_id*/ 8), &mut events)
.await;
reducer
.ingest(
AnalyticsFact::ClientRequest {
connection_id: 8,
request_id: RequestId::Integer(3),
request: Box::new(sample_turn_start_request(
"thread-review",
/*request_id*/ 3,
)),
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 8,
request_id: RequestId::Integer(3),
response: Box::new(sample_turn_start_response("turn-explicit")),
},
&mut events,
)
.await;
ingest_complete_child_turn(&mut reducer, &mut events, "thread-review", "turn-explicit").await;
let [TrackEventRequest::TurnEvent(event)] = events.as_slice() else {
panic!("expected one turn event");
};
assert_eq!(
event.event_params.app_server_client.product_client_id,
DEFAULT_ORIGINATOR
);
}
#[tokio::test]
+9 -3
View File
@@ -1497,17 +1497,23 @@ impl AnalyticsReducer {
let Some(thread_id) = turn_state.thread_id.as_ref() else {
return;
};
let Some(connection_id) = turn_state.connection_id else {
let drop_site = AnalyticsDropSite::turn(thread_id, turn_id);
let connection_id = turn_state.connection_id.or_else(|| {
self.threads
.get(drop_site.thread_id)
.and_then(|thread| thread.connection_id)
});
let Some(connection_id) = connection_id else {
warn_missing_analytics_context(&drop_site, MissingAnalyticsContext::ThreadConnection);
return;
};
let Some(connection_state) = self.connections.get(&connection_id) else {
warn_missing_analytics_context(
&AnalyticsDropSite::turn(thread_id, turn_id),
&drop_site,
MissingAnalyticsContext::Connection { connection_id },
);
return;
};
let drop_site = AnalyticsDropSite::turn(thread_id, turn_id);
let Some(thread_metadata) = self
.threads
.get(drop_site.thread_id)