From 8e2b2cbdf9d6f16d1d8a7ea5c57c81b109bea020 Mon Sep 17 00:00:00 2001 From: Zyy Date: Tue, 23 Jun 2026 17:48:09 +0800 Subject: [PATCH] fix: align responses transport parity Co-authored-by: Codex --- src/proxy/dispatch/responses.rs | 28 ++++++- src/proxy/openai/responses.rs | 19 ++++- src/runtime/bootstrap.rs | 8 +- src/runtime/services.rs | 31 ++++++-- src/upstream/protocol/responses.rs | 6 +- .../dispatch/chat_upstream/responses_http.rs | 6 +- tests/integration/runtime/mod.rs | 1 + tests/integration/runtime/services.rs | 74 +++++++++++++++++++ .../upstream/protocol/openai_responses.rs | 11 ++- 9 files changed, 169 insertions(+), 15 deletions(-) create mode 100644 tests/integration/runtime/services.rs diff --git a/src/proxy/dispatch/responses.rs b/src/proxy/dispatch/responses.rs index 96827970d..7b2b16e82 100644 --- a/src/proxy/dispatch/responses.rs +++ b/src/proxy/dispatch/responses.rs @@ -81,7 +81,7 @@ use super::implicit_resume::{ continuation_input_start, implicit_resume_allowed, ImplicitResumeSnapshot, }; -use crate::proxy::openai::responses::response_failed_sse_event; +use crate::proxy::openai::responses::response_failed_sse_event_with_id; #[derive(Clone, Copy)] enum ExhaustedAccountClass { @@ -2255,7 +2255,7 @@ async fn send_live_response_stream_tail( return None; } } - let failure = premature_close_failed_event(failure_detail); + let failure = premature_close_failed_event(latest_response_id(&body_text), failure_detail); body_text.push_str(&failure); body_bytes.extend_from_slice(failure.as_bytes()); if sender.send(Ok(Bytes::from(failure))).await.is_err() { @@ -2303,12 +2303,32 @@ fn missing_sse_event_separator(body: &str) -> Option<&'static str> { } } -fn premature_close_failed_event(detail: Option<&str>) -> String { +fn latest_response_id(body: &str) -> Option { + parse_sse_events(body).ok().and_then(|events| { + events.iter().rev().find_map(|event| { + serde_json::from_str::(&event.data) + .ok() + .and_then(|data| { + data.pointer("/response/id") + .and_then(Value::as_str) + .filter(|id| !id.trim().is_empty()) + .map(ToString::to_string) + }) + }) + }) +} + +fn premature_close_failed_event(response_id: Option, detail: Option<&str>) -> String { let message = match detail.filter(|value| !value.trim().is_empty()) { Some(detail) => format!("Upstream stream closed before response.completed: {detail}"), None => "Upstream stream closed before response.completed".to_string(), }; - response_failed_sse_event("server_error", "stream_disconnected", &message) + response_failed_sse_event_with_id( + response_id.as_deref(), + "server_error", + "stream_disconnected", + &message, + ) } async fn finalize_live_response_stream(context: LiveResponseStreamContext, body: String) { diff --git a/src/proxy/openai/responses.rs b/src/proxy/openai/responses.rs index 055f70452..40e964111 100644 --- a/src/proxy/openai/responses.rs +++ b/src/proxy/openai/responses.rs @@ -459,15 +459,29 @@ fn non_empty_str(value: Option<&Value>) -> Option<&str> { /// 编码 OpenAI Responses `response.failed` SSE 事件。 pub fn response_failed_sse_event(error_type: &str, code: &str, message: &str) -> String { + response_failed_sse_event_with_id(None, error_type, code, message) +} + +/// 使用指定 response id 编码 OpenAI Responses `response.failed` SSE 事件。 +pub fn response_failed_sse_event_with_id( + response_id: Option<&str>, + error_type: &str, + code: &str, + message: &str, +) -> String { let error = json!({ "type": error_type, "code": code, "message": message, }); + let response_id = response_id + .filter(|value| !value.trim().is_empty()) + .map(ToString::to_string) + .unwrap_or_else(|| format!("resp_proxy_{}", uuid::Uuid::new_v4().simple())); let data = json!({ "type": "response.failed", "response": { - "id": format!("resp_proxy_{}", uuid::Uuid::new_v4().simple()), + "id": response_id, "status": "failed", "error": error, }, @@ -534,6 +548,9 @@ async fn handle_responses( return invalid_responses_request_response().into_response(); }; apply_responses_header_context(&mut openai_request, &headers); + if openai_request.use_websocket.is_none() { + openai_request.use_websocket = Some(true); + } if let Some(subagent) = forced_subagent { force_openai_subagent(&mut openai_request, subagent); } else if let Some(subagent) = openai_subagent_from_headers(&headers) { diff --git a/src/runtime/bootstrap.rs b/src/runtime/bootstrap.rs index ab3147ca6..cbe689b91 100644 --- a/src/runtime/bootstrap.rs +++ b/src/runtime/bootstrap.rs @@ -69,8 +69,12 @@ async fn build_application( .ensure_current_seed(&default_fingerprint) .await?; let runtime_config = RuntimeConfig::from(config.clone()); - let services = - Services::with_installation_id(&config, stores, runtime_fingerprint, Some(installation_id)); + let services = Services::try_with_installation_id( + &config, + stores, + runtime_fingerprint, + Some(installation_id), + )?; let created_default_admin = services .admin_sessions diff --git a/src/runtime/services.rs b/src/runtime/services.rs index 357b72125..8cfacedaa 100644 --- a/src/runtime/services.rs +++ b/src/runtime/services.rs @@ -29,7 +29,9 @@ use crate::{ fingerprint::{Fingerprint, FingerprintRepository}, models::{AdminModelService, ModelService, ModelSnapshotStore}, token_client::{default_openai_token_client, TokenClientConfig}, - transport::{CodexBackendClient, CodexModelCatalogClient}, + transport::{ + build_reqwest_client, CodexBackendClient, CodexModelCatalogClient, CustomCaError, + }, }, }; @@ -79,7 +81,16 @@ pub struct Services { impl Services { pub fn new(config: &AppConfig, stores: BackgroundTaskStores, fingerprint: Fingerprint) -> Self { - Self::with_installation_id(config, stores, fingerprint, None) + Self::try_new(config, stores, fingerprint) + .expect("failed to build runtime services with configured TLS transport") + } + + pub fn try_new( + config: &AppConfig, + stores: BackgroundTaskStores, + fingerprint: Fingerprint, + ) -> Result { + Self::try_with_installation_id(config, stores, fingerprint, None) } pub fn with_installation_id( @@ -88,12 +99,22 @@ impl Services { fingerprint: Fingerprint, installation_id: Option, ) -> Self { + Self::try_with_installation_id(config, stores, fingerprint, installation_id) + .expect("failed to build runtime services with configured TLS transport") + } + + pub fn try_with_installation_id( + config: &AppConfig, + stores: BackgroundTaskStores, + fingerprint: Fingerprint, + installation_id: Option, + ) -> Result { let installation_id = installation_id.filter(|id| !id.trim().is_empty()); let account_store_trait = StdArc::new(stores.accounts.clone()) as StdArc; let codex = { let client = CodexBackendClient::new( - reqwest::Client::new(), + build_reqwest_client(config.tls.force_http11)?, config.api.base_url.clone(), fingerprint.clone(), ); @@ -203,7 +224,7 @@ impl Services { cloudflare_recovery, )); - Self { + Ok(Self { models, admin_models, accounts: account_store_trait, @@ -222,7 +243,7 @@ impl Services { fingerprint, installation_id, background_tasks: stores, - } + }) } } diff --git a/src/upstream/protocol/responses.rs b/src/upstream/protocol/responses.rs index ba249ffd6..66cf54c54 100644 --- a/src/upstream/protocol/responses.rs +++ b/src/upstream/protocol/responses.rs @@ -119,7 +119,11 @@ pub fn transport_for_request(request: &CodexResponsesRequest) -> CodexTransport return CodexTransport::WebSocketRequired; } - CodexTransport::WebSocketPreferred + if request.use_websocket { + return CodexTransport::WebSocketPreferred; + } + + CodexTransport::HttpSse } /// 判断请求在 WebSocket 失败后是否允许 HTTP SSE 回退。 diff --git a/tests/integration/proxy/dispatch/chat_upstream/responses_http.rs b/tests/integration/proxy/dispatch/chat_upstream/responses_http.rs index 87dbf8a94..926ab9b5f 100644 --- a/tests/integration/proxy/dispatch/chat_upstream/responses_http.rs +++ b/tests/integration/proxy/dispatch/chat_upstream/responses_http.rs @@ -1047,7 +1047,10 @@ data: {"delta":"partial before transport failure"} #[tokio::test] async fn responses_stream_should_emit_failed_event_when_upstream_closes_without_completed() { - let first_frame = r#"event: response.output_text.delta + let first_frame = r#"event: response.created +data: {"response":{"id":"resp_clean_close"}} + +event: response.output_text.delta data: {"delta":"partial before clean close"} "#; @@ -1097,6 +1100,7 @@ data: {"delta":"partial before clean close"} assert!(rest.contains("event: response.failed")); assert!(rest.contains("stream_disconnected")); + assert!(rest.contains(r#""id":"resp_clean_close""#)); assert!(rest.ends_with("data: [DONE]\n\n")); } diff --git a/tests/integration/runtime/mod.rs b/tests/integration/runtime/mod.rs index ee7d103a8..2f96872ed 100644 --- a/tests/integration/runtime/mod.rs +++ b/tests/integration/runtime/mod.rs @@ -1,2 +1,3 @@ pub mod account_pool_restore; +pub mod services; pub mod tasks; diff --git a/tests/integration/runtime/services.rs b/tests/integration/runtime/services.rs new file mode 100644 index 000000000..acf1981bf --- /dev/null +++ b/tests/integration/runtime/services.rs @@ -0,0 +1,74 @@ +use std::process::Command; + +use codex_proxy_rs::{ + admin::{ + auth::service::SqliteAdminSessionStore, keys::service::SqliteClientKeyStore, + monitoring::event_store::SqliteEventLogStore, + }, + infra::{crypto::SecretBox, identity::ApiKeyHasher}, + runtime::services::{BackgroundTaskStores, Services}, + upstream::{ + accounts::{ + cookies::SqliteCookieStore, store::SqliteAccountStore, token_refresh::RefreshLeaseStore, + }, + fingerprint::{Fingerprint, FingerprintRepository}, + transport::CODEX_CA_CERT_ENV, + }, +}; + +use crate::support::{config::test_config, sqlite::init_test_db}; + +#[tokio::test] +async fn services_try_new_should_use_configured_tls_transport_builder() { + const CASE_ENV: &str = "CODEX_PROXY_TEST_SERVICES_TLS_CASE"; + + if std::env::var(CASE_ENV).as_deref() == Ok("invalid_ca") { + let (pool, temp_dir) = init_test_db("services-tls.sqlite").await; + let config = test_config(format!( + "sqlite://{}", + temp_dir.path().join("services-tls.sqlite").display() + )); + let secret_box = SecretBox::load_or_create(temp_dir.path().join("master.key")).unwrap(); + let hasher = ApiKeyHasher::load_or_create(temp_dir.path().join("api-key-pepper.key")) + .expect("api key hasher"); + let stores = BackgroundTaskStores { + accounts: SqliteAccountStore::new(pool.clone(), secret_box.clone()), + admin_sessions: SqliteAdminSessionStore::new(pool.clone()), + cookies: SqliteCookieStore::new(pool.clone(), secret_box), + fingerprints: FingerprintRepository::new(pool.clone()), + session_affinity: + codex_proxy_rs::proxy::dispatch::session_affinity::SqliteSessionAffinityStore::new( + pool.clone(), + ), + refresh_leases: RefreshLeaseStore::new(pool.clone()), + client_keys: SqliteClientKeyStore::new(pool.clone(), hasher), + event_logs: SqliteEventLogStore::new(pool), + }; + + let error = match Services::try_new(&config, stores, Fingerprint::default_for_tests()) { + Ok(_) => panic!("invalid custom CA should fail service transport construction"), + Err(error) => error, + }; + assert!(error + .to_string() + .contains("Failed to read CA certificate file")); + return; + } + + let current_exe = std::env::current_exe().expect("current test binary path"); + let output = Command::new(current_exe) + .arg("--exact") + .arg("runtime::services::services_try_new_should_use_configured_tls_transport_builder") + .arg("--nocapture") + .env(CASE_ENV, "invalid_ca") + .env(CODEX_CA_CERT_ENV, "/tmp/codex-proxy-rs-missing-ca.pem") + .output() + .expect("run isolated services TLS test case"); + + assert!( + output.status.success(), + "isolated services TLS test failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} diff --git a/tests/integration/upstream/protocol/openai_responses.rs b/tests/integration/upstream/protocol/openai_responses.rs index bb1864534..08db6c129 100644 --- a/tests/integration/upstream/protocol/openai_responses.rs +++ b/tests/integration/upstream/protocol/openai_responses.rs @@ -8,10 +8,19 @@ fn codex_responses_request_should_enable_http_sse_defaults() { } #[test] -fn codex_responses_transport_should_prefer_websocket_without_history() { +fn codex_responses_transport_should_use_http_sse_by_default_without_history() { let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", vec![json!({})]); request.force_http_sse = false; + assert_eq!(transport_for_request(&request), CodexTransport::HttpSse); + assert!(http_sse_fallback_allowed(&request)); +} + +#[test] +fn codex_responses_transport_should_prefer_websocket_when_requested_without_history() { + let mut request = CodexResponsesRequest::new_http_sse("gpt-5.5", "be brief", vec![json!({})]); + request.use_websocket = true; + assert_eq!( transport_for_request(&request), CodexTransport::WebSocketPreferred