Propagate response tickets to Guardian reviews (#42758)

## What changed

- Request and parse server-issued Guardian tickets for eligible Codex backend responses.
- Scope each ticket to the response that produced a Guardian review or classifier request, including HTTP and WebSocket retries, without adding it to model-visible or persisted context.
- Validate ticket shape, reserve its metadata keys, redact it from debug and request-body logging, and attach it only at the Guardian transport boundary.

## Testing

- Cover ticket parsing, endpoint and authentication routing, retry scoping, classifier retries, and redaction for compressed and uncompressed requests.

GitOrigin-RevId: 6219dd15c9f19ed3387de516aac1b8e7c9cdb0e3
This commit is contained in:
jif authored and copyberry committed 2026-09-04 12:56:26 +00:00
1 parent 97e46694e1
commit cc4b8bdeb8
32 files changed
+562 -44

No files matched your search

+4 -1
View File
@@ -96,7 +96,10 @@ pub struct MemorySummarizeOutput {
#[derive(Debug)] #[derive(Debug)]
pub enum ResponseEvent { pub enum ResponseEvent {
Created, Created {
/// Runtime-only receipt for this exact server response.
guardian_ticket: Option<codex_protocol::guardian_ticket::GuardianTicket>,
},
SafetyBuffering(SafetyBuffering), SafetyBuffering(SafetyBuffering),
OutputItemDone(ResponseItem), OutputItemDone(ResponseItem),
OutputItemAdded(ResponseItem), OutputItemAdded(ResponseItem),
+13 -2
View File
@@ -60,6 +60,8 @@ pub struct ResponsesOptions {
pub extra_headers: HeaderMap, pub extra_headers: HeaderMap,
pub compression: Compression, pub compression: Compression,
pub turn_state: Option<Arc<OnceLock<String>>>, pub turn_state: Option<Arc<OnceLock<String>>>,
/// Runtime-only parent receipt, attached after inference tracing.
pub guardian_ticket: Option<codex_protocol::guardian_ticket::GuardianTicket>,
} }
impl<T: HttpTransport> ResponsesClient<T> { impl<T: HttpTransport> ResponsesClient<T> {
@@ -101,7 +103,7 @@ impl<T: HttpTransport> ResponsesClient<T> {
)] )]
pub async fn stream_request( pub async fn stream_request(
&self, &self,
request: ResponsesApiRequest, mut request: ResponsesApiRequest,
options: ResponsesOptions, options: ResponsesOptions,
) -> Result<ResponseStream, ApiError> { ) -> Result<ResponseStream, ApiError> {
let ResponsesOptions { let ResponsesOptions {
@@ -111,10 +113,19 @@ impl<T: HttpTransport> ResponsesClient<T> {
extra_headers, extra_headers,
compression, compression,
turn_state, turn_state,
guardian_ticket,
} = options; } = 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}")))?; .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; let mut headers = extra_headers;
if let Some(ref thread_id) = thread_id { if let Some(ref thread_id) = thread_id {
@@ -234,9 +234,10 @@ impl ResponsesWebsocketConnection {
)] )]
pub async fn stream_request( pub async fn stream_request(
&self, &self,
request: ResponsesWsRequest<'_>, mut request: ResponsesWsRequest<'_>,
connection_reused: bool, connection_reused: bool,
turn_state: Option<Arc<OnceLock<String>>>, turn_state: Option<Arc<OnceLock<String>>>,
guardian_ticket: Option<&codex_protocol::guardian_ticket::GuardianTicket>,
) -> Result<ResponseStream, ApiError> { ) -> Result<ResponseStream, ApiError> {
let (tx_event, rx_event) = let (tx_event, rx_event) =
mpsc::channel::<std::result::Result<ResponseEvent, ApiError>>(1600); mpsc::channel::<std::result::Result<ResponseEvent, ApiError>>(1600);
@@ -270,6 +271,12 @@ impl ResponsesWebsocketConnection {
warmup: ws_request.generate == Some(false), warmup: ws_request.generate == Some(false),
connection_reused, 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 request_text = serialize_websocket_request(&request)?;
let current_span = Span::current(); let current_span = Span::current();
+33
View File
@@ -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<HashMap<String, String>>,
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;
@@ -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
})
));
}
}
+1
View File
@@ -4,6 +4,7 @@ pub(crate) mod common;
pub(crate) mod endpoint; pub(crate) mod endpoint;
pub(crate) mod error; pub(crate) mod error;
pub(crate) mod files; pub(crate) mod files;
mod guardian_ticket;
pub(crate) mod images; pub(crate) mod images;
pub(crate) mod provider; pub(crate) mod provider;
pub(crate) mod rate_limits; pub(crate) mod rate_limits;
+23 -6
View File
@@ -9,6 +9,8 @@ use crate::telemetry::SseTelemetry;
use codex_client::ByteStream; use codex_client::ByteStream;
use codex_client::StreamResponse; use codex_client::StreamResponse;
use codex_protocol::ResponseUsageMetadata; 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::models::ResponseItem;
use codex_protocol::protocol::MisalignmentErrorDetails; use codex_protocol::protocol::MisalignmentErrorDetails;
use codex_protocol::protocol::ModelVerification; use codex_protocol::protocol::ModelVerification;
@@ -406,8 +408,18 @@ pub fn process_responses_event(
} }
} }
"response.created" => { "response.created" => {
if event.response.is_some() { if let Some(response) = event.response {
return Ok(Some(ResponseEvent::Created {})); 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" => { "response.failed" => {
@@ -1396,7 +1408,7 @@ mod tests {
} }
fn is_created(ev: &ResponseEvent) -> bool { fn is_created(ev: &ResponseEvent) -> bool {
matches!(ev, ResponseEvent::Created) matches!(ev, ResponseEvent::Created { .. })
} }
fn is_output(ev: &ResponseEvent) -> bool { fn is_output(ev: &ResponseEvent) -> bool {
matches!(ev, ResponseEvent::OutputItemDone(_)) matches!(ev, ResponseEvent::OutputItemDone(_))
@@ -1578,7 +1590,7 @@ mod tests {
.await; .await;
assert_eq!(events.len(), 2); assert_eq!(events.len(), 2);
assert_matches!(&events[0], ResponseEvent::Created); assert_matches!(&events[0], ResponseEvent::Created { .. });
assert_matches!( assert_matches!(
&events[1], &events[1],
ResponseEvent::Completed { ResponseEvent::Completed {
@@ -1616,7 +1628,12 @@ mod tests {
&events[0], &events[0],
ResponseEvent::ServerModel(model) if model == CYBER_RESTRICTED_MODEL_FOR_TESTS 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!( assert_matches!(
&events[2], &events[2],
ResponseEvent::Completed { ResponseEvent::Completed {
@@ -1738,7 +1755,7 @@ mod tests {
.await; .await;
assert_eq!(events.len(), 7); assert_eq!(events.len(), 7);
assert_matches!(&events[0], ResponseEvent::Created); assert_matches!(&events[0], ResponseEvent::Created { .. });
assert_matches!( assert_matches!(
&events[1], &events[1],
ResponseEvent::SafetyBuffering(buffering) ResponseEvent::SafetyBuffering(buffering)
+15 -3
View File
@@ -331,7 +331,8 @@ async fn responses_client_uses_guardian_path() -> Result<()> {
async fn responses_client_stream_request_preserves_item_ids() -> Result<()> { async fn responses_client_stream_request_preserves_item_ids() -> Result<()> {
let state = RecordingState::default(); let state = RecordingState::default();
let transport = RecordingTransport::new(state.clone()); 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 { let request = ResponsesApiRequest {
model: "gpt-test".into(), model: "gpt-test".into(),
instructions: "Say hi".into(), instructions: "Say hi".into(),
@@ -356,14 +357,24 @@ async fn responses_client_stream_request_preserves_item_ids() -> Result<()> {
client_metadata: None, client_metadata: None,
access_programs: 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 let _stream = client
.stream_request(request, ResponsesOptions::default()) .stream_request(
request,
ResponsesOptions {
guardian_ticket: Some(ticket),
..Default::default()
},
)
.await?; .await?;
let requests = state.take_stream_requests(); let requests = state.take_stream_requests();
assert_eq!(requests.len(), 1); assert_eq!(requests.len(), 1);
assert!(!format!("{:?}", requests[0]).contains(&raw_ticket));
let prepared = requests[0] let prepared = requests[0]
.prepare_body_for_send() .prepare_body_for_send()
.expect("body should prepare"); .expect("body should prepare");
@@ -578,6 +589,7 @@ async fn azure_store_sends_ids_and_headers() -> Result<()> {
extra_headers, extra_headers,
compression: Compression::None, compression: Compression::None,
turn_state: None, turn_state: None,
guardian_ticket: None,
}, },
) )
.await?; .await?;
+38 -1
View File
@@ -1104,6 +1104,31 @@ impl ModelClient {
&& provider.aws.is_none() && provider.aws.is_none()
} }
fn set_guardian_ticket_request(
&self,
metadata: &mut Option<HashMap<String, String>>,
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( fn build_routing_hint_header(
&self, &self,
auth: Option<&CodexAuth>, auth: Option<&CodexAuth>,
@@ -1326,6 +1351,7 @@ impl ModelClientSession {
}, },
compression, compression,
turn_state: Some(Arc::clone(&self.turn_state)), turn_state: Some(Arc::clone(&self.turn_state)),
guardian_ticket: responses_metadata.guardian_ticket.clone(),
} }
} }
@@ -1615,6 +1641,11 @@ impl ModelClientSession {
service_tier.clone(), service_tier.clone(),
responses_metadata, responses_metadata,
)?; )?;
self.client.set_guardian_ticket_request(
&mut request.client_metadata,
client_setup.auth.as_ref(),
endpoint,
);
if endpoint == ResponsesEndpoint::Guardian { if endpoint == ResponsesEndpoint::Guardian {
request.service_tier = None; request.service_tier = None;
} }
@@ -1859,7 +1890,7 @@ impl ModelClientSession {
.prepare_response_items_for_request(&mut request.input); .prepare_response_items_for_request(&mut request.input);
Some(original_item_ids) Some(original_item_ids)
}; };
let ws_payload = ResponseCreateWsRequest { let mut ws_payload = ResponseCreateWsRequest {
previous_response_id, previous_response_id,
input: incremental_items.as_deref().unwrap_or(&request.input), input: incremental_items.as_deref().unwrap_or(&request.input),
generate: if warmup { Some(false) } else { None }, generate: if warmup { Some(false) } else { None },
@@ -1869,6 +1900,11 @@ impl ModelClientSession {
), ),
..ResponseCreateWsRequest::from(&request) ..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); let mut ws_request = ResponsesWsRequest::ResponseCreate(ws_payload);
stamp_ws_stream_request_start_ms(&mut ws_request); stamp_ws_stream_request_start_ms(&mut ws_request);
if !previous_response_id_from_untraced_warmup { if !previous_response_id_from_untraced_warmup {
@@ -1886,6 +1922,7 @@ impl ModelClientSession {
ws_request, ws_request,
self.websocket_session.connection_reused(), self.websocket_session.connection_reused(),
Some(Arc::clone(&self.turn_state)), Some(Arc::clone(&self.turn_state)),
responses_metadata.guardian_ticket.as_ref(),
) )
.await; .await;
if let Some(original_item_ids) = original_item_ids { if let Some(original_item_ids) = original_item_ids {
+6 -2
View File
@@ -844,7 +844,9 @@ async fn response_stream_records_last_model_feedback_ids() {
.set_default(); .set_default();
let api_stream = futures::stream::iter([ let api_stream = futures::stream::iter([
Ok(ResponseEvent::Created), Ok(ResponseEvent::Created {
guardian_ticket: None,
}),
Ok(ResponseEvent::Completed { Ok(ResponseEvent::Completed {
response_id: "resp-123".to_string(), response_id: "resp-123".to_string(),
token_usage: None, 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 backpressured_item_yielded = Arc::new(Notify::new());
let mut events = VecDeque::new(); let mut events = VecDeque::new();
for _ in 0..super::RESPONSE_STREAM_CHANNEL_CAPACITY { 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( events.push_back(ResponseEvent::OutputItemDone(output_message(
"1", "1",
+18
View File
@@ -85,6 +85,8 @@ const GUARDIAN_RECENT_ENTRY_LIMIT: usize = 40;
/// step-scoped things past their lifetime (like MCP bindings) /// step-scoped things past their lifetime (like MCP bindings)
#[derive(Clone)] #[derive(Clone)]
pub(crate) struct GuardianReviewContext { pub(crate) struct GuardianReviewContext {
/// Ticket from the response currently handled in this execution context.
pub(crate) guardian_ticket: Option<codex_protocol::guardian_ticket::GuardianTicket>,
turn: Arc<TurnContext>, turn: Arc<TurnContext>,
environments: TurnEnvironmentSnapshot, environments: TurnEnvironmentSnapshot,
// Model and reasoning inputs are carried for the follow-up Guardian and V2 migrations. // Model and reasoning inputs are carried for the follow-up Guardian and V2 migrations.
@@ -104,6 +106,11 @@ impl GuardianReviewContext {
settings: &ResolvedStepSettings, settings: &ResolvedStepSettings,
) -> Self { ) -> Self {
Self { Self {
guardian_ticket: turn
.extension_data
.get::<codex_protocol::guardian_ticket::GuardianTicket>()
.as_deref()
.cloned(),
environments: turn.environments.clone(), environments: turn.environments.clone(),
model_info: Arc::clone(&settings.model_info), model_info: Arc::clone(&settings.model_info),
reasoning_effort: settings.reasoning_effort().cloned(), reasoning_effort: settings.reasoning_effort().cloned(),
@@ -126,6 +133,12 @@ impl GuardianReviewContext {
impl From<&Arc<StepContext>> for GuardianReviewContext { impl From<&Arc<StepContext>> for GuardianReviewContext {
fn from(step: &Arc<StepContext>) -> Self { fn from(step: &Arc<StepContext>) -> Self {
Self { Self {
guardian_ticket: step
.turn
.extension_data
.get::<codex_protocol::guardian_ticket::GuardianTicket>()
.as_deref()
.cloned(),
turn: Arc::clone(&step.turn), turn: Arc::clone(&step.turn),
environments: step.environments.clone(), environments: step.environments.clone(),
model_info: Arc::clone(&step.settings.model_info), model_info: Arc::clone(&step.settings.model_info),
@@ -140,6 +153,11 @@ impl From<&Arc<StepContext>> for GuardianReviewContext {
impl From<Arc<TurnContext>> for GuardianReviewContext { impl From<Arc<TurnContext>> for GuardianReviewContext {
fn from(turn: Arc<TurnContext>) -> Self { fn from(turn: Arc<TurnContext>) -> Self {
Self { Self {
guardian_ticket: turn
.extension_data
.get::<codex_protocol::guardian_ticket::GuardianTicket>()
.as_deref()
.cloned(),
environments: turn.environments.clone(), environments: turn.environments.clone(),
model_info: Arc::clone(turn.model_info()), model_info: Arc::clone(turn.model_info()),
reasoning_effort: turn.reasoning_effort().cloned(), reasoning_effort: turn.reasoning_effort().cloned(),
@@ -1226,6 +1226,7 @@ async fn run_review_on_session(
..Default::default() ..Default::default()
}) })
.on_start(TurnStartOptions { .on_start(TurnStartOptions {
guardian_ticket: params.parent_context.guardian_ticket.clone(),
final_output_json_schema: Some(params.schema.clone()), final_output_json_schema: Some(params.schema.clone()),
service_tier: None, service_tier: None,
parent_turn_id: Some(parent_turn.sub_id.clone()), parent_turn_id: Some(parent_turn.sub_id.clone()),
+5
View File
@@ -58,6 +58,8 @@ pub(crate) const WORKSPACES_KEY: &str = "workspaces";
// App-server clients can specify additional metadata in the `responsesapi_client_metadata` param // 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. // when submitting a turn, but they must not override fields owned by core.
const RESERVED_METADATA_KEYS: &[&str] = &[ const RESERVED_METADATA_KEYS: &[&str] = &[
codex_protocol::guardian_ticket::GUARDIAN_TICKET_METADATA_KEY,
"guardian_ticket_requested",
INSTALLATION_ID_KEY, INSTALLATION_ID_KEY,
X_CODEX_INSTALLATION_ID_HEADER, X_CODEX_INSTALLATION_ID_HEADER,
SESSION_ID_KEY, SESSION_ID_KEY,
@@ -217,6 +219,8 @@ pub(crate) enum TurnToolSource {
/// truth. /// truth.
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub struct CodexResponsesMetadata { pub struct CodexResponsesMetadata {
/// Opaque runtime receipt; deliberately omitted from metadata projections.
pub(crate) guardian_ticket: Option<codex_protocol::guardian_ticket::GuardianTicket>,
pub(crate) installation_id: String, pub(crate) installation_id: String,
pub(crate) session_id: String, pub(crate) session_id: String,
pub(crate) thread_id: String, pub(crate) thread_id: String,
@@ -256,6 +260,7 @@ impl CodexResponsesMetadata {
window_id: String, window_id: String,
) -> Self { ) -> Self {
Self { Self {
guardian_ticket: None,
installation_id, installation_id,
session_id, session_id,
thread_id, thread_id,
+1
View File
@@ -140,6 +140,7 @@ pub(super) async fn spawn_review_thread(
extension_data.insert(parent_turn_context.skills_snapshot().as_ref().clone()); extension_data.insert(parent_turn_context.skills_snapshot().as_ref().clone());
let review_turn_context = TurnContext { let review_turn_context = TurnContext {
guardian_ticket: None,
sub_id: review_turn_id.clone(), sub_id: review_turn_id.clone(),
trace_id: current_span_trace_id(), trace_id: current_span_trace_id(),
realtime_active: parent_turn_context.realtime_active, realtime_active: parent_turn_context.realtime_active,
+1
View File
@@ -619,6 +619,7 @@ impl Session {
request_kind, request_kind,
); );
CodexResponsesMetadata { CodexResponsesMetadata {
guardian_ticket: turn_context.guardian_ticket.clone(),
window_number: Some(window_number), window_number: Some(window_number),
context_window_id: Some(context_window_id), context_window_id: Some(context_window_id),
history_ingest_requested: turn_context history_ingest_requested: turn_context
+9 -1
View File
@@ -1437,6 +1437,10 @@ async fn run_sampling_request(
let mut original_input = None; let mut original_input = None;
let mut executed_tool_calls_by_output = HashMap::new(); let mut executed_tool_calls_by_output = HashMap::new();
loop { loop {
// A retry must not lend a previous response's ticket to the next tool call.
turn_context
.extension_data
.remove::<codex_protocol::guardian_ticket::GuardianTicket>();
let prompt_input = if let Some(input) = initial_input.take() { let prompt_input = if let Some(input) = initial_input.take() {
input input
} else { } else {
@@ -2372,7 +2376,11 @@ async fn try_run_sampling_request(
record_turn_ttft_metric(&turn_context, &event).await; record_turn_ttft_metric(&turn_context, &event).await;
match event { match event {
ResponseEvent::Created => {} ResponseEvent::Created { guardian_ticket } => {
if let Some(ticket) = guardian_ticket {
turn_context.extension_data.insert(ticket);
}
}
ResponseEvent::OutputItemDone(mut item) => { ResponseEvent::OutputItemDone(mut item) => {
assign_missing_streamed_response_item_id(&mut item, active_item.as_ref()); 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 { if analytics_tool_call_ids.len() < MAX_ANALYTICS_TOOL_CALL_IDS_PER_RESPONSE {
@@ -185,6 +185,7 @@ impl std::fmt::Debug for TurnEnvironment {
/// Request options for one turn, separate from persistent thread settings. /// Request options for one turn, separate from persistent thread settings.
#[derive(Default)] #[derive(Default)]
pub(crate) struct NewTurnContextOptions { pub(crate) struct NewTurnContextOptions {
pub(crate) guardian_ticket: Option<codex_protocol::guardian_ticket::GuardianTicket>,
pub(crate) final_output_json_schema: Option<Value>, pub(crate) final_output_json_schema: Option<Value>,
pub(crate) cyber_access_program: Option<CyberAccessProgram>, pub(crate) cyber_access_program: Option<CyberAccessProgram>,
} }
@@ -192,6 +193,8 @@ pub(crate) struct NewTurnContextOptions {
/// The context needed for a single turn of the thread. /// The context needed for a single turn of the thread.
#[derive(Debug)] #[derive(Debug)]
pub struct TurnContext { pub struct TurnContext {
/// Receipt selected for this Guardian turn, separate from its own responses.
pub(crate) guardian_ticket: Option<codex_protocol::guardian_ticket::GuardianTicket>,
pub(crate) sub_id: String, pub(crate) sub_id: String,
pub(crate) trace_id: Option<String>, pub(crate) trace_id: Option<String>,
pub(crate) realtime_active: bool, pub(crate) realtime_active: bool,
@@ -537,6 +540,7 @@ impl TurnContext {
windows_sandbox_level: self.windows_sandbox_level, windows_sandbox_level: self.windows_sandbox_level,
available_models, available_models,
unified_exec_shell_mode: self.unified_exec_shell_mode.clone(), 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(), final_output_json_schema: self.final_output_json_schema.clone(),
dynamic_tools: self.dynamic_tools.clone(), dynamic_tools: self.dynamic_tools.clone(),
turn_metadata_state: self.turn_metadata_state.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())); let extension_data = Arc::new(codex_extension_api::ExtensionData::new(sub_id.clone()));
extension_data.insert(skills_snapshot); extension_data.insert(skills_snapshot);
TurnContext { TurnContext {
guardian_ticket: None,
sub_id, sub_id,
trace_id: current_span_trace_id(), trace_id: current_span_trace_id(),
realtime_active: false, realtime_active: false,
@@ -1024,6 +1029,7 @@ impl Session {
turn_context.extension_data.insert(trusted_plugin_roots); turn_context.extension_data.insert(trusted_plugin_roots);
turn_context.realtime_active = self.conversation.running_state().await.is_some(); 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; 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 { if turn_context.config.model_provider_id == codex_model_provider_info::OPENAI_PROVIDER_ID {
turn_context.cyber_access_program = options.cyber_access_program; turn_context.cyber_access_program = options.cyber_access_program;
+2
View File
@@ -121,6 +121,7 @@ impl PreparedTurnInputSettings {
kind: TurnStartKind, kind: TurnStartKind,
) -> CodexResult<Option<Arc<TurnContext>>> { ) -> CodexResult<Option<Arc<TurnContext>>> {
let TurnStartOptions { let TurnStartOptions {
guardian_ticket,
turn_trigger, turn_trigger,
final_output_json_schema, final_output_json_schema,
service_tier, service_tier,
@@ -138,6 +139,7 @@ impl PreparedTurnInputSettings {
updates.service_tier_for_turn = service_tier; updates.service_tier_for_turn = service_tier;
let options = NewTurnContextOptions { let options = NewTurnContextOptions {
guardian_ticket,
final_output_json_schema, final_output_json_schema,
cyber_access_program, cyber_access_program,
}; };
+1
View File
@@ -475,6 +475,7 @@ impl Session {
.new_turn_with_default_settings( .new_turn_with_default_settings(
sub_id, sub_id,
NewTurnContextOptions { NewTurnContextOptions {
guardian_ticket: start_options.guardian_ticket,
final_output_json_schema: start_options.final_output_json_schema, final_output_json_schema: start_options.final_output_json_schema,
cyber_access_program: start_options.cyber_access_program, cyber_access_program: start_options.cyber_access_program,
}, },
+1 -1
View File
@@ -385,7 +385,7 @@ fn response_event_records_turn_ttft(event: &ResponseEvent) -> bool {
| ResponseEvent::ReasoningSummaryDelta { .. } | ResponseEvent::ReasoningSummaryDelta { .. }
| ResponseEvent::ReasoningSummaryDone { .. } | ResponseEvent::ReasoningSummaryDone { .. }
| ResponseEvent::ReasoningContentDelta { .. } => true, | ResponseEvent::ReasoningContentDelta { .. } => true,
ResponseEvent::Created ResponseEvent::Created { .. }
| ResponseEvent::ServerModel(_) | ResponseEvent::ServerModel(_)
| ResponseEvent::ModelVerifications(_) | ResponseEvent::ModelVerifications(_)
| ResponseEvent::TurnModerationMetadata(_) | ResponseEvent::TurnModerationMetadata(_)
+3 -1
View File
@@ -29,7 +29,9 @@ async fn turn_timing_state_records_ttft_only_once_per_turn() {
state.mark_turn_started(Instant::now()).await; state.mark_turn_started(Instant::now()).await;
assert_eq!( assert_eq!(
state state
.record_ttft_for_response_event(&ResponseEvent::Created) .record_ttft_for_response_event(&ResponseEvent::Created {
guardian_ticket: None
})
.await, .await,
None None
); );
+68 -12
View File
@@ -2372,15 +2372,45 @@ async fn code_mode_wait_timeout_reconnects_on_next_exec() -> Result<()> {
Ok(()) Ok(())
} }
#[derive(Default)]
struct GuardianTicketObserver {
tickets: Mutex<Vec<(String, Option<String>)>>,
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::<codex_protocol::guardian_ticket::GuardianTicket>()
.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")] #[cfg_attr(windows, ignore = "no exec_command on Windows")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn code_mode_can_yield_and_resume_with_wait() -> Result<()> { async fn code_mode_can_yield_and_resume_with_wait() -> Result<()> {
skip_if_no_network!(Ok(())); skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await; let server = responses::start_mock_server().await;
let mut builder = test_codex().with_config(move |config| { let observer = Arc::new(GuardianTicketObserver::default());
let _ = config.features.enable(Feature::CodeMode); let mut extensions = ExtensionRegistryBuilder::<Config>::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 test = builder.build(&server).await?;
let phase_2_gate = test.workspace_path("code-mode-phase-2.ready"); let phase_2_gate = test.workspace_path("code-mode-phase-2.ready");
let phase_3_gate = test.workspace_path("code-mode-phase-3.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"); text("phase 1");
yield_control(); yield_control();
{phase_2_wait} {phase_2_wait}
text("phase 2"); text((await tools.exec_command({{cmd: "printf 'phase 2'"}})).output);
{phase_3_wait} {phase_3_wait}
text("phase 3"); text((await tools.exec_command({{cmd: "printf 'phase 3'"}})).output);
"# "#
); );
responses::mount_sse_once( responses::mount_sse_once(
&server, &server,
sse(vec![ 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_custom_tool_call("call-1", "exec", &code),
ev_completed("resp-1"), ev_completed("resp-1"),
]), ]),
@@ -2434,7 +2464,7 @@ text("phase 3");
responses::mount_sse_once( responses::mount_sse_once(
&server, &server,
sse(vec![ 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( responses::ev_function_call(
"call-2", "call-2",
"wait", "wait",
@@ -2456,8 +2486,11 @@ text("phase 3");
) )
.await; .await;
fs::write(&phase_2_gate, "ready")?; tokio::try_join!(test.submit_turn("wait again"), async {
test.submit_turn("wait again").await?; observer.wait_started.notified().await;
fs::write(&phase_2_gate, "ready")?;
anyhow::Ok(())
})?;
let second_request = second_completion.single_request(); let second_request = second_completion.single_request();
let second_items = function_tool_output_items(&second_request, "call-2"); let second_items = function_tool_output_items(&second_request, "call-2");
@@ -2478,7 +2511,7 @@ text("phase 3");
responses::mount_sse_once( responses::mount_sse_once(
&server, &server,
sse(vec![ 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( responses::ev_function_call(
"call-3", "call-3",
"wait", "wait",
@@ -2500,8 +2533,11 @@ text("phase 3");
) )
.await; .await;
fs::write(&phase_3_gate, "ready")?; tokio::try_join!(test.submit_turn("wait for completion"), async {
test.submit_turn("wait for completion").await?; observer.wait_started.notified().await;
fs::write(&phase_3_gate, "ready")?;
anyhow::Ok(())
})?;
let third_request = third_completion.single_request(); let third_request = third_completion.single_request();
let third_items = function_tool_output_items(&third_request, "call-3"); 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"); 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::<Vec<_>>();
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<_>>(),
vec![
("wait".to_owned(), Some("q".repeat(43))),
("wait".to_owned(), Some("r".repeat(43))),
]
);
Ok(()) Ok(())
} }
+83 -10
View File
@@ -130,15 +130,19 @@ impl TimeProvider for RecordingTimeProvider {
} }
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[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::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(), "/backend-api/codex", false, "/backend-api/codex/responses"; "chatgpt_uses_responses_by_default")] #[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(), "/backend-api/codex", true, "/backend-api/codex/guardian"; "chatgpt_uses_guardian_when_enabled")] #[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(), "/v1", true, "/v1/responses"; "custom_openai_url_uses_responses")] #[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( async fn guardian_session_inherits_parent_http_fallback(
auth: CodexAuth, auth: CodexAuth,
provider_name: &str,
base_path: &str, base_path: &str,
free_guardian: bool, free_guardian: bool,
expected_guardian_path: &str, expected_guardian_path: &str,
ticket_issued: bool,
) -> Result<()> { ) -> Result<()> {
skip_if_no_network!(Ok(())); skip_if_no_network!(Ok(()));
skip_if_wine_exec!( skip_if_wine_exec!(
@@ -154,10 +158,16 @@ async fn guardian_session_inherits_parent_http_fallback(
.mount_as_scoped(&server) .mount_as_scoped(&server)
.await; .await;
let parent_ticket = "p".repeat(43);
let responses = mount_sse_sequence( let responses = mount_sse_sequence(
&server, &server,
vec![ 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![ sse(vec![
json!({"type": "response.created", "response": {"id": "parent-tool", "headers": ticket_issued.then(|| json!({"x-codex-guardian-ticket": parent_ticket}))}}),
ev_function_call( ev_function_call(
"call", "call",
"exec_command", "exec_command",
@@ -175,6 +185,7 @@ async fn guardian_session_inherits_parent_http_fallback(
.await; .await;
let base_url = format!("{}{base_path}", server.uri()); let base_url = format!("{}{base_path}", server.uri());
let provider_name = provider_name.to_owned();
let mut builder = test_codex() let mut builder = test_codex()
.with_auth(auth) .with_auth(auth)
.with_pre_build_hook(move |home| { .with_pre_build_hook(move |home| {
@@ -186,7 +197,9 @@ async fn guardian_session_inherits_parent_http_fallback(
}) })
.with_config(move |config| { .with_config(move |config| {
config.model_provider.base_url = Some(base_url); config.model_provider.base_url = Some(base_url);
config.model_provider.name = provider_name;
config.model_provider.supports_websockets = true; config.model_provider.supports_websockets = true;
config.model_provider.stream_max_retries = Some(1);
config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
config.approvals_reviewer = ApprovalsReviewer::User; config.approvals_reviewer = ApprovalsReviewer::User;
}); });
@@ -232,6 +245,34 @@ async fn guardian_session_inherits_parent_http_fallback(
}) })
.expect("Guardian reviewer inference request"); .expect("Guardian reviewer inference request");
assert_eq!(guardian_request.path(), expected_guardian_path); 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(()) Ok(())
} }
@@ -348,15 +389,18 @@ async fn guardian_review_resends_full_transcript_after_reviewer_context_rollover
} }
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[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::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(), "codex-auto-review"; "chatgpt_uses_codex_auto_review")] #[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( async fn guardian_session_prewarms_and_is_reused_for_first_review(
auth: CodexAuth, auth: CodexAuth,
free_guardian: bool,
expected_model: &str, expected_model: &str,
) -> Result<()> { ) -> Result<()> {
skip_if_no_network!(Ok(())); skip_if_no_network!(Ok(()));
let uses_codex_backend = auth.uses_codex_backend(); 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 bundled_models = codex_models_manager::bundled_models_response()?.models;
let catalog_auto_review = bundled_models let catalog_auto_review = bundled_models
.iter() .iter()
@@ -389,11 +433,12 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review(
"justification": "Exercise Guardian approval routing.", "justification": "Exercise Guardian approval routing.",
}) })
.to_string(); .to_string();
let parent_ticket = "w".repeat(43);
let server = start_websocket_server(vec![ let server = start_websocket_server(vec![
vec![vec![ev_response_created("warm-1"), ev_completed("warm-1")]], 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("warm-2"), ev_completed("warm-2")]],
vec![vec![ 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_function_call("approval-call", "exec_command", &tool_args),
ev_completed("approval-request"), 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 backend_base_url = format!("{}/backend-api/codex", server.uri());
let mut builder = test_codex() let mut builder = test_codex()
.with_auth(auth) .with_auth(auth)
.with_pre_build_hook(|home| { .with_pre_build_hook(move |home| {
fs::write( fs::write(
home.join("config.toml"), 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"); .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"); .expect("reviewed parent turn id");
assert_parent_turn(&parent_request, /*expected*/ None)?; assert_parent_turn(&parent_request, /*expected*/ None)?;
assert_parent_turn(&guardian_review, Some(parent_turn_id))?; 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] { for request in [&parent_request, &guardian_review] {
assert_root_turn(request, Some(parent_turn_id))?; 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)]); assert_eq!(guardian_context_windows, vec![Some(258_400)]);
for handshake in server.handshakes() { for handshake in server.handshakes() {
let is_guardian = handshake.header("x-openai-subagent").as_deref() == Some("guardian"); 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!( assert_eq!(
handshake.uri(), handshake.uri(),
if uses_guardian_endpoint { if uses_guardian_endpoint {
@@ -506,6 +506,11 @@ impl GuardianV2Extension {
let thread_id = input.thread_store.level_id().to_owned(); let thread_id = input.thread_store.level_id().to_owned();
let turn_id = input.turn_id.to_owned(); let turn_id = input.turn_id.to_owned();
let root_turn_id = input.root_turn_id.map(str::to_owned); let root_turn_id = input.root_turn_id.map(str::to_owned);
let guardian_ticket = input
.turn_store
.get::<codex_protocol::guardian_ticket::GuardianTicket>()
.as_deref()
.cloned();
let thread_context: Result<_, String> = async { let thread_context: Result<_, String> = async {
let parsed_thread_id = let parsed_thread_id =
ThreadId::from_string(&thread_id).map_err(|error| error.to_string())?; 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 instructions = guardian_config.render_classifier_instructions(policy);
let output = match sampler let output = match sampler
.sample(LunaSamplingRequest { .sample(LunaSamplingRequest {
guardian_ticket,
instructions, instructions,
trusted_review_evidence, trusted_review_evidence,
trusted_tool_context, trusted_tool_context,
@@ -90,6 +90,8 @@ pub struct LunaSamplerConfig {
/// One tool-less Luna classification request over an already-open connection. /// One tool-less Luna classification request over an already-open connection.
pub struct LunaSamplingRequest { pub struct LunaSamplingRequest {
/// Runtime receipt of the response handling the classified tool.
pub guardian_ticket: Option<codex_protocol::guardian_ticket::GuardianTicket>,
/// Trusted instructions describing the requested classification. /// Trusted instructions describing the requested classification.
pub instructions: String, pub instructions: String,
/// Host-supplied Guardian reviews isolated from untrusted transcript entries. /// 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. // A classification is its own inference turn; retries keep that identity.
let turn_id = Uuid::now_v7().to_string(); let turn_id = Uuid::now_v7().to_string();
let guardian_ticket = request.guardian_ticket;
let parent_turn_id = request.parent_turn_id; let parent_turn_id = request.parent_turn_id;
let root_turn_id = request.root_turn_id; let root_turn_id = request.root_turn_id;
let mut input = vec![ let mut input = vec![
@@ -634,6 +637,7 @@ impl LunaSampler {
ResponsesWsRequest::ResponseCreate((&request).into()), ResponsesWsRequest::ResponseCreate((&request).into()),
/*connection_reused*/ true, /*connection_reused*/ true,
/*turn_state*/ None, /*turn_state*/ None,
guardian_ticket.as_ref(),
) )
.await .await
{ {
@@ -192,6 +192,7 @@ async fn connect_sampler(config: LunaSamplerConfig) -> Result<LunaSampler> {
fn sample_request(parent_turn_id: &str) -> LunaSamplingRequest { fn sample_request(parent_turn_id: &str) -> LunaSamplingRequest {
LunaSamplingRequest { LunaSamplingRequest {
guardian_ticket: None,
instructions: "Return high for high risk or low for low risk.".to_owned(), instructions: "Return high for high risk or low for low risk.".to_owned(),
trusted_review_evidence: Vec::new(), trusted_review_evidence: Vec::new(),
trusted_tool_context: None, trusted_tool_context: None,
@@ -446,6 +447,7 @@ async fn preconnected_sampler_reuses_authenticated_websocket_for_classifications
let first = sampler let first = sampler
.sample(LunaSamplingRequest { .sample(LunaSamplingRequest {
guardian_ticket: None,
instructions: "Return high for high risk or low for low risk.".to_owned(), instructions: "Return high for high risk or low for low risk.".to_owned(),
trusted_review_evidence: Vec::new(), trusted_review_evidence: Vec::new(),
trusted_tool_context: None, trusted_tool_context: None,
@@ -477,6 +479,7 @@ async fn preconnected_sampler_reuses_authenticated_websocket_for_classifications
manager.refresh_token_from_authority().await?; manager.refresh_token_from_authority().await?;
let second = sampler let second = sampler
.sample(LunaSamplingRequest { .sample(LunaSamplingRequest {
guardian_ticket: None,
instructions: "Return high for high risk or low for low risk.".to_owned(), instructions: "Return high for high risk or low for low risk.".to_owned(),
trusted_review_evidence: Vec::new(), trusted_review_evidence: Vec::new(),
trusted_tool_context: None, trusted_tool_context: None,
@@ -652,6 +655,7 @@ async fn sampler_returns_classification_token_before_terminal_response_events()
let output = tokio::time::timeout( let output = tokio::time::timeout(
Duration::from_secs(2), Duration::from_secs(2),
sampler.sample(LunaSamplingRequest { sampler.sample(LunaSamplingRequest {
guardian_ticket: None,
instructions: "Return high for high risk or low for low risk.".to_owned(), instructions: "Return high for high risk or low for low risk.".to_owned(),
trusted_review_evidence: Vec::new(), trusted_review_evidence: Vec::new(),
trusted_tool_context: None, trusted_tool_context: None,
@@ -1113,3 +1117,54 @@ async fn sampler_limits_transient_recovery_attempts() -> Result<()> {
Ok(()) 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(())
}
+24 -2
View File
@@ -11,7 +11,7 @@ use std::time::Duration;
/// Clones share the encoded allocation. Internally, the body can also hold the /// Clones share the encoded allocation. Internally, the body can also hold the
/// final compressed wire bytes while retaining the original JSON only when /// final compressed wire bytes while retaining the original JSON only when
/// request-body trace logging is enabled. /// request-body trace logging is enabled.
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Clone, PartialEq, Eq)]
pub struct EncodedJsonBody { pub struct EncodedJsonBody {
bytes: Bytes, bytes: Bytes,
trace_bytes: Option<Bytes>, trace_bytes: Option<Bytes>,
@@ -33,11 +33,27 @@ impl EncodedJsonBody {
&self.bytes &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"<redacted>"));
self
}
pub(crate) fn trace_bytes(&self) -> &[u8] { pub(crate) fn trace_bytes(&self) -> &[u8] {
self.trace_bytes.as_ref().unwrap_or(&self.bytes) 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)] #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum RequestCompression { pub enum RequestCompression {
#[default] #[default]
@@ -120,7 +136,13 @@ impl Request {
self.body, self.body,
Some(RequestBody::Json(_) | RequestBody::EncodedJson(_)) 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) && tracing::enabled!(target: "codex_http_client::transport", tracing::Level::TRACE)
{ {
match self.body.as_ref() { match self.body.as_ref() {
@@ -1,4 +1,7 @@
use super::*; use super::*;
use crate::request::EncodedJsonBody;
use crate::request::RequestCompression;
use pretty_assertions::assert_eq;
use serde_json::json; use serde_json::json;
use std::io::Write; use std::io::Write;
use std::sync::Arc; use std::sync::Arc;
@@ -28,6 +31,35 @@ async fn disabled_request_logging_suppresses_transport_url_and_body() {
assert!(!logs.contains("body-secret")); 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), "<redacted>");
for _ in 0..2 {
request = request.clone().into_prepared().unwrap();
assert_eq!(request_body_for_trace(&request), "<redacted>");
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::<serde_json::Value>(&decoded).unwrap(),
payload
);
}
}
}
#[tokio::test] #[tokio::test]
async fn connection_failures_are_classified_without_exposing_request_urls() { async fn connection_failures_are_classified_without_exposing_request_urls() {
let unavailable_server = let unavailable_server =
@@ -1270,7 +1270,7 @@ impl SessionTelemetry {
fn responses_type(event: &ResponseEvent) -> String { fn responses_type(event: &ResponseEvent) -> String {
match event { match event {
ResponseEvent::Created => "created".into(), ResponseEvent::Created { .. } => "created".into(),
ResponseEvent::OutputItemDone(item) | ResponseEvent::OutputItemAdded(item) => { ResponseEvent::OutputItemDone(item) | ResponseEvent::OutputItemAdded(item) => {
SessionTelemetry::responses_item_type(item) SessionTelemetry::responses_item_type(item)
} }
+37
View File
@@ -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<Self> {
(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])")
}
}
+1
View File
@@ -25,6 +25,7 @@ pub mod dynamic_tools;
mod environment; mod environment;
pub mod error; pub mod error;
pub mod exec_output; pub mod exec_output;
pub mod guardian_ticket;
pub mod items; pub mod items;
mod legacy_events; mod legacy_events;
pub mod local_media; pub mod local_media;
+2
View File
@@ -158,6 +158,8 @@ pub enum CyberAccessProgram {
/// child input, Core also compares root lineage to detect ambiguity. /// child input, Core also compares root lineage to detect ambiguity.
#[derive(Clone, Debug, Default)] #[derive(Clone, Debug, Default)]
pub struct TurnStartOptions { pub struct TurnStartOptions {
/// Parent inference receipt for an internal Guardian turn. Never persisted.
pub guardian_ticket: Option<crate::guardian_ticket::GuardianTicket>,
/// Source classification for the caller that starts a new turn. /// Source classification for the caller that starts a new turn.
/// Ignored when the submitted input steers an active turn. /// Ignored when the submitted input steers an active turn.
pub turn_trigger: Option<String>, pub turn_trigger: Option<String>,