[observability] Cover thread started analytics variants

Expand the thread.started observation conformance test so the observation path is checked against the existing analytics reducer for new, resumed, forked, and subagent thread activations.

This keeps the scope limited to proof coverage for the thread lifecycle taxonomy without adding another event family or instrumentation path.
This commit is contained in:
Albin Cassirer
2026-04-19 16:36:01 -07:00
parent 6922fe3b52
commit 04667d0ea6
+102 -19
View File
@@ -65,6 +65,7 @@ use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::SessionSource as AppServerSessionSource;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadForkResponse;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus;
@@ -202,6 +203,27 @@ fn sample_thread_resume_response_with_source(
}
}
fn sample_thread_fork_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse {
let mut thread = sample_thread(thread_id, ephemeral);
thread.forked_from_id = Some("parent-thread".to_string());
ClientResponse::ThreadFork {
request_id: RequestId::Integer(3),
response: ThreadForkResponse {
thread,
model: model.to_string(),
model_provider: "openai".to_string(),
service_tier: None,
cwd: test_path_buf("/tmp").abs(),
instruction_sources: Vec::new(),
approval_policy: AppServerAskForApproval::OnFailure,
approvals_reviewer: AppServerApprovalsReviewer::User,
sandbox: AppServerSandboxPolicy::DangerFullAccess,
reasoning_effort: None,
},
}
}
fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest {
ClientRequest::TurnStart {
request_id: RequestId::Integer(request_id),
@@ -1714,7 +1736,84 @@ async fn feature_observations_match_legacy_analytics_facts() {
}
#[tokio::test]
async fn thread_started_observation_matches_legacy_response() {
async fn thread_started_observation_matches_legacy_responses() {
assert_thread_started_observation_matches_legacy_response(
sample_thread_start_response("thread-observed", /*ephemeral*/ false, "gpt-5"),
codex_observability::events::ThreadStarted {
thread_id: "thread-observed",
source: codex_observability::events::ThreadSource::User,
parent_thread_id: None,
initialization_mode: codex_observability::events::ThreadInitializationMode::New,
model: "gpt-5",
ephemeral: false,
created_at: 1,
},
)
.await;
assert_thread_started_observation_matches_legacy_response(
sample_thread_resume_response("thread-resumed", /*ephemeral*/ true, "gpt-5.1"),
codex_observability::events::ThreadStarted {
thread_id: "thread-resumed",
source: codex_observability::events::ThreadSource::User,
parent_thread_id: None,
initialization_mode: codex_observability::events::ThreadInitializationMode::Resumed,
model: "gpt-5.1",
ephemeral: true,
created_at: 1,
},
)
.await;
assert_thread_started_observation_matches_legacy_response(
sample_thread_fork_response("thread-forked", /*ephemeral*/ false, "gpt-5.2"),
codex_observability::events::ThreadStarted {
thread_id: "thread-forked",
source: codex_observability::events::ThreadSource::User,
parent_thread_id: None,
initialization_mode: codex_observability::events::ThreadInitializationMode::Forked,
model: "gpt-5.2",
ephemeral: false,
created_at: 1,
},
)
.await;
let parent_thread_id =
codex_protocol::ThreadId::from_string("22222222-2222-2222-2222-222222222222")
.expect("valid parent thread id");
assert_thread_started_observation_matches_legacy_response(
sample_thread_resume_response_with_source(
"thread-subagent",
/*ephemeral*/ false,
"gpt-5",
AppServerSessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
agent_path: None,
agent_nickname: None,
agent_role: None,
}),
),
codex_observability::events::ThreadStarted {
thread_id: "thread-subagent",
source: codex_observability::events::ThreadSource::Subagent(
codex_observability::events::ThreadSubagentKind::ThreadSpawn,
),
parent_thread_id: Some("22222222-2222-2222-2222-222222222222"),
initialization_mode: codex_observability::events::ThreadInitializationMode::Resumed,
model: "gpt-5",
ephemeral: false,
created_at: 1,
},
)
.await;
}
async fn assert_thread_started_observation_matches_legacy_response(
legacy_response: ClientResponse,
observation: codex_observability::events::ThreadStarted<'_>,
) {
let mut legacy_reducer = AnalyticsReducer::default();
let mut observation_reducer = AnalyticsObservationReducer::default();
let mut legacy_events = Vec::new();
@@ -1725,11 +1824,7 @@ async fn thread_started_observation_matches_legacy_response() {
.ingest(
AnalyticsFact::Response {
connection_id: 7,
response: Box::new(sample_thread_start_response(
"thread-observed",
/*ephemeral*/ false,
"gpt-5",
)),
response: Box::new(legacy_response),
},
&mut legacy_events,
)
@@ -1738,19 +1833,7 @@ async fn thread_started_observation_matches_legacy_response() {
observation_reducer
.ingest_existing_fact_for_test(sample_initialize_fact(), &mut observation_events)
.await;
observation_reducer.ingest_thread_started(
7,
codex_observability::events::ThreadStarted {
thread_id: "thread-observed",
source: codex_observability::events::ThreadSource::User,
parent_thread_id: None,
initialization_mode: codex_observability::events::ThreadInitializationMode::New,
model: "gpt-5",
ephemeral: false,
created_at: 1,
},
&mut observation_events,
);
observation_reducer.ingest_thread_started(7, observation, &mut observation_events);
let legacy_payload = serde_json::to_value(&legacy_events).expect("serialize legacy events");
let observation_payload =