fix: align responses transport parity

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

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<String> {
+    parse_sse_events(body).ok().and_then(|events| {
+        events.iter().rev().find_map(|event| {
+            serde_json::from_str::<Value>(&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<String>, 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) {

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) {

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

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, CustomCaError> {
+        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<String>,
     ) -> 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<String>,
+    ) -> Result<Self, CustomCaError> {
         let installation_id = installation_id.filter(|id| !id.trim().is_empty());
         let account_store_trait =
             StdArc::new(stores.accounts.clone()) as StdArc<dyn AccountStoreTrait>;
         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,
-        }
+        })
     }
 }
 

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 回退。

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"));
 }
 

tests/integration/runtime/mod.rs

@@ -1,2 +1,3 @@
 pub mod account_pool_restore;
+pub mod services;
 pub mod tasks;

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)
+    );
+}

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