codex-rs/analytics/src/analytics_client_tests.rs

use crate::client::AnalyticsEventsClient;
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::CodexAppUsedMetadata;
use crate::events::CodexCommandExecutionEventParams;
use crate::events::CodexCommandExecutionEventRequest;
use crate::events::CodexCompactionEventRequest;
use crate::events::CodexHookRunEventRequest;
use crate::events::CodexOnboardingExternalAgentImportFailureEventRequest;
use crate::events::CodexOnboardingExternalAgentImportFailureMetadata;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginInstallFailedEventRequest;
use crate::events::CodexPluginInstallFailedMetadata;
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::GuardianAdditionalPermissions;
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::ToolEventType;
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::current_runtime_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::ArtifactOperation;
use crate::facts::ArtifactOperationInput;
use crate::facts::ArtifactOperationLifecycle;
use crate::facts::CodeModeToolCallFact;
use crate::facts::CodeModeToolCallStatus;
use crate::facts::CodexCompactionEvent;
use crate::facts::CodexErrKind;
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::ControlToolCallFact;
use crate::facts::ControlToolCallStatus;
use crate::facts::CustomAnalyticsFact;
use crate::facts::ElicitationType;
use crate::facts::ExternalAgentConfigImportCompletedInput;
use crate::facts::ExternalAgentConfigImportFailureInput;
use crate::facts::HookRunFact;
use crate::facts::HookRunInput;
use crate::facts::ImageDetailSetting;
use crate::facts::ImagePreparationFact;
use crate::facts::ImagePreparationMetadata;
use crate::facts::InputError;
use crate::facts::InvocationType;
use crate::facts::McpToolCallElicitation;
use crate::facts::PluginInstallFailedInput;
use crate::facts::PluginInstallRequestSource;
use crate::facts::PluginInstallRequested;
use crate::facts::PluginInstallRequestedInput;
use crate::facts::PluginInstallRequestedPlugin;
use crate::facts::PluginInstallSource;
use crate::facts::PluginMeasurementRow;
use crate::facts::PluginMeasurementsInput;
use crate::facts::PluginState;
use crate::facts::PluginStateChangedInput;
use crate::facts::PluginUsedInput;
use crate::facts::SkillInvocation;
use crate::facts::SkillInvocationLocation;
use crate::facts::SkillInvokedInput;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::ThreadInitializationMode;
use crate::facts::TrackEventsContext;
use crate::facts::TurnAnalyticsMetadata;
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::ImageGenerationItem;
use codex_app_server_protocol::ImageReference;
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::McpToolCallAppContext;
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::ThreadArchivedNotification;
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::ThreadUnarchivedNotification;
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::TurnInterruptResponse;
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_app_server_protocol::WebSearchItem;
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::AdditionalPermissionProfile;
use codex_protocol::models::NetworkPermissions as CoreNetworkPermissions;
use codex_protocol::models::PermissionProfile as CorePermissionProfile;
use codex_protocol::models::SandboxPermissions;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::GuardianCommandSource;
use codex_protocol::protocol::HookEventName;
use codex_protocol::protocol::HookExecutionMode;
use codex_protocol::protocol::HookHandlerType;
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::BTreeMap;
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::SystemTime;
use tokio::sync::mpsc;

const TEST_PRODUCT_CLIENT_ID: &str = "codex_work_desktop";

struct TestTurnMetadata {
    root_turn_id: Mutex<Option<String>>,
}

impl TurnAnalyticsMetadata for TestTurnMetadata {
    fn root_turn_id(&self) -> Option<String> {
        self.root_turn_id.lock().expect("root turn ID").clone()
    }

    fn turn_trigger(&self) -> Option<String> {
        None
    }

    fn codex_turn_source(&self) -> Option<String> {
        None
    }
}

fn test_turn_metadata(root_turn_id: Option<&str>) -> Arc<TestTurnMetadata> {
    Arc::new(TestTurnMetadata {
        root_turn_id: Mutex::new(root_turn_id.map(str::to_string)),
    })
}

fn test_tracking_context(thread_id: &str, turn_id: &str) -> TrackEventsContext {
    TrackEventsContext {
        model_slug: "gpt-5".to_string(),
        thread_id: thread_id.to_string(),
        turn_id: turn_id.to_string(),
        product_client_id: TEST_PRODUCT_CLIENT_ID.to_string(),
    }
}

fn sample_thread_with_metadata(
    thread_id: &str,
    ephemeral: bool,
    source: AppServerSessionSource,
    thread_source: Option<AppServerThreadSource>,
    parent_thread_id: Option<String>,
) -> Thread {
    Thread {
        originator: None,
        environments: None,
        id: thread_id.to_string(),
        extra: None,
        session_id: format!("session-{thread_id}"),
        forked_from_id: None,
        parent_thread_id,
        preview: "first prompt".to_string(),
        ephemeral,
        section: None,
        section_entered_at: None,
        project_id: None,
        daybreak_enabled: None,
        history_mode: Default::default(),
        model_provider: "openai".to_string(),
        model: None,
        reasoning_effort: None,
        created_at: 1,
        updated_at: 2,
        recency_at: Some(2),
        status: AppServerThreadStatus::Idle,
        path: None,
        cwd: test_path_buf("/tmp").abs(),
        cli_version: "0.0.0".to_string(),
        source,
        can_accept_direct_input: None,
        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 {
        disabled_plugin_ids: Vec::new(),
        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::OnRequest,
        approvals_reviewer: AppServerApprovalsReviewer::User,
        sandbox: AppServerSandboxPolicy::DangerFullAccess,
        active_permission_profile: None,
        reasoning_effort: None,
        multi_agent_mode: Default::default(),
    })
}

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 {
        disabled_plugin_ids: Vec::new(),
        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::OnRequest,
        approvals_reviewer: AppServerApprovalsReviewer::User,
        sandbox: AppServerSandboxPolicy::DangerFullAccess,
        active_permission_profile: None,
        reasoning_effort: None,
        collaboration_mode: None,
        multi_agent_mode: Default::default(),
        initial_turns_page: None,
        turns_backwards_cursor: None,
        items_backwards_cursor: 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 {
                    image: ImageReference::Inline {
                        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,
            cache_write_input_tokens: 7,
            output_tokens: 140,
            reasoning_output_tokens: 13,
            codex_rollout_budget_units: None,
        },
    }
}

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 {
                misalignment: None,
                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(),
        turn_metadata: test_turn_metadata(/*root_turn_id*/ None),
        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,
        guardian_v2_enabled: false,
        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,
        compaction_ms: 40,
        between_sampling_overhead_ms: 50,
        tool_blocking_ms: 250,
        after_last_sampling_ms: 94,
        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 sample_turn_interrupt_response() -> ClientResponsePayload {
    ClientResponsePayload::TurnInterrupt(TurnInterruptResponse {})
}

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 {
                misalignment: None,
                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",
                )),
                thread_originator: None,
            },
            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",
                    )),
                    thread_originator: None,
                },
                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")),
                thread_originator: None,
            },
            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",
                )),
                thread_originator: None,
            },
            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 plugin_measurements(rows: Vec<PluginMeasurementRow>) -> PluginMeasurementsInput {
    PluginMeasurementsInput {
        thread_id: "thread-1".to_string(),
        turn_id: "turn-1".to_string(),
        item_id: "item-1".to_string(),
        originator: "codex_cli_rs".to_string(),
        model_slug: None,
        reasoning_effort: None,
        plugin_id: "sample@openai-curated".to_string(),
        execution_id: "execution-1".to_string(),
        operation: "security_scan".to_string(),
        rows,
    }
}

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,
                mcp_server_openai_form_elicitation: false,
                extensions: 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,
    }
}

async fn ingest_complete_child_turn(
    reducer: &mut AnalyticsReducer,
    events: &mut Vec<TrackEventRequest>,
    thread_id: &str,
    turn_id: &str,
) {
    for fact in [
        AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
            sample_turn_resolved_config(thread_id, turn_id),
        ))),
        AnalyticsFact::Custom(CustomAnalyticsFact::TurnProfile(Box::new(
            TurnProfileFact {
                turn_id: turn_id.to_string(),
                profile: sample_turn_profile(),
            },
        ))),
        AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
            thread_id,
            turn_id,
            AppServerTurnStatus::Completed,
            /*codex_error_info*/ None,
        ))),
    ] {
        reducer.ingest(fact, events).await;
    }
}

fn sample_command_execution_item(
    status: CommandExecutionStatus,
    exit_code: Option<i32>,
    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 {
        model_context: None,
        id: id.to_string(),
        plugin_id: None,
        script_path: None,
        command: "echo hi".to_string(),
        cwd: test_path_buf("/tmp").abs().into(),
        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>,
    plugin_id: Option<&str>,
    script_path: Option<&str>,
) -> ThreadItem {
    let mut item = sample_command_execution_item(status, exit_code, duration_ms);
    let ThreadItem::CommandExecution {
        command_actions: item_command_actions,
        plugin_id: item_plugin_id,
        script_path: item_script_path,
        ..
    } = &mut item
    else {
        unreachable!("sample command execution item should be CommandExecution");
    };
    *item_command_actions = command_actions;
    *item_plugin_id = plugin_id.map(str::to_string);
    *item_script_path = script_path.map(str::to_string);
    item
}

fn sample_command_approval_request(request_id: i64, approval_id: Option<&str>) -> ServerRequest {
    ServerRequest::CommandExecutionRequestApproval {
        request_id: RequestId::Integer(request_id),
        params: CommandExecutionRequestApprovalParams {
            kind: Default::default(),
            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),
            environment_id: None,
            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().into(),
            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,
    action: GuardianApprovalReviewAction,
) -> 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,
        },
    )
}

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 = test_tracking_context("thread-1", "turn-1");
    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": TEST_PRODUCT_CLIENT_ID,
                "invoke_type": "explicit",
                "model_slug": "gpt-5"
            }
        })
    );
}

#[test]
fn app_used_event_serializes_expected_shape() {
    let tracking = test_tracking_context("thread-2", "turn-2");
    let event = TrackEventRequest::AppUsed(CodexAppUsedEventRequest {
        event_type: "codex_app_used",
        event_params: CodexAppUsedMetadata {
            app: codex_app_metadata(
                &tracking,
                AppInvocation {
                    connector_id: Some("drive".to_string()),
                    app_name: Some("Google Drive".to_string()),
                    invocation_type: Some(InvocationType::Implicit),
                },
            ),
            elicitation_type: None,
        },
    });

    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": TEST_PRODUCT_CLIENT_ID,
                "invoke_type": "implicit",
                "model_slug": "gpt-5",
                "elicitation_type": null
            }
        })
    );
}

#[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: [],
            },
        },
    ));

    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());
}

#[tokio::test]
#[cfg(debug_assertions)]
async fn analytics_flush_delivers_completed_turn_with_file_diff() {
    let nonce = SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .expect("system clock should be after Unix epoch")
        .as_nanos();
    let capture_path = std::env::temp_dir().join(format!(
        "codex-analytics-turn-flush-{}-{nonce}.jsonl",
        std::process::id()
    ));
    let auth_manager = codex_login::AuthManager::from_auth_for_testing(
        codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing(),
    );
    let client = AnalyticsEventsClient::new_for_capture_file(auth_manager, capture_path.clone());

    for fact in [
        sample_initialize_fact(/*connection_id*/ 7),
        AnalyticsFact::ClientResponse {
            connection_id: 7,
            request_id: RequestId::Integer(1),
            response: Box::new(sample_thread_start_response(
                "thread-2", /*ephemeral*/ false, "gpt-5",
            )),
            thread_originator: None,
        },
        AnalyticsFact::ClientRequest {
            connection_id: 7,
            request_id: RequestId::Integer(3),
            request: Box::new(sample_turn_start_request("thread-2", /*request_id*/ 3)),
        },
        AnalyticsFact::ClientResponse {
            connection_id: 7,
            request_id: RequestId::Integer(3),
            response: Box::new(sample_turn_start_response("turn-2")),
            thread_originator: None,
        },
        AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
            sample_turn_resolved_config("thread-2", "turn-2"),
        ))),
        AnalyticsFact::Notification(Box::new(sample_turn_started_notification(
            "thread-2", "turn-2",
        ))),
        AnalyticsFact::Custom(CustomAnalyticsFact::TurnProfile(Box::new(
            TurnProfileFact {
                turn_id: "turn-2".to_string(),
                profile: sample_turn_profile(),
            },
        ))),
        AnalyticsFact::Notification(Box::new(ServerNotification::TurnDiffUpdated(
            TurnDiffUpdatedNotification {
                thread_id: "thread-2".to_string(),
                turn_id: "turn-2".to_string(),
                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 @@
+let value = 1;
"
                .to_string(),
            },
        ))),
        AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
            "thread-2",
            "turn-2",
            AppServerTurnStatus::Completed,
            /*codex_error_info*/ None,
        ))),
    ] {
        client.record_fact(fact);
    }

    client.flush().await;

    let contents = std::fs::read_to_string(&capture_path).expect("read captured analytics events");
    let event_types = contents
        .lines()
        .flat_map(|line| {
            serde_json::from_str::<serde_json::Value>(line)
                .expect("parse captured analytics events")["events"]
                .as_array()
                .expect("captured events should be an array")
                .iter()
                .map(|event| {
                    event["event_type"]
                        .as_str()
                        .expect("captured event type should be a string")
                        .to_string()
                })
                .collect::<Vec<_>>()
        })
        .collect::<Vec<_>>();
    assert!(event_types.iter().any(|event| event == "codex_turn_event"));
    assert!(
        event_types
            .iter()
            .any(|event| event == "codex_accepted_line_fingerprints")
    );

    std::fs::remove_file(capture_path).expect("remove analytics capture file");
}

#[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::ResponsesCompactionV2,
                phase: CompactionPhase::MidTurn,
                strategy: CompactionStrategy::Memento,
                status: CompactionStatus::Completed,
                codex_error_kind: None,
                codex_error_http_status_code: None,
                active_context_tokens_before: 120_000,
                active_context_tokens_after: 18_000,
                retained_image_count: None,
                compaction_summary_tokens: None,
                cached_input_tokens: None,
                cache_write_input_tokens: Some(456),
                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_compaction_v2",
                "phase": "mid_turn",
                "strategy": "memento",
                "status": "completed",
                "codex_error_kind": null,
                "codex_error_http_status_code": null,
                "active_context_tokens_before": 120000,
                "active_context_tokens_after": 18000,
                "retained_image_count": null,
                "compaction_summary_tokens": null,
                "cached_input_tokens": null,
                "cache_write_input_tokens": 456,
                "started_at": 100,
                "completed_at": 106,
                "duration_ms": 6543
            }
        })
    );
}

#[tokio::test]
async fn image_preparation_fact_is_included_in_turn_event() {
    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*/ false,
    )
    .await;

    let metadata = ImagePreparationMetadata {
        message_role: None,
        item_id: Some("call-1".to_string()),
        effective_detail: ImageDetailSetting::High,
        source_width: 2_048,
        source_height: 2_048,
        prepared_width: 1_600,
        prepared_height: 1_600,
    };
    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::ImagePreparation(Box::new(
                ImagePreparationFact {
                    turn_id: "turn-2".to_string(),
                    metadata: metadata.clone(),
                },
            ))),
            &mut events,
        )
        .await;
    reducer
        .ingest(
            AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
                "thread-2",
                "turn-2",
                AppServerTurnStatus::Completed,
                /*codex_error_info*/ None,
            ))),
            &mut events,
        )
        .await;

    let [TrackEventRequest::TurnEvent(event)] = events.as_slice() else {
        panic!("expected one turn event");
    };
    assert_eq!(event.event_params.image_preparations, vec![metadata]);
}

#[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 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,
            is_worktree: Some(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,
                "is_worktree": true,
                "thread_source": "automation",
                "initialization_mode": "new",
                "subagent_source": null,
                "parent_thread_id": null,
                "forked_from_thread_id": null,
                "created_at": 1
            }
        })
    );
}

#[tokio::test]
async fn thread_initialized_classifies_validated_linked_worktrees() {
    let root = std::env::temp_dir().join(format!(
        "codex-analytics-worktree-{}",
        codex_protocol::ThreadId::new()
    ));
    let primary = root.join("primary");
    let linked = root.join("linked");
    let admin = primary.join(".git/worktrees/linked");
    std::fs::create_dir_all(&admin).expect("worktree administrative directory");
    std::fs::create_dir_all(&linked).expect("linked checkout");
    std::fs::write(primary.join(".git/HEAD"), "ref: refs/heads/main\n")
        .expect("primary repository HEAD");
    std::fs::write(admin.join("commondir"), "../..\n").expect("common directory");
    std::fs::write(
        admin.join("gitdir"),
        format!("{}\n", linked.join(".git").display()),
    )
    .expect("linked checkout backlink");
    std::fs::write(
        linked.join(".git"),
        format!("gitdir: {}\n", admin.display()),
    )
    .expect("linked checkout git file");

    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();
    ingest_initialize(&mut reducer, &mut events).await;
    for (cwd, expected) in [
        (primary.as_path(), json!(false)),
        (linked.as_path(), json!(true)),
        (root.as_path(), serde_json::Value::Null),
    ] {
        events.clear();
        let mut response =
            sample_thread_start_response("thread-1", /*ephemeral*/ false, "gpt-5");
        let ClientResponsePayload::ThreadStart(start) = &mut response else {
            panic!("expected thread/start response");
        };
        start.thread.cwd = codex_utils_absolute_path::AbsolutePathBuf::from_absolute_path(cwd)
            .expect("absolute checkout path");
        reducer
            .ingest(
                AnalyticsFact::ClientResponse {
                    connection_id: 7,
                    request_id: RequestId::Integer(1),
                    response: Box::new(response),
                    thread_originator: None,
                },
                &mut events,
            )
            .await;
        let payload = serde_json::to_value(&events).expect("serialize thread event");
        assert_eq!(payload[0]["event_params"]["is_worktree"], expected);
    }

    std::fs::remove_dir_all(root).expect("remove test checkout");
}

#[test]
fn command_execution_event_serializes_expected_shape() {
    let event = TrackEventRequest::CommandExecution(CodexCommandExecutionEventRequest {
        event_type: "codex_command_execution_event",
        event_params: CodexCommandExecutionEventParams {
            model_slug: None,
            reasoning_effort: None,
            base: CodexToolItemEventBase {
                thread_id: "thread-1".to_string(),
                session_id: "session-thread-1".to_string(),
                turn_id: "turn-1".to_string(),
                root_turn_id: Some("root-turn".to_string()),
                item_id: "item-1".to_string(),
                cell_id: None,
                parent_call_id: None,
                originating_response_id: None,
                subsequent_response_id: None,
                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(),
                tool_event_type: Some(ToolEventType::ModelToolCall),
                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,
            },
            plugin_id: Some("sample@openai-curated".to_string()),
            script_path: Some("scripts/run.py".to_string()),
            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");
    let mut expected = json!({
        "event_type": "codex_command_execution_event",
        "event_params": {
            "model_slug": null,
            "reasoning_effort": null,
            "thread_id": "thread-1",
            "session_id": "session-thread-1",
            "turn_id": "turn-1",
            "root_turn_id": "root-turn",
            "item_id": "item-1",
            "cell_id": null,
            "parent_call_id": null,
            "originating_response_id": null,
            "subsequent_response_id": null,
            "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,
            "plugin_id": "sample@openai-curated",
            "script_path": "scripts/run.py",
            "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
        }
    });
    // Keep this field separate to stay within json!'s macro recursion limit.
    expected["event_params"]["tool_event_type"] = json!("model_tool_call");
    assert_eq!(payload, expected);
}

#[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",
                )),
                thread_originator: None,
            },
            &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,
                        mcp_server_openai_form_elicitation: false,
                        extensions: 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",
                )),
                thread_originator: 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_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 thread_originator_overrides_shared_connection_across_thread_events() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();

    reducer
        .ingest(sample_initialize_fact(/*connection_id*/ 7), &mut events)
        .await;
    for (request_id, thread_id, thread_originator) in [
        (1, "thread-work", Some(TEST_PRODUCT_CLIENT_ID.to_string())),
        (2, "thread-default", None),
    ] {
        reducer
            .ingest(
                AnalyticsFact::ClientResponse {
                    connection_id: 7,
                    request_id: RequestId::Integer(request_id),
                    response: Box::new(sample_thread_start_response(
                        thread_id, /*ephemeral*/ false, "gpt-5",
                    )),
                    thread_originator,
                },
                &mut events,
            )
            .await;
    }

    let initialized = serde_json::to_value(&events).expect("serialize thread events");
    assert_eq!(
        initialized
            .as_array()
            .expect("thread events")
            .iter()
            .map(|event| {
                json!({
                    "thread_id": event["event_params"]["thread_id"],
                    "app_server_client": event["event_params"]["app_server_client"],
                })
            })
            .collect::<Vec<_>>(),
        vec![
            json!({
                "thread_id": "thread-work",
                "app_server_client": {
                    "product_client_id": TEST_PRODUCT_CLIENT_ID,
                    "client_name": "codex-tui",
                    "client_version": "1.0.0",
                    "rpc_transport": "websocket",
                    "experimental_api_enabled": false,
                },
            }),
            json!({
                "thread_id": "thread-default",
                "app_server_client": {
                    "product_client_id": DEFAULT_ORIGINATOR,
                    "client_name": "codex-tui",
                    "client_version": "1.0.0",
                    "rpc_transport": "websocket",
                    "experimental_api_enabled": false,
                },
            }),
        ]
    );

    events.clear();
    reducer
        .ingest(
            AnalyticsFact::ClientRequest {
                connection_id: 7,
                request_id: RequestId::Integer(3),
                request: Box::new(sample_turn_start_request(
                    "thread-work",
                    /*request_id*/ 3,
                )),
            },
            &mut events,
        )
        .await;
    reducer
        .ingest(
            AnalyticsFact::ClientResponse {
                connection_id: 7,
                request_id: RequestId::Integer(3),
                response: Box::new(sample_turn_start_response("turn-1")),
                thread_originator: None,
            },
            &mut events,
        )
        .await;
    ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-work", "item-work")
        .await;
    ingest_complete_child_turn(&mut reducer, &mut events, "thread-work", "turn-1").await;
    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(Box::new(
                CodexCompactionEvent {
                    thread_id: "thread-work".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,
                    codex_error_kind: None,
                    codex_error_http_status_code: None,
                    active_context_tokens_before: 131_000,
                    active_context_tokens_after: 64_000,
                    retained_image_count: None,
                    compaction_summary_tokens: None,
                    cached_input_tokens: None,
                    cache_write_input_tokens: None,
                    started_at: 100,
                    completed_at: 101,
                    duration_ms: Some(1200),
                },
            ))),
            &mut events,
        )
        .await;

    let lifecycle = serde_json::to_value(&events).expect("serialize lifecycle events");
    assert_eq!(
        lifecycle
            .as_array()
            .expect("lifecycle events")
            .iter()
            .map(|event| {
                json!({
                    "event_type": event["event_type"],
                    "product_client_id":
                        event["event_params"]["app_server_client"]["product_client_id"],
                })
            })
            .collect::<Vec<_>>(),
        vec![
            json!({
                "event_type": "codex_command_execution_event",
                "product_client_id": TEST_PRODUCT_CLIENT_ID,
            }),
            json!({
                "event_type": "codex_turn_event",
                "product_client_id": TEST_PRODUCT_CLIENT_ID,
            }),
            json!({
                "event_type": "codex_compaction_event",
                "product_client_id": TEST_PRODUCT_CLIENT_ID,
            }),
        ]
    );

    reducer
        .ingest(
            AnalyticsFact::ClientResponse {
                connection_id: 7,
                request_id: RequestId::Integer(4),
                response: Box::new(sample_thread_resume_response_with_source(
                    "thread-private-source",
                    /*ephemeral*/ false,
                    "gpt-5",
                    AppServerSessionSource::Exec,
                    Some(AppServerThreadSource::Feature(
                        "private customer feature label".to_string(),
                    )),
                    Some("019ee5cf-4d15-77a2-8023-01a9f79b6e7d".to_string()),
                )),
                thread_originator: Some(TEST_PRODUCT_CLIENT_ID.to_string()),
            },
            &mut events,
        )
        .await;
    events.clear();

    for notification in [
        ServerNotification::ThreadArchived(ThreadArchivedNotification {
            thread_id: "thread-work".to_string(),
        }),
        ServerNotification::ThreadUnarchived(ThreadUnarchivedNotification {
            thread_id: "thread-default".to_string(),
        }),
        ServerNotification::ThreadArchived(ThreadArchivedNotification {
            thread_id: "thread-private-source".to_string(),
        }),
        ServerNotification::ThreadUnarchived(ThreadUnarchivedNotification {
            thread_id: "thread-without-context".to_string(),
        }),
    ] {
        reducer
            .ingest(
                AnalyticsFact::Notification(Box::new(notification)),
                &mut events,
            )
            .await;
    }

    let mut archives = serde_json::to_value(&events).expect("serialize archive events");
    for event in archives.as_array_mut().expect("archive events") {
        assert!(event["event_params"]["occurred_at_ms"].is_u64());
        event["event_params"]
            .as_object_mut()
            .expect("archive event params")
            .remove("occurred_at_ms");
    }
    assert_eq!(
        archives,
        json!([
            {
                "event_type": "codex_thread_archive_event",
                "event_params": {
                    "thread_id": "thread-work",
                    "action": "archived",
                    "app_server_client": initialized[0]["event_params"]["app_server_client"],
                    "runtime": initialized[0]["event_params"]["runtime"],
                    "thread_source": "user",
                },
            },
            {
                "event_type": "codex_thread_archive_event",
                "event_params": {
                    "thread_id": "thread-default",
                    "action": "unarchived",
                    "app_server_client": initialized[1]["event_params"]["app_server_client"],
                    "runtime": initialized[1]["event_params"]["runtime"],
                    "thread_source": "user",
                },
            },
            {
                "event_type": "codex_thread_archive_event",
                "event_params": {
                    "thread_id": "thread-private-source",
                    "action": "archived",
                    "app_server_client": initialized[0]["event_params"]["app_server_client"],
                    "runtime": initialized[0]["event_params"]["runtime"],
                    "parent_thread_id": "019ee5cf-4d15-77a2-8023-01a9f79b6e7d",
                },
            },
            {
                "event_type": "codex_thread_archive_event",
                "event_params": {
                    "thread_id": "thread-without-context",
                    "action": "unarchived",
                },
            }
        ])
    );
}

#[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")),
                thread_originator: None,
            },
            &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 {},
                )),
                thread_originator: None,
            },
            &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,
                        mcp_server_openai_form_elicitation: false,
                        extensions: 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()),
                )),
                thread_originator: None,
            },
            &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,
                    codex_error_kind: Some(CodexErrKind::ContextWindowExceeded),
                    codex_error_http_status_code: None,
                    active_context_tokens_before: 131_000,
                    active_context_tokens_after: 131_000,
                    retained_image_count: None,
                    compaction_summary_tokens: None,
                    cached_input_tokens: None,
                    cache_write_input_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"]["codex_error_kind"],
        json!("context_window_exceeded")
    );
    assert_eq!(
        payload[0]["event_params"]["codex_error_http_status_code"],
        json!(null)
    );
    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");
}

#[test]
fn execve_serializes_enabled_network_permissions() {
    let permissions: AdditionalPermissionProfile = serde_json::from_value(json!({
        "network": { "enabled": true },
    }))
    .expect("network permissions");

    let action = GuardianReviewedAction::Execve {
        source: GuardianCommandSource::UnifiedExec,
        additional_permissions: Some(GuardianAdditionalPermissions::from(&permissions)),
    };

    assert_eq!(
        serde_json::to_value(action).expect("serialize action"),
        json!({
            "type": "execve",
            "source": "unified_exec",
            "additional_permissions": {
                "network": { "enabled": true },
            },
        }),
    );
}

#[test]
fn unified_exec_serializes_disabled_network_permissions() {
    let permissions: AdditionalPermissionProfile = serde_json::from_value(json!({
        "network": { "enabled": false },
    }))
    .expect("network permissions");

    let action = GuardianReviewedAction::UnifiedExec {
        sandbox_permissions: SandboxPermissions::WithAdditionalPermissions,
        additional_permissions: Some(GuardianAdditionalPermissions::from(&permissions)),
        tty: false,
    };

    assert_eq!(
        serde_json::to_value(action).expect("serialize action"),
        json!({
            "type": "unified_exec",
            "sandbox_permissions": "with_additional_permissions",
            "additional_permissions": {
                "network": { "enabled": false },
            },
            "tty": false,
        }),
    );
}

#[test]
fn permission_metadata_preserves_absent_and_empty_requests() {
    for (input, expected) in [
        (json!(null), json!(null)),
        (json!({}), json!({ "network": null })),
        (
            json!({ "network": { "enabled": null } }),
            json!({
                "network": { "enabled": null },
            }),
        ),
    ] {
        let permissions: Option<AdditionalPermissionProfile> =
            serde_json::from_value(input).expect("optional permissions");
        let metadata = permissions
            .as_ref()
            .map(GuardianAdditionalPermissions::from);

        assert_eq!(
            serde_json::to_value(metadata).expect("serialize permissions"),
            expected,
        );
    }
}

#[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,
                        mcp_server_openai_form_elicitation: false,
                        extensions: 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",
                )),
                thread_originator: None,
            },
            &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,
                    guardian_default_review_model_id: Some("codex-auto-review".to_string()),
                    guardian_catalog_contains_auto_review: Some(false),
                    guardian_review_model_overridden: Some(false),
                    guardian_review_model_override: None,
                    guardian_model_provider_id: Some("openai".to_string()),
                    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,
                    cache_write_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);
    assert_eq!(
        payload[0]["event_params"]["guardian_default_review_model_id"],
        "codex-auto-review"
    );
    assert_eq!(
        payload[0]["event_params"]["guardian_catalog_contains_auto_review"],
        false
    );
    assert_eq!(
        payload[0]["event_params"]["guardian_review_model_overridden"],
        false
    );
    assert_eq!(
        payload[0]["event_params"]["guardian_review_model_override"],
        json!(null)
    );
    assert_eq!(
        payload[0]["event_params"]["guardian_model_provider_id"],
        "openai"
    );
}

#[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;
    for model in ["invoking-model", "later-model"] {
        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: {
                            let mut item = sample_command_execution_item(
                                CommandExecutionStatus::InProgress,
                                /*exit_code*/ None,
                                /*duration_ms*/ None,
                            );
                            if let ThreadItem::CommandExecution { model_context, .. } = &mut item {
                                *model_context =
                                    Some(codex_protocol::items::ModelInvocationContext {
                                        model_slug: model.to_string(),
                                        reasoning_effort: Some("max".to_string()),
                                    });
                            }
                            item
                        },
                    },
                ))),
                &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().into(),
                            },
                            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(),
                            },
                        ],
                        Some("sample@openai-curated"),
                        Some("scripts/run.py"),
                    ),
                },
            ))),
            &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"]["model_slug"], "invoking-model");
    assert_eq!(payload[0]["event_params"]["reasoning_effort"], "max");
    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"]["session_id"], "session-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"]["plugin_id"],
        "sample@openai-curated"
    );
    assert_eq!(payload[0]["event_params"]["script_path"], "scripts/run.py");
    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 plugin_measurement_batch_emits_directly_and_filters_invalid_rows() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();
    let mut too_many_dimensions = BTreeMap::new();
    for index in 0..9 {
        too_many_dimensions.insert(format!("dimension_{index}"), "allowed".to_string());
    }
    let mut measurements = plugin_measurements(vec![
        PluginMeasurementRow {
            measurement_name: "finding_count".to_string(),
            number_value: 3.0,
            dimensions: BTreeMap::from([("severity".to_string(), "high".to_string())]),
        },
        PluginMeasurementRow {
            measurement_name: "non_finite".to_string(),
            number_value: f64::NAN,
            dimensions: BTreeMap::new(),
        },
        PluginMeasurementRow {
            measurement_name: "too_many_dimensions".to_string(),
            number_value: 1.0,
            dimensions: too_many_dimensions,
        },
        PluginMeasurementRow {
            measurement_name: "files_scanned".to_string(),
            number_value: 17.0,
            dimensions: BTreeMap::new(),
        },
    ]);
    measurements.model_slug = Some("invoking-model".to_string());
    measurements.reasoning_effort = Some("max".to_string());
    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::PluginMeasurements(measurements)),
            &mut events,
        )
        .await;
    let payload = serde_json::to_value(&events).expect("serialize events");
    assert_eq!(
        payload,
        json!([
            {
                "event_type": "codex_plugin_measurement_event",
                "event_params": {
                    "thread_id": "thread-1",
                    "turn_id": "turn-1",
                    "item_id": "item-1",
                    "plugin_id": "sample@openai-curated",
                    "execution_id": "execution-1",
                    "operation": "security_scan",
                    "measurement_name": "finding_count",
                    "originator": "codex_cli_rs",
                    "model_slug": "invoking-model",
                    "reasoning_effort": "max",
                    "number_value": 3.0,
                    "dimensions": {"severity": "high"},
                },
            },
            {
                "event_type": "codex_plugin_measurement_event",
                "event_params": {
                    "thread_id": "thread-1",
                    "turn_id": "turn-1",
                    "item_id": "item-1",
                    "plugin_id": "sample@openai-curated",
                    "execution_id": "execution-1",
                    "operation": "security_scan",
                    "measurement_name": "files_scanned",
                    "originator": "codex_cli_rs",
                    "model_slug": "invoking-model",
                    "reasoning_effort": "max",
                    "number_value": 17.0,
                    "dimensions": null,
                },
            },
        ])
    );
}

#[tokio::test]
async fn command_execution_approval_response_publishes_user_review_event() {
    for (kind, approval_id, subject, trigger) in [
        (None, None, "command_execution", "initial"),
        (
            None,
            Some("execve-approval"),
            "command_execution",
            "execve_intercept",
        ),
        (
            Some("writeStdin"),
            Some("stdin-approval"),
            "write_stdin",
            "initial",
        ),
    ] {
        let mut reducer = AnalyticsReducer::default();
        let mut events = Vec::new();

        ingest_review_prerequisites(&mut reducer, &mut events).await;
        let mut request = serde_json::to_value(sample_command_approval_request(
            /*request_id*/ 41,
            approval_id,
        ))
        .expect("serialize approval request");
        // Missing kind models requests from older app-servers.
        if let Some(kind) = kind {
            request["params"]["kind"] = json!(kind);
        } else {
            request["params"].as_object_mut().unwrap().remove("kind");
        }
        reducer
            .ingest(
                AnalyticsFact::ServerRequest {
                    connection_id: 7,
                    request: Box::new(
                        serde_json::from_value(request).expect("deserialize approval request"),
                    ),
                },
                &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"], subject);
        assert_eq!(payload[0]["event_params"]["subject_name"], subject);
        assert_eq!(payload[0]["event_params"]["reviewer"], "user");
        assert_eq!(payload[0]["event_params"]["trigger"], trigger);
        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);

        // Stdin reviews must not count toward the parent command's approval summary.
        events.clear();
        ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-1", "item-1")
            .await;
        let item = serde_json::to_value(&events[0]).expect("serialize tool item event");
        assert_eq!(
            item["event_params"]["review_count"],
            u64::from(kind.is_none())
        );
    }
}

async fn ingest_code_mode_facts(
    reducer: &mut AnalyticsReducer,
    events: &mut Vec<TrackEventRequest>,
    facts: impl IntoIterator<Item = CodeModeToolCallFact>,
) {
    for fact in facts {
        reducer
            .ingest(
                AnalyticsFact::Custom(CustomAnalyticsFact::CodeModeToolCall(fact)),
                events,
            )
            .await;
    }
}

fn sampling_response(
    turn_id: &str,
    response_id: &str,
    tool_call_ids: &[&str],
) -> CodeModeToolCallFact {
    CodeModeToolCallFact::SamplingResponseCompleted {
        thread_id: "thread-1".into(),
        turn_id: turn_id.into(),
        response_id: response_id.into(),
        tool_call_ids: tool_call_ids.iter().map(|id| (*id).into()).collect(),
    }
}

#[tokio::test]
async fn collaborator_tool_events_keep_response_ids_when_completion_races_sampling() {
    for response_first in [false, true] {
        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;
        let item = ThreadItem::CollabAgentToolCall {
            id: "call-1".into(),
            tool: CollabAgentTool::SendMessage,
            status: CollabAgentToolCallStatus::Failed,
            sender_thread_id: "thread-1".into(),
            receiver_thread_ids: Vec::new(),
            prompt: None,
            model: None,
            reasoning_effort: None,
            agents_states: Default::default(),
        };
        reducer
            .ingest(
                AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted(
                    ItemStartedNotification {
                        thread_id: "thread-1".into(),
                        turn_id: "turn-1".into(),
                        started_at_ms: 1_000,
                        item: item.clone(),
                    },
                ))),
                &mut events,
            )
            .await;
        let response = AnalyticsFact::Custom(CustomAnalyticsFact::CodeModeToolCall(
            sampling_response("turn-1", "response-1", &["call-1"]),
        ));
        let completion = AnalyticsFact::Notification(Box::new(ServerNotification::ItemCompleted(
            ItemCompletedNotification {
                thread_id: "thread-1".into(),
                turn_id: "turn-1".into(),
                completed_at_ms: 1_010,
                item,
            },
        )));
        let facts = if response_first {
            [response, completion]
        } else {
            [completion, response]
        };
        for fact in facts {
            reducer.ingest(fact, &mut events).await;
            assert!(events.is_empty(), "emitted before response correlation");
        }
        ingest_code_mode_facts(
            &mut reducer,
            &mut events,
            [sampling_response("turn-1", "response-2", &[])],
        )
        .await;
        let payload = serde_json::to_value(&events).expect("serialize collaborator event");
        assert_eq!(payload.as_array().expect("events array").len(), 1);
        let params = &payload[0]["event_params"];
        assert_eq!(
            json!({
                "type": payload[0]["event_type"],
                "item": params["item_id"],
                "origin": params["originating_response_id"],
                "subsequent": params["subsequent_response_id"],
                "status": params["terminal_status"],
                "tool_event_type": params["tool_event_type"],
            }),
            json!({
                "type": "codex_collab_agent_tool_call_event",
                "item": "call-1",
                "origin": "response-1",
                "subsequent": "response-2",
                "status": "failed",
                "tool_event_type": "model_tool_call",
            }),
        );
    }
}

#[tokio::test]
async fn code_mode_exec_wait_and_child_events_share_cell_and_response_ids() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();
    ingest_review_prerequisites(&mut reducer, &mut events).await;
    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
                TurnResolvedConfigFact {
                    turn_metadata: test_turn_metadata(Some("root-a")),
                    ..sample_turn_resolved_config("thread-1", "turn-1")
                },
            ))),
            &mut events,
        )
        .await;

    let completed = |turn_id: &str, root_turn_id: &str, call_id: &str, tool_name: &str| {
        CodeModeToolCallFact::Completed {
            thread_id: "thread-1".into(),
            turn_id: turn_id.into(),
            turn_metadata: test_turn_metadata(Some(root_turn_id)),
            call_id: call_id.into(),
            cell_id: Some("cell-1".into()),
            tool_name: tool_name.into(),
            started_at_ms: 1_000,
            completed_at_ms: 1_010,
            status: CodeModeToolCallStatus::Completed,
        }
    };
    ingest_code_mode_facts(
        &mut reducer,
        &mut events,
        [
            sampling_response("turn-1", "resp-a", &["exec-1"]),
            CodeModeToolCallFact::CellStarted {
                thread_id: "thread-1".into(),
                turn_id: "turn-1".into(),
                call_id: "exec-1".into(),
                cell_id: "cell-1".into(),
            },
            CodeModeToolCallFact::ChildStarted {
                thread_id: "thread-1".into(),
                turn_id: "turn-1".into(),
                call_id: "child-1".into(),
                cell_id: "cell-1".into(),
            },
            completed("turn-1", "root-a", "exec-1", "exec"),
        ],
    )
    .await;
    ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-1", "child-1").await;
    assert!(events.is_empty());

    ingest_code_mode_facts(
        &mut reducer,
        &mut events,
        [
            sampling_response("turn-1", "resp-b", &[]),
            sampling_response("turn-2", "resp-c", &["wait-1"]),
            completed("turn-2", "root-b", "wait-1", "wait"),
            sampling_response("turn-2", "resp-d", &[]),
        ],
    )
    .await;

    let actual = events
        .iter()
        .map(|event| {
            let event = serde_json::to_value(event).expect("serialize tool event");
            serde_json::json!({
                "item": event["event_params"]["item_id"],
                "tool_event_type": event["event_params"]["tool_event_type"],
                "root": event["event_params"]["root_turn_id"],
                "cell": event["event_params"]["cell_id"],
                "parent": event["event_params"]["parent_call_id"],
                "origin": event["event_params"]["originating_response_id"],
                "subsequent": event["event_params"]["subsequent_response_id"],
            })
        })
        .collect::<Vec<_>>();
    assert_eq!(
        actual,
        vec![
            serde_json::json!({"tool_event_type":"model_tool_call","item":"exec-1","root":"root-a","cell":"cell-1","parent":null,"origin":"resp-a","subsequent":"resp-b"}),
            serde_json::json!({"tool_event_type":"inner_tool_call","item":"child-1","root":"root-a","cell":"cell-1","parent":"exec-1","origin":"resp-a","subsequent":"resp-b"}),
            serde_json::json!({"tool_event_type":"model_tool_call","item":"wait-1","root":"root-b","cell":"cell-1","parent":"exec-1","origin":"resp-c","subsequent":"resp-d"}),
        ]
    );
}

#[tokio::test]
async fn tool_event_types_require_exact_unambiguous_call_origin() {
    for (sampled_ids, child_ids, expected) in [
        (
            vec!["command-1", "control-1"],
            vec![],
            json!("model_tool_call"),
        ),
        (
            vec![],
            vec!["command-1", "control-1"],
            json!("inner_tool_call"),
        ),
        (vec![], vec![], json!(null)),
        (
            vec!["command-1", "control-1"],
            vec!["command-1", "control-1"],
            json!(null),
        ),
        (vec!["other-call"], vec!["other-child"], json!(null)),
    ] {
        let mut reducer = AnalyticsReducer::default();
        let mut events = Vec::new();
        ingest_review_prerequisites(&mut reducer, &mut events).await;
        ingest_code_mode_facts(
            &mut reducer,
            &mut events,
            [
                sampling_response("turn-1", "response-1", &sampled_ids),
                CodeModeToolCallFact::CellStarted {
                    thread_id: "thread-1".into(),
                    turn_id: "turn-1".into(),
                    call_id: "exec-1".into(),
                    cell_id: "cell-1".into(),
                },
            ],
        )
        .await;
        ingest_code_mode_facts(
            &mut reducer,
            &mut events,
            child_ids
                .into_iter()
                .map(|call_id| CodeModeToolCallFact::ChildStarted {
                    thread_id: "thread-1".into(),
                    turn_id: "turn-1".into(),
                    call_id: call_id.into(),
                    cell_id: "cell-1".into(),
                }),
        )
        .await;
        ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-1", "command-1")
            .await;
        reducer
            .ingest(
                AnalyticsFact::Custom(CustomAnalyticsFact::ControlToolCall(ControlToolCallFact {
                    thread_id: "thread-1".into(),
                    turn_id: "turn-1".into(),
                    turn_metadata: test_turn_metadata(/*root_turn_id*/ None),
                    call_id: "control-1".into(),
                    cell_id: Some("cell-1".into()),
                    tool_name: "view_image".into(),
                    started_at_ms: 1_000,
                    completed_at_ms: 1_042,
                    status: ControlToolCallStatus::Completed,
                })),
                &mut events,
            )
            .await;
        reducer.flush(&mut events);
        let payload = serde_json::to_value(&events).expect("serialize tool events");
        assert_eq!(
            payload
                .as_array()
                .expect("events")
                .iter()
                .map(|event| json!({
                    "event": event["event_type"],
                    "type": event["event_params"]["tool_event_type"],
                    "status": event["event_params"]["terminal_status"],
                    "duration": event["event_params"]["duration_ms"],
                }))
                .collect::<Vec<_>>(),
            vec![
                json!({"event": "codex_command_execution_event", "type": expected, "status": "completed", "duration": 42}),
                json!({"event": "codex_control_tool_call_event", "type": expected, "status": "completed", "duration": 42}),
            ],
        );
    }
}

#[tokio::test]
async fn tool_event_emitted_before_sampling_evidence_keeps_unknown_origin() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();
    ingest_review_prerequisites(&mut reducer, &mut events).await;
    ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-1", "call-1").await;
    assert_eq!(events.len(), 1);
    ingest_code_mode_facts(
        &mut reducer,
        &mut events,
        [sampling_response("turn-1", "response-1", &["call-1"])],
    )
    .await;
    reducer.flush(&mut events);
    let payload = serde_json::to_value(&events).expect("serialize tool events");
    assert_eq!(
        payload
            .as_array()
            .expect("events")
            .iter()
            .map(|event| json!({
                "item": event["event_params"]["item_id"],
                "type": event["event_params"]["tool_event_type"],
                "origin": event["event_params"]["originating_response_id"],
            }))
            .collect::<Vec<_>>(),
        vec![json!({"item": "call-1", "type": null, "origin": null})],
    );
}

#[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,
                GuardianApprovalReviewAction::Command {
                    source: AppServerGuardianCommandSource::Shell,
                    command: "echo hi".to_string(),
                    cwd: test_path_buf("/tmp").abs().into(),
                },
            ))),
            &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();

    reducer
        .ingest(
            AnalyticsFact::Notification(Box::new(sample_guardian_review_completed(
                "guardian-stdin-review-1",
                Some("item-1"),
                GuardianApprovalReviewStatus::Denied,
                GuardianApprovalReviewAction::WriteStdin {
                    approval_id: "stdin-approval-1".to_string(),
                    process_id: "42".to_string(),
                    stdin: "confirm\n".to_string(),
                    cwd: test_path_buf("/tmp").abs().into(),
                },
            ))),
            &mut events,
        )
        .await;
    let review_payload = serde_json::to_value(&events[0]).expect("serialize review event");
    assert_eq!(
        review_payload["event_params"]["subject_kind"],
        "write_stdin"
    );
    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",
                )),
                thread_originator: None,
            },
            &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: Some("codex-tui".to_string()),
            client_version: Some("1.0.0".to_string()),
            model: "gpt-5".to_string(),
            ephemeral: false,
            thread_source: Some(ThreadSource::Subagent),
            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: Some("codex-tui".to_string()),
            client_version: Some("1.0.0".to_string()),
            model: "gpt-5".to_string(),
            ephemeral: true,
            thread_source: Some(ThreadSource::Subagent),
            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: Some("codex-tui".to_string()),
            client_version: Some("1.0.0".to_string()),
            model: "gpt-5".to_string(),
            ephemeral: false,
            thread_source: Some(ThreadSource::Subagent),
            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: Some("codex-tui".to_string()),
            client_version: Some("1.0.0".to_string()),
            model: "gpt-5".to_string(),
            ephemeral: false,
            thread_source: Some(ThreadSource::GuardianReview),
            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"]["thread_source"], "guardian_review");
    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: Some("codex-tui".to_string()),
            client_version: Some("1.0.0".to_string()),
            model: "gpt-5".to_string(),
            ephemeral: false,
            thread_source: Some(ThreadSource::GuardianReview),
            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: Some("codex-tui".to_string()),
                    client_version: Some("1.0.0".to_string()),
                    model: "gpt-5".to_string(),
                    ephemeral: false,
                    thread_source: Some(ThreadSource::Subagent),
                    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 guardian_events_keep_thread_source_and_originator_with_explicit_turn_connection() {
    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",
                )),
                thread_originator: None,
            },
            &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: Some("parent-client".to_string()),
                    client_version: Some("1.0.0".to_string()),
                    model: "gpt-5".to_string(),
                    ephemeral: false,
                    thread_source: Some(ThreadSource::GuardianReview),
                    subagent_source: SubAgentSource::Other("guardian".to_string()),
                    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,
                    codex_error_kind: None,
                    codex_error_http_status_code: None,
                    active_context_tokens_before: 131_000,
                    active_context_tokens_after: 64_000,
                    retained_image_count: None,
                    compaction_summary_tokens: None,
                    cached_input_tokens: None,
                    cache_write_input_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"
    );

    events.clear();
    ingest_complete_child_turn(&mut reducer, &mut events, "thread-review", "turn-inherited").await;
    let [TrackEventRequest::TurnEvent(event)] = events.as_slice() else {
        panic!("expected one turn event");
    };
    let params = &event.event_params;
    assert_eq!(params.session_id, "session-root");
    assert_eq!(params.thread_source, Some(ThreadSource::GuardianReview));
    assert_eq!(params.subagent_source.as_deref(), Some("guardian"));
    assert_eq!(
        params.parent_thread_id.as_deref(),
        Some("44444444-4444-4444-4444-444444444444")
    );
    assert_eq!(params.app_server_client.product_client_id, "parent-client");
    assert_eq!(params.runtime.codex_rs_version, "0.1.0");

    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::TurnTokenUsage(Box::new(
                sample_turn_token_usage_fact("thread-review", "turn-inherited"),
            ))),
            &mut events,
        )
        .await;
    assert_eq!(events.len(), 1);

    events.clear();
    reducer
        .ingest(sample_initialize_fact(/*connection_id*/ 8), &mut events)
        .await;
    reducer
        .ingest(
            AnalyticsFact::ClientRequest {
                connection_id: 8,
                request_id: RequestId::Integer(3),
                request: Box::new(sample_turn_start_request(
                    "thread-review",
                    /*request_id*/ 3,
                )),
            },
            &mut events,
        )
        .await;
    reducer
        .ingest(
            AnalyticsFact::ClientResponse {
                connection_id: 8,
                request_id: RequestId::Integer(3),
                response: Box::new(sample_turn_start_response("turn-explicit")),
                thread_originator: None,
            },
            &mut events,
        )
        .await;
    ingest_complete_child_turn(&mut reducer, &mut events, "thread-review", "turn-explicit").await;
    let [TrackEventRequest::TurnEvent(event)] = events.as_slice() else {
        panic!("expected one turn event");
    };
    assert_eq!(
        event.event_params.app_server_client.product_client_id,
        "parent-client"
    );
    assert_eq!(
        event.event_params.app_server_client.client_name.as_deref(),
        Some("codex-tui")
    );
}

#[tokio::test]
async fn subagent_tool_items_inherit_parent_connection_metadata() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();

    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted(
                SubAgentThreadStartedInput {
                    session_id: "session-thread-1".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: Some("codex-tui".to_string()),
                    client_version: Some("1.0.0".to_string()),
                    model: "gpt-5".to_string(),
                    ephemeral: false,
                    thread_source: Some(ThreadSource::Subagent),
                    subagent_source: SubAgentSource::Review,
                    created_at: 128,
                },
            )),
            &mut events,
        )
        .await;
    ingest_review_prerequisites(&mut reducer, &mut events).await;
    for (thread_id, turn_id, root_turn_id) in [
        ("thread-1", "turn-parent", "parent-current-root"),
        ("thread-subagent", "turn-subagent", "child-causal-root"),
    ] {
        reducer
            .ingest(
                AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
                    TurnResolvedConfigFact {
                        turn_metadata: test_turn_metadata(Some(root_turn_id)),
                        ..sample_turn_resolved_config(thread_id, turn_id)
                    },
                ))),
                &mut events,
            )
            .await;
    }
    reducer
        .ingest(
            AnalyticsFact::Notification(Box::new(sample_turn_started_notification(
                "thread-subagent",
                "turn-subagent",
            ))),
            &mut events,
        )
        .await;

    ingest_code_mode_facts(
        &mut reducer,
        &mut events,
        [CodeModeToolCallFact::SamplingResponseCompleted {
            thread_id: "thread-subagent".into(),
            turn_id: "turn-subagent".into(),
            response_id: "response-subagent".into(),
            tool_call_ids: vec!["item-1".into(), "exec-1".into()],
        }],
    )
    .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;

    ingest_code_mode_facts(
        &mut reducer,
        &mut events,
        [CodeModeToolCallFact::Completed {
            thread_id: "thread-subagent".into(),
            turn_id: "turn-subagent".into(),
            turn_metadata: test_turn_metadata(Some("child-causal-root")),
            call_id: "exec-1".into(),
            cell_id: None,
            tool_name: "exec".into(),
            started_at_ms: 1_000,
            completed_at_ms: 1_042,
            status: CodeModeToolCallStatus::Completed,
        }],
    )
    .await;
    reducer.flush(&mut events);

    let payload = serde_json::to_value(&events).expect("serialize events");
    assert_eq!(payload.as_array().expect("events array").len(), 2);
    assert_eq!(
        payload
            .as_array()
            .expect("events array")
            .iter()
            .map(|event| json!({
                "turn_id": event["event_params"]["turn_id"],
                "root_turn_id": event["event_params"]["root_turn_id"],
                "tool_event_type": event["event_params"]["tool_event_type"],
            }))
            .collect::<Vec<_>>(),
        vec![
            json!({"turn_id": "turn-subagent", "root_turn_id": "child-causal-root", "tool_event_type": "model_tool_call"}),
            json!({"turn_id": "turn-subagent", "root_turn_id": "child-causal-root", "tool_event_type": "model_tool_call"}),
        ]
    );
    assert_eq!(payload[0]["event_type"], "codex_command_execution_event");
    assert_eq!(payload[0]["event_params"]["thread_id"], "thread-subagent");
    assert_eq!(payload[0]["event_params"]["session_id"], "session-thread-1");
    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"
    );
    assert_eq!(payload[1]["event_type"], "codex_dynamic_tool_call_event");
    assert_eq!(payload[1]["event_params"]["parent_thread_id"], "thread-1");
}

#[test]
fn plugin_used_event_serializes_expected_shape() {
    let tracking = test_tracking_context("thread-3", "turn-3");
    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",
                "remote_plugin_id": null,
                "plugin_name": "sample",
                "marketplace_name": "test",
                "has_skills": true,
                "mcp_server_count": 2,
                "connector_ids": ["calendar", "drive"],
                "product_client_id": TEST_PRODUCT_CLIENT_ID,
                "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",
                "remote_plugin_id": null,
                "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_install_failed_event_serializes_expected_shape() {
    let event = TrackEventRequest::PluginInstallFailed(CodexPluginInstallFailedEventRequest {
        event_type: "codex_plugin_install_failed",
        event_params: CodexPluginInstallFailedMetadata {
            plugin: codex_plugin_metadata(sample_plugin_metadata()),
            source: PluginInstallSource::Manual,
            error_type: "store_io".to_string(),
            sub_error_type: Some("failed_to_copy_plugin_file".to_string()),
        },
    });

    let payload = serde_json::to_value(&event).expect("serialize plugin install failed event");

    assert_eq!(
        payload,
        json!({
            "event_type": "codex_plugin_install_failed",
            "event_params": {
                "plugin_id": "sample@test",
                "remote_plugin_id": null,
                "plugin_name": "sample",
                "marketplace_name": "test",
                "has_skills": true,
                "mcp_server_count": 2,
                "connector_ids": ["calendar", "drive"],
                "product_client_id": originator().value,
                "source": "manual",
                "error_type": "store_io",
                "sub_error_type": "failed_to_copy_plugin_file"
            }
        })
    );
}

#[test]
fn plugin_management_event_keeps_plugin_id_local_when_remote_id_exists() {
    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,
        json!({
            "event_type": "codex_plugin_installed",
            "event_params": {
                "plugin_id": "sample@test",
                "remote_plugin_id": "plugins~Plugin_remote",
                "plugin_name": "sample",
                "marketplace_name": "test",
                "has_skills": true,
                "mcp_server_count": 2,
                "connector_ids": ["calendar", "drive"],
                "product_client_id": originator().value
            }
        })
    );
}

#[test]
fn hook_run_event_serializes_expected_shape() {
    let tracking = test_tracking_context("thread-3", "turn-3");
    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,
                handler_type: HookHandlerType::McpTool,
                execution_mode: HookExecutionMode::Sync,
                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",
                "product_client_id": TEST_PRODUCT_CLIENT_ID,
                "model_slug": "gpt-5",
                "hook_name": "PreToolUse",
                "hook_source": "user",
                "handler_type": "mcp_tool",
                "execution_mode": "sync",
                "status": "completed"
            }
        })
    );
}

#[test]
fn hook_run_metadata_maps_sources_and_statuses() {
    let tracking = test_tracking_context("thread-1", "turn-1");

    let system = serde_json::to_value(codex_hook_run_metadata(
        &tracking,
        HookRunFact {
            event_name: HookEventName::SessionStart,
            hook_source: HookSource::System,
            handler_type: HookHandlerType::Command,
            execution_mode: HookExecutionMode::Sync,
            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,
            handler_type: HookHandlerType::Prompt,
            execution_mode: HookExecutionMode::Async,
            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,
            handler_type: HookHandlerType::Agent,
            execution_mode: HookExecutionMode::Sync,
            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,
            handler_type: HookHandlerType::Command,
            execution_mode: HookExecutionMode::Async,
            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 = test_tracking_context("thread-1", "turn-1");

    let stopped = serde_json::to_value(codex_hook_run_metadata(
        &tracking,
        HookRunFact {
            event_name: HookEventName::Stop,
            hook_source: HookSource::User,
            handler_type: HookHandlerType::Command,
            execution_mode: HookExecutionMode::Sync,
            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 = test_tracking_context("thread-1", "turn-1");
    let turn_2 = test_tracking_context("thread-1", "turn-2");

    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_artifact_operation_fact() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();
    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::ArtifactOperation(
                ArtifactOperationInput {
                    tracking: test_tracking_context("thread-1", "turn-1"),
                    operation: ArtifactOperation {
                        item_id: "call-1".to_string(),
                        lifecycle: ArtifactOperationLifecycle::Started,
                        occurred_at_ms: 1_786_000_000_000,
                        plugin_id: "presentations@openai-primary-runtime".to_string(),
                        script_path: "skills/presentations/container_tools/mark_artifact_operation_started.mjs".to_string(),
                        skill: "presentations".to_string(),
                        artifact_type: "presentation".to_string(),
                        operation_kind: "create".to_string(),
                        expected_output_count: 2,
                        output_format: "pptx".to_string(),
                        execution_backend: "unified_exec".to_string(),
                    },
                },
            )),
            &mut events,
        )
        .await;

    assert!(events[0].can_send_with_api_key_auth());
    assert_eq!(
        serde_json::to_value(events).expect("serialize events"),
        json!([{
            "event_type": "codex_artifact_operation",
            "event_params": {
                "thread_id": "thread-1",
                "turn_id": "turn-1",
                "item_id": "call-1",
                "lifecycle": "started",
                "occurred_at_ms": 1_786_000_000_000_u64,
                "product_client_id": TEST_PRODUCT_CLIENT_ID,
                "runtime": serde_json::to_value(current_runtime_metadata())
                    .expect("serialize runtime metadata"),
                "model_slug": "gpt-5",
                "plugin_id": "presentations@openai-primary-runtime",
                "script_path": "skills/presentations/container_tools/mark_artifact_operation_started.mjs",
                "skill": "presentations",
                "artifact_type": "presentation",
                "operation_kind": "create",
                "expected_output_count": 2,
                "output_format": "pptx",
                "execution_backend": "unified_exec"
            }
        }])
    );
}

#[tokio::test]
async fn reducer_ingests_skill_invoked_fact() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();
    let tracking = test_tracking_context("thread-1", "turn-1");
    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(),
                    location: SkillInvocationLocation::Host {
                        path: skill_path,
                        scope: codex_protocol::protocol::SkillScope::User,
                    },
                    plugin_id: None,
                    remote_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": TEST_PRODUCT_CLIENT_ID,
                "skill_scope": "user",
                "plugin_id": null,
                "remote_plugin_id": null,
                "thread_id": "thread-1",
                "turn_id": "turn-1",
                "invoke_type": "explicit",
                "model_slug": "gpt-5"
            }
        }])
    );
}

#[tokio::test]
async fn reducer_includes_plugin_ids_for_plugin_skill_invocations() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();
    let tracking = test_tracking_context("thread-1", "turn-1");
    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(),
                    location: SkillInvocationLocation::Host {
                        path: skill_path,
                        scope: codex_protocol::protocol::SkillScope::User,
                    },
                    plugin_id: Some("sample@test".to_string()),
                    remote_plugin_id: Some("plugins~Plugin_sample".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"],
            &payload[0]["event_params"]["remote_plugin_id"],
        ),
        (&json!("sample@test"), &json!("plugins~Plugin_sample"))
    );
}

#[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: test_tracking_context("thread-1", "turn-1"),
                hook: HookRunFact {
                    event_name: HookEventName::PostToolUse,
                    hook_source: HookSource::Unknown,
                    handler_type: HookHandlerType::Agent,
                    execution_mode: HookExecutionMode::Async,
                    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"]["handler_type"], "agent");
    assert_eq!(payload[0]["event_params"]["execution_mode"], "async");
    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 = test_tracking_context("thread-1", "turn-1");

    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),
                },
                elicitation_type: Some(ElicitationType::AuthOrLink),
            })),
            &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[1]["event_params"]["elicitation_type"],
        "auth_or_link"
    );
    assert_eq!(payload[2]["event_type"], "codex_plugin_used");
    assert_eq!(
        payload[0]["event_params"]["product_client_id"],
        TEST_PRODUCT_CLIENT_ID
    );
    assert_eq!(
        payload[1]["event_params"]["product_client_id"],
        TEST_PRODUCT_CLIENT_ID
    );
    assert_eq!(
        payload[2]["event_params"]["product_client_id"],
        TEST_PRODUCT_CLIENT_ID
    );
}

#[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",
                "remote_plugin_id": null,
                "plugin_name": "sample",
                "marketplace_name": "test",
                "has_skills": true,
                "mcp_server_count": 2,
                "connector_ids": ["calendar", "drive"],
                "product_client_id": originator().value
            }
        }])
    );
}

#[tokio::test]
async fn reducer_ingests_plugin_install_requested_fact() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();
    let tracking = test_tracking_context("thread-1", "turn-1");
    let request = PluginInstallRequested {
        suggestion_id: "request_plugin_install_call-1".to_string(),
        plugins: vec![
            PluginInstallRequestedPlugin {
                plugin_id: "calendar@openai-curated-remote".to_string(),
                remote_plugin_id: Some("plugin_calendar".to_string()),
                plugin_name: "Calendar".to_string(),
                connector_ids: vec!["connector_calendar".to_string()],
            },
            PluginInstallRequestedPlugin {
                plugin_id: "github@openai-curated-remote".to_string(),
                remote_plugin_id: None,
                plugin_name: "GitHub".to_string(),
                connector_ids: vec!["connector_github".to_string()],
            },
        ],
        source: PluginInstallRequestSource::EndpointRecommendation,
    };

    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::PluginInstallRequested(
                PluginInstallRequestedInput { tracking, request },
            )),
            &mut events,
        )
        .await;

    assert_eq!(
        serde_json::to_value(&events).expect("serialize events"),
        json!([{
            "event_type": "codex_plugin_install_requested",
            "event_params": {
                "suggestion_id": "request_plugin_install_call-1",
                "plugins": [{
                    "plugin_id": "calendar@openai-curated-remote",
                    "remote_plugin_id": "plugin_calendar",
                    "plugin_name": "Calendar",
                    "connector_ids": ["connector_calendar"],
                }, {
                    "plugin_id": "github@openai-curated-remote",
                    "remote_plugin_id": null,
                    "plugin_name": "GitHub",
                    "connector_ids": ["connector_github"],
                }],
                "source": "endpoint_recommendation",
                "thread_id": "thread-1",
                "turn_id": "turn-1",
                "model_slug": "gpt-5",
                "product_client_id": originator().value,
            }
        }])
    );
}

#[tokio::test]
async fn reducer_ingests_plugin_install_failed_fact() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();

    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::PluginInstallFailed(
                PluginInstallFailedInput {
                    plugin: sample_plugin_metadata(),
                    source: PluginInstallSource::ExternalAgentMigration,
                    error_type: "invalid_plugin".to_string(),
                    sub_error_type: Some("failed_to_copy_plugin_file".to_string()),
                },
            )),
            &mut events,
        )
        .await;

    let payload = serde_json::to_value(&events).expect("serialize events");
    assert_eq!(
        payload,
        json!([{
            "event_type": "codex_plugin_install_failed",
            "event_params": {
                "plugin_id": "sample@test",
                "remote_plugin_id": null,
                "plugin_name": "sample",
                "marketplace_name": "test",
                "has_skills": true,
                "mcp_server_count": 2,
                "connector_ids": ["calendar", "drive"],
                "product_client_id": originator().value,
                "source": "external_agent_migration",
                "error_type": "invalid_plugin",
                "sub_error_type": "failed_to_copy_plugin_file"
            }
        }])
    );
}

#[tokio::test]
async fn reducer_ingests_plugin_install_failed_fact_without_detail() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();
    let plugin = PluginTelemetryMetadata {
        plugin_id: None,
        remote_plugin_id: Some("plugins~Plugin_00000000000000000000000000000000".to_string()),
        capability_summary: None,
    };

    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::PluginInstallFailed(
                PluginInstallFailedInput {
                    plugin,
                    source: PluginInstallSource::Manual,
                    error_type: "remote_catalog_unexpected_status".to_string(),
                    sub_error_type: None,
                },
            )),
            &mut events,
        )
        .await;

    let payload = serde_json::to_value(&events).expect("serialize events");
    assert_eq!(
        payload,
        json!([{
            "event_type": "codex_plugin_install_failed",
            "event_params": {
                "plugin_id": null,
                "remote_plugin_id": "plugins~Plugin_00000000000000000000000000000000",
                "plugin_name": null,
                "marketplace_name": null,
                "has_skills": null,
                "mcp_server_count": null,
                "connector_ids": null,
                "product_client_id": originator().value,
                "source": "manual",
                "error_type": "remote_catalog_unexpected_status",
                "sub_error_type": null
            }
        }])
    );
}

#[tokio::test]
async fn reducer_ingests_external_agent_config_import_completed_fact() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();

    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::ExternalAgentConfigImportCompleted(
                ExternalAgentConfigImportCompletedInput {
                    import_id: "import-1".to_string(),
                    source: "app_server".to_string(),
                    provider_id: "test-provider-42".to_string(),
                    item_type: "PLUGINS".to_string(),
                    success_count: 2,
                    failed_count: 1,
                },
            )),
            &mut events,
        )
        .await;

    let payload = serde_json::to_value(&events).expect("serialize events");
    assert_eq!(
        payload,
        json!([{
            "event_type": "codex_onboarding_external_agent_import_complete",
            "event_params": {
                "import_id": "import-1",
                "source": "app_server",
                "provider_id": "test-provider-42",
                "type": "PLUGINS",
                "success_count": 2,
                "failed_count": 1,
                "product_client_id": originator().value,
            }
        }])
    );
}

#[test]
fn external_agent_config_import_failure_event_serializes_expected_shape() {
    let event = TrackEventRequest::ExternalAgentConfigImportFailure(
        CodexOnboardingExternalAgentImportFailureEventRequest {
            event_type: "codex_onboarding_external_agent_import_failure",
            event_params: CodexOnboardingExternalAgentImportFailureMetadata {
                import_id: "import-1".to_string(),
                source: "app_server".to_string(),
                provider_id: "test-provider-42".to_string(),
                item_type: "PLUGINS".to_string(),
                failure_stage: "plugin_import".to_string(),
                error_type: "plugin_import".to_string(),
                sub_error_type: Some("failed_to_copy_plugin_file".to_string()),
                product_client_id: Some(originator().value),
            },
        },
    );

    let payload = serde_json::to_value(&event).expect("serialize import failure event");

    assert_eq!(
        payload,
        json!({
            "event_type": "codex_onboarding_external_agent_import_failure",
            "event_params": {
                "import_id": "import-1",
                "source": "app_server",
                "provider_id": "test-provider-42",
                "type": "PLUGINS",
                "failure_stage": "plugin_import",
                "error_type": "plugin_import",
                "sub_error_type": "failed_to_copy_plugin_file",
                "product_client_id": originator().value,
            }
        })
    );
}

#[tokio::test]
async fn reducer_ingests_external_agent_config_import_failure_fact() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();

    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::ExternalAgentConfigImportFailure(
                ExternalAgentConfigImportFailureInput {
                    import_id: "import-1".to_string(),
                    source: "app_server".to_string(),
                    provider_id: "test-provider-42".to_string(),
                    item_type: "PLUGINS".to_string(),
                    failure_stage: "plugin_import".to_string(),
                    error_type: "plugin_import".to_string(),
                    sub_error_type: Some("failed_to_copy_plugin_file".to_string()),
                },
            )),
            &mut events,
        )
        .await;

    let payload = serde_json::to_value(&events).expect("serialize events");
    assert_eq!(
        payload,
        json!([{
            "event_type": "codex_onboarding_external_agent_import_failure",
            "event_params": {
                "import_id": "import-1",
                "source": "app_server",
                "provider_id": "test-provider-42",
                "type": "PLUGINS",
                "failure_stage": "plugin_import",
                "error_type": "plugin_import",
                "sub_error_type": "failed_to_copy_plugin_file",
                "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(),
            root_turn_id: Some("turn-2".to_string()),
            turn_trigger: Some("user".to_string()),
            codex_turn_source: Some("composer".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(),
            guardian_v2_enabled: true,
            sandbox_network_access: true,
            collaboration_mode: Some("plan"),
            personality: Some("pragmatic".to_string()),
            workspace_kind: Some("projectless".to_string()),
            num_input_images: 2,
            image_preparations: vec![ImagePreparationMetadata {
                message_role: Some("user".to_string()),
                item_id: None,
                effective_detail: ImageDetailSetting::High,
                source_width: 2_048,
                source_height: 2_048,
                prepared_width: 1_600,
                prepared_height: 1_600,
            }],
            is_first_turn: true,
            status: Some(TurnStatus::Completed),
            explicit_client_interrupt_requested_at_ms: None,
            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,
            cache_write_input_tokens: None,
            output_tokens: None,
            reasoning_output_tokens: None,
            total_tokens: None,
            before_first_sampling_ms: 100,
            sampling_ms: 700,
            compaction_ms: 40,
            between_sampling_overhead_ms: 50,
            tool_blocking_ms: 250,
            after_last_sampling_ms: 94,
            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",
                "root_turn_id": "turn-2",
                "turn_trigger": "user",
                "codex_turn_source": "composer",
                "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",
                "guardian_v2_enabled": true,
                "sandbox_network_access": true,
                "collaboration_mode": "plan",
                "personality": "pragmatic",
                "workspace_kind": "projectless",
                "num_input_images": 2,
                "image_preparations": [{
                    "message_role": "user",
                    "item_id": null,
                    "effective_detail": "high",
                    "source_width": 2048,
                    "source_height": 2048,
                    "prepared_width": 1600,
                    "prepared_height": 1600
                }],
                "is_first_turn": true,
                "status": "completed",
                "explicit_client_interrupt_requested_at_ms": null,
                "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,
                "cache_write_input_tokens": null,
                "output_tokens": null,
                "reasoning_output_tokens": null,
                "total_tokens": null,
                "before_first_sampling_ms": 100,
                "sampling_ms": 700,
                "compaction_ms": 40,
                "between_sampling_overhead_ms": 50,
                "tool_blocking_ms": 250,
                "after_last_sampling_ms": 94,
                "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")),
                thread_originator: None,
            },
            &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")),
                thread_originator: None,
            },
            &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"].get("turn_trigger"),
            payload["event_params"].get("codex_turn_source"),
        ),
        (
            Some(&serde_json::Value::Null),
            Some(&serde_json::Value::Null)
        )
    );
    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"]["guardian_v2_enabled"], json!(false));
    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"]["cache_write_input_tokens"],
        json!(7)
    );
    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 image_generation_events_preserve_transparent_background_metadata() {
    for (status, transparent_background) in [
        ("completed", Some(true)),
        ("completed", Some(false)),
        ("completed", None),
        ("failed", None),
    ] {
        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 item = ThreadItem::ImageGeneration(ImageGenerationItem {
            id: "image-1".to_string(),
            status: status.to_string(),
            revised_prompt: None,
            result: "ok".to_string(),
            transparent_background,
            failure: None,
            saved_path: None,
            imagegen_request_id: None,
            generation_id: None,
        });

        reducer
            .ingest(
                AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted(
                    ItemStartedNotification {
                        thread_id: "thread-2".to_string(),
                        turn_id: "turn-2".to_string(),
                        started_at_ms: 998,
                        item: item.clone(),
                    },
                ))),
                &mut out,
            )
            .await;
        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;

        let event = out
            .iter()
            .find(|event| matches!(event, TrackEventRequest::ImageGeneration(_)))
            .expect("image generation event should be emitted");
        let payload = serde_json::to_value(event).expect("serialize image generation event");

        assert_eq!(
            payload["event_params"].get("transparent_background"),
            Some(&json!(transparent_background))
        );
        assert_eq!(payload["event_params"]["terminal_status"], json!(status));
    }
}

#[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*/ false,
        /*include_started*/ true,
        /*include_token_usage*/ false,
    )
    .await;
    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
                TurnResolvedConfigFact {
                    turn_metadata: test_turn_metadata(Some("root-ancestor")),
                    ..sample_turn_resolved_config("thread-2", "turn-2")
                },
            ))),
            &mut out,
        )
        .await;

    let mcp_tool_call_item = |status, duration_ms| ThreadItem::McpToolCall {
        id: "mcp-1".to_string(),
        server: "server".to_string(),
        tool: "search".to_string(),
        status,
        arguments: json!({}),
        app_context: Some(McpToolCallAppContext {
            connector_id: "connector-test".to_string(),
            link_id: None,
            resource_uri: None,
            app_name: None,
            action_name: None,
        }),
        mcp_app_resource_uri: None,
        mcp_app_ui: None,
        plugin_id: Some("sample@test".to_string()),
        read_only_hint: None,
        result: None,
        error: None,
        duration_ms,
    };
    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,
        },
        mcp_tool_call_item(McpToolCallStatus::Completed, Some(2)),
        ThreadItem::DynamicToolCall {
            id: "dynamic-1".to_string(),
            // A name collision with code-mode exec must not change the event boundary.
            namespace: Some("custom".to_string()),
            tool: "exec".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::SubAgentActivity {
            id: "sub-agent-activity-completed".to_string(),
            kind: SubAgentActivityKind::Completed,
            agent_thread_id: "thread-child".to_string(),
            agent_path: "/root/child".to_string(),
        },
        ThreadItem::WebSearch(WebSearchItem {
            id: "web-1".to_string(),
            query: "codex".to_string(),
            action: None,
            results: None,
        }),
        ThreadItem::ImageGeneration(ImageGenerationItem {
            id: "image-1".to_string(),
            status: "completed".to_string(),
            revised_prompt: None,
            result: "ok".to_string(),
            transparent_background: None,
            failure: None,
            saved_path: None,
            imagegen_request_id: Some("req-imagegen-123".to_string()),
            generation_id: Some("gen-image-123".to_string()),
        }),
    ];

    for item in &completed_tool_items {
        reducer
            .ingest(
                AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted(
                    ItemStartedNotification {
                        thread_id: "thread-2".to_string(),
                        turn_id: "turn-2".to_string(),
                        started_at_ms: 998,
                        item: item.clone(),
                    },
                ))),
                &mut out,
            )
            .await;
    }

    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::Custom(CustomAnalyticsFact::ControlToolCall(ControlToolCallFact {
                thread_id: "thread-2".to_string(),
                turn_id: "turn-2".to_string(),
                turn_metadata: test_turn_metadata(Some("root-ancestor")),
                call_id: "control-1".to_string(),
                cell_id: None,
                tool_name: "update_plan".to_string(),
                started_at_ms: 998,
                completed_at_ms: 1_000,
                status: ControlToolCallStatus::Completed,
            })),
            &mut out,
        )
        .await;
    reducer.flush(&mut out);

    let payload = serde_json::to_value(&out).expect("serialize tool item events");
    let emitted_tool_events = payload
        .as_array()
        .expect("tool item events array")
        .iter()
        .map(|event| {
            (
                event["event_type"].as_str().expect("tool item event type"),
                event["event_params"]["session_id"]
                    .as_str()
                    .expect("tool item event session ID"),
                event["event_params"]["turn_id"]
                    .as_str()
                    .expect("tool item event turn ID"),
                event["event_params"]["root_turn_id"]
                    .as_str()
                    .expect("tool item event root turn ID"),
                event["event_params"]["tool_event_type"].as_str(),
            )
        })
        .collect::<Vec<_>>();
    assert_eq!(
        emitted_tool_events,
        [
            ("codex_command_execution_event", None),
            ("codex_file_change_event", None),
            ("codex_mcp_tool_call_event", None),
            ("codex_dynamic_tool_call_event", None),
            ("codex_web_search_event", None),
            ("codex_image_generation_event", None),
            ("codex_collab_agent_tool_call_event", None),
            ("codex_control_tool_call_event", None),
        ]
        .map(|(event_type, tool_event_type)| {
            (
                event_type,
                "session-thread-2",
                "turn-2",
                "root-ancestor",
                tool_event_type,
            )
        })
        .to_vec()
    );

    let image_generation_event = out
        .iter()
        .find(|event| matches!(event, TrackEventRequest::ImageGeneration(_)))
        .expect("image generation event should be emitted");
    let payload =
        serde_json::to_value(image_generation_event).expect("serialize image generation event");
    assert_eq!(
        payload["event_params"]["imagegen_request_id"],
        json!("req-imagegen-123")
    );
    assert_eq!(
        payload["event_params"]["generation_id"],
        json!("gen-image-123")
    );

    let mcp_tool_call_event = out
        .iter()
        .find(|event| matches!(event, TrackEventRequest::McpToolCall(_)))
        .expect("MCP tool call event should be emitted");
    let payload = serde_json::to_value(mcp_tool_call_event).expect("serialize MCP tool call event");
    assert_eq!(payload["event_params"]["plugin_id"], json!("sample@test"));
    assert_eq!(
        payload["event_params"]["connector_id"],
        json!("connector-test")
    );

    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"]["root_turn_id"], "root-ancestor");
    assert_eq!(payload["event_params"]["total_tool_call_count"], json!(9));
    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 mcp_elicitation_classification_survives_turn_completion_and_preserves_call_grain() {
    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*/ false,
    )
    .await;

    let mut items = Vec::new();
    for (item_id, connector_id, elicitation_type) in [
        ("auth", "calendar", Some(ElicitationType::AuthOrLink)),
        ("retry", "calendar", None),
        ("denied", "drive", Some(ElicitationType::Approval)),
    ] {
        let item = ThreadItem::McpToolCall {
            id: item_id.to_string(),
            server: "server".to_string(),
            tool: "search".to_string(),
            status: McpToolCallStatus::Completed,
            arguments: json!({ "token": "synthetic-private-input" }),
            app_context: Some(McpToolCallAppContext {
                connector_id: connector_id.to_string(),
                link_id: None,
                resource_uri: None,
                app_name: None,
                action_name: None,
            }),
            mcp_app_resource_uri: None,
            mcp_app_ui: None,
            plugin_id: None,
            read_only_hint: None,
            result: None,
            error: None,
            duration_ms: Some(2),
        };
        reducer
            .ingest(
                AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted(
                    ItemStartedNotification {
                        thread_id: "thread-2".to_string(),
                        turn_id: "turn-2".to_string(),
                        started_at_ms: 998,
                        item: item.clone(),
                    },
                ))),
                &mut events,
            )
            .await;
        if let Some(elicitation_type) = elicitation_type {
            reducer
                .ingest(
                    AnalyticsFact::Custom(CustomAnalyticsFact::McpToolCallElicitation(
                        McpToolCallElicitation {
                            thread_id: "thread-2".to_string(),
                            turn_id: "turn-2".to_string(),
                            item_id: item_id.to_string(),
                            elicitation_type,
                        },
                    )),
                    &mut events,
                )
                .await;
        }
        items.push(item);
    }
    reducer
        .ingest(
            AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
                "thread-2",
                "turn-2",
                AppServerTurnStatus::Completed,
                /*codex_error_info*/ None,
            ))),
            &mut events,
        )
        .await;
    for item in 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 events,
            )
            .await;
    }

    let payload = serde_json::to_value(&events).expect("serialize analytics events");
    let classifications = payload
        .as_array()
        .expect("analytics events array")
        .iter()
        .filter(|event| event["event_type"] == "codex_mcp_tool_call_event")
        .map(|event| {
            json!({
                "item_id": event["event_params"]["item_id"],
                "connector_id": event["event_params"]["connector_id"],
                "elicitation_type": event["event_params"].get("elicitation_type")
                    .expect("elicitation_type must be present"),
            })
        })
        .collect::<Vec<_>>();
    assert_eq!(
        classifications,
        vec![
            json!({"item_id": "auth", "connector_id": "calendar", "elicitation_type": "auth_or_link"}),
            json!({"item_id": "retry", "connector_id": "calendar", "elicitation_type": null}),
            json!({"item_id": "denied", "connector_id": "drive", "elicitation_type": "approval"}),
        ]
    );
    assert!(!payload.to_string().contains("synthetic-private-input"));
}

#[tokio::test]
async fn turn_and_tool_events_read_current_trusted_root() {
    let mut reducer = AnalyticsReducer::default();
    let mut events = Vec::new();
    ingest_review_prerequisites(&mut reducer, &mut events).await;
    let turn_metadata = test_turn_metadata(Some("root-ancestor"));
    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
                TurnResolvedConfigFact {
                    turn_metadata: turn_metadata.clone(),
                    ..sample_turn_resolved_config("thread-1", "turn-1")
                },
            ))),
            &mut events,
        )
        .await;

    ingest_completed_command_execution_item(
        &mut reducer,
        &mut events,
        "thread-1",
        "before-conflict",
    )
    .await;
    let queued_control_call = ControlToolCallFact {
        thread_id: "thread-1".to_string(),
        turn_id: "turn-1".to_string(),
        turn_metadata: turn_metadata.clone(),
        call_id: "queued-control".to_string(),
        cell_id: None,
        tool_name: "view_image".to_string(),
        started_at_ms: 998,
        completed_at_ms: 1_000,
        status: ControlToolCallStatus::Completed,
    };
    let queued_code_mode_call = CodeModeToolCallFact::Completed {
        thread_id: "thread-1".to_string(),
        turn_id: "turn-1".to_string(),
        turn_metadata: turn_metadata.clone(),
        call_id: "queued-code-mode".to_string(),
        cell_id: None,
        tool_name: "exec".to_string(),
        started_at_ms: 998,
        completed_at_ms: 1_000,
        status: CodeModeToolCallStatus::Completed,
    };
    *turn_metadata.root_turn_id.lock().expect("root turn ID") = None;
    ingest_completed_command_execution_item(
        &mut reducer,
        &mut events,
        "thread-1",
        "after-conflict",
    )
    .await;
    for fact in [
        AnalyticsFact::Custom(CustomAnalyticsFact::ControlToolCall(queued_control_call)),
        AnalyticsFact::Custom(CustomAnalyticsFact::CodeModeToolCall(queued_code_mode_call)),
        AnalyticsFact::Custom(CustomAnalyticsFact::TurnProfile(Box::new(
            TurnProfileFact {
                turn_id: "turn-1".to_string(),
                profile: sample_turn_profile(),
            },
        ))),
        AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
            "thread-1",
            "turn-1",
            AppServerTurnStatus::Completed,
            /*codex_error_info*/ None,
        ))),
    ] {
        reducer.ingest(fact, &mut events).await;
    }

    let payload = serde_json::to_value(&events).expect("serialize events");
    assert_eq!(
        payload
            .as_array()
            .expect("events array")
            .iter()
            .map(|event| json!({
                "event_type": event["event_type"],
                "turn_id": event["event_params"]["turn_id"],
                "item_id": event["event_params"].get("item_id"),
                "root_turn_id": event["event_params"].get("root_turn_id").expect("root field"),
            }))
            .collect::<Vec<_>>(),
        vec![
            json!({"event_type": "codex_command_execution_event", "turn_id": "turn-1", "item_id": "before-conflict", "root_turn_id": "root-ancestor"}),
            json!({"event_type": "codex_command_execution_event", "turn_id": "turn-1", "item_id": "after-conflict", "root_turn_id": null}),
            json!({"event_type": "codex_control_tool_call_event", "turn_id": "turn-1", "item_id": "queued-control", "root_turn_id": null}),
            json!({"event_type": "codex_dynamic_tool_call_event", "turn_id": "turn-1", "item_id": "queued-code-mode", "root_turn_id": null}),
            json!({"event_type": "codex_turn_event", "turn_id": "turn-1", "item_id": null, "root_turn_id": null}),
        ]
    );
}

#[tokio::test]
async fn completed_background_tool_item_emits_after_turn_event() {
    let mut reducer = AnalyticsReducer::default();
    let mut out = Vec::new();
    let turn_metadata = test_turn_metadata(Some("root-background"));

    ingest_turn_prerequisites(
        &mut reducer,
        &mut out,
        /*include_initialize*/ true,
        /*include_resolved_config*/ false,
        /*include_started*/ true,
        /*include_token_usage*/ false,
    )
    .await;
    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::TurnResolvedConfig(Box::new(
                TurnResolvedConfigFact {
                    turn_metadata: turn_metadata.clone(),
                    ..sample_turn_resolved_config("thread-2", "turn-2")
                },
            ))),
            &mut out,
        )
        .await;

    reducer
        .ingest(
            AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted(
                ItemStartedNotification {
                    thread_id: "thread-2".to_string(),
                    turn_id: "turn-2".to_string(),
                    started_at_ms: 998,
                    item: sample_command_execution_item(
                        CommandExecutionStatus::InProgress,
                        /*exit_code*/ None,
                        /*duration_ms*/ None,
                    ),
                },
            ))),
            &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_eq!(
        out.iter()
            .filter(|event| matches!(event, TrackEventRequest::TurnEvent(_)))
            .count(),
        1
    );
    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;

    assert_eq!(
        out.iter()
            .filter(|event| matches!(event, TrackEventRequest::TurnEvent(_)))
            .count(),
        1
    );
    assert!(
        out.iter()
            .any(|event| matches!(event, TrackEventRequest::CommandExecution(_)))
    );
    reducer
        .ingest(
            AnalyticsFact::Custom(CustomAnalyticsFact::ControlToolCall(ControlToolCallFact {
                thread_id: "thread-2".to_string(),
                turn_id: "turn-2".to_string(),
                turn_metadata: turn_metadata.clone(),
                call_id: "late-view-image".to_string(),
                cell_id: None,
                tool_name: "view_image".to_string(),
                started_at_ms: 998,
                completed_at_ms: 1_001,
                status: ControlToolCallStatus::Completed,
            })),
            &mut out,
        )
        .await;
    ingest_code_mode_facts(
        &mut reducer,
        &mut out,
        [CodeModeToolCallFact::Completed {
            thread_id: "thread-2".to_string(),
            turn_id: "turn-2".to_string(),
            turn_metadata,
            call_id: "late-code-mode".to_string(),
            cell_id: None,
            tool_name: "exec".to_string(),
            started_at_ms: 998,
            completed_at_ms: 1_001,
            status: CodeModeToolCallStatus::Interrupted,
        }],
    )
    .await;
    reducer.flush(&mut out);

    assert_eq!(
        out.iter()
            .map(|event| {
                let event = serde_json::to_value(event).expect("serialize event");
                json!({
                    "event_type": event["event_type"],
                    "turn_id": event["event_params"]["turn_id"],
                    "root_turn_id": event["event_params"]["root_turn_id"],
                })
            })
            .collect::<Vec<_>>(),
        vec![
            json!({"event_type": "codex_turn_event", "turn_id": "turn-2", "root_turn_id": "root-background"}),
            json!({"event_type": "codex_command_execution_event", "turn_id": "turn-2", "root_turn_id": "root-background"}),
            json!({"event_type": "codex_control_tool_call_event", "turn_id": "turn-2", "root_turn_id": "root-background"}),
            json!({"event_type": "codex_dynamic_tool_call_event", "turn_id": "turn-2", "root_turn_id": "root-background"}),
        ]
    );
}

#[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")),
                thread_originator: None,
            },
            &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")),
                thread_originator: None,
            },
            &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 rejected_turn_interrupt_does_not_tag_interrupted_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::ExplicitClientInterruptRequest {
                connection_id: 7,
                request_id: RequestId::Integer(4),
                turn_id: "turn-2".to_string(),
                requested_at_ms: 1716000000123,
            },
            &mut out,
        )
        .await;
    reducer
        .ingest(
            AnalyticsFact::ErrorResponse {
                connection_id: 7,
                request_id: RequestId::Integer(4),
                error: no_active_turn_steer_error(),
                error_type: None,
            },
            &mut out,
        )
        .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"]["explicit_client_interrupt_requested_at_ms"],
        json!(null)
    );
    assert_eq!(payload["event_params"]["turn_error"], json!(null));
    assert_eq!(payload["event_params"]["codex_error_kind"], json!(null));
}

#[tokio::test]
async fn accepted_turn_interrupt_records_requested_at_on_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::ExplicitClientInterruptRequest {
                connection_id: 7,
                request_id: RequestId::Integer(4),
                turn_id: "turn-2".to_string(),
                requested_at_ms: 1716000000123,
            },
            &mut out,
        )
        .await;
    reducer
        .ingest(
            AnalyticsFact::ClientResponse {
                connection_id: 7,
                request_id: RequestId::Integer(4),
                response: Box::new(sample_turn_interrupt_response()),
                thread_originator: None,
            },
            &mut out,
        )
        .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"]["explicit_client_interrupt_requested_at_ms"],
        json!(1716000000123_u64)
    );
}

#[tokio::test]
async fn accepted_turn_interrupt_retries_preserve_earliest_requested_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*/ true,
        /*include_token_usage*/ false,
    )
    .await;
    reducer
        .ingest(
            AnalyticsFact::ExplicitClientInterruptRequest {
                connection_id: 7,
                request_id: RequestId::Integer(4),
                turn_id: "turn-2".to_string(),
                requested_at_ms: 1716000000123,
            },
            &mut out,
        )
        .await;
    reducer
        .ingest(
            AnalyticsFact::ExplicitClientInterruptRequest {
                connection_id: 7,
                request_id: RequestId::Integer(5),
                turn_id: "turn-2".to_string(),
                requested_at_ms: 1716000000456,
            },
            &mut out,
        )
        .await;
    reducer
        .ingest(
            AnalyticsFact::ClientResponse {
                connection_id: 7,
                request_id: RequestId::Integer(5),
                response: Box::new(sample_turn_interrupt_response()),
                thread_originator: None,
            },
            &mut out,
        )
        .await;
    reducer
        .ingest(
            AnalyticsFact::ClientResponse {
                connection_id: 7,
                request_id: RequestId::Integer(4),
                response: Box::new(sample_turn_interrupt_response()),
                thread_originator: None,
            },
            &mut out,
        )
        .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"]["explicit_client_interrupt_requested_at_ms"],
        json!(1716000000123_u64)
    );
}

#[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: Some(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(),
            plugin_namespace: None,
            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()),
            ],
        }),
    }
}