From eeded5ba1a1db09c627e9e7caa80c66cd7d3291d Mon Sep 17 00:00:00 2001 From: jif Date: Tue, 15 Sep 2026 16:54:10 +0000 Subject: [PATCH] Route Guardian requests through `/responses` with identifying headers (#45736) ## 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 --- codex-rs/codex-api/src/endpoint/mod.rs | 1 - codex-rs/codex-api/src/endpoint/responses.rs | 38 +---- .../src/endpoint/responses_websocket.rs | 27 +--- codex-rs/codex-api/src/lib.rs | 1 - codex-rs/codex-api/tests/clients.rs | 21 +-- codex-rs/core/config.schema.json | 2 +- codex-rs/core/src/client.rs | 132 +++++++++--------- codex-rs/core/src/client_tests.rs | 86 ++++-------- codex-rs/core/src/config/mod.rs | 11 -- codex-rs/core/src/lib.rs | 2 + codex-rs/core/src/responses_headers.rs | 24 ++++ codex-rs/core/src/session/session.rs | 3 +- codex-rs/core/tests/common/responses.rs | 2 +- codex-rs/core/tests/suite/guardian_review.rs | 28 ++-- .../guardian-v2/src/async_scorer/sampler.rs | 2 - .../async_scorer/sampler/connection_pool.rs | 70 ++++++---- .../sampler/connection_pool_tests.rs | 19 +-- .../src/async_scorer/sampler/execution.rs | 6 +- .../src/async_scorer/sampler_tests.rs | 40 +++--- .../guardian-v2/src/async_scorer/startup.rs | 1 - .../ext/guardian-v2/src/sync_reviewer/mod.rs | 14 ++ codex-rs/features/src/feature_configs.rs | 2 +- 22 files changed, 246 insertions(+), 286 deletions(-) create mode 100644 codex-rs/core/src/responses_headers.rs diff --git a/codex-rs/codex-api/src/endpoint/mod.rs b/codex-rs/codex-api/src/endpoint/mod.rs index c9f44a8f91..53a2b72b88 100644 --- a/codex-rs/codex-api/src/endpoint/mod.rs +++ b/codex-rs/codex-api/src/endpoint/mod.rs @@ -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; diff --git a/codex-rs/codex-api/src/endpoint/responses.rs b/codex-rs/codex-api/src/endpoint/responses.rs index 78a08232f9..9c41a802ea 100644 --- a/codex-rs/codex-api/src/endpoint/responses.rs +++ b/codex-rs/codex-api/src/endpoint/responses.rs @@ -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 { session: EndpointSession, sse_telemetry: Option>, - endpoint: ResponsesEndpoint, } #[derive(Default)] @@ -67,16 +43,9 @@ impl ResponsesClient { 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>, @@ -85,7 +54,6 @@ impl ResponsesClient { Self { session: self.session.with_request_telemetry(request), sse_telemetry: sse, - endpoint: self.endpoint, } } @@ -96,7 +64,7 @@ impl ResponsesClient { 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 ResponsesClient { 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 ResponsesClient { .session .stream_encoded_json_with( Method::POST, - self.endpoint.path(), + "/responses", extra_headers, Some(body), |req| { diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index 95d41e0d91..48e99d0106 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -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>>, - 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", &"") - .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, telemetry: Option>, - 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 { 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 { 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 = diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index 92507807f6..165bf6cc50 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -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; diff --git a/codex-rs/codex-api/tests/clients.rs b/codex-rs/codex-api/tests/clients.rs index af43022417..d2ac1175c0 100644 --- a/codex-rs/codex-api/tests/clients.rs +++ b/codex-rs/codex-api/tests/clients.rs @@ -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(()) } diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 72fdd44907..b53c690aea 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -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": { diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 9e4cf7df6f..b60f810b99 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -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, agent_identity_policy: AgentIdentityAuthPolicy, prompt_cache_key_override: Option, - free_guardian_enabled: bool, + codex_responses_headers: Option>, event_sender: Option>, http_client_factory: HttpClientFactory, } @@ -292,7 +292,7 @@ struct LastResponse { #[derive(Debug, Default)] struct WebsocketSession { connection: Option, - endpoint: Option, + responses_headers: ApiHeaderMap, /// Owner of the cached state, including before a connection is opened. auth_owner_generation: Option, last_request: Option, @@ -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, event_sender: Sender, + codex_responses_headers: Option>, ) -> 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>, 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 { - 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( diff --git a/codex-rs/core/src/client_tests.rs b/codex-rs/core/src/client_tests.rs index 8936f5afab..36d54c9cc4 100644 --- a/codex-rs/core/src/client_tests.rs +++ b/codex-rs/core/src/client_tests.rs @@ -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(), ); } diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 12a1472ac4..38a832452d 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -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, diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index a8f76ae359..ea8e1b29f5 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -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; diff --git a/codex-rs/core/src/responses_headers.rs b/codex-rs/core/src/responses_headers.rs new file mode 100644 index 0000000000..44d9a5d28e --- /dev/null +++ b/codex-rs/core/src/responses_headers.rs @@ -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() + } +} diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 7e0cd79a0a..259ef9ba83 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1600,6 +1600,7 @@ impl Session { &config.features, &initial_history, ); + let codex_responses_headers = thread_extension_data.get::(); 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( diff --git a/codex-rs/core/tests/common/responses.rs b/codex-rs/core/tests/common/responses.rs index f22e511f65..408a6234b3 100644 --- a/codex-rs/core/tests/common/responses.rs +++ b/codex-rs/core/tests/common/responses.rs @@ -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) } diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index a88ff05af7..b7d44dfbe1 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/codex-rs/core/tests/suite/guardian_review.rs @@ -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); } } diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs index 4ad0c0e2ea..5c88c51917 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs @@ -52,8 +52,6 @@ pub struct LunaSamplerConfig { pub thread_id: String, /// Optional host-resolved request originator. pub originator: Option, - /// Whether this thread may use the unmetered Guardian classifier endpoint. - pub free_guardian: bool, /// Optional inference service tier. pub service_tier: Option, /// Luna model's host-resolved encrypted-compaction compatibility hash. diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler/connection_pool.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler/connection_pool.rs index 13adc5f1bc..0358d1ecc2 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler/connection_pool.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler/connection_pool.rs @@ -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, pub(super) idle_connections: Mutex>, @@ -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, _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 { + fn headers( + &self, + thread_id: &str, + request_kind: RequestMode, + ) -> Result { 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() }, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler/connection_pool_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler/connection_pool_tests.rs index 42ed27c7ee..97d6eacefa 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler/connection_pool_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler/connection_pool_tests.rs @@ -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); diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler/execution.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler/execution.rs index 642587e94b..cd9d5451d4 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler/execution.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler/execution.rs @@ -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()); diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs index 9a5018b028..e7183f4b21 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs @@ -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)); } diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/startup.rs b/codex-rs/ext/guardian-v2/src/async_scorer/startup.rs index 34a87d643a..df05e4a933 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/startup.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/startup.rs @@ -56,7 +56,6 @@ pub(super) async fn sampler_config( .thread_store .get::() .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, diff --git a/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs b/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs index ec162a6a0c..f1bf7772c8 100644 --- a/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs +++ b/codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs @@ -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 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); diff --git a/codex-rs/features/src/feature_configs.rs b/codex-rs/features/src/feature_configs.rs index 1482439206..e6977676cc 100644 --- a/codex-rs/features/src/feature_configs.rs +++ b/codex-rs/features/src/feature_configs.rs @@ -127,7 +127,7 @@ pub struct GuardianV2ReviewScopeConfigToml { pub struct GuardianV2ConfigToml { #[serde(skip_serializing_if = "Option::is_none")] pub enabled: Option, - /// 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, /// Use thread-owned context for sync and async Guardian. Defaults to false.