diff --git a/codex-rs/codex-api/src/common.rs b/codex-rs/codex-api/src/common.rs index 5adfe8db6a..c457285afd 100644 --- a/codex-rs/codex-api/src/common.rs +++ b/codex-rs/codex-api/src/common.rs @@ -96,7 +96,10 @@ pub struct MemorySummarizeOutput { #[derive(Debug)] pub enum ResponseEvent { - Created, + Created { + /// Runtime-only receipt for this exact server response. + guardian_ticket: Option, + }, SafetyBuffering(SafetyBuffering), OutputItemDone(ResponseItem), OutputItemAdded(ResponseItem), diff --git a/codex-rs/codex-api/src/endpoint/responses.rs b/codex-rs/codex-api/src/endpoint/responses.rs index 69a8c686d1..fa746f0e58 100644 --- a/codex-rs/codex-api/src/endpoint/responses.rs +++ b/codex-rs/codex-api/src/endpoint/responses.rs @@ -60,6 +60,8 @@ pub struct ResponsesOptions { pub extra_headers: HeaderMap, pub compression: Compression, pub turn_state: Option>>, + /// Runtime-only parent receipt, attached after inference tracing. + pub guardian_ticket: Option, } impl ResponsesClient { @@ -101,7 +103,7 @@ impl ResponsesClient { )] pub async fn stream_request( &self, - request: ResponsesApiRequest, + mut request: ResponsesApiRequest, options: ResponsesOptions, ) -> Result { let ResponsesOptions { @@ -111,10 +113,19 @@ impl ResponsesClient { extra_headers, compression, turn_state, + guardian_ticket, } = options; + crate::guardian_ticket::attach( + &mut request.client_metadata, + guardian_ticket.as_ref(), + self.endpoint, + ); - let body = EncodedJsonBody::encode(&request) + let mut body = EncodedJsonBody::encode(&request) .map_err(|e| ApiError::Stream(format!("failed to encode responses request: {e}")))?; + if guardian_ticket.is_some() { + body = body.without_body_logging(); + } let mut headers = extra_headers; if let Some(ref thread_id) = thread_id { diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index 95d41e0d91..0f928d1cff 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -234,9 +234,10 @@ impl ResponsesWebsocketConnection { )] pub async fn stream_request( &self, - request: ResponsesWsRequest<'_>, + mut request: ResponsesWsRequest<'_>, connection_reused: bool, turn_state: Option>>, + guardian_ticket: Option<&codex_protocol::guardian_ticket::GuardianTicket>, ) -> Result { let (tx_event, rx_event) = mpsc::channel::>(1600); @@ -270,6 +271,12 @@ impl ResponsesWebsocketConnection { warmup: ws_request.generate == Some(false), connection_reused, }; + let ResponsesWsRequest::ResponseCreate(ws_request) = &mut request; + crate::guardian_ticket::attach( + &mut ws_request.client_metadata, + guardian_ticket, + self.endpoint, + ); let request_text = serialize_websocket_request(&request)?; let current_span = Span::current(); diff --git a/codex-rs/codex-api/src/guardian_ticket.rs b/codex-rs/codex-api/src/guardian_ticket.rs new file mode 100644 index 0000000000..94a7660055 --- /dev/null +++ b/codex-rs/codex-api/src/guardian_ticket.rs @@ -0,0 +1,33 @@ +//! Adds opaque receipts only to Guardian requests at the transport boundary. + +use std::collections::HashMap; + +use crate::endpoint::responses::ResponsesEndpoint; + +use codex_protocol::guardian_ticket::GUARDIAN_TICKET_METADATA_KEY; +use codex_protocol::guardian_ticket::GuardianTicket; + +pub(crate) fn attach( + metadata: &mut Option>, + ticket: Option<&GuardianTicket>, + endpoint: ResponsesEndpoint, +) { + if let Some(metadata) = metadata.as_mut() { + metadata.remove(GUARDIAN_TICKET_METADATA_KEY); + if endpoint != ResponsesEndpoint::Responses { + metadata.remove("guardian_ticket_requested"); + } + } + if endpoint != ResponsesEndpoint::Responses + && let Some(ticket) = ticket + { + metadata.get_or_insert_with(HashMap::new).insert( + GUARDIAN_TICKET_METADATA_KEY.to_owned(), + ticket.as_str().to_owned(), + ); + } +} + +#[cfg(test)] +#[path = "guardian_ticket_tests.rs"] +mod tests; diff --git a/codex-rs/codex-api/src/guardian_ticket_tests.rs b/codex-rs/codex-api/src/guardian_ticket_tests.rs new file mode 100644 index 0000000000..ba7307f703 --- /dev/null +++ b/codex-rs/codex-api/src/guardian_ticket_tests.rs @@ -0,0 +1,60 @@ +use super::*; +use crate::common::ResponseEvent; +use crate::sse::responses::ResponsesStreamEvent; +use crate::sse::responses::process_responses_event; +use pretty_assertions::assert_eq; +use serde_json::json; + +#[test] +fn response_created_receipt_is_runtime_only_and_redacted() { + let raw = "a".repeat(43); + let event: ResponsesStreamEvent = serde_json::from_value(json!({ + "type": "response.created", + "response": {"id": "resp-1", "headers": {"x-codex-guardian-ticket": raw}} + })) + .unwrap(); + let ResponseEvent::Created { guardian_ticket } = + process_responses_event(event).unwrap().unwrap() + else { + panic!("expected response.created"); + }; + let ticket = guardian_ticket.unwrap(); + assert_eq!(ticket.as_str(), raw); + assert!(!format!("{ticket:?}").contains(&raw)); + let mut metadata = Some(HashMap::from([("turn_id".into(), "turn-1".into())])); + attach(&mut metadata, Some(&ticket), ResponsesEndpoint::Guardian); + assert_eq!( + metadata, + Some(HashMap::from([ + ("turn_id".into(), "turn-1".into()), + (GUARDIAN_TICKET_METADATA_KEY.into(), raw), + ])) + ); + attach( + &mut metadata, + /*ticket*/ None, + ResponsesEndpoint::Guardian, + ); + assert_eq!( + metadata, + Some(HashMap::from([("turn_id".into(), "turn-1".into())])) + ); +} + +#[test] +fn missing_or_malformed_response_receipts_do_not_become_tickets() { + for response in [ + json!({}), + json!({"headers": {"x-codex-guardian-ticket": "invalid"}}), + ] { + let event = + serde_json::from_value(json!({"type": "response.created", "response": response})) + .unwrap(); + assert!(matches!( + process_responses_event(event).unwrap(), + Some(ResponseEvent::Created { + guardian_ticket: None + }) + )); + } +} diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index 17e3ceb69b..a302c2595f 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -4,6 +4,7 @@ pub(crate) mod common; pub(crate) mod endpoint; pub(crate) mod error; pub(crate) mod files; +mod guardian_ticket; pub(crate) mod images; pub(crate) mod provider; pub(crate) mod rate_limits; diff --git a/codex-rs/codex-api/src/sse/responses.rs b/codex-rs/codex-api/src/sse/responses.rs index abef2468cc..8f2c5f2fa8 100644 --- a/codex-rs/codex-api/src/sse/responses.rs +++ b/codex-rs/codex-api/src/sse/responses.rs @@ -9,6 +9,8 @@ use crate::telemetry::SseTelemetry; use codex_client::ByteStream; use codex_client::StreamResponse; use codex_protocol::ResponseUsageMetadata; +use codex_protocol::guardian_ticket::GUARDIAN_TICKET_HEADER; +use codex_protocol::guardian_ticket::GuardianTicket; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::MisalignmentErrorDetails; use codex_protocol::protocol::ModelVerification; @@ -406,8 +408,18 @@ pub fn process_responses_event( } } "response.created" => { - if event.response.is_some() { - return Ok(Some(ResponseEvent::Created {})); + if let Some(response) = event.response { + let guardian_ticket = response + .get("headers") + .and_then(Value::as_object) + .and_then(|headers| { + headers + .iter() + .find(|(name, _)| name.eq_ignore_ascii_case(GUARDIAN_TICKET_HEADER)) + }) + .and_then(|(_, value)| value.as_str()) + .and_then(GuardianTicket::from_server); + return Ok(Some(ResponseEvent::Created { guardian_ticket })); } } "response.failed" => { @@ -1396,7 +1408,7 @@ mod tests { } fn is_created(ev: &ResponseEvent) -> bool { - matches!(ev, ResponseEvent::Created) + matches!(ev, ResponseEvent::Created { .. }) } fn is_output(ev: &ResponseEvent) -> bool { matches!(ev, ResponseEvent::OutputItemDone(_)) @@ -1578,7 +1590,7 @@ mod tests { .await; assert_eq!(events.len(), 2); - assert_matches!(&events[0], ResponseEvent::Created); + assert_matches!(&events[0], ResponseEvent::Created { .. }); assert_matches!( &events[1], ResponseEvent::Completed { @@ -1616,7 +1628,12 @@ mod tests { &events[0], ResponseEvent::ServerModel(model) if model == CYBER_RESTRICTED_MODEL_FOR_TESTS ); - assert_matches!(&events[1], ResponseEvent::Created); + assert_matches!( + &events[1], + ResponseEvent::Created { + guardian_ticket: None + } + ); assert_matches!( &events[2], ResponseEvent::Completed { @@ -1738,7 +1755,7 @@ mod tests { .await; assert_eq!(events.len(), 7); - assert_matches!(&events[0], ResponseEvent::Created); + assert_matches!(&events[0], ResponseEvent::Created { .. }); assert_matches!( &events[1], ResponseEvent::SafetyBuffering(buffering) diff --git a/codex-rs/codex-api/tests/clients.rs b/codex-rs/codex-api/tests/clients.rs index af43022417..07c34b3ba5 100644 --- a/codex-rs/codex-api/tests/clients.rs +++ b/codex-rs/codex-api/tests/clients.rs @@ -331,7 +331,8 @@ async fn responses_client_uses_guardian_path() -> Result<()> { async fn responses_client_stream_request_preserves_item_ids() -> Result<()> { let state = RecordingState::default(); let transport = RecordingTransport::new(state.clone()); - let client = ResponsesClient::new(transport, provider("openai"), Arc::new(NoAuth)); + let client = ResponsesClient::new(transport, provider("openai"), Arc::new(NoAuth)) + .with_endpoint(ResponsesEndpoint::Guardian); let request = ResponsesApiRequest { model: "gpt-test".into(), instructions: "Say hi".into(), @@ -356,14 +357,24 @@ async fn responses_client_stream_request_preserves_item_ids() -> Result<()> { client_metadata: None, access_programs: None, }; - let expected = serde_json::to_value(&request)?; + let raw_ticket = "t".repeat(43); + let ticket = codex_protocol::guardian_ticket::GuardianTicket::from_server(&raw_ticket).unwrap(); + let mut expected = serde_json::to_value(&request)?; + expected["client_metadata"] = serde_json::json!({"guardian_ticket": raw_ticket}); let _stream = client - .stream_request(request, ResponsesOptions::default()) + .stream_request( + request, + ResponsesOptions { + guardian_ticket: Some(ticket), + ..Default::default() + }, + ) .await?; let requests = state.take_stream_requests(); assert_eq!(requests.len(), 1); + assert!(!format!("{:?}", requests[0]).contains(&raw_ticket)); let prepared = requests[0] .prepare_body_for_send() .expect("body should prepare"); @@ -578,6 +589,7 @@ async fn azure_store_sends_ids_and_headers() -> Result<()> { extra_headers, compression: Compression::None, turn_state: None, + guardian_ticket: None, }, ) .await?; diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 9f03c939e0..9bef3b34cf 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -1104,6 +1104,31 @@ impl ModelClient { && provider.aws.is_none() } + fn set_guardian_ticket_request( + &self, + metadata: &mut Option>, + auth: Option<&CodexAuth>, + endpoint: ResponsesEndpoint, + ) { + if let Some(metadata) = metadata.as_mut() { + metadata.remove("guardian_ticket_requested"); + } + if self.free_guardian_enabled + && endpoint == ResponsesEndpoint::Responses + && !crate::guardian::is_basic_session_source(&self.state.session_source) + && matches!( + auth, + Some(CodexAuth::Chatgpt(_) | CodexAuth::ChatgptAuthTokens(_)) + ) + && self.uses_codex_backend(auth) + && self.state.provider.info().supports_codex_backend_routes() + { + metadata + .get_or_insert_with(HashMap::new) + .insert("guardian_ticket_requested".to_owned(), "true".to_owned()); + } + } + fn build_routing_hint_header( &self, auth: Option<&CodexAuth>, @@ -1326,6 +1351,7 @@ impl ModelClientSession { }, compression, turn_state: Some(Arc::clone(&self.turn_state)), + guardian_ticket: responses_metadata.guardian_ticket.clone(), } } @@ -1615,6 +1641,11 @@ impl ModelClientSession { service_tier.clone(), responses_metadata, )?; + self.client.set_guardian_ticket_request( + &mut request.client_metadata, + client_setup.auth.as_ref(), + endpoint, + ); if endpoint == ResponsesEndpoint::Guardian { request.service_tier = None; } @@ -1859,7 +1890,7 @@ impl ModelClientSession { .prepare_response_items_for_request(&mut request.input); Some(original_item_ids) }; - let ws_payload = ResponseCreateWsRequest { + let mut ws_payload = ResponseCreateWsRequest { previous_response_id, input: incremental_items.as_deref().unwrap_or(&request.input), generate: if warmup { Some(false) } else { None }, @@ -1869,6 +1900,11 @@ impl ModelClientSession { ), ..ResponseCreateWsRequest::from(&request) }; + self.client.set_guardian_ticket_request( + &mut ws_payload.client_metadata, + client_setup.auth.as_ref(), + endpoint, + ); let mut ws_request = ResponsesWsRequest::ResponseCreate(ws_payload); stamp_ws_stream_request_start_ms(&mut ws_request); if !previous_response_id_from_untraced_warmup { @@ -1886,6 +1922,7 @@ impl ModelClientSession { ws_request, self.websocket_session.connection_reused(), Some(Arc::clone(&self.turn_state)), + responses_metadata.guardian_ticket.as_ref(), ) .await; if let Some(original_item_ids) = original_item_ids { diff --git a/codex-rs/core/src/client_tests.rs b/codex-rs/core/src/client_tests.rs index 6eda1c86bf..ba97459c49 100644 --- a/codex-rs/core/src/client_tests.rs +++ b/codex-rs/core/src/client_tests.rs @@ -844,7 +844,9 @@ async fn response_stream_records_last_model_feedback_ids() { .set_default(); let api_stream = futures::stream::iter([ - Ok(ResponseEvent::Created), + Ok(ResponseEvent::Created { + guardian_ticket: None, + }), Ok(ResponseEvent::Completed { response_id: "resp-123".to_string(), token_usage: None, @@ -1060,7 +1062,9 @@ async fn dropped_backpressured_response_stream_traces_cancelled_partial_output() let backpressured_item_yielded = Arc::new(Notify::new()); let mut events = VecDeque::new(); for _ in 0..super::RESPONSE_STREAM_CHANNEL_CAPACITY { - events.push_back(ResponseEvent::Created); + events.push_back(ResponseEvent::Created { + guardian_ticket: None, + }); } events.push_back(ResponseEvent::OutputItemDone(output_message( "1", diff --git a/codex-rs/core/src/guardian/mod.rs b/codex-rs/core/src/guardian/mod.rs index 8eea38dad5..d3d40a7491 100644 --- a/codex-rs/core/src/guardian/mod.rs +++ b/codex-rs/core/src/guardian/mod.rs @@ -85,6 +85,8 @@ const GUARDIAN_RECENT_ENTRY_LIMIT: usize = 40; /// step-scoped things past their lifetime (like MCP bindings) #[derive(Clone)] pub(crate) struct GuardianReviewContext { + /// Ticket from the response currently handled in this execution context. + pub(crate) guardian_ticket: Option, turn: Arc, environments: TurnEnvironmentSnapshot, // Model and reasoning inputs are carried for the follow-up Guardian and V2 migrations. @@ -104,6 +106,11 @@ impl GuardianReviewContext { settings: &ResolvedStepSettings, ) -> Self { Self { + guardian_ticket: turn + .extension_data + .get::() + .as_deref() + .cloned(), environments: turn.environments.clone(), model_info: Arc::clone(&settings.model_info), reasoning_effort: settings.reasoning_effort().cloned(), @@ -126,6 +133,12 @@ impl GuardianReviewContext { impl From<&Arc> for GuardianReviewContext { fn from(step: &Arc) -> Self { Self { + guardian_ticket: step + .turn + .extension_data + .get::() + .as_deref() + .cloned(), turn: Arc::clone(&step.turn), environments: step.environments.clone(), model_info: Arc::clone(&step.settings.model_info), @@ -140,6 +153,11 @@ impl From<&Arc> for GuardianReviewContext { impl From> for GuardianReviewContext { fn from(turn: Arc) -> Self { Self { + guardian_ticket: turn + .extension_data + .get::() + .as_deref() + .cloned(), environments: turn.environments.clone(), model_info: Arc::clone(turn.model_info()), reasoning_effort: turn.reasoning_effort().cloned(), diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index c7ea001e5e..51af7c0f9a 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -1226,6 +1226,7 @@ async fn run_review_on_session( ..Default::default() }) .on_start(TurnStartOptions { + guardian_ticket: params.parent_context.guardian_ticket.clone(), final_output_json_schema: Some(params.schema.clone()), service_tier: None, parent_turn_id: Some(parent_turn.sub_id.clone()), diff --git a/codex-rs/core/src/responses_metadata.rs b/codex-rs/core/src/responses_metadata.rs index 5c97c16a24..6f64929ec8 100644 --- a/codex-rs/core/src/responses_metadata.rs +++ b/codex-rs/core/src/responses_metadata.rs @@ -58,6 +58,8 @@ pub(crate) const WORKSPACES_KEY: &str = "workspaces"; // App-server clients can specify additional metadata in the `responsesapi_client_metadata` param // when submitting a turn, but they must not override fields owned by core. const RESERVED_METADATA_KEYS: &[&str] = &[ + codex_protocol::guardian_ticket::GUARDIAN_TICKET_METADATA_KEY, + "guardian_ticket_requested", INSTALLATION_ID_KEY, X_CODEX_INSTALLATION_ID_HEADER, SESSION_ID_KEY, @@ -217,6 +219,8 @@ pub(crate) enum TurnToolSource { /// truth. #[derive(Clone, Debug)] pub struct CodexResponsesMetadata { + /// Opaque runtime receipt; deliberately omitted from metadata projections. + pub(crate) guardian_ticket: Option, pub(crate) installation_id: String, pub(crate) session_id: String, pub(crate) thread_id: String, @@ -256,6 +260,7 @@ impl CodexResponsesMetadata { window_id: String, ) -> Self { Self { + guardian_ticket: None, installation_id, session_id, thread_id, diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index cf7392ca26..a2453d227b 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -140,6 +140,7 @@ pub(super) async fn spawn_review_thread( extension_data.insert(parent_turn_context.skills_snapshot().as_ref().clone()); let review_turn_context = TurnContext { + guardian_ticket: None, sub_id: review_turn_id.clone(), trace_id: current_span_trace_id(), realtime_active: parent_turn_context.realtime_active, diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 56c0107c35..8d50a6885d 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -619,6 +619,7 @@ impl Session { request_kind, ); CodexResponsesMetadata { + guardian_ticket: turn_context.guardian_ticket.clone(), window_number: Some(window_number), context_window_id: Some(context_window_id), history_ingest_requested: turn_context diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 98b98db3dc..1482e540fe 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -1437,6 +1437,10 @@ async fn run_sampling_request( let mut original_input = None; let mut executed_tool_calls_by_output = HashMap::new(); loop { + // A retry must not lend a previous response's ticket to the next tool call. + turn_context + .extension_data + .remove::(); let prompt_input = if let Some(input) = initial_input.take() { input } else { @@ -2372,7 +2376,11 @@ async fn try_run_sampling_request( record_turn_ttft_metric(&turn_context, &event).await; match event { - ResponseEvent::Created => {} + ResponseEvent::Created { guardian_ticket } => { + if let Some(ticket) = guardian_ticket { + turn_context.extension_data.insert(ticket); + } + } ResponseEvent::OutputItemDone(mut item) => { assign_missing_streamed_response_item_id(&mut item, active_item.as_ref()); if analytics_tool_call_ids.len() < MAX_ANALYTICS_TOOL_CALL_IDS_PER_RESPONSE { diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index e5a371e9d0..c11e556901 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -185,6 +185,7 @@ impl std::fmt::Debug for TurnEnvironment { /// Request options for one turn, separate from persistent thread settings. #[derive(Default)] pub(crate) struct NewTurnContextOptions { + pub(crate) guardian_ticket: Option, pub(crate) final_output_json_schema: Option, pub(crate) cyber_access_program: Option, } @@ -192,6 +193,8 @@ pub(crate) struct NewTurnContextOptions { /// The context needed for a single turn of the thread. #[derive(Debug)] pub struct TurnContext { + /// Receipt selected for this Guardian turn, separate from its own responses. + pub(crate) guardian_ticket: Option, pub(crate) sub_id: String, pub(crate) trace_id: Option, pub(crate) realtime_active: bool, @@ -537,6 +540,7 @@ impl TurnContext { windows_sandbox_level: self.windows_sandbox_level, available_models, unified_exec_shell_mode: self.unified_exec_shell_mode.clone(), + guardian_ticket: self.guardian_ticket.clone(), final_output_json_schema: self.final_output_json_schema.clone(), dynamic_tools: self.dynamic_tools.clone(), turn_metadata_state: self.turn_metadata_state.clone(), @@ -781,6 +785,7 @@ impl Session { let extension_data = Arc::new(codex_extension_api::ExtensionData::new(sub_id.clone())); extension_data.insert(skills_snapshot); TurnContext { + guardian_ticket: None, sub_id, trace_id: current_span_trace_id(), realtime_active: false, @@ -1024,6 +1029,7 @@ impl Session { turn_context.extension_data.insert(trusted_plugin_roots); turn_context.realtime_active = self.conversation.running_state().await.is_some(); + turn_context.guardian_ticket = options.guardian_ticket; turn_context.final_output_json_schema = options.final_output_json_schema; if turn_context.config.model_provider_id == codex_model_provider_info::OPENAI_PROVIDER_ID { turn_context.cyber_access_program = options.cyber_access_program; diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index 2882931bca..79df5facfa 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -121,6 +121,7 @@ impl PreparedTurnInputSettings { kind: TurnStartKind, ) -> CodexResult>> { let TurnStartOptions { + guardian_ticket, turn_trigger, final_output_json_schema, service_tier, @@ -138,6 +139,7 @@ impl PreparedTurnInputSettings { updates.service_tier_for_turn = service_tier; let options = NewTurnContextOptions { + guardian_ticket, final_output_json_schema, cyber_access_program, }; diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 5bf41019d5..9cf988a295 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -475,6 +475,7 @@ impl Session { .new_turn_with_default_settings( sub_id, NewTurnContextOptions { + guardian_ticket: start_options.guardian_ticket, final_output_json_schema: start_options.final_output_json_schema, cyber_access_program: start_options.cyber_access_program, }, diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index b54da9a405..bc90f81a11 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/codex-rs/core/src/turn_timing.rs @@ -385,7 +385,7 @@ fn response_event_records_turn_ttft(event: &ResponseEvent) -> bool { | ResponseEvent::ReasoningSummaryDelta { .. } | ResponseEvent::ReasoningSummaryDone { .. } | ResponseEvent::ReasoningContentDelta { .. } => true, - ResponseEvent::Created + ResponseEvent::Created { .. } | ResponseEvent::ServerModel(_) | ResponseEvent::ModelVerifications(_) | ResponseEvent::TurnModerationMetadata(_) diff --git a/codex-rs/core/src/turn_timing_tests.rs b/codex-rs/core/src/turn_timing_tests.rs index 0feb9a19e0..ed79be2cd5 100644 --- a/codex-rs/core/src/turn_timing_tests.rs +++ b/codex-rs/core/src/turn_timing_tests.rs @@ -29,7 +29,9 @@ async fn turn_timing_state_records_ttft_only_once_per_turn() { state.mark_turn_started(Instant::now()).await; assert_eq!( state - .record_ttft_for_response_event(&ResponseEvent::Created) + .record_ttft_for_response_event(&ResponseEvent::Created { + guardian_ticket: None + }) .await, None ); diff --git a/codex-rs/core/tests/suite/code_mode.rs b/codex-rs/core/tests/suite/code_mode.rs index 112740dfea..e1ce2c5b73 100644 --- a/codex-rs/core/tests/suite/code_mode.rs +++ b/codex-rs/core/tests/suite/code_mode.rs @@ -2372,15 +2372,45 @@ async fn code_mode_wait_timeout_reconnects_on_next_exec() -> Result<()> { Ok(()) } +#[derive(Default)] +struct GuardianTicketObserver { + tickets: Mutex)>>, + wait_started: tokio::sync::Notify, +} + +impl ToolLifecycleContributor for GuardianTicketObserver { + fn on_tool_start<'a>(&'a self, input: ToolStartInput<'a>) -> ToolLifecycleFuture<'a> { + Box::pin(async move { + if matches!(input.tool_name.name.as_str(), "exec_command" | "wait") { + self.tickets.lock().unwrap().push(( + input.tool_name.name.clone(), + input + .turn_store + .get::() + .map(|ticket| ticket.as_str().to_owned()), + )); + if input.tool_name.name == "wait" { + self.wait_started.notify_one(); + } + } + }) + } +} + #[cfg_attr(windows, ignore = "no exec_command on Windows")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn code_mode_can_yield_and_resume_with_wait() -> Result<()> { skip_if_no_network!(Ok(())); let server = responses::start_mock_server().await; - let mut builder = test_codex().with_config(move |config| { - let _ = config.features.enable(Feature::CodeMode); - }); + let observer = Arc::new(GuardianTicketObserver::default()); + let mut extensions = ExtensionRegistryBuilder::::new(); + extensions.tool_lifecycle_contributor(observer.clone()); + let mut builder = test_codex() + .with_extensions(Arc::new(extensions.build())) + .with_config(move |config| { + let _ = config.features.enable(Feature::CodeMode); + }); let test = builder.build(&server).await?; let phase_2_gate = test.workspace_path("code-mode-phase-2.ready"); let phase_3_gate = test.workspace_path("code-mode-phase-3.ready"); @@ -2392,16 +2422,16 @@ async fn code_mode_can_yield_and_resume_with_wait() -> Result<()> { text("phase 1"); yield_control(); {phase_2_wait} -text("phase 2"); +text((await tools.exec_command({{cmd: "printf 'phase 2'"}})).output); {phase_3_wait} -text("phase 3"); +text((await tools.exec_command({{cmd: "printf 'phase 3'"}})).output); "# ); responses::mount_sse_once( &server, sse(vec![ - ev_response_created("resp-1"), + serde_json::json!({"type": "response.created", "response": {"id": "resp-1", "headers": {"x-codex-guardian-ticket": "p".repeat(43)}}}), ev_custom_tool_call("call-1", "exec", &code), ev_completed("resp-1"), ]), @@ -2434,7 +2464,7 @@ text("phase 3"); responses::mount_sse_once( &server, sse(vec![ - ev_response_created("resp-3"), + serde_json::json!({"type": "response.created", "response": {"id": "resp-3", "headers": {"x-codex-guardian-ticket": "q".repeat(43)}}}), responses::ev_function_call( "call-2", "wait", @@ -2456,8 +2486,11 @@ text("phase 3"); ) .await; - fs::write(&phase_2_gate, "ready")?; - test.submit_turn("wait again").await?; + tokio::try_join!(test.submit_turn("wait again"), async { + observer.wait_started.notified().await; + fs::write(&phase_2_gate, "ready")?; + anyhow::Ok(()) + })?; let second_request = second_completion.single_request(); let second_items = function_tool_output_items(&second_request, "call-2"); @@ -2478,7 +2511,7 @@ text("phase 3"); responses::mount_sse_once( &server, sse(vec![ - ev_response_created("resp-5"), + serde_json::json!({"type": "response.created", "response": {"id": "resp-5", "headers": {"x-codex-guardian-ticket": "r".repeat(43)}}}), responses::ev_function_call( "call-3", "wait", @@ -2500,8 +2533,11 @@ text("phase 3"); ) .await; - fs::write(&phase_3_gate, "ready")?; - test.submit_turn("wait for completion").await?; + tokio::try_join!(test.submit_turn("wait for completion"), async { + observer.wait_started.notified().await; + fs::write(&phase_3_gate, "ready")?; + anyhow::Ok(()) + })?; let third_request = third_completion.single_request(); let third_items = function_tool_output_items(&third_request, "call-3"); @@ -2515,6 +2551,26 @@ text("phase 3"); ); assert_eq!(text_item(&third_items, /*index*/ 1), "phase 3"); + let observed = observer.tickets.lock().unwrap(); + let mut nested_tickets = observed + .iter() + .filter(|(tool, _)| tool == "exec_command") + .filter_map(|(_, ticket)| ticket.clone()) + .skip_while(|ticket| ticket != &"q".repeat(43)) + .collect::>(); + nested_tickets.dedup(); + assert_eq!(nested_tickets, vec!["q".repeat(43), "r".repeat(43)]); + assert_eq!( + observed + .iter() + .filter(|(tool, _)| tool == "wait") + .cloned() + .collect::>(), + vec![ + ("wait".to_owned(), Some("q".repeat(43))), + ("wait".to_owned(), Some("r".repeat(43))), + ] + ); Ok(()) } diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index 19fc625a84..9a790463a0 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/codex-rs/core/tests/suite/guardian_review.rs @@ -130,15 +130,19 @@ impl TimeProvider for RecordingTimeProvider { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -#[test_case(CodexAuth::from_api_key("test-api-key"), "/v1", true, "/v1/responses"; "api_key_uses_responses")] -#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "/backend-api/codex", false, "/backend-api/codex/responses"; "chatgpt_uses_responses_by_default")] -#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "/backend-api/codex", true, "/backend-api/codex/guardian"; "chatgpt_uses_guardian_when_enabled")] -#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "/v1", true, "/v1/responses"; "custom_openai_url_uses_responses")] +#[test_case(CodexAuth::from_api_key("test-api-key"), "OpenAI", "/v1", true, "/v1/responses", true; "api_key_uses_responses")] +#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "OpenAI", "/backend-api/codex", false, "/backend-api/codex/responses", true; "chatgpt_uses_responses_by_default")] +#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "OpenAI", "/backend-api/codex", true, "/backend-api/codex/guardian", true; "chatgpt_uses_guardian_when_enabled")] +#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "OpenAI", "/v1", true, "/v1/responses", true; "custom_openai_url_uses_responses")] +#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "Custom", "/backend-api/codex", true, "/backend-api/codex/responses", true; "custom_provider_uses_responses")] +#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "OpenAI", "/backend-api/codex", true, "/backend-api/codex/guardian", false; "server_without_tickets")] async fn guardian_session_inherits_parent_http_fallback( auth: CodexAuth, + provider_name: &str, base_path: &str, free_guardian: bool, expected_guardian_path: &str, + ticket_issued: bool, ) -> Result<()> { skip_if_no_network!(Ok(())); skip_if_wine_exec!( @@ -154,10 +158,16 @@ async fn guardian_session_inherits_parent_http_fallback( .mount_as_scoped(&server) .await; + let parent_ticket = "p".repeat(43); let responses = mount_sse_sequence( &server, vec![ + // A failed sampling attempt must not lend its receipt to the retried action. + sse(vec![json!({"type": "response.created", "response": { + "id": "failed-parent", "headers": {"x-codex-guardian-ticket": "f".repeat(43)} + }})]), sse(vec![ + json!({"type": "response.created", "response": {"id": "parent-tool", "headers": ticket_issued.then(|| json!({"x-codex-guardian-ticket": parent_ticket}))}}), ev_function_call( "call", "exec_command", @@ -175,6 +185,7 @@ async fn guardian_session_inherits_parent_http_fallback( .await; let base_url = format!("{}{base_path}", server.uri()); + let provider_name = provider_name.to_owned(); let mut builder = test_codex() .with_auth(auth) .with_pre_build_hook(move |home| { @@ -186,7 +197,9 @@ async fn guardian_session_inherits_parent_http_fallback( }) .with_config(move |config| { config.model_provider.base_url = Some(base_url); + config.model_provider.name = provider_name; config.model_provider.supports_websockets = true; + config.model_provider.stream_max_retries = Some(1); config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); config.approvals_reviewer = ApprovalsReviewer::User; }); @@ -232,6 +245,34 @@ async fn guardian_session_inherits_parent_http_fallback( }) .expect("Guardian reviewer inference request"); assert_eq!(guardian_request.path(), expected_guardian_path); + let tickets_enabled = expected_guardian_path.ends_with("/guardian"); + let body = guardian_request.body_json(); + assert_eq!( + ( + body["client_metadata"].get("guardian_ticket").cloned(), + body["client_metadata"].get("guardian_ticket_requested"), + ), + ( + (tickets_enabled && ticket_issued).then(|| json!(parent_ticket)), + None + ) + ); + assert!(!body["input"].to_string().contains(&parent_ticket)); + for request in responses.requests() { + let body = request.body_json(); + if body["client_metadata"]["x-openai-subagent"] != "guardian" { + assert_eq!( + ( + body["client_metadata"] + .get("guardian_ticket_requested") + .cloned(), + body["client_metadata"].get("guardian_ticket"), + ), + (tickets_enabled.then(|| json!("true")), None) + ); + } + } + test.codex.shutdown_and_wait().await?; Ok(()) } @@ -348,15 +389,18 @@ async fn guardian_review_resends_full_transcript_after_reviewer_context_rollover } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -#[test_case(CodexAuth::from_api_key("test-api-key"), "gpt-5.6-luna"; "api_key_uses_luna_with_responses_lite")] -#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "codex-auto-review"; "chatgpt_uses_codex_auto_review")] +#[test_case(CodexAuth::from_api_key("test-api-key"), true, "gpt-5.6-luna"; "api_key_uses_luna_with_responses_lite")] +#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), true, "codex-auto-review"; "chatgpt_uses_codex_auto_review")] +#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), false, "codex-auto-review"; "chatgpt_without_free_guardian")] async fn guardian_session_prewarms_and_is_reused_for_first_review( auth: CodexAuth, + free_guardian: bool, expected_model: &str, ) -> Result<()> { skip_if_no_network!(Ok(())); let uses_codex_backend = auth.uses_codex_backend(); + let tickets_enabled = free_guardian && uses_codex_backend; let bundled_models = codex_models_manager::bundled_models_response()?.models; let catalog_auto_review = bundled_models .iter() @@ -389,11 +433,12 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review( "justification": "Exercise Guardian approval routing.", }) .to_string(); + let parent_ticket = "w".repeat(43); let server = start_websocket_server(vec![ vec![vec![ev_response_created("warm-1"), ev_completed("warm-1")]], vec![vec![ev_response_created("warm-2"), ev_completed("warm-2")]], vec![vec![ - ev_response_created("approval-request"), + json!({"type": "response.created", "response": {"id": "approval-request", "headers": {"x-codex-guardian-ticket": parent_ticket}}}), ev_function_call("approval-call", "exec_command", &tool_args), ev_completed("approval-request"), ]], @@ -419,10 +464,10 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review( let backend_base_url = format!("{}/backend-api/codex", server.uri()); let mut builder = test_codex() .with_auth(auth) - .with_pre_build_hook(|home| { + .with_pre_build_hook(move |home| { fs::write( home.join("config.toml"), - "[features.guardianv2]\nfree_guardian = true\n", + format!("[features.guardianv2]\nfree_guardian = {free_guardian}\n"), ) .expect("Guardian endpoint configuration should be written"); }) @@ -546,6 +591,34 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review( .expect("reviewed parent turn id"); assert_parent_turn(&parent_request, /*expected*/ None)?; assert_parent_turn(&guardian_review, Some(parent_turn_id))?; + assert_eq!( + ( + guardian_review["client_metadata"] + .get("guardian_ticket") + .cloned(), + guardian_review["client_metadata"].get("guardian_ticket_requested"), + parent_request["client_metadata"] + .get("guardian_ticket_requested") + .cloned(), + parent_request["client_metadata"].get("guardian_ticket"), + ), + ( + tickets_enabled.then(|| json!(parent_ticket)), + None, + tickets_enabled.then(|| json!("true")), + None, + ) + ); + assert!( + guardian_prewarm["client_metadata"] + .get("guardian_ticket") + .is_none() + ); + assert!( + !guardian_review["input"] + .to_string() + .contains(&parent_ticket) + ); for request in [&parent_request, &guardian_review] { assert_root_turn(request, Some(parent_turn_id))?; } @@ -628,7 +701,7 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review( assert_eq!(guardian_context_windows, vec![Some(258_400)]); for handshake in server.handshakes() { let is_guardian = handshake.header("x-openai-subagent").as_deref() == Some("guardian"); - let uses_guardian_endpoint = uses_codex_backend && is_guardian; + let uses_guardian_endpoint = tickets_enabled && is_guardian; assert_eq!( handshake.uri(), if uses_guardian_endpoint { diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs index 64b8bafb3f..c6a41bec6d 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs @@ -506,6 +506,11 @@ impl GuardianV2Extension { let thread_id = input.thread_store.level_id().to_owned(); let turn_id = input.turn_id.to_owned(); let root_turn_id = input.root_turn_id.map(str::to_owned); + let guardian_ticket = input + .turn_store + .get::() + .as_deref() + .cloned(); let thread_context: Result<_, String> = async { let parsed_thread_id = ThreadId::from_string(&thread_id).map_err(|error| error.to_string())?; @@ -816,6 +821,7 @@ impl GuardianV2Extension { let instructions = guardian_config.render_classifier_instructions(policy); let output = match sampler .sample(LunaSamplingRequest { + guardian_ticket, instructions, trusted_review_evidence, trusted_tool_context, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs index 6a822cfd71..f2b7bcc14b 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs @@ -90,6 +90,8 @@ pub struct LunaSamplerConfig { /// One tool-less Luna classification request over an already-open connection. pub struct LunaSamplingRequest { + /// Runtime receipt of the response handling the classified tool. + pub guardian_ticket: Option, /// Trusted instructions describing the requested classification. pub instructions: String, /// Host-supplied Guardian reviews isolated from untrusted transcript entries. @@ -452,6 +454,7 @@ impl LunaSampler { } // A classification is its own inference turn; retries keep that identity. let turn_id = Uuid::now_v7().to_string(); + let guardian_ticket = request.guardian_ticket; let parent_turn_id = request.parent_turn_id; let root_turn_id = request.root_turn_id; let mut input = vec![ @@ -634,6 +637,7 @@ impl LunaSampler { ResponsesWsRequest::ResponseCreate((&request).into()), /*connection_reused*/ true, /*turn_state*/ None, + guardian_ticket.as_ref(), ) .await { diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs index 4750b3d7f8..4d61c61fe5 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs @@ -192,6 +192,7 @@ async fn connect_sampler(config: LunaSamplerConfig) -> Result { fn sample_request(parent_turn_id: &str) -> LunaSamplingRequest { LunaSamplingRequest { + guardian_ticket: None, instructions: "Return high for high risk or low for low risk.".to_owned(), trusted_review_evidence: Vec::new(), trusted_tool_context: None, @@ -446,6 +447,7 @@ async fn preconnected_sampler_reuses_authenticated_websocket_for_classifications let first = sampler .sample(LunaSamplingRequest { + guardian_ticket: None, instructions: "Return high for high risk or low for low risk.".to_owned(), trusted_review_evidence: Vec::new(), trusted_tool_context: None, @@ -477,6 +479,7 @@ async fn preconnected_sampler_reuses_authenticated_websocket_for_classifications manager.refresh_token_from_authority().await?; let second = sampler .sample(LunaSamplingRequest { + guardian_ticket: None, instructions: "Return high for high risk or low for low risk.".to_owned(), trusted_review_evidence: Vec::new(), trusted_tool_context: None, @@ -652,6 +655,7 @@ async fn sampler_returns_classification_token_before_terminal_response_events() let output = tokio::time::timeout( Duration::from_secs(2), sampler.sample(LunaSamplingRequest { + guardian_ticket: None, instructions: "Return high for high risk or low for low risk.".to_owned(), trusted_review_evidence: Vec::new(), trusted_tool_context: None, @@ -1113,3 +1117,54 @@ async fn sampler_limits_transient_recovery_attempts() -> Result<()> { Ok(()) } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn guardian_ticket_survives_classifier_transport_retry() -> Result<()> { + skip_if_no_network!(Ok(())); + for free_guardian in [false, true] { + let healthy = responses::start_websocket_server(vec![vec![vec![ + ev_assistant_message("resp-review", "low"), + ev_completed("resp-review"), + ]]]) + .await; + let expired = responses::start_websocket_server(vec![vec![vec![json!({ + "type": "error", "status": 400, + "error": {"type": "invalid_request_error", "code": "websocket_connection_limit_reached", "message": "expired"} + })]]]).await; + let base_url = proxy_websocket_servers(&[&healthy, &expired]).await?; + let mut config = sampler_config(base_url.clone()); + if free_guardian { + config.free_guardian = true; + config.provider = create_model_provider( + ModelProviderInfo::create_openai_provider(Some(format!( + "{}/backend-api/codex", + base_url.trim_end_matches("/v1") + ))), + Some(AuthManager::from_auth_for_testing( + CodexAuth::create_dummy_chatgpt_auth_for_testing(), + )), + ); + } + let sampler = connect_sampler(config).await?; + let raw_ticket = "t".repeat(43); + let mut request = sample_request("turn-1"); + request.guardian_ticket = + codex_protocol::guardian_ticket::GuardianTicket::from_server(&raw_ticket); + assert_eq!(sampler.sample(request).await?, "low"); + for server in [&expired, &healthy] { + let requests = server.single_connection(); + assert_eq!(requests.len(), 1); + let body = requests[0].body_json(); + assert_eq!( + ( + body["client_metadata"].get("guardian_ticket").cloned(), + body["client_metadata"].get("guardian_ticket_requested"), + ), + (free_guardian.then(|| json!(raw_ticket)), None) + ); + assert!(!body["input"].to_string().contains(&raw_ticket)); + } + } + + Ok(()) +} diff --git a/codex-rs/http-client/src/request.rs b/codex-rs/http-client/src/request.rs index 7608860c3d..84b9a730d9 100644 --- a/codex-rs/http-client/src/request.rs +++ b/codex-rs/http-client/src/request.rs @@ -11,7 +11,7 @@ use std::time::Duration; /// Clones share the encoded allocation. Internally, the body can also hold the /// final compressed wire bytes while retaining the original JSON only when /// request-body trace logging is enabled. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Clone, PartialEq, Eq)] pub struct EncodedJsonBody { bytes: Bytes, trace_bytes: Option, @@ -33,11 +33,27 @@ impl EncodedJsonBody { &self.bytes } + /// Omits credential-bearing bodies from tracing and debug output, including retries. + pub fn without_body_logging(mut self) -> Self { + self.trace_bytes = Some(Bytes::from_static(b"")); + self + } + pub(crate) fn trace_bytes(&self) -> &[u8] { self.trace_bytes.as_ref().unwrap_or(&self.bytes) } } +impl std::fmt::Debug for EncodedJsonBody { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("EncodedJsonBody") + .field("body", &String::from_utf8_lossy(self.trace_bytes())) + .field("prepared", &self.prepared) + .finish() + } +} + #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub enum RequestCompression { #[default] @@ -120,7 +136,13 @@ impl Request { self.body, Some(RequestBody::Json(_) | RequestBody::EncodedJson(_)) ); - let trace_bytes = if self.compression != RequestCompression::None + let retained_trace_bytes = match self.body.as_ref() { + Some(RequestBody::EncodedJson(body)) => body.trace_bytes.clone(), + Some(RequestBody::Json(_) | RequestBody::Raw(_)) | None => None, + }; + let trace_bytes = if retained_trace_bytes.is_some() { + retained_trace_bytes + } else if self.compression != RequestCompression::None && tracing::enabled!(target: "codex_http_client::transport", tracing::Level::TRACE) { match self.body.as_ref() { diff --git a/codex-rs/http-client/src/transport_tests.rs b/codex-rs/http-client/src/transport_tests.rs index a8ab11aba5..6f28df9025 100644 --- a/codex-rs/http-client/src/transport_tests.rs +++ b/codex-rs/http-client/src/transport_tests.rs @@ -1,4 +1,7 @@ use super::*; +use crate::request::EncodedJsonBody; +use crate::request::RequestCompression; +use pretty_assertions::assert_eq; use serde_json::json; use std::io::Write; use std::sync::Arc; @@ -28,6 +31,35 @@ async fn disabled_request_logging_suppresses_transport_url_and_body() { assert!(!logs.contains("body-secret")); } +#[test] +fn sensitive_encoded_body_stays_redacted_after_preparation_and_retries() { + let payload = json!({"client_metadata": {"guardian_ticket": "secret-receipt"}}); + for compression in [RequestCompression::None, RequestCompression::Zstd] { + let mut request = Request::new(Method::POST, "https://example.com/responses".into()) + .with_compression(compression); + request.body = Some(RequestBody::EncodedJson( + EncodedJsonBody::encode(&payload) + .unwrap() + .without_body_logging(), + )); + assert_eq!(request_body_for_trace(&request), ""); + for _ in 0..2 { + request = request.clone().into_prepared().unwrap(); + assert_eq!(request_body_for_trace(&request), ""); + assert!(!format!("{request:?}").contains("secret-receipt")); + let body = request.prepare_body_for_send().unwrap().body.unwrap(); + let decoded = match compression { + RequestCompression::None => body.to_vec(), + RequestCompression::Zstd => zstd::stream::decode_all(body.as_ref()).unwrap(), + }; + assert_eq!( + serde_json::from_slice::(&decoded).unwrap(), + payload + ); + } + } +} + #[tokio::test] async fn connection_failures_are_classified_without_exposing_request_urls() { let unavailable_server = diff --git a/codex-rs/otel/src/events/session_telemetry.rs b/codex-rs/otel/src/events/session_telemetry.rs index b266b71b2a..ddcd7fb2bb 100644 --- a/codex-rs/otel/src/events/session_telemetry.rs +++ b/codex-rs/otel/src/events/session_telemetry.rs @@ -1270,7 +1270,7 @@ impl SessionTelemetry { fn responses_type(event: &ResponseEvent) -> String { match event { - ResponseEvent::Created => "created".into(), + ResponseEvent::Created { .. } => "created".into(), ResponseEvent::OutputItemDone(item) | ResponseEvent::OutputItemAdded(item) => { SessionTelemetry::responses_item_type(item) } diff --git a/codex-rs/protocol/src/guardian_ticket.rs b/codex-rs/protocol/src/guardian_ticket.rs new file mode 100644 index 0000000000..942a463ae8 --- /dev/null +++ b/codex-rs/protocol/src/guardian_ticket.rs @@ -0,0 +1,37 @@ +//! Opaque parent-inference receipts kept out of persisted and model-visible context. + +use std::fmt; + +/// Header carried on the parent response's metadata, for both SSE and WebSocket. +pub const GUARDIAN_TICKET_HEADER: &str = "x-codex-guardian-ticket"; +/// Request metadata consumed by Codex backend before forwarding reviewer inference. +pub const GUARDIAN_TICKET_METADATA_KEY: &str = "guardian_ticket"; + +/// A server-issued receipt. Parsing checks its shape, not its authenticity. +/// +/// Only the backend can validate ownership, expiry, and remaining allowance. This +/// type intentionally has no serialization implementation or unredacted Debug. +#[derive(Clone, PartialEq, Eq)] +pub struct GuardianTicket(String); + +impl GuardianTicket { + /// Accept the bounded URL-safe representation used by the receipt issuer. + pub fn from_server(value: &str) -> Option { + (value.len() == 43 + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))) + .then(|| Self(value.to_owned())) + } + + /// Expose the receipt only when writing transport metadata. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl fmt::Debug for GuardianTicket { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("GuardianTicket([REDACTED])") + } +} diff --git a/codex-rs/protocol/src/lib.rs b/codex-rs/protocol/src/lib.rs index 10af9c7317..80531f4586 100644 --- a/codex-rs/protocol/src/lib.rs +++ b/codex-rs/protocol/src/lib.rs @@ -25,6 +25,7 @@ pub mod dynamic_tools; mod environment; pub mod error; pub mod exec_output; +pub mod guardian_ticket; pub mod items; mod legacy_events; pub mod local_media; diff --git a/codex-rs/protocol/src/turn_input.rs b/codex-rs/protocol/src/turn_input.rs index 4819d0fb9b..2715fd61cf 100644 --- a/codex-rs/protocol/src/turn_input.rs +++ b/codex-rs/protocol/src/turn_input.rs @@ -158,6 +158,8 @@ pub enum CyberAccessProgram { /// child input, Core also compares root lineage to detect ambiguity. #[derive(Clone, Debug, Default)] pub struct TurnStartOptions { + /// Parent inference receipt for an internal Guardian turn. Never persisted. + pub guardian_ticket: Option, /// Source classification for the caller that starts a new turn. /// Ignored when the submitted input steers an active turn. pub turn_trigger: Option,