From 07316174c6f62deb20278c53add2dbd9dcf6da70 Mon Sep 17 00:00:00 2001 From: jif Date: Thu, 22 Oct 2026 09:10:55 +0000 Subject: [PATCH] Forward Guardian reviewer events to parent raw-event subscribers (#56124) ## What changed Add the disabled-by-default `guardian_recorded_items` feature to forward recorded reviewer items and interruptions through the parent's raw-event subscription, preserving reviewer thread and turn IDs without modifying the parent's history or completing its turn. Emit reviewer items as `rawResponseItem/completed` notifications and interruptions as interrupted `turn/completed` notifications. Mark opted-in reviewer requests with `guardian_recorded_items` metadata. ## Testing Extend Guardian integration tests to cover reviewer identity, recorded-item metadata replay, concurrent reviewer interruptions, independent parent completion, and behavior without the feature enabled. GitOrigin-RevId: beb4dbf6c6f2a800a2fbf3cfd11ae174a1382386 --- .../app-server/src/bespoke_event_handling.rs | 40 +++++++ .../request_processors/thread_lifecycle.rs | 2 +- .../suite/v2/guardian_sync_session_tests.rs | 107 +++++++++++++++++- codex-rs/core/config.schema.json | 6 + .../core/src/guardian/review_session_setup.rs | 7 ++ codex-rs/core/src/responses_metadata.rs | 8 ++ codex-rs/core/src/session/mod.rs | 26 +++++ codex-rs/core/src/session/session.rs | 5 + codex-rs/core/src/session/tests.rs | 2 + codex-rs/core/src/session/turn.rs | 1 + codex-rs/features/src/lib.rs | 8 ++ codex-rs/protocol/src/protocol.rs | 20 ++++ codex-rs/rollout-trace/src/protocol_event.rs | 2 + codex-rs/rollout/src/policy.rs | 1 + 14 files changed, 228 insertions(+), 7 deletions(-) diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 11f0555325..d0d363e401 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -104,6 +104,7 @@ use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ExecApprovalRequestEvent; use codex_protocol::protocol::Op; +use codex_protocol::protocol::PrivateSessionEventPayload; use codex_protocol::protocol::RealtimeEvent; use codex_protocol::protocol::ReviewDecision; use codex_protocol::protocol::SubAgentActivityKind; @@ -1174,6 +1175,45 @@ pub(crate) async fn apply_bespoke_event_handling( .send_server_notification(ServerNotification::HookCompleted(notification)) .await; } + EventMsg::PrivateSessionEvent(private) => match private.event { + PrivateSessionEventPayload::RecordedItem(mut recorded) => { + recorded.item.clear_executed_tool_calls(); + let mut notification = ServerNotification::RawResponseItemCompleted( + RawResponseItemCompletedNotification { + thread_id: private.thread_id.to_string(), + turn_id: event_turn_id, + item: recorded.item, + }, + ); + if conversation.enabled(Feature::OmitAppServerNotificationMedia) { + notification = without_notification_media(notification); + } + outgoing.send_server_notification(notification).await; + } + PrivateSessionEventPayload::TurnAborted(aborted) => { + // Notify observers without completing or clearing the parent's turn. + emit_turn_completed_with_status( + private.thread_id, + event_turn_id, + TurnCompletionMetadata { + status: TurnStatus::Interrupted, + root_turn_id: aborted.root_turn_id, + error: aborted.error.map(|error| TurnError { + message: error.message, + codex_error_info: error.codex_error_info.map(Into::into), + misalignment: error.misalignment.map(Into::into), + additional_details: None, + }), + last_agent_message: None, + started_at: aborted.started_at, + completed_at: aborted.completed_at, + duration_ms: aborted.duration_ms, + }, + &outgoing, + ) + .await; + } + }, EventMsg::RawResponseItem(mut raw_response_item_event) => { // Keep warehouse metadata out of app-server notifications. raw_response_item_event.item.clear_executed_tool_calls(); diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index ab13ac6b0d..f864f0f341 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -349,7 +349,7 @@ pub(super) async fn ensure_listener_task_running( }; if matches!( &event.msg, - EventMsg::RawResponseItem(_) | EventMsg::RawResponseCompleted(_) + EventMsg::RawResponseItem(_) | EventMsg::RawResponseCompleted(_) | EventMsg::PrivateSessionEvent(_) ) && !raw_events_enabled { continue; diff --git a/codex-rs/app-server/tests/suite/v2/guardian_sync_session_tests.rs b/codex-rs/app-server/tests/suite/v2/guardian_sync_session_tests.rs index 0830c48079..7d690e6df4 100644 --- a/codex-rs/app-server/tests/suite/v2/guardian_sync_session_tests.rs +++ b/codex-rs/app-server/tests/suite/v2/guardian_sync_session_tests.rs @@ -5,6 +5,7 @@ use app_test_support::create_final_assistant_message_sse_response as assistant_r use codex_app_server_protocol::CodexErrorInfo; use codex_app_server_protocol::GuardianWarningNotification; use codex_app_server_protocol::JSONRPCErrorError; +use codex_app_server_protocol::RawResponseItemCompletedNotification; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ReviewDelivery; use codex_app_server_protocol::ReviewStartParams; @@ -191,6 +192,9 @@ async fn managed_reviewer_refreshes_global_instructions_before_reuse() -> Result let codex_home = TempDir::new()?; let instructions_path = codex_home.path().join("AGENTS.md"); sync_config(server.uri()) + .with_provider_name("OpenAI") + .enable_feature(Feature::ContentItemKinds) + .enable_feature(Feature::GuardianRecordedItems) .with_extra_config(&format!( "[mcp_servers.{TEST_SERVER_NAME}]\nurl = \"{mcp_url}/mcp\"\ndefault_tools_approval_mode = \"prompt\"" )) @@ -202,8 +206,15 @@ async fn managed_reviewer_refreshes_global_instructions_before_reuse() -> Result let old_instructions = "Keep this project private."; let new_instructions = "Ask before sharing this project's source code."; std::fs::write(&instructions_path, old_instructions)?; - let parent = app.start_thread(ThreadStartParams::default()).await?.thread; + let parent = app + .start_thread(ThreadStartParams { + experimental_raw_events: true, + ..Default::default() + }) + .await? + .thread; let mut reviewer_ids = Vec::new(); + let mut previous_recorded: Option<(String, Value)> = None; for (index, instructions) in [ old_instructions, old_instructions, @@ -234,6 +245,44 @@ async fn managed_reviewer_refreshes_global_instructions_before_reuse() -> Result assert_eq!(requests.len(), (index + 1) * 3); let review: Value = serde_json::from_slice(&requests[index * 3 + 1])?; assert!(review["input"].to_string().contains(instructions)); + let turn_metadata: Value = serde_json::from_str( + review["client_metadata"]["x-codex-turn-metadata"] + .as_str() + .expect("turn metadata"), + )?; + assert_eq!(turn_metadata["guardian_recorded_items"], true); + if let Some((thread_id, item)) = &previous_recorded + && review["client_metadata"]["thread_id"] == thread_id.as_str() + { + let replayed = review["input"] + .as_array() + .expect("review input") + .iter() + .find(|replayed| replayed["content"] == item["content"]) + .expect("previous Guardian reply is replayed"); + assert_eq!( + replayed["internal_chat_message_metadata_passthrough"], + item["internal_chat_message_metadata_passthrough"], + ); + } + // Private reviewer events arrive on the parent's subscription with the + // reviewer's identity and the metadata that its next request replays. + let (recorded, item) = loop { + let recorded: RawResponseItemCompletedNotification = + app.read_notification("rawResponseItem/completed").await?; + let item = serde_json::to_value(&recorded.item)?; + if recorded.thread_id != parent.id + && item["role"] == "assistant" + && item["content"][0]["text"] == ALLOW + { + break (recorded, item); + } + }; + assert_eq!( + json!([recorded.thread_id, recorded.turn_id]), + json!([turn_metadata["thread_id"], turn_metadata["turn_id"]]), + ); + previous_recorded = Some((recorded.thread_id, item)); reviewer_ids.push( review["client_metadata"]["thread_id"] .as_str() @@ -250,11 +299,13 @@ async fn managed_reviewer_refreshes_global_instructions_before_reuse() -> Result Ok(()) } -#[test_case(ReviewEnding::Complete; "completed reviews")] -#[test_case(ReviewEnding::Interrupt; "cancelled concurrent reviews")] +#[test_case(ReviewEnding::Complete, true; "completed reviews")] +#[test_case(ReviewEnding::Interrupt, true; "cancelled concurrent reviews")] +#[test_case(ReviewEnding::Interrupt, false; "raw events without reviewer opt-in")] #[tokio::test] async fn managed_reviewers_reuse_fork_and_resume_after_parent_shutdown( ending: ReviewEnding, + recorded_items: bool, ) -> Result<()> { skip_if_no_network!(Ok(())); let (continue_parent, parent_gate) = oneshot::channel(); @@ -291,6 +342,7 @@ async fn managed_reviewers_reuse_fork_and_resume_after_parent_shutdown( "Keep this project private.", )?; sync_config(server.uri()) + .with_features(&[(Feature::GuardianRecordedItems, recorded_items)].into()) .with_extra_config(&format!( "[mcp_servers.{TEST_SERVER_NAME}]\nurl = \"{mcp_url}/mcp\"\ndefault_tools_approval_mode = \"prompt\"" )) @@ -299,7 +351,13 @@ async fn managed_reviewers_reuse_fork_and_resume_after_parent_shutdown( .with_codex_home(codex_home.path()) .build_initialized_with_timeout(TIMEOUT) .await?; - let parent = app.start_thread(ThreadStartParams::default()).await?.thread; + let parent = app + .start_thread(ThreadStartParams { + experimental_raw_events: true, + ..Default::default() + }) + .await? + .thread; let started: TurnStartResponse = app .request(|request_id| ClientRequest::TurnStart { request_id, @@ -413,8 +471,45 @@ async fn managed_reviewers_reuse_fork_and_resume_after_parent_shutdown( finish_first.send(()).expect("finish first review"); finish_second.send(()).expect("finish second review"); } - let completed: TurnCompletedNotification = - timeout(TIMEOUT, app.read_notification("turn/completed")).await??; + let mut parent_completed = None; + let mut interrupted_reviewers = std::collections::BTreeSet::new(); + let expected_interruptions = match ending { + ReviewEnding::Interrupt if recorded_items => 2, + ReviewEnding::Complete | ReviewEnding::Interrupt => 0, + }; + while parent_completed.is_none() || interrupted_reviewers.len() < expected_interruptions { + let completed: TurnCompletedNotification = + timeout(TIMEOUT, app.read_notification("turn/completed")).await??; + if completed.thread_id == parent.id { + parent_completed = Some(completed); + } else { + assert_eq!(completed.turn.status, TurnStatus::Interrupted); + interrupted_reviewers.insert((completed.thread_id, completed.turn.id)); + } + } + assert_eq!(interrupted_reviewers.len(), expected_interruptions); + if matches!(ending, ReviewEnding::Interrupt) && recorded_items { + let mut expected = std::collections::BTreeSet::new(); + for review in [&first, &second] { + let metadata: Value = serde_json::from_str( + review["client_metadata"]["x-codex-turn-metadata"] + .as_str() + .expect("turn metadata"), + )?; + expected.insert(( + metadata["thread_id"] + .as_str() + .expect("reviewer thread ID") + .to_owned(), + metadata["turn_id"] + .as_str() + .expect("reviewer turn ID") + .to_owned(), + )); + } + assert_eq!(interrupted_reviewers, expected); + } + let completed = parent_completed.expect("parent completes independently of its reviewers"); let expected_status = match ending { ReviewEnding::Complete => TurnStatus::Completed, ReviewEnding::Interrupt => TurnStatus::Interrupted, diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 2b2c978b69..cc0aa777c3 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -964,6 +964,9 @@ "guardian_node_repl_transcript_images": { "type": "boolean" }, + "guardian_recorded_items": { + "type": "boolean" + }, "guardian_reuse_parent_compaction": { "type": "boolean" }, @@ -7479,6 +7482,9 @@ "guardian_node_repl_transcript_images": { "type": "boolean" }, + "guardian_recorded_items": { + "type": "boolean" + }, "guardian_reuse_parent_compaction": { "type": "boolean" }, diff --git a/codex-rs/core/src/guardian/review_session_setup.rs b/codex-rs/core/src/guardian/review_session_setup.rs index 7131d38025..39052af7cd 100644 --- a/codex-rs/core/src/guardian/review_session_setup.rs +++ b/codex-rs/core/src/guardian/review_session_setup.rs @@ -2,11 +2,13 @@ //! Checkpoint recovery allows one fresh attempt under the original review deadline. //! Both attempts share captured context and use separate recovery flags. //! Checkpoints are captured before final authorization validation and published without yielding. +//! Recorded reviewer items and interruptions are forwarded to the parent's raw-event subscription. use std::sync::atomic::AtomicBool; use super::*; use crate::guardian::input_budget::CheckpointRecovery; +use codex_features::Feature; use codex_guardian_reviewer::ReviewerPool; use codex_guardian_reviewer::ReviewerRequest; use codex_protocol::protocol::TurnEnvironmentSelection; @@ -185,6 +187,11 @@ impl PreparedGuardianContext { cancellation: CancellationToken, ) -> GuardianReviewSession { let session = Arc::clone(&thread.session); + if self.parent.enabled(Feature::GuardianRecordedItems) { + session + .parent_observer_event_tx + .get_or_init(|| self.parent.get_tx_event()); + } let io = SessionIo { tx_sub: thread.io.tx_sub.clone(), rx_event: thread.io.rx_event.clone(), diff --git a/codex-rs/core/src/responses_metadata.rs b/codex-rs/core/src/responses_metadata.rs index bcc2f9e945..2d82bb719e 100644 --- a/codex-rs/core/src/responses_metadata.rs +++ b/codex-rs/core/src/responses_metadata.rs @@ -67,6 +67,7 @@ pub(crate) const WORKSPACES_KEY: &str = "workspaces"; // when submitting a turn, but they must not override fields owned by core. const RESERVED_METADATA_KEYS: &[&str] = &[ "guardian_credits_requested", + "guardian_recorded_items", MCP_ATTRIBUTION_CLIENT_METADATA_KEY, INSTALLATION_ID_KEY, X_CODEX_INSTALLATION_ID_HEADER, @@ -108,6 +109,7 @@ const RESERVED_METADATA_KEYS: &[&str] = &[ // These keys were previously valid user configuration. Accept existing configs while filtering // their values before constructing Core-owned request metadata. const BACKWARD_COMPATIBLE_RESERVED_METADATA_KEYS: &[&str] = &[ + "guardian_recorded_items", WINDOW_NUMBER_KEY, FORKED_FROM_ORDINAL_EXCLUSIVE_KEY, ANALYTICS_ENABLED_KEY, @@ -232,6 +234,8 @@ pub(crate) enum TurnToolSource { /// truth. #[derive(Clone, Debug)] pub struct CodexResponsesMetadata { + /// The private reviewer forwards recorded items through its parent's raw-event subscription. + pub(crate) guardian_recorded_items: bool, /// Guardian parent reference; projected only onto a Guardian request. pub(crate) parent_response_id: Option, pub(crate) installation_id: String, @@ -280,6 +284,7 @@ impl CodexResponsesMetadata { ) -> Self { Self { parent_response_id: None, + guardian_recorded_items: false, installation_id, session_id, thread_id, @@ -420,6 +425,7 @@ impl CodexResponsesMetadata { let has_request_identity = request_kind.is_some_and(CodexResponsesRequestKind::has_thread_identity); CodexTurnMetadataPayload { + guardian_recorded_items: self.guardian_recorded_items.then_some(true), installation_id: has_request_identity.then_some(self.installation_id.as_str()), session_id: has_thread_identity.then_some(self.session_id.as_str()), thread_id: has_thread_identity.then_some(self.thread_id.as_str()), @@ -546,6 +552,8 @@ fn non_empty_workspaces( #[derive(Serialize)] struct CodexTurnMetadataPayload<'a> { + #[serde(default, skip_serializing_if = "Option::is_none")] + guardian_recorded_items: Option, #[serde(default, skip_serializing_if = "Option::is_none")] installation_id: Option<&'a str>, #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 96cfaa1970..c9332bdf64 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -149,6 +149,8 @@ use codex_protocol::protocol::ItemCompletedEvent; use codex_protocol::protocol::ItemStartedEvent; use codex_protocol::protocol::MULTI_AGENT_MODE_OPEN_TAG; use codex_protocol::protocol::MultiAgentVersion; +use codex_protocol::protocol::PrivateSessionEvent; +use codex_protocol::protocol::PrivateSessionEventPayload; use codex_protocol::protocol::RawResponseItemEvent; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; @@ -2741,6 +2743,30 @@ impl Session { } async fn deliver_event_raw(&self, event: Event) { + if let Some(parent_tx) = self.parent_observer_event_tx.get() { + let forwarded = match &event.msg { + EventMsg::RawResponseItem(item) => { + Some(PrivateSessionEventPayload::RecordedItem(item.clone())) + } + EventMsg::TurnAborted(aborted) => { + Some(PrivateSessionEventPayload::TurnAborted(aborted.clone())) + } + _ => None, + }; + if let Some(forwarded) = forwarded { + // Queue observer events before the private consumer can finish the review. + // They bypass the parent's history and retain the reviewer's identity. + let _ = parent_tx + .send(Event { + id: event.id.clone(), + msg: EventMsg::PrivateSessionEvent(PrivateSessionEvent { + thread_id: self.thread_id, + event: forwarded, + }), + }) + .await; + } + } // Record the last known agent status. if let Some(status) = agent_status_from_event(&event.msg) { self.agent_status.send_replace(status); diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index e3bcb2befb..3431138b77 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -61,6 +61,9 @@ pub(crate) struct Session { pub(crate) thread_id: ThreadId, pub(crate) installation_id: String, pub(super) tx_event: Sender, + /// Installed by Guardian before its first review; forwards items and cancellation without + /// transferring ownership of the private session's event receiver. + pub(crate) parent_observer_event_tx: OnceLock>, pub(super) agent_status: watch::Sender, pub(super) state: Mutex, /// Orders accepted settings commits and their persisted events with compaction checkpoints. @@ -717,6 +720,7 @@ impl Session { request_kind, ); ExecutionMetadata::from_settings(&step_context.settings).apply_to(&mut responses_metadata); + responses_metadata.guardian_recorded_items = self.parent_observer_event_tx.get().is_some(); responses_metadata.tool_namespaces_info = if step_context .turn .config @@ -1855,6 +1859,7 @@ impl Session { thread_id, installation_id, tx_event: tx_event.clone(), + parent_observer_event_tx: OnceLock::new(), agent_status, state: Mutex::new(state), thread_settings_persistence: Semaphore::new(/*permits*/ 1), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 7bd3a385bc..6ee2ccf4d1 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6989,6 +6989,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { thread_id, installation_id: "11111111-1111-4111-8111-111111111111".to_string(), tx_event, + parent_observer_event_tx: OnceLock::new(), agent_status: agent_status_tx, state: Mutex::new(state), thread_settings_persistence: Semaphore::new(/*permits*/ 1), @@ -9262,6 +9263,7 @@ where thread_id, installation_id: "11111111-1111-4111-8111-111111111111".to_string(), tx_event, + parent_observer_event_tx: OnceLock::new(), agent_status: agent_status_tx, state: Mutex::new(state), thread_settings_persistence: Semaphore::new(/*permits*/ 1), diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 32fdf7bd27..8531b098ba 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -2278,6 +2278,7 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option Option Option<&'static s | EventMsg::PlanUpdate(_) | EventMsg::EnteredReviewMode(_) | EventMsg::ExitedReviewMode(_) + | EventMsg::PrivateSessionEvent(_) | EventMsg::RawResponseItem(_) | EventMsg::RawResponseCompleted(_) | EventMsg::ItemStarted(_) diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 51997cd91b..93a1e3a5d8 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -183,6 +183,7 @@ fn persisted_event_msg( | EventMsg::ModelVerification(_) | EventMsg::TurnModerationMetadata(_) | EventMsg::AgentReasoningSectionBreak(_) + | EventMsg::PrivateSessionEvent(_) | EventMsg::RawResponseItem(_) | EventMsg::RawResponseCompleted(_) | EventMsg::SessionConfigured(_)