Files
codex/codex-rs/analytics/src/analytics_client_tests.rs
T
kbazzi ccf1a18518 [codex] Retry transient Guardian review failures (#27062)
## Background

Codex can use **Auto Review** for permission requests. Instead of asking
the user immediately, Codex starts a separate locked-down reviewer
session called **Guardian**, which returns a structured `allow` or
`deny` assessment.

The Guardian reviewer is itself a Codex session, so its model request
can fail for transient infrastructure reasons such as model overload,
HTTP connection failure, or response-stream disconnect. Today, any such
failure immediately ends the Auto Review attempt and blocks the action.

This PR adds bounded retries for failures that the existing protocol
explicitly identifies as transient.

Linear context:
[CA-539](https://linear.app/openai/issue/CA-539/retry-auto-review-infrastructure-failures-and-fall-back-to-manual)

## What changes

A Guardian review can now make at most **three total attempts**:

1. Run the review normally.
2. Retry after a jittered delay of roughly 180–220 ms if the first
attempt fails with an eligible error.
3. Retry after a jittered delay of roughly 360–440 ms if the second
attempt also fails with an eligible error.

All attempts share the original review deadline. Jitter spreads retries
from concurrent clients to reduce synchronized load during broader
outages. The retries do not reset the user's maximum wait time, and the
backoff waits terminate early if the review is cancelled or the deadline
expires.

Before retrying, the existing Guardian session lifecycle decides whether
the session remains usable. Healthy trunks are reused, broken trunks are
removed by the existing cleanup path, and ephemeral sessions continue to
clean themselves up.

The review still emits one logical lifecycle to clients. Recoverable
intermediate failures do not produce warnings or terminal events.

## Retry policy

### Retried up to twice

- model/server overload
- HTTP connection failure
- response-stream connection failure
- response-stream disconnect
- internal server error
- a final reviewer message that cannot be parsed as the required
Guardian assessment

### Not retried

- bad or invalid requests
- authentication failures
- usage limits
- cyber-policy failures
- errors without a structured category
- a request that already exhausted the lower-level Responses retry
budget
- a completed Guardian turn with no assessment payload
- prompt-construction failures
- Guardian review timeout
- cancellation or abort
- a valid `deny` assessment

The session-error classification uses `ErrorEvent.codex_error_info`; it
does not inspect error-message strings.

## Implementation notes

- `wait_for_guardian_review` preserves the complete `ErrorEvent`,
including structured `codex_error_info`.
- Guardian session failures preserve the original message and optional
structured `CodexErrorInfo`.
- The retry policy classifies the explicitly transient `CodexErrorInfo`
variants; unknown, absent, and deterministic categories are not retried.
- The Guardian session manager receives the caller's deadline rather
than creating a new timeout per attempt.
- Analytics record the final `attempt_count`.
- Retry orchestration does not add a separate session-cleanup protocol;
it relies on the existing trunk and ephemeral lifecycle decisions.

## Automated testing

Focused Guardian coverage verifies:

- every supported transient `CodexErrorInfo` is classified as retryable,
while absent and non-transient categories are not;
- structured transient session failure -> retry -> approval with the
healthy trunk reused;
- two invalid Guardian responses -> third attempt -> approval, with
exactly three requests;
- three invalid responses -> existing fail-closed result, with exactly
three requests and one terminal lifecycle;
- valid denial, missing payload, invalid request, timeout, cancellation,
and prompt/session construction failures are not retried;
- retry eligibility ends after the third attempt;
- retry delays use the shared exponential backoff helper and remain
within the expected jitter bounds;
- cancellation and deadline expiry interrupt the backoff wait;
- healthy trunks are reused across retryable failures;
- broken event streams remove the trunk through the existing lifecycle
cleanup;
- an ephemeral retry does not disturb a concurrent trunk review.

Validation performed:

- `just test -p codex-core guardian_review_
guardian_ephemeral_retry_preserves_parallel_trunk_and_fork_history
run_review_removes_trunk_when_event_stream_is_broken` — **42 passed**;
- `just test -p codex-analytics` — **71 passed**;
- scoped Clippy fixes for `codex-core` and `codex-analytics` passed.

A prior full `codex-core` run had unrelated environment-sensitive
failures outside Guardian coverage.

## Manual QA

The focused integration tests use the local mock Responses server to
inspect exact request counts and emitted lifecycle events. They confirm
that retries are internal, a successful later attempt supplies the final
decision, non-retryable failures issue only one request, and exhausted
retries emit only one terminal result.
2026-06-10 11:46:57 -07:00

4203 lines
145 KiB
Rust

use crate::client::AnalyticsEventsQueue;
use crate::events::AppServerRpcTransport;
use crate::events::CodexAcceptedLineFingerprintsEventParams;
use crate::events::CodexAcceptedLineFingerprintsEventRequest;
use crate::events::CodexAppMentionedEventRequest;
use crate::events::CodexAppServerClientMetadata;
use crate::events::CodexAppUsedEventRequest;
use crate::events::CodexCommandExecutionEventParams;
use crate::events::CodexCommandExecutionEventRequest;
use crate::events::CodexCompactionEventRequest;
use crate::events::CodexHookRunEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexReviewEventParams;
use crate::events::CodexReviewEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::CodexToolItemEventBase;
use crate::events::CodexTurnEventRequest;
use crate::events::FinalApprovalOutcome;
use crate::events::GuardianApprovalRequestSource;
use crate::events::GuardianReviewDecision;
use crate::events::GuardianReviewEventParams;
use crate::events::GuardianReviewFailureReason;
use crate::events::GuardianReviewTerminalStatus;
use crate::events::GuardianReviewedAction;
use crate::events::ReviewResolution;
use crate::events::ReviewStatus;
use crate::events::ReviewSubjectKind;
use crate::events::ReviewTrigger;
use crate::events::Reviewer;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::ToolItemTerminalStatus;
use crate::events::TrackEventRequest;
use crate::events::codex_app_metadata;
use crate::events::codex_hook_run_metadata;
use crate::events::codex_plugin_metadata;
use crate::events::codex_plugin_used_metadata;
use crate::events::subagent_thread_started_event_request;
use crate::facts::AnalyticsFact;
use crate::facts::AnalyticsJsonRpcError;
use crate::facts::AppInvocation;
use crate::facts::AppMentionedInput;
use crate::facts::AppUsedInput;
use crate::facts::CodexCompactionEvent;
use crate::facts::CompactionImplementation;
use crate::facts::CompactionPhase;
use crate::facts::CompactionReason;
use crate::facts::CompactionStatus;
use crate::facts::CompactionStrategy;
use crate::facts::CompactionTrigger;
use crate::facts::CustomAnalyticsFact;
use crate::facts::HookRunFact;
use crate::facts::HookRunInput;
use crate::facts::InputError;
use crate::facts::InvocationType;
use crate::facts::PluginState;
use crate::facts::PluginStateChangedInput;
use crate::facts::PluginUsedInput;
use crate::facts::SkillInvocation;
use crate::facts::SkillInvokedInput;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::ThreadInitializationMode;
use crate::facts::TrackEventsContext;
use crate::facts::TurnCodexErrorFact;
use crate::facts::TurnProfile;
use crate::facts::TurnProfileFact;
use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnStatus;
use crate::facts::TurnSteerRequestError;
use crate::facts::TurnTokenUsageFact;
use crate::reducer::AnalyticsReducer;
use crate::reducer::normalize_path_for_skill_id;
use crate::reducer::skill_id_for_local_skill;
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::ClientResponsePayload;
use codex_app_server_protocol::CodexErrorInfo;
use codex_app_server_protocol::CollabAgentTool;
use codex_app_server_protocol::CollabAgentToolCallStatus;
use codex_app_server_protocol::CommandAction;
use codex_app_server_protocol::CommandExecutionApprovalDecision;
use codex_app_server_protocol::CommandExecutionRequestApprovalParams;
use codex_app_server_protocol::CommandExecutionRequestApprovalResponse;
use codex_app_server_protocol::CommandExecutionSource;
use codex_app_server_protocol::CommandExecutionStatus;
use codex_app_server_protocol::DynamicToolCallStatus;
use codex_app_server_protocol::GuardianApprovalReview;
use codex_app_server_protocol::GuardianApprovalReviewAction;
use codex_app_server_protocol::GuardianApprovalReviewStatus;
use codex_app_server_protocol::GuardianCommandSource as AppServerGuardianCommandSource;
use codex_app_server_protocol::InitializeCapabilities;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::ItemCompletedNotification;
use codex_app_server_protocol::ItemGuardianApprovalReviewCompletedNotification;
use codex_app_server_protocol::ItemStartedNotification;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::McpToolCallStatus;
use codex_app_server_protocol::NonSteerableTurnKind;
use codex_app_server_protocol::PatchApplyStatus;
use codex_app_server_protocol::PermissionsRequestApprovalParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::RequestPermissionProfile;
use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ServerResponse;
use codex_app_server_protocol::SessionSource as AppServerSessionSource;
use codex_app_server_protocol::SubAgentActivityKind;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadArchiveParams;
use codex_app_server_protocol::ThreadArchiveResponse;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadSource as AppServerThreadSource;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnDiffUpdatedNotification;
use codex_app_server_protocol::TurnError as AppServerTurnError;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartedNotification;
use codex_app_server_protocol::TurnStatus as AppServerTurnStatus;
use codex_app_server_protocol::TurnSteerParams;
use codex_app_server_protocol::TurnSteerResponse;
use codex_app_server_protocol::UserInput;
use codex_login::default_client::DEFAULT_ORIGINATOR;
use codex_login::default_client::originator;
use codex_plugin::AppConnectorId;
use codex_plugin::PluginCapabilitySummary;
use codex_plugin::PluginId;
use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::approvals::NetworkApprovalProtocol;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::ModeKind;
use codex_protocol::error::CodexErr;
use codex_protocol::models::NetworkPermissions as CoreNetworkPermissions;
use codex_protocol::models::PermissionProfile as CorePermissionProfile;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::HookEventName;
use codex_protocol::protocol::HookRunStatus;
use codex_protocol::protocol::HookSource;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::request_permissions::PermissionGrantScope as CorePermissionGrantScope;
use codex_protocol::request_permissions::RequestPermissionProfile as CoreRequestPermissionProfile;
use codex_protocol::request_permissions::RequestPermissionsResponse as CoreRequestPermissionsResponse;
use codex_utils_absolute_path::test_support::PathBufExt;
use codex_utils_absolute_path::test_support::test_path_buf;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use tokio::sync::mpsc;
fn sample_thread_with_metadata(
thread_id: &str,
ephemeral: bool,
source: AppServerSessionSource,
thread_source: Option<AppServerThreadSource>,
parent_thread_id: Option<String>,
) -> Thread {
Thread {
id: thread_id.to_string(),
session_id: format!("session-{thread_id}"),
forked_from_id: None,
parent_thread_id,
preview: "first prompt".to_string(),
ephemeral,
model_provider: "openai".to_string(),
created_at: 1,
updated_at: 2,
status: AppServerThreadStatus::Idle,
path: None,
cwd: test_path_buf("/tmp").abs(),
cli_version: "0.0.0".to_string(),
source,
thread_source,
agent_nickname: None,
agent_role: None,
git_info: None,
name: None,
turns: Vec::new(),
}
}
fn sample_thread_start_response(
thread_id: &str,
ephemeral: bool,
model: &str,
) -> ClientResponsePayload {
ClientResponsePayload::ThreadStart(ThreadStartResponse {
thread: sample_thread_with_metadata(
thread_id,
ephemeral,
AppServerSessionSource::Exec,
Some(AppServerThreadSource::User),
/*parent_thread_id*/ None,
),
model: model.to_string(),
model_provider: "openai".to_string(),
service_tier: None,
cwd: test_path_buf("/tmp").abs(),
runtime_workspace_roots: Vec::new(),
instruction_sources: Vec::new(),
approval_policy: AppServerAskForApproval::OnFailure,
approvals_reviewer: AppServerApprovalsReviewer::User,
sandbox: AppServerSandboxPolicy::DangerFullAccess,
active_permission_profile: None,
reasoning_effort: None,
})
}
fn sample_app_server_client_metadata() -> CodexAppServerClientMetadata {
CodexAppServerClientMetadata {
product_client_id: DEFAULT_ORIGINATOR.to_string(),
client_name: Some("codex-tui".to_string()),
client_version: Some("1.0.0".to_string()),
rpc_transport: AppServerRpcTransport::Stdio,
experimental_api_enabled: Some(true),
}
}
fn sample_runtime_metadata() -> CodexRuntimeMetadata {
CodexRuntimeMetadata {
codex_rs_version: "0.1.0".to_string(),
runtime_os: "macos".to_string(),
runtime_os_version: "15.3.1".to_string(),
runtime_arch: "aarch64".to_string(),
}
}
fn sample_thread_resume_response(
thread_id: &str,
ephemeral: bool,
model: &str,
) -> ClientResponsePayload {
sample_thread_resume_response_with_source(
thread_id,
ephemeral,
model,
AppServerSessionSource::Exec,
Some(AppServerThreadSource::User),
/*parent_thread_id*/ None,
)
}
fn sample_thread_resume_response_with_source(
thread_id: &str,
ephemeral: bool,
model: &str,
source: AppServerSessionSource,
thread_source: Option<AppServerThreadSource>,
parent_thread_id: Option<String>,
) -> ClientResponsePayload {
ClientResponsePayload::ThreadResume(ThreadResumeResponse {
thread: sample_thread_with_metadata(
thread_id,
ephemeral,
source,
thread_source,
parent_thread_id,
),
model: model.to_string(),
model_provider: "openai".to_string(),
service_tier: None,
cwd: test_path_buf("/tmp").abs(),
runtime_workspace_roots: Vec::new(),
instruction_sources: Vec::new(),
approval_policy: AppServerAskForApproval::OnFailure,
approvals_reviewer: AppServerApprovalsReviewer::User,
sandbox: AppServerSandboxPolicy::DangerFullAccess,
active_permission_profile: None,
reasoning_effort: None,
initial_turns_page: None,
})
}
fn sample_turn_start_request(thread_id: &str, request_id: i64) -> ClientRequest {
ClientRequest::TurnStart {
request_id: RequestId::Integer(request_id),
params: TurnStartParams {
thread_id: thread_id.to_string(),
client_user_message_id: None,
input: vec![
UserInput::Text {
text: "hello".to_string(),
text_elements: vec![],
},
UserInput::Image {
url: "https://example.com/a.png".to_string(),
detail: None,
},
],
..Default::default()
},
}
}
fn sample_turn_start_response(turn_id: &str) -> ClientResponsePayload {
ClientResponsePayload::TurnStart(codex_app_server_protocol::TurnStartResponse {
turn: Turn {
id: turn_id.to_string(),
items_view: codex_app_server_protocol::TurnItemsView::Full,
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 {
ServerNotification::TurnStarted(TurnStartedNotification {
thread_id: thread_id.to_string(),
turn: Turn {
id: turn_id.to_string(),
items_view: codex_app_server_protocol::TurnItemsView::Full,
items: vec![],
status: AppServerTurnStatus::InProgress,
error: None,
started_at: Some(455),
completed_at: None,
duration_ms: None,
},
})
}
fn sample_turn_token_usage_fact(thread_id: &str, turn_id: &str) -> TurnTokenUsageFact {
TurnTokenUsageFact {
thread_id: thread_id.to_string(),
turn_id: turn_id.to_string(),
token_usage: TokenUsage {
total_tokens: 321,
input_tokens: 123,
cached_input_tokens: 45,
output_tokens: 140,
reasoning_output_tokens: 13,
},
}
}
fn sample_turn_completed_notification(
thread_id: &str,
turn_id: &str,
status: AppServerTurnStatus,
codex_error_info: Option<codex_app_server_protocol::CodexErrorInfo>,
) -> ServerNotification {
ServerNotification::TurnCompleted(TurnCompletedNotification {
thread_id: thread_id.to_string(),
turn: Turn {
id: turn_id.to_string(),
items_view: codex_app_server_protocol::TurnItemsView::Full,
items: vec![],
status,
error: codex_error_info.map(|codex_error_info| AppServerTurnError {
message: "turn failed".to_string(),
codex_error_info: Some(codex_error_info),
additional_details: None,
}),
started_at: None,
completed_at: Some(456),
duration_ms: Some(1234),
},
})
}
fn sample_turn_resolved_config(thread_id: &str, turn_id: &str) -> TurnResolvedConfigFact {
TurnResolvedConfigFact {
turn_id: turn_id.to_string(),
thread_id: thread_id.to_string(),
num_input_images: 1,
submission_type: None,
ephemeral: false,
session_source: SessionSource::Exec,
model: "gpt-5".to_string(),
model_provider: "openai".to_string(),
permission_profile: CorePermissionProfile::read_only(),
permission_profile_cwd: PathBuf::from("/tmp"),
reasoning_effort: None,
reasoning_summary: None,
service_tier: None,
approval_policy: AskForApproval::OnRequest,
approvals_reviewer: ApprovalsReviewer::AutoReview,
sandbox_network_access: true,
collaboration_mode: ModeKind::Plan,
personality: None,
workspace_kind: None,
is_first_turn: true,
}
}
fn sample_turn_profile() -> TurnProfile {
TurnProfile {
before_first_sampling_ms: 100,
sampling_ms: 700,
between_sampling_overhead_ms: 50,
tool_blocking_ms: 250,
after_last_sampling_ms: 134,
sampling_request_count: 2,
sampling_retry_count: 1,
}
}
fn sample_turn_steer_request(
thread_id: &str,
expected_turn_id: &str,
request_id: i64,
) -> ClientRequest {
ClientRequest::TurnSteer {
request_id: RequestId::Integer(request_id),
params: TurnSteerParams {
thread_id: thread_id.to_string(),
expected_turn_id: expected_turn_id.to_string(),
client_user_message_id: None,
input: vec![
UserInput::Text {
text: "more".to_string(),
text_elements: vec![],
},
UserInput::LocalImage {
path: "/tmp/a.png".into(),
detail: None,
},
],
responsesapi_client_metadata: None,
additional_context: None,
},
}
}
fn sample_turn_steer_response(turn_id: &str) -> ClientResponsePayload {
ClientResponsePayload::TurnSteer(TurnSteerResponse {
turn_id: turn_id.to_string(),
})
}
fn no_active_turn_steer_error() -> JSONRPCErrorError {
JSONRPCErrorError {
code: -32600,
message: "no active turn to steer".to_string(),
data: None,
}
}
fn no_active_turn_steer_error_type() -> AnalyticsJsonRpcError {
AnalyticsJsonRpcError::TurnSteer(TurnSteerRequestError::NoActiveTurn)
}
fn non_steerable_review_error() -> JSONRPCErrorError {
JSONRPCErrorError {
code: -32600,
message: "cannot steer a review turn".to_string(),
data: Some(
serde_json::to_value(AppServerTurnError {
message: "cannot steer a review turn".to_string(),
codex_error_info: Some(CodexErrorInfo::ActiveTurnNotSteerable {
turn_kind: NonSteerableTurnKind::Review,
}),
additional_details: None,
})
.expect("serialize turn error"),
),
}
}
fn non_steerable_review_error_type() -> AnalyticsJsonRpcError {
AnalyticsJsonRpcError::TurnSteer(TurnSteerRequestError::NonSteerableReview)
}
fn input_too_large_steer_error() -> JSONRPCErrorError {
JSONRPCErrorError {
code: -32602,
message: "Input exceeds the maximum length of 1048576 characters.".to_string(),
data: Some(json!({
"input_error_code": "input_too_large",
"actual_chars": 1048577,
"max_chars": 1048576,
})),
}
}
fn input_too_large_error_type() -> AnalyticsJsonRpcError {
AnalyticsJsonRpcError::Input(InputError::TooLarge)
}
async fn ingest_rejected_turn_steer(
reducer: &mut AnalyticsReducer,
out: &mut Vec<TrackEventRequest>,
error: JSONRPCErrorError,
error_type: Option<AnalyticsJsonRpcError>,
) -> serde_json::Value {
ingest_turn_prerequisites(
reducer, out, /*include_initialize*/ true, /*include_resolved_config*/ false,
/*include_started*/ false, /*include_token_usage*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Initialize {
connection_id: 8,
params: InitializeParams {
client_info: ClientInfo {
name: "codex-web".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: None,
},
product_client_id: "codex-web".to_string(),
runtime: sample_runtime_metadata(),
rpc_transport: AppServerRpcTransport::Stdio,
},
out,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 8,
request_id: RequestId::Integer(6),
response: Box::new(sample_thread_resume_response(
"thread-2", /*ephemeral*/ false, "gpt-5",
)),
},
out,
)
.await;
out.clear();
reducer
.ingest(
AnalyticsFact::ClientRequest {
connection_id: 7,
request_id: RequestId::Integer(4),
request: Box::new(sample_turn_steer_request(
"thread-2", "turn-2", /*request_id*/ 4,
)),
},
out,
)
.await;
reducer
.ingest(
AnalyticsFact::ErrorResponse {
connection_id: 7,
request_id: RequestId::Integer(4),
error,
error_type,
},
out,
)
.await;
assert_eq!(out.len(), 1);
serde_json::to_value(&out[0]).expect("serialize turn steer event")
}
async fn ingest_initialize(reducer: &mut AnalyticsReducer, out: &mut Vec<TrackEventRequest>) {
reducer
.ingest(
AnalyticsFact::Initialize {
connection_id: 7,
params: InitializeParams {
client_info: ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: None,
},
product_client_id: "codex-tui".to_string(),
runtime: sample_runtime_metadata(),
rpc_transport: AppServerRpcTransport::Stdio,
},
out,
)
.await;
}
async fn ingest_turn_prerequisites(
reducer: &mut AnalyticsReducer,
out: &mut Vec<TrackEventRequest>,
include_initialize: bool,
include_resolved_config: bool,
include_started: bool,
include_token_usage: bool,
) {
if include_initialize {
ingest_initialize(reducer, out).await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(1),
response: Box::new(sample_thread_start_response(
"thread-2", /*ephemeral*/ false, "gpt-5",
)),
},
out,
)
.await;
out.clear();
}
reducer
.ingest(
AnalyticsFact::ClientRequest {
connection_id: 7,
request_id: RequestId::Integer(3),
request: Box::new(sample_turn_start_request("thread-2", /*request_id*/ 3)),
},
out,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(3),
response: Box::new(sample_turn_start_response("turn-2")),
},
out,
)
.await;
if include_resolved_config {
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
sample_turn_resolved_config("thread-2", "turn-2"),
))),
out,
)
.await;
}
if include_started {
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_started_notification(
"thread-2", "turn-2",
))),
out,
)
.await;
}
if include_token_usage {
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnTokenUsage(Box::new(
sample_turn_token_usage_fact("thread-2", "turn-2"),
))),
out,
)
.await;
}
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnProfile(Box::new(
TurnProfileFact {
turn_id: "turn-2".to_string(),
profile: sample_turn_profile(),
},
))),
out,
)
.await;
}
async fn ingest_review_prerequisites(
reducer: &mut AnalyticsReducer,
events: &mut Vec<TrackEventRequest>,
) {
reducer
.ingest(sample_initialize_fact(/*connection_id*/ 7), events)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(1),
response: Box::new(sample_thread_start_response(
"thread-1", /*ephemeral*/ false, "gpt-5",
)),
},
events,
)
.await;
events.clear();
}
async fn ingest_completed_command_execution_item(
reducer: &mut AnalyticsReducer,
events: &mut Vec<TrackEventRequest>,
thread_id: &str,
item_id: &str,
) {
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_started_notification(
thread_id, "turn-1",
))),
events,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted(
ItemStartedNotification {
thread_id: thread_id.to_string(),
turn_id: "turn-1".to_string(),
started_at_ms: 1_000,
item: sample_command_execution_item_with_id(
item_id,
CommandExecutionStatus::InProgress,
/*exit_code*/ None,
/*duration_ms*/ None,
),
},
))),
events,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::ItemCompleted(
ItemCompletedNotification {
thread_id: thread_id.to_string(),
turn_id: "turn-1".to_string(),
completed_at_ms: 1_042,
item: sample_command_execution_item_with_id(
item_id,
CommandExecutionStatus::Completed,
Some(0),
Some(42),
),
},
))),
events,
)
.await;
}
fn sample_initialize_fact(connection_id: u64) -> AnalyticsFact {
AnalyticsFact::Initialize {
connection_id,
params: InitializeParams {
client_info: ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: Some(InitializeCapabilities {
experimental_api: false,
request_attestation: false,
opt_out_notification_methods: None,
}),
},
product_client_id: DEFAULT_ORIGINATOR.to_string(),
runtime: CodexRuntimeMetadata {
codex_rs_version: "0.99.0".to_string(),
runtime_os: "linux".to_string(),
runtime_os_version: "24.04".to_string(),
runtime_arch: "x86_64".to_string(),
},
rpc_transport: AppServerRpcTransport::Websocket,
}
}
fn sample_command_execution_item(
status: CommandExecutionStatus,
exit_code: Option<i32>,
duration_ms: Option<i64>,
) -> ThreadItem {
sample_command_execution_item_with_id("item-1", status, exit_code, duration_ms)
}
fn sample_command_execution_item_with_id(
id: &str,
status: CommandExecutionStatus,
exit_code: Option<i32>,
duration_ms: Option<i64>,
) -> ThreadItem {
ThreadItem::CommandExecution {
id: id.to_string(),
command: "echo hi".to_string(),
cwd: test_path_buf("/tmp").abs(),
process_id: Some("pid-1".to_string()),
source: CommandExecutionSource::Agent,
status,
command_actions: Vec::new(),
aggregated_output: None,
exit_code,
duration_ms,
}
}
fn sample_command_execution_item_with_actions(
status: CommandExecutionStatus,
exit_code: Option<i32>,
duration_ms: Option<i64>,
command_actions: Vec<CommandAction>,
) -> ThreadItem {
let mut item = sample_command_execution_item(status, exit_code, duration_ms);
let ThreadItem::CommandExecution {
command_actions: item_command_actions,
..
} = &mut item
else {
unreachable!("sample command execution item should be CommandExecution");
};
*item_command_actions = command_actions;
item
}
fn sample_command_approval_request(request_id: i64, approval_id: Option<&str>) -> ServerRequest {
ServerRequest::CommandExecutionRequestApproval {
request_id: RequestId::Integer(request_id),
params: CommandExecutionRequestApprovalParams {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
item_id: "item-1".to_string(),
started_at_ms: 1_000,
approval_id: approval_id.map(str::to_string),
reason: None,
network_approval_context: None,
command: Some("echo hi".to_string()),
cwd: None,
command_actions: None,
additional_permissions: None,
proposed_execpolicy_amendment: None,
proposed_network_policy_amendments: None,
available_decisions: None,
},
}
}
fn sample_command_approval_response(
request_id: i64,
decision: CommandExecutionApprovalDecision,
) -> ServerResponse {
ServerResponse::CommandExecutionRequestApproval {
request_id: RequestId::Integer(request_id),
response: CommandExecutionRequestApprovalResponse { decision },
}
}
fn sample_permissions_approval_request(request_id: i64) -> ServerRequest {
ServerRequest::PermissionsRequestApproval {
request_id: RequestId::Integer(request_id),
params: PermissionsRequestApprovalParams {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
item_id: "permissions-1".to_string(),
environment_id: None,
started_at_ms: 1_000,
cwd: test_path_buf("/tmp").abs(),
reason: Some("need network".to_string()),
permissions: RequestPermissionProfile {
network: Some(codex_app_server_protocol::AdditionalNetworkPermissions {
enabled: Some(true),
}),
file_system: None,
},
},
}
}
fn sample_effective_permissions_approval_response(
permissions: CoreRequestPermissionProfile,
scope: CorePermissionGrantScope,
) -> CoreRequestPermissionsResponse {
CoreRequestPermissionsResponse {
permissions,
scope,
strict_auto_review: false,
}
}
fn sample_guardian_review_completed(
review_id: &str,
target_item_id: Option<&str>,
status: GuardianApprovalReviewStatus,
) -> ServerNotification {
ServerNotification::ItemGuardianApprovalReviewCompleted(
ItemGuardianApprovalReviewCompletedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
started_at_ms: 1_000,
completed_at_ms: 1_042,
review_id: review_id.to_string(),
target_item_id: target_item_id.map(str::to_string),
decision_source: codex_app_server_protocol::AutoReviewDecisionSource::Agent,
review: GuardianApprovalReview {
status,
risk_level: None,
user_authorization: None,
rationale: None,
},
action: GuardianApprovalReviewAction::Command {
source: AppServerGuardianCommandSource::Shell,
command: "echo hi".to_string(),
cwd: test_path_buf("/tmp").abs(),
},
},
)
}
fn expected_absolute_path(path: &PathBuf) -> String {
std::fs::canonicalize(path)
.unwrap_or_else(|_| path.to_path_buf())
.to_string_lossy()
.replace('\\', "/")
}
#[test]
fn normalize_path_for_skill_id_repo_scoped_uses_relative_path() {
let repo_root = PathBuf::from("/repo/root");
let skill_path = PathBuf::from("/repo/root/.codex/skills/doc/SKILL.md");
let path = normalize_path_for_skill_id(
Some("https://example.com/repo.git"),
Some(repo_root.as_path()),
skill_path.as_path(),
);
assert_eq!(path, ".codex/skills/doc/SKILL.md");
}
#[test]
fn normalize_path_for_skill_id_user_scoped_uses_absolute_path() {
let skill_path = PathBuf::from("/Users/abc/.codex/skills/doc/SKILL.md");
let path = normalize_path_for_skill_id(
/*repo_url*/ None,
/*repo_root*/ None,
skill_path.as_path(),
);
let expected = expected_absolute_path(&skill_path);
assert_eq!(path, expected);
}
#[test]
fn normalize_path_for_skill_id_admin_scoped_uses_absolute_path() {
let skill_path = PathBuf::from("/etc/codex/skills/doc/SKILL.md");
let path = normalize_path_for_skill_id(
/*repo_url*/ None,
/*repo_root*/ None,
skill_path.as_path(),
);
let expected = expected_absolute_path(&skill_path);
assert_eq!(path, expected);
}
#[test]
fn normalize_path_for_skill_id_repo_root_not_in_skill_path_uses_absolute_path() {
let repo_root = PathBuf::from("/repo/root");
let skill_path = PathBuf::from("/other/path/.codex/skills/doc/SKILL.md");
let path = normalize_path_for_skill_id(
Some("https://example.com/repo.git"),
Some(repo_root.as_path()),
skill_path.as_path(),
);
let expected = expected_absolute_path(&skill_path);
assert_eq!(path, expected);
}
#[test]
fn app_mentioned_event_serializes_expected_shape() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let event = TrackEventRequest::AppMentioned(CodexAppMentionedEventRequest {
event_type: "codex_app_mentioned",
event_params: codex_app_metadata(
&tracking,
AppInvocation {
connector_id: Some("calendar".to_string()),
app_name: Some("Calendar".to_string()),
invocation_type: Some(InvocationType::Explicit),
},
),
});
let payload = serde_json::to_value(&event).expect("serialize app mentioned event");
assert_eq!(
payload,
json!({
"event_type": "codex_app_mentioned",
"event_params": {
"connector_id": "calendar",
"thread_id": "thread-1",
"turn_id": "turn-1",
"app_name": "Calendar",
"product_client_id": originator().value,
"invoke_type": "explicit",
"model_slug": "gpt-5"
}
})
);
}
#[test]
fn app_used_event_serializes_expected_shape() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
};
let event = TrackEventRequest::AppUsed(CodexAppUsedEventRequest {
event_type: "codex_app_used",
event_params: codex_app_metadata(
&tracking,
AppInvocation {
connector_id: Some("drive".to_string()),
app_name: Some("Google Drive".to_string()),
invocation_type: Some(InvocationType::Implicit),
},
),
});
let payload = serde_json::to_value(&event).expect("serialize app used event");
assert_eq!(
payload,
json!({
"event_type": "codex_app_used",
"event_params": {
"connector_id": "drive",
"thread_id": "thread-2",
"turn_id": "turn-2",
"app_name": "Google Drive",
"product_client_id": originator().value,
"invoke_type": "implicit",
"model_slug": "gpt-5"
}
})
);
}
#[test]
fn accepted_line_fingerprints_event_serializes_expected_shape() {
let event = TrackEventRequest::AcceptedLineFingerprints(Box::new(
CodexAcceptedLineFingerprintsEventRequest {
event_type: "codex_accepted_line_fingerprints",
event_params: CodexAcceptedLineFingerprintsEventParams {
event_type: "codex.accepted_line_fingerprints",
turn_id: "turn-1".to_string(),
thread_id: "thread-1".to_string(),
product_surface: Some("codex".to_string()),
model_slug: Some("gpt-5.1-codex".to_string()),
completed_at: 1710000000,
repo_hash: Some("repo-hash-1".to_string()),
accepted_added_lines: 42,
accepted_deleted_lines: 40,
line_fingerprints: Vec::new(),
},
},
));
let payload = serde_json::to_value(&event).expect("serialize accepted line fingerprints event");
assert_eq!(
payload,
json!({
"event_type": "codex_accepted_line_fingerprints",
"event_params": {
"event_type": "codex.accepted_line_fingerprints",
"turn_id": "turn-1",
"thread_id": "thread-1",
"product_surface": "codex",
"model_slug": "gpt-5.1-codex",
"completed_at": 1710000000,
"repo_hash": "repo-hash-1",
"accepted_added_lines": 42,
"accepted_deleted_lines": 40,
"line_fingerprints": []
}
})
);
}
#[tokio::test]
async fn reducer_emits_large_accepted_line_aggregates_without_fingerprints() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut events,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
/*include_token_usage*/ true,
)
.await;
events.clear();
let mut diff = "\
diff --git a/src/lib.rs b/src/lib.rs
index 1111111..2222222
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -0,0 +1,20000 @@
"
.to_string();
for index in 0..20_000 {
diff.push_str(&format!("+let value_{index} = {index};\n"));
}
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::TurnDiffUpdated(
TurnDiffUpdatedNotification {
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
diff,
},
))),
&mut events,
)
.await;
assert!(events.is_empty());
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut events,
)
.await;
let accepted_line_events = events
.iter()
.filter_map(|event| match event {
TrackEventRequest::AcceptedLineFingerprints(event) => Some(event),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(accepted_line_events.len(), 1);
let event = accepted_line_events[0];
assert_eq!(event.event_params.turn_id, "turn-2");
assert_eq!(event.event_params.thread_id, "thread-2");
assert_eq!(event.event_params.accepted_added_lines, 20_000);
assert_eq!(event.event_params.accepted_deleted_lines, 0);
assert!(event.event_params.line_fingerprints.is_empty());
assert!(serde_json::to_vec(event).expect("serialize event").len() < 2_100_000);
}
#[tokio::test]
async fn reducer_emits_accepted_line_fingerprints_once_from_latest_turn_diff_on_completion() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut events,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
/*include_token_usage*/ true,
)
.await;
events.clear();
for line in ["let old_value = 1;", "let latest_value = 2;"] {
let diff = format!(
"\
diff --git a/src/lib.rs b/src/lib.rs
index 1111111..2222222
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -0,0 +1 @@
+{line}
"
);
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::TurnDiffUpdated(
TurnDiffUpdatedNotification {
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
diff,
},
))),
&mut events,
)
.await;
}
assert!(events.is_empty());
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut events,
)
.await;
let accepted_line_events = events
.iter()
.filter_map(|event| match event {
TrackEventRequest::AcceptedLineFingerprints(event) => Some(event),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(accepted_line_events.len(), 1);
let event = accepted_line_events[0];
assert_eq!(event.event_params.accepted_added_lines, 1);
assert!(event.event_params.line_fingerprints.is_empty());
}
#[test]
fn compaction_event_serializes_expected_shape() {
let event = TrackEventRequest::Compaction(Box::new(CodexCompactionEventRequest {
event_type: "codex_compaction_event",
event_params: crate::events::codex_compaction_event_params(
CodexCompactionEvent {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
trigger: CompactionTrigger::Auto,
reason: CompactionReason::ContextLimit,
implementation: CompactionImplementation::ResponsesCompact,
phase: CompactionPhase::MidTurn,
strategy: CompactionStrategy::Memento,
status: CompactionStatus::Completed,
error: None,
active_context_tokens_before: 120_000,
active_context_tokens_after: 18_000,
retained_image_count: None,
compaction_summary_tokens: None,
started_at: 100,
completed_at: 106,
duration_ms: Some(6543),
},
"session-thread-1".to_string(),
sample_app_server_client_metadata(),
sample_runtime_metadata(),
Some(ThreadSource::User),
/*subagent_source*/ None,
/*parent_thread_id*/ None,
),
}));
let payload = serde_json::to_value(&event).expect("serialize compaction event");
assert_eq!(
payload,
json!({
"event_type": "codex_compaction_event",
"event_params": {
"thread_id": "thread-1",
"session_id": "session-thread-1",
"turn_id": "turn-1",
"app_server_client": {
"product_client_id": DEFAULT_ORIGINATOR,
"client_name": "codex-tui",
"client_version": "1.0.0",
"rpc_transport": "stdio",
"experimental_api_enabled": true
},
"runtime": {
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64"
},
"thread_source": "user",
"subagent_source": null,
"parent_thread_id": null,
"trigger": "auto",
"reason": "context_limit",
"implementation": "responses_compact",
"phase": "mid_turn",
"strategy": "memento",
"status": "completed",
"error": null,
"active_context_tokens_before": 120000,
"active_context_tokens_after": 18000,
"retained_image_count": null,
"compaction_summary_tokens": null,
"started_at": 100,
"completed_at": 106,
"duration_ms": 6543
}
})
);
}
#[test]
fn compaction_implementation_serializes_remote_v2() {
let payload = serde_json::to_value(CompactionImplementation::ResponsesCompactionV2)
.expect("serialize compaction implementation");
assert_eq!(payload, json!("responses_compaction_v2"));
}
#[test]
fn app_used_dedupe_is_keyed_by_turn_and_connector() {
let (sender, _receiver) = mpsc::channel(1);
let queue = AnalyticsEventsQueue {
sender,
app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
};
let app = AppInvocation {
connector_id: Some("calendar".to_string()),
app_name: Some("Calendar".to_string()),
invocation_type: Some(InvocationType::Implicit),
};
let turn_1 = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let turn_2 = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-2".to_string(),
};
assert_eq!(queue.should_enqueue_app_used(&turn_1, &app), true);
assert_eq!(queue.should_enqueue_app_used(&turn_1, &app), false);
assert_eq!(queue.should_enqueue_app_used(&turn_2, &app), true);
}
#[test]
fn thread_initialized_event_serializes_expected_shape() {
let event = TrackEventRequest::ThreadInitialized(ThreadInitializedEvent {
event_type: "codex_thread_initialized",
event_params: ThreadInitializedEventParams {
thread_id: "thread-0".to_string(),
session_id: "session-thread-0".to_string(),
app_server_client: CodexAppServerClientMetadata {
product_client_id: DEFAULT_ORIGINATOR.to_string(),
client_name: Some("codex-tui".to_string()),
client_version: Some("1.0.0".to_string()),
rpc_transport: AppServerRpcTransport::Stdio,
experimental_api_enabled: Some(true),
},
runtime: CodexRuntimeMetadata {
codex_rs_version: "0.1.0".to_string(),
runtime_os: "macos".to_string(),
runtime_os_version: "15.3.1".to_string(),
runtime_arch: "aarch64".to_string(),
},
model: "gpt-5".to_string(),
ephemeral: true,
thread_source: Some(ThreadSource::Feature("automation".to_string())),
initialization_mode: ThreadInitializationMode::New,
subagent_source: None,
parent_thread_id: None,
forked_from_thread_id: None,
created_at: 1,
},
});
let payload = serde_json::to_value(&event).expect("serialize thread initialized event");
assert_eq!(
payload,
json!({
"event_type": "codex_thread_initialized",
"event_params": {
"thread_id": "thread-0",
"session_id": "session-thread-0",
"app_server_client": {
"product_client_id": DEFAULT_ORIGINATOR,
"client_name": "codex-tui",
"client_version": "1.0.0",
"rpc_transport": "stdio",
"experimental_api_enabled": true
},
"runtime": {
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64"
},
"model": "gpt-5",
"ephemeral": true,
"thread_source": "automation",
"initialization_mode": "new",
"subagent_source": null,
"parent_thread_id": null,
"forked_from_thread_id": null,
"created_at": 1
}
})
);
}
#[test]
fn command_execution_event_serializes_expected_shape() {
let event = TrackEventRequest::CommandExecution(CodexCommandExecutionEventRequest {
event_type: "codex_command_execution_event",
event_params: CodexCommandExecutionEventParams {
base: CodexToolItemEventBase {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
item_id: "item-1".to_string(),
app_server_client: CodexAppServerClientMetadata {
product_client_id: "codex_tui".to_string(),
client_name: Some("codex-tui".to_string()),
client_version: Some("1.2.3".to_string()),
rpc_transport: AppServerRpcTransport::Websocket,
experimental_api_enabled: Some(true),
},
runtime: CodexRuntimeMetadata {
codex_rs_version: "0.99.0".to_string(),
runtime_os: "macos".to_string(),
runtime_os_version: "15.3.1".to_string(),
runtime_arch: "aarch64".to_string(),
},
thread_source: Some(ThreadSource::User),
subagent_source: None,
parent_thread_id: None,
tool_name: "shell".to_string(),
started_at_ms: 123_000,
completed_at_ms: 125_000,
duration_ms: Some(2000),
execution_duration_ms: Some(1900),
review_count: 0,
guardian_review_count: 0,
user_review_count: 0,
final_approval_outcome: FinalApprovalOutcome::NotNeeded,
terminal_status: ToolItemTerminalStatus::Completed,
failure_kind: None,
requested_additional_permissions: false,
requested_network_access: false,
},
command_execution_source: CommandExecutionSource::Agent,
exit_code: Some(0),
command_total_action_count: 4,
command_read_action_count: 1,
command_list_files_action_count: 1,
command_search_action_count: 1,
command_unknown_action_count: 1,
},
});
let payload = serde_json::to_value(&event).expect("serialize command execution event");
assert_eq!(
payload,
json!({
"event_type": "codex_command_execution_event",
"event_params": {
"thread_id": "thread-1",
"turn_id": "turn-1",
"item_id": "item-1",
"app_server_client": {
"product_client_id": "codex_tui",
"client_name": "codex-tui",
"client_version": "1.2.3",
"rpc_transport": "websocket",
"experimental_api_enabled": true
},
"runtime": {
"codex_rs_version": "0.99.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64"
},
"thread_source": "user",
"subagent_source": null,
"parent_thread_id": null,
"tool_name": "shell",
"started_at_ms": 123000,
"completed_at_ms": 125000,
"duration_ms": 2000,
"execution_duration_ms": 1900,
"review_count": 0,
"guardian_review_count": 0,
"user_review_count": 0,
"final_approval_outcome": "not_needed",
"terminal_status": "completed",
"failure_kind": null,
"requested_additional_permissions": false,
"requested_network_access": false,
"command_execution_source": "agent",
"exit_code": 0,
"command_total_action_count": 4,
"command_read_action_count": 1,
"command_list_files_action_count": 1,
"command_search_action_count": 1,
"command_unknown_action_count": 1
}
})
);
}
#[test]
fn review_event_serializes_expected_shape() {
let event = TrackEventRequest::ReviewEvent(CodexReviewEventRequest {
event_type: "codex_review_event",
event_params: CodexReviewEventParams {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
item_id: None,
review_id: "review-1".to_string(),
app_server_client: CodexAppServerClientMetadata {
product_client_id: "codex_tui".to_string(),
client_name: Some("codex-tui".to_string()),
client_version: Some("1.2.3".to_string()),
rpc_transport: AppServerRpcTransport::Websocket,
experimental_api_enabled: Some(true),
},
runtime: CodexRuntimeMetadata {
codex_rs_version: "0.99.0".to_string(),
runtime_os: "macos".to_string(),
runtime_os_version: "15.3.1".to_string(),
runtime_arch: "aarch64".to_string(),
},
thread_source: Some(ThreadSource::Subagent),
subagent_source: Some("thread_spawn".to_string()),
parent_thread_id: Some("parent-thread-1".to_string()),
subject_kind: ReviewSubjectKind::NetworkAccess,
subject_name: "network_access".to_string(),
reviewer: Reviewer::User,
trigger: ReviewTrigger::NetworkPolicyDenial,
status: ReviewStatus::Approved,
resolution: ReviewResolution::NetworkPolicyAmendment,
started_at_ms: 123,
completed_at_ms: 125,
duration_ms: Some(2),
},
});
let payload = serde_json::to_value(&event).expect("serialize review event");
assert_eq!(
payload,
json!({
"event_type": "codex_review_event",
"event_params": {
"thread_id": "thread-1",
"turn_id": "turn-1",
"item_id": null,
"review_id": "review-1",
"app_server_client": {
"product_client_id": "codex_tui",
"client_name": "codex-tui",
"client_version": "1.2.3",
"rpc_transport": "websocket",
"experimental_api_enabled": true
},
"runtime": {
"codex_rs_version": "0.99.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64"
},
"thread_source": "subagent",
"subagent_source": "thread_spawn",
"parent_thread_id": "parent-thread-1",
"subject_kind": "network_access",
"subject_name": "network_access",
"reviewer": "user",
"trigger": "network_policy_denial",
"status": "approved",
"resolution": "network_policy_amendment",
"started_at_ms": 123,
"completed_at_ms": 125,
"duration_ms": 2
}
})
);
}
#[tokio::test]
async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialized() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(1),
response: Box::new(sample_thread_start_response(
"thread-no-client",
/*ephemeral*/ false,
"gpt-5",
)),
},
&mut events,
)
.await;
assert!(events.is_empty(), "thread events should require initialize");
reducer
.ingest(
AnalyticsFact::Initialize {
connection_id: 7,
params: InitializeParams {
client_info: ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: Some(InitializeCapabilities {
experimental_api: false,
request_attestation: false,
opt_out_notification_methods: None,
}),
},
product_client_id: DEFAULT_ORIGINATOR.to_string(),
runtime: CodexRuntimeMetadata {
codex_rs_version: "0.99.0".to_string(),
runtime_os: "linux".to_string(),
runtime_os_version: "24.04".to_string(),
runtime_arch: "x86_64".to_string(),
},
rpc_transport: AppServerRpcTransport::Websocket,
},
&mut events,
)
.await;
assert!(events.is_empty(), "initialize should not publish by itself");
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(2),
response: Box::new(sample_thread_resume_response(
"thread-1", /*ephemeral*/ true, "gpt-5",
)),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_thread_initialized");
assert_eq!(payload[0]["event_params"]["session_id"], "session-thread-1");
assert_eq!(
payload[0]["event_params"]["app_server_client"]["product_client_id"],
DEFAULT_ORIGINATOR
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["client_name"],
"codex-tui"
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["client_version"],
"1.0.0"
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["rpc_transport"],
"websocket"
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["experimental_api_enabled"],
false
);
assert_eq!(
payload[0]["event_params"]["runtime"]["codex_rs_version"],
"0.99.0"
);
assert_eq!(payload[0]["event_params"]["runtime"]["runtime_os"], "linux");
assert_eq!(
payload[0]["event_params"]["runtime"]["runtime_os_version"],
"24.04"
);
assert_eq!(
payload[0]["event_params"]["runtime"]["runtime_arch"],
"x86_64"
);
}
#[tokio::test]
async fn unrelated_client_requests_are_ignored_by_reducer() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
reducer
.ingest(
AnalyticsFact::ClientRequest {
connection_id: 7,
request_id: RequestId::Integer(3),
request: Box::new(ClientRequest::ThreadArchive {
request_id: RequestId::Integer(3),
params: ThreadArchiveParams {
thread_id: "thread-2".to_string(),
},
}),
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(3),
response: Box::new(sample_turn_start_response("turn-2")),
},
&mut events,
)
.await;
assert!(
events.is_empty(),
"unrelated requests must not create pending turn state"
);
}
#[tokio::test]
async fn unrelated_client_responses_are_ignored_by_reducer() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_initialize(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(9),
response: Box::new(ClientResponsePayload::ThreadArchive(
ThreadArchiveResponse {},
)),
},
&mut events,
)
.await;
assert!(events.is_empty());
}
#[tokio::test]
async fn compaction_event_ingests_custom_fact() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
let parent_thread_id =
codex_protocol::ThreadId::from_string("22222222-2222-2222-2222-222222222222")
.expect("valid parent thread id");
reducer
.ingest(
AnalyticsFact::Initialize {
connection_id: 7,
params: InitializeParams {
client_info: ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: Some(InitializeCapabilities {
experimental_api: false,
request_attestation: false,
opt_out_notification_methods: None,
}),
},
product_client_id: DEFAULT_ORIGINATOR.to_string(),
runtime: sample_runtime_metadata(),
rpc_transport: AppServerRpcTransport::Websocket,
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(2),
response: Box::new(sample_thread_resume_response_with_source(
"thread-1",
/*ephemeral*/ false,
"gpt-5",
AppServerSessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
agent_path: None,
agent_nickname: None,
agent_role: None,
}),
Some(AppServerThreadSource::Subagent),
Some(parent_thread_id.to_string()),
)),
},
&mut events,
)
.await;
events.clear();
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(Box::new(
CodexCompactionEvent {
thread_id: "thread-1".to_string(),
turn_id: "turn-compact".to_string(),
trigger: CompactionTrigger::Manual,
reason: CompactionReason::UserRequested,
implementation: CompactionImplementation::Responses,
phase: CompactionPhase::StandaloneTurn,
strategy: CompactionStrategy::Memento,
status: CompactionStatus::Failed,
error: Some("context limit exceeded".to_string()),
active_context_tokens_before: 131_000,
active_context_tokens_after: 131_000,
retained_image_count: None,
compaction_summary_tokens: None,
started_at: 100,
completed_at: 101,
duration_ms: Some(1200),
},
))),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_compaction_event");
assert_eq!(payload[0]["event_params"]["session_id"], "session-thread-1");
assert_eq!(payload[0]["event_params"]["thread_id"], "thread-1");
assert_eq!(payload[0]["event_params"]["turn_id"], "turn-compact");
assert_eq!(
payload[0]["event_params"]["app_server_client"]["product_client_id"],
DEFAULT_ORIGINATOR
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["client_name"],
"codex-tui"
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["rpc_transport"],
"websocket"
);
assert_eq!(
payload[0]["event_params"]["runtime"]["codex_rs_version"],
"0.1.0"
);
assert_eq!(payload[0]["event_params"]["thread_source"], "subagent");
assert_eq!(
payload[0]["event_params"]["subagent_source"],
"thread_spawn"
);
assert_eq!(
payload[0]["event_params"]["parent_thread_id"],
"22222222-2222-2222-2222-222222222222"
);
assert_eq!(payload[0]["event_params"]["trigger"], "manual");
assert_eq!(payload[0]["event_params"]["reason"], "user_requested");
assert_eq!(payload[0]["event_params"]["implementation"], "responses");
assert_eq!(payload[0]["event_params"]["phase"], "standalone_turn");
assert_eq!(payload[0]["event_params"]["strategy"], "memento");
assert_eq!(payload[0]["event_params"]["status"], "failed");
}
#[tokio::test]
async fn guardian_review_event_ingests_custom_fact_with_optional_target_item() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
reducer
.ingest(
AnalyticsFact::Initialize {
connection_id: 7,
params: InitializeParams {
client_info: ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: Some(InitializeCapabilities {
experimental_api: false,
request_attestation: false,
opt_out_notification_methods: None,
}),
},
product_client_id: DEFAULT_ORIGINATOR.to_string(),
runtime: sample_runtime_metadata(),
rpc_transport: AppServerRpcTransport::Websocket,
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(1),
response: Box::new(sample_thread_start_response(
"thread-guardian",
/*ephemeral*/ false,
"gpt-5",
)),
},
&mut events,
)
.await;
events.clear();
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::GuardianReview(Box::new(
GuardianReviewEventParams {
thread_id: "thread-guardian".to_string(),
turn_id: "turn-guardian".to_string(),
review_id: "review-guardian".to_string(),
target_item_id: None,
approval_request_source: GuardianApprovalRequestSource::DelegatedSubagent,
reviewed_action: GuardianReviewedAction::NetworkAccess {
protocol: NetworkApprovalProtocol::Https,
port: 443,
},
reviewed_action_truncated: false,
decision: GuardianReviewDecision::Denied,
terminal_status: GuardianReviewTerminalStatus::TimedOut,
failure_reason: Some(GuardianReviewFailureReason::Timeout),
attempt_count: 1,
risk_level: None,
user_authorization: None,
outcome: None,
guardian_thread_id: None,
guardian_session_kind: None,
guardian_model: None,
guardian_reasoning_effort: None,
had_prior_review_context: None,
review_timeout_ms: 90_000,
tool_call_count: None,
time_to_first_token_ms: None,
completion_latency_ms: Some(90_000),
started_at: 100,
completed_at: Some(190),
input_tokens: None,
cached_input_tokens: None,
output_tokens: None,
reasoning_output_tokens: None,
total_tokens: None,
},
))),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_guardian_review");
assert_eq!(
payload[0]["event_params"]["session_id"],
"session-thread-guardian"
);
assert_eq!(payload[0]["event_params"]["thread_id"], "thread-guardian");
assert_eq!(payload[0]["event_params"]["turn_id"], "turn-guardian");
assert_eq!(payload[0]["event_params"]["review_id"], "review-guardian");
assert_eq!(payload[0]["event_params"]["target_item_id"], json!(null));
assert_eq!(
payload[0]["event_params"]["approval_request_source"],
"delegated_subagent"
);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["product_client_id"],
DEFAULT_ORIGINATOR
);
assert_eq!(
payload[0]["event_params"]["runtime"]["codex_rs_version"],
"0.1.0"
);
assert_eq!(
payload[0]["event_params"]["reviewed_action"]["type"],
"network_access"
);
assert_eq!(
payload[0]["event_params"]["reviewed_action"]["protocol"],
"https"
);
assert_eq!(payload[0]["event_params"]["reviewed_action"]["port"], 443);
assert!(payload[0]["event_params"].get("retry_reason").is_none());
assert!(payload[0]["event_params"].get("rationale").is_none());
assert!(
payload[0]["event_params"]["reviewed_action"]
.get("target")
.is_none()
);
assert!(
payload[0]["event_params"]["reviewed_action"]
.get("host")
.is_none()
);
assert_eq!(payload[0]["event_params"]["terminal_status"], "timed_out");
assert_eq!(payload[0]["event_params"]["failure_reason"], "timeout");
assert_eq!(payload[0]["event_params"]["attempt_count"], 1);
assert_eq!(payload[0]["event_params"]["review_timeout_ms"], 90_000);
}
#[tokio::test]
async fn item_lifecycle_notifications_publish_command_execution_event() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_started_notification(
"thread-1", "turn-1",
))),
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted(
ItemStartedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
started_at_ms: 1_000,
item: sample_command_execution_item(
CommandExecutionStatus::InProgress,
/*exit_code*/ None,
/*duration_ms*/ None,
),
},
))),
&mut events,
)
.await;
assert!(
events.is_empty(),
"tool item event should emit on completion"
);
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::ItemCompleted(
ItemCompletedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
completed_at_ms: 1_045,
item: sample_command_execution_item_with_actions(
CommandExecutionStatus::Completed,
Some(0),
Some(42),
vec![
CommandAction::Read {
command: "cat README.md".to_string(),
name: "README.md".to_string(),
path: test_path_buf("/tmp/README.md").abs(),
},
CommandAction::ListFiles {
command: "ls".to_string(),
path: None,
},
CommandAction::Search {
command: "rg TODO".to_string(),
query: Some("TODO".to_string()),
path: None,
},
CommandAction::Unknown {
command: "cargo test".to_string(),
},
],
),
},
))),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_command_execution_event");
assert_eq!(payload[0]["event_params"]["thread_id"], "thread-1");
assert_eq!(payload[0]["event_params"]["turn_id"], "turn-1");
assert_eq!(payload[0]["event_params"]["item_id"], "item-1");
assert_eq!(payload[0]["event_params"]["tool_name"], "shell");
assert_eq!(
payload[0]["event_params"]["command_execution_source"],
"agent"
);
assert_eq!(payload[0]["event_params"]["terminal_status"], "completed");
assert_eq!(
payload[0]["event_params"]["final_approval_outcome"],
"unknown"
);
assert_eq!(
payload[0]["event_params"]["failure_kind"],
serde_json::Value::Null
);
assert_eq!(payload[0]["event_params"]["exit_code"], 0);
assert_eq!(payload[0]["event_params"]["command_total_action_count"], 4);
assert_eq!(payload[0]["event_params"]["command_read_action_count"], 1);
assert_eq!(
payload[0]["event_params"]["command_list_files_action_count"],
1
);
assert_eq!(payload[0]["event_params"]["command_search_action_count"], 1);
assert_eq!(
payload[0]["event_params"]["command_unknown_action_count"],
1
);
assert_eq!(payload[0]["event_params"]["started_at_ms"], 1_000);
assert_eq!(payload[0]["event_params"]["completed_at_ms"], 1_045);
assert_eq!(payload[0]["event_params"]["duration_ms"], 45);
assert_eq!(payload[0]["event_params"]["execution_duration_ms"], 42);
assert_eq!(
payload[0]["event_params"]["app_server_client"]["client_name"],
"codex-tui"
);
assert_eq!(payload[0]["event_params"]["thread_source"], "user");
}
#[tokio::test]
async fn command_execution_approval_response_publishes_user_review_event() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_command_approval_request(
/*request_id*/ 41, /*approval_id*/ None,
)),
},
&mut events,
)
.await;
assert!(events.is_empty());
reducer
.ingest(
AnalyticsFact::ServerResponse {
completed_at_ms: 1_042,
response: Box::new(sample_command_approval_response(
/*request_id*/ 41,
CommandExecutionApprovalDecision::Accept,
)),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_review_event");
assert_eq!(payload[0]["event_params"]["thread_id"], "thread-1");
assert_eq!(payload[0]["event_params"]["turn_id"], "turn-1");
assert_eq!(payload[0]["event_params"]["item_id"], "item-1");
assert_eq!(payload[0]["event_params"]["review_id"], "user:41");
assert_eq!(payload[0]["event_params"]["thread_source"], "user");
assert_eq!(
payload[0]["event_params"]["subject_kind"],
"command_execution"
);
assert_eq!(
payload[0]["event_params"]["subject_name"],
"command_execution"
);
assert_eq!(payload[0]["event_params"]["reviewer"], "user");
assert_eq!(payload[0]["event_params"]["trigger"], "initial");
assert_eq!(payload[0]["event_params"]["status"], "approved");
assert_eq!(payload[0]["event_params"]["started_at_ms"], 1_000);
assert_eq!(payload[0]["event_params"]["completed_at_ms"], 1_042);
assert_eq!(payload[0]["event_params"]["duration_ms"], 42);
}
#[tokio::test]
async fn permissions_reviews_emit_events_without_denormalizing_onto_tool_items() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_permissions_approval_request(/*request_id*/ 51)),
},
&mut events,
)
.await;
assert!(events.is_empty());
reducer
.ingest(
AnalyticsFact::EffectivePermissionsApprovalResponse {
completed_at_ms: 1_042,
request_id: RequestId::Integer(51),
response: Box::new(sample_effective_permissions_approval_response(
CoreRequestPermissionProfile::default(),
CorePermissionGrantScope::Turn,
)),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_review_event");
assert_eq!(payload[0]["event_params"]["review_id"], "user:51");
assert_eq!(payload[0]["event_params"]["subject_kind"], "permissions");
assert_eq!(payload[0]["event_params"]["reviewer"], "user");
assert_eq!(payload[0]["event_params"]["status"], "denied");
assert_eq!(payload[0]["event_params"]["resolution"], "none");
events.clear();
ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-1", "permissions-1")
.await;
let payload = serde_json::to_value(&events[0]).expect("serialize tool item event");
assert_eq!(payload["event_params"]["item_id"], "permissions-1");
assert_eq!(payload["event_params"]["review_count"], 0);
assert_eq!(payload["event_params"]["user_review_count"], 0);
assert_eq!(payload["event_params"]["guardian_review_count"], 0);
}
#[tokio::test]
async fn effective_session_permissions_response_publishes_session_user_review_event() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_permissions_approval_request(/*request_id*/ 52)),
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::EffectivePermissionsApprovalResponse {
completed_at_ms: 1_042,
request_id: RequestId::Integer(52),
response: Box::new(sample_effective_permissions_approval_response(
CoreRequestPermissionProfile {
network: Some(CoreNetworkPermissions {
enabled: Some(true),
}),
file_system: None,
},
CorePermissionGrantScope::Session,
)),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_review_event");
assert_eq!(payload[0]["event_params"]["review_id"], "user:52");
assert_eq!(payload[0]["event_params"]["subject_kind"], "permissions");
assert_eq!(payload[0]["event_params"]["reviewer"], "user");
assert_eq!(payload[0]["event_params"]["status"], "approved");
assert_eq!(payload[0]["event_params"]["resolution"], "session_approval");
}
#[tokio::test]
async fn aborted_server_request_publishes_aborted_user_review_event_once() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_command_approval_request(
/*request_id*/ 61, /*approval_id*/ None,
)),
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ServerRequestAborted {
completed_at_ms: 1_042,
request_id: RequestId::Integer(61),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_params"]["review_id"], "user:61");
assert_eq!(payload[0]["event_params"]["status"], "aborted");
assert_eq!(payload[0]["event_params"]["resolution"], "none");
events.clear();
reducer
.ingest(
AnalyticsFact::ServerResponse {
completed_at_ms: 1_043,
response: Box::new(sample_command_approval_response(
/*request_id*/ 61,
CommandExecutionApprovalDecision::Accept,
)),
},
&mut events,
)
.await;
assert!(events.is_empty());
}
#[tokio::test]
async fn guardian_completed_notification_publishes_review_event_with_thread_metadata() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_guardian_review_completed(
"guardian-review-1",
Some("item-1"),
GuardianApprovalReviewStatus::Denied,
))),
&mut events,
)
.await;
let payload = serde_json::to_value(&events[0]).expect("serialize review event");
assert_eq!(payload["event_type"], "codex_review_event");
assert_eq!(payload["event_params"]["review_id"], "guardian-review-1");
assert_eq!(payload["event_params"]["item_id"], "item-1");
assert_eq!(payload["event_params"]["thread_source"], "user");
assert_eq!(payload["event_params"]["subject_kind"], "command_execution");
assert_eq!(payload["event_params"]["reviewer"], "guardian");
assert_eq!(payload["event_params"]["status"], "denied");
assert_eq!(payload["event_params"]["started_at_ms"], 1_000);
assert_eq!(payload["event_params"]["completed_at_ms"], 1_042);
assert_eq!(payload["event_params"]["duration_ms"], 42);
}
#[tokio::test]
async fn terminal_reviews_denormalize_counts_onto_tool_item_events() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_command_approval_request(
/*request_id*/ 71, /*approval_id*/ None,
)),
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ServerResponse {
completed_at_ms: 1_042,
response: Box::new(sample_command_approval_response(
/*request_id*/ 71,
CommandExecutionApprovalDecision::AcceptForSession,
)),
},
&mut events,
)
.await;
events.clear();
ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-1", "item-1").await;
let payload = serde_json::to_value(&events[0]).expect("serialize tool item event");
assert_eq!(payload["event_params"]["review_count"], 1);
assert_eq!(payload["event_params"]["user_review_count"], 1);
assert_eq!(payload["event_params"]["guardian_review_count"], 0);
assert_eq!(
payload["event_params"]["final_approval_outcome"],
"user_approved_for_session"
);
}
#[tokio::test]
async fn item_review_summaries_do_not_cross_threads_with_reused_item_ids() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(2),
response: Box::new(sample_thread_start_response(
"thread-2", /*ephemeral*/ false, "gpt-5",
)),
},
&mut events,
)
.await;
events.clear();
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_command_approval_request(
/*request_id*/ 72, /*approval_id*/ None,
)),
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ServerResponse {
completed_at_ms: 1_042,
response: Box::new(sample_command_approval_response(
/*request_id*/ 72,
CommandExecutionApprovalDecision::Accept,
)),
},
&mut events,
)
.await;
events.clear();
ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-2", "item-1").await;
let payload = serde_json::to_value(&events[0]).expect("serialize tool item event");
assert_eq!(payload["event_params"]["thread_id"], "thread-2");
assert_eq!(payload["event_params"]["item_id"], "item-1");
assert_eq!(payload["event_params"]["review_count"], 0);
assert_eq!(payload["event_params"]["user_review_count"], 0);
assert_eq!(payload["event_params"]["guardian_review_count"], 0);
assert_eq!(payload["event_params"]["final_approval_outcome"], "unknown");
}
#[test]
fn subagent_thread_started_review_serializes_expected_shape() {
let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request(
SubAgentThreadStartedInput {
session_id: "session-root".to_string(),
thread_id: "thread-review".to_string(),
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(),
model: "gpt-5".to_string(),
ephemeral: false,
subagent_source: SubAgentSource::Review,
created_at: 123,
},
));
let payload = serde_json::to_value(&event).expect("serialize review subagent event");
assert_eq!(payload["event_params"]["thread_source"], "subagent");
assert_eq!(
payload["event_params"]["app_server_client"]["product_client_id"],
"codex-tui"
);
assert_eq!(
payload["event_params"]["app_server_client"]["client_name"],
"codex-tui"
);
assert_eq!(
payload["event_params"]["app_server_client"]["client_version"],
"1.0.0"
);
assert_eq!(
payload["event_params"]["app_server_client"]["rpc_transport"],
"in_process"
);
assert_eq!(payload["event_params"]["created_at"], 123);
assert_eq!(payload["event_params"]["initialization_mode"], "new");
assert_eq!(payload["event_params"]["subagent_source"], "review");
assert_eq!(payload["event_params"]["parent_thread_id"], json!(null));
assert_eq!(
payload["event_params"]["forked_from_thread_id"],
json!(null)
);
}
#[test]
fn subagent_thread_started_thread_spawn_serializes_thread_lineage() {
let parent_thread_id =
codex_protocol::ThreadId::from_string("11111111-1111-1111-1111-111111111111")
.expect("valid thread id");
let forked_from_thread_id =
codex_protocol::ThreadId::from_string("22222222-2222-4222-8222-222222222222")
.expect("valid thread id");
let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request(
SubAgentThreadStartedInput {
session_id: "session-root".to_string(),
thread_id: "thread-spawn".to_string(),
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(),
model: "gpt-5".to_string(),
ephemeral: true,
subagent_source: SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
agent_path: None,
agent_nickname: None,
agent_role: None,
},
created_at: 124,
},
));
let payload = serde_json::to_value(&event).expect("serialize thread spawn subagent event");
assert_eq!(payload["event_params"]["thread_id"], "thread-spawn");
assert_eq!(payload["event_params"]["thread_source"], "subagent");
assert_eq!(payload["event_params"]["subagent_source"], "thread_spawn");
assert_eq!(
payload["event_params"]["parent_thread_id"],
"11111111-1111-1111-1111-111111111111"
);
assert_eq!(
payload["event_params"]["forked_from_thread_id"],
"22222222-2222-4222-8222-222222222222"
);
assert_eq!(payload["event_params"]["session_id"], "session-root");
}
#[test]
fn subagent_thread_started_memory_consolidation_serializes_expected_shape() {
let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request(
SubAgentThreadStartedInput {
session_id: "session-root".to_string(),
thread_id: "thread-memory".to_string(),
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(),
model: "gpt-5".to_string(),
ephemeral: false,
subagent_source: SubAgentSource::MemoryConsolidation,
created_at: 125,
},
));
let payload =
serde_json::to_value(&event).expect("serialize memory consolidation subagent event");
assert_eq!(
payload["event_params"]["subagent_source"],
"memory_consolidation"
);
assert_eq!(payload["event_params"]["parent_thread_id"], json!(null));
}
#[test]
fn subagent_thread_started_other_serializes_expected_shape() {
let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request(
SubAgentThreadStartedInput {
session_id: "session-root".to_string(),
thread_id: "thread-guardian".to_string(),
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(),
model: "gpt-5".to_string(),
ephemeral: false,
subagent_source: SubAgentSource::Other("guardian".to_string()),
created_at: 126,
},
));
let payload = serde_json::to_value(&event).expect("serialize other subagent event");
assert_eq!(payload["event_params"]["subagent_source"], "guardian");
assert_eq!(payload["event_params"]["parent_thread_id"], json!(null));
}
#[test]
fn subagent_thread_started_other_serializes_explicit_parent_thread_id() {
let parent_thread_id =
codex_protocol::ThreadId::from_string("33333333-3333-4333-8333-333333333333")
.expect("valid thread id");
let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request(
SubAgentThreadStartedInput {
session_id: "session-root".to_string(),
thread_id: "thread-guardian".to_string(),
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(),
model: "gpt-5".to_string(),
ephemeral: false,
subagent_source: SubAgentSource::Other("guardian".to_string()),
created_at: 126,
},
));
let payload = serde_json::to_value(&event).expect("serialize auto-review subagent event");
assert_eq!(payload["event_params"]["subagent_source"], "guardian");
assert_eq!(
payload["event_params"]["parent_thread_id"],
"33333333-3333-4333-8333-333333333333"
);
}
#[tokio::test]
async fn subagent_thread_started_publishes_without_initialize() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted(
SubAgentThreadStartedInput {
session_id: "session-root".to_string(),
thread_id: "thread-review".to_string(),
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(),
model: "gpt-5".to_string(),
ephemeral: false,
subagent_source: SubAgentSource::Review,
created_at: 127,
},
)),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_thread_initialized");
assert_eq!(
payload[0]["event_params"]["app_server_client"]["product_client_id"],
"codex-tui"
);
assert_eq!(payload[0]["event_params"]["thread_source"], "subagent");
assert_eq!(payload[0]["event_params"]["subagent_source"], "review");
}
#[tokio::test]
async fn subagent_thread_started_inherits_parent_connection_for_new_thread() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
let parent_thread_id =
codex_protocol::ThreadId::from_string("44444444-4444-4444-4444-444444444444")
.expect("valid parent thread id");
let parent_thread_id_string = parent_thread_id.to_string();
reducer
.ingest(
AnalyticsFact::Initialize {
connection_id: 7,
params: InitializeParams {
client_info: ClientInfo {
name: "parent-client".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: None,
},
product_client_id: "parent-client".to_string(),
runtime: sample_runtime_metadata(),
rpc_transport: AppServerRpcTransport::Stdio,
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(1),
response: Box::new(sample_thread_start_response(
&parent_thread_id_string,
/*ephemeral*/ false,
"gpt-5",
)),
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted(
SubAgentThreadStartedInput {
session_id: "session-root".to_string(),
thread_id: "thread-review".to_string(),
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(),
model: "gpt-5".to_string(),
ephemeral: false,
subagent_source: SubAgentSource::ThreadSpawn {
parent_thread_id,
depth: 1,
agent_path: None,
agent_nickname: None,
agent_role: None,
},
created_at: 130,
},
)),
&mut events,
)
.await;
events.clear();
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(Box::new(
CodexCompactionEvent {
thread_id: "thread-review".to_string(),
turn_id: "turn-compact".to_string(),
trigger: CompactionTrigger::Manual,
reason: CompactionReason::UserRequested,
implementation: CompactionImplementation::Responses,
phase: CompactionPhase::StandaloneTurn,
strategy: CompactionStrategy::Memento,
status: CompactionStatus::Completed,
error: None,
active_context_tokens_before: 131_000,
active_context_tokens_after: 64_000,
retained_image_count: None,
compaction_summary_tokens: None,
started_at: 100,
completed_at: 101,
duration_ms: Some(1200),
},
))),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload[0]["event_params"]["session_id"], "session-root");
assert_eq!(payload[0]["event_params"]["thread_id"], "thread-review");
assert_eq!(
payload[0]["event_params"]["app_server_client"]["product_client_id"],
"parent-client"
);
assert_eq!(
payload[0]["event_params"]["parent_thread_id"],
"44444444-4444-4444-4444-444444444444"
);
}
#[tokio::test]
async fn subagent_tool_items_inherit_parent_connection_metadata() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted(
SubAgentThreadStartedInput {
session_id: "session-root".to_string(),
thread_id: "thread-subagent".to_string(),
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(),
model: "gpt-5".to_string(),
ephemeral: false,
subagent_source: SubAgentSource::Review,
created_at: 128,
},
)),
&mut events,
)
.await;
events.clear();
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_started_notification(
"thread-subagent",
"turn-subagent",
))),
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted(
ItemStartedNotification {
thread_id: "thread-subagent".to_string(),
turn_id: "turn-subagent".to_string(),
started_at_ms: 1_000,
item: sample_command_execution_item(
CommandExecutionStatus::InProgress,
/*exit_code*/ None,
/*duration_ms*/ None,
),
},
))),
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::ItemCompleted(
ItemCompletedNotification {
thread_id: "thread-subagent".to_string(),
turn_id: "turn-subagent".to_string(),
completed_at_ms: 1_042,
item: sample_command_execution_item(
CommandExecutionStatus::Completed,
Some(0),
Some(42),
),
},
))),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_command_execution_event");
assert_eq!(payload[0]["event_params"]["thread_source"], "subagent");
assert_eq!(payload[0]["event_params"]["subagent_source"], "review");
assert_eq!(payload[0]["event_params"]["parent_thread_id"], "thread-1");
assert_eq!(
payload[0]["event_params"]["app_server_client"]["client_name"],
"codex-tui"
);
}
#[test]
fn plugin_used_event_serializes_expected_shape() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-3".to_string(),
turn_id: "turn-3".to_string(),
};
let event = TrackEventRequest::PluginUsed(CodexPluginUsedEventRequest {
event_type: "codex_plugin_used",
event_params: codex_plugin_used_metadata(&tracking, sample_plugin_metadata()),
});
let payload = serde_json::to_value(&event).expect("serialize plugin used event");
assert_eq!(
payload,
json!({
"event_type": "codex_plugin_used",
"event_params": {
"plugin_id": "sample@test",
"plugin_name": "sample",
"marketplace_name": "test",
"has_skills": true,
"mcp_server_count": 2,
"connector_ids": ["calendar", "drive"],
"product_client_id": originator().value,
"mcp_server_names": ["mcp-1", "mcp-2"],
"thread_id": "thread-3",
"turn_id": "turn-3",
"model_slug": "gpt-5"
}
})
);
}
#[test]
fn plugin_management_event_serializes_expected_shape() {
let event = TrackEventRequest::PluginInstalled(CodexPluginEventRequest {
event_type: "codex_plugin_installed",
event_params: codex_plugin_metadata(sample_plugin_metadata()),
});
let payload = serde_json::to_value(&event).expect("serialize plugin installed event");
assert_eq!(
payload,
json!({
"event_type": "codex_plugin_installed",
"event_params": {
"plugin_id": "sample@test",
"plugin_name": "sample",
"marketplace_name": "test",
"has_skills": true,
"mcp_server_count": 2,
"connector_ids": ["calendar", "drive"],
"product_client_id": originator().value
}
})
);
}
#[test]
fn plugin_management_event_can_use_remote_plugin_id_override() {
let mut plugin = sample_plugin_metadata();
plugin.remote_plugin_id = Some("plugins~Plugin_remote".to_string());
let event = TrackEventRequest::PluginInstalled(CodexPluginEventRequest {
event_type: "codex_plugin_installed",
event_params: codex_plugin_metadata(plugin),
});
let payload = serde_json::to_value(&event).expect("serialize plugin installed event");
assert_eq!(
payload["event_params"]["plugin_id"],
"plugins~Plugin_remote"
);
assert_eq!(payload["event_params"]["plugin_name"], "sample");
assert_eq!(payload["event_params"]["marketplace_name"], "test");
}
#[test]
fn hook_run_event_serializes_expected_shape() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-3".to_string(),
turn_id: "turn-3".to_string(),
};
let event = TrackEventRequest::HookRun(CodexHookRunEventRequest {
event_type: "codex_hook_run",
event_params: codex_hook_run_metadata(
&tracking,
HookRunFact {
event_name: HookEventName::PreToolUse,
hook_source: HookSource::User,
status: HookRunStatus::Completed,
},
),
});
let payload = serde_json::to_value(&event).expect("serialize hook run event");
assert_eq!(
payload,
json!({
"event_type": "codex_hook_run",
"event_params": {
"thread_id": "thread-3",
"turn_id": "turn-3",
"model_slug": "gpt-5",
"hook_name": "PreToolUse",
"hook_source": "user",
"status": "completed"
}
})
);
}
#[test]
fn hook_run_metadata_maps_sources_and_statuses() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let system = serde_json::to_value(codex_hook_run_metadata(
&tracking,
HookRunFact {
event_name: HookEventName::SessionStart,
hook_source: HookSource::System,
status: HookRunStatus::Completed,
},
))
.expect("serialize system hook");
let project = serde_json::to_value(codex_hook_run_metadata(
&tracking,
HookRunFact {
event_name: HookEventName::Stop,
hook_source: HookSource::Project,
status: HookRunStatus::Blocked,
},
))
.expect("serialize project hook");
let cloud_requirements = serde_json::to_value(codex_hook_run_metadata(
&tracking,
HookRunFact {
event_name: HookEventName::Stop,
hook_source: HookSource::CloudRequirements,
status: HookRunStatus::Blocked,
},
))
.expect("serialize cloud requirements hook");
let unknown = serde_json::to_value(codex_hook_run_metadata(
&tracking,
HookRunFact {
event_name: HookEventName::UserPromptSubmit,
hook_source: HookSource::Unknown,
status: HookRunStatus::Failed,
},
))
.expect("serialize unknown hook");
assert_eq!(system["hook_source"], "system");
assert_eq!(system["status"], "completed");
assert_eq!(project["hook_source"], "project");
assert_eq!(project["status"], "blocked");
assert_eq!(cloud_requirements["hook_source"], "cloud_requirements");
assert_eq!(cloud_requirements["status"], "blocked");
assert_eq!(unknown["hook_source"], "unknown");
assert_eq!(unknown["status"], "failed");
}
#[test]
fn hook_run_metadata_maps_stopped_status() {
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let stopped = serde_json::to_value(codex_hook_run_metadata(
&tracking,
HookRunFact {
event_name: HookEventName::Stop,
hook_source: HookSource::User,
status: HookRunStatus::Stopped,
},
))
.expect("serialize stopped hook");
assert_eq!(stopped["hook_source"], "user");
assert_eq!(stopped["status"], "stopped");
}
#[test]
fn plugin_used_dedupe_is_keyed_by_turn_and_plugin() {
let (sender, _receiver) = mpsc::channel(1);
let queue = AnalyticsEventsQueue {
sender,
app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
};
let plugin = sample_plugin_metadata();
let turn_1 = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let turn_2 = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-2".to_string(),
};
assert_eq!(queue.should_enqueue_plugin_used(&turn_1, &plugin), true);
assert_eq!(queue.should_enqueue_plugin_used(&turn_1, &plugin), false);
assert_eq!(queue.should_enqueue_plugin_used(&turn_2, &plugin), true);
}
#[tokio::test]
async fn reducer_ingests_skill_invoked_fact() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let skill_path = PathBuf::from("/Users/abc/.codex/skills/doc/SKILL.md");
let expected_skill_id = skill_id_for_local_skill(
/*repo_url*/ None,
/*repo_root*/ None,
skill_path.as_path(),
"doc",
);
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::SkillInvoked(SkillInvokedInput {
tracking,
invocations: vec![SkillInvocation {
skill_name: "doc".to_string(),
skill_scope: codex_protocol::protocol::SkillScope::User,
skill_path,
plugin_id: None,
invocation_type: InvocationType::Explicit,
}],
})),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(
payload,
json!([{
"event_type": "skill_invocation",
"skill_id": expected_skill_id,
"skill_name": "doc",
"event_params": {
"product_client_id": originator().value,
"skill_scope": "user",
"plugin_id": null,
"repo_url": null,
"thread_id": "thread-1",
"turn_id": "turn-1",
"invoke_type": "explicit",
"model_slug": "gpt-5"
}
}])
);
}
#[tokio::test]
async fn reducer_includes_plugin_id_for_plugin_skill_invocations() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let skill_path =
PathBuf::from("/Users/abc/.codex/plugins/cache/test/sample/skills/doc/SKILL.md");
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::SkillInvoked(SkillInvokedInput {
tracking,
invocations: vec![SkillInvocation {
skill_name: "sample:doc".to_string(),
skill_scope: codex_protocol::protocol::SkillScope::User,
skill_path,
plugin_id: Some("sample@test".to_string()),
invocation_type: InvocationType::Explicit,
}],
})),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(
payload[0]["event_params"]["plugin_id"],
json!("sample@test")
);
}
#[tokio::test]
async fn reducer_ingests_hook_run_fact() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::HookRun(HookRunInput {
tracking: TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
},
hook: HookRunFact {
event_name: HookEventName::PostToolUse,
hook_source: HookSource::Unknown,
status: HookRunStatus::Failed,
},
})),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_hook_run");
assert_eq!(payload[0]["event_params"]["hook_name"], "PostToolUse");
assert_eq!(payload[0]["event_params"]["hook_source"], "unknown");
assert_eq!(payload[0]["event_params"]["status"], "failed");
}
#[tokio::test]
async fn reducer_ingests_app_and_plugin_facts() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::AppMentioned(AppMentionedInput {
tracking: tracking.clone(),
mentions: vec![AppInvocation {
connector_id: Some("calendar".to_string()),
app_name: Some("Calendar".to_string()),
invocation_type: Some(InvocationType::Explicit),
}],
})),
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(AppUsedInput {
tracking: tracking.clone(),
app: AppInvocation {
connector_id: Some("drive".to_string()),
app_name: Some("Drive".to_string()),
invocation_type: Some(InvocationType::Implicit),
},
})),
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::PluginUsed(PluginUsedInput {
tracking,
plugin: sample_plugin_metadata(),
})),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 3);
assert_eq!(payload[0]["event_type"], "codex_app_mentioned");
assert_eq!(payload[1]["event_type"], "codex_app_used");
assert_eq!(payload[2]["event_type"], "codex_plugin_used");
}
#[tokio::test]
async fn reducer_ingests_plugin_state_changed_fact() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::PluginStateChanged(
PluginStateChangedInput {
plugin: sample_plugin_metadata(),
state: PluginState::Disabled,
},
)),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(
payload,
json!([{
"event_type": "codex_plugin_disabled",
"event_params": {
"plugin_id": "sample@test",
"plugin_name": "sample",
"marketplace_name": "test",
"has_skills": true,
"mcp_server_count": 2,
"connector_ids": ["calendar", "drive"],
"product_client_id": originator().value
}
}])
);
}
#[test]
fn turn_event_serializes_expected_shape() {
let event = TrackEventRequest::TurnEvent(Box::new(CodexTurnEventRequest {
event_type: "codex_turn_event",
event_params: crate::events::CodexTurnEventParams {
thread_id: "thread-2".to_string(),
session_id: "session-thread-2".to_string(),
turn_id: "turn-2".to_string(),
app_server_client: sample_app_server_client_metadata(),
runtime: sample_runtime_metadata(),
submission_type: None,
ephemeral: false,
thread_source: Some(ThreadSource::User),
initialization_mode: ThreadInitializationMode::New,
subagent_source: None,
parent_thread_id: None,
model: Some("gpt-5".to_string()),
model_provider: "openai".to_string(),
sandbox_policy: Some("read_only"),
reasoning_effort: Some("high".to_string()),
reasoning_summary: Some("detailed".to_string()),
service_tier: "flex".to_string(),
approval_policy: "on-request".to_string(),
approvals_reviewer: "auto_review".to_string(),
sandbox_network_access: true,
collaboration_mode: Some("plan"),
personality: Some("pragmatic".to_string()),
workspace_kind: Some("projectless".to_string()),
num_input_images: 2,
is_first_turn: true,
status: Some(TurnStatus::Completed),
turn_error: None,
codex_error_kind: None,
codex_error_http_status_code: None,
steer_count: Some(0),
total_tool_call_count: None,
shell_command_count: None,
file_change_count: None,
mcp_tool_call_count: None,
dynamic_tool_call_count: None,
subagent_tool_call_count: None,
web_search_count: None,
image_generation_count: None,
input_tokens: None,
cached_input_tokens: None,
output_tokens: None,
reasoning_output_tokens: None,
total_tokens: None,
before_first_sampling_ms: 100,
sampling_ms: 700,
between_sampling_overhead_ms: 50,
tool_blocking_ms: 250,
after_last_sampling_ms: 134,
sampling_request_count: 2,
sampling_retry_count: 1,
duration_ms: Some(1234),
started_at: Some(455),
completed_at: Some(456),
},
}));
let payload = serde_json::to_value(&event).expect("serialize turn event");
let expected = serde_json::from_str::<serde_json::Value>(
r#"{
"event_type": "codex_turn_event",
"event_params": {
"thread_id": "thread-2",
"session_id": "session-thread-2",
"turn_id": "turn-2",
"submission_type": null,
"app_server_client": {
"product_client_id": "codex_cli_rs",
"client_name": "codex-tui",
"client_version": "1.0.0",
"rpc_transport": "stdio",
"experimental_api_enabled": true
},
"runtime": {
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64"
},
"ephemeral": false,
"thread_source": "user",
"initialization_mode": "new",
"subagent_source": null,
"parent_thread_id": null,
"model": "gpt-5",
"model_provider": "openai",
"sandbox_policy": "read_only",
"reasoning_effort": "high",
"reasoning_summary": "detailed",
"service_tier": "flex",
"approval_policy": "on-request",
"approvals_reviewer": "auto_review",
"sandbox_network_access": true,
"collaboration_mode": "plan",
"personality": "pragmatic",
"workspace_kind": "projectless",
"num_input_images": 2,
"is_first_turn": true,
"status": "completed",
"turn_error": null,
"codex_error_kind": null,
"codex_error_http_status_code": null,
"steer_count": 0,
"total_tool_call_count": null,
"shell_command_count": null,
"file_change_count": null,
"mcp_tool_call_count": null,
"dynamic_tool_call_count": null,
"subagent_tool_call_count": null,
"web_search_count": null,
"image_generation_count": null,
"input_tokens": null,
"cached_input_tokens": null,
"output_tokens": null,
"reasoning_output_tokens": null,
"total_tokens": null,
"before_first_sampling_ms": 100,
"sampling_ms": 700,
"between_sampling_overhead_ms": 50,
"tool_blocking_ms": 250,
"after_last_sampling_ms": 134,
"sampling_request_count": 2,
"sampling_retry_count": 1,
"duration_ms": 1234,
"started_at": 455,
"completed_at": 456
}
}"#,
)
.expect("parse expected turn event");
assert_eq!(payload, expected);
}
#[tokio::test]
async fn accepted_turn_steer_emits_expected_event() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ false,
/*include_started*/ false,
/*include_token_usage*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientRequest {
connection_id: 7,
request_id: RequestId::Integer(4),
request: Box::new(sample_turn_steer_request(
"thread-2", "turn-2", /*request_id*/ 4,
)),
},
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(4),
response: Box::new(sample_turn_steer_response("turn-2")),
},
&mut out,
)
.await;
assert_eq!(out.len(), 1);
let payload = serde_json::to_value(&out[0]).expect("serialize turn steer event");
assert_eq!(payload["event_type"], json!("codex_turn_steer_event"));
assert_eq!(payload["event_params"]["thread_id"], json!("thread-2"));
assert_eq!(
payload["event_params"]["session_id"],
json!("session-thread-2")
);
assert_eq!(payload["event_params"]["expected_turn_id"], json!("turn-2"));
assert_eq!(payload["event_params"]["accepted_turn_id"], json!("turn-2"));
assert_eq!(payload["event_params"]["num_input_images"], json!(1));
assert_eq!(payload["event_params"]["result"], json!("accepted"));
assert_eq!(payload["event_params"]["rejection_reason"], json!(null));
assert!(
payload["event_params"]["created_at"]
.as_u64()
.expect("created_at")
> 0
);
assert_eq!(
payload["event_params"]["app_server_client"]["product_client_id"],
json!("codex-tui")
);
assert_eq!(
payload["event_params"]["runtime"]["codex_rs_version"],
json!("0.1.0")
);
assert_eq!(payload["event_params"]["thread_source"], json!("user"));
assert_eq!(payload["event_params"]["subagent_source"], json!(null));
assert_eq!(payload["event_params"]["parent_thread_id"], json!(null));
assert!(payload["event_params"].get("product_client_id").is_none());
}
#[tokio::test]
async fn rejected_turn_steer_uses_request_connection_metadata() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
let payload = ingest_rejected_turn_steer(
&mut reducer,
&mut out,
no_active_turn_steer_error(),
Some(no_active_turn_steer_error_type()),
)
.await;
assert_eq!(payload["event_type"], json!("codex_turn_steer_event"));
assert_eq!(payload["event_params"]["thread_id"], json!("thread-2"));
assert_eq!(payload["event_params"]["expected_turn_id"], json!("turn-2"));
assert_eq!(payload["event_params"]["accepted_turn_id"], json!(null));
assert_eq!(payload["event_params"]["num_input_images"], json!(1));
assert_eq!(
payload["event_params"]["app_server_client"]["product_client_id"],
json!("codex-tui")
);
assert_eq!(
payload["event_params"]["runtime"]["codex_rs_version"],
json!("0.1.0")
);
assert_eq!(payload["event_params"]["thread_source"], json!("user"));
assert_eq!(payload["event_params"]["subagent_source"], json!(null));
assert_eq!(payload["event_params"]["parent_thread_id"], json!(null));
assert_eq!(payload["event_params"]["result"], json!("rejected"));
assert_eq!(
payload["event_params"]["rejection_reason"],
json!("no_active_turn")
);
assert!(
payload["event_params"]["created_at"]
.as_u64()
.expect("created_at")
> 0
);
}
#[tokio::test]
async fn rejected_turn_steer_maps_active_turn_not_steerable_error_type() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
let payload = ingest_rejected_turn_steer(
&mut reducer,
&mut out,
non_steerable_review_error(),
Some(non_steerable_review_error_type()),
)
.await;
assert_eq!(
payload["event_params"]["rejection_reason"],
json!("non_steerable_review")
);
}
#[tokio::test]
async fn rejected_turn_steer_maps_input_too_large_error_type() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
let payload = ingest_rejected_turn_steer(
&mut reducer,
&mut out,
input_too_large_steer_error(),
Some(input_too_large_error_type()),
)
.await;
assert_eq!(
payload["event_params"]["rejection_reason"],
json!("input_too_large")
);
}
#[tokio::test]
async fn turn_steer_does_not_emit_without_pending_request() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
reducer
.ingest(
AnalyticsFact::ErrorResponse {
connection_id: 7,
request_id: RequestId::Integer(4),
error: no_active_turn_steer_error(),
error_type: Some(no_active_turn_steer_error_type()),
},
&mut out,
)
.await;
assert!(out.is_empty());
}
#[tokio::test]
async fn turn_start_error_response_discards_pending_start_request() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_initialize(&mut reducer, &mut out).await;
reducer
.ingest(
AnalyticsFact::ClientRequest {
connection_id: 7,
request_id: RequestId::Integer(3),
request: Box::new(sample_turn_start_request("thread-2", /*request_id*/ 3)),
},
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::ErrorResponse {
connection_id: 7,
request_id: RequestId::Integer(3),
error: no_active_turn_steer_error(),
error_type: None,
},
&mut out,
)
.await;
// A late/synthetic response for the same request id must not resurrect the
// failed turn/start request and attach request-scoped connection metadata.
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(3),
response: Box::new(sample_turn_start_response("turn-2")),
},
&mut out,
)
.await;
assert!(out.is_empty());
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
sample_turn_resolved_config("thread-2", "turn-2"),
))),
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
assert!(out.is_empty());
}
#[tokio::test]
async fn turn_lifecycle_emits_turn_event() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
/*include_token_usage*/ true,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
assert_eq!(out.len(), 1);
let payload = serde_json::to_value(&out[0]).expect("serialize turn event");
assert_eq!(payload["event_type"], json!("codex_turn_event"));
assert_eq!(payload["event_params"]["thread_id"], json!("thread-2"));
assert_eq!(
payload["event_params"]["session_id"],
json!("session-thread-2")
);
assert_eq!(payload["event_params"]["turn_id"], json!("turn-2"));
assert_eq!(
payload["event_params"]["app_server_client"],
json!({
"product_client_id": "codex-tui",
"client_name": "codex-tui",
"client_version": "1.0.0",
"rpc_transport": "stdio",
"experimental_api_enabled": null,
})
);
assert_eq!(
payload["event_params"]["runtime"],
json!({
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64",
})
);
assert!(payload["event_params"].get("product_client_id").is_none());
assert_eq!(payload["event_params"]["ephemeral"], json!(false));
assert_eq!(payload["event_params"]["workspace_kind"], json!(null));
assert_eq!(payload["event_params"]["num_input_images"], json!(1));
assert_eq!(payload["event_params"]["status"], json!("completed"));
assert_eq!(payload["event_params"]["steer_count"], json!(0));
assert_eq!(payload["event_params"]["total_tool_call_count"], json!(0));
assert_eq!(payload["event_params"]["shell_command_count"], json!(0));
assert_eq!(payload["event_params"]["file_change_count"], json!(0));
assert_eq!(payload["event_params"]["mcp_tool_call_count"], json!(0));
assert_eq!(payload["event_params"]["dynamic_tool_call_count"], json!(0));
assert_eq!(
payload["event_params"]["subagent_tool_call_count"],
json!(0)
);
assert_eq!(payload["event_params"]["web_search_count"], json!(0));
assert_eq!(payload["event_params"]["image_generation_count"], json!(0));
assert_eq!(payload["event_params"]["started_at"], json!(455));
assert_eq!(payload["event_params"]["completed_at"], json!(456));
assert_eq!(payload["event_params"]["duration_ms"], json!(1234));
assert_eq!(payload["event_params"]["input_tokens"], json!(123));
assert_eq!(payload["event_params"]["cached_input_tokens"], json!(45));
assert_eq!(payload["event_params"]["output_tokens"], json!(140));
assert_eq!(
payload["event_params"]["reasoning_output_tokens"],
json!(13)
);
assert_eq!(payload["event_params"]["total_tokens"], json!(321));
}
#[tokio::test]
async fn turn_event_counts_completed_tool_items() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
/*include_token_usage*/ false,
)
.await;
let completed_tool_items = vec![
sample_command_execution_item(CommandExecutionStatus::Completed, Some(0), Some(1)),
ThreadItem::FileChange {
id: "file-change-1".to_string(),
changes: Vec::new(),
status: PatchApplyStatus::Completed,
},
ThreadItem::McpToolCall {
id: "mcp-1".to_string(),
server: "server".to_string(),
tool: "search".to_string(),
status: McpToolCallStatus::Completed,
arguments: json!({}),
mcp_app_resource_uri: None,
plugin_id: None,
result: None,
error: None,
duration_ms: Some(2),
},
ThreadItem::DynamicToolCall {
id: "dynamic-1".to_string(),
namespace: None,
tool: "render".to_string(),
arguments: json!({}),
status: DynamicToolCallStatus::Completed,
content_items: None,
success: Some(true),
duration_ms: Some(3),
},
ThreadItem::CollabAgentToolCall {
id: "collab-1".to_string(),
tool: CollabAgentTool::SpawnAgent,
status: CollabAgentToolCallStatus::Completed,
sender_thread_id: "thread-2".to_string(),
receiver_thread_ids: vec!["thread-child".to_string()],
prompt: Some("help".to_string()),
model: Some("gpt-5".to_string()),
reasoning_effort: None,
agents_states: Default::default(),
},
ThreadItem::SubAgentActivity {
id: "sub-agent-activity-1".to_string(),
kind: SubAgentActivityKind::Interacted,
agent_thread_id: "thread-child".to_string(),
agent_path: "/root/child".to_string(),
},
ThreadItem::WebSearch {
id: "web-1".to_string(),
query: "codex".to_string(),
action: None,
},
ThreadItem::ImageGeneration {
id: "image-1".to_string(),
status: "completed".to_string(),
revised_prompt: None,
result: "ok".to_string(),
saved_path: None,
},
];
for item in completed_tool_items {
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::ItemCompleted(
ItemCompletedNotification {
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
completed_at_ms: 1_000,
item,
},
))),
&mut out,
)
.await;
}
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
let turn_event = out
.iter()
.find(|event| matches!(event, TrackEventRequest::TurnEvent(_)))
.expect("turn event should be emitted");
let payload = serde_json::to_value(turn_event).expect("serialize turn event");
assert_eq!(payload["event_params"]["total_tool_call_count"], json!(8));
assert_eq!(payload["event_params"]["shell_command_count"], json!(1));
assert_eq!(payload["event_params"]["file_change_count"], json!(1));
assert_eq!(payload["event_params"]["mcp_tool_call_count"], json!(1));
assert_eq!(payload["event_params"]["dynamic_tool_call_count"], json!(1));
assert_eq!(
payload["event_params"]["subagent_tool_call_count"],
json!(2)
);
assert_eq!(payload["event_params"]["web_search_count"], json!(1));
assert_eq!(payload["event_params"]["image_generation_count"], json!(1));
}
#[tokio::test]
async fn item_completed_without_turn_state_does_not_create_turn_state() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::ItemCompleted(
ItemCompletedNotification {
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
completed_at_ms: 1_000,
item: sample_command_execution_item(
CommandExecutionStatus::Completed,
Some(0),
Some(1),
),
},
))),
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
assert!(out.is_empty());
}
#[tokio::test]
async fn accepted_steers_increment_turn_steer_count() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
/*include_token_usage*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientRequest {
connection_id: 7,
request_id: RequestId::Integer(4),
request: Box::new(sample_turn_steer_request(
"thread-2", "turn-2", /*request_id*/ 4,
)),
},
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(4),
response: Box::new(sample_turn_steer_response("turn-2")),
},
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientRequest {
connection_id: 7,
request_id: RequestId::Integer(5),
request: Box::new(sample_turn_steer_request(
"thread-2", "turn-2", /*request_id*/ 5,
)),
},
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::ErrorResponse {
connection_id: 7,
request_id: RequestId::Integer(5),
error: no_active_turn_steer_error(),
error_type: Some(no_active_turn_steer_error_type()),
},
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientRequest {
connection_id: 7,
request_id: RequestId::Integer(6),
request: Box::new(sample_turn_steer_request(
"thread-2", "turn-2", /*request_id*/ 6,
)),
},
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(6),
response: Box::new(sample_turn_steer_response("turn-2")),
},
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
let turn_event = out
.iter()
.find(|event| matches!(event, TrackEventRequest::TurnEvent(_)))
.expect("turn event should be emitted");
let payload = serde_json::to_value(turn_event).expect("serialize turn event");
assert_eq!(payload["event_params"]["steer_count"], json!(2));
}
#[tokio::test]
async fn turn_does_not_emit_without_required_prerequisites() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ false,
/*include_resolved_config*/ true,
/*include_started*/ false,
/*include_token_usage*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
assert!(out.is_empty());
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ false,
/*include_started*/ false,
/*include_token_usage*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
assert!(out.is_empty());
}
#[tokio::test]
async fn turn_lifecycle_emits_failed_turn_event() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
/*include_token_usage*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnCodexError(Box::new(
TurnCodexErrorFact::from_codex_err(
"thread-2".to_string(),
"turn-2".to_string(),
&CodexErr::InvalidRequest("unknown turn environment id `env-2`".to_string()),
),
))),
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Failed,
Some(codex_app_server_protocol::CodexErrorInfo::BadRequest),
))),
&mut out,
)
.await;
assert_eq!(out.len(), 1);
let payload = serde_json::to_value(&out[0]).expect("serialize turn event");
assert_eq!(payload["event_params"]["status"], json!("failed"));
assert_eq!(payload["event_params"]["turn_error"], json!("badRequest"));
assert_eq!(
payload["event_params"]["codex_error_kind"],
json!("invalid_request")
);
assert_eq!(
payload["event_params"]["codex_error_http_status_code"],
json!(null)
);
}
#[tokio::test]
async fn turn_lifecycle_emits_interrupted_turn_event_without_error() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
/*include_token_usage*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Interrupted,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
assert_eq!(out.len(), 1);
let payload = serde_json::to_value(&out[0]).expect("serialize turn event");
assert_eq!(payload["event_params"]["status"], json!("interrupted"));
assert_eq!(payload["event_params"]["turn_error"], json!(null));
assert_eq!(payload["event_params"]["codex_error_kind"], json!(null));
}
#[tokio::test]
async fn turn_completed_without_started_notification_emits_null_started_at() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ false,
/*include_token_usage*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
let payload = serde_json::to_value(&out[0]).expect("serialize turn event");
assert_eq!(payload["event_params"]["started_at"], json!(null));
assert_eq!(payload["event_params"]["duration_ms"], json!(1234));
assert_eq!(payload["event_params"]["input_tokens"], json!(null));
assert_eq!(payload["event_params"]["cached_input_tokens"], json!(null));
assert_eq!(payload["event_params"]["output_tokens"], json!(null));
assert_eq!(
payload["event_params"]["reasoning_output_tokens"],
json!(null)
);
assert_eq!(payload["event_params"]["total_tokens"], json!(null));
}
fn sample_plugin_metadata() -> PluginTelemetryMetadata {
PluginTelemetryMetadata {
plugin_id: PluginId::parse("sample@test").expect("valid plugin id"),
remote_plugin_id: None,
capability_summary: Some(PluginCapabilitySummary {
config_name: "sample@test".to_string(),
display_name: "sample".to_string(),
description: None,
has_skills: true,
mcp_server_names: vec!["mcp-1".to_string(), "mcp-2".to_string()],
app_connector_ids: vec![
AppConnectorId("calendar".to_string()),
AppConnectorId("drive".to_string()),
],
}),
}
}