openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
8f8ace78cbac2901b0030a0101ae00146148c237
codex/codex-rs/analytics/src/client_tests.rs
T
acrognale-oai 8f8ace78cb Enforce application network policy for AWS auth and telemetry (#47408) 
## Why

AWS credential discovery, analytics, and telemetry used clients that did not share the application's network policy. These requests need to honor destination restrictions and permission revocation.

## What changed

- Route AWS credential and region HTTP requests through the shared HTTP client, and check the signing destination before loading credentials. Skip discovery for static access keys with an explicit region.
- Apply account-scoped policy to Bedrock authentication. Require unrestricted policy for credential exporters and reauthentication commands, and cancel active work when permission is revoked. Document that AWS profile `credential_process` network traffic remains outside the application's HTTP policy.
- Send analytics through the authenticated account's HTTP client factory.
- Guard OTLP log, trace, and metric exports with revocable permits, disable them under destination restrictions, and make managed HTTP exports cancellable. Suppress global Statsig settings while managed policy is active.

## Testing

Add coverage for metadata credential request cancellation, allowed and denied SigV4 destinations, AWS credential precedence and endpoint configuration, and blocked credential exporter recovery. Update telemetry tests for account transitions and managed-policy Statsig suppression.

GitOrigin-RevId: d8a2018bfbbe971ec430699b0d3f86a7a9e1e072
2026-09-23 00:47:34 +00:00

1058 lines
38 KiB
Rust
Raw Blame History

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::ThreadRealtimeItemAddedNotification;
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;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
#[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()),
voice_session_id: None,
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 {
turn_metadata: None,
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,
voice_session_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();
let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault);
send_track_events_request(&auth, &destination, vec![event], &factory).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) {
let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault);
send_track_events_request(&auth, &destination, batch, &factory).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 {
turn_metadata: None,
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 realtime_handoff_tracks_only_marker_without_transcript() {
let (client, mut receiver) = client_with_receiver();
client.track_notification(&ServerNotification::ThreadRealtimeItemAdded(
ThreadRealtimeItemAddedNotification {
thread_id: "thread-1".to_string(),
item: serde_json::json!({
"type": "handoff_request",
"input_transcript": "private speech",
}),
},
));
let Ok(AnalyticsEventsQueueMessage::Fact(input)) = receiver.try_recv() else {
panic!("expected realtime handoff marker");
};
assert!(matches!(
*input,
AnalyticsFact::RealtimeHandoffRequested { thread_id } if thread_id == "thread-1"
));
client.track_notification(&ServerNotification::ThreadRealtimeItemAdded(
ThreadRealtimeItemAddedNotification {
thread_id: "thread-1".to_string(),
item: serde_json::json!({"type": "input_transcript"}),
},
));
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());
}
Reference in New Issue View Git Blame Copy Permalink