diff --git a/backend/crates/gateway-api/src/openai/responses/mod.rs b/backend/crates/gateway-api/src/openai/responses/mod.rs index 367c70354..799c6f4a6 100644 --- a/backend/crates/gateway-api/src/openai/responses/mod.rs +++ b/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 { diff --git a/backend/crates/gateway-api/src/openai/responses/request.rs b/backend/crates/gateway-api/src/openai/responses/request.rs index 1669eb5f9..e1af6fa91 100644 --- a/backend/crates/gateway-api/src/openai/responses/request.rs +++ b/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, turn_metadata: Option, beta_features: Option, @@ -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 { - 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, diff --git a/backend/crates/gateway-api/src/openai/responses/response.rs b/backend/crates/gateway-api/src/openai/responses/response.rs index a0a2c4560..83610b3a1 100644 --- a/backend/crates/gateway-api/src/openai/responses/response.rs +++ b/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, -} - -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) { - (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, 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, - ) -> Result { - 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, ResponseEncodeError> { self.started = Some(meta.clone()); let response = response_base( diff --git a/backend/crates/gateway-api/src/openai/responses/websocket.rs b/backend/crates/gateway-api/src/openai/responses/websocket.rs index d592c49f9..fe5b03956 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket.rs +++ b/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 { - 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, diff --git a/backend/crates/gateway-api/tests/openai/responses/mod.rs b/backend/crates/gateway-api/tests/openai/responses/mod.rs index 379f03d8e..983c46fc7 100644 --- a/backend/crates/gateway-api/tests/openai/responses/mod.rs +++ b/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 { + decode_request_with_headers(body, &HeaderMap::new(), &openai_provider()) +} + +fn decode_response_create( + payload: &str, +) -> Result { + 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, +} + +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 { + 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 { 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"); diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket.rs b/backend/crates/gateway-api/tests/openai/responses/websocket.rs index ec59be568..c50385250 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket.rs +++ b/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( diff --git a/backend/crates/gateway-core/src/routing/mod.rs b/backend/crates/gateway-core/src/routing/mod.rs index 4879c93dd..ff4349a1e 100644 --- a/backend/crates/gateway-core/src/routing/mod.rs +++ b/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 diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index d0b6b7630..3da902b01 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -335,7 +335,7 @@ fn complete_stream(total_tokens: Option) -> Vec 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, ) -> (Arc, Arc) { 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( diff --git a/backend/crates/gateway-protocol/src/openai/sse.rs b/backend/crates/gateway-protocol/src/openai/sse.rs index c2816a5f4..122e30b53 100644 --- a/backend/crates/gateway-protocol/src/openai/sse.rs +++ b/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 { sse_frame_separator_bytes(bytes).map(|(position, separator_len)| position + separator_len) diff --git a/backend/crates/gateway-protocol/tests/openai/sse.rs b/backend/crates/gateway-protocol/tests/openai/sse.rs index c6192e892..11fd54f3d 100644 --- a/backend/crates/gateway-protocol/tests/openai/sse.rs +++ b/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!( diff --git a/backend/crates/providers/openai/src/credential/admin.rs b/backend/crates/providers/openai/src/credential/admin.rs index 322d54bb5..4bcc3c45d 100644 --- a/backend/crates/providers/openai/src/credential/admin.rs +++ b/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( diff --git a/backend/crates/providers/openai/src/credential/mod.rs b/backend/crates/providers/openai/src/credential/mod.rs index 3db07717f..57003f7a7 100644 --- a/backend/crates/providers/openai/src/credential/mod.rs +++ b/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, }; diff --git a/backend/crates/providers/openai/src/credential/repository.rs b/backend/crates/providers/openai/src/credential/repository.rs index ce0270f1f..fff684cff 100644 --- a/backend/crates/providers/openai/src/credential/repository.rs +++ b/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 { - 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, diff --git a/backend/crates/providers/openai/src/credential/types.rs b/backend/crates/providers/openai/src/credential/types.rs index 37b443337..131d459d3 100644 --- a/backend/crates/providers/openai/src/credential/types.rs +++ b/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>, - 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, diff --git a/backend/crates/providers/openai/src/lib.rs b/backend/crates/providers/openai/src/lib.rs index 4a8e59b24..6d35a83b4 100644 --- a/backend/crates/providers/openai/src/lib.rs +++ b/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 初始化后交给组装根的最小能力集。 diff --git a/backend/crates/providers/openai/src/transport/diagnostics.rs b/backend/crates/providers/openai/src/transport/diagnostics.rs index b1c3eaf70..c0bc4c78e 100644 --- a/backend/crates/providers/openai/src/transport/diagnostics.rs +++ b/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 { diff --git a/backend/crates/providers/openai/src/transport/endpoints.rs b/backend/crates/providers/openai/src/transport/endpoints.rs index 518b9b80a..94f5890eb 100644 --- a/backend/crates/providers/openai/src/transport/endpoints.rs +++ b/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 { usage_endpoint_paths(base_url) diff --git a/backend/crates/providers/openai/src/transport/headers.rs b/backend/crates/providers/openai/src/transport/headers.rs index 3b561f14b..c94554e76 100644 --- a/backend/crates/providers/openai/src/transport/headers.rs +++ b/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 { - 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, diff --git a/backend/crates/providers/openai/src/transport/mod.rs b/backend/crates/providers/openai/src/transport/mod.rs index a747264a8..95dd299a8 100644 --- a/backend/crates/providers/openai/src/transport/mod.rs +++ b/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::{ diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index 4648ac025..39e5b87fd 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/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::(&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, - instructions: impl Into, - input: Vec, - ) -> 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 { &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) { - 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") diff --git a/backend/crates/providers/openai/src/transport/protocol/websocket.rs b/backend/crates/providers/openai/src/transport/protocol/websocket.rs index 92d6e3197..46f2411bd 100644 --- a/backend/crates/providers/openai/src/transport/protocol/websocket.rs +++ b/backend/crates/providers/openai/src/transport/protocol/websocket.rs @@ -181,14 +181,6 @@ pub fn websocket_response_completed_id(raw: &str) -> Result, 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); diff --git a/backend/crates/providers/openai/src/transport/request.rs b/backend/crates/providers/openai/src/transport/request.rs index 8ae1ae01b..855e505d5 100644 --- a/backend/crates/providers/openai/src/transport/request.rs +++ b/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, CodexRequestEncodeError> { let mut input = Vec::new(); for message in request.messages() { diff --git a/backend/crates/providers/openai/src/transport/websocket/breaker.rs b/backend/crates/providers/openai/src/transport/websocket/breaker.rs index 9e7db6ade..aec6bfb0e 100644 --- a/backend/crates/providers/openai/src/transport/websocket/breaker.rs +++ b/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 { diff --git a/backend/crates/providers/openai/tests/admin.rs b/backend/crates/providers/openai/tests/admin.rs index 812f1ea85..1f3b10915 100644 --- a/backend/crates/providers/openai/tests/admin.rs +++ b/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( diff --git a/backend/crates/providers/openai/tests/credential/admin.rs b/backend/crates/providers/openai/tests/credential/admin.rs index ed37d02ac..a2528db95 100644 --- a/backend/crates/providers/openai/tests/credential/admin.rs +++ b/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()) diff --git a/backend/crates/providers/openai/tests/credential/catalog.rs b/backend/crates/providers/openai/tests/credential/catalog.rs index faa6e6d38..aadfa9c6d 100644 --- a/backend/crates/providers/openai/tests/credential/catalog.rs +++ b/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, 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) -> CodexCredentialCatalogService { diff --git a/backend/crates/providers/openai/tests/credential/contract.rs b/backend/crates/providers/openai/tests/credential/contract.rs index b10b6d6e5..4f3bf80c3 100644 --- a/backend/crates/providers/openai/tests/credential/contract.rs +++ b/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, 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) -> AttemptContext { diff --git a/backend/crates/providers/openai/tests/credential/oauth.rs b/backend/crates/providers/openai/tests/credential/oauth.rs index 0a0dbec71..b1f0f7eb5 100644 --- a/backend/crates/providers/openai/tests/credential/oauth.rs +++ b/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 { diff --git a/backend/crates/providers/openai/tests/credential/quota.rs b/backend/crates/providers/openai/tests/credential/quota.rs index 53b1e4d3c..50835966a 100644 --- a/backend/crates/providers/openai/tests/credential/quota.rs +++ b/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) -> 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 { diff --git a/backend/crates/providers/openai/tests/credential/refresh.rs b/backend/crates/providers/openai/tests/credential/refresh.rs index d86e23f06..90c07f824 100644 --- a/backend/crates/providers/openai/tests/credential/refresh.rs +++ b/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 { diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index de6c0595e..a154fb02d 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/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) { 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) { 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, account_id: &str, value: &str) { diff --git a/backend/crates/providers/openai/tests/support.rs b/backend/crates/providers/openai/tests/support.rs index 5b9a70d95..84f99d1a9 100644 --- a/backend/crates/providers/openai/tests/support.rs +++ b/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 { let id = ProviderAccountId::new(id).ok()?; self.accounts diff --git a/backend/crates/providers/openai/tests/transport/diagnostics.rs b/backend/crates/providers/openai/tests/transport/diagnostics.rs index 3b242cf29..c10b59ece 100644 --- a/backend/crates/providers/openai/tests/transport/diagnostics.rs +++ b/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")) ); } diff --git a/backend/crates/providers/openai/tests/transport/endpoints.rs b/backend/crates/providers/openai/tests/transport/endpoints.rs index 10a8ab35b..14731ffb3 100644 --- a/backend/crates/providers/openai/tests/transport/endpoints.rs +++ b/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" ); } diff --git a/backend/crates/providers/openai/tests/transport/headers.rs b/backend/crates/providers/openai/tests/transport/headers.rs index 0625ac7c0..b95317483 100644 --- a/backend/crates/providers/openai/tests/transport/headers.rs +++ b/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()); diff --git a/backend/crates/providers/openai/tests/transport/http_client.rs b/backend/crates/providers/openai/tests/transport/http_client.rs index 14950b5b4..358edcb2a 100644 --- a/backend/crates/providers/openai/tests/transport/http_client.rs +++ b/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 diff --git a/backend/crates/providers/openai/tests/transport/latency.rs b/backend/crates/providers/openai/tests/transport/latency.rs index 2ef623672..36f78d076 100644 --- a/backend/crates/providers/openai/tests/transport/latency.rs +++ b/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; } diff --git a/backend/crates/providers/openai/tests/transport/mod.rs b/backend/crates/providers/openai/tests/transport/mod.rs index e01451a6c..da9d74552 100644 --- a/backend/crates/providers/openai/tests/transport/mod.rs +++ b/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, + instructions: impl Into, + input: Vec, +) -> CodexResponsesRequest { + CodexResponsesRequest::from_body(codex_request_body(model, instructions, input)) +} + +fn codex_request_with_prompt_cache_key( + model: impl Into, + instructions: impl Into, + input: Vec, + prompt_cache_key: impl Into, +) -> 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, + instructions: impl Into, + input: Vec, +) -> Map { + 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 { .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]; diff --git a/backend/crates/providers/openai/tests/transport/protocol/responses.rs b/backend/crates/providers/openai/tests/transport/protocol/responses.rs index fb11159ef..d6d9bc28f 100644 --- a/backend/crates/providers/openai/tests/transport/protocol/responses.rs +++ b/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::(&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), diff --git a/backend/crates/providers/openai/tests/transport/protocol/websocket.rs b/backend/crates/providers/openai/tests/transport/protocol/websocket.rs index c56e69962..63af06136 100644 --- a/backend/crates/providers/openai/tests/transport/protocol/websocket.rs +++ b/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::(&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"})], diff --git a/backend/crates/providers/openai/tests/transport/request.rs b/backend/crates/providers/openai/tests/transport/request.rs index e05c5b16a..566e8a98f 100644 --- a/backend/crates/providers/openai/tests/transport/request.rs +++ b/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); diff --git a/backend/crates/providers/openai/tests/transport/websocket.rs b/backend/crates/providers/openai/tests/transport/websocket.rs index d1b64c062..d9922104e 100644 --- a/backend/crates/providers/openai/tests/transport/websocket.rs +++ b/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))); diff --git a/backend/crates/providers/openai/tests/transport/websocket_pool.rs b/backend/crates/providers/openai/tests/transport/websocket_pool.rs index 240d50c9e..de6ede43f 100644 --- a/backend/crates/providers/openai/tests/transport/websocket_pool.rs +++ b/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( diff --git a/backend/crates/providers/xai/src/credential/types.rs b/backend/crates/providers/xai/src/credential/types.rs index 60d50a710..8dd6c70f0 100644 --- a/backend/crates/providers/xai/src/credential/types.rs +++ b/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( diff --git a/backend/crates/providers/xai/src/transport/network.rs b/backend/crates/providers/xai/src/transport/network.rs index 3f3885c8a..c1a199875 100644 --- a/backend/crates/providers/xai/src/transport/network.rs +++ b/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, diff --git a/backend/crates/providers/xai/tests/credential/refresh.rs b/backend/crates/providers/xai/tests/credential/refresh.rs index 0761c1b11..48a5d90a2 100644 --- a/backend/crates/providers/xai/tests/credential/refresh.rs +++ b/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(); diff --git a/backend/crates/providers/xai/tests/transport/network.rs b/backend/crates/providers/xai/tests/transport/network.rs index c62c064fd..94148c90b 100644 --- a/backend/crates/providers/xai/tests/transport/network.rs +++ b/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" ); } }