[app-server] collapse client response analytics facts

This commit is contained in:
Roy Han
2026-04-28 07:33:36 -07:00
parent 114566ae9d
commit 0caa739982
6 changed files with 115 additions and 215 deletions
+100 -119
View File
@@ -61,7 +61,7 @@ use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer;
use codex_app_server_protocol::AskForApproval as AppServerAskForApproval;
use codex_app_server_protocol::ClientInfo;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::ClientResponsePayload;
use codex_app_server_protocol::CodexErrorInfo;
use codex_app_server_protocol::InitializeCapabilities;
use codex_app_server_protocol::InitializeParams;
@@ -147,23 +147,24 @@ fn sample_thread_with_source(
}
}
fn sample_thread_start_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse {
ClientResponse::ThreadStart {
request_id: RequestId::Integer(1),
response: ThreadStartResponse {
thread: sample_thread(thread_id, ephemeral),
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,
permission_profile: Some(sample_permission_profile()),
reasoning_effort: None,
},
}
fn sample_thread_start_response(
thread_id: &str,
ephemeral: bool,
model: &str,
) -> ClientResponsePayload {
ClientResponsePayload::ThreadStart(ThreadStartResponse {
thread: sample_thread(thread_id, ephemeral),
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,
permission_profile: Some(sample_permission_profile()),
reasoning_effort: None,
})
}
fn sample_permission_profile() -> AppServerPermissionProfile {
@@ -189,7 +190,11 @@ fn sample_runtime_metadata() -> CodexRuntimeMetadata {
}
}
fn sample_thread_resume_response(thread_id: &str, ephemeral: bool, model: &str) -> ClientResponse {
fn sample_thread_resume_response(
thread_id: &str,
ephemeral: bool,
model: &str,
) -> ClientResponsePayload {
sample_thread_resume_response_with_source(
thread_id,
ephemeral,
@@ -203,23 +208,20 @@ fn sample_thread_resume_response_with_source(
ephemeral: bool,
model: &str,
source: AppServerSessionSource,
) -> ClientResponse {
ClientResponse::ThreadResume {
request_id: RequestId::Integer(2),
response: ThreadResumeResponse {
thread: sample_thread_with_source(thread_id, ephemeral, source),
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,
permission_profile: Some(sample_permission_profile()),
reasoning_effort: None,
},
}
) -> ClientResponsePayload {
ClientResponsePayload::ThreadResume(ThreadResumeResponse {
thread: sample_thread_with_source(thread_id, ephemeral, source),
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,
permission_profile: Some(sample_permission_profile()),
reasoning_effort: None,
})
}
fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest {
@@ -241,21 +243,18 @@ fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest
}
}
fn sample_turn_start_response(turn_id: &str, request_id: i64) -> ClientResponse {
ClientResponse::TurnStart {
request_id: RequestId::Integer(request_id),
response: codex_app_server_protocol::TurnStartResponse {
turn: Turn {
id: turn_id.to_string(),
items: vec![],
status: AppServerTurnStatus::InProgress,
error: None,
started_at: None,
completed_at: None,
duration_ms: None,
},
fn sample_turn_start_response(turn_id: &str) -> ClientResponsePayload {
ClientResponsePayload::TurnStart(codex_app_server_protocol::TurnStartResponse {
turn: Turn {
id: turn_id.to_string(),
items: vec![],
status: AppServerTurnStatus::InProgress,
error: None,
started_at: None,
completed_at: None,
duration_ms: None,
},
}
})
}
fn sample_turn_started_notification(thread_id: &str, turn_id: &str) -> ServerNotification {
@@ -361,12 +360,21 @@ fn sample_turn_steer_request(
}
}
fn sample_turn_steer_response(turn_id: &str, request_id: i64) -> ClientResponse {
ClientResponse::TurnSteer {
fn sample_turn_steer_response(turn_id: &str) -> ClientResponsePayload {
ClientResponsePayload::TurnSteer(TurnSteerResponse {
turn_id: turn_id.to_string(),
})
}
fn client_response_fact(
connection_id: u64,
request_id: i64,
response: ClientResponsePayload,
) -> AnalyticsFact {
AnalyticsFact::ClientResponse {
connection_id,
request_id: RequestId::Integer(request_id),
response: TurnSteerResponse {
turn_id: turn_id.to_string(),
},
response: Box::new(response),
}
}
@@ -492,12 +500,11 @@ async fn ingest_turn_prerequisites(
ingest_initialize(reducer, out).await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_thread_start_response(
"thread-2", /*ephemeral*/ false, "gpt-5",
)),
},
client_response_fact(
7,
1,
sample_thread_start_response("thread-2", /*ephemeral*/ false, "gpt-5"),
),
out,
)
.await;
@@ -516,10 +523,7 @@ async fn ingest_turn_prerequisites(
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_turn_start_response("turn-2", /*request_id*/ 3)),
},
client_response_fact(7, 3, sample_turn_start_response("turn-2")),
out,
)
.await;
@@ -899,14 +903,11 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_thread_start_response(
"thread-no-client",
/*ephemeral*/ false,
"gpt-5",
)),
},
client_response_fact(
7,
1,
sample_thread_start_response("thread-no-client", /*ephemeral*/ false, "gpt-5"),
),
&mut events,
)
.await;
@@ -943,12 +944,11 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_thread_resume_response(
"thread-1", /*ephemeral*/ true, "gpt-5",
)),
},
client_response_fact(
7,
2,
sample_thread_resume_response("thread-1", /*ephemeral*/ true, "gpt-5"),
),
&mut events,
)
.await;
@@ -1013,10 +1013,7 @@ async fn unrelated_client_requests_are_ignored_by_reducer() {
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_turn_start_response("turn-2", /*request_id*/ 3)),
},
client_response_fact(7, 3, sample_turn_start_response("turn-2")),
&mut events,
)
.await;
@@ -1035,13 +1032,11 @@ async fn unrelated_client_responses_are_ignored_by_reducer() {
ingest_initialize(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(ClientResponse::ThreadArchive {
request_id: RequestId::Integer(9),
response: ThreadArchiveResponse {},
}),
},
client_response_fact(
7,
9,
ClientResponsePayload::ThreadArchive(ThreadArchiveResponse {}),
),
&mut events,
)
.await;
@@ -1081,9 +1076,10 @@ async fn compaction_event_ingests_custom_fact() {
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_thread_resume_response_with_source(
client_response_fact(
7,
2,
sample_thread_resume_response_with_source(
"thread-1",
/*ephemeral*/ false,
"gpt-5",
@@ -1094,8 +1090,8 @@ async fn compaction_event_ingests_custom_fact() {
agent_nickname: None,
agent_role: None,
}),
)),
},
),
),
&mut events,
)
.await;
@@ -1192,14 +1188,11 @@ async fn guardian_review_event_ingests_custom_fact_with_optional_target_item() {
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_thread_start_response(
"thread-guardian",
/*ephemeral*/ false,
"gpt-5",
)),
},
client_response_fact(
7,
1,
sample_thread_start_response("thread-guardian", /*ephemeral*/ false, "gpt-5"),
),
&mut events,
)
.await;
@@ -1974,10 +1967,7 @@ async fn accepted_turn_steer_emits_expected_event() {
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 4)),
},
client_response_fact(7, 4, sample_turn_steer_response("turn-2")),
&mut out,
)
.await;
@@ -2140,10 +2130,7 @@ async fn turn_start_error_response_discards_pending_start_request() {
// failed turn/start request and attach request-scoped connection metadata.
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_turn_start_response("turn-2", /*request_id*/ 3)),
},
client_response_fact(7, 3, sample_turn_start_response("turn-2")),
&mut out,
)
.await;
@@ -2269,10 +2256,7 @@ async fn accepted_steers_increment_turn_steer_count() {
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 4)),
},
client_response_fact(7, 4, sample_turn_steer_response("turn-2")),
&mut out,
)
.await;
@@ -2315,10 +2299,7 @@ async fn accepted_steers_increment_turn_steer_count() {
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
response: Box::new(sample_turn_steer_response("turn-2", /*request_id*/ 6)),
},
client_response_fact(7, 6, sample_turn_steer_response("turn-2")),
&mut out,
)
.await;
+14 -44
View File
@@ -22,7 +22,6 @@ use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnTokenUsageFact;
use crate::reducer::AnalyticsReducer;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::ClientResponsePayload;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::JSONRPCErrorError;
@@ -225,7 +224,10 @@ impl AnalyticsEventsClient {
}
pub fn track_request(&self, connection_id: u64, request_id: RequestId, request: ClientRequest) {
if !tracks_client_request(&request) {
if !matches!(
request,
ClientRequest::TurnStart { .. } | ClientRequest::TurnSteer { .. }
) {
return;
}
self.record_fact(AnalyticsFact::ClientRequest {
@@ -325,26 +327,23 @@ impl AnalyticsEventsClient {
}
}
pub fn track_response(&self, connection_id: u64, response: ClientResponse) {
if !tracks_client_response(&response) {
return;
}
self.record_fact(AnalyticsFact::ClientResponse {
connection_id,
response: Box::new(response),
});
}
pub fn track_response_payload(
pub fn track_response(
&self,
connection_id: u64,
request_id: RequestId,
response: Box<ClientResponsePayload>,
) {
if !tracks_client_response_payload(response.as_ref()) {
if !matches!(
response.as_ref(),
ClientResponsePayload::ThreadStart(_)
| ClientResponsePayload::ThreadResume(_)
| ClientResponsePayload::ThreadFork(_)
| ClientResponsePayload::TurnStart(_)
| ClientResponsePayload::TurnSteer(_)
) {
return;
}
self.record_fact(AnalyticsFact::ResponsePayload {
self.record_fact(AnalyticsFact::ClientResponse {
connection_id,
request_id,
response,
@@ -385,35 +384,6 @@ impl AnalyticsEventsClient {
}
}
fn tracks_client_request(request: &ClientRequest) -> bool {
matches!(
request,
ClientRequest::TurnStart { .. } | ClientRequest::TurnSteer { .. }
)
}
fn tracks_client_response(response: &ClientResponse) -> bool {
matches!(
response,
ClientResponse::ThreadStart { .. }
| ClientResponse::ThreadResume { .. }
| ClientResponse::ThreadFork { .. }
| ClientResponse::TurnStart { .. }
| ClientResponse::TurnSteer { .. }
)
}
fn tracks_client_response_payload(response: &ClientResponsePayload) -> bool {
matches!(
response,
ClientResponsePayload::ThreadStart(_)
| ClientResponsePayload::ThreadResume(_)
| ClientResponsePayload::ThreadFork(_)
| ClientResponsePayload::TurnStart(_)
| ClientResponsePayload::TurnSteer(_)
)
}
async fn send_track_events(
auth_manager: &AuthManager,
base_url: &str,
-5
View File
@@ -2,7 +2,6 @@ use crate::events::AppServerRpcTransport;
use crate::events::CodexRuntimeMetadata;
use crate::events::GuardianReviewEventParams;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::ClientResponsePayload;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::JSONRPCErrorError;
@@ -281,10 +280,6 @@ pub(crate) enum AnalyticsFact {
request: Box<ClientRequest>,
},
ClientResponse {
connection_id: u64,
response: Box<ClientResponse>,
},
ResponsePayload {
connection_id: u64,
request_id: RequestId,
response: Box<ClientResponsePayload>,
-6
View File
@@ -179,12 +179,6 @@ impl AnalyticsReducer {
self.ingest_request(connection_id, request_id, *request);
}
AnalyticsFact::ClientResponse {
connection_id,
response,
} => {
self.ingest_response(connection_id, *response, out);
}
AnalyticsFact::ResponsePayload {
connection_id,
request_id,
response,
@@ -42,7 +42,6 @@ use codex_app_server_protocol::CancelLoginAccountParams;
use codex_app_server_protocol::CancelLoginAccountResponse;
use codex_app_server_protocol::CancelLoginAccountStatus;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponse;
use codex_app_server_protocol::ClientResponsePayload;
use codex_app_server_protocol::CodexErrorInfo;
use codex_app_server_protocol::CollaborationModeListParams;
@@ -2786,16 +2785,6 @@ impl CodexMessageProcessor {
match result {
Ok((response, notif)) => {
listener_task_context
.analytics_events_client
.track_response(
request_id.connection_id.0,
ClientResponse::ThreadStart {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
listener_task_context
.outgoing
.send_response(
@@ -4381,13 +4370,6 @@ impl CodexMessageProcessor {
permission_profile,
reasoning_effort: session_configured.reasoning_effort,
};
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::ThreadResume {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
let connection_id = request_id.connection_id;
let token_usage_thread = include_turns.then(|| response.thread.clone());
@@ -5005,14 +4987,6 @@ impl CodexMessageProcessor {
}
};
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::ThreadFork {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
let connection_id = request_id.connection_id;
let token_usage_thread = include_turns.then(|| response.thread.clone());
self.outgoing
@@ -6681,13 +6655,6 @@ impl CodexMessageProcessor {
match result {
Ok(response) => {
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::TurnStart {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
self.outgoing
.send_response(
request_id,
@@ -6865,13 +6832,6 @@ impl CodexMessageProcessor {
match result {
Ok(response) => {
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::TurnSteer {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
self.outgoing
.send_response(
request_id,
+1 -1
View File
@@ -507,7 +507,7 @@ impl OutgoingMessageSender {
.into_jsonrpc_parts_and_payload(request_id.request_id.clone())
.map(|(id, result, response)| {
if let Some(response) = response {
self.analytics_events_client.track_response_payload(
self.analytics_events_client.track_response(
connection_id.0,
request_id_for_analytics,
response,