codex-rs/analytics/src/client_tests.rs

use super::AnalyticsEventsClient;
use super::AnalyticsEventsDestination;
use super::AnalyticsEventsQueue;
use super::AnalyticsEventsQueueMessage;
#[cfg(debug_assertions)]
use super::capture_track_events_request;
#[cfg(debug_assertions)]
use super::send_track_events;
#[cfg(debug_assertions)]
use super::send_track_events_request;
use super::track_event_request_batches;
#[cfg(debug_assertions)]
use crate::events::AppServerRpcTransport;
use crate::events::CodexAcceptedLineFingerprintsEventParams;
use crate::events::CodexAcceptedLineFingerprintsEventRequest;
#[cfg(debug_assertions)]
use crate::events::CodexAppServerClientMetadata;
#[cfg(debug_assertions)]
use crate::events::CodexMcpToolCallEventParams;
#[cfg(debug_assertions)]
use crate::events::CodexMcpToolCallEventRequest;
#[cfg(debug_assertions)]
use crate::events::CodexPluginMeasurementEventParams;
#[cfg(debug_assertions)]
use crate::events::CodexPluginMeasurementEventRequest;
#[cfg(debug_assertions)]
use crate::events::CodexPluginMetadata;
#[cfg(debug_assertions)]
use crate::events::CodexPluginUsedEventRequest;
#[cfg(debug_assertions)]
use crate::events::CodexPluginUsedMetadata;
#[cfg(debug_assertions)]
use crate::events::CodexRuntimeMetadata;
#[cfg(debug_assertions)]
use crate::events::CodexToolItemEventBase;
#[cfg(debug_assertions)]
use crate::events::FinalApprovalOutcome;
use crate::events::SkillInvocationEventParams;
use crate::events::SkillInvocationEventRequest;
#[cfg(debug_assertions)]
use crate::events::ThreadArchiveAction;
#[cfg(debug_assertions)]
use crate::events::ThreadArchiveEvent;
#[cfg(debug_assertions)]
use crate::events::ThreadArchiveEventParams;
#[cfg(debug_assertions)]
use crate::events::ToolItemTerminalStatus;
use crate::events::TrackEventRequest;
#[cfg(debug_assertions)]
use crate::events::codex_artifact_operation_event_request;
use crate::facts::AnalyticsFact;
use crate::facts::AppInvocation;
#[cfg(debug_assertions)]
use crate::facts::ArtifactOperation;
#[cfg(debug_assertions)]
use crate::facts::ArtifactOperationLifecycle;
use crate::facts::CustomAnalyticsFact;
use crate::facts::ElicitationType;
use crate::facts::InvocationType;
use crate::facts::PluginMeasurementRow;
use crate::facts::PluginMeasurementsInput;
use crate::facts::TrackEventsContext;
use crate::reducer::MAX_PLUGIN_MEASUREMENTS_PER_BATCH;
use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer;
use codex_app_server_protocol::AskForApproval as AppServerAskForApproval;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponsePayload;
use codex_app_server_protocol::CommandExecutionOutputDeltaNotification;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::SessionSource as AppServerSessionSource;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadArchiveParams;
use codex_app_server_protocol::ThreadArchiveResponse;
use codex_app_server_protocol::ThreadArchivedNotification;
use codex_app_server_protocol::ThreadForkResponse;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnDiffUpdatedNotification;
use codex_app_server_protocol::TurnInterruptParams;
use codex_app_server_protocol::TurnInterruptResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus as AppServerTurnStatus;
use codex_app_server_protocol::TurnSteerParams;
use codex_app_server_protocol::TurnSteerResponse;
#[cfg(debug_assertions)]
use codex_login::AuthManager;
use codex_utils_absolute_path::test_support::PathBufExt;
use codex_utils_absolute_path::test_support::test_path_buf;
use pretty_assertions::assert_eq;
use std::collections::BTreeMap;
use std::collections::HashSet;
#[cfg(debug_assertions)]
use std::fs;
#[cfg(debug_assertions)]
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
#[cfg(debug_assertions)]
use std::time::SystemTime;
use tokio::sync::mpsc;
use tokio::sync::mpsc::error::TryRecvError;

#[cfg(debug_assertions)]
impl AnalyticsEventsClient {
    pub(crate) fn new_for_capture_file(auth_manager: Arc<AuthManager>, path: PathBuf) -> Self {
        Self {
            queue: Some(AnalyticsEventsQueue::new(
                auth_manager,
                AnalyticsEventsDestination::CaptureFile { path },
            )),
        }
    }
}

fn sample_accepted_line_fingerprint_event(thread_id: &str) -> TrackEventRequest {
    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_id.to_string(),
                product_surface: Some("codex".to_string()),
                model_slug: Some("gpt-5.1-codex".to_string()),
                completed_at: 1,
                repo_hash: None,
                accepted_added_lines: 1,
                accepted_deleted_lines: 0,
                line_fingerprints: [],
            },
        },
    ))
}

fn sample_skill_track_event(thread_id: &str, plugin_id: Option<&str>) -> TrackEventRequest {
    TrackEventRequest::SkillInvocation(SkillInvocationEventRequest {
        event_type: "skill_invocation",
        skill_id: format!("skill-{thread_id}"),
        skill_name: "doc".to_string(),
        event_params: SkillInvocationEventParams {
            product_client_id: None,
            skill_scope: None,
            plugin_id: plugin_id.map(str::to_string),
            remote_plugin_id: None,
            thread_id: Some(thread_id.to_string()),
            turn_id: Some("turn-1".to_string()),
            invoke_type: Some(InvocationType::Explicit),
            model_slug: Some("gpt-5.1-codex".to_string()),
        },
    })
}

#[cfg(debug_assertions)]
fn sample_artifact_operation_event(thread_id: &str) -> TrackEventRequest {
    TrackEventRequest::ArtifactOperation(codex_artifact_operation_event_request(
        TrackEventsContext {
            model_slug: "gpt-5.1-codex".to_string(),
            thread_id: thread_id.to_string(),
            turn_id: "turn-1".to_string(),
            product_client_id: "codex_desktop".to_string(),
        },
        ArtifactOperation {
            item_id: format!("item-{thread_id}"),
            lifecycle: ArtifactOperationLifecycle::Started,
            occurred_at_ms: 1,
            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: 1,
            output_format: "pptx".to_string(),
            execution_backend: "unified_exec".to_string(),
        },
    ))
}

fn sample_regular_track_event(thread_id: &str) -> TrackEventRequest {
    sample_skill_track_event(thread_id, /*plugin_id*/ None)
}

#[cfg(debug_assertions)]
fn sample_mcp_tool_call_event(thread_id: &str, plugin_id: Option<&str>) -> TrackEventRequest {
    TrackEventRequest::McpToolCall(CodexMcpToolCallEventRequest {
        event_type: "codex_mcp_tool_call_event",
        event_params: CodexMcpToolCallEventParams {
            base: CodexToolItemEventBase {
                thread_id: thread_id.to_string(),
                session_id: format!("session-{thread_id}"),
                turn_id: "turn-1".to_string(),
                root_turn_id: None,
                item_id: format!("item-{thread_id}"),
                cell_id: None,
                parent_call_id: None,
                originating_response_id: None,
                subsequent_response_id: None,
                app_server_client: CodexAppServerClientMetadata {
                    product_client_id: "codex_desktop".to_string(),
                    client_name: None,
                    client_version: None,
                    rpc_transport: AppServerRpcTransport::InProcess,
                    experimental_api_enabled: None,
                },
                runtime: CodexRuntimeMetadata {
                    codex_rs_version: "0.0.0".to_string(),
                    runtime_os: "test".to_string(),
                    runtime_os_version: "test".to_string(),
                    runtime_arch: "test".to_string(),
                },
                thread_source: None,
                subagent_source: None,
                parent_thread_id: None,
                tool_name: "search".to_string(),
                tool_event_type: None,
                started_at_ms: 1,
                completed_at_ms: 2,
                duration_ms: Some(1),
                execution_duration_ms: Some(1),
                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,
            },
            mcp_server_name: "sample".to_string(),
            mcp_tool_name: "search".to_string(),
            mcp_error_present: false,
            plugin_id: plugin_id.map(str::to_string),
            connector_id: None,
            elicitation_type: None,
        },
    })
}

#[cfg(debug_assertions)]
fn sample_plugin_used_track_event(thread_id: &str, plugin_id: Option<&str>) -> TrackEventRequest {
    TrackEventRequest::PluginUsed(CodexPluginUsedEventRequest {
        event_type: "codex_plugin_used",
        event_params: CodexPluginUsedMetadata {
            plugin: CodexPluginMetadata {
                plugin_id: plugin_id.map(str::to_string),
                remote_plugin_id: None,
                plugin_name: Some("sample".to_string()),
                marketplace_name: Some("test".to_string()),
                has_skills: Some(true),
                mcp_server_count: Some(1),
                connector_ids: Some(vec!["calendar".to_string()]),
                product_client_id: Some("codex_desktop".to_string()),
            },
            mcp_server_names: Some(vec!["mcp-1".to_string()]),
            thread_id: Some(thread_id.to_string()),
            turn_id: Some("turn-1".to_string()),
            model_slug: Some("gpt-5.1-codex".to_string()),
        },
    })
}

#[cfg(debug_assertions)]
fn unique_capture_path(name: &str) -> PathBuf {
    let nonce = SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .expect("system clock should be after Unix epoch")
        .as_nanos();
    std::env::temp_dir().join(format!(
        "codex-analytics-{name}-{}-{nonce}.jsonl",
        std::process::id()
    ))
}

fn client_with_receiver() -> (
    AnalyticsEventsClient,
    mpsc::Receiver<AnalyticsEventsQueueMessage>,
) {
    let (sender, receiver) = mpsc::channel(8);
    let queue = AnalyticsEventsQueue {
        sender,
        app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
        plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())),
    };
    (AnalyticsEventsClient { queue: Some(queue) }, receiver)
}

#[test]
#[cfg(debug_assertions)]
fn analytics_destination_uses_explicit_capture_file() {
    let capture_path = unique_capture_path("destination");
    let destination = AnalyticsEventsDestination::from_base_url_and_capture_file(
        "https://chatgpt.com/backend-api/".to_string(),
        Some(capture_path.clone()),
    );

    assert_eq!(
        destination,
        AnalyticsEventsDestination::CaptureFile {
            path: capture_path.clone()
        }
    );
    assert_eq!(
        fs::read_to_string(&capture_path).expect("read capture file"),
        ""
    );
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;

        let mode = fs::metadata(&capture_path)
            .expect("read capture file metadata")
            .permissions()
            .mode();
        assert_eq!(mode & 0o777, 0o600);
    }
    fs::remove_file(capture_path).expect("remove capture file");
}

#[test]
fn analytics_destination_uses_http_without_capture_file() {
    let destination = AnalyticsEventsDestination::from_base_url_and_capture_file(
        "https://chatgpt.com/backend-api/".to_string(),
        /*capture_file*/ None,
    );

    assert_eq!(
        destination,
        AnalyticsEventsDestination::Http {
            url: "https://chatgpt.com/backend-api/codex/analytics-events/events".to_string()
        }
    );
}

#[test]
#[cfg(not(debug_assertions))]
fn analytics_destination_ignores_capture_file_in_release() {
    let destination = AnalyticsEventsDestination::from_base_url_and_capture_file(
        "https://chatgpt.com/backend-api/".to_string(),
        Some(std::path::PathBuf::from("ignored.jsonl")),
    );

    assert_eq!(
        destination,
        AnalyticsEventsDestination::Http {
            url: "https://chatgpt.com/backend-api/codex/analytics-events/events".to_string()
        }
    );
}

#[tokio::test]
#[cfg(debug_assertions)]
async fn capture_file_writes_exact_serialized_request() {
    let capture_path = unique_capture_path("single");
    let destination = AnalyticsEventsDestination::CaptureFile {
        path: capture_path.clone(),
    };
    let event = sample_regular_track_event("thread-1");
    let expected_event = serde_json::to_value(&event).expect("serialize expected event");
    let auth = codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing();

    send_track_events_request(&auth, &destination, vec![event]).await;

    let contents = fs::read_to_string(&capture_path).expect("read capture file");
    let lines = contents.lines().collect::<Vec<_>>();
    assert_eq!(lines.len(), 1);
    let payload: serde_json::Value =
        serde_json::from_str(lines[0]).expect("parse captured payload");
    assert_eq!(payload, serde_json::json!({"events": [expected_event]}));

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

#[tokio::test]
#[cfg(debug_assertions)]
async fn capture_file_writes_final_batches_as_separate_lines() {
    let capture_path = unique_capture_path("batches");
    let destination = AnalyticsEventsDestination::CaptureFile {
        path: capture_path.clone(),
    };
    let auth = codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing();
    let events = vec![
        sample_regular_track_event("thread-1"),
        sample_accepted_line_fingerprint_event("thread-2"),
        sample_regular_track_event("thread-3"),
    ];

    for batch in track_event_request_batches(events) {
        send_track_events_request(&auth, &destination, batch).await;
    }

    let contents = fs::read_to_string(&capture_path).expect("read capture file");
    let payloads = contents
        .lines()
        .map(|line| serde_json::from_str::<serde_json::Value>(line).expect("parse capture line"))
        .collect::<Vec<_>>();
    assert_eq!(payloads.len(), 3);
    assert_eq!(payloads[0]["events"][0]["skill_id"], "skill-thread-1");
    assert_eq!(
        payloads[1]["events"][0]["event_type"],
        "codex_accepted_line_fingerprints"
    );
    assert_eq!(payloads[2]["events"][0]["skill_id"], "skill-thread-3");

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

#[tokio::test]
#[cfg(debug_assertions)]
async fn api_key_auth_sends_only_plugin_events_to_codex_backend() {
    let capture_path = unique_capture_path("api-key-plugin-events");
    let destination = AnalyticsEventsDestination::CaptureFile {
        path: capture_path.clone(),
    };
    let auth_manager = codex_login::AuthManager::from_auth_for_testing(
        codex_login::CodexAuth::from_api_key("sk-test"),
    );
    let plugin_measurement = |thread_id: &str, plugin_id: &str| {
        TrackEventRequest::PluginMeasurement(CodexPluginMeasurementEventRequest {
            event_type: "codex_plugin_measurement_event",
            event_params: CodexPluginMeasurementEventParams {
                model_slug: None,
                reasoning_effort: None,
                thread_id: thread_id.to_string(),
                turn_id: "turn-1".to_string(),
                item_id: "item-1".to_string(),
                originator: "codex_cli_rs".to_string(),
                plugin_id: plugin_id.to_string(),
                execution_id: "execution-1".to_string(),
                operation: "security_scan".to_string(),
                measurement_name: "findings".to_string(),
                number_value: 1.0,
                dimensions: None,
            },
        })
    };

    send_track_events(
        &auth_manager,
        &destination,
        vec![
            sample_regular_track_event("non-plugin-skill"),
            sample_mcp_tool_call_event("non-plugin-mcp", /*plugin_id*/ None),
            sample_plugin_used_track_event("non-plugin-used", /*plugin_id*/ None),
            plugin_measurement("non-plugin-measurement", /*plugin_id*/ ""),
            sample_accepted_line_fingerprint_event("other-event"),
            TrackEventRequest::ThreadArchive(ThreadArchiveEvent {
                event_type: "codex_thread_archive_event",
                event_params: ThreadArchiveEventParams {
                    thread_id: "non-plugin-thread-archive".to_string(),
                    action: ThreadArchiveAction::Archived,
                    occurred_at_ms: 1,
                    app_server_client: None,
                    runtime: None,
                    thread_source: None,
                    parent_thread_id: None,
                },
            }),
            sample_plugin_used_track_event("plugin-used", Some("sample@test")),
            sample_skill_track_event("plugin-skill", Some("sample@test")),
            sample_mcp_tool_call_event("plugin-mcp", Some("sample@test")),
            sample_artifact_operation_event("plugin-artifact"),
            plugin_measurement("plugin-measurement", "sample@test"),
        ],
    )
    .await;

    let contents = fs::read_to_string(&capture_path).expect("read capture file");
    let lines = contents.lines().collect::<Vec<_>>();
    assert_eq!(lines.len(), 1);
    let payload: serde_json::Value =
        serde_json::from_str(lines[0]).expect("parse captured payload");
    let events = payload["events"].as_array().expect("events array");
    for event in events {
        let event_params = event["event_params"].as_object().expect("event params");
        for server_owned_field in [
            "auth_mode",
            "api_organization_id",
            "api_project_id",
            "api_key_tracking_id",
        ] {
            assert!(!event_params.contains_key(server_owned_field));
        }
    }
    let delivered_events = events
        .iter()
        .map(|event| {
            serde_json::json!({
                "event_type": event["event_type"],
                "plugin_id": event["event_params"]["plugin_id"],
                "thread_id": event["event_params"]["thread_id"],
            })
        })
        .collect::<Vec<_>>();
    assert_eq!(
        delivered_events,
        vec![
            serde_json::json!({
                "event_type": "codex_plugin_used",
                "plugin_id": "sample@test",
                "thread_id": "plugin-used",
            }),
            serde_json::json!({
                "event_type": "skill_invocation",
                "plugin_id": "sample@test",
                "thread_id": "plugin-skill",
            }),
            serde_json::json!({
                "event_type": "codex_mcp_tool_call_event",
                "plugin_id": "sample@test",
                "thread_id": "plugin-mcp",
            }),
            serde_json::json!({
                "event_type": "codex_artifact_operation",
                "plugin_id": "presentations@openai-primary-runtime",
                "thread_id": "plugin-artifact",
            }),
            serde_json::json!({
                "event_type": "codex_plugin_measurement_event",
                "plugin_id": "sample@test",
                "thread_id": "plugin-measurement",
            }),
        ]
    );

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

#[test]
#[cfg(debug_assertions)]
fn capture_write_failure_still_consumes_delivery() {
    let capture_path = unique_capture_path("missing-parent").join("events.jsonl");
    let destination = AnalyticsEventsDestination::CaptureFile { path: capture_path };
    let payload = crate::events::TrackEventsRequest {
        events: vec![sample_regular_track_event("thread-1")],
    };

    assert!(capture_track_events_request(&destination, &payload));
}

fn sample_turn_start_request() -> ClientRequest {
    ClientRequest::TurnStart {
        request_id: RequestId::Integer(1),
        params: TurnStartParams {
            thread_id: "thread-1".to_string(),
            client_user_message_id: None,
            input: Vec::new(),
            ..Default::default()
        },
    }
}

fn sample_turn_steer_request() -> ClientRequest {
    ClientRequest::TurnSteer {
        request_id: RequestId::Integer(2),
        params: TurnSteerParams {
            thread_id: "thread-1".to_string(),
            expected_turn_id: "turn-1".to_string(),
            client_user_message_id: None,
            input: Vec::new(),
            responsesapi_client_metadata: None,
            additional_context: None,
        },
    }
}

fn sample_turn_interrupt_request(turn_id: &str) -> ClientRequest {
    ClientRequest::TurnInterrupt {
        request_id: RequestId::Integer(3),
        params: TurnInterruptParams {
            thread_id: "thread-1".to_string(),
            turn_id: turn_id.to_string(),
        },
    }
}

fn sample_turn_interrupt_response() -> ClientResponsePayload {
    ClientResponsePayload::TurnInterrupt(TurnInterruptResponse {})
}

fn sample_thread_archive_request() -> ClientRequest {
    ClientRequest::ThreadArchive {
        request_id: RequestId::Integer(3),
        params: ThreadArchiveParams {
            thread_id: "thread-1".to_string(),
        },
    }
}

fn sample_thread(thread_id: &str) -> 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: None,
        preview: "first prompt".to_string(),
        ephemeral: false,
        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: AppServerSessionSource::Exec,
        can_accept_direct_input: None,
        thread_source: None,
        agent_nickname: None,
        agent_role: None,
        git_info: None,
        name: None,
        turns: Vec::new(),
    }
}

fn sample_thread_start_response() -> ClientResponsePayload {
    ClientResponsePayload::ThreadStart(ThreadStartResponse {
        disabled_plugin_ids: Vec::new(),
        thread: sample_thread("thread-1"),
        model: "gpt-5".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_thread_resume_response() -> ClientResponsePayload {
    ClientResponsePayload::ThreadResume(ThreadResumeResponse {
        disabled_plugin_ids: Vec::new(),
        thread: sample_thread("thread-2"),
        model: "gpt-5".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_thread_fork_response() -> ClientResponsePayload {
    ClientResponsePayload::ThreadFork(ThreadForkResponse {
        disabled_plugin_ids: Vec::new(),
        thread: sample_thread("thread-3"),
        model: "gpt-5".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_turn_start_response() -> ClientResponsePayload {
    ClientResponsePayload::TurnStart(TurnStartResponse {
        turn: Turn {
            id: "turn-1".to_string(),
            items_view: codex_app_server_protocol::TurnItemsView::Full,
            items: Vec::new(),
            status: AppServerTurnStatus::InProgress,
            error: None,
            started_at: None,
            completed_at: None,
            duration_ms: None,
        },
    })
}

fn sample_turn_steer_response() -> ClientResponsePayload {
    ClientResponsePayload::TurnSteer(TurnSteerResponse {
        turn_id: "turn-2".to_string(),
    })
}

#[test]
fn track_plugin_measurements_rejects_unbounded_inputs_before_queueing() {
    let (client, mut receiver) = client_with_receiver();
    let measurements = |row_count| 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: vec![
            PluginMeasurementRow {
                measurement_name: "finding_count".to_string(),
                number_value: 1.0,
                dimensions: BTreeMap::new(),
            };
            row_count
        ],
    };

    client.track_plugin_measurements(measurements(MAX_PLUGIN_MEASUREMENTS_PER_BATCH + 1));
    assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));

    let mut oversized_operation = measurements(1);
    oversized_operation.operation = "o".repeat(65);
    client.track_plugin_measurements(oversized_operation);
    assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));

    let mut mixed_rows = measurements(4);
    mixed_rows.rows[0].measurement_name = "m".repeat(65);
    mixed_rows.rows[1]
        .dimensions
        .insert("d".repeat(65), "valid".to_string());
    mixed_rows.rows[2]
        .dimensions
        .insert("valid".to_string(), "v".repeat(65));
    client.track_plugin_measurements(mixed_rows);
    assert!(matches!(
        receiver.try_recv(),
        Ok(AnalyticsEventsQueueMessage::Fact(fact))
            if matches!(
                fact.as_ref(),
                AnalyticsFact::Custom(CustomAnalyticsFact::PluginMeasurements(input))
                    if input.rows.len() == 1
                        && input.rows[0].measurement_name == "finding_count"
            )
    ));

    client.track_plugin_measurements(measurements(MAX_PLUGIN_MEASUREMENTS_PER_BATCH));
    assert!(matches!(
        receiver.try_recv(),
        Ok(AnalyticsEventsQueueMessage::Fact(fact))
            if matches!(
                fact.as_ref(),
                AnalyticsFact::Custom(CustomAnalyticsFact::PluginMeasurements(input))
                    if input.rows.len() == MAX_PLUGIN_MEASUREMENTS_PER_BATCH
            )
    ));
}

#[test]
fn track_request_only_enqueues_analytics_relevant_requests() {
    let (client, mut receiver) = client_with_receiver();

    for (request_id, request) in [
        (RequestId::Integer(1), sample_turn_start_request()),
        (RequestId::Integer(2), sample_turn_steer_request()),
    ] {
        client.track_request(/*connection_id*/ 7, request_id, &request);
        assert!(matches!(
            receiver.try_recv(),
            Ok(AnalyticsEventsQueueMessage::Fact(input))
                if matches!(*input, AnalyticsFact::ClientRequest { .. })
        ));
    }

    client.track_request(
        /*connection_id*/ 7,
        RequestId::Integer(3),
        &sample_turn_interrupt_request("turn-1"),
    );
    assert!(matches!(
        receiver.try_recv(),
        Ok(AnalyticsEventsQueueMessage::Fact(input))
            if matches!(
                *input,
                AnalyticsFact::ExplicitClientInterruptRequest {
                    ref turn_id,
                    requested_at_ms,
                    ..
                } if turn_id == "turn-1" && requested_at_ms > 0
            )
    ));

    let ignored_request = sample_thread_archive_request();
    client.track_request(
        /*connection_id*/ 7,
        RequestId::Integer(3),
        &ignored_request,
    );
    assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));

    client.track_request(
        /*connection_id*/ 7,
        RequestId::Integer(4),
        &sample_turn_interrupt_request(""),
    );
    assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));
}

#[test]
fn track_response_only_enqueues_analytics_relevant_responses() {
    let (client, mut receiver) = client_with_receiver();

    for (request_id, response) in [
        (RequestId::Integer(1), sample_thread_start_response()),
        (RequestId::Integer(2), sample_thread_resume_response()),
        (RequestId::Integer(3), sample_thread_fork_response()),
        (RequestId::Integer(4), sample_turn_start_response()),
        (RequestId::Integer(5), sample_turn_steer_response()),
        (RequestId::Integer(6), sample_turn_interrupt_response()),
    ] {
        client.track_response(/*connection_id*/ 7, request_id, &response);
        assert!(matches!(
            receiver.try_recv(),
            Ok(AnalyticsEventsQueueMessage::Fact(input))
                if matches!(*input, AnalyticsFact::ClientResponse { .. })
        ));
    }

    client.track_response(
        /*connection_id*/ 7,
        RequestId::Integer(7),
        &ClientResponsePayload::ThreadArchive(ThreadArchiveResponse {}),
    );
    assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));
}

#[cfg(unix)]
#[test]
fn track_response_ignores_unserializable_thread_responses() {
    use std::ffi::OsString;
    use std::os::unix::ffi::OsStringExt;

    let (client, mut receiver) = client_with_receiver();
    let mut response = sample_thread_start_response();
    let ClientResponsePayload::ThreadStart(thread_start) = &mut response else {
        panic!("expected thread/start response");
    };
    thread_start.cwd = codex_utils_absolute_path::AbsolutePathBuf::from_absolute_path(
        std::path::PathBuf::from(OsString::from_vec(vec![b'/', b'b', b'a', b'd', 0xff])),
    )
    .expect("non-UTF-8 Unix paths are valid absolute paths");

    client.track_response(/*connection_id*/ 7, RequestId::Integer(1), &response);

    assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));
}

#[tokio::test]
async fn flush_waits_for_preceding_fact_delivery() {
    let (client, mut receiver) = client_with_receiver();
    client.track_request(
        /*connection_id*/ 7,
        RequestId::Integer(1),
        &sample_turn_start_request(),
    );

    let flush = tokio::spawn(async move { client.flush().await });
    assert!(matches!(
        receiver.recv().await,
        Some(AnalyticsEventsQueueMessage::Fact(input))
            if matches!(*input, AnalyticsFact::ClientRequest { .. })
    ));
    let done_tx = match receiver.recv().await {
        Some(AnalyticsEventsQueueMessage::Flush(done_tx)) => done_tx,
        _ => panic!("expected analytics flush barrier"),
    };
    tokio::time::sleep(std::time::Duration::from_millis(25)).await;
    assert!(!flush.is_finished());
    done_tx.send(()).expect("flush receiver should remain open");
    flush.await.expect("flush task should complete");
}

#[tokio::test]
async fn flush_is_noop_when_analytics_is_disabled() {
    let client = AnalyticsEventsClient::new(
        codex_login::AuthManager::from_auth_for_testing(
            codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing(),
        ),
        "https://chatgpt.com/backend-api".to_string(),
        /*analytics_enabled*/ Some(false),
    );
    client.track_notification(&ServerNotification::ThreadArchived(
        ThreadArchivedNotification {
            thread_id: "thread-1".to_string(),
        },
    ));
    assert!(client.queue.is_none());
    client.flush().await;
}

#[test]
fn app_used_preserves_first_classification_and_emits_again_next_turn() {
    let (client, mut receiver) = client_with_receiver();
    let tracking = TrackEventsContext {
        model_slug: "gpt-5".to_string(),
        thread_id: "thread-1".to_string(),
        turn_id: "turn-1".to_string(),
        product_client_id: "codex_desktop".to_string(),
    };
    for (turn_id, elicitation_type) in [
        ("turn-1", Some(ElicitationType::AuthOrLink)),
        ("turn-1", None),
        ("turn-2", None),
    ] {
        client.track_app_used(
            TrackEventsContext {
                turn_id: turn_id.to_string(),
                ..tracking.clone()
            },
            AppInvocation {
                connector_id: Some("calendar".to_string()),
                app_name: Some("Calendar".to_string()),
                invocation_type: Some(InvocationType::Implicit),
            },
            elicitation_type,
        );
    }
    for (turn_id, elicitation_type) in [
        ("turn-1", Some(ElicitationType::AuthOrLink)),
        ("turn-2", None),
    ] {
        let Ok(AnalyticsEventsQueueMessage::Fact(input)) = receiver.try_recv() else {
            panic!("expected app-used analytics fact");
        };
        let AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(input)) = *input else {
            panic!("expected app-used analytics fact");
        };
        assert_eq!(
            (
                input.tracking.turn_id.as_str(),
                input.app.connector_id.as_deref(),
                input.elicitation_type,
            ),
            (turn_id, Some("calendar"), elicitation_type)
        );
    }
    assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));
}

#[test]
fn track_notification_only_enqueues_analytics_relevant_notifications() {
    let (client, mut receiver) = client_with_receiver();
    let tracked_payload = TurnDiffUpdatedNotification {
        thread_id: "thread-1".to_string(),
        turn_id: "turn-1".to_string(),
        diff: "diff".to_string(),
    };
    let tracked_notification = ServerNotification::TurnDiffUpdated(tracked_payload.clone());

    client.track_notification(&tracked_notification);

    let Ok(AnalyticsEventsQueueMessage::Fact(input)) = receiver.try_recv() else {
        panic!("expected analytics notification");
    };
    let AnalyticsFact::Notification(notification) = *input else {
        panic!("expected analytics notification fact");
    };
    let ServerNotification::TurnDiffUpdated(notification) = *notification else {
        panic!("expected turn diff notification");
    };
    assert_eq!(notification, tracked_payload);

    let ignored_notification =
        ServerNotification::CommandExecutionOutputDelta(CommandExecutionOutputDeltaNotification {
            thread_id: "thread-1".to_string(),
            turn_id: "turn-1".to_string(),
            item_id: "item-1".to_string(),
            delta: "output".to_string(),
        });

    client.track_notification(&ignored_notification);
    assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty)));
}

#[test]
fn track_event_request_batches_only_isolates_accepted_line_fingerprint_events() {
    let batches = track_event_request_batches(vec![
        sample_regular_track_event("thread-1"),
        sample_regular_track_event("thread-2"),
        sample_accepted_line_fingerprint_event("thread-3"),
        sample_accepted_line_fingerprint_event("thread-4"),
        sample_regular_track_event("thread-5"),
        sample_regular_track_event("thread-6"),
    ]);

    assert_eq!(batches.len(), 4);
    assert_eq!(batches[0].len(), 2);
    assert_eq!(batches[1].len(), 1);
    assert_eq!(batches[2].len(), 1);
    assert_eq!(batches[3].len(), 2);
    assert!(batches[1][0].should_send_in_isolated_request());
    assert!(batches[2][0].should_send_in_isolated_request());
}