agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

[codex-analytics] emit terminal review events (#18748)

Browse Source 
## Why

Review telemetry should describe reviews as first-class events, not only
as counters denormalized onto terminal tool-item events. That lets us
analyze guardian and user reviews consistently across command execution,
file changes, permissions, and network access, while still preserving
the terminal item summaries that existing tool analytics need.

To make those review events accurate, analytics also needs the observed
completion time for each review and enough command metadata to
distinguish `shell` from `unified_exec` reviews.

## What changed

- emit generic `codex_review_event` rows for completed user and guardian
reviews, with review subjects, reviewer, trigger, terminal status,
resolution, and observed duration
- reduce approval request / response / abort facts into review events
for command execution, file change, and permissions flows
- keep denormalized review counts, final approval outcome, and
permission-request flags on terminal tool-item events for
item-associated reviews
- plumb review completion timing so user-review responses and aborts use
app-server-observed completion times, while guardian analytics reuse the
same terminal timestamps emitted on guardian assessment events
- carry command approval `source` through the protocol and app-server
layers so review analytics can distinguish `shell` from `unified_exec`
- add analytics coverage for user-review emission, guardian-review
emission, permission reviews that should not denormalize onto tool
items, item-summary isolation across threads, and the serialized
review-event shape

## Verification

- `cargo test -p codex-analytics`

---
[//]: # (BEGIN SAPLING FOOTER)
Stack created with [Sapling](https://sapling-scm.com). Best reviewed
with [ReviewStack](https://reviewstack.dev/openai/codex/pull/18748).
* __->__ #18748
* #21434
* #18747
* #17090
* #17089
* #20514
This commit is contained in:
rhan-oai
2026-05-11 15:13:32 -07:00
committed by GitHub
Unverified
parent aa9e8f0262
commit a175ddacc0
8 changed files with 1382 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/analytics/src/analytics_client_tests.rs
+588 -6
View File
@@ -11,18 +11,25 @@ use crate::events::CodexCompactionEventRequest;
use crate::events::CodexHookRunEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexReviewEventParams;
use crate::events::CodexReviewEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::CodexToolItemEventBase;
use crate::events::CodexTurnEventRequest;
use crate::events::FinalApprovalOutcome;
use crate::events::GuardianApprovalRequestSource;
use crate::events::GuardianReviewDecision;
use crate::events::GuardianReviewEventParams;
use crate::events::GuardianReviewFailureReason;
use crate::events::GuardianReviewTerminalStatus;
use crate::events::GuardianReviewedAction;
use crate::events::ReviewResolution;
use crate::events::ReviewStatus;
use crate::events::ReviewSubjectKind;
use crate::events::ReviewTrigger;
use crate::events::Reviewer;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::ToolItemFinalApprovalOutcome;
use crate::events::ToolItemTerminalStatus;
use crate::events::TrackEventRequest;
use crate::events::codex_app_metadata;
@@ -72,20 +79,32 @@ use codex_app_server_protocol::CodexErrorInfo;
use codex_app_server_protocol::CollabAgentTool;
use codex_app_server_protocol::CollabAgentToolCallStatus;
use codex_app_server_protocol::CommandAction;
use codex_app_server_protocol::CommandExecutionApprovalDecision;
use codex_app_server_protocol::CommandExecutionRequestApprovalParams;
use codex_app_server_protocol::CommandExecutionRequestApprovalResponse;
use codex_app_server_protocol::CommandExecutionSource;
use codex_app_server_protocol::CommandExecutionStatus;
use codex_app_server_protocol::DynamicToolCallStatus;
use codex_app_server_protocol::GuardianApprovalReview;
use codex_app_server_protocol::GuardianApprovalReviewAction;
use codex_app_server_protocol::GuardianApprovalReviewStatus;
use codex_app_server_protocol::GuardianCommandSource as AppServerGuardianCommandSource;
use codex_app_server_protocol::InitializeCapabilities;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::ItemCompletedNotification;
use codex_app_server_protocol::ItemGuardianApprovalReviewCompletedNotification;
use codex_app_server_protocol::ItemStartedNotification;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::McpToolCallStatus;
use codex_app_server_protocol::NonSteerableTurnKind;
use codex_app_server_protocol::PatchApplyStatus;
use codex_app_server_protocol::PermissionsRequestApprovalParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::RequestPermissionProfile;
use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ServerResponse;
use codex_app_server_protocol::SessionSource as AppServerSessionSource;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadArchiveParams;
@@ -114,6 +133,7 @@ use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::approvals::NetworkApprovalProtocol;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::ModeKind;
use codex_protocol::models::NetworkPermissions as CoreNetworkPermissions;
use codex_protocol::models::PermissionProfile as CorePermissionProfile;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::HookEventName;
@@ -124,6 +144,9 @@ 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;
@@ -612,7 +635,7 @@ async fn ingest_turn_prerequisites(
}
}
async fn ingest_tool_review_prerequisites(
async fn ingest_review_prerequisites(
reducer: &mut AnalyticsReducer,
events: &mut Vec<TrackEventRequest>,
) {
@@ -634,6 +657,58 @@ async fn ingest_tool_review_prerequisites(
events.clear();
}
async fn ingest_completed_command_execution_item(
reducer: &mut AnalyticsReducer,
events: &mut Vec<TrackEventRequest>,
thread_id: &str,
item_id: &str,
) {
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_started_notification(
thread_id, "turn-1",
))),
events,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::ItemStarted(
ItemStartedNotification {
thread_id: thread_id.to_string(),
turn_id: "turn-1".to_string(),
started_at_ms: 1_000,
item: sample_command_execution_item_with_id(
item_id,
CommandExecutionStatus::InProgress,
/*exit_code*/ None,
/*duration_ms*/ None,
),
},
))),
events,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(ServerNotification::ItemCompleted(
ItemCompletedNotification {
thread_id: thread_id.to_string(),
turn_id: "turn-1".to_string(),
completed_at_ms: 1_042,
item: sample_command_execution_item_with_id(
item_id,
CommandExecutionStatus::Completed,
Some(0),
Some(42),
),
},
))),
events,
)
.await;
}
fn sample_initialize_fact(connection_id: u64) -> AnalyticsFact {
AnalyticsFact::Initialize {
connection_id,
@@ -664,9 +739,18 @@ fn sample_command_execution_item(
status: CommandExecutionStatus,
exit_code: Option<i32>,
duration_ms: Option<i64>,
) -> ThreadItem {
sample_command_execution_item_with_id("item-1", status, exit_code, duration_ms)
}
fn sample_command_execution_item_with_id(
id: &str,
status: CommandExecutionStatus,
exit_code: Option<i32>,
duration_ms: Option<i64>,
) -> ThreadItem {
ThreadItem::CommandExecution {
id: "item-1".to_string(),
id: id.to_string(),
command: "echo hi".to_string(),
cwd: test_path_buf("/tmp").abs(),
process_id: Some("pid-1".to_string()),
@@ -697,6 +781,98 @@ fn sample_command_execution_item_with_actions(
item
}
fn sample_command_approval_request(request_id: i64, approval_id: Option<&str>) -> ServerRequest {
ServerRequest::CommandExecutionRequestApproval {
request_id: RequestId::Integer(request_id),
params: CommandExecutionRequestApprovalParams {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
item_id: "item-1".to_string(),
started_at_ms: 1_000,
approval_id: approval_id.map(str::to_string),
reason: None,
network_approval_context: None,
command: Some("echo hi".to_string()),
cwd: None,
command_actions: None,
additional_permissions: None,
proposed_execpolicy_amendment: None,
proposed_network_policy_amendments: None,
available_decisions: None,
},
}
}
fn sample_command_approval_response(
request_id: i64,
decision: CommandExecutionApprovalDecision,
) -> ServerResponse {
ServerResponse::CommandExecutionRequestApproval {
request_id: RequestId::Integer(request_id),
response: CommandExecutionRequestApprovalResponse { decision },
}
}
fn sample_permissions_approval_request(request_id: i64) -> ServerRequest {
ServerRequest::PermissionsRequestApproval {
request_id: RequestId::Integer(request_id),
params: PermissionsRequestApprovalParams {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
item_id: "permissions-1".to_string(),
started_at_ms: 1_000,
cwd: test_path_buf("/tmp").abs(),
reason: Some("need network".to_string()),
permissions: RequestPermissionProfile {
network: Some(codex_app_server_protocol::AdditionalNetworkPermissions {
enabled: Some(true),
}),
file_system: None,
},
},
}
}
fn sample_effective_permissions_approval_response(
permissions: CoreRequestPermissionProfile,
scope: CorePermissionGrantScope,
) -> CoreRequestPermissionsResponse {
CoreRequestPermissionsResponse {
permissions,
scope,
strict_auto_review: false,
}
}
fn sample_guardian_review_completed(
review_id: &str,
target_item_id: Option<&str>,
status: GuardianApprovalReviewStatus,
) -> ServerNotification {
ServerNotification::ItemGuardianApprovalReviewCompleted(
ItemGuardianApprovalReviewCompletedNotification {
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
started_at_ms: 1_000,
completed_at_ms: 1_042,
review_id: review_id.to_string(),
target_item_id: target_item_id.map(str::to_string),
decision_source: codex_app_server_protocol::AutoReviewDecisionSource::Agent,
review: GuardianApprovalReview {
status,
risk_level: None,
user_authorization: None,
rationale: None,
},
action: GuardianApprovalReviewAction::Command {
source: AppServerGuardianCommandSource::Shell,
command: "echo hi".to_string(),
cwd: test_path_buf("/tmp").abs(),
},
},
)
}
fn expected_absolute_path(path: &PathBuf) -> String {
std::fs::canonicalize(path)
.unwrap_or_else(|_| path.to_path_buf())
@@ -1233,7 +1409,7 @@ fn command_execution_event_serializes_expected_shape() {
review_count: 0,
guardian_review_count: 0,
user_review_count: 0,
final_approval_outcome: ToolItemFinalApprovalOutcome::NotNeeded,
final_approval_outcome: FinalApprovalOutcome::NotNeeded,
terminal_status: ToolItemTerminalStatus::Completed,
failure_kind: None,
requested_additional_permissions: false,
@@ -1299,6 +1475,82 @@ fn command_execution_event_serializes_expected_shape() {
);
}
#[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();
@@ -1713,7 +1965,7 @@ async fn item_lifecycle_notifications_publish_command_execution_event() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_tool_review_prerequisites(&mut reducer, &mut events).await;
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_started_notification(
@@ -1824,6 +2076,336 @@ async fn item_lifecycle_notifications_publish_command_execution_event() {
assert_eq!(payload[0]["event_params"]["thread_source"], "user");
}
#[tokio::test]
async fn command_execution_approval_response_publishes_user_review_event() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_command_approval_request(
/*request_id*/ 41, /*approval_id*/ None,
)),
},
&mut events,
)
.await;
assert!(events.is_empty());
reducer
.ingest(
AnalyticsFact::ServerResponse {
completed_at_ms: 1_042,
response: Box::new(sample_command_approval_response(
/*request_id*/ 41,
CommandExecutionApprovalDecision::Accept,
)),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_review_event");
assert_eq!(payload[0]["event_params"]["thread_id"], "thread-1");
assert_eq!(payload[0]["event_params"]["turn_id"], "turn-1");
assert_eq!(payload[0]["event_params"]["item_id"], "item-1");
assert_eq!(payload[0]["event_params"]["review_id"], "user:41");
assert_eq!(payload[0]["event_params"]["thread_source"], "user");
assert_eq!(
payload[0]["event_params"]["subject_kind"],
"command_execution"
);
assert_eq!(
payload[0]["event_params"]["subject_name"],
"command_execution"
);
assert_eq!(payload[0]["event_params"]["reviewer"], "user");
assert_eq!(payload[0]["event_params"]["trigger"], "initial");
assert_eq!(payload[0]["event_params"]["status"], "approved");
assert_eq!(payload[0]["event_params"]["started_at_ms"], 1_000);
assert_eq!(payload[0]["event_params"]["completed_at_ms"], 1_042);
assert_eq!(payload[0]["event_params"]["duration_ms"], 42);
}
#[tokio::test]
async fn permissions_reviews_emit_events_without_denormalizing_onto_tool_items() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_permissions_approval_request(/*request_id*/ 51)),
},
&mut events,
)
.await;
assert!(events.is_empty());
reducer
.ingest(
AnalyticsFact::EffectivePermissionsApprovalResponse {
completed_at_ms: 1_042,
request_id: RequestId::Integer(51),
response: Box::new(sample_effective_permissions_approval_response(
CoreRequestPermissionProfile::default(),
CorePermissionGrantScope::Turn,
)),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_review_event");
assert_eq!(payload[0]["event_params"]["review_id"], "user:51");
assert_eq!(payload[0]["event_params"]["subject_kind"], "permissions");
assert_eq!(payload[0]["event_params"]["reviewer"], "user");
assert_eq!(payload[0]["event_params"]["status"], "denied");
assert_eq!(payload[0]["event_params"]["resolution"], "none");
events.clear();
ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-1", "permissions-1")
.await;
let payload = serde_json::to_value(&events[0]).expect("serialize tool item event");
assert_eq!(payload["event_params"]["item_id"], "permissions-1");
assert_eq!(payload["event_params"]["review_count"], 0);
assert_eq!(payload["event_params"]["user_review_count"], 0);
assert_eq!(payload["event_params"]["guardian_review_count"], 0);
}
#[tokio::test]
async fn effective_session_permissions_response_publishes_session_user_review_event() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_permissions_approval_request(/*request_id*/ 52)),
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::EffectivePermissionsApprovalResponse {
completed_at_ms: 1_042,
request_id: RequestId::Integer(52),
response: Box::new(sample_effective_permissions_approval_response(
CoreRequestPermissionProfile {
network: Some(CoreNetworkPermissions {
enabled: Some(true),
}),
file_system: None,
},
CorePermissionGrantScope::Session,
)),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_review_event");
assert_eq!(payload[0]["event_params"]["review_id"], "user:52");
assert_eq!(payload[0]["event_params"]["subject_kind"], "permissions");
assert_eq!(payload[0]["event_params"]["reviewer"], "user");
assert_eq!(payload[0]["event_params"]["status"], "approved");
assert_eq!(payload[0]["event_params"]["resolution"], "session_approval");
}
#[tokio::test]
async fn aborted_server_request_publishes_aborted_user_review_event_once() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_command_approval_request(
/*request_id*/ 61, /*approval_id*/ None,
)),
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ServerRequestAborted {
completed_at_ms: 1_042,
request_id: RequestId::Integer(61),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_params"]["review_id"], "user:61");
assert_eq!(payload[0]["event_params"]["status"], "aborted");
assert_eq!(payload[0]["event_params"]["resolution"], "none");
events.clear();
reducer
.ingest(
AnalyticsFact::ServerResponse {
completed_at_ms: 1_043,
response: Box::new(sample_command_approval_response(
/*request_id*/ 61,
CommandExecutionApprovalDecision::Accept,
)),
},
&mut events,
)
.await;
assert!(events.is_empty());
}
#[tokio::test]
async fn guardian_completed_notification_publishes_review_event_with_thread_metadata() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_guardian_review_completed(
"guardian-review-1",
Some("item-1"),
GuardianApprovalReviewStatus::Denied,
))),
&mut events,
)
.await;
let payload = serde_json::to_value(&events[0]).expect("serialize review event");
assert_eq!(payload["event_type"], "codex_review_event");
assert_eq!(payload["event_params"]["review_id"], "guardian-review-1");
assert_eq!(payload["event_params"]["item_id"], "item-1");
assert_eq!(payload["event_params"]["thread_source"], "user");
assert_eq!(payload["event_params"]["subject_kind"], "command_execution");
assert_eq!(payload["event_params"]["reviewer"], "guardian");
assert_eq!(payload["event_params"]["status"], "denied");
assert_eq!(payload["event_params"]["started_at_ms"], 1_000);
assert_eq!(payload["event_params"]["completed_at_ms"], 1_042);
assert_eq!(payload["event_params"]["duration_ms"], 42);
}
#[tokio::test]
async fn terminal_reviews_denormalize_counts_onto_tool_item_events() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_command_approval_request(
/*request_id*/ 71, /*approval_id*/ None,
)),
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ServerResponse {
completed_at_ms: 1_042,
response: Box::new(sample_command_approval_response(
/*request_id*/ 71,
CommandExecutionApprovalDecision::AcceptForSession,
)),
},
&mut events,
)
.await;
events.clear();
ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-1", "item-1").await;
let payload = serde_json::to_value(&events[0]).expect("serialize tool item event");
assert_eq!(payload["event_params"]["review_count"], 1);
assert_eq!(payload["event_params"]["user_review_count"], 1);
assert_eq!(payload["event_params"]["guardian_review_count"], 0);
assert_eq!(
payload["event_params"]["final_approval_outcome"],
"user_approved_for_session"
);
}
#[tokio::test]
async fn item_review_summaries_do_not_cross_threads_with_reused_item_ids() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(2),
response: Box::new(sample_thread_start_response(
"thread-2", /*ephemeral*/ false, "gpt-5",
)),
},
&mut events,
)
.await;
events.clear();
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(sample_command_approval_request(
/*request_id*/ 72, /*approval_id*/ None,
)),
},
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ServerResponse {
completed_at_ms: 1_042,
response: Box::new(sample_command_approval_response(
/*request_id*/ 72,
CommandExecutionApprovalDecision::Accept,
)),
},
&mut events,
)
.await;
events.clear();
ingest_completed_command_execution_item(&mut reducer, &mut events, "thread-2", "item-1").await;
let payload = serde_json::to_value(&events[0]).expect("serialize tool item event");
assert_eq!(payload["event_params"]["thread_id"], "thread-2");
assert_eq!(payload["event_params"]["item_id"], "item-1");
assert_eq!(payload["event_params"]["review_count"], 0);
assert_eq!(payload["event_params"]["user_review_count"], 0);
assert_eq!(payload["event_params"]["guardian_review_count"], 0);
assert_eq!(payload["event_params"]["final_approval_outcome"], "unknown");
}
#[test]
fn subagent_thread_started_review_serializes_expected_shape() {
let event = TrackEventRequest::ThreadInitialized(subagent_thread_started_event_request(
@@ -2112,7 +2694,7 @@ async fn subagent_tool_items_inherit_parent_connection_metadata() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_tool_review_prerequisites(&mut reducer, &mut events).await;
ingest_review_prerequisites(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::SubAgentThreadStarted(
codex-rs/analytics/src/client.rs
+23 -1
View File
@@ -33,6 +33,7 @@ use codex_login::AuthManager;
use codex_login::CodexAuth;
use codex_login::default_client::create_client;
use codex_plugin::PluginTelemetryMetadata;
use codex_protocol::request_permissions::RequestPermissionsResponse;
use std::collections::HashSet;
use std::sync::Arc;
use std::sync::Mutex;
@@ -172,9 +173,10 @@ impl AnalyticsEventsClient {
&self,
tracking: &GuardianReviewTrackContext,
result: GuardianReviewAnalyticsResult,
completed_at_ms: u64,
) {
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::GuardianReview(
Box::new(tracking.event_params(result)),
Box::new(tracking.event_params(result, completed_at_ms)),
)));
}
@@ -348,6 +350,26 @@ impl AnalyticsEventsClient {
});
}
pub fn track_effective_permissions_approval_response(
&self,
completed_at_ms: u64,
request_id: RequestId,
response: RequestPermissionsResponse,
) {
self.record_fact(AnalyticsFact::EffectivePermissionsApprovalResponse {
completed_at_ms,
request_id,
response: Box::new(response),
});
}
pub fn track_server_request_aborted(&self, completed_at_ms: u64, request_id: RequestId) {
self.record_fact(AnalyticsFact::ServerRequestAborted {
completed_at_ms,
request_id,
});
}
pub fn track_notification(&self, notification: ServerNotification) {
if !matches!(
notification,
codex-rs/analytics/src/events.rs
+6 -6
View File
@@ -20,7 +20,6 @@ use crate::facts::TurnSteerRejectionReason;
use crate::facts::TurnSteerResult;
use crate::facts::TurnSubmissionType;
use crate::now_unix_millis;
use crate::now_unix_seconds;
use codex_app_server_protocol::CodexErrorInfo;
use codex_app_server_protocol::CommandExecutionSource;
use codex_login::default_client::originator;
@@ -320,6 +319,7 @@ impl GuardianReviewTrackContext {
pub(crate) fn event_params(
&self,
result: GuardianReviewAnalyticsResult,
completed_at_ms: u64,
) -> GuardianReviewEventParams {
GuardianReviewEventParams {
thread_id: self.thread_id.clone(),
@@ -346,7 +346,7 @@ impl GuardianReviewTrackContext {
time_to_first_token_ms: result.time_to_first_token_ms,
completion_latency_ms: Some(self.started_instant.elapsed().as_millis() as u64),
started_at: self.started_at_ms / 1_000,
completed_at: Some(now_unix_seconds()),
completed_at: Some(completed_at_ms / 1_000),
input_tokens: result.token_usage.as_ref().map(|usage| usage.input_tokens),
cached_input_tokens: result
.token_usage
@@ -429,7 +429,7 @@ pub(crate) struct GuardianReviewEventPayload {
#[allow(dead_code)]
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub(crate) enum ToolItemFinalApprovalOutcome {
pub(crate) enum FinalApprovalOutcome {
Unknown,
NotNeeded,
ConfigAllowed,
@@ -486,7 +486,7 @@ pub(crate) struct CodexToolItemEventBase {
pub(crate) review_count: u64,
pub(crate) guardian_review_count: u64,
pub(crate) user_review_count: u64,
pub(crate) final_approval_outcome: ToolItemFinalApprovalOutcome,
pub(crate) final_approval_outcome: FinalApprovalOutcome,
pub(crate) terminal_status: ToolItemTerminalStatus,
pub(crate) failure_kind: Option<ToolItemFailureKind>,
pub(crate) requested_additional_permissions: bool,
@@ -553,8 +553,8 @@ pub(crate) struct CodexReviewEventParams {
pub(crate) thread_source: Option<ThreadSource>,
pub(crate) subagent_source: Option<String>,
pub(crate) parent_thread_id: Option<String>,
pub(crate) tool_kind: ReviewSubjectKind,
pub(crate) tool_name: String,
pub(crate) subject_kind: ReviewSubjectKind,
pub(crate) subject_name: String,
pub(crate) reviewer: Reviewer,
pub(crate) trigger: ReviewTrigger,
pub(crate) status: ReviewStatus,
codex-rs/analytics/src/facts.rs
+10
View File
@@ -25,6 +25,7 @@ use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SkillScope;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::request_permissions::RequestPermissionsResponse;
use serde::Serialize;
use std::path::PathBuf;
@@ -305,6 +306,15 @@ pub(crate) enum AnalyticsFact {
completed_at_ms: u64,
response: Box<ServerResponse>,
},
EffectivePermissionsApprovalResponse {
completed_at_ms: u64,
request_id: RequestId,
response: Box<RequestPermissionsResponse>,
},
ServerRequestAborted {
completed_at_ms: u64,
request_id: RequestId,
},
Notification(Box<ServerNotification>),
// Facts that do not naturally exist on the app-server protocol surface, or
// would require non-trivial protocol reshaping on this branch.
codex-rs/analytics/src/reducer.rs
+691 -34
View File
@@ -22,6 +22,8 @@ use crate::events::CodexMcpToolCallEventParams;
use crate::events::CodexMcpToolCallEventRequest;
use crate::events::CodexPluginEventRequest;
use crate::events::CodexPluginUsedEventRequest;
use crate::events::CodexReviewEventParams;
use crate::events::CodexReviewEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::CodexToolItemEventBase;
use crate::events::CodexTurnEventParams;
@@ -30,15 +32,20 @@ use crate::events::CodexTurnSteerEventParams;
use crate::events::CodexTurnSteerEventRequest;
use crate::events::CodexWebSearchEventParams;
use crate::events::CodexWebSearchEventRequest;
use crate::events::FinalApprovalOutcome;
use crate::events::GuardianReviewEventParams;
use crate::events::GuardianReviewEventPayload;
use crate::events::GuardianReviewEventRequest;
use crate::events::ReviewResolution;
use crate::events::ReviewStatus;
use crate::events::ReviewSubjectKind;
use crate::events::ReviewTrigger;
use crate::events::Reviewer;
use crate::events::SkillInvocationEventParams;
use crate::events::SkillInvocationEventRequest;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::ToolItemFailureKind;
use crate::events::ToolItemFinalApprovalOutcome;
use crate::events::ToolItemTerminalStatus;
use crate::events::TrackEventRequest;
use crate::events::WebSearchActionKind;
@@ -80,16 +87,24 @@ use codex_app_server_protocol::CollabAgentStatus;
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::CommandExecutionSource;
use codex_app_server_protocol::CommandExecutionStatus;
use codex_app_server_protocol::DynamicToolCallOutputContentItem;
use codex_app_server_protocol::DynamicToolCallStatus;
use codex_app_server_protocol::FileChangeApprovalDecision;
use codex_app_server_protocol::GuardianApprovalReviewAction;
use codex_app_server_protocol::GuardianApprovalReviewStatus;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::McpToolCallStatus;
use codex_app_server_protocol::NetworkPolicyRuleAction;
use codex_app_server_protocol::PatchApplyStatus;
use codex_app_server_protocol::PatchChangeKind;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::RequestPermissionProfile;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ServerResponse;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::TurnSteerResponse;
use codex_app_server_protocol::UserInput;
@@ -105,6 +120,8 @@ use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SkillScope;
use codex_protocol::protocol::ThreadSource;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::request_permissions::PermissionGrantScope as CorePermissionGrantScope;
use codex_protocol::request_permissions::RequestPermissionsResponse as CoreRequestPermissionsResponse;
use sha1::Digest;
use std::collections::HashMap;
use std::path::Path;
@@ -117,6 +134,8 @@ pub(crate) struct AnalyticsReducer {
connections: HashMap<u64, ConnectionState>,
threads: HashMap<String, ThreadAnalyticsState>,
tool_items_started_at_ms: HashMap<ToolItemKey, u64>,
pending_reviews: HashMap<RequestId, PendingReviewState>,
item_review_summaries: HashMap<ToolItemKey, ItemReviewSummary>,
}
struct ConnectionState {
@@ -150,6 +169,16 @@ impl<'a> AnalyticsDropSite<'a> {
}
}
fn review(input: &'a PendingReviewState) -> Self {
Self {
event_name: "review",
thread_id: &input.thread_id,
turn_id: Some(&input.turn_id),
review_id: Some(&input.review_id),
item_id: input.item_id.as_deref(),
}
}
fn compaction(input: &'a CodexCompactionEvent) -> Self {
Self {
event_name: "compaction",
@@ -200,6 +229,30 @@ enum MissingAnalyticsContext {
ThreadMetadata,
}
#[derive(Clone)]
struct PendingReviewState {
thread_id: String,
turn_id: String,
item_id: Option<String>,
review_id: String,
subject_kind: ReviewSubjectKind,
subject_name: String,
trigger: ReviewTrigger,
started_at_ms: u64,
requested_additional_permissions: bool,
requested_network_access: bool,
}
#[derive(Clone, Default)]
struct ItemReviewSummary {
review_count: u64,
guardian_review_count: u64,
user_review_count: u64,
final_approval_outcome: Option<FinalApprovalOutcome>,
requested_additional_permissions: bool,
requested_network_access: bool,
}
#[derive(Clone)]
struct ThreadMetadataState {
thread_source: Option<ThreadSource>,
@@ -363,13 +416,35 @@ impl AnalyticsReducer {
self.ingest_notification(*notification, out).await;
}
AnalyticsFact::ServerRequest {
connection_id: _connection_id,
request: _request,
} => {}
connection_id,
request,
} => {
self.ingest_server_request(connection_id, *request);
}
AnalyticsFact::ServerResponse {
response: _response,
..
} => {}
completed_at_ms,
response,
} => {
self.ingest_server_response(completed_at_ms, *response, out);
}
AnalyticsFact::EffectivePermissionsApprovalResponse {
completed_at_ms,
request_id,
response,
} => {
self.ingest_effective_permissions_approval_response(
completed_at_ms,
request_id,
*response,
out,
);
}
AnalyticsFact::ServerRequestAborted {
completed_at_ms,
request_id,
} => {
self.ingest_server_request_aborted(completed_at_ms, request_id, out);
}
AnalyticsFact::Custom(input) => match input {
CustomAnalyticsFact::SubAgentThreadStarted(input) => {
self.ingest_subagent_thread_started(input, out);
@@ -740,6 +815,207 @@ impl AnalyticsReducer {
}
}
fn ingest_server_request(&mut self, _connection_id: u64, request: ServerRequest) {
match request {
ServerRequest::CommandExecutionRequestApproval { request_id, params } => {
let is_network_access_review = params.network_approval_context.is_some();
let requested_network_access = is_network_access_review
|| params
.proposed_network_policy_amendments
.as_ref()
.is_some_and(|amendments| !amendments.is_empty())
|| params
.additional_permissions
.as_ref()
.and_then(|permissions| permissions.network.as_ref())
.and_then(|network| network.enabled)
.unwrap_or(false);
let requested_additional_permissions = params.additional_permissions.is_some();
let trigger = if params.approval_id.is_some() {
ReviewTrigger::ExecveIntercept
} else if requested_network_access {
ReviewTrigger::NetworkPolicyDenial
} else if requested_additional_permissions {
ReviewTrigger::SandboxDenial
} else {
ReviewTrigger::Initial
};
let Some(started_at_ms) = option_i64_to_u64(Some(params.started_at_ms)) else {
return;
};
self.pending_reviews.insert(
request_id.clone(),
PendingReviewState {
thread_id: params.thread_id,
turn_id: params.turn_id,
item_id: Some(params.item_id),
review_id: user_review_id(&request_id),
subject_kind: if is_network_access_review {
ReviewSubjectKind::NetworkAccess
} else {
ReviewSubjectKind::CommandExecution
},
subject_name: if is_network_access_review {
"network_access".to_string()
} else {
"command_execution".to_string()
},
trigger,
started_at_ms,
requested_additional_permissions,
requested_network_access,
},
);
}
ServerRequest::FileChangeRequestApproval { request_id, params } => {
let requested_additional_permissions = params.grant_root.is_some();
let Some(started_at_ms) = option_i64_to_u64(Some(params.started_at_ms)) else {
return;
};
self.pending_reviews.insert(
request_id.clone(),
PendingReviewState {
thread_id: params.thread_id,
turn_id: params.turn_id,
item_id: Some(params.item_id),
review_id: user_review_id(&request_id),
subject_kind: ReviewSubjectKind::FileChange,
subject_name: "apply_patch".to_string(),
trigger: if requested_additional_permissions {
ReviewTrigger::SandboxDenial
} else {
ReviewTrigger::Initial
},
started_at_ms,
requested_additional_permissions,
requested_network_access: false,
},
);
}
ServerRequest::PermissionsRequestApproval { request_id, params } => {
let requested_network_access = params
.permissions
.network
.as_ref()
.and_then(|network| network.enabled)
.unwrap_or(false);
let requested_additional_permissions =
requested_network_access || params.permissions.file_system.is_some();
let trigger = if requested_network_access {
ReviewTrigger::NetworkPolicyDenial
} else if requested_additional_permissions {
ReviewTrigger::SandboxDenial
} else {
ReviewTrigger::Initial
};
let Some(started_at_ms) = option_i64_to_u64(Some(params.started_at_ms)) else {
return;
};
self.pending_reviews.insert(
request_id.clone(),
PendingReviewState {
thread_id: params.thread_id,
turn_id: params.turn_id,
item_id: Some(params.item_id),
review_id: user_review_id(&request_id),
subject_kind: ReviewSubjectKind::Permissions,
subject_name: "permissions".to_string(),
trigger,
started_at_ms,
requested_additional_permissions,
requested_network_access,
},
);
}
_ => {}
}
}
fn ingest_server_response(
&mut self,
completed_at_ms: u64,
response: ServerResponse,
out: &mut Vec<TrackEventRequest>,
) {
match response {
ServerResponse::CommandExecutionRequestApproval {
request_id,
response,
} => {
let Some(pending_review) = self.pending_reviews.remove(&request_id) else {
return;
};
let (status, resolution) = command_execution_review_result(response.decision);
self.emit_review_event(
pending_review,
Reviewer::User,
status,
resolution,
completed_at_ms,
out,
);
}
ServerResponse::FileChangeRequestApproval {
request_id,
response,
} => {
let Some(pending_review) = self.pending_reviews.remove(&request_id) else {
return;
};
let (status, resolution) = file_change_review_result(response.decision);
self.emit_review_event(
pending_review,
Reviewer::User,
status,
resolution,
completed_at_ms,
out,
);
}
_ => {}
}
}
fn ingest_effective_permissions_approval_response(
&mut self,
completed_at_ms: u64,
request_id: RequestId,
response: CoreRequestPermissionsResponse,
out: &mut Vec<TrackEventRequest>,
) {
let Some(pending_review) = self.pending_reviews.remove(&request_id) else {
return;
};
let (status, resolution) = effective_permissions_review_result(&response);
self.emit_review_event(
pending_review,
Reviewer::User,
status,
resolution,
completed_at_ms,
out,
);
}
fn ingest_server_request_aborted(
&mut self,
completed_at_ms: u64,
request_id: RequestId,
out: &mut Vec<TrackEventRequest>,
) {
let Some(pending_review) = self.pending_reviews.remove(&request_id) else {
return;
};
self.emit_review_event(
pending_review,
Reviewer::User,
ReviewStatus::Aborted,
ReviewResolution::None,
completed_at_ms,
out,
);
}
fn ingest_error_response(
&mut self,
connection_id: u64,
@@ -850,17 +1126,25 @@ impl AnalyticsReducer {
else {
return;
};
if let Some(event) = tool_item_event(
&notification.thread_id,
&notification.turn_id,
&notification.item,
if let Some(event) = tool_item_event(ToolItemEventInput {
thread_id: &notification.thread_id,
turn_id: &notification.turn_id,
item: &notification.item,
started_at_ms,
completed_at_ms,
connection_state,
thread_metadata,
) {
review_summary: self.item_review_summaries.get(&key),
}) {
out.push(event);
}
self.item_review_summaries.remove(&key);
}
ServerNotification::ItemGuardianApprovalReviewStarted(notification) => {
let _ = notification;
}
ServerNotification::ItemGuardianApprovalReviewCompleted(notification) => {
self.ingest_guardian_review_completed(notification, out);
}
ServerNotification::TurnStarted(notification) => {
let turn_state = self.turns.entry(notification.turn.id).or_insert(TurnState {
@@ -1003,6 +1287,48 @@ impl AnalyticsReducer {
)));
}
fn ingest_guardian_review_completed(
&mut self,
notification: codex_app_server_protocol::ItemGuardianApprovalReviewCompletedNotification,
out: &mut Vec<TrackEventRequest>,
) {
let Some((status, resolution)) = guardian_review_result(notification.review.status) else {
return;
};
let (subject_kind, subject_name, trigger) =
guardian_review_subject_metadata(&notification.action);
let Some(started_at_ms) = option_i64_to_u64(Some(notification.started_at_ms)) else {
return;
};
let pending_review = PendingReviewState {
thread_id: notification.thread_id,
turn_id: notification.turn_id,
item_id: notification.target_item_id,
review_id: notification.review_id,
subject_kind,
subject_name,
trigger,
started_at_ms,
requested_additional_permissions: guardian_review_requested_additional_permissions(
&notification.action,
),
requested_network_access: guardian_review_requested_network_access(
&notification.action,
),
};
let Some(completed_at_ms) = option_i64_to_u64(Some(notification.completed_at_ms)) else {
return;
};
self.emit_review_event(
pending_review,
Reviewer::Guardian,
status,
resolution,
completed_at_ms,
out,
);
}
fn ingest_turn_steer_response(
&mut self,
connection_id: u64,
@@ -1068,6 +1394,73 @@ impl AnalyticsReducer {
}));
}
fn emit_review_event(
&mut self,
pending_review: PendingReviewState,
reviewer: Reviewer,
status: ReviewStatus,
resolution: ReviewResolution,
completed_at_ms: u64,
out: &mut Vec<TrackEventRequest>,
) {
if let Some(item_key) = item_review_summary_key(&pending_review) {
self.record_item_review_summary(
item_key,
reviewer,
status,
resolution,
&pending_review,
);
}
let Some((connection_state, thread_metadata)) =
self.thread_context_or_warn(AnalyticsDropSite::review(&pending_review))
else {
return;
};
out.push(TrackEventRequest::ReviewEvent(CodexReviewEventRequest {
event_type: "codex_review_event",
event_params: CodexReviewEventParams {
thread_id: pending_review.thread_id,
turn_id: pending_review.turn_id,
item_id: pending_review.item_id,
review_id: pending_review.review_id,
app_server_client: connection_state.app_server_client.clone(),
runtime: connection_state.runtime.clone(),
thread_source: thread_metadata.thread_source,
subagent_source: thread_metadata.subagent_source.clone(),
parent_thread_id: thread_metadata.parent_thread_id.clone(),
subject_kind: pending_review.subject_kind,
subject_name: pending_review.subject_name,
reviewer,
trigger: pending_review.trigger,
status,
resolution,
started_at_ms: pending_review.started_at_ms,
completed_at_ms,
duration_ms: observed_duration_ms(pending_review.started_at_ms, completed_at_ms),
},
}));
}
fn record_item_review_summary(
&mut self,
item_key: ToolItemKey,
reviewer: Reviewer,
status: ReviewStatus,
resolution: ReviewResolution,
pending_review: &PendingReviewState,
) {
let summary = self.item_review_summaries.entry(item_key).or_default();
summary.review_count += 1;
match reviewer {
Reviewer::Guardian => summary.guardian_review_count += 1,
Reviewer::User => summary.user_review_count += 1,
}
summary.final_approval_outcome = Some(final_approval_outcome(reviewer, status, resolution));
summary.requested_additional_permissions |= pending_review.requested_additional_permissions;
summary.requested_network_access |= pending_review.requested_network_access;
}
async fn maybe_emit_turn_event(&mut self, turn_id: &str, out: &mut Vec<TrackEventRequest>) {
let Some(turn_state) = self.turns.get(turn_id) else {
return;
@@ -1204,21 +1597,41 @@ fn tracked_tool_item_id(item: &ThreadItem) -> Option<&str> {
}
}
fn tool_item_event(
thread_id: &str,
turn_id: &str,
item: &ThreadItem,
fn item_review_summary_key(pending_review: &PendingReviewState) -> Option<ToolItemKey> {
match pending_review.subject_kind {
ReviewSubjectKind::CommandExecution
| ReviewSubjectKind::FileChange
| ReviewSubjectKind::McpToolCall => Some(ToolItemKey {
thread_id: pending_review.thread_id.clone(),
turn_id: pending_review.turn_id.clone(),
item_id: pending_review.item_id.clone()?,
}),
ReviewSubjectKind::Permissions | ReviewSubjectKind::NetworkAccess => None,
}
}
struct ToolItemEventInput<'a> {
thread_id: &'a str,
turn_id: &'a str,
item: &'a ThreadItem,
started_at_ms: u64,
completed_at_ms: u64,
connection_state: &ConnectionState,
thread_metadata: &ThreadMetadataState,
) -> Option<TrackEventRequest> {
let context = ToolItemContext {
connection_state: &'a ConnectionState,
thread_metadata: &'a ThreadMetadataState,
review_summary: Option<&'a ItemReviewSummary>,
}
fn tool_item_event(input: ToolItemEventInput<'_>) -> Option<TrackEventRequest> {
let ToolItemEventInput {
thread_id,
turn_id,
item,
started_at_ms,
completed_at_ms,
connection_state,
thread_metadata,
};
review_summary,
} = input;
match item {
ThreadItem::CommandExecution {
id,
@@ -1241,7 +1654,13 @@ fn tool_item_event(
failure_kind,
execution_duration_ms: option_i64_to_u64(*duration_ms),
},
context,
ToolItemContext {
started_at_ms,
completed_at_ms,
connection_state,
thread_metadata,
review_summary,
},
);
Some(TrackEventRequest::CommandExecution(
CodexCommandExecutionEventRequest {
@@ -1276,7 +1695,13 @@ fn tool_item_event(
failure_kind,
execution_duration_ms: None,
},
context,
ToolItemContext {
started_at_ms,
completed_at_ms,
connection_state,
thread_metadata,
review_summary,
},
);
Some(TrackEventRequest::FileChange(CodexFileChangeEventRequest {
event_type: "codex_file_change_event",
@@ -1310,7 +1735,13 @@ fn tool_item_event(
failure_kind,
execution_duration_ms: option_i64_to_u64(*duration_ms),
},
context,
ToolItemContext {
started_at_ms,
completed_at_ms,
connection_state,
thread_metadata,
review_summary,
},
);
Some(TrackEventRequest::McpToolCall(
CodexMcpToolCallEventRequest {
@@ -1347,7 +1778,13 @@ fn tool_item_event(
failure_kind,
execution_duration_ms: option_i64_to_u64(*duration_ms),
},
context,
ToolItemContext {
started_at_ms,
completed_at_ms,
connection_state,
thread_metadata,
review_summary,
},
);
Some(TrackEventRequest::DynamicToolCall(
CodexDynamicToolCallEventRequest {
@@ -1385,7 +1822,13 @@ fn tool_item_event(
failure_kind,
execution_duration_ms: None,
},
context,
ToolItemContext {
started_at_ms,
completed_at_ms,
connection_state,
thread_metadata,
review_summary,
},
);
Some(TrackEventRequest::CollabAgentToolCall(
CodexCollabAgentToolCallEventRequest {
@@ -1434,7 +1877,13 @@ fn tool_item_event(
failure_kind: None,
execution_duration_ms: None,
},
context,
ToolItemContext {
started_at_ms,
completed_at_ms,
connection_state,
thread_metadata,
review_summary,
},
);
Some(TrackEventRequest::WebSearch(CodexWebSearchEventRequest {
event_type: "codex_web_search_event",
@@ -1464,7 +1913,13 @@ fn tool_item_event(
failure_kind,
execution_duration_ms: None,
},
context,
ToolItemContext {
started_at_ms,
completed_at_ms,
connection_state,
thread_metadata,
review_summary,
},
);
Some(TrackEventRequest::ImageGeneration(
CodexImageGenerationEventRequest {
@@ -1518,6 +1973,7 @@ struct ToolItemContext<'a> {
completed_at_ms: u64,
connection_state: &'a ConnectionState,
thread_metadata: &'a ThreadMetadataState,
review_summary: Option<&'a ItemReviewSummary>,
}
fn tool_item_base(
@@ -1529,6 +1985,7 @@ fn tool_item_base(
context: ToolItemContext<'_>,
) -> CodexToolItemEventBase {
let thread_metadata = context.thread_metadata;
let review_summary = context.review_summary.cloned().unwrap_or_default();
CodexToolItemEventBase {
thread_id: thread_id.to_string(),
turn_id: turn_id.to_string(),
@@ -1546,14 +2003,16 @@ fn tool_item_base(
// full upstream execution time.
duration_ms: observed_duration_ms(context.started_at_ms, context.completed_at_ms),
execution_duration_ms: outcome.execution_duration_ms,
review_count: 0,
guardian_review_count: 0,
user_review_count: 0,
final_approval_outcome: ToolItemFinalApprovalOutcome::Unknown,
review_count: review_summary.review_count,
guardian_review_count: review_summary.guardian_review_count,
user_review_count: review_summary.user_review_count,
final_approval_outcome: review_summary
.final_approval_outcome
.unwrap_or(FinalApprovalOutcome::Unknown),
terminal_status: outcome.terminal_status,
failure_kind: outcome.failure_kind,
requested_additional_permissions: false,
requested_network_access: false,
requested_additional_permissions: review_summary.requested_additional_permissions,
requested_network_access: review_summary.requested_network_access,
}
}
@@ -1561,6 +2020,195 @@ fn observed_duration_ms(started_at_ms: u64, completed_at_ms: u64) -> Option<u64>
completed_at_ms.checked_sub(started_at_ms)
}
fn user_review_id(request_id: &RequestId) -> String {
format!("user:{request_id}")
}
fn command_execution_review_result(
decision: CommandExecutionApprovalDecision,
) -> (ReviewStatus, ReviewResolution) {
match decision {
CommandExecutionApprovalDecision::Accept => {
(ReviewStatus::Approved, ReviewResolution::None)
}
CommandExecutionApprovalDecision::AcceptForSession => {
(ReviewStatus::Approved, ReviewResolution::SessionApproval)
}
CommandExecutionApprovalDecision::AcceptWithExecpolicyAmendment { .. } => (
ReviewStatus::Approved,
ReviewResolution::ExecPolicyAmendment,
),
CommandExecutionApprovalDecision::ApplyNetworkPolicyAmendment {
network_policy_amendment,
} => match network_policy_amendment.action {
NetworkPolicyRuleAction::Allow => (
ReviewStatus::Approved,
ReviewResolution::NetworkPolicyAmendment,
),
NetworkPolicyRuleAction::Deny => (
ReviewStatus::Denied,
ReviewResolution::NetworkPolicyAmendment,
),
},
CommandExecutionApprovalDecision::Decline => (ReviewStatus::Denied, ReviewResolution::None),
CommandExecutionApprovalDecision::Cancel => (ReviewStatus::Aborted, ReviewResolution::None),
}
}
fn file_change_review_result(
decision: FileChangeApprovalDecision,
) -> (ReviewStatus, ReviewResolution) {
match decision {
FileChangeApprovalDecision::Accept => (ReviewStatus::Approved, ReviewResolution::None),
FileChangeApprovalDecision::AcceptForSession => {
(ReviewStatus::Approved, ReviewResolution::SessionApproval)
}
FileChangeApprovalDecision::Decline => (ReviewStatus::Denied, ReviewResolution::None),
FileChangeApprovalDecision::Cancel => (ReviewStatus::Aborted, ReviewResolution::None),
}
}
fn effective_permissions_review_result(
response: &CoreRequestPermissionsResponse,
) -> (ReviewStatus, ReviewResolution) {
if response.permissions.is_empty() {
return (ReviewStatus::Denied, ReviewResolution::None);
}
match response.scope {
CorePermissionGrantScope::Turn => (ReviewStatus::Approved, ReviewResolution::None),
CorePermissionGrantScope::Session => {
(ReviewStatus::Approved, ReviewResolution::SessionApproval)
}
}
}
fn guardian_review_result(
status: GuardianApprovalReviewStatus,
) -> Option<(ReviewStatus, ReviewResolution)> {
match status {
GuardianApprovalReviewStatus::InProgress => None,
GuardianApprovalReviewStatus::Approved => {
Some((ReviewStatus::Approved, ReviewResolution::None))
}
GuardianApprovalReviewStatus::Denied => {
Some((ReviewStatus::Denied, ReviewResolution::None))
}
GuardianApprovalReviewStatus::TimedOut => {
Some((ReviewStatus::TimedOut, ReviewResolution::None))
}
GuardianApprovalReviewStatus::Aborted => {
Some((ReviewStatus::Aborted, ReviewResolution::None))
}
}
}
fn guardian_review_subject_metadata(
action: &GuardianApprovalReviewAction,
) -> (ReviewSubjectKind, String, ReviewTrigger) {
match action {
GuardianApprovalReviewAction::Command { .. } => (
ReviewSubjectKind::CommandExecution,
"command_execution".to_string(),
ReviewTrigger::Initial,
),
GuardianApprovalReviewAction::Execve { .. } => (
ReviewSubjectKind::CommandExecution,
"command_execution".to_string(),
ReviewTrigger::ExecveIntercept,
),
GuardianApprovalReviewAction::ApplyPatch { .. } => (
ReviewSubjectKind::FileChange,
"apply_patch".to_string(),
ReviewTrigger::SandboxDenial,
),
GuardianApprovalReviewAction::NetworkAccess { .. } => (
ReviewSubjectKind::NetworkAccess,
"network_access".to_string(),
ReviewTrigger::NetworkPolicyDenial,
),
GuardianApprovalReviewAction::RequestPermissions { permissions, .. } => {
let requested_network_access = permissions
.network
.as_ref()
.and_then(|network| network.enabled)
.unwrap_or(false);
let trigger = if requested_network_access {
ReviewTrigger::NetworkPolicyDenial
} else if permissions.file_system.is_some() {
ReviewTrigger::SandboxDenial
} else {
ReviewTrigger::Initial
};
(
ReviewSubjectKind::Permissions,
"permissions".to_string(),
trigger,
)
}
GuardianApprovalReviewAction::McpToolCall { tool_name, .. } => (
ReviewSubjectKind::McpToolCall,
tool_name.clone(),
ReviewTrigger::Initial,
),
}
}
fn guardian_review_requested_additional_permissions(action: &GuardianApprovalReviewAction) -> bool {
match action {
GuardianApprovalReviewAction::ApplyPatch { .. }
| GuardianApprovalReviewAction::NetworkAccess { .. } => true,
GuardianApprovalReviewAction::RequestPermissions { permissions, .. } => {
guardian_review_request_permissions_network_enabled(permissions)
|| permissions.file_system.is_some()
}
GuardianApprovalReviewAction::Command { .. }
| GuardianApprovalReviewAction::Execve { .. }
| GuardianApprovalReviewAction::McpToolCall { .. } => false,
}
}
fn guardian_review_requested_network_access(action: &GuardianApprovalReviewAction) -> bool {
match action {
GuardianApprovalReviewAction::NetworkAccess { .. } => true,
GuardianApprovalReviewAction::RequestPermissions { permissions, .. } => {
guardian_review_request_permissions_network_enabled(permissions)
}
GuardianApprovalReviewAction::ApplyPatch { .. }
| GuardianApprovalReviewAction::Command { .. }
| GuardianApprovalReviewAction::Execve { .. }
| GuardianApprovalReviewAction::McpToolCall { .. } => false,
}
}
fn guardian_review_request_permissions_network_enabled(
permissions: &RequestPermissionProfile,
) -> bool {
permissions
.network
.as_ref()
.and_then(|network| network.enabled)
.unwrap_or(false)
}
fn final_approval_outcome(
reviewer: Reviewer,
status: ReviewStatus,
resolution: ReviewResolution,
) -> FinalApprovalOutcome {
match (reviewer, status, resolution) {
(Reviewer::Guardian, ReviewStatus::Approved, _) => FinalApprovalOutcome::GuardianApproved,
(Reviewer::Guardian, ReviewStatus::Denied, _) => FinalApprovalOutcome::GuardianDenied,
(Reviewer::Guardian, _, _) => FinalApprovalOutcome::GuardianAborted,
(Reviewer::User, ReviewStatus::Approved, ReviewResolution::SessionApproval) => {
FinalApprovalOutcome::UserApprovedForSession
}
(Reviewer::User, ReviewStatus::Approved, _) => FinalApprovalOutcome::UserApproved,
(Reviewer::User, ReviewStatus::Denied, _) => FinalApprovalOutcome::UserDenied,
(Reviewer::User, _, _) => FinalApprovalOutcome::UserAborted,
}
}
fn command_execution_tool_name(source: CommandExecutionSource) -> &'static str {
match source {
CommandExecutionSource::UnifiedExecStartup
@@ -1990,4 +2638,13 @@ mod tests {
"external_sandbox"
);
}
#[test]
fn guardian_review_result_maps_terminal_statuses() {
assert!(guardian_review_result(GuardianApprovalReviewStatus::InProgress).is_none());
assert!(matches!(
guardian_review_result(GuardianApprovalReviewStatus::TimedOut),
Some((ReviewStatus::TimedOut, ReviewResolution::None))
));
}
}
codex-rs/app-server/src/bespoke_event_handling.rs
+5 -1
View File
@@ -772,6 +772,7 @@ pub(crate) async fn apply_bespoke_event_handling(
requested_permissions,
request_cwd,
pending_request_id,
outgoing,
receiver: rx,
request_permissions_guard: permission_guard,
};
@@ -1745,11 +1746,12 @@ async fn on_request_permissions_response(
requested_permissions,
request_cwd,
pending_request_id,
outgoing,
receiver,
request_permissions_guard,
} = pending_response;
let response = receiver.await;
resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await;
resolve_server_request_on_thread_listener(&thread_state, pending_request_id.clone()).await;
drop(request_permissions_guard);
let Some(response) = request_permissions_response_from_client_result(
requested_permissions,
@@ -1758,6 +1760,7 @@ async fn on_request_permissions_response(
) else {
return;
};
outgoing.track_effective_permissions_approval_response(pending_request_id, response.clone());
if let Err(err) = conversation
.submit(Op::RequestPermissionsResponse {
@@ -1775,6 +1778,7 @@ struct PendingRequestPermissionsResponse {
requested_permissions: CoreRequestPermissionProfile,
request_cwd: AbsolutePathBuf,
pending_request_id: RequestId,
outgoing: ThreadScopedOutgoingMessageSender,
receiver: oneshot::Receiver<ClientRequestResult>,
request_permissions_guard: ThreadWatchActiveGuard,
}
codex-rs/app-server/src/outgoing_message.rs
+45 -14
View File
@@ -13,9 +13,11 @@ use codex_app_server_protocol::Result;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ServerRequestPayload;
use codex_app_server_protocol::ServerResponse;
use codex_otel::span_w3c_trace_context;
use codex_protocol::ThreadId;
use codex_protocol::protocol::W3cTraceContext;
use codex_protocol::request_permissions::RequestPermissionsResponse;
use tokio::sync::Mutex;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
@@ -141,6 +143,20 @@ impl ThreadScopedOutgoingMessageSender {
.await
}
pub(crate) fn track_effective_permissions_approval_response(
&self,
request_id: RequestId,
response: RequestPermissionsResponse,
) {
self.outgoing
.analytics_events_client
.track_effective_permissions_approval_response(
now_unix_timestamp_ms(),
request_id,
response,
);
}
pub(crate) async fn send_server_notification(&self, notification: ServerNotification) {
self.outgoing
.analytics_events_client
@@ -360,7 +376,9 @@ impl OutgoingMessageSender {
match entry {
Some((id, entry)) => {
let completed_at_ms = now_unix_timestamp_ms();
if let Ok(response) = entry.request.response_from_result(result.clone()) {
if let Ok(response) = entry.request.response_from_result(result.clone())
&& !matches!(response, ServerResponse::PermissionsRequestApproval { .. })
{
self.analytics_events_client
.track_server_response(completed_at_ms, response);
}
@@ -380,6 +398,8 @@ impl OutgoingMessageSender {
match entry {
Some((id, entry)) => {
warn!("client responded with error for {id:?}: {error:?}");
self.analytics_events_client
.track_server_request_aborted(now_unix_timestamp_ms(), id.clone());
if let Err(err) = entry.callback.send(Err(error)) {
warn!("could not notify callback for {id:?} due to: {err:?}");
}
@@ -391,7 +411,14 @@ impl OutgoingMessageSender {
}
pub(crate) async fn cancel_request(&self, id: &RequestId) -> bool {
self.take_request_callback(id).await.is_some()
let entry = self.take_request_callback(id).await;
if let Some((request_id, _entry)) = entry {
self.analytics_events_client
.track_server_request_aborted(now_unix_timestamp_ms(), request_id);
true
} else {
false
}
}
pub(crate) async fn cancel_all_requests(&self, error: Option<JSONRPCErrorError>) {
@@ -403,12 +430,14 @@ impl OutgoingMessageSender {
.collect::<Vec<_>>()
};
if let Some(error) = error {
for entry in entries {
if let Err(err) = entry.callback.send(Err(error.clone())) {
let request_id = entry.request.id();
warn!("could not notify callback for {request_id:?} due to: {err:?}");
}
for entry in entries {
self.analytics_events_client
.track_server_request_aborted(now_unix_timestamp_ms(), entry.request.id().clone());
if let Some(error) = error.as_ref()
&& let Err(err) = entry.callback.send(Err(error.clone()))
{
let request_id = entry.request.id();
warn!("could not notify callback for {request_id:?} due to: {err:?}");
}
}
}
@@ -459,12 +488,14 @@ impl OutgoingMessageSender {
entries
};
if let Some(error) = error {
for entry in entries {
if let Err(err) = entry.callback.send(Err(error.clone())) {
let request_id = entry.request.id();
warn!("could not notify callback for {request_id:?} due to: {err:?}",);
}
for entry in entries {
self.analytics_events_client
.track_server_request_aborted(now_unix_timestamp_ms(), entry.request.id().clone());
if let Some(error) = error.as_ref()
&& let Err(err) = entry.callback.send(Err(error.clone()))
{
let request_id = entry.request.id();
warn!("could not notify callback for {request_id:?} due to: {err:?}",);
}
}
}
codex-rs/core/src/guardian/review.rs
+14 -7
View File
@@ -1,5 +1,3 @@
use std::sync::Arc;
use codex_analytics::GuardianApprovalRequestSource;
use codex_analytics::GuardianReviewAnalyticsResult;
use codex_analytics::GuardianReviewDecision;
@@ -18,6 +16,7 @@ use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::WarningEvent;
use std::sync::Arc;
use tokio::sync::oneshot;
use tokio_util::sync::CancellationToken;
@@ -164,11 +163,12 @@ fn track_guardian_review(
session: &Session,
tracking: &GuardianReviewTrackContext,
result: GuardianReviewAnalyticsResult,
completed_at_ms: u64,
) {
session
.services
.analytics_events_client
.track_guardian_review(tracking, result);
.track_guardian_review(tracking, result, completed_at_ms);
}
async fn record_guardian_non_denial(session: &Arc<Session>, turn_id: &str) {
@@ -277,6 +277,7 @@ async fn run_guardian_review(
.as_ref()
.is_some_and(CancellationToken::is_cancelled)
{
let completed_at_ms = now_unix_timestamp_ms();
track_guardian_review(
session.as_ref(),
&review_tracking,
@@ -286,6 +287,7 @@ async fn run_guardian_review(
failure_reason: Some(GuardianReviewFailureReason::Cancelled),
..GuardianReviewAnalyticsResult::without_session()
},
completed_at_ms.try_into().unwrap_or_default(),
);
session
.send_event(
@@ -295,7 +297,7 @@ async fn run_guardian_review(
target_item_id,
turn_id: assessment_turn_id.clone(),
started_at_ms,
completed_at_ms: Some(now_unix_timestamp_ms()),
completed_at_ms: Some(completed_at_ms),
status: GuardianAssessmentStatus::Aborted,
risk_level: None,
user_authorization: None,
@@ -321,6 +323,7 @@ async fn run_guardian_review(
))
.await;
let completed_at_ms = now_unix_timestamp_ms();
let (assessment, count_denial_for_circuit_breaker) = match outcome {
GuardianReviewOutcome::Completed(assessment) => {
let approved = matches!(assessment.outcome, GuardianAssessmentOutcome::Allow);
@@ -344,6 +347,7 @@ async fn run_guardian_review(
outcome: Some(assessment.outcome),
..analytics_result
},
completed_at_ms.try_into().unwrap_or_default(),
);
let count_denial_for_circuit_breaker =
matches!(assessment.outcome, GuardianAssessmentOutcome::Deny);
@@ -363,6 +367,7 @@ async fn run_guardian_review(
failure_reason: Some(error.failure_reason()),
..analytics_result
},
completed_at_ms.try_into().unwrap_or_default(),
);
session
.send_event(
@@ -380,7 +385,7 @@ async fn run_guardian_review(
target_item_id,
turn_id: assessment_turn_id.clone(),
started_at_ms,
completed_at_ms: Some(now_unix_timestamp_ms()),
completed_at_ms: Some(completed_at_ms),
status: GuardianAssessmentStatus::TimedOut,
risk_level: None,
user_authorization: None,
@@ -403,6 +408,7 @@ async fn run_guardian_review(
failure_reason: Some(error.failure_reason()),
..analytics_result
},
completed_at_ms.try_into().unwrap_or_default(),
);
session
.send_event(
@@ -412,7 +418,7 @@ async fn run_guardian_review(
target_item_id,
turn_id: assessment_turn_id.clone(),
started_at_ms,
completed_at_ms: Some(now_unix_timestamp_ms()),
completed_at_ms: Some(completed_at_ms),
status: GuardianAssessmentStatus::Aborted,
risk_level: None,
user_authorization: None,
@@ -446,6 +452,7 @@ async fn run_guardian_review(
failure_reason: Some(error.failure_reason()),
..analytics_result
},
completed_at_ms.try_into().unwrap_or_default(),
);
(
GuardianAssessment {
@@ -507,7 +514,7 @@ async fn run_guardian_review(
target_item_id,
turn_id: assessment_turn_id.clone(),
started_at_ms,
completed_at_ms: Some(now_unix_timestamp_ms()),
completed_at_ms: Some(completed_at_ms),
status,
risk_level: Some(assessment.risk_level),
user_authorization: Some(assessment.user_authorization),