github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 16:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Route Guardian requests through /responses with identifying headers (#45736)

Browse Source 
## What changed

- Replace dedicated Guardian endpoints with `/responses`, sending `x-codex-guardian: reviewer` or `x-codex-guardian: classifier` for eligible Codex backend requests over HTTP and WebSocket.
- Add model-scoped thread headers, recheck backend authentication on each request attempt, and reconnect WebSockets when the applicable headers change.
- Retain `features.guardianv2.free_guardian` for configuration compatibility while removing its routing gate; the backend now controls Guardian billing.

## Testing

Update tests for reviewer and classifier headers, model and authentication scoping, HTTP fallback, WebSocket reuse, and parent-response metadata across retries.

GitOrigin-RevId: 1c2b3c458ab77fb40aae4ee6784f6d827c60e76c
This commit is contained in:
jif
2026-09-15 17:09:11 +00:00
committed by copyberry
parent d4e11a9b97
commit eeded5ba1a
22 changed files with 246 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/codex-api/src/endpoint/mod.rs
-1
View File
@@ -25,7 +25,6 @@ pub use realtime_websocket::RealtimeWebsocketEvents;
pub use realtime_websocket::RealtimeWebsocketWriter;
pub use realtime_websocket::session_update_session_json;
pub use responses::ResponsesClient;
pub use responses::ResponsesEndpoint;
pub use responses::ResponsesOptions;
pub use responses_websocket::ResponsesWebsocketClient;
pub use responses_websocket::ResponsesWebsocketClose;
codex-rs/codex-api/src/endpoint/responses.rs
+3 -35
View File
@@ -23,33 +23,9 @@ use std::sync::Arc;
use std::sync::OnceLock;
use tracing::instrument;
/// Responses-compatible inference routes supported by Codex backend.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub enum ResponsesEndpoint {
/// Regular user-owned model inference.
#[default]
Responses,
/// Full Guardian approval-review agent inference.
Guardian,
/// Lightweight asynchronous Guardian risk classification.
GuardianClassifier,
}
impl ResponsesEndpoint {
/// Returns the provider-relative path for this inference surface.
pub const fn path(self) -> &'static str {
match self {
Self::Responses => "/responses",
Self::Guardian => "/guardian",
Self::GuardianClassifier => "/guardian-classifier",
}
}
}
pub struct ResponsesClient<T: HttpTransport> {
session: EndpointSession<T>,
sse_telemetry: Option<Arc<dyn SseTelemetry>>,
endpoint: ResponsesEndpoint,
}
#[derive(Default)]
@@ -67,16 +43,9 @@ impl<T: HttpTransport> ResponsesClient<T> {
Self {
session: EndpointSession::new(transport, provider, auth),
sse_telemetry: None,
endpoint: ResponsesEndpoint::Responses,
}
}
/// Selects a Responses-compatible backend route for subsequent requests.
pub fn with_endpoint(mut self, endpoint: ResponsesEndpoint) -> Self {
self.endpoint = endpoint;
self
}
pub fn with_telemetry(
self,
request: Option<Arc<dyn RequestTelemetry>>,
@@ -85,7 +54,6 @@ impl<T: HttpTransport> ResponsesClient<T> {
Self {
session: self.session.with_request_telemetry(request),
sse_telemetry: sse,
endpoint: self.endpoint,
}
}
@@ -96,7 +64,7 @@ impl<T: HttpTransport> ResponsesClient<T> {
fields(
transport = "responses_http",
http.method = "POST",
api.path = self.endpoint.path()
api.path = "/responses"
)
)]
pub async fn stream_request(
@@ -135,7 +103,7 @@ impl<T: HttpTransport> ResponsesClient<T> {
fields(
transport = "responses_http",
http.method = "POST",
api.path = self.endpoint.path(),
api.path = "/responses",
turn.has_state = turn_state.is_some()
)
)]
@@ -168,7 +136,7 @@ impl<T: HttpTransport> ResponsesClient<T> {
.session
.stream_encoded_json_with(
Method::POST,
self.endpoint.path(),
"/responses",
extra_headers,
Some(body),
|req| {
codex-rs/codex-api/src/endpoint/responses_websocket.rs
+5 -22
View File
@@ -4,7 +4,6 @@ use crate::common::ResponseStream;
use crate::common::ResponsesWsRequest;
use crate::common::SafetyBufferingTreatment;
use crate::common::WS_REQUEST_HEADER_TRACEPARENT_CLIENT_METADATA_KEY;
use crate::endpoint::responses::ResponsesEndpoint;
use crate::error::ApiError;
use crate::provider::Provider;
use crate::rate_limits::parse_rate_limit_event;
@@ -182,7 +181,6 @@ struct ResponsesWebsocketTimingLogContext {
pub struct ResponsesWebsocketConnection {
stream: Arc<Mutex<Option<WsStream>>>,
endpoint: ResponsesEndpoint,
// TODO (pakrym): is this the right place for timeout?
idle_timeout: Duration,
server_reasoning_included: bool,
@@ -194,7 +192,6 @@ impl std::fmt::Debug for ResponsesWebsocketConnection {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ResponsesWebsocketConnection")
.field("stream", &"<ws-stream>")
.field("endpoint", &self.endpoint)
.field("idle_timeout", &self.idle_timeout)
.field("server_reasoning_included", &self.server_reasoning_included)
.field("server_model", &self.server_model)
@@ -210,11 +207,9 @@ impl ResponsesWebsocketConnection {
server_reasoning_included: bool,
server_model: Option<String>,
telemetry: Option<Arc<dyn WebsocketTelemetry>>,
endpoint: ResponsesEndpoint,
) -> Self {
Self {
stream: Arc::new(Mutex::new(Some(stream))),
endpoint,
idle_timeout,
server_reasoning_included,
server_model,
@@ -230,7 +225,7 @@ impl ResponsesWebsocketConnection {
name = "responses_websocket.stream_request",
level = "info",
skip_all,
fields(transport = "responses_websocket", api.path = self.endpoint.path())
fields(transport = "responses_websocket", api.path = "/responses")
)]
pub async fn stream_request(
&self,
@@ -345,7 +340,6 @@ impl ResponsesWebsocketConnection {
pub struct ResponsesWebsocketClient {
provider: Provider,
auth: SharedAuthProvider,
endpoint: ResponsesEndpoint,
}
/// Close frame information captured by a handshake probe.
@@ -375,24 +369,14 @@ pub struct ResponsesWebsocketProbe {
impl ResponsesWebsocketClient {
/// Creates a Responses WebSocket client for an already-resolved provider and auth source.
pub fn new(provider: Provider, auth: SharedAuthProvider) -> Self {
Self {
provider,
auth,
endpoint: ResponsesEndpoint::Responses,
}
}
/// Selects a Responses-compatible backend route for subsequent connections.
pub fn with_endpoint(mut self, endpoint: ResponsesEndpoint) -> Self {
self.endpoint = endpoint;
self
Self { provider, auth }
}
#[instrument(
name = "responses_websocket.connect",
level = "info",
skip_all,
fields(transport = "responses_websocket", api.path = self.endpoint.path())
fields(transport = "responses_websocket", api.path = "/responses")
)]
pub async fn connect(
&self,
@@ -404,7 +388,7 @@ impl ResponsesWebsocketClient {
) -> Result<ResponsesWebsocketConnection, ApiError> {
let ws_url = self
.provider
.websocket_url_for_path(self.endpoint.path())
.websocket_url_for_path("/responses")
.map_err(|err| ApiError::Stream(format!("failed to build websocket URL: {err}")))?;
let mut headers =
@@ -419,7 +403,6 @@ impl ResponsesWebsocketClient {
server_reasoning_included,
server_model,
telemetry,
self.endpoint,
))
}
@@ -439,7 +422,7 @@ impl ResponsesWebsocketClient {
) -> Result<ResponsesWebsocketProbe, ApiError> {
let ws_url = self
.provider
.websocket_url_for_path(self.endpoint.path())
.websocket_url_for_path("/responses")
.map_err(|err| ApiError::Stream(format!("failed to build websocket URL: {err}")))?;
let mut headers =
codex-rs/codex-api/src/lib.rs
-1
View File
@@ -65,7 +65,6 @@ pub use crate::endpoint::RealtimeWebsocketConnection;
pub use crate::endpoint::RealtimeWebsocketEvents;
pub use crate::endpoint::RealtimeWebsocketWriter;
pub use crate::endpoint::ResponsesClient;
pub use crate::endpoint::ResponsesEndpoint;
pub use crate::endpoint::ResponsesOptions;
pub use crate::endpoint::ResponsesWebsocketClient;
pub use crate::endpoint::ResponsesWebsocketClose;
codex-rs/codex-api/tests/clients.rs
+13 -8
View File
@@ -12,7 +12,6 @@ use codex_api::Compression;
use codex_api::Provider;
use codex_api::ResponsesApiRequest;
use codex_api::ResponsesClient;
use codex_api::ResponsesEndpoint;
use codex_api::ResponsesOptions;
use codex_client::HttpTransport;
use codex_client::Request;
@@ -308,22 +307,28 @@ async fn responses_client_uses_responses_path() -> Result<()> {
}
#[tokio::test]
async fn responses_client_uses_guardian_path() -> Result<()> {
async fn responses_client_sends_extra_headers() -> 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 headers = HeaderMap::from_iter([(
http::HeaderName::from_static("x-custom-request"),
HeaderValue::from_static("example"),
)]);
let _stream = client
.stream(
serde_json::json!({ "echo": true }),
HeaderMap::new(),
headers,
Compression::None,
/*turn_state*/ None,
)
.await?;
assert_path_ends_with(&state.take_stream_requests(), "/guardian");
let requests = state.take_stream_requests();
assert_path_ends_with(&requests, "/responses");
assert_eq!(
requests[0].headers.get("x-custom-request"),
Some(&HeaderValue::from_static("example")),
);
Ok(())
}
codex-rs/core/config.schema.json
+1 -1
View File
@@ -1617,7 +1617,7 @@
"type": "boolean"
},
"free_guardian": {
"description": "Route Guardian review and classification through the unmetered Codex endpoints.",
"description": "Legacy setting retained for config compatibility; the backend now controls Guardian billing.",
"type": "boolean"
},
"max_action_tokens": {
codex-rs/core/src/client.rs
+67 -65
View File
@@ -32,6 +32,7 @@ use std::sync::OnceLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use crate::CodexResponsesHeaders;
use async_channel::Sender;
use codex_api::AgentIdentityTelemetry;
use codex_api::ApiError;
@@ -51,7 +52,6 @@ use codex_api::ReqwestTransport;
use codex_api::ResponseCreateWsRequest;
use codex_api::ResponsesApiRequest;
use codex_api::ResponsesClient as ApiResponsesClient;
use codex_api::ResponsesEndpoint;
use codex_api::ResponsesOptions as ApiResponsesOptions;
use codex_api::ResponsesWebsocketClient as ApiWebSocketResponsesClient;
use codex_api::ResponsesWebsocketConnection as ApiWebSocketConnection;
@@ -248,7 +248,7 @@ pub struct ModelClient {
state: Arc<ModelClientState>,
agent_identity_policy: AgentIdentityAuthPolicy,
prompt_cache_key_override: Option<String>,
free_guardian_enabled: bool,
codex_responses_headers: Option<Arc<CodexResponsesHeaders>>,
event_sender: Option<Sender<ProtocolEvent>>,
http_client_factory: HttpClientFactory,
}
@@ -292,7 +292,7 @@ struct LastResponse {
#[derive(Debug, Default)]
struct WebsocketSession {
connection: Option<ApiWebSocketConnection>,
endpoint: Option<ResponsesEndpoint>,
responses_headers: ApiHeaderMap,
/// Owner of the cached state, including before a connection is opened.
auth_owner_generation: Option<u64>,
last_request: Option<ResponsesApiRequest>,
@@ -473,24 +473,21 @@ impl ModelClient {
}),
agent_identity_policy,
prompt_cache_key_override: None,
free_guardian_enabled: false,
codex_responses_headers: None,
event_sender: None,
http_client_factory,
}
}
pub(crate) fn with_free_guardian_enabled(mut self, free_guardian_enabled: bool) -> Self {
self.free_guardian_enabled = free_guardian_enabled;
self
}
pub(crate) fn with_session_context(
mut self,
prompt_cache_key_override: Option<String>,
event_sender: Sender<ProtocolEvent>,
codex_responses_headers: Option<Arc<CodexResponsesHeaders>>,
) -> Self {
self.prompt_cache_key_override = prompt_cache_key_override;
self.event_sender = Some(event_sender);
self.codex_responses_headers = codex_responses_headers;
self
}
@@ -971,17 +968,16 @@ impl ModelClient {
})
}
fn responses_endpoint(&self, auth: Option<&CodexAuth>, model: &str) -> ResponsesEndpoint {
if self.free_guardian_enabled
&& crate::guardian::is_basic_session_source(&self.state.session_source)
&& self.uses_codex_backend(auth)
&& self.state.provider.info().supports_codex_backend_routes()
&& model == self.state.provider.approval_review_preferred_model()
{
ResponsesEndpoint::Guardian
} else {
ResponsesEndpoint::Responses
}
fn responses_headers(&self, auth: Option<&CodexAuth>, model: &str) -> ApiHeaderMap {
self.codex_responses_headers
.as_ref()
.filter(|config| {
config.model == model
&& self.uses_codex_backend(auth)
&& self.state.provider.info().supports_codex_backend_routes()
})
.map(|config| config.headers.clone())
.unwrap_or_default()
}
fn uses_codex_backend(&self, auth: Option<&CodexAuth>) -> bool {
@@ -1000,22 +996,22 @@ impl ModelClient {
metadata: &mut Option<HashMap<String, String>>,
parent_response_id: Option<&str>,
auth: Option<&CodexAuth>,
endpoint: ResponsesEndpoint,
responses_headers: &ApiHeaderMap,
) {
if let Some(metadata) = metadata.as_mut() {
metadata.remove("guardian_credits_requested");
metadata.remove("parent_response_id");
}
if endpoint == ResponsesEndpoint::Guardian
&& let Some(parent_response_id) = parent_response_id
{
let guardian_reviewer = responses_headers
.get("x-codex-guardian")
.is_some_and(|value| value == "reviewer");
if guardian_reviewer && 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
if !guardian_reviewer
&& !crate::guardian::is_basic_session_source(&self.state.session_source)
&& matches!(
auth,
@@ -1079,9 +1075,10 @@ impl ModelClient {
responses_metadata: &CodexResponsesMetadata,
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
endpoint: ResponsesEndpoint,
responses_headers: &ApiHeaderMap,
) -> std::result::Result<ApiWebSocketConnection, ApiError> {
let headers = self.build_websocket_headers(responses_metadata).await;
let mut headers = self.build_websocket_headers(responses_metadata).await;
headers.extend(responses_headers.clone());
let websocket_telemetry = ModelClientSession::build_websocket_telemetry(
session_telemetry,
auth_context.clone(),
@@ -1092,15 +1089,13 @@ impl ModelClient {
let start = Instant::now();
let result = match tokio::time::timeout(
websocket_connect_timeout,
ApiWebSocketResponsesClient::new(api_provider, api_auth)
.with_endpoint(endpoint)
.connect(
&self.http_client_factory,
headers,
codex_login::default_client::default_headers(),
/*turn_state*/ None,
Some(websocket_telemetry),
),
ApiWebSocketResponsesClient::new(api_provider, api_auth).connect(
&self.http_client_factory,
headers,
codex_login::default_client::default_headers(),
/*turn_state*/ None,
Some(websocket_telemetry),
),
)
.await
{
@@ -1207,7 +1202,7 @@ impl Drop for ModelClientSession {
impl ModelClientSession {
fn reset_websocket_session(&mut self) {
self.websocket_session.connection = None;
self.websocket_session.endpoint = None;
self.websocket_session.responses_headers.clear();
self.websocket_session.last_request = None;
self.websocket_session.last_response_rx = None;
self.websocket_session.last_response_from_untraced_warmup = false;
@@ -1360,9 +1355,9 @@ impl ModelClientSession {
client_setup.agent_identity_telemetry.clone(),
PendingUnauthorizedRetry::default(),
);
let endpoint = self
let responses_headers = self
.client
.responses_endpoint(client_setup.auth.as_ref(), &model_info.slug);
.responses_headers(client_setup.auth.as_ref(), &model_info.slug);
self.websocket_connection(WebsocketConnectParams {
session_telemetry,
api_provider: client_setup.api_provider,
@@ -1370,8 +1365,8 @@ impl ModelClientSession {
auth_owner_generation: client_setup.auth_owner_generation,
responses_metadata,
auth_context,
request_route_telemetry: RequestRouteTelemetry::for_endpoint(endpoint.path()),
endpoint,
request_route_telemetry: RequestRouteTelemetry::for_endpoint("/responses"),
responses_headers: &responses_headers,
})
.await?;
Ok(())
@@ -1385,7 +1380,7 @@ impl ModelClientSession {
provider = %self.client.state.provider.info().name,
wire_api = %self.client.state.provider.info().wire_api,
transport = "responses_websocket",
api.path = params.endpoint.path(),
api.path = "/responses",
turn.has_metadata_header = params.responses_metadata.has_turn_metadata()
)
)]
@@ -1401,11 +1396,12 @@ impl ModelClientSession {
responses_metadata,
auth_context,
request_route_telemetry,
endpoint,
responses_headers,
} = params;
let needs_new = match self.websocket_session.connection.as_ref() {
Some(conn) => {
self.websocket_session.endpoint != Some(endpoint) || conn.is_closed().await
self.websocket_session.responses_headers != *responses_headers
|| conn.is_closed().await
}
None => true,
};
@@ -1427,7 +1423,7 @@ impl ModelClientSession {
responses_metadata,
auth_context,
request_route_telemetry,
endpoint,
responses_headers,
)
.await
{
@@ -1440,7 +1436,7 @@ impl ModelClientSession {
}
};
self.websocket_session.connection = Some(new_conn);
self.websocket_session.endpoint = Some(endpoint);
self.websocket_session.responses_headers = responses_headers.clone();
self.websocket_session.auth_owner_generation = auth_owner_generation;
self.websocket_session
.set_connection_reused(/*connection_reused*/ false);
@@ -1504,13 +1500,13 @@ impl ModelClientSession {
let mut pending_retry = PendingUnauthorizedRetry::default();
loop {
let client_setup = self.client.current_client_setup().await?;
let endpoint = self
let responses_headers = self
.client
.responses_endpoint(client_setup.auth.as_ref(), &model_info.slug);
tracing::Span::current().record("api.path", endpoint.path());
.responses_headers(client_setup.auth.as_ref(), &model_info.slug);
tracing::Span::current().record("api.path", "/responses");
let transport = self
.client
.build_api_transport(&client_setup.api_provider, endpoint.path())?;
.build_api_transport(&client_setup.api_provider, "/responses")?;
let request_auth_context = AuthRequestTelemetryContext::new(
client_setup.auth.as_ref().map(CodexAuth::auth_mode),
client_setup.api_auth.as_ref(),
@@ -1520,7 +1516,7 @@ impl ModelClientSession {
let (request_telemetry, sse_telemetry) = Self::build_streaming_telemetry(
session_telemetry,
request_auth_context,
RequestRouteTelemetry::for_endpoint(endpoint.path()),
RequestRouteTelemetry::for_endpoint("/responses"),
self.client.state.auth_env_telemetry.clone(),
);
let compression = self.responses_request_compression(client_setup.auth.as_ref());
@@ -1548,12 +1544,15 @@ impl ModelClientSession {
&mut request.client_metadata,
responses_metadata.parent_response_id.as_deref(),
client_setup.auth.as_ref(),
endpoint,
&responses_headers,
);
if endpoint == ResponsesEndpoint::Guardian {
let guardian_reviewer = responses_headers
.get("x-codex-guardian")
.is_some_and(|value| value == "reviewer");
if guardian_reviewer {
request.service_tier = None;
}
if endpoint == ResponsesEndpoint::Responses
if !guardian_reviewer
&& let Some(header_value) = self.client.build_routing_hint_header(
client_setup.auth.as_ref(),
&request.model,
@@ -1575,6 +1574,7 @@ impl ModelClientSession {
}
let request_session_telemetry =
session_telemetry_for_request(session_telemetry, &request);
options.extra_headers.extend(responses_headers);
let inference_trace_attempt = inference_trace.start_attempt();
inference_trace_attempt.add_request_headers(&mut options.extra_headers);
inference_trace_attempt.record_started(&request);
@@ -1583,7 +1583,6 @@ impl ModelClientSession {
client_setup.api_provider,
client_setup.api_auth,
)
.with_endpoint(endpoint)
.with_telemetry(Some(request_telemetry), Some(sse_telemetry));
let stream_result = client.stream_request(request, options).await;
@@ -1678,10 +1677,10 @@ impl ModelClientSession {
let mut pending_retry = PendingUnauthorizedRetry::default();
loop {
let client_setup = self.client.current_client_setup().await?;
let endpoint = self
let responses_headers = self
.client
.responses_endpoint(client_setup.auth.as_ref(), &model_info.slug);
tracing::Span::current().record("api.path", endpoint.path());
.responses_headers(client_setup.auth.as_ref(), &model_info.slug);
tracing::Span::current().record("api.path", "/responses");
let request_auth_context = AuthRequestTelemetryContext::new(
client_setup.auth.as_ref().map(CodexAuth::auth_mode),
client_setup.api_auth.as_ref(),
@@ -1700,7 +1699,10 @@ impl ModelClientSession {
&mut request.input,
&client_setup.api_provider,
);
if endpoint == ResponsesEndpoint::Guardian {
let guardian_reviewer = responses_headers
.get("x-codex-guardian")
.is_some_and(|value| value == "reviewer");
if guardian_reviewer {
request.service_tier = None;
}
request.access_programs = cyber_access_program::for_auth(
@@ -1708,7 +1710,7 @@ impl ModelClientSession {
prompt.cyber_access_program,
);
let mut websocket_metadata = responses_metadata.clone();
websocket_metadata.routing_hint = if endpoint == ResponsesEndpoint::Responses {
websocket_metadata.routing_hint = if !guardian_reviewer {
self.client.build_routing_hint_header(
client_setup.auth.as_ref(),
&request.model,
@@ -1731,8 +1733,8 @@ impl ModelClientSession {
auth_owner_generation: client_setup.auth_owner_generation,
responses_metadata: &websocket_metadata,
auth_context: request_auth_context,
request_route_telemetry: RequestRouteTelemetry::for_endpoint(endpoint.path()),
endpoint,
request_route_telemetry: RequestRouteTelemetry::for_endpoint("/responses"),
responses_headers: &responses_headers,
})
.await
{
@@ -1822,7 +1824,7 @@ impl ModelClientSession {
&mut ws_payload.client_metadata,
responses_metadata.parent_response_id.as_deref(),
client_setup.auth.as_ref(),
endpoint,
&responses_headers,
);
let mut ws_request = ResponsesWsRequest::ResponseCreate(ws_payload);
stamp_ws_stream_request_start_ms(&mut ws_request);
@@ -2345,7 +2347,7 @@ struct WebsocketConnectParams<'a> {
responses_metadata: &'a CodexResponsesMetadata,
auth_context: AuthRequestTelemetryContext,
request_route_telemetry: RequestRouteTelemetry,
endpoint: ResponsesEndpoint,
responses_headers: &'a ApiHeaderMap,
}
fn emit_auth_recovery_event(
codex-rs/core/src/client_tests.rs
+25 -61
View File
@@ -17,7 +17,6 @@ use crate::test_support::responses_metadata as test_responses_metadata;
use codex_api::AgentIdentityTelemetry;
use codex_api::ApiError;
use codex_api::ResponseEvent;
use codex_api::ResponsesEndpoint;
use codex_api::TransportError;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
@@ -1203,76 +1202,41 @@ fn model_client_with_counting_attestation(
}
#[test]
fn guardian_reviewer_uses_dedicated_endpoint_only_with_codex_backend_auth() {
fn thread_responses_headers_are_scoped_to_model_and_backend_auth() {
let (mut model_client, _) =
model_client_with_counting_attestation(/*include_attestation*/ true);
Arc::get_mut(&mut model_client.state)
.expect("test client should have unique session state")
.session_source = SessionSource::SubAgent(SubAgentSource::Other("guardian".to_owned()));
assert_eq!(
model_client.responses_endpoint(
Some(&CodexAuth::create_dummy_chatgpt_auth_for_testing()),
"codex-auto-review",
let headers = http::HeaderMap::from_iter([(
http::HeaderName::from_static("x-custom-request"),
http::HeaderValue::from_static("example"),
)]);
model_client.codex_responses_headers = Some(Arc::new(crate::CodexResponsesHeaders {
model: "selected-model".to_owned(),
headers: headers.clone(),
}));
let chatgpt_auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
let api_key_auth = CodexAuth::from_api_key("test-api-key");
for (auth, model, expected) in [
(Some(&chatgpt_auth), "selected-model", headers),
(Some(&chatgpt_auth), "other-model", http::HeaderMap::new()),
(
Some(&api_key_auth),
"selected-model",
http::HeaderMap::new(),
),
ResponsesEndpoint::Responses
);
model_client = model_client.with_free_guardian_enabled(/*free_guardian_enabled*/ true);
assert_eq!(
model_client.responses_endpoint(
Some(&CodexAuth::create_dummy_chatgpt_auth_for_testing()),
"codex-auto-review",
),
ResponsesEndpoint::Guardian
);
assert_eq!(
model_client.responses_endpoint(
Some(&CodexAuth::create_dummy_chatgpt_auth_for_testing()),
"required-reviewer-model",
),
ResponsesEndpoint::Responses
);
assert_eq!(
model_client.responses_endpoint(
Some(&CodexAuth::create_dummy_chatgpt_auth_for_testing()),
"parent-fallback-model",
),
ResponsesEndpoint::Responses
);
assert_eq!(
model_client.responses_endpoint(
Some(&CodexAuth::from_api_key("test-api-key")),
"codex-auto-review",
),
ResponsesEndpoint::Responses
);
(None, "selected-model", http::HeaderMap::new()),
] {
assert_eq!(model_client.responses_headers(auth, model), expected);
}
Arc::get_mut(&mut model_client.state)
.expect("test client should have unique session state")
.provider = create_model_provider(
ModelProviderInfo::create_openai_provider(Some("https://proxy.example.com/v1".to_owned())),
Some(AuthManager::from_auth_for_testing(
CodexAuth::create_dummy_chatgpt_auth_for_testing(),
)),
Some(AuthManager::from_auth_for_testing(chatgpt_auth.clone())),
);
assert_eq!(
model_client.responses_endpoint(
Some(&CodexAuth::create_dummy_chatgpt_auth_for_testing()),
"codex-auto-review",
),
ResponsesEndpoint::Responses
);
Arc::get_mut(&mut model_client.state)
.expect("test client should have unique session state")
.session_source = SessionSource::Exec;
assert_eq!(
model_client.responses_endpoint(
Some(&CodexAuth::create_dummy_chatgpt_auth_for_testing()),
"codex-auto-review",
),
ResponsesEndpoint::Responses
model_client.responses_headers(Some(&chatgpt_auth), "selected-model"),
http::HeaderMap::new(),
);
}
codex-rs/core/src/config/mod.rs
-11
View File
@@ -1522,17 +1522,6 @@ impl Config {
&self.sqlite
}
/// Whether Guardian may use the unmetered Codex inference endpoints.
pub fn free_guardian_enabled(&self) -> bool {
self.config_layer_stack
.effective_config()
.get("features")
.and_then(|features| features.get("guardianv2"))
.and_then(|guardian| guardian.get("free_guardian"))
.and_then(toml::Value::as_bool)
.unwrap_or(false)
}
/// Resolves the configured, reviewer-catalog, or bundled Guardian policy.
pub fn resolve_guardian_policy<'a>(
&'a self,
codex-rs/core/src/lib.rs
+2
View File
@@ -13,6 +13,8 @@ mod realtime_context;
mod realtime_conversation;
mod realtime_history;
mod realtime_prompt;
mod responses_headers;
pub use responses_headers::CodexResponsesHeaders;
mod responses_metadata;
mod responses_retry;
pub(crate) mod session;
codex-rs/core/src/responses_headers.rs
+24
View File
@@ -0,0 +1,24 @@
//! Header overrides for host-created threads on the Codex Responses endpoint.
//! Headers are scoped to a model and rechecked against current auth on each attempt.
use http::HeaderMap;
/// Seed in `StartThreadOptions::thread_extension_init` to supply Responses headers.
///
/// Applied only to the selected model with Codex backend auth and routing. The headers
/// are sent on HTTP requests and WebSocket handshakes; changes require a new socket.
/// Other provider endpoints and unrelated threads do not inherit these headers.
#[derive(Clone)]
pub struct CodexResponsesHeaders {
pub model: String,
pub headers: HeaderMap,
}
impl std::fmt::Debug for CodexResponsesHeaders {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CodexResponsesHeaders")
.field("model", &self.model)
.field("header_names", &self.headers.keys())
.finish()
}
}
codex-rs/core/src/session/session.rs
+2 -1
View File
@@ -1600,6 +1600,7 @@ impl Session {
&config.features,
&initial_history,
);
let codex_responses_headers = thread_extension_data.get::<crate::CodexResponsesHeaders>();
let services = SessionServices {
// Start with an empty connection set. The initialized set is
// published after SessionConfigured so MCP events follow it.
@@ -1672,7 +1673,6 @@ impl Session {
attestation_provider,
config.http_client_factory(),
)
.with_free_guardian_enabled(config.free_guardian_enabled())
.with_session_context(
crate::guardian::prompt_cache_key_override_for_review_session(
&session_configuration.session_source,
@@ -1680,6 +1680,7 @@ impl Session {
)
.or(fork_cache_key),
tx_event.clone(),
codex_responses_headers,
),
executed_tool_calls: executed_tool_calls.clone(),
code_mode_service: crate::tools::code_mode::CodeModeService::new(
codex-rs/core/tests/common/responses.rs
+1 -1
View File
@@ -1079,7 +1079,7 @@ where
fn base_mock() -> (MockBuilder, ResponseMock) {
let response_mock = ResponseMock::new();
let mock = Mock::given(method("POST"))
.and(path_regex(".*/(responses|guardian|guardian-classifier)$"))
.and(path_regex(".*/responses$"))
.and(response_mock.clone());
(mock, response_mock)
}
codex-rs/core/tests/suite/guardian_review.rs
+18 -10
View File
@@ -138,11 +138,11 @@ impl TimeProvider for RecordingTimeProvider {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[test_case(CodexAuth::from_api_key("test-api-key"), "OpenAI", "/v1", true, "/v1/responses", true; "api_key_uses_responses")]
#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "OpenAI", "/backend-api/codex", false, "/backend-api/codex/responses", true; "chatgpt_uses_responses_by_default")]
#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "OpenAI", "/backend-api/codex", true, "/backend-api/codex/guardian", true; "chatgpt_uses_guardian_when_enabled")]
#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "OpenAI", "/backend-api/codex", false, "/backend-api/codex/responses", true; "chatgpt_marks_guardian_by_default")]
#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "OpenAI", "/backend-api/codex", true, "/backend-api/codex/responses", true; "legacy_opt_in_still_accepted")]
#[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; "retry_without_response_id_keeps_last_parent")]
#[test_case(CodexAuth::create_dummy_chatgpt_auth_for_testing(), "OpenAI", "/backend-api/codex", true, "/backend-api/codex/responses", false; "retry_without_response_id_keeps_last_parent")]
async fn guardian_session_inherits_parent_http_fallback(
auth: CodexAuth,
provider_name: &str,
@@ -151,6 +151,8 @@ async fn guardian_session_inherits_parent_http_fallback(
expected_guardian_path: &str,
response_id_present: bool,
) -> Result<()> {
let credits_enabled =
auth.uses_codex_backend() && provider_name == "OpenAI" && base_path == "/backend-api/codex";
skip_if_no_network!(Ok(()));
let configured_policy = "Use the task-configured Guardian policy.";
@@ -256,7 +258,10 @@ async fn guardian_session_inherits_parent_http_fallback(
.body_contains_text("Configured template: Use the task-configured Guardian policy.")
);
assert_eq!(guardian_request.path(), expected_guardian_path);
let credits_enabled = expected_guardian_path.ends_with("/guardian");
assert_eq!(
guardian_request.header("x-codex-guardian").as_deref(),
credits_enabled.then_some("reviewer")
);
let body = guardian_request.body_json();
assert_eq!(
(
@@ -278,6 +283,7 @@ async fn guardian_session_inherits_parent_http_fallback(
for request in responses.requests() {
let body = request.body_json();
if body["client_metadata"]["x-openai-subagent"] != "guardian" {
assert_eq!(request.header("x-codex-guardian"), None);
assert_eq!(
(
body["client_metadata"]
@@ -772,7 +778,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 credits_enabled = free_guardian && uses_codex_backend;
let credits_enabled = uses_codex_backend;
let bundled_models = codex_models_manager::bundled_models_response()?.models;
let catalog_auto_review = bundled_models
.iter()
@@ -1086,18 +1092,20 @@ 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 = credits_enabled && is_guardian;
let is_guardian_request = credits_enabled && is_guardian;
assert_eq!(
handshake.uri(),
if uses_guardian_endpoint {
"/backend-api/codex/guardian"
} else if uses_codex_backend {
if uses_codex_backend {
"/backend-api/codex/responses"
} else {
"/v1/responses"
}
);
if uses_guardian_endpoint {
assert_eq!(
handshake.header("x-codex-guardian").as_deref(),
is_guardian_request.then_some("reviewer")
);
if is_guardian_request {
assert_eq!(handshake.header("x-codex-routing-hint"), None);
}
}
codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs
-2
View File
@@ -52,8 +52,6 @@ pub struct LunaSamplerConfig {
pub thread_id: String,
/// Optional host-resolved request originator.
pub originator: Option<String>,
/// Whether this thread may use the unmetered Guardian classifier endpoint.
pub free_guardian: bool,
/// Optional inference service tier.
pub service_tier: Option<String>,
/// Luna model's host-resolved encrypted-compaction compatibility hash.
codex-rs/ext/guardian-v2/src/async_scorer/sampler/connection_pool.rs
+40 -30
View File
@@ -13,7 +13,6 @@ use codex_api::ReqwestTransport;
use codex_api::ResponseStream;
use codex_api::ResponsesApiRequest;
use codex_api::ResponsesClient;
use codex_api::ResponsesEndpoint;
use codex_api::ResponsesOptions;
use codex_api::ResponsesWebsocketClient;
use codex_api::ResponsesWebsocketConnection;
@@ -44,6 +43,12 @@ const CONNECT_COOLDOWN: Duration = Duration::from_secs(5 * 60);
const MAX_WEBSOCKET_AGE: Duration = Duration::from_secs(55 * 60);
const RESPONSES_WEBSOCKETS_BETA: &str = "responses_websockets=2026-02-06";
#[derive(Clone, Copy, PartialEq, Eq)]
pub(super) enum RequestMode {
Regular,
GuardianClassifier,
}
pub(super) struct ConnectionPool {
config: Arc<LunaSamplerConfig>,
pub(super) idle_connections: Mutex<Vec<PooledConnection>>,
@@ -57,7 +62,7 @@ pub(super) struct ConnectionPool {
pub(super) struct PooledConnection {
connection: ResponsesWebsocketConnection,
endpoint: ResponsesEndpoint,
request_kind: RequestMode,
// The bridge routes by thread ID, so each socket needs its own identity.
thread_id: String,
pub(super) expires_at: Instant,
@@ -72,7 +77,7 @@ enum Connection {
pub(super) struct ConnectionLease {
pub(super) thread_id: String,
pub(super) endpoint: ResponsesEndpoint,
pub(super) request_kind: RequestMode,
connection: Connection,
pool: Arc<ConnectionPool>,
_permit: OwnedSemaphorePermit,
@@ -177,19 +182,19 @@ impl ConnectionPool {
None => break None,
}
};
let (connection, thread_id, endpoint) = match connection {
let (connection, thread_id, request_kind) = match connection {
Some(connection) => {
let thread_id = connection.thread_id.clone();
let endpoint = connection.endpoint;
(Connection::Websocket(connection), thread_id, endpoint)
let request_kind = connection.request_kind;
(Connection::Websocket(connection), thread_id, request_kind)
}
None => {
self.replenish();
let (mut provider, auth) = self.client_setup().await?;
// Sampling owns the retry budget across both transports.
provider.retry.max_attempts = 0;
let endpoint = self.responses_endpoint().await;
let url = provider.url_for_path(endpoint.path());
let request_kind = self.responses_request_kind().await;
let url = provider.url_for_path("/responses");
let transport = {
let mut cached = self
.http_transport
@@ -222,18 +227,17 @@ impl ConnectionPool {
})
.await?
.clone();
let client =
ResponsesClient::new(transport, provider, auth).with_endpoint(endpoint);
let client = ResponsesClient::new(transport, provider, auth);
(
Connection::Http(client),
ThreadId::new().to_string(),
endpoint,
request_kind,
)
}
};
Ok(ConnectionLease {
thread_id,
endpoint,
request_kind,
connection,
pool: Arc::clone(self),
_permit: permit,
@@ -261,11 +265,18 @@ impl ConnectionPool {
Ok((provider, auth))
}
fn headers(&self, thread_id: &str) -> Result<HeaderMap, LunaSamplerError> {
fn headers(
&self,
thread_id: &str,
request_kind: RequestMode,
) -> Result<HeaderMap, LunaSamplerError> {
let mut headers = build_session_headers(
Some(self.config.session_id.clone()),
Some(thread_id.to_owned()),
);
if request_kind == RequestMode::GuardianClassifier {
headers.insert("x-codex-guardian", HeaderValue::from_static("classifier"));
}
headers.insert("x-openai-subagent", HeaderValue::from_static("guardian"));
headers.insert(
"x-codex-window-id",
@@ -287,16 +298,15 @@ impl ConnectionPool {
}
Ok(headers)
}
async fn responses_endpoint(&self) -> ResponsesEndpoint {
async fn responses_request_kind(&self) -> RequestMode {
let provider = self.config.provider.info();
if self.config.free_guardian
&& self
.config
.provider
.auth()
.await
.as_ref()
.is_some_and(CodexAuth::uses_codex_backend)
if self
.config
.provider
.auth()
.await
.as_ref()
.is_some_and(CodexAuth::uses_codex_backend)
&& provider.supports_codex_backend_routes()
&& provider.requires_openai_auth
&& provider.env_key.is_none()
@@ -304,9 +314,9 @@ impl ConnectionPool {
&& provider.auth.is_none()
&& provider.aws.is_none()
{
ResponsesEndpoint::GuardianClassifier
RequestMode::GuardianClassifier
} else {
ResponsesEndpoint::Responses
RequestMode::Regular
}
}
@@ -318,15 +328,15 @@ impl ConnectionPool {
let auth_changes = auth_manager.map(|manager| manager.auth_change_receiver());
let (provider, auth) = self.client_setup().await?;
let thread_id = ThreadId::new().to_string();
let mut headers = self.headers(&thread_id)?;
let request_kind = self.responses_request_kind().await;
let mut headers = self.headers(&thread_id, request_kind)?;
headers.insert(
"openai-beta",
HeaderValue::from_static(RESPONSES_WEBSOCKETS_BETA),
);
let provider_info = self.config.provider.info();
let endpoint = self.responses_endpoint().await;
let client = ResponsesWebsocketClient::new(provider, auth).with_endpoint(endpoint);
let client = ResponsesWebsocketClient::new(provider, auth);
let connect = client.connect(
&self.config.http_client_factory,
headers,
@@ -341,7 +351,7 @@ impl ConnectionPool {
.and_then(|result| result.map_err(LunaSamplerError::Api));
if let Some(metrics) = self.config.metrics.as_deref() {
let outcome = if result.is_ok() { "success" } else { "failure" };
let mut tags = vec![("endpoint", endpoint.path()), ("outcome", outcome)];
let mut tags = vec![("endpoint", "/responses"), ("outcome", outcome)];
if let Err(error) = &result {
tags.push(("failure_reason", sampler_failure_reason(error)));
}
@@ -363,7 +373,7 @@ impl ConnectionPool {
Ok(PooledConnection {
connection,
endpoint,
request_kind,
thread_id,
expires_at: Instant::now() + MAX_WEBSOCKET_AGE,
auth_changes,
@@ -399,7 +409,7 @@ impl ConnectionLease {
thread_id: Some(self.thread_id.clone()),
extra_headers: self
.pool
.headers(&self.thread_id)
.headers(&self.thread_id, self.request_kind)
.map_err(|error| ApiError::Stream(error.to_string()))?,
..Default::default()
},
codex-rs/ext/guardian-v2/src/async_scorer/sampler/connection_pool_tests.rs
+11 -8
View File
@@ -81,7 +81,7 @@ impl Drop for Gateway {
#[tokio::test]
async fn cold_pool_uses_http_during_open_timeout_then_recovers_after_cooldown() -> Result<()> {
skip_if_no_network!(Ok(()));
for free_guardian in [false, true] {
for uses_codex_backend in [false, true] {
let http = responses::start_mock_server().await;
let events = vec![
responses::ev_output_text_delta("low"),
@@ -93,7 +93,7 @@ async fn cold_pool_uses_http_during_open_timeout_then_recovers_after_cooldown()
connections.push(vec![events.clone()]);
let ws = responses::start_websocket_server(connections).await;
let gateway = Gateway::new(&http.uri(), ws.uri()).await?;
let base_path = if free_guardian {
let base_path = if uses_codex_backend {
"/backend-api/codex"
} else {
"/v1"
@@ -102,13 +102,12 @@ async fn cold_pool_uses_http_during_open_timeout_then_recovers_after_cooldown()
let mut config = sampler_config(base_url.clone());
config.provider = create_model_provider(
ModelProviderInfo::create_openai_provider(Some(base_url)),
Some(AuthManager::from_auth_for_testing(if free_guardian {
Some(AuthManager::from_auth_for_testing(if uses_codex_backend {
CodexAuth::create_dummy_chatgpt_auth_for_testing()
} else {
CodexAuth::from_api_key("test-api-key")
})),
);
config.free_guardian = free_guardian;
config.service_tier = Some("priority".to_owned());
let sampler = LunaSampler::new(config);
let opener = sampler.connections.replenish().unwrap();
@@ -145,17 +144,21 @@ async fn cold_pool_uses_http_during_open_timeout_then_recovers_after_cooldown()
let requests = http_mock.requests();
let first = &requests[0];
let expected_path = if free_guardian {
"/backend-api/codex/guardian-classifier"
let expected_path = if uses_codex_backend {
"/backend-api/codex/responses"
} else {
"/v1/responses"
};
assert_eq!(first.path(), expected_path);
assert_eq!(
first.header("x-codex-guardian").as_deref(),
uses_codex_backend.then_some("classifier")
);
assert!(first.header("authorization").is_some());
let body = first.body_json();
assert_eq!(
body["service_tier"].as_str(),
if free_guardian {
if uses_codex_backend {
None
} else {
Some("priority")
@@ -163,7 +166,7 @@ async fn cold_pool_uses_http_during_open_timeout_then_recovers_after_cooldown()
);
assert_eq!(
body["client_metadata"]["parent_response_id"].as_str(),
free_guardian.then_some("resp-parent")
uses_codex_backend.then_some("resp-parent")
);
gateway.allowed_opens.store(usize::MAX, Ordering::SeqCst);
codex-rs/ext/guardian-v2/src/async_scorer/sampler/execution.rs
+3 -3
View File
@@ -6,10 +6,10 @@ use super::ConnectionPool;
use super::LunaSamplerConfig;
use super::LunaSamplerError;
use super::MAX_OUTPUT_BYTES;
use super::connection_pool::RequestMode;
use codex_api::ApiError;
use codex_api::ResponseEvent;
use codex_api::ResponsesApiRequest;
use codex_api::ResponsesEndpoint;
use codex_api::TransportError;
use codex_extension_api::ExtensionMetrics;
use codex_login::UnauthorizedRecovery;
@@ -127,7 +127,7 @@ impl SamplingExecution {
return Err(error);
}
};
self.request.service_tier = if lease.endpoint == ResponsesEndpoint::GuardianClassifier {
self.request.service_tier = if lease.request_kind == RequestMode::GuardianClassifier {
None
} else {
self.config.service_tier.clone()
@@ -157,7 +157,7 @@ impl SamplingExecution {
turn_metadata["root_turn_id"] = json!(root_turn_id);
}
client_metadata.insert(TURN_METADATA_KEY.to_owned(), turn_metadata.to_string());
if lease.endpoint == ResponsesEndpoint::GuardianClassifier
if lease.request_kind == RequestMode::GuardianClassifier
&& let Some(parent_response_id) = &self.parent_response_id
{
client_metadata.insert("parent_response_id".to_owned(), parent_response_id.clone());
codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs
+16 -24
View File
@@ -206,7 +206,7 @@ pub(super) fn sampler_config(base_url: String) -> LunaSamplerConfig {
session_id: "session-1".to_owned(),
thread_id: "thread-1".to_owned(),
originator: Some("guardian-v2-test".to_owned()),
free_guardian: false,
service_tier: None,
luna_compaction_hash: None,
max_input_tokens: codex_guardian_context::DEFAULT_MAX_INPUT_TOKENS,
@@ -363,36 +363,26 @@ impl ExternalAuth for RefreshableAuth {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn classifier_uses_free_endpoint_only_with_codex_backend_auth() -> Result<()> {
async fn classifier_sends_guardian_header_only_with_codex_backend_auth() -> Result<()> {
skip_if_no_network!(Ok(()));
for (auth, base_path, free_guardian, expected_path, expected_service_tier) in [
for (auth, base_path, expected_header, expected_service_tier) in [
(
CodexAuth::create_dummy_chatgpt_auth_for_testing(),
"/backend-api/codex",
false,
"/backend-api/codex/responses",
Some("priority"),
),
(
CodexAuth::create_dummy_chatgpt_auth_for_testing(),
"/backend-api/codex",
true,
"/backend-api/codex/guardian-classifier",
Some("classifier"),
None,
),
(
CodexAuth::create_dummy_chatgpt_auth_for_testing(),
"/v1",
true,
"/v1/responses",
None,
Some("priority"),
),
(
CodexAuth::from_api_key("test-api-key"),
"/v1",
true,
"/v1/responses",
None,
Some("priority"),
),
] {
@@ -412,13 +402,16 @@ async fn classifier_uses_free_endpoint_only_with_codex_backend_auth() -> Result<
ModelProviderInfo::create_openai_provider(Some(base_url)),
Some(AuthManager::from_auth_for_testing(auth)),
);
config.free_guardian = free_guardian;
config.service_tier = Some("priority".to_owned());
let sampler = connect_sampler(config).await?;
assert_eq!(sampler.sample(sample_request("turn-1")).await?, "low");
for handshake in server.handshakes() {
assert_eq!(handshake.uri(), expected_path);
assert_eq!(handshake.uri(), format!("{base_path}/responses"));
assert_eq!(
handshake.header("x-codex-guardian").as_deref(),
expected_header
);
assert_eq!(handshake.header("x-codex-routing-hint"), None);
}
let request = server
@@ -489,7 +482,7 @@ async fn preconnected_sampler_reuses_authenticated_websocket_for_classifications
session_id: "session-1".to_owned(),
thread_id: "thread-1".to_owned(),
originator: Some("guardian-v2-test".to_owned()),
free_guardian: false,
service_tier: None,
luna_compaction_hash: None,
max_input_tokens: codex_guardian_context::DEFAULT_MAX_INPUT_TOKENS,
@@ -733,7 +726,7 @@ async fn sampler_returns_classification_token_before_terminal_response_events()
session_id: "session-1".to_owned(),
thread_id: "thread-1".to_owned(),
originator: None,
free_guardian: false,
service_tier: None,
luna_compaction_hash: None,
max_input_tokens: codex_guardian_context::DEFAULT_MAX_INPUT_TOKENS,
@@ -1144,7 +1137,7 @@ async fn sampler_limits_transient_recovery_attempts() -> Result<()> {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn parent_response_id_survives_classifier_transport_retry() -> Result<()> {
skip_if_no_network!(Ok(()));
for free_guardian in [false, true] {
for uses_codex_backend in [false, true] {
let healthy = responses::start_websocket_server(vec![vec![vec![
ev_assistant_message("resp-review", "low"),
ev_completed("resp-review"),
@@ -1156,8 +1149,7 @@ async fn parent_response_id_survives_classifier_transport_retry() -> Result<()>
})]]]).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;
if uses_codex_backend {
config.provider = create_model_provider(
ModelProviderInfo::create_openai_provider(Some(format!(
"{}/backend-api/codex",
@@ -1182,7 +1174,7 @@ async fn parent_response_id_survives_classifier_transport_retry() -> Result<()>
body["client_metadata"].get("parent_response_id").cloned(),
body["client_metadata"].get("guardian_credits_requested"),
),
(free_guardian.then(|| json!(parent_response_id)), None)
(uses_codex_backend.then(|| json!(parent_response_id)), None)
);
assert!(!body["input"].to_string().contains(parent_response_id));
}
codex-rs/ext/guardian-v2/src/async_scorer/startup.rs
-1
View File
@@ -56,7 +56,6 @@ pub(super) async fn sampler_config(
.thread_store
.get::<ThreadOriginator>()
.map(|originator| originator.0.clone()),
free_guardian: input.config.free_guardian_enabled(),
service_tier: input.config.service_tier.clone(),
luna_compaction_hash,
max_input_tokens,
codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs
+14
View File
@@ -5,6 +5,7 @@
use std::sync::Arc;
use std::sync::Weak;
use codex_core::CodexResponsesHeaders;
use codex_core::ThreadManager;
use codex_core::config::Config;
use codex_core::config::Constrained;
@@ -85,6 +86,19 @@ impl ThreadLifecycleContributor<Config> for GuardianExtension {
options.session_source =
Some(SessionSource::Internal(InternalSessionSource::Guardian));
options.thread_source = Some(ThreadSource::GuardianReview);
// This is the backend reviewer model, independent of current login.
// Core checks the selected model and auth on each request attempt.
let provider = codex_model_provider::create_model_provider(
options.config.model_provider.clone(),
/*auth_manager*/ None,
);
options.thread_extension_init.insert(CodexResponsesHeaders {
model: provider.approval_review_preferred_model().to_owned(),
headers: http::HeaderMap::from_iter([(
http::HeaderName::from_static("x-codex-guardian"),
http::HeaderValue::from_static("reviewer"),
)]),
});
options
.thread_extension_init
.insert(SessionIsolation::Isolated);
codex-rs/features/src/feature_configs.rs
+1 -1
View File
@@ -127,7 +127,7 @@ pub struct GuardianV2ReviewScopeConfigToml {
pub struct GuardianV2ConfigToml {
#[serde(skip_serializing_if = "Option::is_none")]
pub enabled: Option<bool>,
/// Route Guardian review and classification through the unmetered Codex endpoints.
/// Legacy setting retained for config compatibility; the backend now controls Guardian billing.
#[serde(skip_serializing_if = "Option::is_none")]
pub free_guardian: Option<bool>,
/// Use thread-owned context for sync and async Guardian. Defaults to false.