refactor(gateway): remove test-only production APIs

Move protocol and provider fixtures and observations into integration tests while exercising the production decoder, encoder, transport, DNS, and lease paths.

Remove direct credential seeding and routing test setters while preserving OpenAI transparent proxy and WebSocket/SSE fallback behavior.

Co-authored-by: Codex <noreply@openai.com>

backend/crates/gateway-api/src/openai/responses/mod.rs

@@ -10,12 +10,12 @@ pub use error::{ProtocolErrorBody, RequestDecodeError, ResponseEncodeError};
 pub use http::{collect_execution_response, stream_execution_response};
 pub(crate) use http::{responses, review_responses};
 pub use request::{
-    ContinuationIntent, DecodedResponsesRequest, PROVIDER_OPTIONS_VERSION,
-    ResponsesRequestMetadata, decode_request,
+    ContinuationIntent, DecodedResponsesRequest, OpenAiRequestHeaders, PROVIDER_OPTIONS_VERSION,
+    ResponsesRequestMetadata, decode_request_with_headers,
 };
-pub use response::{CollectedResponses, OpenAiResponsesEncoder, ResponsesCollector};
+pub use response::{OpenAiResponsesEncoder, ResponsesCollector};
 pub(crate) use websocket::responses_websocket;
-pub use websocket::{ResponseCreateFrameError, decode_response_create};
+pub use websocket::{ResponseCreateFrameError, decode_response_create_with_context};
 
 pub(super) fn safe_response_header_name(name: &str) -> Option<&'static str> {
     match name {

backend/crates/gateway-api/src/openai/responses/request.rs

@@ -30,7 +30,7 @@ const OPENAI_SUBAGENT_KEY: &str = "x-openai-subagent";
 const MAX_PROMPT_CACHE_KEY_BYTES: usize = 1_024;
 
 #[derive(Clone, Default)]
-pub(super) struct OpenAiRequestHeaders {
+pub struct OpenAiRequestHeaders {
     turn_state: Option<String>,
     turn_metadata: Option<String>,
     beta_features: Option<String>,
@@ -52,7 +52,9 @@ pub(super) struct OpenAiRequestHeaders {
 }
 
 impl OpenAiRequestHeaders {
-    pub(super) fn from_headers(headers: &HeaderMap) -> Self {
+    /// 提取 OpenAI/Codex 连接级请求头上下文。
+    #[must_use]
+    pub fn from_headers(headers: &HeaderMap) -> Self {
         let prompt_cache_seed = [
             "x-claude-code-session-id",
             "x-session-id",
@@ -307,17 +309,12 @@ impl fmt::Debug for DecodedResponsesRequest {
     }
 }
 
-/// 解码普通 `POST /v1/responses` 请求。
+/// 使用下游 OpenAI/Codex 请求头解码 `POST /v1/responses`。
 ///
 /// # Errors
 ///
 /// JSON 非法、顶层不是 object，或网关必须解释的路由字段无效时返回安全错误。
-pub fn decode_request(body: &[u8]) -> Result<DecodedResponsesRequest, RequestDecodeError> {
-    decode_request_inner(body, false, &OpenAiRequestHeaders::default(), None)
-}
-
-/// 使用下游 OpenAI/Codex 请求头解码 `POST /v1/responses`。
-pub(super) fn decode_request_with_headers(
+pub fn decode_request_with_headers(
     body: &[u8],
     headers: &HeaderMap,
     provider_kind: &ProviderKind,

backend/crates/gateway-api/src/openai/responses/response.rs

@@ -270,33 +270,6 @@ fn rewrite_response_identity(value: &mut Value, upstream: &str, gateway: &str) {
     }
 }
 
-/// 同一 canonical event 流生成的两种 Responses 表达。
-#[derive(Debug, Clone, PartialEq)]
-pub struct CollectedResponses {
-    response: Value,
-    sse_frames: Vec<String>,
-}
-
-impl CollectedResponses {
-    /// 返回非流式 Responses JSON。
-    #[must_use]
-    pub const fn response(&self) -> &Value {
-        &self.response
-    }
-
-    /// 返回按事件边界编码的 SSE frame。
-    #[must_use]
-    pub fn sse_frames(&self) -> &[String] {
-        &self.sse_frames
-    }
-
-    /// 拆分两种表达。
-    #[must_use]
-    pub fn into_parts(self) -> (Value, Vec<String>) {
-        (self.response, self.sse_frames)
-    }
-}
-
 /// OpenAI Responses event collector。
 ///
 /// Collector 只做协议编码；它不拥有 commit、retry 或 Provider 生命周期。
@@ -352,31 +325,6 @@ impl ResponsesCollector {
         }
     }
 
-    /// 消费一个 canonical event，并返回 Responses WebSocket 文本消息。
-    ///
-    /// WebSocket 与 SSE 共用同一状态机和事件 JSON；这里只移除 SSE framing，
-    /// 避免两个下游 transport 各自维护一套协议投影。
-    ///
-    /// # Errors
-    ///
-    /// 继承 [`Self::push`] 的 canonical 编码错误；内部 SSE framing 若不满足编码器
-    /// 自身的不变量则返回 [`ResponseEncodeError::InvalidEventEncoding`]。
-    pub fn push_websocket_events(
-        &mut self,
-        event: &GatewayEvent,
-    ) -> Result<Vec<String>, ResponseEncodeError> {
-        self.push(event)?
-            .into_iter()
-            .map(|frame| {
-                frame
-                    .lines()
-                    .find_map(|line| line.strip_prefix("data: "))
-                    .map(ToOwned::to_owned)
-                    .ok_or(ResponseEncodeError::InvalidEventEncoding)
-            })
-            .collect()
-    }
-
     /// 校验终态并返回非流式 JSON。
     ///
     /// # Errors
@@ -388,27 +336,6 @@ impl ResponsesCollector {
             .ok_or_else(|| gateway_core::event::EventSequenceError::MissingCompleted.into())
     }
 
-    /// 一次性收集完整事件序列，同时保留 stateful collector 的相同实现路径。
-    ///
-    /// # Errors
-    ///
-    /// 任意事件无法编码或序列不完整时返回错误。
-    pub fn collect<'a>(
-        created_at: u64,
-        events: impl IntoIterator<Item = &'a GatewayEvent>,
-    ) -> Result<CollectedResponses, ResponseEncodeError> {
-        let mut collector = Self::new(created_at);
-        let mut sse_frames = Vec::new();
-        for event in events {
-            sse_frames.extend(collector.push(event)?);
-        }
-        let response = collector.finish()?;
-        Ok(CollectedResponses {
-            response,
-            sse_frames,
-        })
-    }
-
     fn started(&mut self, meta: &ResponseMeta) -> Result<Vec<String>, ResponseEncodeError> {
         self.started = Some(meta.clone());
         let response = response_base(

backend/crates/gateway-api/src/openai/responses/websocket.rs

@@ -522,7 +522,7 @@ async fn next_active_input(
     }
 }
 
-/// 解码一个官方 `response.create` 文本帧。
+/// 使用连接级请求头和 Provider 上下文解码官方 `response.create` 文本帧。
 ///
 /// 缺省 `stream` 等价于 WebSocket 固有的流式语义；显式 `false` 会被拒绝。
 ///
@@ -530,13 +530,7 @@ async fn next_active_input(
 ///
 /// 帧不是合法 JSON object、消息类型错误、显式关闭 stream，或 Responses 请求
 /// 无法映射到 canonical operation 时返回不包含正文内容的稳定错误。
-pub fn decode_response_create(
-    payload: &str,
-) -> Result<DecodedResponsesRequest, ResponseCreateFrameError> {
-    decode_response_create_inner(payload, &OpenAiRequestHeaders::default(), None)
-}
-
-fn decode_response_create_with_context(
+pub fn decode_response_create_with_context(
     payload: &str,
     request_headers: &OpenAiRequestHeaders,
     provider_kind: &ProviderKind,

backend/crates/gateway-api/tests/openai/responses/mod.rs

@@ -1,9 +1,11 @@
 mod http;
 mod websocket;
 
+use axum::http::HeaderMap;
 use gateway_api::openai::responses::{
-    ContinuationIntent, OpenAiResponsesEncoder, RequestDecodeError, ResponseEncodeError,
-    ResponsesCollector, decode_request, decode_response_create,
+    ContinuationIntent, DecodedResponsesRequest, OpenAiRequestHeaders, OpenAiResponsesEncoder,
+    RequestDecodeError, ResponseCreateFrameError, ResponseEncodeError, ResponsesCollector,
+    decode_request_with_headers, decode_response_create_with_context,
 };
 use gateway_core::{
     accounting::{CalculatedCost, ProviderReportedCost, Usage},
@@ -17,10 +19,29 @@ use gateway_core::{
         ContentPart, Feature, MessageRole, Operation, OutputFormat, ReasoningEffort,
         ReasoningSummary, ResponsePersistence,
     },
+    routing::ProviderKind,
 };
 use gateway_protocol::openai::sse::parse_sse_events;
 use serde_json::{Value, json};
 
+fn openai_provider() -> ProviderKind {
+    ProviderKind::new("openai").expect("OpenAI provider kind")
+}
+
+fn decode_request(body: &[u8]) -> Result<DecodedResponsesRequest, RequestDecodeError> {
+    decode_request_with_headers(body, &HeaderMap::new(), &openai_provider())
+}
+
+fn decode_response_create(
+    payload: &str,
+) -> Result<DecodedResponsesRequest, ResponseCreateFrameError> {
+    decode_response_create_with_context(
+        payload,
+        &OpenAiRequestHeaders::default(),
+        &openai_provider(),
+    )
+}
+
 fn generate_request(body: Value) -> gateway_api::openai::responses::DecodedResponsesRequest {
     decode_request(body.to_string().as_bytes()).expect("test request should decode")
 }
@@ -61,6 +82,36 @@ fn usage() -> Usage {
     usage
 }
 
+struct CollectedResponses {
+    response: Value,
+    sse_frames: Vec<String>,
+}
+
+impl CollectedResponses {
+    fn response(&self) -> &Value {
+        &self.response
+    }
+
+    fn sse_frames(&self) -> &[String] {
+        &self.sse_frames
+    }
+}
+
+fn collect_responses(
+    created_at: u64,
+    events: &[GatewayEvent],
+) -> Result<CollectedResponses, ResponseEncodeError> {
+    let mut collector = ResponsesCollector::new(created_at);
+    let mut sse_frames = Vec::new();
+    for event in events {
+        sse_frames.extend(collector.push(event)?);
+    }
+    Ok(CollectedResponses {
+        response: collector.finish()?,
+        sse_frames,
+    })
+}
+
 fn text_events() -> Vec<GatewayEvent> {
     vec![
         GatewayEvent::Started(response_meta()),
@@ -706,8 +757,8 @@ fn request_errors_should_not_disclose_prompt_or_raw_body() {
 #[test]
 fn stream_terminal_response_should_equal_non_stream_response() {
     let events = text_events();
-    let collected = ResponsesCollector::collect(1_700_000_000, &events)
-        .expect("canonical events should encode");
+    let collected =
+        collect_responses(1_700_000_000, &events).expect("canonical events should encode");
     let sse = collected.sse_frames().join("");
     let parsed = parse_sse_events(&sse).expect("generated SSE should parse");
     let terminal = parsed
@@ -722,13 +773,16 @@ fn stream_terminal_response_should_equal_non_stream_response() {
 #[test]
 fn websocket_events_should_reuse_the_exact_sse_event_json() {
     let events = text_events();
-    let mut sse_collector = ResponsesCollector::new(1_700_000_000);
-    let mut websocket_collector = ResponsesCollector::new(1_700_000_000);
+    let mut sse_encoder = OpenAiResponsesEncoder::new(1_700_000_000);
+    let mut websocket_encoder = OpenAiResponsesEncoder::new(1_700_000_000);
 
     for event in &events {
-        let sse_frames = sse_collector.push(event).expect("SSE event encoding");
-        let websocket_events = websocket_collector
-            .push_websocket_events(event)
+        let provider_event = ProviderEvent::canonical(event.clone());
+        let sse_frames = sse_encoder
+            .push_sse(&provider_event)
+            .expect("SSE event encoding");
+        let websocket_events = websocket_encoder
+            .push_websocket(&provider_event)
             .expect("WebSocket event encoding");
         let parsed = parse_sse_events(&sse_frames.join(""))
             .expect("generated SSE frames remain parseable")
@@ -743,8 +797,8 @@ fn websocket_events_should_reuse_the_exact_sse_event_json() {
 #[test]
 fn stream_should_emit_first_terminal_and_usage_once() {
     let events = text_events();
-    let collected = ResponsesCollector::collect(1_700_000_000, &events)
-        .expect("canonical events should encode");
+    let collected =
+        collect_responses(1_700_000_000, &events).expect("canonical events should encode");
     let parsed =
         parse_sse_events(&collected.sse_frames().join("")).expect("generated SSE should parse");
     let first = parsed
@@ -786,7 +840,7 @@ fn collector_should_encode_reasoning_and_whole_function_call() {
         }),
         GatewayEvent::Completed(response_meta().with_finish_reason(FinishReason::ToolCall)),
     ];
-    let collected = ResponsesCollector::collect(1_700_000_000, &events)
+    let collected = collect_responses(1_700_000_000, &events)
         .expect("reasoning and function call should encode");
 
     assert_eq!(
@@ -810,10 +864,8 @@ fn collector_should_encode_reasoning_and_whole_function_call() {
 #[test]
 fn collector_should_generate_deterministic_output_ids() {
     let events = text_events();
-    let first =
-        ResponsesCollector::collect(1_700_000_000, &events).expect("first encoding should work");
-    let second =
-        ResponsesCollector::collect(1_700_000_000, &events).expect("second encoding should work");
+    let first = collect_responses(1_700_000_000, &events).expect("first encoding should work");
+    let second = collect_responses(1_700_000_000, &events).expect("second encoding should work");
 
     assert_eq!(
         first.response().pointer("/output/0/id"),
@@ -826,8 +878,8 @@ fn collector_should_encode_length_termination_as_incomplete() {
     let mut events = text_events();
     let terminal = events.last_mut().expect("terminal event");
     *terminal = GatewayEvent::Completed(response_meta().with_finish_reason(FinishReason::Length));
-    let collected = ResponsesCollector::collect(1_700_000_000, &events)
-        .expect("length termination should encode");
+    let collected =
+        collect_responses(1_700_000_000, &events).expect("length termination should encode");
     let parsed =
         parse_sse_events(&collected.sse_frames().join("")).expect("generated SSE should parse");
 
@@ -872,7 +924,7 @@ fn collector_should_keep_accounting_costs_out_of_the_responses_wire() {
         GatewayEvent::ProviderCost(cost),
         GatewayEvent::Completed(completed_meta()),
     ];
-    let collected = ResponsesCollector::collect(1_700_000_000, &events)
+    let collected = collect_responses(1_700_000_000, &events)
         .expect("provider cost is accounting-only metadata");
 
     assert!(collected.response().get("cost").is_none());
@@ -1085,7 +1137,7 @@ fn typed_compaction_output_should_encode_the_codex_remote_compaction_v2_contract
         )),
         GatewayEvent::Completed(completed_meta()),
     ];
-    let collected = ResponsesCollector::collect(1_700_000_000, &events)
+    let collected = collect_responses(1_700_000_000, &events)
         .expect("typed compaction output should be representable as OpenAI Responses");
     let parsed =
         parse_sse_events(&collected.sse_frames().join("")).expect("compaction SSE should be valid");

backend/crates/gateway-api/tests/openai/responses/websocket.rs

@@ -1,7 +1,9 @@
-use gateway_api::openai::responses::{ResponseCreateFrameError, decode_response_create};
+use gateway_api::openai::responses::ResponseCreateFrameError;
 use gateway_core::operation::Operation;
 use serde_json::json;
 
+use super::decode_response_create;
+
 #[test]
 fn response_create_should_default_to_the_websocket_streaming_contract() {
     let decoded = decode_response_create(

backend/crates/gateway-core/src/routing/mod.rs

@@ -389,13 +389,6 @@ pub struct RoutingPlan {
 }
 
 impl RoutingPlan {
-    /// 冻结本请求允许的最大上游调用次数。
-    #[must_use]
-    pub fn with_max_attempts(mut self, max_attempts: NonZeroU32) -> Self {
-        self.max_attempts = max_attempts;
-        self
-    }
-
     #[must_use]
     pub const fn config_revision(&self) -> ConfigRevision {
         self.config_revision

backend/crates/gateway-core/tests/engine/coordinator.rs

@@ -335,7 +335,7 @@ fn complete_stream(total_tokens: Option<u64>) -> Vec<Result<GatewayEvent, Provid
     events
 }
 
-fn plan(operation: &Operation, max_attempts: u32, instance_count: u32) -> RoutingPlan {
+fn plan(operation: &Operation, instance_count: u32) -> RoutingPlan {
     let provider = ProviderKind::new("openai").expect("provider");
     let public_model = PublicModelId::new("gpt-5").expect("public model");
     let capabilities = ModelCapabilities::new(
@@ -382,7 +382,6 @@ fn plan(operation: &Operation, max_attempts: u32, instance_count: u32) -> Routin
         },
     )
     .expect("routing plan")
-    .with_max_attempts(NonZeroU32::new(max_attempts).expect("max attempts"))
 }
 
 fn model_request(deadline: SystemTime) -> NewModelRequest {
@@ -432,7 +431,7 @@ fn terminal_non_idempotent_failure(
     continuation: Option<ContinuationBinding>,
 ) -> (Arc<FakeStore>, Arc<ScriptedProvider>) {
     let operation = operation(RetrySafety::NonIdempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Stream {
             account_id: "acct_first",
@@ -463,7 +462,7 @@ fn terminal_non_idempotent_failure(
 #[test]
 fn success_updates_one_model_request_and_persists_usage() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, _) = coordinator(vec![Script::Stream {
         account_id: "acct_one",
         items: complete_stream(Some(12)),
@@ -525,7 +524,7 @@ fn success_updates_one_model_request_and_persists_usage() {
 #[test]
 fn response_observation_is_persisted_but_never_delivered() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 1, 1);
+    let route_plan = plan(&operation, 1);
     let observation = ProviderResponseObservation::new(
         UpstreamTransport::new("http_sse").expect("actual transport"),
     )
@@ -580,7 +579,7 @@ fn response_observation_is_persisted_but_never_delivered() {
 #[test]
 fn unknown_wire_event_before_response_identity_is_discarded_with_retried_attempt() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let unknown = ProtocolWireEvent::json(
         "openai",
         Some("response.future_event".to_owned()),
@@ -626,7 +625,7 @@ fn unknown_wire_event_before_response_identity_is_discarded_with_retried_attempt
 #[test]
 fn discarded_attempt_observation_does_not_leak_into_retry_result() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let first_observation = ProviderResponseObservation::new(
         UpstreamTransport::new("websocket").expect("first transport"),
     )
@@ -707,7 +706,7 @@ fn discarded_attempt_observation_does_not_leak_into_retry_result() {
 #[test]
 fn websocket_success_keeps_client_http_status_absent() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 1, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, _) = coordinator(vec![Script::Stream {
         account_id: "acct_one",
         items: complete_stream(None),
@@ -733,7 +732,7 @@ fn websocket_success_keeps_client_http_status_absent() {
 #[test]
 fn authenticated_native_continuation_reaches_every_attempt_context() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 1, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, _, provider) = coordinator(vec![Script::Stream {
         account_id: "acct_one",
         items: complete_stream(None),
@@ -800,7 +799,7 @@ fn native_continuation_replays_owner_before_safely_switching_account() {
     let operation = Operation::Generate(generate.with_provider_session_state(
         ProviderSessionState::new("openai", payload).expect("provider session state"),
     ));
-    let route_plan = plan(&operation, 3, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, _, provider) = coordinator(vec![
         Script::Stream {
             account_id: "acct_one",
@@ -874,7 +873,7 @@ fn native_continuation_replays_owner_before_safely_switching_account() {
 #[test]
 fn required_account_reaches_provider_and_matching_metadata_succeeds() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 2);
+    let route_plan = plan(&operation, 2);
     let (coordinator, _, provider) = coordinator(vec![Script::Stream {
         account_id: "acct_required",
         items: complete_stream(None),
@@ -900,7 +899,7 @@ fn required_account_reaches_provider_and_matching_metadata_succeeds() {
 #[test]
 fn provider_metadata_for_another_account_fails_closed() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 2);
+    let route_plan = plan(&operation, 2);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Stream {
             account_id: "acct_wrong",
@@ -937,7 +936,7 @@ fn provider_metadata_for_another_account_fails_closed() {
 #[test]
 fn required_account_disables_instance_fallback_before_stream() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 2);
+    let route_plan = plan(&operation, 2);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Error(ProviderError::new(
             ProviderErrorKind::Unavailable,
@@ -977,7 +976,7 @@ fn required_account_disables_instance_fallback_before_stream() {
 #[test]
 fn local_selection_failure_falls_back_without_instance_failure_observation() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 2);
+    let route_plan = plan(&operation, 2);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Error(ProviderError::new(
             ProviderErrorKind::Unavailable,
@@ -1020,7 +1019,7 @@ fn local_selection_failure_falls_back_without_instance_failure_observation() {
 #[test]
 fn required_account_disables_account_retry_after_stream_creation() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Stream {
             account_id: "acct_required",
@@ -1068,7 +1067,7 @@ fn required_account_disables_account_retry_after_stream_creation() {
 #[test]
 fn latest_provider_reported_cost_is_persisted_as_known_usd_total() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 1, 1);
+    let route_plan = plan(&operation, 1);
     let events = vec![
         Ok(GatewayEvent::Started(ResponseMeta::new(
             "native-cost",
@@ -1111,7 +1110,7 @@ fn latest_provider_reported_cost_is_persisted_as_known_usd_total() {
 #[test]
 fn calculated_cost_is_persisted_when_provider_does_not_report_cost() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 1, 1);
+    let route_plan = plan(&operation, 1);
     let events = vec![
         Ok(GatewayEvent::Started(ResponseMeta::new(
             "calculated-cost",
@@ -1149,7 +1148,7 @@ fn calculated_cost_is_persisted_when_provider_does_not_report_cost() {
 #[test]
 fn provider_reported_cost_should_not_be_replaced_by_calculated_cost() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 1, 1);
+    let route_plan = plan(&operation, 1);
     let events = vec![
         Ok(GatewayEvent::Started(ResponseMeta::new(
             "reported-cost",
@@ -1198,7 +1197,7 @@ fn provider_reported_cost_should_not_be_replaced_by_calculated_cost() {
 #[test]
 fn discarded_attempt_cost_never_leaks_into_retry_result() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, _) = coordinator(vec![
         Script::Stream {
             account_id: "acct_first",
@@ -1243,7 +1242,7 @@ fn discarded_attempt_cost_never_leaks_into_retry_result() {
 #[test]
 fn pre_commit_failure_excludes_account_and_retries_same_target() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Stream {
             account_id: "acct_first",
@@ -1308,7 +1307,7 @@ fn pre_commit_failure_excludes_account_and_retries_same_target() {
 #[test]
 fn recovered_credential_retries_the_same_account_exactly_once() {
     let operation = operation(RetrySafety::NonIdempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Stream {
             account_id: "acct_first",
@@ -1353,7 +1352,7 @@ fn recovered_credential_retries_the_same_account_exactly_once() {
 #[test]
 fn non_idempotent_explicit_429_rejection_rotates_account_before_output() {
     let operation = operation(RetrySafety::NonIdempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Stream {
             account_id: "acct_first",
@@ -1433,7 +1432,7 @@ fn explicit_429_with_ambiguous_send_state_is_not_retried() {
 #[test]
 fn explicit_429_after_structural_event_should_retry_before_commit() {
     let operation = operation(RetrySafety::NonIdempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Stream {
             account_id: "acct_first",
@@ -1527,7 +1526,7 @@ fn external_continuation_explicit_429_is_not_retried() {
 #[test]
 fn non_idempotent_not_sent_failure_can_fallback_target() {
     let operation = operation(RetrySafety::NonIdempotent);
-    let route_plan = plan(&operation, 2, 2);
+    let route_plan = plan(&operation, 2);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Error(ProviderError::new(
             ProviderErrorKind::Unavailable,
@@ -1566,7 +1565,7 @@ fn non_idempotent_not_sent_failure_can_fallback_target() {
 #[test]
 fn ambiguous_send_state_stops_retry() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Stream {
             account_id: "acct_first",
@@ -1612,7 +1611,7 @@ fn ambiguous_send_state_stops_retry() {
 #[test]
 fn structural_event_before_replay_safe_failure_should_switch_account_before_commit() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 2, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, provider) = coordinator(vec![
         Script::Stream {
             account_id: "acct_first",
@@ -1671,7 +1670,7 @@ fn structural_event_before_replay_safe_failure_should_switch_account_before_comm
 #[test]
 fn cancellation_before_pending_delivery_commit_reaches_terminal_state() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 1, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, _) = coordinator(vec![Script::Stream {
         account_id: "acct_first",
         items: complete_stream(None),
@@ -1705,7 +1704,7 @@ fn cancellation_before_pending_delivery_commit_reaches_terminal_state() {
 #[test]
 fn local_unavailability_before_stream_does_not_create_attempt_or_instance_failure() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 1, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, _) = coordinator(vec![Script::Error(
         ProviderError::new(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent)
             .with_upstream_code(SafeUpstreamValue::new("overloaded").expect("provider code"))
@@ -1742,7 +1741,7 @@ fn local_unavailability_before_stream_does_not_create_attempt_or_instance_failur
 #[test]
 fn expired_deadline_finalizes_without_calling_provider() {
     let operation = operation(RetrySafety::Idempotent);
-    let route_plan = plan(&operation, 1, 1);
+    let route_plan = plan(&operation, 1);
     let (coordinator, store, provider) = coordinator(vec![]);
 
     let error = match block_on(coordinator.start(

backend/crates/gateway-protocol/src/openai/sse.rs

@@ -22,16 +22,6 @@ pub const MAX_SSE_EVENT_BUFFER_BYTES: usize = 64 * 1024 * 1024;
 /// SSE 流结束标记帧。
 pub const DONE_SSE_FRAME: &str = "data: [DONE]\n\n";
 
-/// 判断一个完整 SSE frame 是否为传输层 `[DONE]` 控制帧。
-pub fn sse_frame_is_done(frame: &str) -> bool {
-    let mut data = frame.lines().filter_map(|raw_line| {
-        let line = raw_line.strip_suffix('\r').unwrap_or(raw_line);
-        let (field, value) = split_sse_field(line);
-        (field == "data").then_some(value)
-    });
-    matches!((data.next(), data.next()), (Some("[DONE]"), None))
-}
-
 /// 用于处理任意分块边界的增量 SSE 解码器。
 #[derive(Debug, Default)]
 pub struct SseEventDecoder {
@@ -121,12 +111,6 @@ pub fn response_failed_sse_data_with_id(
     })
 }
 
-/// 判断 SSE 文本是否已经包含结束标记。
-pub fn sse_body_has_done(body: &str) -> bool {
-    body.trim_end_matches(['\r', '\n'])
-        .ends_with(DONE_SSE_FRAME.trim_end_matches(['\r', '\n']))
-}
-
 /// 返回下一个完整 SSE 帧结束位置（含分隔符）。
 pub fn sse_frame_end(bytes: &[u8]) -> Option<usize> {
     sse_frame_separator_bytes(bytes).map(|(position, separator_len)| position + separator_len)

backend/crates/gateway-protocol/tests/openai/sse.rs

@@ -1,9 +1,25 @@
 use gateway_protocol::openai::sse::{
     DONE_SSE_FRAME, MAX_SSE_EVENT_BUFFER_BYTES, SseError, SseEvent, SseEventDecoder,
     encode_sse_event, encode_sse_event_with_metadata, parse_sse_events, response_failed_sse_event,
-    response_failed_sse_event_with_id, sse_body_has_done, sse_frame_end, sse_frame_is_done,
+    response_failed_sse_event_with_id, sse_frame_end,
 };
 
+fn sse_frame_is_done(frame: &str) -> bool {
+    let mut data = frame.lines().filter_map(|raw_line| {
+        let line = raw_line.strip_suffix('\r').unwrap_or(raw_line);
+        let (field, value) = line.split_once(':').map_or((line, ""), |(field, value)| {
+            (field, value.strip_prefix(' ').unwrap_or(value))
+        });
+        (field == "data").then_some(value)
+    });
+    matches!((data.next(), data.next()), (Some("[DONE]"), None))
+}
+
+fn sse_body_has_done(body: &str) -> bool {
+    body.trim_end_matches(['\r', '\n'])
+        .ends_with(DONE_SSE_FRAME.trim_end_matches(['\r', '\n']))
+}
+
 #[test]
 fn parser_should_preserve_multiline_data_id_and_retry() {
     let input = concat!(

backend/crates/providers/openai/src/credential/admin.rs

@@ -278,11 +278,6 @@ impl std::fmt::Debug for PreparedCodexCredentialRotation {
 }
 
 impl PreparedCodexCredentialRotation {
-    #[must_use]
-    pub const fn holds_refresh_lease(&self) -> bool {
-        self.refresh_guards.is_some()
-    }
-
     /// 将 command 与 lease 一起交给 App；App 必须让返回的 guard 活到 CAS 提交结束。
     #[must_use]
     pub fn into_parts(

backend/crates/providers/openai/src/credential/mod.rs

@@ -56,6 +56,6 @@ pub use selector::{
 };
 pub use types::{
     CodexAccountProfile, CodexCookie, CodexCookieCaptureOutcome, CodexCredentialData,
-    CodexCredentialPrincipal, CodexOAuthSecret, CreateCodexCredential, CredentialRecord,
-    RotateCodexCredential, RuntimeCodexCookie, UpsertCodexCookie,
+    CodexCredentialPrincipal, CodexOAuthSecret, RotateCodexCredential, RuntimeCodexCookie,
+    UpsertCodexCookie,
 };

backend/crates/providers/openai/src/credential/repository.rs

@@ -6,18 +6,17 @@ use std::time::SystemTime;
 use chrono::{DateTime, Utc};
 use gateway_core::engine::credential::{
     AccountAvailability, AccountStateChange, CredentialCasOutcome, CredentialCasUpdate,
-    CredentialRevision, LoadedCredential, NewProviderAccount, ProviderAccount, ProviderAccountId,
-    ProviderAccountStore, ProviderAccountUpdate,
+    CredentialRevision, LoadedCredential, ProviderAccount, ProviderAccountId, ProviderAccountStore,
+    ProviderAccountUpdate,
 };
-use gateway_core::routing::{ProviderInstanceId, ProviderKind};
+use gateway_core::routing::ProviderInstanceId;
 use secrecy::ExposeSecret;
 use thiserror::Error;
 
 use super::identity::CodexSignedIdentity;
 use super::security::{CodexCredentialCodec, CodexCredentialDataError, CodexRuntimeCredential};
 use super::types::{
-    CodexAccountProfile, CodexCredentialData, CodexOAuthSecret, CreateCodexCredential,
-    CredentialRecord, RotateCodexCredential,
+    CodexAccountProfile, CodexCredentialData, CodexOAuthSecret, RotateCodexCredential,
 };
 
 const PROVIDER_NAME: &str = "openai";
@@ -38,53 +37,6 @@ impl CodexCredentialRepository {
         &self.store
     }
 
-    pub async fn create_oauth_credential(
-        &self,
-        input: CreateCodexCredential,
-    ) -> Result<CredentialRecord, CredentialRepositoryError> {
-        let account_id = ProviderAccountId::new(input.account_id)
-            .map_err(|_| CredentialRepositoryError::InvalidInput("account_id"))?;
-        let instance = ProviderInstanceId::new(input.provider_instance_id)
-            .map_err(|_| CredentialRepositoryError::InvalidInput("provider_instance_id"))?;
-        let provider = ProviderKind::new(PROVIDER_NAME)
-            .map_err(|_| CredentialRepositoryError::InvalidCredentialData)?;
-        let revision = CredentialRevision::new(1)
-            .map_err(|_| CredentialRepositoryError::InvalidCredentialData)?;
-        let access_token_expires_at = required_time(input.account.access_token_expires_at)?;
-        let upstream_user_id = input.account.chatgpt_user_id.clone();
-        let credential =
-            CodexCredentialCodec::encode_new(&input.secret, &input.account, Vec::new())?;
-        let has_refresh_token = input.secret.refresh_token.is_some();
-        let next_refresh_at = optional_time(input.next_refresh_at);
-        let account = ProviderAccount::new(
-            account_id.clone(),
-            instance.clone(),
-            provider,
-            input.name,
-            upstream_user_id,
-            revision,
-            access_token_expires_at,
-        )
-        .with_profile(
-            input.account.email,
-            Some(input.account.chatgpt_account_id),
-            input.account.plan_type,
-        )
-        .with_runtime_state(input.enabled, AccountAvailability::Ready, None)
-        .with_refresh_schedule(has_refresh_token, next_refresh_at);
-        self.store
-            .create_account(NewProviderAccount {
-                account,
-                credential,
-            })
-            .await?;
-        Ok(CredentialRecord {
-            account_id: account_id.to_string(),
-            provider_instance_id: instance.to_string(),
-            credential_revision: revision.get(),
-        })
-    }
-
     pub async fn rotate_oauth_secret(
         &self,
         input: RotateCodexCredential,

backend/crates/providers/openai/src/credential/types.rs

@@ -141,32 +141,6 @@ impl fmt::Debug for CodexCredentialData {
     }
 }
 
-/// Admin/OAuth 导入一个统一 Provider 账号。
-pub struct CreateCodexCredential {
-    pub account_id: String,
-    pub provider_instance_id: String,
-    pub name: String,
-    pub secret: CodexOAuthSecret,
-    pub account: CodexAccountProfile,
-    pub next_refresh_at: Option<DateTime<Utc>>,
-    pub enabled: bool,
-}
-
-impl fmt::Debug for CreateCodexCredential {
-    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
-        formatter
-            .debug_struct("CreateCodexCredential")
-            .field("account_id", &self.account_id)
-            .field("provider_instance_id", &self.provider_instance_id)
-            .field("name", &self.name)
-            .field("secret", &self.secret)
-            .field("account", &self.account)
-            .field("next_refresh_at", &self.next_refresh_at)
-            .field("enabled", &self.enabled)
-            .finish()
-    }
-}
-
 /// 刷新成功后的 CAS 输入。
 pub struct RotateCodexCredential {
     pub account_id: String,
@@ -192,13 +166,6 @@ impl fmt::Debug for RotateCodexCredential {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct CredentialRecord {
-    pub account_id: String,
-    pub provider_instance_id: String,
-    pub credential_revision: u64,
-}
-
 /// 运行时使用的 Cookie；值受 `secrecy` 保护。
 pub struct RuntimeCodexCookie {
     pub name: String,

backend/crates/providers/openai/src/lib.rs

@@ -40,8 +40,8 @@ pub mod credential;
 pub mod transport;
 
 pub use transport::{
-    CodexCanonicalDecoder, CodexCanonicalError, CodexRequestEncodeError, codex_request_semantics,
-    encode_generate_request, openai_billing_breakdown,
+    CodexCanonicalDecoder, CodexCanonicalError, CodexRequestEncodeError, encode_generate_request,
+    openai_billing_breakdown,
 };
 
 /// OpenAI 初始化后交给组装根的最小能力集。

backend/crates/providers/openai/src/transport/diagnostics.rs

@@ -314,13 +314,6 @@ impl CodexUpstreamDiagnostics {
             && self.identity_error_code.is_none()
             && self.trace_headers.is_empty()
     }
-
-    pub fn cf_ray(&self) -> Option<&str> {
-        self.trace_headers
-            .iter()
-            .find(|(name, _)| name.eq_ignore_ascii_case("cf-ray"))
-            .map(|(_, value)| value.as_str())
-    }
 }
 
 fn pair_value(headers: &[(String, String)], name: &str) -> Option<String> {

backend/crates/providers/openai/src/transport/endpoints.rs

@@ -16,22 +16,6 @@ pub fn endpoint_url(base_url: &str, endpoint_path: &str) -> String {
     )
 }
 
-/// 计算请求层路径。
-pub fn endpoint_request_path(base_url: &str, endpoint_path: &str) -> String {
-    let endpoint_path = endpoint_path.trim_start_matches('/');
-    let base_path = reqwest::Url::parse(base_url)
-        .ok()
-        .map(|url| url.path().trim_end_matches('/').to_string())
-        .filter(|path| !path.is_empty())
-        .unwrap_or_default();
-
-    if base_path.is_empty() {
-        format!("/{endpoint_path}")
-    } else {
-        format!("{base_path}/{endpoint_path}")
-    }
-}
-
 /// 返回 usage 相关 endpoint 的完整 URL 列表。
 pub fn usage_endpoint_urls(base_url: &str) -> Vec<String> {
     usage_endpoint_paths(base_url)

backend/crates/providers/openai/src/transport/headers.rs

@@ -1,5 +1,5 @@
 use crate::transport::profile::CodexWireProfile;
-use reqwest::header::{ACCEPT, AUTHORIZATION, HeaderMap, HeaderName, HeaderValue, USER_AGENT};
+use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderName, HeaderValue, USER_AGENT};
 
 use super::client::CodexClientResult;
 
@@ -23,28 +23,6 @@ pub fn build_codex_base_headers(
     Ok(headers)
 }
 
-/// 构造含通用请求上下文的 Codex 模型请求头。
-pub fn build_codex_headers(
-    profile: &CodexWireProfile,
-    access_token: &str,
-    account_id: Option<&str>,
-    turn_state: Option<&str>,
-    request_id: &str,
-) -> CodexClientResult<HeaderMap> {
-    let mut headers = build_codex_base_headers(profile, access_token, account_id)?;
-    headers.insert(
-        HeaderName::from_static("x-openai-internal-codex-residency"),
-        HeaderValue::from_static("us"),
-    );
-    headers.insert(
-        HeaderName::from_static("x-client-request-id"),
-        HeaderValue::from_str(request_id)?,
-    );
-    insert_optional_header(&mut headers, "x-codex-turn-state", turn_state)?;
-    headers.insert(ACCEPT, HeaderValue::from_static("text/event-stream"));
-    Ok(headers)
-}
-
 pub(super) fn insert_optional_header(
     headers: &mut HeaderMap,
     name: &'static str,

backend/crates/providers/openai/src/transport/mod.rs

@@ -32,10 +32,10 @@ pub use self::{
     diagnostics::{CodexUpstreamDiagnostics, CodexUpstreamSendPhase},
     endpoints::{
         CODEX_RESPONSES_PATH, CODEX_USAGE_API_PATH, CODEX_USAGE_PATH, WHAM_USAGE_PATH,
-        endpoint_request_path, endpoint_url, usage_endpoint_urls,
+        endpoint_url, usage_endpoint_urls,
     },
-    headers::{build_codex_base_headers, build_codex_headers},
-    request::{CodexRequestEncodeError, codex_request_semantics, encode_generate_request},
+    headers::build_codex_base_headers,
+    request::{CodexRequestEncodeError, encode_generate_request},
     response_meta::CodexResponseMetadata,
     usage::{MAX_CODEX_USAGE_BODY_BYTES, openai_billing_breakdown},
     websocket::{

backend/crates/providers/openai/src/transport/protocol/responses.rs

@@ -290,41 +290,6 @@ pub fn response_event_signals(event_type: Option<&str>, value: &Value) -> Respon
     signals
 }
 
-/// 判断已收到的 Responses SSE 内容是否包含首个完整的语义输出事件。
-pub fn response_body_has_semantic_output(body_bytes: &[u8]) -> bool {
-    response_body_signals(body_bytes).semantic_output
-}
-
-fn response_body_signals(body_bytes: &[u8]) -> ResponseEventSignals {
-    let body = String::from_utf8_lossy(body_bytes);
-    let Some(complete_body) = complete_sse_body_prefix(&body) else {
-        return ResponseEventSignals::default();
-    };
-    let Ok(events) = parse_sse_events(complete_body) else {
-        return ResponseEventSignals::default();
-    };
-    events
-        .iter()
-        .fold(ResponseEventSignals::default(), |mut signals, event| {
-            let Ok(value) = serde_json::from_str::<Value>(&event.data) else {
-                return signals;
-            };
-            let event_type = event.event.as_deref().or_else(|| {
-                value
-                    .get("type")
-                    .and_then(Value::as_str)
-                    .filter(|value| !value.trim().is_empty())
-            });
-            merge_event_signals(&mut signals, response_event_signals(event_type, &value));
-            signals
-        })
-}
-
-fn merge_event_signals(target: &mut ResponseEventSignals, source: ResponseEventSignals) {
-    target.protocol_progress |= source.protocol_progress;
-    merge_output_signals(target, source);
-}
-
 fn merge_output_signals(target: &mut ResponseEventSignals, source: ResponseEventSignals) {
     target.semantic_output |= source.semantic_output;
     target.reasoning_output |= source.reasoning_output;
@@ -400,16 +365,6 @@ fn is_tool_execution_event(event_type: &str) -> bool {
         )
 }
 
-fn complete_sse_body_prefix(body: &str) -> Option<&str> {
-    let lf_end = body.rfind("\n\n").map(|index| index + 2);
-    let crlf_end = body.rfind("\r\n\r\n").map(|index| index + 4);
-    lf_end
-        .into_iter()
-        .chain(crlf_end)
-        .max()
-        .map(|end| &body[..end])
-}
-
 /// 从 Codex SSE 收集出的非流式 Responses 结果。
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum CollectedResponse {
@@ -704,22 +659,6 @@ impl CodexResponsesRequest {
         }
     }
 
-    /// 构造默认的 HTTP SSE 请求（测试与内部构造用）。
-    pub fn new_http_sse(
-        model: impl Into<String>,
-        instructions: impl Into<String>,
-        input: Vec<Value>,
-    ) -> Self {
-        let mut body = Map::new();
-        body.insert("model".to_string(), Value::String(model.into()));
-        body.insert(
-            "instructions".to_string(),
-            Value::String(instructions.into()),
-        );
-        body.insert("input".to_string(), Value::Array(input));
-        Self::from_body(body)
-    }
-
     /// 上游 body 的只读视图。
     pub fn body(&self) -> &Map<String, Value> {
         &self.body
@@ -796,11 +735,6 @@ impl CodexResponsesRequest {
             .unwrap_or(true)
     }
 
-    /// 设置上游存储标志。
-    pub fn set_store(&mut self, store: bool) {
-        self.body.insert("store".to_string(), Value::Bool(store));
-    }
-
     /// reasoning 配置（透传，不规整）。
     pub fn reasoning(&self) -> Option<&Value> {
         self.body.get("reasoning")
@@ -851,19 +785,6 @@ impl CodexResponsesRequest {
         self.body.get("prompt_cache_key").and_then(Value::as_str)
     }
 
-    /// 设置提示缓存键。
-    pub fn set_prompt_cache_key(&mut self, prompt_cache_key: Option<String>) {
-        match prompt_cache_key {
-            Some(value) => {
-                self.body
-                    .insert("prompt_cache_key".to_string(), Value::String(value));
-            }
-            None => {
-                self.body.remove("prompt_cache_key");
-            }
-        }
-    }
-
     /// client metadata（透传原值）。
     pub fn client_metadata(&self) -> Option<&Value> {
         self.body.get("client_metadata")

backend/crates/providers/openai/src/transport/protocol/websocket.rs

@@ -181,14 +181,6 @@ pub fn websocket_response_completed_id(raw: &str) -> Result<Option<String>, Stri
     Ok(Some(completed.id))
 }
 
-/// 判断 WebSocket 事件是否会结束当前响应流。
-pub fn is_terminal_websocket_event(event: &str) -> bool {
-    matches!(
-        event,
-        "response.completed" | "response.incomplete" | "response.failed" | "error"
-    )
-}
-
 /// 生成 Responses WebSocket payload 审计快照。
 pub fn websocket_payload_audit_snapshot(request: &CodexResponsesRequest) -> PayloadAuditSnapshot {
     let body = websocket_response_create_payload(request);

backend/crates/providers/openai/src/transport/request.rs

@@ -8,7 +8,7 @@ use gateway_core::operation::{
 use serde_json::{Map, Value, json};
 use sha2::{Digest, Sha256};
 
-use crate::transport::protocol::responses::{CodexRequestSemantics, CodexResponsesRequest};
+use crate::transport::protocol::responses::CodexResponsesRequest;
 
 const CODEX_OPTION_FIELDS: &[&str] = &[
     "beta_features",
@@ -191,14 +191,6 @@ pub fn encode_generate_request(
     Ok(encoded)
 }
 
-/// 复用正式 Codex 编码规则提取请求诊断语义；编码失败时不伪造 Provider 事实。
-#[must_use]
-pub fn codex_request_semantics(request: &GenerateRequest) -> CodexRequestSemantics {
-    encode_generate_request(request, "observability")
-        .map(|request| request.semantics())
-        .unwrap_or_default()
-}
-
 fn encode_messages(request: &GenerateRequest) -> Result<Vec<Value>, CodexRequestEncodeError> {
     let mut input = Vec::new();
     for message in request.messages() {

backend/crates/providers/openai/src/transport/websocket/breaker.rs

@@ -181,10 +181,6 @@ impl WebSocketOriginBreakerPermit {
         }
     }
 
-    pub fn is_half_open_probe(&self) -> bool {
-        self.probe_id.is_some()
-    }
-
     /// 创建可跨前台预算与后台 opening 生命周期共享的快路径观察器。
     pub(crate) fn fast_path_reporter(&self) -> WebSocketOriginFastPathReporter {
         WebSocketOriginFastPathReporter {

backend/crates/providers/openai/tests/admin.rs

@@ -48,7 +48,7 @@ use gateway_core::routing::{
 };
 use gateway_core::task::{WorkerContribution, WorkerKind, WorkerRunnable};
 use provider_openai::config::{CodexWireProfileConfig, OpenAiConfig};
-use provider_openai::credential::CreateCodexCredential;
+use provider_openai::credential::ImportCodexOAuthCredential;
 use provider_openai::transport::profile::APPCAST_POLL_INTERVAL;
 use serde_json::{Map, Value, json};
 use std::collections::BTreeSet;
@@ -399,18 +399,16 @@ async fn openai_admin_provider_persists_the_full_pending_envelope_and_binds_owne
 async fn openai_admin_provider_projects_cached_quota_models_and_canonical_export() {
     let store = Arc::new(MemoryAccountStore::default());
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_admin_projection".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "admin projection".to_owned(),
             secret: secret("admin-projection-access"),
-            account: profile("chatgpt-admin-projection"),
+            verified_account: profile("chatgpt-admin-projection"),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
     let account = store
         .account("acct_admin_projection")
         .expect("stored account");
@@ -487,18 +485,16 @@ async fn openai_admin_provider_projects_cached_quota_models_and_canonical_export
 async fn openai_admin_provider_rejects_unprepared_mutations_before_store_commit() {
     let store = Arc::new(MemoryAccountStore::default());
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_admin_invalid".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "admin invalid".to_owned(),
             secret: secret("admin-invalid-access"),
-            account: profile("chatgpt-admin-invalid"),
+            verified_account: profile("chatgpt-admin-invalid"),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
     let account = store.account("acct_admin_invalid").expect("stored account");
     let record = account_record(&account);
     let bundle = provider_openai::initialize(

backend/crates/providers/openai/tests/credential/admin.rs

@@ -17,9 +17,9 @@ use provider_openai::credential::{
     CodexAccountIdentityService, CodexAccountIdentityVerifier, CodexCredentialAdmin,
     CodexCredentialAdminError, CodexCredentialAdminService, CodexCredentialCodec,
     CodexIdentityExpectation, CodexIdentityVerification, CodexIdentityVerificationError,
-    CodexJwtIdentityVerifier, CreateCodexCredential, ExportManagedCodexCredential,
-    ImportCodexOAuthCredential, ImportCodexOAuthCredentialBatch,
-    ReqwestCodexAuthenticatedAccountSource, ReqwestOpenAiJwksSource, RotateManagedCodexCredential,
+    CodexJwtIdentityVerifier, ExportManagedCodexCredential, ImportCodexOAuthCredential,
+    ImportCodexOAuthCredentialBatch, ReqwestCodexAuthenticatedAccountSource,
+    ReqwestOpenAiJwksSource, RotateManagedCodexCredential,
 };
 use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState};
 use secrecy::{ExposeSecret, SecretString};
@@ -345,18 +345,16 @@ async fn manual_refresh_fixture(
 ) {
     let store = Arc::new(MemoryAccountStore::default());
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_manual_refresh".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "manual refresh".to_owned(),
             secret: secret("old-access"),
-            account: profile("chatgpt-acct_manual_refresh"),
+            verified_account: profile("chatgpt-acct_manual_refresh"),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
     let refresher = Arc::new(ManualRefresher {
         outcome: Mutex::new(Some(outcome)),
         seen: Mutex::new(Vec::new()),
@@ -397,7 +395,6 @@ async fn manual_refresh_prepares_revision_fenced_rotation_without_store_mutation
         .await
         .expect("manual refresh");
 
-    assert!(prepared.holds_refresh_lease());
     assert_eq!(prepared.credential.account_id(), &account_id);
     assert_eq!(prepared.credential.expected_revision().get(), 1);
     let runtime = CodexCredentialCodec::decode(prepared.credential.credential())

backend/crates/providers/openai/tests/credential/catalog.rs

@@ -9,7 +9,7 @@ use gateway_core::engine::credential::{
 };
 use gateway_core::routing::{InstanceHealth, ProviderInstance, ProviderKind};
 use provider_openai::credential::{
-    CodexCredentialCatalogError, CodexCredentialCatalogService, CreateCodexCredential,
+    CodexCredentialCatalogError, CodexCredentialCatalogService, ImportCodexOAuthCredential,
 };
 use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState};
 use wiremock::matchers::{header, method, path};
@@ -46,18 +46,16 @@ fn instance(base_url: &str) -> ProviderInstance {
 
 async fn create(store: &Arc<MemoryAccountStore>, id: &str, token: &str) {
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: id.to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: id.to_owned(),
             secret: secret(token),
-            account: profile(&format!("chatgpt-{id}")),
+            verified_account: profile(&format!("chatgpt-{id}")),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
 }
 
 fn service(store: &Arc<MemoryAccountStore>) -> CodexCredentialCatalogService {

backend/crates/providers/openai/tests/credential/contract.rs

@@ -14,8 +14,8 @@ use gateway_core::engine::{
 use gateway_core::policy::ClientApiKeyId;
 use provider_openai::credential::{
     CodexAccountFailure, CodexCookiePolicy, CodexCredentialCatalogService, CodexCredentialCodec,
-    CodexCredentialQuotaService, CodexCredentialSelector, CreateCodexCredential,
-    CredentialSelectionError, SelectCodexCredential,
+    CodexCredentialQuotaService, CodexCredentialSelector, CredentialSelectionError,
+    ImportCodexOAuthCredential, SelectCodexCredential,
 };
 use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState};
 use provider_openai::{CodexOriginPolicy, OfficialCodexOriginPolicy};
@@ -27,20 +27,15 @@ use crate::support::{
 };
 
 fn create_account(store: &Arc<MemoryAccountStore>, id: &str, token: &str) {
-    block_on(
-        store
-            .repository()
-            .create_oauth_credential(CreateCodexCredential {
-                account_id: id.to_owned(),
-                provider_instance_id: instance_id().to_string(),
-                name: id.to_owned(),
-                secret: secret(token),
-                account: profile(&format!("chatgpt-{id}")),
-                next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
-                enabled: true,
-            }),
-    )
-    .expect("create account");
+    block_on(store.seed_oauth_credential(ImportCodexOAuthCredential {
+        account_id: id.to_owned(),
+        provider_instance_id: instance_id().to_string(),
+        name: id.to_owned(),
+        secret: secret(token),
+        verified_account: profile(&format!("chatgpt-{id}")),
+        next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
+        enabled: true,
+    }));
 }
 
 fn attempt(excluded_accounts: BTreeSet<ProviderAccountId>) -> AttemptContext {

backend/crates/providers/openai/tests/credential/oauth.rs

@@ -17,7 +17,7 @@ use provider_openai::credential::{
     CodexIdentityVerification, CodexIdentityVerificationError, CodexOAuthAdmin,
     CodexOAuthAdminError, CodexOAuthAdminService, CodexOAuthPendingStore,
     CodexOAuthPendingStoreError, CodexOAuthSecret, CodexPendingAuthorization,
-    CompleteCodexOAuthAuthorization, CompletedCodexOAuthCredential, CreateCodexCredential,
+    CompleteCodexOAuthAuthorization, CompletedCodexOAuthCredential, ImportCodexOAuthCredential,
     StartCodexOAuthAuthorization, StoredCodexPendingAuthorization,
 };
 use secrecy::{ExposeSecret, SecretString};
@@ -315,18 +315,16 @@ fn oauth_command_debug_redacts_owner_flow_and_callback() {
 async fn reauthorization_binds_account_revision_and_returns_only_prepared_rotation() {
     let store = Arc::new(MemoryAccountStore::default());
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_oauth_reauth".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "reauthorize".to_owned(),
             secret: secret("old-oauth-access"),
-            account: profile("chatgpt-oauth"),
+            verified_account: profile("chatgpt-oauth"),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
     let original_account = store
         .account("acct_oauth_reauth")
         .expect("original account");
@@ -414,18 +412,16 @@ impl CodexAccountIdentityVerifier for RebindingVerifier {
 async fn reauthorization_rejects_identity_rebinding() {
     let store = Arc::new(MemoryAccountStore::default());
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_oauth_rebind".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "reauthorize".to_owned(),
             secret: secret("old-oauth-access"),
-            account: profile("chatgpt-oauth"),
+            verified_account: profile("chatgpt-oauth"),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
     let service = service_with_store(store, Arc::new(RebindingVerifier));
     let started = service
         .start_authorization(StartCodexOAuthAuthorization {

backend/crates/providers/openai/tests/credential/quota.rs

@@ -9,7 +9,7 @@ use gateway_core::engine::credential::{
 use gateway_core::routing::{InstanceHealth, ProviderInstance, ProviderKind};
 use provider_openai::credential::{
     CodexCredentialQuotaService, CodexQuotaSyncSummary, CodexQuotaWindowKind,
-    CreateCodexCredential, parse_codex_quota_usage,
+    ImportCodexOAuthCredential, parse_codex_quota_usage,
 };
 use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState};
 use serde_json::json;
@@ -149,18 +149,16 @@ fn quota_service(store: &Arc<MemoryAccountStore>) -> CodexCredentialQuotaService
 async fn concurrent_cold_scheduling_hydration_reads_quota_once() {
     let store = Arc::new(MemoryAccountStore::default());
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_hydration".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "hydration".to_owned(),
             secret: secret("hydration-token"),
-            account: profile("chatgpt-acct_hydration"),
+            verified_account: profile("chatgpt-acct_hydration"),
             next_refresh_at: Some(Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
     let account = store.account("acct_hydration").expect("created account");
     let service = quota_service(&store);
 
@@ -206,18 +204,16 @@ async fn quota_service_stores_raw_provider_json_and_projects_common_state() {
         .await;
     let store = Arc::new(MemoryAccountStore::default());
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_quota".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "quota".to_owned(),
             secret: secret("quota-token"),
-            account: profile("chatgpt-acct_quota"),
+            verified_account: profile("chatgpt-acct_quota"),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
     let service = quota_service(&store);
     let instance = ProviderInstance::new(
         instance_id(),
@@ -290,18 +286,16 @@ async fn quota_refresh_success_must_not_clear_newer_future_cooldown() {
         .await;
     let store = Arc::new(MemoryAccountStore::default());
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_quota_cooldown".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "quota cooldown".to_owned(),
             secret: secret("quota-cooldown-token"),
-            account: profile("chatgpt-acct_quota_cooldown"),
+            verified_account: profile("chatgpt-acct_quota_cooldown"),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
     let account = store.account("acct_quota_cooldown").expect("quota account");
     let service = quota_service(&store);
     let instance = ProviderInstance::new(
@@ -362,18 +356,16 @@ async fn non_exhausted_quota_refresh_should_restore_quota_exhausted_account() {
 async fn passive_rate_limit_headers_update_quota_and_account_state_with_revision_fence() {
     let store = Arc::new(MemoryAccountStore::default());
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_passive_quota".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "passive quota".to_owned(),
             secret: secret("passive-quota-token"),
-            account: profile("chatgpt-acct_passive_quota"),
+            verified_account: profile("chatgpt-acct_passive_quota"),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
     let account = store.account("acct_passive_quota").expect("account");
     let service = quota_service(&store);
     let reset_at = 1_900_000_000_i64;
@@ -500,18 +492,16 @@ async fn synchronize_account_from_state(
         .await;
     let store = Arc::new(MemoryAccountStore::default());
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: account_id.to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: account_id.to_owned(),
             secret: secret(&format!("token-{account_id}")),
-            account: profile(&format!("chatgpt-{account_id}")),
+            verified_account: profile(&format!("chatgpt-{account_id}")),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create quota account");
+        .await;
     let account = store.account(account_id).expect("quota account");
     store
         .apply_state_change(AccountStateChange {

backend/crates/providers/openai/tests/credential/refresh.rs

@@ -13,7 +13,7 @@ use provider_openai::credential::token_client::{RefreshFailure, TokenPair, Token
 use provider_openai::credential::{
     CodexAccountIdentityVerifier, CodexCredentialRefreshOutcome, CodexCredentialRefreshService,
     CodexIdentityExpectation, CodexIdentityVerification, CodexIdentityVerificationError,
-    CodexOAuthSecret, CodexSignedIdentity, CreateCodexCredential, RotateCodexCredential,
+    CodexOAuthSecret, CodexSignedIdentity, ImportCodexOAuthCredential, RotateCodexCredential,
 };
 use secrecy::{ExposeSecret, SecretString};
 
@@ -225,18 +225,16 @@ async fn setup(
 ) {
     let store = Arc::new(MemoryAccountStore::default());
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_refresh".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "refresh".to_owned(),
             secret: secret("old-access"),
-            account: profile("chatgpt-acct_refresh"),
+            verified_account: profile("chatgpt-acct_refresh"),
             next_refresh_at: Some(chrono::Utc::now() - chrono::Duration::seconds(1)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
     let refresher = Arc::new(Refresher::new(outcome));
     let service = CodexCredentialRefreshService::new(
         store.repository(),
@@ -361,18 +359,16 @@ async fn unavailable_usage_preserves_rotated_tokens_and_persists_backoff() {
     account_profile.poid = signed.poid().map(str::to_owned);
     account_profile.chatgpt_user_id = "user-signed".to_owned();
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_signed_only".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "signed only".to_owned(),
             secret: secret("old-access"),
-            account: account_profile,
+            verified_account: account_profile,
             next_refresh_at: Some(chrono::Utc::now() - chrono::Duration::seconds(1)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
     let service = CodexCredentialRefreshService::new(
         store.repository(),
         Arc::new(Refresher::new(Ok(success_tokens()))),
@@ -537,18 +533,16 @@ async fn malformed_account_refresh_does_not_stop_later_accounts() {
     let store = Arc::new(MemoryAccountStore::default());
     for account_id in ["acct_bad", "acct_good"] {
         store
-            .repository()
-            .create_oauth_credential(CreateCodexCredential {
+            .seed_oauth_credential(ImportCodexOAuthCredential {
                 account_id: account_id.to_owned(),
                 provider_instance_id: instance_id().to_string(),
                 name: account_id.to_owned(),
                 secret: secret(account_id),
-                account: profile(&format!("chatgpt-{account_id}")),
+                verified_account: profile(&format!("chatgpt-{account_id}")),
                 next_refresh_at: Some(chrono::Utc::now() - chrono::Duration::seconds(1)),
                 enabled: true,
             })
-            .await
-            .expect("create account");
+            .await;
     }
     let refresher = Arc::new(Refresher::scripted([
         Ok(TokenPair {

backend/crates/providers/openai/tests/provider/contract.rs

@@ -29,7 +29,7 @@ use gateway_core::routing::{
 use provider_openai::CodexProvider;
 use provider_openai::credential::{
     CodexCookie, CodexCookiePolicy, CodexCredentialCatalogService, CodexCredentialQuotaService,
-    CodexCredentialSelector, CreateCodexCredential, RotateCodexCredential,
+    CodexCredentialSelector, ImportCodexOAuthCredential, RotateCodexCredential,
 };
 use provider_openai::transport::CodexWebSocketPool;
 use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState};
@@ -61,34 +61,30 @@ fn wire_profile() -> CodexWireProfileState {
 
 async fn create_account(store: &Arc<MemoryAccountStore>) {
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_provider".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "provider".to_owned(),
             secret: secret("provider-access"),
-            account: profile("chatgpt-acct_provider"),
+            verified_account: profile("chatgpt-acct_provider"),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create account");
+        .await;
 }
 
 async fn create_other_account(store: &Arc<MemoryAccountStore>) {
     store
-        .repository()
-        .create_oauth_credential(CreateCodexCredential {
+        .seed_oauth_credential(ImportCodexOAuthCredential {
             account_id: "acct_provider_other".to_owned(),
             provider_instance_id: instance_id().to_string(),
             name: "provider-other".to_owned(),
             secret: secret("provider-other-access"),
-            account: profile("chatgpt-acct_provider_other"),
+            verified_account: profile("chatgpt-acct_provider_other"),
             next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(30)),
             enabled: true,
         })
-        .await
-        .expect("create other account");
+        .await;
 }
 
 async fn attach_cookie(store: &Arc<MemoryAccountStore>, account_id: &str, value: &str) {

backend/crates/providers/openai/tests/support.rs

@@ -23,7 +23,8 @@ use gateway_core::provider_ports::{
 use gateway_core::routing::{ProviderInstanceId, ProviderKind};
 use provider_openai::CodexOriginPolicy;
 use provider_openai::credential::{
-    CodexAccountProfile, CodexCredentialRepository, CodexOAuthSecret,
+    CodexAccountProfile, CodexCredentialAdmin, CodexCredentialRepository, CodexOAuthSecret,
+    ImportCodexOAuthCredential,
 };
 use secrecy::SecretString;
 use url::{Host, Url};
@@ -46,6 +47,15 @@ impl MemoryAccountStore {
         CodexCredentialRepository::new(self.clone())
     }
 
+    pub(crate) async fn seed_oauth_credential(&self, input: ImportCodexOAuthCredential) {
+        let account = CodexCredentialAdmin
+            .prepare_import(input)
+            .expect("prepare test OAuth credential");
+        self.create_account(account)
+            .await
+            .expect("seed test OAuth credential");
+    }
+
     pub(crate) fn account(&self, id: &str) -> Option<ProviderAccount> {
         let id = ProviderAccountId::new(id).ok()?;
         self.accounts

backend/crates/providers/openai/tests/transport/diagnostics.rs

@@ -2,6 +2,14 @@ use base64::{Engine as _, engine::general_purpose::STANDARD};
 use provider_openai::transport::CodexUpstreamDiagnostics;
 use reqwest::header::{HeaderMap, HeaderValue};
 
+fn trace_header<'a>(diagnostics: &'a CodexUpstreamDiagnostics, name: &str) -> Option<&'a str> {
+    diagnostics
+        .trace_headers
+        .iter()
+        .find(|(candidate, _)| candidate.eq_ignore_ascii_case(name))
+        .map(|(_, value)| value.as_str())
+}
+
 #[test]
 fn diagnostics_should_extract_request_id_and_trace_headers() {
     let mut headers = HeaderMap::new();
@@ -14,7 +22,7 @@ fn diagnostics_should_extract_request_id_and_trace_headers() {
         (
             diagnostics.status_code,
             diagnostics.request_id.as_deref(),
-            diagnostics.cf_ray(),
+            trace_header(&diagnostics, "cf-ray"),
         ),
         (Some(429), Some("req_1"), Some("ray_1"))
     );
@@ -28,7 +36,10 @@ fn diagnostics_should_not_treat_cf_ray_as_a_request_id() {
     let diagnostics = CodexUpstreamDiagnostics::from_headers(Some(403), &headers);
 
     assert_eq!(
-        (diagnostics.request_id.as_deref(), diagnostics.cf_ray()),
+        (
+            diagnostics.request_id.as_deref(),
+            trace_header(&diagnostics, "cf-ray"),
+        ),
         (None, Some("ray_1"))
     );
 }

backend/crates/providers/openai/tests/transport/endpoints.rs

@@ -1,16 +1,10 @@
-use provider_openai::transport::{endpoint_request_path, endpoint_url, usage_endpoint_urls};
+use provider_openai::transport::{endpoint_url, usage_endpoint_urls};
 
 #[test]
 fn endpoints_should_join_origin_and_backend_paths_without_double_slashes() {
     assert_eq!(
-        (
-            endpoint_url("https://api.example.com/", "/codex/responses"),
-            endpoint_request_path("https://api.example.com/backend-api", "/codex/usage"),
-        ),
-        (
-            "https://api.example.com/codex/responses".to_owned(),
-            "/backend-api/codex/usage".to_owned(),
-        )
+        endpoint_url("https://api.example.com/", "/codex/responses"),
+        "https://api.example.com/codex/responses"
     );
 }
 

backend/crates/providers/openai/tests/transport/headers.rs

@@ -1,64 +1,8 @@
-use chrono::{TimeZone, Utc};
-use provider_openai::transport::build_codex_headers;
-use provider_openai::transport::profile::CodexWireProfile;
 use provider_openai::transport::websocket::CodexWebSocketConnection;
 use serde_json::Value;
 
 use super::*;
 
-#[test]
-fn codex_http_headers_should_use_the_wire_profile_without_browser_metadata() {
-    let profile = CodexWireProfile {
-        originator: "Codex Desktop".to_owned(),
-        codex_version: "0.144.2".to_owned(),
-        desktop_version: "26.707.72221".to_owned(),
-        desktop_build: "72221".to_owned(),
-        os_type: "Mac OS".to_owned(),
-        os_version: "15.7.1".to_owned(),
-        arch: "arm64".to_owned(),
-        terminal: "unknown".to_owned(),
-        verified_at: Utc
-            .with_ymd_and_hms(2026, 7, 18, 0, 0, 0)
-            .single()
-            .expect("valid fixture time"),
-    };
-
-    let headers = build_codex_headers(
-        &profile,
-        "access-token",
-        Some("acct-1"),
-        Some("turn-1"),
-        "req-1",
-    )
-    .expect("valid profile headers");
-    let value = |name: &str| headers.get(name).and_then(|value| value.to_str().ok());
-
-    assert_eq!(value("authorization"), Some("Bearer access-token"));
-    assert_eq!(value("chatgpt-account-id"), Some("acct-1"));
-    assert_eq!(value("originator"), Some("Codex Desktop"));
-    assert_eq!(
-        value("user-agent"),
-        Some("Codex Desktop/0.144.2 (Mac OS 15.7.1; arm64) unknown (Codex Desktop; 26.707.72221)")
-    );
-    assert_eq!(value("x-client-request-id"), Some("req-1"));
-    assert_eq!(value("x-codex-turn-state"), Some("turn-1"));
-    assert_eq!(value("accept"), Some("text/event-stream"));
-    for browser_header in [
-        "sec-ch-ua",
-        "sec-ch-ua-mobile",
-        "sec-ch-ua-platform",
-        "sec-fetch-site",
-        "sec-fetch-mode",
-        "sec-fetch-dest",
-        "accept-language",
-    ] {
-        assert!(
-            headers.get(browser_header).is_none(),
-            "sent {browser_header}"
-        );
-    }
-}
-
 #[test]
 fn websocket_connection_should_preserve_endpoint_and_header_order() {
     let connection = CodexWebSocketConnection::new(
@@ -200,11 +144,11 @@ async fn backend_websocket_should_forward_context_headers_and_preserve_payload_f
             .expect("send terminal event");
         payload
     });
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-test", "be brief", Vec::new());
+    let mut request =
+        codex_request_with_prompt_cache_key("gpt-test", "be brief", Vec::new(), "client-thread");
     request.use_websocket = true;
     request.responses_lite = Some("true".to_owned());
     request.memgen_request = Some("true".to_owned());
-    request.set_prompt_cache_key(Some("client-thread".to_owned()));
     request.set_client_metadata(Some(json!({
         "safe": "yes",
         "x-openai-subagent": "review",
@@ -306,7 +250,7 @@ async fn backend_http_should_send_codex_context_without_browser_headers() {
         write_completed_sse_response(&mut stream).await;
         request
     });
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-test", "", Vec::new());
+    let mut request = codex_request("gpt-test", "", Vec::new());
     request.force_http_sse = true;
     request.turn_metadata = Some("turn-meta".to_owned());
     request.beta_features = Some("beta-a".to_owned());
@@ -314,13 +258,15 @@ async fn backend_http_should_send_codex_context_without_browser_headers() {
     request.version = Some("26.707.51957".to_owned());
     request.codex_window_id = Some("cw_1".to_owned());
     request.parent_thread_id = Some("parent-1".to_owned());
+    let profile = test_wire_profile();
+    let expected_user_agent = profile.snapshot().user_agent();
     let client = CodexBackendClient::new(
         reqwest::Client::builder()
             .no_proxy()
             .build()
             .expect("HTTP client"),
         format!("http://{address}"),
-        test_wire_profile(),
+        profile,
     );
 
     client
@@ -350,6 +296,28 @@ async fn backend_http_should_send_codex_context_without_browser_headers() {
 
     let raw_request = server.await.expect("HTTP header server task");
     let header_names = read_header_names(&raw_request);
+    assert_eq!(
+        (
+            read_header_value(&raw_request, "authorization"),
+            read_header_value(&raw_request, "chatgpt-account-id"),
+            read_header_value(&raw_request, "originator"),
+            read_header_value(&raw_request, "user-agent"),
+            read_header_value(&raw_request, "accept"),
+            read_header_value(&raw_request, "x-openai-internal-codex-residency"),
+            read_header_value(&raw_request, "x-client-request-id"),
+            read_header_value(&raw_request, "x-codex-turn-state"),
+        ),
+        (
+            Some("Bearer access-token"),
+            Some("chatgpt-account"),
+            Some("codex_cli_rs"),
+            Some(expected_user_agent.as_str()),
+            Some("text/event-stream"),
+            Some("us"),
+            Some("req_order"),
+            Some("turn-state"),
+        )
+    );
     for required in [
         "authorization",
         "chatgpt-account-id",
@@ -469,7 +437,7 @@ async fn websocket_should_keep_an_exact_chain_while_new_connections_adopt_the_la
         profile.clone(),
     )
     .with_websocket_pool(pool);
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-test", "", Vec::new());
+    let mut request = codex_request("gpt-test", "", Vec::new());
     request.use_websocket = true;
     request.local_conversation_id = Some("profile-rotation".to_owned());
 

backend/crates/providers/openai/tests/transport/http_client.rs

@@ -24,13 +24,6 @@ fn endpoints_should_join_backend_paths() {
         provider_openai::transport::endpoint_url("https://api.example.com/", "/codex/responses"),
         "https://api.example.com/codex/responses"
     );
-    assert_eq!(
-        provider_openai::transport::endpoint_request_path(
-            "https://api.example.com/backend-api",
-            "/codex/usage"
-        ),
-        "/backend-api/codex/usage"
-    );
 }
 
 #[test]
@@ -124,7 +117,7 @@ async fn codex_backend_client_should_bound_and_redact_oversized_error_body() {
         server.uri(),
         test_wire_profile(),
     );
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "", Vec::new());
+    let mut request = codex_request("gpt-5.5", "", Vec::new());
     request.force_http_sse = true;
 
     let result = client
@@ -176,7 +169,7 @@ async fn codex_backend_client_should_parse_retry_after_from_rate_limit_error_bod
         server.uri(),
         test_wire_profile(),
     );
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "", Vec::new());
+    let mut request = codex_request("gpt-5.5", "", Vec::new());
     request.force_http_sse = true;
 
     let result = client
@@ -235,7 +228,7 @@ async fn codex_backend_http_sse_should_fail_after_five_minutes_without_stream_da
         base_url,
         test_wire_profile(),
     );
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "", Vec::new());
+    let mut request = codex_request("gpt-5.5", "", Vec::new());
     request.force_http_sse = true;
 
     let mut response = client
@@ -281,7 +274,7 @@ async fn codex_backend_client_should_capture_only_safe_response_metadata() {
         server.uri(),
         test_wire_profile(),
     );
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "", Vec::new());
+    let mut request = codex_request("gpt-5.5", "", Vec::new());
     request.force_http_sse = true;
 
     let response = client

backend/crates/providers/openai/tests/transport/latency.rs

@@ -9,7 +9,7 @@ use provider_openai::transport::{
 use super::*;
 
 fn new_chain_request(conversation_id: &str) -> CodexResponsesRequest {
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request("gpt-5.5", "be brief", Vec::new());
     request.use_websocket = true;
     request.local_conversation_id = Some(conversation_id.to_string());
     request
@@ -955,7 +955,6 @@ async fn origin_breaker_should_open_then_allow_only_one_half_open_probe() {
     else {
         panic!("expired open state should allow one probe");
     };
-    assert!(probe.is_half_open_probe());
     assert!(matches!(
         breaker.try_acquire("https://example.test:443"),
         WebSocketOriginBreakerDecision::HalfOpenBusy
@@ -1026,7 +1025,10 @@ async fn cancelled_half_open_opening_should_allow_another_probe() {
     else {
         panic!("cancelled half-open probe should release its ownership");
     };
-    assert!(next_probe.is_half_open_probe());
+    assert!(matches!(
+        breaker.try_acquire(&origin_key),
+        WebSocketOriginBreakerDecision::HalfOpenBusy
+    ));
     next_probe.succeed();
     pool.shutdown().await;
 }

backend/crates/providers/openai/tests/transport/mod.rs

@@ -21,7 +21,7 @@ use provider_openai::transport::{
     CodexBackendClient, CodexBackendTransport, CodexClientError, CodexRequestContext,
     CodexTransportDecision, CodexWebSocketPool, CodexWebSocketPoolConfig,
 };
-use serde_json::json;
+use serde_json::{Map, Value, json};
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::{TcpListener, TcpStream},
@@ -146,8 +146,45 @@ fn request_context<'a>(
     }
 }
 
+fn codex_request(
+    model: impl Into<String>,
+    instructions: impl Into<String>,
+    input: Vec<Value>,
+) -> CodexResponsesRequest {
+    CodexResponsesRequest::from_body(codex_request_body(model, instructions, input))
+}
+
+fn codex_request_with_prompt_cache_key(
+    model: impl Into<String>,
+    instructions: impl Into<String>,
+    input: Vec<Value>,
+    prompt_cache_key: impl Into<String>,
+) -> CodexResponsesRequest {
+    let mut body = codex_request_body(model, instructions, input);
+    body.insert(
+        "prompt_cache_key".to_owned(),
+        Value::String(prompt_cache_key.into()),
+    );
+    CodexResponsesRequest::from_body(body)
+}
+
+fn codex_request_body(
+    model: impl Into<String>,
+    instructions: impl Into<String>,
+    input: Vec<Value>,
+) -> Map<String, Value> {
+    Map::from_iter([
+        ("model".to_owned(), Value::String(model.into())),
+        (
+            "instructions".to_owned(),
+            Value::String(instructions.into()),
+        ),
+        ("input".to_owned(), Value::Array(input)),
+    ])
+}
+
 fn prepared_websocket_request(base_url: &str) -> CodexWebSocketRequest {
-    let request = CodexResponsesRequest::new_http_sse("gpt-test", "be brief", Vec::new());
+    let request = codex_request("gpt-test", "be brief", Vec::new());
     CodexWebSocketConnection::responses_create_request(
         base_url,
         "dGhlIHNhbXBsZSBub25jZQ==",
@@ -158,10 +195,10 @@ fn prepared_websocket_request(base_url: &str) -> CodexWebSocketRequest {
 }
 
 fn pooled_websocket_request(conversation_id: &str) -> CodexResponsesRequest {
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-test", "be brief", Vec::new());
+    let mut request =
+        codex_request_with_prompt_cache_key("gpt-test", "be brief", Vec::new(), conversation_id);
     request.set_previous_response_id(Some("resp_previous".to_owned()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
-    request.set_prompt_cache_key(Some(conversation_id.to_owned()));
     request.local_conversation_id = Some(conversation_id.to_owned());
     request
 }
@@ -201,6 +238,16 @@ fn read_header_names(request: &str) -> Vec<String> {
         .collect()
 }
 
+fn read_header_value<'a>(request: &'a str, name: &str) -> Option<&'a str> {
+    request
+        .lines()
+        .skip(1)
+        .take_while(|line| !line.is_empty())
+        .filter_map(|line| line.split_once(':'))
+        .find(|(candidate, _)| candidate.eq_ignore_ascii_case(name))
+        .map(|(_, value)| value.trim())
+}
+
 async fn read_http_request(stream: &mut TcpStream) -> String {
     let mut request = Vec::new();
     let mut buffer = [0_u8; 1024];

backend/crates/providers/openai/tests/transport/protocol/responses.rs

@@ -1,8 +1,35 @@
+use gateway_protocol::openai::sse::parse_sse_events;
 use provider_openai::transport::protocol::responses::{
-    CodexResponsesRequest, PreviousResponseScope, TransportRequirement,
-    response_body_has_semantic_output, response_event_signals, transport_requirement,
+    CodexResponsesRequest, PreviousResponseScope, TransportRequirement, response_event_signals,
+    transport_requirement,
 };
-use serde_json::json;
+use serde_json::{Value, json};
+
+use super::super::codex_request;
+
+fn response_body_has_semantic_output(body_bytes: &[u8]) -> bool {
+    let body = String::from_utf8_lossy(body_bytes);
+    let lf_end = body.rfind("\n\n").map(|index| index + 2);
+    let crlf_end = body.rfind("\r\n\r\n").map(|index| index + 4);
+    let Some(end) = lf_end.into_iter().chain(crlf_end).max() else {
+        return false;
+    };
+    let Ok(events) = parse_sse_events(&body[..end]) else {
+        return false;
+    };
+    events.iter().any(|event| {
+        let Ok(value) = serde_json::from_str::<Value>(&event.data) else {
+            return false;
+        };
+        let event_type = event.event.as_deref().or_else(|| {
+            value
+                .get("type")
+                .and_then(Value::as_str)
+                .filter(|value| !value.trim().is_empty())
+        });
+        response_event_signals(event_type, &value).semantic_output
+    })
+}
 
 #[test]
 fn semantic_output_should_detect_a_text_delta() {
@@ -165,7 +192,7 @@ fn semantic_output_should_accept_crlf_frame_boundaries() {
 
 #[test]
 fn transport_should_prefer_websocket_when_requested_without_history() {
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-test", "be brief", Vec::new());
+    let mut request = codex_request("gpt-test", "be brief", Vec::new());
     request.use_websocket = true;
 
     assert_eq!(
@@ -177,7 +204,7 @@ fn transport_should_prefer_websocket_when_requested_without_history() {
 
 #[test]
 fn transport_should_mark_unknown_previous_response_as_external() {
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-test", "be brief", Vec::new());
+    let mut request = codex_request("gpt-test", "be brief", Vec::new());
     request.set_previous_response_id(Some("resp_previous".to_owned()));
 
     assert_eq!(
@@ -189,7 +216,7 @@ fn transport_should_mark_unknown_previous_response_as_external() {
 
 #[test]
 fn transport_should_allow_forced_http_sse() {
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-test", "be brief", Vec::new());
+    let mut request = codex_request("gpt-test", "be brief", Vec::new());
     request.set_previous_response_id(Some("resp_previous".to_owned()));
     request.use_websocket = true;
     request.force_http_sse = true;
@@ -202,7 +229,7 @@ fn transport_should_allow_forced_http_sse() {
 
 #[test]
 fn transport_should_require_exact_websocket_for_connection_local_continuation() {
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-test", "be brief", Vec::new());
+    let mut request = codex_request("gpt-test", "be brief", Vec::new());
     request.set_previous_response_id(Some("resp_previous".to_owned()));
     request.previous_response_scope = Some(PreviousResponseScope::ConnectionLocal);
 
@@ -215,7 +242,7 @@ fn transport_should_require_exact_websocket_for_connection_local_continuation()
 
 #[test]
 fn exact_continuation_should_remain_stricter_than_forced_websocket() {
-    let mut preferred = CodexResponsesRequest::new_http_sse("gpt-test", "be brief", Vec::new());
+    let mut preferred = codex_request("gpt-test", "be brief", Vec::new());
     preferred.use_websocket = true;
     assert_eq!(
         transport_requirement(&preferred),

backend/crates/providers/openai/tests/transport/protocol/websocket.rs

@@ -1,11 +1,13 @@
 use provider_openai::transport::protocol::responses::CodexResponsesRequest;
 use provider_openai::transport::protocol::websocket::{
-    OpeningAuditSnapshot, is_terminal_websocket_event, websocket_audit_artifact_from_attempt,
-    websocket_event_to_sse_frame, websocket_metadata_turn_state, websocket_payload_audit_snapshot,
+    OpeningAuditSnapshot, websocket_audit_artifact_from_attempt, websocket_event_to_sse_frame,
+    websocket_metadata_turn_state, websocket_payload_audit_snapshot,
     websocket_response_completed_id, websocket_response_create_payload_text,
 };
 use serde_json::json;
 
+use super::super::{codex_request, codex_request_with_prompt_cache_key};
+
 #[test]
 fn websocket_payload_audit_should_redact_sensitive_content_and_preserve_key_order() {
     let mut request = CodexResponsesRequest::from_body(
@@ -14,13 +16,13 @@ fn websocket_payload_audit_should_redact_sensitive_content_and_preserve_key_orde
             "instructions": "private instructions",
             "input": [{"role": "user", "content": "private prompt"}],
             "tools": [{"type": "function", "name": "private-tool"}],
-            "service_tier": "flex"
+            "service_tier": "flex",
+            "prompt_cache_key": "cache-secret"
         })
         .as_object()
         .expect("request object")
         .clone(),
     );
-    request.set_prompt_cache_key(Some("cache-secret".to_owned()));
     request.set_client_metadata(Some(json!({"thread_id": "thread-secret"})));
 
     let snapshot = websocket_payload_audit_snapshot(&request);
@@ -53,12 +55,12 @@ fn websocket_payload_audit_should_redact_sensitive_content_and_preserve_key_orde
 
 #[test]
 fn websocket_response_create_payload_should_preserve_transparent_body_order() {
-    let mut request = CodexResponsesRequest::new_http_sse(
+    let mut request = codex_request_with_prompt_cache_key(
         "gpt-test",
         "capture instructions",
         vec![json!({"role": "user", "content": "capture prompt"})],
+        "session-1",
     );
-    request.set_prompt_cache_key(Some("session-1".to_owned()));
     request.set_client_metadata(Some(json!({"thread_id": "capture-thread"})));
 
     let payload = websocket_response_create_payload_text(&request).expect("serialize payload");
@@ -81,7 +83,7 @@ fn websocket_response_create_payload_should_preserve_transparent_body_order() {
 
 #[test]
 fn websocket_response_create_payload_should_keep_explicit_empty_instructions() {
-    let request = CodexResponsesRequest::new_http_sse("gpt-test", "", Vec::new());
+    let request = codex_request("gpt-test", "", Vec::new());
     let payload = websocket_response_create_payload_text(&request).expect("serialize payload");
     let value = serde_json::from_str::<serde_json::Value>(&payload).expect("payload JSON");
 
@@ -145,19 +147,6 @@ fn websocket_metadata_turn_state_should_accept_case_insensitive_header() {
     );
 }
 
-#[test]
-fn websocket_terminal_event_should_cover_all_current_terminal_types() {
-    for event in [
-        "response.completed",
-        "response.incomplete",
-        "response.failed",
-        "error",
-    ] {
-        assert!(is_terminal_websocket_event(event));
-    }
-    assert!(!is_terminal_websocket_event("response.output_text.delta"));
-}
-
 #[test]
 fn websocket_completed_id_should_validate_the_official_shape() {
     let valid = json!({
@@ -204,7 +193,7 @@ fn websocket_typed_events_should_remain_transparent_without_schema_filtering() {
 
 #[test]
 fn websocket_audit_artifact_should_record_opening_and_redacted_payload() {
-    let mut request = CodexResponsesRequest::new_http_sse(
+    let mut request = codex_request(
         "gpt-test",
         "private instructions",
         vec![json!({"role": "user", "content": "private prompt"})],

backend/crates/providers/openai/tests/transport/request.rs

@@ -4,7 +4,7 @@ use gateway_core::operation::{
 };
 use serde_json::{Map, Value, json};
 
-use provider_openai::{CodexRequestEncodeError, codex_request_semantics, encode_generate_request};
+use provider_openai::{CodexRequestEncodeError, encode_generate_request};
 
 #[test]
 fn encoder_should_preserve_openai_wire_fields_without_deriving_accountless_pool_identity() {
@@ -291,7 +291,9 @@ fn observability_semantics_should_reuse_codex_turn_metadata() {
         .expect("generate")
         .with_provider_options(providers);
 
-    let semantics = codex_request_semantics(&request);
+    let semantics = encode_generate_request(&request, "observability")
+        .expect("observability request should encode")
+        .semantics();
 
     assert_eq!(semantics.request_kind.as_deref(), Some("compaction"));
     assert_eq!(semantics.subagent_kind.as_deref(), Some("review"));
@@ -324,7 +326,9 @@ fn observability_semantics_should_use_the_transparent_openai_payload() {
     .expect("OpenAI payload");
     let request = GenerateRequest::from_protocol_payload(Vec::new(), payload);
 
-    let semantics = codex_request_semantics(&request);
+    let semantics = encode_generate_request(&request, "observability")
+        .expect("transparent OpenAI request should encode")
+        .semantics();
 
     assert_eq!(semantics.reasoning_preset, Some("ultra"));
     assert!(semantics.compact);

backend/crates/providers/openai/tests/transport/websocket.rs

@@ -133,7 +133,7 @@ async fn codex_backend_client_should_decode_permessage_deflate_context_takeover_
         test_wire_profile(),
     )
     .with_websocket_pool(pool);
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request("gpt-5.5", "be brief", Vec::new());
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
 
@@ -156,7 +156,7 @@ async fn codex_backend_client_should_decode_permessage_deflate_context_takeover_
 
 #[test]
 fn websocket_connection_should_prepare_response_create_payload_text() {
-    let request = CodexResponsesRequest::new_http_sse(
+    let request = codex_request(
         "gpt-5.5",
         "be brief",
         vec![json!({
@@ -192,15 +192,15 @@ fn websocket_connection_should_prepare_response_create_payload_text() {
 
 #[test]
 fn websocket_connection_should_prepare_capture_payload_with_canonical_field_order() {
-    let mut request = CodexResponsesRequest::new_http_sse(
+    let mut request = codex_request_with_prompt_cache_key(
         "gpt-5.5",
         "private capture instructions",
         vec![json!({
             "role": "user",
             "content": "private capture prompt",
         })],
+        "session-1",
     );
-    request.set_prompt_cache_key(Some("session-1".to_string()));
     request.set_client_metadata(Some(json!({
         "thread_id": "capture-thread-secret",
         "safe": "capture",
@@ -263,7 +263,7 @@ async fn websocket_execute_response_create_request_should_collect_completed_sse(
         websocket.close(None).await.unwrap();
         payload
     });
-    let request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let request = codex_request("gpt-5.5", "be brief", Vec::new());
     let prepared = CodexWebSocketConnection::responses_create_request(
         &format!("http://{addr}"),
         "dGhlIHNhbXBsZSBub25jZQ==",
@@ -313,7 +313,7 @@ async fn websocket_execute_response_create_request_should_return_business_coded_
             .unwrap();
         websocket.close(None).await.unwrap();
     });
-    let request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let request = codex_request("gpt-5.5", "be brief", Vec::new());
     let prepared = CodexWebSocketConnection::responses_create_request(
         &format!("http://{addr}"),
         "dGhlIHNhbXBsZSBub25jZQ==",
@@ -565,7 +565,7 @@ async fn websocket_execute_response_create_request_should_forward_typed_events_w
             .unwrap();
         websocket.close(None).await.unwrap();
     });
-    let request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let request = codex_request("gpt-5.5", "be brief", Vec::new());
     let prepared = CodexWebSocketConnection::responses_create_request(
         &format!("http://{addr}"),
         "dGhlIHNhbXBsZSBub25jZQ==",
@@ -767,7 +767,7 @@ async fn websocket_execute_response_create_request_should_forward_incomplete_res
             .unwrap();
         websocket.close(None).await.unwrap();
     });
-    let request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let request = codex_request("gpt-5.5", "be brief", Vec::new());
     let prepared = CodexWebSocketConnection::responses_create_request(
         &format!("http://{addr}"),
         "dGhlIHNhbXBsZSBub25jZQ==",
@@ -817,7 +817,7 @@ async fn websocket_execute_response_create_request_should_reject_invalid_complet
             .unwrap();
         websocket.close(None).await.unwrap();
     });
-    let request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let request = codex_request("gpt-5.5", "be brief", Vec::new());
     let prepared = CodexWebSocketConnection::responses_create_request(
         &format!("http://{addr}"),
         "dGhlIHNhbXBsZSBub25jZQ==",
@@ -860,7 +860,7 @@ async fn websocket_execute_response_create_request_should_reject_completed_witho
             .unwrap();
         websocket.close(None).await.unwrap();
     });
-    let request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let request = codex_request("gpt-5.5", "be brief", Vec::new());
     let prepared = CodexWebSocketConnection::responses_create_request(
         &format!("http://{addr}"),
         "dGhlIHNhbXBsZSBub25jZQ==",
@@ -909,7 +909,7 @@ async fn websocket_execute_response_create_request_should_return_error_terminal_
             .unwrap();
         websocket.close(None).await.unwrap();
     });
-    let request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let request = codex_request("gpt-5.5", "be brief", Vec::new());
     let prepared = CodexWebSocketConnection::responses_create_request(
         &format!("http://{addr}"),
         "dGhlIHNhbXBsZSBub25jZQ==",
@@ -951,7 +951,7 @@ async fn codex_backend_client_should_timeout_when_upstream_is_silent() {
         test_wire_profile(),
     )
     .with_websocket_initial_event_timeout(Some(Duration::from_millis(30)));
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request("gpt-5.5", "be brief", Vec::new());
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
 
@@ -1061,7 +1061,7 @@ async fn websocket_execute_response_create_request_should_reply_to_server_ping_b
             .unwrap();
         websocket.close(None).await.unwrap();
     });
-    let request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let request = codex_request("gpt-5.5", "be brief", Vec::new());
     let prepared = CodexWebSocketConnection::responses_create_request(
         &format!("http://{addr}"),
         "dGhlIHNhbXBsZSBub25jZQ==",
@@ -1097,7 +1097,7 @@ async fn codex_backend_client_stream_should_reject_binary_websocket_event() {
             .unwrap();
         websocket.close(None).await.unwrap();
     });
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request("gpt-5.5", "be brief", Vec::new());
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
     request.force_http_sse = false;
@@ -1230,7 +1230,7 @@ async fn codex_backend_client_stream_should_error_when_websocket_closes_before_t
             .unwrap();
         websocket.close(None).await.unwrap();
     });
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request("gpt-5.5", "be brief", Vec::new());
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
     request.force_http_sse = false;
@@ -1475,7 +1475,7 @@ async fn codex_backend_client_stream_should_wait_for_terminal_after_active_webso
             .unwrap();
         websocket.close(None).await.unwrap();
     });
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request("gpt-5.5", "be brief", Vec::new());
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
     request.force_http_sse = false;
@@ -1537,7 +1537,7 @@ async fn codex_backend_client_stream_should_timeout_when_active_websocket_stalls
             .unwrap();
         futures::future::pending::<()>().await;
     });
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request("gpt-5.5", "be brief", Vec::new());
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
     request.force_http_sse = false;
@@ -1632,7 +1632,7 @@ async fn codex_backend_client_should_use_websocket_when_previous_response_id_is_
         websocket.close(None).await.unwrap();
         payload
     });
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request("gpt-5.5", "be brief", Vec::new());
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
     let pool = Arc::new(CodexWebSocketPool::new(8, Duration::from_mins(1)));

backend/crates/providers/openai/tests/transport/websocket_pool.rs

@@ -42,10 +42,10 @@ async fn codex_backend_client_should_reuse_pooled_websocket_for_same_account_and
         test_wire_profile(),
     )
     .with_websocket_pool(Arc::clone(&pool));
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request =
+        codex_request_with_prompt_cache_key("gpt-5.5", "be brief", Vec::new(), "conversation-pool");
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
-    request.set_prompt_cache_key(Some("conversation-pool".to_string()));
 
     let first = backend
         .create_response(
@@ -98,8 +98,7 @@ async fn exact_continuation_reuses_owning_socket_after_connection_profile_update
         profile.clone(),
     )
     .with_websocket_pool(Arc::new(CodexWebSocketPool::new(1, Duration::from_mins(1))));
-    let mut first_request = pooled_websocket_request("wire-profile-continuation");
-    first_request.set_store(false);
+    let first_request = pooled_websocket_request("wire-profile-continuation");
     let first = backend
         .create_response(
             &first_request,
@@ -180,10 +179,14 @@ async fn codex_backend_client_should_open_fresh_socket_when_idle_pooled_websocke
         test_wire_profile(),
     )
     .with_websocket_pool(Arc::clone(&pool));
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request_with_prompt_cache_key(
+        "gpt-5.5",
+        "be brief",
+        Vec::new(),
+        "conversation-pool-silent",
+    );
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
-    request.set_prompt_cache_key(Some("conversation-pool-silent".to_string()));
 
     let first = backend
         .create_response(
@@ -471,10 +474,10 @@ async fn codex_backend_client_should_not_reuse_pooled_websocket_across_local_acc
         test_wire_profile(),
     )
     .with_websocket_pool(Arc::clone(&pool));
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request =
+        codex_request_with_prompt_cache_key("gpt-5.5", "be brief", Vec::new(), "conversation-pool");
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
-    request.set_prompt_cache_key(Some("conversation-pool".to_string()));
 
     let first = backend
         .create_response_with_pool_account_started_at(
@@ -1448,10 +1451,14 @@ async fn codex_backend_client_should_keep_idle_pooled_websocket_alive_across_rep
         test_wire_profile(),
     )
     .with_websocket_pool(Arc::clone(&pool));
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request_with_prompt_cache_key(
+        "gpt-5.5",
+        "be brief",
+        Vec::new(),
+        "conversation-pool-background",
+    );
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
-    request.set_prompt_cache_key(Some("conversation-pool-background".to_string()));
 
     let first = backend
         .create_response(
@@ -1611,10 +1618,14 @@ async fn codex_backend_client_should_close_idle_pooled_websocket_when_account_is
         test_wire_profile(),
     )
     .with_websocket_pool(Arc::clone(&pool));
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request_with_prompt_cache_key(
+        "gpt-5.5",
+        "be brief",
+        Vec::new(),
+        "conversation-pool-evict",
+    );
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
-    request.set_prompt_cache_key(Some("conversation-pool-evict".to_string()));
 
     let first = backend
         .create_response(
@@ -1691,10 +1702,14 @@ async fn codex_backend_client_should_stop_reusing_pooled_websockets_after_shutdo
         test_wire_profile(),
     )
     .with_websocket_pool(Arc::clone(&pool));
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request_with_prompt_cache_key(
+        "gpt-5.5",
+        "be brief",
+        Vec::new(),
+        "conversation-pool-shutdown",
+    );
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
-    request.set_prompt_cache_key(Some("conversation-pool-shutdown".to_string()));
 
     let first = backend
         .create_response(
@@ -1800,10 +1815,14 @@ async fn codex_backend_client_should_close_idle_pooled_websocket_after_liveness_
         test_wire_profile(),
     )
     .with_websocket_pool(Arc::clone(&pool));
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request = codex_request_with_prompt_cache_key(
+        "gpt-5.5",
+        "be brief",
+        Vec::new(),
+        "conversation-pool-liveness",
+    );
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
-    request.set_prompt_cache_key(Some("conversation-pool-liveness".to_string()));
 
     let first = backend
         .create_response(
@@ -1894,10 +1913,10 @@ async fn codex_backend_client_should_discard_pooled_websocket_after_error_termin
         test_wire_profile(),
     )
     .with_websocket_pool(pool);
-    let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", Vec::new());
+    let mut request =
+        codex_request_with_prompt_cache_key("gpt-5.5", "be brief", Vec::new(), "conversation-pool");
     request.set_previous_response_id(Some("resp_previous".to_string()));
     request.previous_response_scope = Some(PreviousResponseScope::Persisted);
-    request.set_prompt_cache_key(Some("conversation-pool".to_string()));
 
     let first = backend
         .create_response(

backend/crates/providers/xai/src/credential/types.rs

@@ -180,11 +180,6 @@ impl PreparedGrokCredentialRotation {
         self
     }
 
-    #[must_use]
-    pub const fn holds_refresh_lease(&self) -> bool {
-        self.refresh_guards.is_some()
-    }
-
     /// 将 command 与 lease 一起交给 App；返回的 guard 必须活到 CAS 提交结束。
     #[must_use]
     pub fn into_parts(

backend/crates/providers/xai/src/transport/network.rs

@@ -172,13 +172,6 @@ pub trait GrokEndpointPolicy: fmt::Debug + Send + Sync {
 #[derive(Debug, Default)]
 pub struct OfficialGrokEndpointPolicy;
 
-impl OfficialGrokEndpointPolicy {
-    #[must_use]
-    pub fn accepts_resolved_address(address: IpAddr) -> bool {
-        is_public_ip(address)
-    }
-}
-
 impl GrokEndpointPolicy for OfficialGrokEndpointPolicy {
     fn build_oauth_client(
         &self,

backend/crates/providers/xai/tests/credential/refresh.rs

@@ -255,7 +255,6 @@ async fn manual_refresh_returns_prepared_rotation_without_writing_store() {
         .prepare_manual_refresh(&id, CredentialRevision::new(1).expect("revision"))
         .await
         .expect("prepare manual refresh");
-    assert!(prepared.holds_refresh_lease());
     assert_eq!(store.account(&id).expect("account").revision().get(), 1);
     assert_eq!(refresher.prepare_calls.load(Ordering::SeqCst), 1);
     let (_profile, credential, guard) = prepared.into_parts();

backend/crates/providers/xai/tests/transport/network.rs

@@ -374,6 +374,7 @@ async fn billing_transport_should_get_exact_credits_resource_without_redirect()
 
 #[test]
 fn endpoint_policy_should_reject_private_and_documentation_addresses() {
+    let policy = GrokDnsResolutionPolicy::official_oauth();
     for address in [
         "127.0.0.1",
         "10.0.0.1",
@@ -382,11 +383,19 @@ fn endpoint_policy_should_reject_private_and_documentation_addresses() {
         "2001:db8::1",
         "::1",
     ] {
+        let address = address.parse().expect("fixture address");
+        assert_eq!(
+            policy
+                .plan_system_resolution("auth.x.ai", &[address])
+                .expect("official host"),
+            GrokDnsResolutionPlan::TrustedDoh,
+            "{address} must require trusted fallback"
+        );
         assert!(
-            !OfficialGrokEndpointPolicy::accepts_resolved_address(
-                address.parse().expect("fixture address")
-            ),
-            "{address} must be rejected"
+            policy
+                .validate_trusted_doh_resolution("auth.x.ai", &[address])
+                .is_err(),
+            "{address} must be rejected after trusted resolution"
         );
     }
 }