From df90ed05fe07b0d8f400864bc7e4099056b5ae2c Mon Sep 17 00:00:00 2001 From: Albin Cassirer Date: Sun, 19 Apr 2026 17:22:16 -0700 Subject: [PATCH] Add review completed observation projection\n\nIntroduce a generic review.completed observation that separates the reviewed action\nfrom the reviewer response. Model user and guardian reviewer responses in the\nshared event stream, while projecting only guardian responses into the current\nlegacy analytics event.\n\nKeep the guardian analytics mapping isolated in a dedicated projection module\nand cover the reviewed-action variants with targeted projection tests. Update the\ndesign doc to clarify that field uses are the analytics selection marker, with\ndetail/class metadata retained for review, auditing, and sink-specific behavior. --- .../analytics/src/analytics_client_tests.rs | 160 ++++++ codex-rs/analytics/src/lib.rs | 1 + codex-rs/analytics/src/observation_reducer.rs | 25 + .../src/review_observation_projection.rs | 471 ++++++++++++++++++ codex-rs/observability/src/events.rs | 2 + codex-rs/observability/src/events/review.rs | 272 ++++++++++ docs/observability-event-stream-design.md | 26 +- 7 files changed, 946 insertions(+), 11 deletions(-) create mode 100644 codex-rs/analytics/src/review_observation_projection.rs create mode 100644 codex-rs/observability/src/events/review.rs diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 6087fcc24a..1ae1c60544 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -2428,6 +2428,166 @@ async fn compaction_ended_observation_matches_legacy_fact() { assert_eq!(observation_payload, legacy_payload); } +#[tokio::test] +async fn review_completed_observation_matches_legacy_guardian_fact() { + use crate::events as analytics_events; + use codex_observability::events as observation_events; + + let mut legacy_reducer = AnalyticsReducer::default(); + let mut observation_reducer = AnalyticsObservationReducer::default(); + let mut legacy_events = Vec::new(); + let mut observation_events = Vec::new(); + + ingest_initialize(&mut legacy_reducer, &mut legacy_events).await; + legacy_reducer + .ingest( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_start_response( + "thread-guardian", + /*ephemeral*/ false, + "gpt-5", + )), + }, + &mut legacy_events, + ) + .await; + legacy_events.clear(); + + observation_reducer + .ingest_existing_fact_for_test(sample_initialize_fact(), &mut observation_events) + .await; + observation_reducer + .ingest_existing_fact_for_test( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_start_response( + "thread-guardian", + /*ephemeral*/ false, + "gpt-5", + )), + }, + &mut observation_events, + ) + .await; + observation_events.clear(); + + legacy_reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::GuardianReview(Box::new( + analytics_events::GuardianReviewEventParams { + thread_id: "thread-guardian".to_string(), + turn_id: "turn-guardian".to_string(), + review_id: "review-1".to_string(), + target_item_id: "item-1".to_string(), + retry_reason: Some("parse_retry".to_string()), + approval_request_source: + analytics_events::GuardianApprovalRequestSource::DelegatedSubagent, + reviewed_action: analytics_events::GuardianReviewedAction::McpToolCall { + server: "drive".to_string(), + tool_name: "search".to_string(), + connector_id: Some("drive-connector".to_string()), + connector_name: Some("Drive".to_string()), + tool_title: Some("Search Drive".to_string()), + }, + reviewed_action_truncated: false, + decision: analytics_events::GuardianReviewDecision::Aborted, + terminal_status: analytics_events::GuardianReviewTerminalStatus::FailedClosed, + failure_reason: Some(analytics_events::GuardianReviewFailureReason::ParseError), + risk_level: Some(analytics_events::GuardianReviewRiskLevel::High), + user_authorization: Some( + analytics_events::GuardianReviewUserAuthorization::Medium, + ), + outcome: Some(analytics_events::GuardianReviewOutcome::Deny), + rationale: Some("review could not parse final policy".to_string()), + guardian_thread_id: Some("guardian-thread-1".to_string()), + guardian_session_kind: Some( + analytics_events::GuardianReviewSessionKind::TrunkNew, + ), + guardian_model: Some("gpt-5".to_string()), + guardian_reasoning_effort: Some("medium".to_string()), + had_prior_review_context: Some(true), + review_timeout_ms: 30_000, + tool_call_count: 2, + time_to_first_token_ms: Some(100), + completion_latency_ms: Some(1500), + started_at: 100, + completed_at: Some(102), + input_tokens: Some(123), + cached_input_tokens: Some(45), + output_tokens: Some(140), + reasoning_output_tokens: Some(13), + total_tokens: Some(321), + }, + ))), + &mut legacy_events, + ) + .await; + + observation_reducer + .ingest_review_completed( + observation_events::ReviewCompleted { + thread_id: "thread-guardian", + turn_id: "turn-guardian", + review_id: "review-1", + target_item_id: "item-1", + retry_reason: Some("parse_retry"), + request_source: observation_events::ReviewRequestSource::DelegatedSubagent, + reviewed_action: observation_events::ReviewedAction::McpToolCall { + server: "drive", + tool_name: "search", + connector_id: Some("drive-connector"), + connector_name: Some("Drive"), + tool_title: Some("Search Drive"), + }, + reviewed_action_truncated: false, + response: observation_events::ReviewResponse::Guardian( + observation_events::GuardianReviewResponse { + decision: observation_events::ReviewDecision::Aborted, + terminal_status: observation_events::ReviewTerminalStatus::FailedClosed { + failure_reason: Some( + observation_events::ReviewFailureReason::ParseError, + ), + }, + risk_level: Some(observation_events::ReviewRiskLevel::High), + user_authorization: Some( + observation_events::ReviewUserAuthorization::Medium, + ), + outcome: Some(observation_events::ReviewOutcome::Deny), + rationale: Some("review could not parse final policy"), + session: Some(observation_events::GuardianReviewSession { + guardian_thread_id: "guardian-thread-1", + session_kind: observation_events::GuardianReviewSessionKind::TrunkNew, + model: "gpt-5", + reasoning_effort: Some("medium"), + had_prior_review_context: true, + }), + review_timeout_ms: 30_000, + tool_call_count: 2, + time_to_first_token_ms: Some(100), + completion_latency_ms: Some(1500), + token_usage: Some(observation_events::TurnTokenUsage { + input_tokens: 123, + cached_input_tokens: 45, + output_tokens: 140, + reasoning_output_tokens: 13, + total_tokens: 321, + }), + }, + ), + started_at: 100, + ended_at: 102, + }, + &mut observation_events, + ) + .await; + + let legacy_payload = serde_json::to_value(&legacy_events).expect("serialize legacy events"); + let observation_payload = + serde_json::to_value(&observation_events).expect("serialize observation events"); + assert_eq!(observation_payload, legacy_payload); +} + fn legacy_event_created_at(event: &TrackEventRequest) -> i64 { serde_json::to_value(event) .expect("serialize event") diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index 1283206986..26acc6c311 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -4,6 +4,7 @@ mod facts; mod observation_projection; mod observation_reducer; mod reducer; +mod review_observation_projection; use std::time::SystemTime; use std::time::UNIX_EPOCH; diff --git a/codex-rs/analytics/src/observation_reducer.rs b/codex-rs/analytics/src/observation_reducer.rs index 093bed6be3..8423934b82 100644 --- a/codex-rs/analytics/src/observation_reducer.rs +++ b/codex-rs/analytics/src/observation_reducer.rs @@ -11,6 +11,7 @@ use crate::facts::AnalyticsFact; use crate::facts::CustomAnalyticsFact; use crate::observation_projection; use crate::reducer::AnalyticsReducer; +use crate::review_observation_projection; use codex_observability::events; /// Analytics reducer entrypoint for typed observations. @@ -156,6 +157,30 @@ impl AnalyticsObservationReducer { .await; } + /// Ingests a review.completed observation for the legacy guardian analytics event. + /// + /// User review responses are represented in the shared observation type but + /// do not have a legacy analytics event today. + pub(crate) async fn ingest_review_completed( + &mut self, + observation: events::ReviewCompleted<'_>, + out: &mut Vec, + ) { + let Some(guardian_review) = + review_observation_projection::legacy_guardian_review_event(observation) + else { + return; + }; + self.legacy + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::GuardianReview(Box::new( + guardian_review, + ))), + out, + ) + .await; + } + /// Ingests a hook.run_completed observation and emits the current analytics event. pub(crate) fn ingest_hook_run_completed( &mut self, diff --git a/codex-rs/analytics/src/review_observation_projection.rs b/codex-rs/analytics/src/review_observation_projection.rs new file mode 100644 index 0000000000..7afa3b46cd --- /dev/null +++ b/codex-rs/analytics/src/review_observation_projection.rs @@ -0,0 +1,471 @@ +//! Projection from review observations into the current analytics schema. + +use crate::events::GuardianApprovalRequestSource as AnalyticsGuardianApprovalRequestSource; +use crate::events::GuardianCommandSource as AnalyticsGuardianCommandSource; +use crate::events::GuardianReviewDecision as AnalyticsGuardianReviewDecision; +use crate::events::GuardianReviewEventParams; +use crate::events::GuardianReviewFailureReason as AnalyticsGuardianReviewFailureReason; +use crate::events::GuardianReviewOutcome as AnalyticsGuardianReviewOutcome; +use crate::events::GuardianReviewRiskLevel as AnalyticsGuardianReviewRiskLevel; +use crate::events::GuardianReviewSessionKind as AnalyticsGuardianReviewSessionKind; +use crate::events::GuardianReviewTerminalStatus as AnalyticsGuardianReviewTerminalStatus; +use crate::events::GuardianReviewUserAuthorization as AnalyticsGuardianReviewUserAuthorization; +use crate::events::GuardianReviewedAction as AnalyticsGuardianReviewedAction; +use codex_observability::events; +use codex_protocol::approvals::NetworkApprovalProtocol; +use codex_protocol::models::PermissionProfile; +use codex_protocol::models::SandboxPermissions; + +/// Projects a generic review completion into the legacy guardian analytics event. +/// +/// This preserves the existing guardian analytics payload, including reviewed +/// action details and guardian rationale that are explicitly marked for +/// analytics on the observation fields. +/// +/// User review responses intentionally return None: the current analytics +/// schema only has a guardian review event, while the shared observation is +/// generic enough to represent both user and guardian review completions. +pub(crate) fn legacy_guardian_review_event( + observation: events::ReviewCompleted<'_>, +) -> Option { + let events::ReviewResponse::Guardian(response) = observation.response else { + return None; + }; + let (terminal_status, failure_reason) = match response.terminal_status { + events::ReviewTerminalStatus::Approved => { + (AnalyticsGuardianReviewTerminalStatus::Approved, None) + } + events::ReviewTerminalStatus::Denied => { + (AnalyticsGuardianReviewTerminalStatus::Denied, None) + } + events::ReviewTerminalStatus::Aborted { failure_reason } => ( + AnalyticsGuardianReviewTerminalStatus::Aborted, + failure_reason.map(review_failure_reason), + ), + events::ReviewTerminalStatus::TimedOut { failure_reason } => ( + AnalyticsGuardianReviewTerminalStatus::TimedOut, + failure_reason.map(review_failure_reason), + ), + events::ReviewTerminalStatus::FailedClosed { failure_reason } => ( + AnalyticsGuardianReviewTerminalStatus::FailedClosed, + failure_reason.map(review_failure_reason), + ), + }; + let token_usage = response.token_usage; + let guardian_session = response.session; + + Some(GuardianReviewEventParams { + thread_id: observation.thread_id.to_string(), + turn_id: observation.turn_id.to_string(), + review_id: observation.review_id.to_string(), + target_item_id: observation.target_item_id.to_string(), + retry_reason: observation.retry_reason.map(str::to_string), + approval_request_source: match observation.request_source { + events::ReviewRequestSource::MainTurn => { + AnalyticsGuardianApprovalRequestSource::MainTurn + } + events::ReviewRequestSource::DelegatedSubagent => { + AnalyticsGuardianApprovalRequestSource::DelegatedSubagent + } + }, + reviewed_action: reviewed_action(observation.reviewed_action), + reviewed_action_truncated: observation.reviewed_action_truncated, + decision: match response.decision { + events::ReviewDecision::Approved => AnalyticsGuardianReviewDecision::Approved, + events::ReviewDecision::Denied => AnalyticsGuardianReviewDecision::Denied, + events::ReviewDecision::Aborted => AnalyticsGuardianReviewDecision::Aborted, + }, + terminal_status, + failure_reason, + risk_level: response.risk_level.map(|risk_level| match risk_level { + events::ReviewRiskLevel::Low => AnalyticsGuardianReviewRiskLevel::Low, + events::ReviewRiskLevel::Medium => AnalyticsGuardianReviewRiskLevel::Medium, + events::ReviewRiskLevel::High => AnalyticsGuardianReviewRiskLevel::High, + events::ReviewRiskLevel::Critical => AnalyticsGuardianReviewRiskLevel::Critical, + }), + user_authorization: response.user_authorization.map(|user_authorization| { + match user_authorization { + events::ReviewUserAuthorization::Unknown => { + AnalyticsGuardianReviewUserAuthorization::Unknown + } + events::ReviewUserAuthorization::Low => { + AnalyticsGuardianReviewUserAuthorization::Low + } + events::ReviewUserAuthorization::Medium => { + AnalyticsGuardianReviewUserAuthorization::Medium + } + events::ReviewUserAuthorization::High => { + AnalyticsGuardianReviewUserAuthorization::High + } + } + }), + outcome: response.outcome.map(|outcome| match outcome { + events::ReviewOutcome::Allow => AnalyticsGuardianReviewOutcome::Allow, + events::ReviewOutcome::Deny => AnalyticsGuardianReviewOutcome::Deny, + }), + rationale: response.rationale.map(str::to_string), + guardian_thread_id: guardian_session.map(|session| session.guardian_thread_id.to_string()), + guardian_session_kind: guardian_session.map(|session| match session.session_kind { + events::GuardianReviewSessionKind::TrunkNew => { + AnalyticsGuardianReviewSessionKind::TrunkNew + } + events::GuardianReviewSessionKind::TrunkReused => { + AnalyticsGuardianReviewSessionKind::TrunkReused + } + events::GuardianReviewSessionKind::EphemeralForked => { + AnalyticsGuardianReviewSessionKind::EphemeralForked + } + }), + guardian_model: guardian_session.map(|session| session.model.to_string()), + guardian_reasoning_effort: guardian_session + .and_then(|session| session.reasoning_effort.map(str::to_string)), + had_prior_review_context: guardian_session.map(|session| session.had_prior_review_context), + review_timeout_ms: response.review_timeout_ms, + tool_call_count: response.tool_call_count, + time_to_first_token_ms: response.time_to_first_token_ms, + completion_latency_ms: response.completion_latency_ms, + started_at: u64::try_from(observation.started_at).unwrap_or_default(), + completed_at: Some(u64::try_from(observation.ended_at).unwrap_or_default()), + input_tokens: token_usage.map(|token_usage| token_usage.input_tokens), + cached_input_tokens: token_usage.map(|token_usage| token_usage.cached_input_tokens), + output_tokens: token_usage.map(|token_usage| token_usage.output_tokens), + reasoning_output_tokens: token_usage.map(|token_usage| token_usage.reasoning_output_tokens), + total_tokens: token_usage.map(|token_usage| token_usage.total_tokens), + }) +} + +fn reviewed_action(action: events::ReviewedAction<'_>) -> AnalyticsGuardianReviewedAction { + match action { + events::ReviewedAction::Shell { + command, + command_display, + cwd, + sandbox_permissions, + additional_permissions, + justification, + } => AnalyticsGuardianReviewedAction::Shell { + command: command.to_vec(), + command_display: command_display.to_string(), + cwd: cwd.to_string(), + sandbox_permissions: sandbox_permissions_for_review(sandbox_permissions), + additional_permissions: additional_permissions.and_then(permission_profile_for_review), + justification: justification.map(str::to_string), + }, + events::ReviewedAction::UnifiedExec { + command, + command_display, + cwd, + sandbox_permissions, + additional_permissions, + justification, + tty, + } => AnalyticsGuardianReviewedAction::UnifiedExec { + command: command.to_vec(), + command_display: command_display.to_string(), + cwd: cwd.to_string(), + sandbox_permissions: sandbox_permissions_for_review(sandbox_permissions), + additional_permissions: additional_permissions.and_then(permission_profile_for_review), + justification: justification.map(str::to_string), + tty, + }, + events::ReviewedAction::ProcessExec { + source, + program, + argv, + cwd, + additional_permissions, + } => AnalyticsGuardianReviewedAction::Execve { + source: match source { + events::ReviewCommandSource::Shell => AnalyticsGuardianCommandSource::Shell, + events::ReviewCommandSource::UnifiedExec => { + AnalyticsGuardianCommandSource::UnifiedExec + } + }, + program: program.to_string(), + argv: argv.to_vec(), + cwd: cwd.to_string(), + additional_permissions: additional_permissions.and_then(permission_profile_for_review), + }, + events::ReviewedAction::ApplyPatch { cwd, files } => { + AnalyticsGuardianReviewedAction::ApplyPatch { + cwd: cwd.to_string(), + files: files.to_vec(), + } + } + events::ReviewedAction::NetworkAccess { + target, + host, + protocol, + port, + } => AnalyticsGuardianReviewedAction::NetworkAccess { + target: target.to_string(), + host: host.to_string(), + protocol: match protocol { + events::ReviewNetworkApprovalProtocol::Http => NetworkApprovalProtocol::Http, + events::ReviewNetworkApprovalProtocol::Https => NetworkApprovalProtocol::Https, + events::ReviewNetworkApprovalProtocol::Socks5Tcp => { + NetworkApprovalProtocol::Socks5Tcp + } + events::ReviewNetworkApprovalProtocol::Socks5Udp => { + NetworkApprovalProtocol::Socks5Udp + } + }, + port, + }, + events::ReviewedAction::McpToolCall { + server, + tool_name, + connector_id, + connector_name, + tool_title, + } => AnalyticsGuardianReviewedAction::McpToolCall { + server: server.to_string(), + tool_name: tool_name.to_string(), + connector_id: connector_id.map(str::to_string), + connector_name: connector_name.map(str::to_string), + tool_title: tool_title.map(str::to_string), + }, + } +} + +fn sandbox_permissions_for_review( + sandbox_permissions: events::ReviewSandboxPermissions, +) -> SandboxPermissions { + match sandbox_permissions { + events::ReviewSandboxPermissions::UseDefault => SandboxPermissions::UseDefault, + events::ReviewSandboxPermissions::RequireEscalated => SandboxPermissions::RequireEscalated, + events::ReviewSandboxPermissions::WithAdditionalPermissions => { + SandboxPermissions::WithAdditionalPermissions + } + } +} + +fn permission_profile_for_review( + profile: events::ReviewPermissionProfile<'_>, +) -> Option { + // Keep observability independent of codex-protocol types. The protocol + // serde shape is the compatibility boundary for this nested payload. + let value = serde_json::to_value(profile).ok()?; + serde_json::from_value(value).ok() +} + +fn review_failure_reason( + failure_reason: events::ReviewFailureReason, +) -> AnalyticsGuardianReviewFailureReason { + match failure_reason { + events::ReviewFailureReason::Timeout => AnalyticsGuardianReviewFailureReason::Timeout, + events::ReviewFailureReason::Cancelled => AnalyticsGuardianReviewFailureReason::Cancelled, + events::ReviewFailureReason::PromptBuildError => { + AnalyticsGuardianReviewFailureReason::PromptBuildError + } + events::ReviewFailureReason::SessionError => { + AnalyticsGuardianReviewFailureReason::SessionError + } + events::ReviewFailureReason::ParseError => AnalyticsGuardianReviewFailureReason::ParseError, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use pretty_assertions::assert_eq; + use serde_json::json; + + fn projected_reviewed_action(action: events::ReviewedAction<'_>) -> serde_json::Value { + let event = + legacy_guardian_review_event(review_completed_with_action(action)).expect("project"); + serde_json::to_value(event.reviewed_action).expect("serialize reviewed action") + } + + fn review_completed_with_action( + reviewed_action: events::ReviewedAction<'_>, + ) -> events::ReviewCompleted<'_> { + events::ReviewCompleted { + thread_id: "thread-1", + turn_id: "turn-1", + review_id: "review-1", + target_item_id: "item-1", + retry_reason: None, + request_source: events::ReviewRequestSource::MainTurn, + reviewed_action, + reviewed_action_truncated: false, + response: events::ReviewResponse::Guardian(events::GuardianReviewResponse { + decision: events::ReviewDecision::Approved, + terminal_status: events::ReviewTerminalStatus::Approved, + risk_level: None, + user_authorization: None, + outcome: None, + rationale: None, + session: None, + review_timeout_ms: 30_000, + tool_call_count: 0, + time_to_first_token_ms: None, + completion_latency_ms: None, + token_usage: None, + }), + started_at: 1, + ended_at: 2, + } + } + + #[test] + fn legacy_guardian_projection_ignores_user_review_responses() { + let observation = events::ReviewCompleted { + thread_id: "thread-1", + turn_id: "turn-1", + review_id: "review-1", + target_item_id: "item-1", + retry_reason: None, + request_source: events::ReviewRequestSource::MainTurn, + reviewed_action: events::ReviewedAction::ApplyPatch { + cwd: "/repo", + files: &[], + }, + reviewed_action_truncated: false, + response: events::ReviewResponse::User(events::UserReviewResponse { + decision: events::ReviewDecision::Approved, + }), + started_at: 1, + ended_at: 2, + }; + + assert!(legacy_guardian_review_event(observation).is_none()); + } + + #[test] + fn projects_shell_reviewed_action_with_permission_profile() { + let command = vec!["git".to_string(), "status".to_string()]; + let read_paths = vec!["/repo".to_string()]; + let write_paths = vec!["/repo/tmp".to_string()]; + + let action = events::ReviewedAction::Shell { + command: &command, + command_display: "git status", + cwd: "/repo", + sandbox_permissions: events::ReviewSandboxPermissions::WithAdditionalPermissions, + additional_permissions: Some(events::ReviewPermissionProfile { + network: Some(events::ReviewNetworkPermissions { + enabled: Some(true), + }), + file_system: Some(events::ReviewFileSystemPermissions { + read: Some(&read_paths), + write: Some(&write_paths), + }), + }), + justification: Some("inspect repository state"), + }; + + assert_eq!( + projected_reviewed_action(action), + json!({ + "type": "shell", + "command": ["git", "status"], + "command_display": "git status", + "cwd": "/repo", + "sandbox_permissions": "with_additional_permissions", + "additional_permissions": { + "network": { + "enabled": true + }, + "file_system": { + "read": ["/repo"], + "write": ["/repo/tmp"] + } + }, + "justification": "inspect repository state" + }) + ); + } + + #[test] + fn projects_remaining_reviewed_action_variants() { + let unified_command = vec!["cargo".to_string(), "test".to_string()]; + assert_eq!( + projected_reviewed_action(events::ReviewedAction::UnifiedExec { + command: &unified_command, + command_display: "cargo test", + cwd: "/repo", + sandbox_permissions: events::ReviewSandboxPermissions::RequireEscalated, + additional_permissions: None, + justification: None, + tty: true, + }), + json!({ + "type": "unified_exec", + "command": ["cargo", "test"], + "command_display": "cargo test", + "cwd": "/repo", + "sandbox_permissions": "require_escalated", + "additional_permissions": null, + "justification": null, + "tty": true + }) + ); + + let argv = vec!["git".to_string(), "diff".to_string()]; + assert_eq!( + projected_reviewed_action(events::ReviewedAction::ProcessExec { + source: events::ReviewCommandSource::UnifiedExec, + program: "git", + argv: &argv, + cwd: "/repo", + additional_permissions: None, + }), + json!({ + "type": "execve", + "source": "unified_exec", + "program": "git", + "argv": ["git", "diff"], + "cwd": "/repo", + "additional_permissions": null + }) + ); + + let files = vec!["src/lib.rs".to_string(), "Cargo.toml".to_string()]; + assert_eq!( + projected_reviewed_action(events::ReviewedAction::ApplyPatch { + cwd: "/repo", + files: &files, + }), + json!({ + "type": "apply_patch", + "cwd": "/repo", + "files": ["src/lib.rs", "Cargo.toml"] + }) + ); + + assert_eq!( + projected_reviewed_action(events::ReviewedAction::NetworkAccess { + target: "https://example.com", + host: "example.com", + protocol: events::ReviewNetworkApprovalProtocol::Https, + port: 443, + }), + json!({ + "type": "network_access", + "target": "https://example.com", + "host": "example.com", + "protocol": "https", + "port": 443 + }) + ); + + assert_eq!( + projected_reviewed_action(events::ReviewedAction::McpToolCall { + server: "drive", + tool_name: "search", + connector_id: Some("drive-connector"), + connector_name: Some("Drive"), + tool_title: Some("Search Drive"), + }), + json!({ + "type": "mcp_tool_call", + "server": "drive", + "tool_name": "search", + "connector_id": "drive-connector", + "connector_name": "Drive", + "tool_title": "Search Drive" + }) + ); + } +} diff --git a/codex-rs/observability/src/events.rs b/codex-rs/observability/src/events.rs index a3aef0cd8b..f728d05e3c 100644 --- a/codex-rs/observability/src/events.rs +++ b/codex-rs/observability/src/events.rs @@ -4,10 +4,12 @@ use crate::Observation; use serde::Serialize; mod compaction; +mod review; mod thread; mod turn; pub use compaction::*; +pub use review::*; pub use thread::*; pub use turn::*; diff --git a/codex-rs/observability/src/events/review.rs b/codex-rs/observability/src/events/review.rs new file mode 100644 index 0000000000..2398c46cb8 --- /dev/null +++ b/codex-rs/observability/src/events/review.rs @@ -0,0 +1,272 @@ +//! Review observation event definitions. + +use crate::Observation; +use serde::Serialize; + +/// Final decision returned by a review. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReviewDecision { + Approved, + Denied, + Aborted, +} + +/// Terminal state of a review. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReviewTerminalStatus { + Approved, + Denied, + Aborted { + failure_reason: Option, + }, + TimedOut { + failure_reason: Option, + }, + FailedClosed { + failure_reason: Option, + }, +} + +/// Stable failure category for review terminals. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReviewFailureReason { + Timeout, + Cancelled, + PromptBuildError, + SessionError, + ParseError, +} + +/// Source of the request sent for review. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReviewRequestSource { + MainTurn, + DelegatedSubagent, +} + +/// Per-command sandbox override requested by the reviewed action. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReviewSandboxPermissions { + UseDefault, + RequireEscalated, + WithAdditionalPermissions, +} + +/// Additional filesystem permissions requested for a reviewed action. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +pub struct ReviewFileSystemPermissions<'a> { + pub read: Option<&'a [String]>, + pub write: Option<&'a [String]>, +} + +/// Additional network permissions requested for a reviewed action. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +pub struct ReviewNetworkPermissions { + pub enabled: Option, +} + +/// Additional permissions requested for a reviewed action. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +pub struct ReviewPermissionProfile<'a> { + pub network: Option, + pub file_system: Option>, +} + +/// Source tool that produced an exec-style reviewed action. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReviewCommandSource { + Shell, + UnifiedExec, +} + +/// Network protocol involved in a reviewed network access request. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReviewNetworkApprovalProtocol { + Http, + Https, + Socks5Tcp, + Socks5Udp, +} + +/// Action that was evaluated by a review. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum ReviewedAction<'a> { + Shell { + command: &'a [String], + command_display: &'a str, + cwd: &'a str, + sandbox_permissions: ReviewSandboxPermissions, + additional_permissions: Option>, + justification: Option<&'a str>, + }, + UnifiedExec { + command: &'a [String], + command_display: &'a str, + cwd: &'a str, + sandbox_permissions: ReviewSandboxPermissions, + additional_permissions: Option>, + justification: Option<&'a str>, + tty: bool, + }, + ProcessExec { + source: ReviewCommandSource, + program: &'a str, + argv: &'a [String], + cwd: &'a str, + additional_permissions: Option>, + }, + ApplyPatch { + cwd: &'a str, + files: &'a [String], + }, + NetworkAccess { + target: &'a str, + host: &'a str, + protocol: ReviewNetworkApprovalProtocol, + port: u16, + }, + McpToolCall { + server: &'a str, + tool_name: &'a str, + connector_id: Option<&'a str>, + connector_name: Option<&'a str>, + tool_title: Option<&'a str>, + }, +} + +/// Risk level assigned by an automated reviewer. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReviewRiskLevel { + Low, + Medium, + High, + Critical, +} + +/// User authorization level observed by an automated reviewer. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReviewUserAuthorization { + Unknown, + Low, + Medium, + High, +} + +/// Policy outcome recommended by an automated reviewer. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReviewOutcome { + Allow, + Deny, +} + +/// How guardian review obtained a model session. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum GuardianReviewSessionKind { + TrunkNew, + TrunkReused, + EphemeralForked, +} + +/// Guardian model session used to perform a review. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +pub struct GuardianReviewSession<'a> { + pub guardian_thread_id: &'a str, + pub session_kind: GuardianReviewSessionKind, + pub model: &'a str, + /// Absent when the selected model/provider has no explicit effort setting. + pub reasoning_effort: Option<&'a str>, + pub had_prior_review_context: bool, +} + +/// Response produced by a user reviewer. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +pub struct UserReviewResponse { + pub decision: ReviewDecision, +} + +/// Response produced by guardian review. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +pub struct GuardianReviewResponse<'a> { + pub decision: ReviewDecision, + /// Terminal runtime state; non-success states carry their stable failure category. + pub terminal_status: ReviewTerminalStatus, + /// Absent when review stops before the model reports a risk classification. + pub risk_level: Option, + /// Absent when review stops before user authorization is assessed. + pub user_authorization: Option, + /// Absent when review stops before a policy outcome is produced. + pub outcome: Option, + /// Model-authored rationale text returned by guardian review. + pub rationale: Option<&'a str>, + /// Absent when review fails before a guardian model session is created or reused. + pub session: Option>, + pub review_timeout_ms: u64, + pub tool_call_count: u64, + /// Absent when the guardian session did not stream model tokens. + pub time_to_first_token_ms: Option, + /// Absent when review ended before a model completion was received. + pub completion_latency_ms: Option, + /// Absent when the guardian model did not report token accounting. + pub token_usage: Option, +} + +/// Reviewer response that completed a review. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(tag = "reviewer", rename_all = "snake_case")] +pub enum ReviewResponse<'a> { + User(UserReviewResponse), + Guardian(GuardianReviewResponse<'a>), +} + +/// Observation emitted when review of an action reaches a terminal state. +#[derive(Observation)] +#[observation(name = "review.completed", crate = "crate", uses = ["analytics"])] +pub struct ReviewCompleted<'a> { + #[obs(level = "basic", class = "identifier")] + pub thread_id: &'a str, + + #[obs(level = "basic", class = "identifier")] + pub turn_id: &'a str, + + #[obs(level = "basic", class = "identifier")] + pub review_id: &'a str, + + #[obs(level = "basic", class = "identifier")] + pub target_item_id: &'a str, + + /// Absent on the first review attempt; present when this review retries an earlier attempt. + #[obs(level = "basic", class = "operational")] + pub retry_reason: Option<&'a str>, + + #[obs(level = "basic", class = "operational")] + pub request_source: ReviewRequestSource, + + /// The reviewed action may contain command text, paths, or tool names. + #[obs(level = "basic", class = "content")] + pub reviewed_action: ReviewedAction<'a>, + + #[obs(level = "basic", class = "operational")] + pub reviewed_action_truncated: bool, + + /// Contains the reviewer-specific terminal result. + #[obs(level = "basic", class = "content")] + pub response: ReviewResponse<'a>, + + #[obs(level = "basic", class = "operational")] + pub started_at: i64, + + #[obs(level = "basic", class = "operational")] + pub ended_at: i64, +} diff --git a/docs/observability-event-stream-design.md b/docs/observability-event-stream-design.md index 2ca1bfd5b1..08f132dcaf 100644 --- a/docs/observability-event-stream-design.md +++ b/docs/observability-event-stream-design.md @@ -223,9 +223,9 @@ Each field should carry at least: - **Data class**: `identifier`, `operational`, `environment`, `content`, or `secret_risk`. -Use markers express intent. Detail level and data class are guardrails. A sink -must first select only fields explicitly marked for that sink, then enforce its -detail/class policy before serializing values. +Use markers are the allow marker: a sink should consume only fields explicitly +marked for that sink. Detail level and data class are descriptive metadata for +review, auditing, and sinks that need coarse verbosity or redaction modes. Event structs may define default use markers when most fields feed the same projection; field-level use markers override that default for mixed events. @@ -234,11 +234,10 @@ remote export, and a trace-level field can be trace-level because it is large rather than sensitive. Data class is also not enough for selection: analytics must not consume every basic operational field just because it would be safe. -Expected sink policies: +Expected sink behavior: -- Analytics selects only fields marked `analytics`, then allows basic - identifiers/operational fields plus selected environment fields; it denies - content and secret-risk fields even if they are marked accidentally. +- Analytics consumes fields marked `analytics`. Content-bearing analytics fields + should be rare and called out on the field or projection that exports them. - Rollout trace allows rich local fields, with explicit redaction rules for secret-risk material. - OTEL logs preserve today's log export behavior, including account/email only @@ -276,7 +275,7 @@ Initial workflow coverage should be chosen by conformance need: | Tools | `tool_call.started`, `tool_call.approval_resolved`, `tool_call.ended` | OTEL, rollout | | Compaction | `compaction.started`, `compaction.installed`, `compaction.ended` | analytics, rollout | | Agents | `agent.task_sent`, `agent.message_sent`, `agent.result_delivered`, `agent.closed` | rollout, analytics subset | -| Product features | `app.mentioned`, `app.used`, `hook.run_completed`, `plugin.used`, `plugin.state_changed`, `guardian.review_completed` | analytics | +| Product features | `app.mentioned`, `app.used`, `hook.run_completed`, `plugin.used`, `plugin.state_changed`, `review.completed` | analytics | | Transport/auth | `transport.api_request_completed`, `transport.websocket_request_completed`, `auth.recovery_step_completed` | OTEL | This table is not a complete schema. Each event still needs a typed Rust @@ -293,9 +292,14 @@ output schemas. | --- | --- | | `thread.started` | `codex_thread_initialized` | | `turn.started` with resolved config, `turn.ended` with token usage | `codex_turn_event` | -| `turn.steer_resolved` | `codex_turn_steer_event` | +| `turn.steer` | `codex_turn_steer_event` | | compaction lifecycle observations | `codex_compaction_event` | -| skill/app/plugin/guardian observations | existing feature events | +| skill/app/plugin/review observations | existing feature events | + +`review.completed` is generic, but the current analytics projection only emits +the legacy guardian event for guardian reviewer responses. User reviewer +responses remain represented in the observation stream until a consumer needs +an analytics output for them. The analytics crate should keep the observation-to-legacy-schema translation in a private projection module. The reducer then stays responsible for ingestion @@ -354,7 +358,7 @@ tests. Recommended first scenarios: - failed or interrupted turn - accepted and rejected turn steering - compaction -- skill, app, plugin, and guardian events +- skill, app, plugin, and review events - subagent thread start Rollout trace can start with observation-to-trace unit tests plus one or two