mirror of
https://github.com/openai/codex.git
synced 2026-10-28 18:58:46 +00:00
Replace Guardian tickets with parent response IDs (#43002)
## What changed Capture the ID from `response.created` and pass it as `parent_response_id` in Guardian review and classifier request metadata. Clear the current response ID before sampling retries so tool reviews cannot inherit a failed response's ID. Request Guardian credits with `guardian_credits_requested` on eligible parent requests. Remove the ticket protocol, transport attachment, and ticket-specific body redaction. Keep ordinary app-server client `parent_response_id` values inside `x-codex-turn-metadata` rather than promoting them to Guardian request metadata. ## Testing Update coverage for HTTP fallback, WebSocket review reuse, missing response IDs, classifier transport retries, and code-mode yield/resume attribution. Extend app-server tests to verify client metadata placement over HTTP and WebSocket. GitOrigin-RevId: c1665e56382ecec8d260fc2ca0ecd5e20b0b2391
This commit is contained in:
@@ -73,6 +73,10 @@ async fn turn_start_forwards_client_metadata_to_responses_request_v2() -> Result
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(thread_req)).await??;
|
||||
|
||||
let client_metadata = HashMap::from([
|
||||
(
|
||||
"parent_response_id".to_string(),
|
||||
"client-correlation".to_string(),
|
||||
),
|
||||
("fiber_run_id".to_string(), "fiber-start-123".to_string()),
|
||||
("origin".to_string(), "gaas".to_string()),
|
||||
(
|
||||
@@ -109,6 +113,20 @@ async fn turn_start_forwards_client_metadata_to_responses_request_v2() -> Result
|
||||
.as_deref()
|
||||
.map(parse_json_header)
|
||||
.expect("x-codex-turn-metadata header should be present");
|
||||
let body = request.body_json();
|
||||
let body_turn_metadata = parse_json_header(
|
||||
body["client_metadata"]["x-codex-turn-metadata"]
|
||||
.as_str()
|
||||
.expect("turn metadata"),
|
||||
);
|
||||
assert_eq!(
|
||||
(
|
||||
metadata["parent_response_id"].as_str(),
|
||||
body_turn_metadata["parent_response_id"].as_str(),
|
||||
body["client_metadata"].get("parent_response_id"),
|
||||
),
|
||||
(Some("client-correlation"), Some("client-correlation"), None),
|
||||
);
|
||||
assert_eq!(metadata["fiber_run_id"].as_str(), Some("fiber-start-123"));
|
||||
assert_eq!(metadata["origin"].as_str(), Some("gaas"));
|
||||
assert_eq!(metadata["thread_source"].as_str(), Some("automation"));
|
||||
@@ -603,6 +621,10 @@ async fn turn_start_forwards_client_metadata_to_responses_websocket_request_body
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(thread_req)).await??;
|
||||
|
||||
let client_metadata = HashMap::from([
|
||||
(
|
||||
"parent_response_id".to_string(),
|
||||
"client-correlation".to_string(),
|
||||
),
|
||||
("fiber_run_id".to_string(), "fiber-start-123".to_string()),
|
||||
("origin".to_string(), "gaas".to_string()),
|
||||
]);
|
||||
@@ -646,6 +668,13 @@ async fn turn_start_forwards_client_metadata_to_responses_websocket_request_body
|
||||
.as_str()
|
||||
.map(parse_json_header)
|
||||
.expect("websocket x-codex-turn-metadata client metadata should be present");
|
||||
assert_eq!(
|
||||
(
|
||||
metadata["parent_response_id"].as_str(),
|
||||
request["client_metadata"].get("parent_response_id"),
|
||||
),
|
||||
(Some("client-correlation"), None),
|
||||
);
|
||||
assert_eq!(metadata["fiber_run_id"].as_str(), Some("fiber-start-123"));
|
||||
assert_eq!(metadata["origin"].as_str(), Some("gaas"));
|
||||
assert_eq!(metadata["thread_source"].as_str(), Some("automation"));
|
||||
|
||||
@@ -94,11 +94,15 @@ pub struct MemorySummarizeOutput {
|
||||
pub memory_summary: String,
|
||||
}
|
||||
|
||||
/// The server response currently being handled, shared with tool-review extensions.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct ResponseId(pub String);
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum ResponseEvent {
|
||||
Created {
|
||||
/// Runtime-only receipt for this exact server response.
|
||||
guardian_ticket: Option<codex_protocol::guardian_ticket::GuardianTicket>,
|
||||
/// Existing server response ID, when supplied by the stream.
|
||||
response_id: Option<String>,
|
||||
},
|
||||
SafetyBuffering(SafetyBuffering),
|
||||
OutputItemDone(ResponseItem),
|
||||
|
||||
@@ -60,8 +60,6 @@ pub struct ResponsesOptions {
|
||||
pub extra_headers: HeaderMap,
|
||||
pub compression: Compression,
|
||||
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> {
|
||||
@@ -103,7 +101,7 @@ impl<T: HttpTransport> ResponsesClient<T> {
|
||||
)]
|
||||
pub async fn stream_request(
|
||||
&self,
|
||||
mut request: ResponsesApiRequest,
|
||||
request: ResponsesApiRequest,
|
||||
options: ResponsesOptions,
|
||||
) -> Result<ResponseStream, ApiError> {
|
||||
let ResponsesOptions {
|
||||
@@ -113,19 +111,9 @@ impl<T: HttpTransport> ResponsesClient<T> {
|
||||
extra_headers,
|
||||
compression,
|
||||
turn_state,
|
||||
guardian_ticket,
|
||||
} = options;
|
||||
crate::guardian_ticket::attach(
|
||||
&mut request.client_metadata,
|
||||
guardian_ticket.as_ref(),
|
||||
self.endpoint,
|
||||
);
|
||||
|
||||
let mut body = EncodedJsonBody::encode(&request)
|
||||
let 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 {
|
||||
|
||||
@@ -234,10 +234,9 @@ impl ResponsesWebsocketConnection {
|
||||
)]
|
||||
pub async fn stream_request(
|
||||
&self,
|
||||
mut request: ResponsesWsRequest<'_>,
|
||||
request: ResponsesWsRequest<'_>,
|
||||
connection_reused: bool,
|
||||
turn_state: Option<Arc<OnceLock<String>>>,
|
||||
guardian_ticket: Option<&codex_protocol::guardian_ticket::GuardianTicket>,
|
||||
) -> Result<ResponseStream, ApiError> {
|
||||
let (tx_event, rx_event) =
|
||||
mpsc::channel::<std::result::Result<ResponseEvent, ApiError>>(1600);
|
||||
@@ -271,12 +270,6 @@ 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();
|
||||
|
||||
@@ -1,33 +0,0 @@
|
||||
//! 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;
|
||||
@@ -1,60 +0,0 @@
|
||||
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
|
||||
})
|
||||
));
|
||||
}
|
||||
}
|
||||
@@ -4,7 +4,6 @@ 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;
|
||||
@@ -40,6 +39,7 @@ pub use crate::common::ReasoningContext;
|
||||
pub use crate::common::ReasoningSummaryDelivery;
|
||||
pub use crate::common::ResponseCreateWsRequest;
|
||||
pub use crate::common::ResponseEvent;
|
||||
pub use crate::common::ResponseId;
|
||||
pub use crate::common::ResponseStream;
|
||||
pub use crate::common::ResponsesApiRequest;
|
||||
pub use crate::common::ResponsesApiTools;
|
||||
|
||||
@@ -9,8 +9,6 @@ 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;
|
||||
@@ -409,17 +407,11 @@ pub fn process_responses_event(
|
||||
}
|
||||
"response.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 }));
|
||||
let response_id = response
|
||||
.get("id")
|
||||
.and_then(Value::as_str)
|
||||
.map(str::to_owned);
|
||||
return Ok(Some(ResponseEvent::Created { response_id }));
|
||||
}
|
||||
}
|
||||
"response.failed" => {
|
||||
@@ -1630,9 +1622,7 @@ mod tests {
|
||||
);
|
||||
assert_matches!(
|
||||
&events[1],
|
||||
ResponseEvent::Created {
|
||||
guardian_ticket: None
|
||||
}
|
||||
ResponseEvent::Created { response_id: Some(id) } if id == "resp-1"
|
||||
);
|
||||
assert_matches!(
|
||||
&events[2],
|
||||
|
||||
@@ -331,8 +331,7 @@ 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))
|
||||
.with_endpoint(ResponsesEndpoint::Guardian);
|
||||
let client = ResponsesClient::new(transport, provider("openai"), Arc::new(NoAuth));
|
||||
let request = ResponsesApiRequest {
|
||||
model: "gpt-test".into(),
|
||||
instructions: "Say hi".into(),
|
||||
@@ -357,24 +356,14 @@ async fn responses_client_stream_request_preserves_item_ids() -> Result<()> {
|
||||
client_metadata: None,
|
||||
access_programs: None,
|
||||
};
|
||||
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 expected = serde_json::to_value(&request)?;
|
||||
|
||||
let _stream = client
|
||||
.stream_request(
|
||||
request,
|
||||
ResponsesOptions {
|
||||
guardian_ticket: Some(ticket),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.stream_request(request, ResponsesOptions::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");
|
||||
@@ -589,7 +578,6 @@ async fn azure_store_sends_ids_and_headers() -> Result<()> {
|
||||
extra_headers,
|
||||
compression: Compression::None,
|
||||
turn_state: None,
|
||||
guardian_ticket: None,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
|
||||
@@ -1104,14 +1104,24 @@ impl ModelClient {
|
||||
&& provider.aws.is_none()
|
||||
}
|
||||
|
||||
fn set_guardian_ticket_request(
|
||||
fn set_guardian_metadata(
|
||||
&self,
|
||||
metadata: &mut Option<HashMap<String, String>>,
|
||||
parent_response_id: Option<&str>,
|
||||
auth: Option<&CodexAuth>,
|
||||
endpoint: ResponsesEndpoint,
|
||||
) {
|
||||
if let Some(metadata) = metadata.as_mut() {
|
||||
metadata.remove("guardian_ticket_requested");
|
||||
metadata.remove("guardian_credits_requested");
|
||||
metadata.remove("parent_response_id");
|
||||
}
|
||||
if endpoint == ResponsesEndpoint::Guardian
|
||||
&& let Some(parent_response_id) = parent_response_id
|
||||
{
|
||||
metadata.get_or_insert_with(HashMap::new).insert(
|
||||
"parent_response_id".to_owned(),
|
||||
parent_response_id.to_owned(),
|
||||
);
|
||||
}
|
||||
if self.free_guardian_enabled
|
||||
&& endpoint == ResponsesEndpoint::Responses
|
||||
@@ -1125,7 +1135,7 @@ impl ModelClient {
|
||||
{
|
||||
metadata
|
||||
.get_or_insert_with(HashMap::new)
|
||||
.insert("guardian_ticket_requested".to_owned(), "true".to_owned());
|
||||
.insert("guardian_credits_requested".to_owned(), "true".to_owned());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1351,7 +1361,6 @@ impl ModelClientSession {
|
||||
},
|
||||
compression,
|
||||
turn_state: Some(Arc::clone(&self.turn_state)),
|
||||
guardian_ticket: responses_metadata.guardian_ticket.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1641,8 +1650,9 @@ impl ModelClientSession {
|
||||
service_tier.clone(),
|
||||
responses_metadata,
|
||||
)?;
|
||||
self.client.set_guardian_ticket_request(
|
||||
self.client.set_guardian_metadata(
|
||||
&mut request.client_metadata,
|
||||
responses_metadata.parent_response_id.as_deref(),
|
||||
client_setup.auth.as_ref(),
|
||||
endpoint,
|
||||
);
|
||||
@@ -1900,8 +1910,9 @@ impl ModelClientSession {
|
||||
),
|
||||
..ResponseCreateWsRequest::from(&request)
|
||||
};
|
||||
self.client.set_guardian_ticket_request(
|
||||
self.client.set_guardian_metadata(
|
||||
&mut ws_payload.client_metadata,
|
||||
responses_metadata.parent_response_id.as_deref(),
|
||||
client_setup.auth.as_ref(),
|
||||
endpoint,
|
||||
);
|
||||
@@ -1922,7 +1933,6 @@ 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 {
|
||||
|
||||
@@ -844,9 +844,7 @@ async fn response_stream_records_last_model_feedback_ids() {
|
||||
.set_default();
|
||||
|
||||
let api_stream = futures::stream::iter([
|
||||
Ok(ResponseEvent::Created {
|
||||
guardian_ticket: None,
|
||||
}),
|
||||
Ok(ResponseEvent::Created { response_id: None }),
|
||||
Ok(ResponseEvent::Completed {
|
||||
response_id: "resp-123".to_string(),
|
||||
token_usage: None,
|
||||
@@ -1062,9 +1060,7 @@ 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 {
|
||||
guardian_ticket: None,
|
||||
});
|
||||
events.push_back(ResponseEvent::Created { response_id: None });
|
||||
}
|
||||
events.push_back(ResponseEvent::OutputItemDone(output_message(
|
||||
"1",
|
||||
|
||||
@@ -88,8 +88,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<codex_protocol::guardian_ticket::GuardianTicket>,
|
||||
/// The response currently handled in this execution context.
|
||||
pub(crate) parent_response_id: Option<String>,
|
||||
turn: Arc<TurnContext>,
|
||||
environments: TurnEnvironmentSnapshot,
|
||||
// Model and reasoning inputs are carried for the follow-up Guardian and V2 migrations.
|
||||
@@ -109,11 +109,10 @@ impl GuardianReviewContext {
|
||||
settings: &ResolvedStepSettings,
|
||||
) -> Self {
|
||||
Self {
|
||||
guardian_ticket: turn
|
||||
parent_response_id: turn
|
||||
.extension_data
|
||||
.get::<codex_protocol::guardian_ticket::GuardianTicket>()
|
||||
.as_deref()
|
||||
.cloned(),
|
||||
.get::<codex_api::ResponseId>()
|
||||
.map(|id| id.0.clone()),
|
||||
environments: turn.environments.clone(),
|
||||
model_info: Arc::clone(&settings.model_info),
|
||||
reasoning_effort: settings.reasoning_effort().cloned(),
|
||||
@@ -136,12 +135,11 @@ impl GuardianReviewContext {
|
||||
impl From<&Arc<StepContext>> for GuardianReviewContext {
|
||||
fn from(step: &Arc<StepContext>) -> Self {
|
||||
Self {
|
||||
guardian_ticket: step
|
||||
parent_response_id: step
|
||||
.turn
|
||||
.extension_data
|
||||
.get::<codex_protocol::guardian_ticket::GuardianTicket>()
|
||||
.as_deref()
|
||||
.cloned(),
|
||||
.get::<codex_api::ResponseId>()
|
||||
.map(|id| id.0.clone()),
|
||||
turn: Arc::clone(&step.turn),
|
||||
environments: step.environments.clone(),
|
||||
model_info: Arc::clone(&step.settings.model_info),
|
||||
@@ -156,11 +154,10 @@ impl From<&Arc<StepContext>> for GuardianReviewContext {
|
||||
impl From<Arc<TurnContext>> for GuardianReviewContext {
|
||||
fn from(turn: Arc<TurnContext>) -> Self {
|
||||
Self {
|
||||
guardian_ticket: turn
|
||||
parent_response_id: turn
|
||||
.extension_data
|
||||
.get::<codex_protocol::guardian_ticket::GuardianTicket>()
|
||||
.as_deref()
|
||||
.cloned(),
|
||||
.get::<codex_api::ResponseId>()
|
||||
.map(|id| id.0.clone()),
|
||||
environments: turn.environments.clone(),
|
||||
model_info: Arc::clone(turn.model_info()),
|
||||
reasoning_effort: turn.reasoning_effort().cloned(),
|
||||
|
||||
@@ -1300,8 +1300,14 @@ async fn run_review_on_session(
|
||||
}),
|
||||
..Default::default()
|
||||
})
|
||||
.with_responses_metadata(
|
||||
params
|
||||
.parent_context
|
||||
.parent_response_id
|
||||
.as_ref()
|
||||
.map(|id| HashMap::from([("parent_response_id".to_owned(), id.clone())])),
|
||||
)
|
||||
.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()),
|
||||
|
||||
@@ -58,8 +58,7 @@ 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",
|
||||
"guardian_credits_requested",
|
||||
INSTALLATION_ID_KEY,
|
||||
X_CODEX_INSTALLATION_ID_HEADER,
|
||||
SESSION_ID_KEY,
|
||||
@@ -219,8 +218,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<codex_protocol::guardian_ticket::GuardianTicket>,
|
||||
/// Guardian parent reference; projected only onto a Guardian request.
|
||||
pub(crate) parent_response_id: Option<String>,
|
||||
pub(crate) installation_id: String,
|
||||
pub(crate) session_id: String,
|
||||
pub(crate) thread_id: String,
|
||||
@@ -260,7 +259,7 @@ impl CodexResponsesMetadata {
|
||||
window_id: String,
|
||||
) -> Self {
|
||||
Self {
|
||||
guardian_ticket: None,
|
||||
parent_response_id: None,
|
||||
installation_id,
|
||||
session_id,
|
||||
thread_id,
|
||||
|
||||
@@ -137,7 +137,6 @@ 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,
|
||||
|
||||
@@ -619,7 +619,6 @@ 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
|
||||
|
||||
@@ -1437,10 +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.
|
||||
// A retry must not attribute the next tool call to the previous response.
|
||||
turn_context
|
||||
.extension_data
|
||||
.remove::<codex_protocol::guardian_ticket::GuardianTicket>();
|
||||
.remove::<codex_api::ResponseId>();
|
||||
let prompt_input = if let Some(input) = initial_input.take() {
|
||||
input
|
||||
} else {
|
||||
@@ -2376,9 +2376,11 @@ async fn try_run_sampling_request(
|
||||
record_turn_ttft_metric(&turn_context, &event).await;
|
||||
|
||||
match event {
|
||||
ResponseEvent::Created { guardian_ticket } => {
|
||||
if let Some(ticket) = guardian_ticket {
|
||||
turn_context.extension_data.insert(ticket);
|
||||
ResponseEvent::Created { response_id } => {
|
||||
if let Some(response_id) = response_id {
|
||||
turn_context
|
||||
.extension_data
|
||||
.insert(codex_api::ResponseId(response_id));
|
||||
}
|
||||
}
|
||||
ResponseEvent::OutputItemDone(mut item) => {
|
||||
|
||||
@@ -185,7 +185,6 @@ 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<codex_protocol::guardian_ticket::GuardianTicket>,
|
||||
pub(crate) final_output_json_schema: Option<Value>,
|
||||
pub(crate) cyber_access_program: Option<CyberAccessProgram>,
|
||||
}
|
||||
@@ -193,8 +192,6 @@ 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<codex_protocol::guardian_ticket::GuardianTicket>,
|
||||
pub(crate) sub_id: String,
|
||||
pub(crate) trace_id: Option<String>,
|
||||
pub(crate) realtime_active: bool,
|
||||
@@ -540,7 +537,6 @@ 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(),
|
||||
@@ -785,7 +781,6 @@ 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,
|
||||
@@ -1040,7 +1035,6 @@ 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;
|
||||
|
||||
@@ -121,7 +121,6 @@ impl PreparedTurnInputSettings {
|
||||
kind: TurnStartKind,
|
||||
) -> CodexResult<Option<Arc<TurnContext>>> {
|
||||
let TurnStartOptions {
|
||||
guardian_ticket,
|
||||
turn_trigger,
|
||||
final_output_json_schema,
|
||||
service_tier,
|
||||
@@ -139,7 +138,6 @@ impl PreparedTurnInputSettings {
|
||||
updates.service_tier_for_turn = service_tier;
|
||||
|
||||
let options = NewTurnContextOptions {
|
||||
guardian_ticket,
|
||||
final_output_json_schema,
|
||||
cyber_access_program,
|
||||
};
|
||||
|
||||
@@ -473,7 +473,6 @@ 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,
|
||||
},
|
||||
|
||||
@@ -393,6 +393,14 @@ impl TurnMetadataState {
|
||||
.read()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.clone();
|
||||
// Extract Guardian's internal parent before filtering configured metadata keys.
|
||||
// Ordinary app-server client metadata stays in `extra`.
|
||||
let parent_response_id =
|
||||
if self.subagent_header.as_deref() == Some(crate::guardian::GUARDIAN_REVIEWER_NAME) {
|
||||
extra.remove("parent_response_id")
|
||||
} else {
|
||||
None
|
||||
};
|
||||
for key in self
|
||||
.responses_api_metadata
|
||||
.read()
|
||||
@@ -402,6 +410,7 @@ impl TurnMetadataState {
|
||||
extra.remove(key);
|
||||
}
|
||||
let mut metadata = CodexResponsesMetadata {
|
||||
parent_response_id,
|
||||
turn_id: Some(self.turn_id.clone()),
|
||||
agent_name: Some(self.agent_name.clone()),
|
||||
forked_from_thread_id: self.forked_from_thread_id,
|
||||
|
||||
@@ -29,9 +29,7 @@ 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 {
|
||||
guardian_ticket: None
|
||||
})
|
||||
.record_ttft_for_response_event(&ResponseEvent::Created { response_id: None })
|
||||
.await,
|
||||
None
|
||||
);
|
||||
|
||||
@@ -2373,21 +2373,21 @@ async fn code_mode_wait_timeout_reconnects_on_next_exec() -> Result<()> {
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct GuardianTicketObserver {
|
||||
tickets: Mutex<Vec<(String, Option<String>)>>,
|
||||
struct ResponseIdObserver {
|
||||
response_ids: Mutex<Vec<(String, Option<String>)>>,
|
||||
wait_started: tokio::sync::Notify,
|
||||
}
|
||||
|
||||
impl ToolLifecycleContributor for GuardianTicketObserver {
|
||||
impl ToolLifecycleContributor for ResponseIdObserver {
|
||||
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((
|
||||
self.response_ids.lock().unwrap().push((
|
||||
input.tool_name.name.clone(),
|
||||
input
|
||||
.turn_store
|
||||
.get::<codex_protocol::guardian_ticket::GuardianTicket>()
|
||||
.map(|ticket| ticket.as_str().to_owned()),
|
||||
.get::<codex_api::ResponseId>()
|
||||
.map(|id| id.0.clone()),
|
||||
));
|
||||
if input.tool_name.name == "wait" {
|
||||
self.wait_started.notify_one();
|
||||
@@ -2403,7 +2403,7 @@ async fn code_mode_can_yield_and_resume_with_wait() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
let server = responses::start_mock_server().await;
|
||||
let observer = Arc::new(GuardianTicketObserver::default());
|
||||
let observer = Arc::new(ResponseIdObserver::default());
|
||||
let mut extensions = ExtensionRegistryBuilder::<Config>::new();
|
||||
extensions.tool_lifecycle_contributor(observer.clone());
|
||||
let mut builder = test_codex()
|
||||
@@ -2431,7 +2431,7 @@ text((await tools.exec_command({{cmd: "printf 'phase 3'"}})).output);
|
||||
responses::mount_sse_once(
|
||||
&server,
|
||||
sse(vec![
|
||||
serde_json::json!({"type": "response.created", "response": {"id": "resp-1", "headers": {"x-codex-guardian-ticket": "p".repeat(43)}}}),
|
||||
responses::ev_response_created("resp-1"),
|
||||
ev_custom_tool_call("call-1", "exec", &code),
|
||||
ev_completed("resp-1"),
|
||||
]),
|
||||
@@ -2464,7 +2464,7 @@ text((await tools.exec_command({{cmd: "printf 'phase 3'"}})).output);
|
||||
responses::mount_sse_once(
|
||||
&server,
|
||||
sse(vec![
|
||||
serde_json::json!({"type": "response.created", "response": {"id": "resp-3", "headers": {"x-codex-guardian-ticket": "q".repeat(43)}}}),
|
||||
responses::ev_response_created("resp-3"),
|
||||
responses::ev_function_call(
|
||||
"call-2",
|
||||
"wait",
|
||||
@@ -2511,7 +2511,7 @@ text((await tools.exec_command({{cmd: "printf 'phase 3'"}})).output);
|
||||
responses::mount_sse_once(
|
||||
&server,
|
||||
sse(vec![
|
||||
serde_json::json!({"type": "response.created", "response": {"id": "resp-5", "headers": {"x-codex-guardian-ticket": "r".repeat(43)}}}),
|
||||
responses::ev_response_created("resp-5"),
|
||||
responses::ev_function_call(
|
||||
"call-3",
|
||||
"wait",
|
||||
@@ -2551,15 +2551,18 @@ text((await tools.exec_command({{cmd: "printf 'phase 3'"}})).output);
|
||||
);
|
||||
assert_eq!(text_item(&third_items, /*index*/ 1), "phase 3");
|
||||
|
||||
let observed = observer.tickets.lock().unwrap();
|
||||
let mut nested_tickets = observed
|
||||
let observed = observer.response_ids.lock().unwrap();
|
||||
let mut nested_response_ids = observed
|
||||
.iter()
|
||||
.filter(|(tool, _)| tool == "exec_command")
|
||||
.filter_map(|(_, ticket)| ticket.clone())
|
||||
.skip_while(|ticket| ticket != &"q".repeat(43))
|
||||
.filter_map(|(_, response_id)| response_id.clone())
|
||||
.skip_while(|response_id| response_id != &"resp-3".to_owned())
|
||||
.collect::<Vec<_>>();
|
||||
nested_tickets.dedup();
|
||||
assert_eq!(nested_tickets, vec!["q".repeat(43), "r".repeat(43)]);
|
||||
nested_response_ids.dedup();
|
||||
assert_eq!(
|
||||
nested_response_ids,
|
||||
vec!["resp-3".to_owned(), "resp-5".to_owned()]
|
||||
);
|
||||
assert_eq!(
|
||||
observed
|
||||
.iter()
|
||||
@@ -2567,8 +2570,8 @@ text((await tools.exec_command({{cmd: "printf 'phase 3'"}})).output);
|
||||
.cloned()
|
||||
.collect::<Vec<_>>(),
|
||||
vec![
|
||||
("wait".to_owned(), Some("q".repeat(43))),
|
||||
("wait".to_owned(), Some("r".repeat(43))),
|
||||
("wait".to_owned(), Some("resp-3".to_owned())),
|
||||
("wait".to_owned(), Some("resp-5".to_owned())),
|
||||
]
|
||||
);
|
||||
Ok(())
|
||||
|
||||
@@ -135,14 +135,14 @@ impl TimeProvider for RecordingTimeProvider {
|
||||
#[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")]
|
||||
#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "OpenAI", "/backend-api/codex", true, "/backend-api/codex/guardian", false; "server_without_response_id")]
|
||||
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,
|
||||
response_id_present: bool,
|
||||
) -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
@@ -154,16 +154,16 @@ async fn guardian_session_inherits_parent_http_fallback(
|
||||
.mount_as_scoped(&server)
|
||||
.await;
|
||||
|
||||
let parent_ticket = "p".repeat(43);
|
||||
let parent_response_id = "parent-tool";
|
||||
let responses = mount_sse_sequence(
|
||||
&server,
|
||||
vec![
|
||||
// A failed sampling attempt must not lend its receipt to the retried action.
|
||||
// A failed sampling attempt must not supply the retried action's parent ID.
|
||||
sse(vec![json!({"type": "response.created", "response": {
|
||||
"id": "failed-parent", "headers": {"x-codex-guardian-ticket": "f".repeat(43)}
|
||||
"id": "failed-parent"
|
||||
}})]),
|
||||
sse(vec![
|
||||
json!({"type": "response.created", "response": {"id": "parent-tool", "headers": ticket_issued.then(|| json!({"x-codex-guardian-ticket": parent_ticket}))}}),
|
||||
json!({"type": "response.created", "response": {"id": response_id_present.then_some(parent_response_id)}}),
|
||||
ev_function_call(
|
||||
"call",
|
||||
"exec_command",
|
||||
@@ -241,30 +241,30 @@ 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 credits_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"),
|
||||
body["client_metadata"].get("parent_response_id").cloned(),
|
||||
body["client_metadata"].get("guardian_credits_requested"),
|
||||
),
|
||||
(
|
||||
(tickets_enabled && ticket_issued).then(|| json!(parent_ticket)),
|
||||
(credits_enabled && response_id_present).then(|| json!(parent_response_id)),
|
||||
None
|
||||
)
|
||||
);
|
||||
assert!(!body["input"].to_string().contains(&parent_ticket));
|
||||
assert!(!body["input"].to_string().contains(parent_response_id));
|
||||
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")
|
||||
.get("guardian_credits_requested")
|
||||
.cloned(),
|
||||
body["client_metadata"].get("guardian_ticket"),
|
||||
body["client_metadata"].get("parent_response_id"),
|
||||
),
|
||||
(tickets_enabled.then(|| json!("true")), None)
|
||||
(credits_enabled.then(|| json!("true")), None)
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -409,7 +409,7 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review(
|
||||
skip_if_no_network!(Ok(()));
|
||||
|
||||
let uses_codex_backend = auth.uses_codex_backend();
|
||||
let tickets_enabled = free_guardian && uses_codex_backend;
|
||||
let credits_enabled = free_guardian && uses_codex_backend;
|
||||
let bundled_models = codex_models_manager::bundled_models_response()?.models;
|
||||
let catalog_auto_review = bundled_models
|
||||
.iter()
|
||||
@@ -442,12 +442,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 parent_response_id = "approval-request";
|
||||
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![
|
||||
json!({"type": "response.created", "response": {"id": "approval-request", "headers": {"x-codex-guardian-ticket": parent_ticket}}}),
|
||||
ev_response_created(parent_response_id),
|
||||
ev_function_call("approval-call", "exec_command", &tool_args),
|
||||
ev_completed("approval-request"),
|
||||
]],
|
||||
@@ -603,30 +603,30 @@ async fn guardian_session_prewarms_and_is_reused_for_first_review(
|
||||
assert_eq!(
|
||||
(
|
||||
guardian_review["client_metadata"]
|
||||
.get("guardian_ticket")
|
||||
.get("parent_response_id")
|
||||
.cloned(),
|
||||
guardian_review["client_metadata"].get("guardian_ticket_requested"),
|
||||
guardian_review["client_metadata"].get("guardian_credits_requested"),
|
||||
parent_request["client_metadata"]
|
||||
.get("guardian_ticket_requested")
|
||||
.get("guardian_credits_requested")
|
||||
.cloned(),
|
||||
parent_request["client_metadata"].get("guardian_ticket"),
|
||||
parent_request["client_metadata"].get("parent_response_id"),
|
||||
),
|
||||
(
|
||||
tickets_enabled.then(|| json!(parent_ticket)),
|
||||
credits_enabled.then(|| json!(parent_response_id)),
|
||||
None,
|
||||
tickets_enabled.then(|| json!("true")),
|
||||
credits_enabled.then(|| json!("true")),
|
||||
None,
|
||||
)
|
||||
);
|
||||
assert!(
|
||||
guardian_prewarm["client_metadata"]
|
||||
.get("guardian_ticket")
|
||||
.get("parent_response_id")
|
||||
.is_none()
|
||||
);
|
||||
assert!(
|
||||
!guardian_review["input"]
|
||||
.to_string()
|
||||
.contains(&parent_ticket)
|
||||
.contains(parent_response_id)
|
||||
);
|
||||
for request in [&parent_request, &guardian_review] {
|
||||
assert_root_turn(request, Some(parent_turn_id))?;
|
||||
@@ -710,7 +710,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 = tickets_enabled && is_guardian;
|
||||
let uses_guardian_endpoint = credits_enabled && is_guardian;
|
||||
assert_eq!(
|
||||
handshake.uri(),
|
||||
if uses_guardian_endpoint {
|
||||
|
||||
@@ -512,11 +512,10 @@ 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
|
||||
let parent_response_id = input
|
||||
.turn_store
|
||||
.get::<codex_protocol::guardian_ticket::GuardianTicket>()
|
||||
.as_deref()
|
||||
.cloned();
|
||||
.get::<codex_api::ResponseId>()
|
||||
.map(|id| id.0.clone());
|
||||
let thread_context: Result<_, String> = async {
|
||||
let parsed_thread_id =
|
||||
ThreadId::from_string(&thread_id).map_err(|error| error.to_string())?;
|
||||
@@ -839,7 +838,7 @@ impl GuardianV2Extension {
|
||||
let instructions = guardian_config.render_classifier_instructions(policy);
|
||||
let output = match sampler
|
||||
.sample(LunaSamplingRequest {
|
||||
guardian_ticket,
|
||||
parent_response_id,
|
||||
instructions,
|
||||
trusted_review_evidence,
|
||||
trusted_tool_context,
|
||||
|
||||
@@ -90,8 +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<codex_protocol::guardian_ticket::GuardianTicket>,
|
||||
/// ID of the response handling the classified tool.
|
||||
pub parent_response_id: Option<String>,
|
||||
/// Trusted instructions describing the requested classification.
|
||||
pub instructions: String,
|
||||
/// Host-supplied Guardian reviews isolated from untrusted transcript entries.
|
||||
@@ -454,7 +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_response_id = request.parent_response_id;
|
||||
let parent_turn_id = request.parent_turn_id;
|
||||
let root_turn_id = request.root_turn_id;
|
||||
let mut input = vec![
|
||||
@@ -629,6 +629,11 @@ impl LunaSampler {
|
||||
turn_metadata["root_turn_id"] = json!(root_turn_id);
|
||||
}
|
||||
client_metadata.insert(TURN_METADATA_KEY.to_owned(), turn_metadata.to_string());
|
||||
if lease.connection.endpoint == ResponsesEndpoint::GuardianClassifier
|
||||
&& let Some(parent_response_id) = &parent_response_id
|
||||
{
|
||||
client_metadata.insert("parent_response_id".to_owned(), parent_response_id.clone());
|
||||
}
|
||||
request.client_metadata = Some(client_metadata);
|
||||
let mut stream = match lease
|
||||
.connection
|
||||
@@ -637,7 +642,6 @@ impl LunaSampler {
|
||||
ResponsesWsRequest::ResponseCreate((&request).into()),
|
||||
/*connection_reused*/ true,
|
||||
/*turn_state*/ None,
|
||||
guardian_ticket.as_ref(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
|
||||
@@ -192,7 +192,7 @@ async fn connect_sampler(config: LunaSamplerConfig) -> Result<LunaSampler> {
|
||||
|
||||
fn sample_request(parent_turn_id: &str) -> LunaSamplingRequest {
|
||||
LunaSamplingRequest {
|
||||
guardian_ticket: None,
|
||||
parent_response_id: None,
|
||||
instructions: "Return high for high risk or low for low risk.".to_owned(),
|
||||
trusted_review_evidence: Vec::new(),
|
||||
trusted_tool_context: None,
|
||||
@@ -447,7 +447,7 @@ async fn preconnected_sampler_reuses_authenticated_websocket_for_classifications
|
||||
|
||||
let first = sampler
|
||||
.sample(LunaSamplingRequest {
|
||||
guardian_ticket: None,
|
||||
parent_response_id: None,
|
||||
instructions: "Return high for high risk or low for low risk.".to_owned(),
|
||||
trusted_review_evidence: Vec::new(),
|
||||
trusted_tool_context: None,
|
||||
@@ -479,7 +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,
|
||||
parent_response_id: None,
|
||||
instructions: "Return high for high risk or low for low risk.".to_owned(),
|
||||
trusted_review_evidence: Vec::new(),
|
||||
trusted_tool_context: None,
|
||||
@@ -655,7 +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,
|
||||
parent_response_id: None,
|
||||
instructions: "Return high for high risk or low for low risk.".to_owned(),
|
||||
trusted_review_evidence: Vec::new(),
|
||||
trusted_tool_context: None,
|
||||
@@ -1119,7 +1119,7 @@ async fn sampler_limits_transient_recovery_attempts() -> Result<()> {
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn guardian_ticket_survives_classifier_transport_retry() -> Result<()> {
|
||||
async fn parent_response_id_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![
|
||||
@@ -1146,10 +1146,9 @@ async fn guardian_ticket_survives_classifier_transport_retry() -> Result<()> {
|
||||
);
|
||||
}
|
||||
let sampler = connect_sampler(config).await?;
|
||||
let raw_ticket = "t".repeat(43);
|
||||
let parent_response_id = "resp-parent";
|
||||
let mut request = sample_request("turn-1");
|
||||
request.guardian_ticket =
|
||||
codex_protocol::guardian_ticket::GuardianTicket::from_server(&raw_ticket);
|
||||
request.parent_response_id = Some(parent_response_id.to_owned());
|
||||
assert_eq!(sampler.sample(request).await?, "low");
|
||||
for server in [&expired, &healthy] {
|
||||
let requests = server.single_connection();
|
||||
@@ -1157,12 +1156,12 @@ async fn guardian_ticket_survives_classifier_transport_retry() -> Result<()> {
|
||||
let body = requests[0].body_json();
|
||||
assert_eq!(
|
||||
(
|
||||
body["client_metadata"].get("guardian_ticket").cloned(),
|
||||
body["client_metadata"].get("guardian_ticket_requested"),
|
||||
body["client_metadata"].get("parent_response_id").cloned(),
|
||||
body["client_metadata"].get("guardian_credits_requested"),
|
||||
),
|
||||
(free_guardian.then(|| json!(raw_ticket)), None)
|
||||
(free_guardian.then(|| json!(parent_response_id)), None)
|
||||
);
|
||||
assert!(!body["input"].to_string().contains(&raw_ticket));
|
||||
assert!(!body["input"].to_string().contains(parent_response_id));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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(Clone, PartialEq, Eq)]
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct EncodedJsonBody {
|
||||
bytes: Bytes,
|
||||
trace_bytes: Option<Bytes>,
|
||||
@@ -33,27 +33,11 @@ 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"<redacted>"));
|
||||
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]
|
||||
@@ -136,13 +120,7 @@ impl Request {
|
||||
self.body,
|
||||
Some(RequestBody::Json(_) | RequestBody::EncodedJson(_))
|
||||
);
|
||||
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
|
||||
let trace_bytes = if self.compression != RequestCompression::None
|
||||
&& tracing::enabled!(target: "codex_http_client::transport", tracing::Level::TRACE)
|
||||
{
|
||||
match self.body.as_ref() {
|
||||
|
||||
@@ -1,7 +1,4 @@
|
||||
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;
|
||||
@@ -31,35 +28,6 @@ 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), "<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]
|
||||
async fn connection_failures_are_classified_without_exposing_request_urls() {
|
||||
let unavailable_server =
|
||||
|
||||
@@ -1,37 +0,0 @@
|
||||
//! 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])")
|
||||
}
|
||||
}
|
||||
@@ -25,7 +25,6 @@ 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;
|
||||
|
||||
@@ -158,8 +158,6 @@ 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<crate::guardian_ticket::GuardianTicket>,
|
||||
/// Source classification for the caller that starts a new turn.
|
||||
/// Ignored when the submitted input steers an active turn.
|
||||
pub turn_trigger: Option<String>,
|
||||
|
||||
Reference in New Issue
Block a user