starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 08:06:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(openai): preserve upstream failures during fallback

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-07-31 15:33:22 +08:00
co-authored by Codex
parent e012f24ef2
commit 5c9da8a39d
17 changed files with 804 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-api/src/openai/error.rs
+12 -15
View File
@@ -7,8 +7,8 @@ use axum::{
response::{IntoResponse, Response},
};
use gateway_core::{
engine::{EngineError, UpstreamSendState},
error::{GatewayError, GatewayErrorKind, ProviderErrorKind},
engine::EngineError,
error::{GatewayError, GatewayErrorKind},
};
use serde_json::{Value, json};
@@ -72,19 +72,6 @@ pub fn model_not_found_response() -> (StatusCode, Json<Value>) {
#[must_use]
pub fn gateway_error_from_engine(error: &EngineError) -> GatewayError {
match error {
EngineError::Provider(error)
if error.kind() == ProviderErrorKind::Unavailable
&& error.send_state() == UpstreamSendState::NotSent =>
{
let gateway = GatewayError::new(
GatewayErrorKind::NoAvailableProvider,
"no upstream provider is currently available for this request",
);
match error.client_visible_upstream_error() {
Some(upstream) => gateway.with_client_visible_upstream_error(upstream.clone()),
None => gateway,
}
}
EngineError::Provider(error) => GatewayError::from_provider(error),
EngineError::Cancelled => {
GatewayError::new(GatewayErrorKind::Cancelled, "request was cancelled")
@@ -185,6 +172,16 @@ pub const fn gateway_error_contract(
"server_error",
"no_available_provider",
),
GatewayErrorKind::AccountCapacityUnavailable => (
StatusCode::SERVICE_UNAVAILABLE,
"server_error",
"account_capacity_unavailable",
),
GatewayErrorKind::ProviderInfrastructureUnavailable => (
StatusCode::SERVICE_UNAVAILABLE,
"server_error",
"provider_infrastructure_unavailable",
),
GatewayErrorKind::RateLimited => (
StatusCode::TOO_MANY_REQUESTS,
"rate_limit_error",
backend/crates/gateway-api/tests/openai/error.rs
+60 -15
View File
@@ -58,6 +58,30 @@ fn no_available_provider_error_should_map_to_service_unavailable() {
);
}
#[test]
fn account_capacity_error_should_map_to_service_unavailable() {
assert_eq!(
gateway_error_contract(GatewayErrorKind::AccountCapacityUnavailable),
(
StatusCode::SERVICE_UNAVAILABLE,
"server_error",
"account_capacity_unavailable",
)
);
}
#[test]
fn provider_infrastructure_error_should_map_to_service_unavailable() {
assert_eq!(
gateway_error_contract(GatewayErrorKind::ProviderInfrastructureUnavailable),
(
StatusCode::SERVICE_UNAVAILABLE,
"server_error",
"provider_infrastructure_unavailable",
)
);
}
#[test]
fn rate_limited_error_should_map_to_openai_retryable_status() {
assert_eq!(
@@ -122,29 +146,32 @@ fn engine_provider_error_should_preserve_retry_classification() {
#[test]
fn local_provider_capacity_exhaustion_should_map_to_service_unavailable() {
let error = EngineError::Provider(ProviderError::new(
ProviderErrorKind::Unavailable,
ProviderErrorKind::AccountCapacityUnavailable,
UpstreamSendState::NotSent,
));
assert_eq!(
gateway_error_from_engine(&error),
GatewayError::new(
GatewayErrorKind::NoAvailableProvider,
"no upstream provider is currently available for this request"
GatewayErrorKind::AccountCapacityUnavailable,
"all eligible upstream accounts are temporarily busy"
)
);
}
#[test]
fn local_provider_capacity_exhaustion_should_preserve_safe_provider_detail() {
fn no_eligible_provider_account_should_preserve_safe_provider_detail() {
let detail = ClientVisibleUpstreamError::new(
"no account is eligible for the requested model",
Some("no_eligible_account".to_owned()),
Some("account_unavailable_error".to_owned()),
);
let error = EngineError::Provider(
ProviderError::new(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent)
.with_client_visible_upstream_error(detail),
ProviderError::new(
ProviderErrorKind::NoEligibleAccount,
UpstreamSendState::NotSent,
)
.with_client_visible_upstream_error(detail),
);
let gateway = gateway_error_from_engine(&error);
@@ -164,6 +191,22 @@ fn local_provider_capacity_exhaustion_should_preserve_safe_provider_detail() {
);
}
#[test]
fn provider_infrastructure_failure_should_have_a_distinct_client_contract() {
let error = EngineError::Provider(ProviderError::new(
ProviderErrorKind::ProviderInfrastructureUnavailable,
UpstreamSendState::NotSent,
));
assert_eq!(
gateway_error_from_engine(&error),
GatewayError::new(
GatewayErrorKind::ProviderInfrastructureUnavailable,
"provider account infrastructure is temporarily unavailable"
)
);
}
#[test]
fn no_eligible_provider_account_should_map_to_service_unavailable() {
let error = EngineError::Provider(ProviderError::new(
@@ -181,16 +224,18 @@ fn no_eligible_provider_account_should_map_to_service_unavailable() {
}
#[test]
fn sent_provider_unavailability_should_remain_bad_gateway() {
let error = EngineError::Provider(ProviderError::new(
ProviderErrorKind::Unavailable,
UpstreamSendState::Sent,
));
fn provider_unavailability_should_remain_bad_gateway_regardless_of_send_state() {
for send_state in [UpstreamSendState::NotSent, UpstreamSendState::Sent] {
let error = EngineError::Provider(ProviderError::new(
ProviderErrorKind::Unavailable,
send_state,
));
assert_eq!(
gateway_error_from_engine(&error).kind(),
GatewayErrorKind::UpstreamUnavailable
);
assert_eq!(
gateway_error_from_engine(&error).kind(),
GatewayErrorKind::UpstreamUnavailable
);
}
}
#[test]
backend/crates/gateway-core/src/engine/coordinator.rs
+24 -24
View File
@@ -165,8 +165,8 @@ where
timings: ModelRequestTimings::default(),
client_response_id: None,
upstream_response_id: None,
last_account_exhaustion: None,
last_account_exhaustion_events: Vec::new(),
last_retryable_failure: None,
last_retryable_failure_events: Vec::new(),
provider_attempt_outcomes: Vec::new(),
pending_terminal_failure: None,
};
@@ -245,9 +245,10 @@ pub struct ResponseExecutionSession<S: ?Sized> {
timings: ModelRequestTimings,
client_response_id: Option<String>,
upstream_response_id: Option<String>,
last_account_exhaustion: Option<ProviderError>,
/// 最近一次账号耗尽在重试时被丢弃的原始失败批次；只用于后续空选路终态。
last_account_exhaustion_events: Vec<ProviderEvent>,
/// 最近一次为无感恢复而被丢弃的原始上游错误；只在后续空选路时成为终态。
last_retryable_failure: Option<ProviderError>,
/// 与 `last_retryable_failure` 同属一个 attempt 的原始失败批次。
last_retryable_failure_events: Vec<ProviderEvent>,
provider_attempt_outcomes: Vec<ProviderAttemptOutcome>,
/// 原子失败批次已交给协议层、但尚待下游提交后收敛的原 Provider 错误。
pending_terminal_failure: Option<PendingTerminalFailure>,
@@ -659,25 +660,30 @@ where
ProviderBoundary::Result(result) => match *result {
Ok(stream) => stream,
Err(error) => {
if error.kind() == ProviderErrorKind::NoEligibleAccount
&& let Some(exhaustion) = self.last_account_exhaustion.take()
if matches!(
error.kind(),
ProviderErrorKind::AccountCapacityUnavailable
| ProviderErrorKind::NoEligibleAccount
) && let Some(last_failure) = self.last_retryable_failure.take()
{
let send_state = self.current_send_state();
let mut events = std::mem::take(&mut self.last_account_exhaustion_events);
let mut events = std::mem::take(&mut self.last_retryable_failure_events);
if !events.is_empty() {
self.observe_atomic_terminal_events(&mut events);
return Ok(Some(PullOutcome::TerminalFailure {
events,
error: exhaustion,
error: last_failure,
send_state,
}));
}
self.finish_provider_error(&exhaustion).await?;
return Err(provider_engine_error(exhaustion));
self.finish_provider_error(&last_failure).await?;
return Err(provider_engine_error(last_failure));
}
if !(matches!(
error.kind(),
ProviderErrorKind::Unavailable | ProviderErrorKind::NoEligibleAccount
ProviderErrorKind::AccountCapacityUnavailable
| ProviderErrorKind::NoEligibleAccount
| ProviderErrorKind::ProviderInfrastructureUnavailable
) && error.send_state() == UpstreamSendState::NotSent)
{
self.record_provider_failure(candidate.provider().clone(), error.kind());
@@ -909,14 +915,10 @@ where
&mut self,
mut error: ProviderError,
) -> Result<StreamErrorOutcome, EngineError> {
// 原始 wire 只能活在本次决策栈内；后续 clone、attempt 记录与终态对象均只
// 接触已剥离的稳定错误字段。
// 原始 wire 只活在 request-local 决策状态；clone、attempt 记录与持久化终态
// 均只接触已剥离的稳定错误字段。
let mut atomic_client_events = error.take_atomic_client_events();
let current = self.current.take().ok_or(EngineError::NoActiveAttempt)?;
let account_exhausted = matches!(
error.kind(),
ProviderErrorKind::RateLimited | ProviderErrorKind::QuotaExhausted
);
self.record_provider_failure(current.metadata.provider().clone(), error.kind());
// attempt_send_state 是本 attempt 自身的发送事实，驱动重试门；
// 持久化与终态用请求级水位，二者不可混用（水位会把早先 attempt 的
@@ -976,8 +978,8 @@ where
continuation_retry || same_account_retry || ordinary_retry || pre_delivery_retry;
if retryable {
// 普通 clone 只保留稳定事实；原始失败响应只能随最后一次客户端可见
// 的账号耗尽错误继续，不能进入中间失败记录。
// 普通 clone 只保留稳定事实；原始 wire/HTTP response 由 request-local
// 所有权保留到下一次 attempt 成功，或最终空选路时返回客户端。
let persistence_error = self.request_persisted.then(|| error.clone());
if same_account_retry {
if let Some(account) = current.metadata.provider_account_id() {
@@ -992,10 +994,8 @@ where
{
self.excluded_accounts.insert(account.clone());
}
if account_exhausted {
self.last_account_exhaustion_events = atomic_client_events;
self.last_account_exhaustion = Some(error);
}
self.last_retryable_failure_events = atomic_client_events;
self.last_retryable_failure = Some(error);
if let Some(error) = persistence_error {
best_effort_store_write(
"record_intermediate_failure",
backend/crates/gateway-core/src/error.rs
+20
View File
@@ -129,8 +129,12 @@ pub enum ProviderErrorKind {
RateLimited,
/// Credential 配额耗尽。
QuotaExhausted,
/// 仍有符合条件的账号，但它们暂时没有可调度容量。
AccountCapacityUnavailable,
/// Provider 已确认当前请求无法选出可用账号。
NoEligibleAccount,
/// Provider 的账号存储、租约协调或本地凭据数据不可用。
ProviderInfrastructureUnavailable,
/// 请求超时。
Timeout,
/// 网络或 transport 失败。
@@ -156,7 +160,9 @@ impl ProviderErrorKind {
Self::PermissionDenied => "permission_denied",
Self::RateLimited => "rate_limited",
Self::QuotaExhausted => "quota_exhausted",
Self::AccountCapacityUnavailable => "account_capacity_unavailable",
Self::NoEligibleAccount => "no_eligible_account",
Self::ProviderInfrastructureUnavailable => "provider_infrastructure_unavailable",
Self::Timeout => "timeout",
Self::Transport => "transport",
Self::Protocol => "protocol",
@@ -711,6 +717,10 @@ pub enum GatewayErrorKind {
ModelNotFound,
/// 当前没有可用 target。
NoAvailableProvider,
/// 符合条件的上游账号暂时没有调度容量。
AccountCapacityUnavailable,
/// Provider 的本地账号基础设施不可用。
ProviderInfrastructureUnavailable,
/// 上游限流。
RateLimited,
/// 上游暂不可用。
@@ -734,6 +744,8 @@ impl GatewayErrorKind {
Self::PolicyDenied => "policy_denied",
Self::ModelNotFound => "model_not_found",
Self::NoAvailableProvider => "no_available_provider",
Self::AccountCapacityUnavailable => "account_capacity_unavailable",
Self::ProviderInfrastructureUnavailable => "provider_infrastructure_unavailable",
Self::RateLimited => "rate_limited",
Self::UpstreamUnavailable => "upstream_unavailable",
Self::Timeout => "timeout",
@@ -782,10 +794,18 @@ impl GatewayError {
GatewayErrorKind::RateLimited,
"upstream capacity is temporarily unavailable",
),
ProviderErrorKind::AccountCapacityUnavailable => Self::new(
GatewayErrorKind::AccountCapacityUnavailable,
"all eligible upstream accounts are temporarily busy",
),
ProviderErrorKind::NoEligibleAccount => Self::new(
GatewayErrorKind::NoAvailableProvider,
"no upstream provider is currently available for this request",
),
ProviderErrorKind::ProviderInfrastructureUnavailable => Self::new(
GatewayErrorKind::ProviderInfrastructureUnavailable,
"provider account infrastructure is temporarily unavailable",
),
ProviderErrorKind::Timeout => {
Self::new(GatewayErrorKind::Timeout, "upstream request timed out")
}
backend/crates/gateway-core/tests/engine/coordinator.rs
+125 -2
View File
@@ -29,8 +29,9 @@ use gateway_core::engine::{
ProviderAttemptOutcome, RecoveryReport, UpstreamSendState,
};
use gateway_core::error::{
ClientVisibleUpstreamResponse, ContinuationFailure, GatewayErrorKind, OpaqueUpstreamValue,
ProviderError, ProviderErrorKind, StoreError, StoreErrorKind,
ClientVisibleUpstreamError, ClientVisibleUpstreamResponse, ContinuationFailure,
GatewayErrorKind, OpaqueUpstreamValue, ProviderError, ProviderErrorKind, StoreError,
StoreErrorKind,
};
use gateway_core::event::{
ContentItem, ContentKind, GatewayEvent, ProtocolWireEvent, ProviderEvent,
@@ -2554,6 +2555,128 @@ fn pre_delivery_transport_failure_rotates_account_for_non_idempotent_generation(
assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded);
}
#[test]
fn final_capacity_exhaustion_returns_the_last_retryable_upstream_failure() {
let operation = operation(RetrySafety::NonIdempotent);
let route_plan = plan(&operation);
let raw_body = Bytes::from_static(b"{\"error\":\"last upstream body\"}\x00");
let upstream_error =
ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous)
.with_pre_delivery_retry()
.with_client_visible_upstream_error(ClientVisibleUpstreamError::new(
"message too big",
Some("1009".to_owned()),
Some("websocket_close_error".to_owned()),
))
.with_client_visible_upstream_response(ClientVisibleUpstreamResponse::new(
502,
Some(b"application/json".to_vec()),
raw_body.clone(),
));
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
account_id: "acct_first",
items: vec![Err(upstream_error)],
},
Script::Error(ProviderError::new(
ProviderErrorKind::AccountCapacityUnavailable,
UpstreamSendState::NotSent,
)),
]);
let mut session = block_on(coordinator.start(
model_request(&operation, SystemTime::now() + Duration::from_secs(30)),
operation,
route_plan,
None,
None,
CancellationToken::new(),
))
.expect("start execution");
let error = block_on(session.collect_uncommitted())
.expect_err("empty selection must return the retained upstream failure");
let EngineError::Provider(error) = error else {
panic!("expected retained provider error");
};
assert_eq!(error.kind(), ProviderErrorKind::Transport);
let detail = error
.client_visible_upstream_error()
.expect("structured upstream detail");
assert_eq!(detail.message(), "message too big");
assert_eq!(detail.code(), Some("1009"));
assert_eq!(detail.error_type(), Some("websocket_close_error"));
let response = error
.client_visible_upstream_response()
.expect("original upstream response");
assert_eq!(response.status(), 502);
assert_eq!(response.body(), &raw_body);
assert_eq!(provider.contexts.lock().expect("contexts lock").len(), 2);
let state = store.state.lock().expect("store lock");
assert_eq!(state.intermediate_failures, 1);
assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed);
assert_eq!(
state.finalizations[0].send_state,
UpstreamSendState::Ambiguous
);
}
#[test]
fn provider_infrastructure_failure_does_not_restore_the_last_upstream_failure() {
let operation = operation(RetrySafety::NonIdempotent);
let route_plan = plan(&operation);
let upstream_error =
ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous)
.with_pre_delivery_retry()
.with_client_visible_upstream_error(ClientVisibleUpstreamError::new(
"upstream marker",
Some("upstream_marker".to_owned()),
Some("upstream_error".to_owned()),
))
.with_client_visible_upstream_response(ClientVisibleUpstreamResponse::new(
502,
Some(b"application/json".to_vec()),
Bytes::from_static(b"{\"error\":\"upstream marker\"}"),
));
let (coordinator, store, provider) = coordinator(vec![
Script::Stream {
account_id: "acct_first",
items: vec![Err(upstream_error)],
},
Script::Error(ProviderError::new(
ProviderErrorKind::ProviderInfrastructureUnavailable,
UpstreamSendState::NotSent,
)),
]);
let mut session = block_on(coordinator.start(
model_request(&operation, SystemTime::now() + Duration::from_secs(30)),
operation,
route_plan,
None,
None,
CancellationToken::new(),
))
.expect("start execution");
let error = block_on(session.collect_uncommitted())
.expect_err("provider infrastructure failure must remain terminal");
let EngineError::Provider(error) = error else {
panic!("expected provider error");
};
assert_eq!(
error.kind(),
ProviderErrorKind::ProviderInfrastructureUnavailable
);
assert!(error.client_visible_upstream_error().is_none());
assert!(error.client_visible_upstream_response().is_none());
assert_eq!(provider.contexts.lock().expect("contexts lock").len(), 2);
let state = store.state.lock().expect("store lock");
assert_eq!(state.intermediate_failures, 1);
assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed);
}
#[test]
fn pre_delivery_retry_marker_is_ignored_after_downstream_commit() {
let operation = operation(RetrySafety::NonIdempotent);
backend/crates/providers/openai/src/provider.rs
+108 -11
View File
@@ -1767,7 +1767,10 @@ fn map_request_error(error: CodexRequestEncodeError) -> ProviderError {
fn map_selection_error(error: CredentialSelectionError) -> ProviderError {
match error {
CredentialSelectionError::CapacityUnavailable { retry_after } => {
let error = provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent);
let error = provider_error(
ProviderErrorKind::AccountCapacityUnavailable,
UpstreamSendState::NotSent,
);
match retry_after {
Some(retry) => error.with_retry_after(retry),
None => error,
@@ -1780,9 +1783,10 @@ fn map_selection_error(error: CredentialSelectionError) -> ProviderError {
CredentialSelectionError::InvalidCredential
| CredentialSelectionError::Store
| CredentialSelectionError::Coordinator
| CredentialSelectionError::CookiePolicy => {
provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent)
}
| CredentialSelectionError::CookiePolicy => provider_error(
ProviderErrorKind::ProviderInfrastructureUnavailable,
UpstreamSendState::NotSent,
),
}
}
@@ -2063,7 +2067,7 @@ fn stream_transport_allows_pre_delivery_retry(error: &CodexClientError) -> bool
CodexWebSocketExchangeError::Transport(_)
| CodexWebSocketExchangeError::PostSendAmbiguous { .. }
| CodexWebSocketExchangeError::SendTimeout { .. }
| CodexWebSocketExchangeError::ClosedBeforeTerminal
| CodexWebSocketExchangeError::ClosedBeforeTerminal(_)
| CodexWebSocketExchangeError::ReceiveIdleTimeout { .. }
| CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent { .. }
| CodexWebSocketExchangeError::InitialEventTimeout { .. }
@@ -2151,10 +2155,19 @@ fn map_client_error(
send_state,
))
}
CodexClientError::WebSocket(error) => MappedProviderFailure::plain(provider_error(
websocket_error_kind(&error),
websocket_send_state(&error),
)),
CodexClientError::WebSocket(error) => {
let client_visible_error = websocket_client_visible_error(&error);
let mut failure = MappedProviderFailure::plain(provider_error(
websocket_error_kind(&error),
websocket_send_state(&error),
));
if let Some(client_visible_error) = client_visible_error {
failure.error = failure
.error
.with_client_visible_upstream_error(client_visible_error);
}
failure
}
};
if let Some(continuation_failure) = continuation_failure {
failure.error = failure
@@ -2165,6 +2178,21 @@ fn map_client_error(
failure
}
fn websocket_client_visible_error(
error: &CodexWebSocketExchangeError,
) -> Option<ClientVisibleUpstreamError> {
let close = error.close_before_terminal()?;
let message = close
.reason()
.filter(|reason| !reason.is_empty())
.map_or_else(|| close.to_string(), str::to_owned);
Some(ClientVisibleUpstreamError::new(
message,
close.code().map(|code| code.to_string()),
Some("websocket_close_error".to_owned()),
))
}
fn map_upstream_failure(
mut failure: CodexUpstreamFailure,
observation: Option<ProviderResponseObservation>,
@@ -2333,7 +2361,7 @@ const fn websocket_send_state(error: &CodexWebSocketExchangeError) -> UpstreamSe
CodexWebSocketExchangeError::Transport(_)
| CodexWebSocketExchangeError::PostSendAmbiguous { .. }
| CodexWebSocketExchangeError::SendTimeout { .. }
| CodexWebSocketExchangeError::ClosedBeforeTerminal
| CodexWebSocketExchangeError::ClosedBeforeTerminal(_)
| CodexWebSocketExchangeError::ReceiveIdleTimeout { .. }
| CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent { .. }
| CodexWebSocketExchangeError::InitialEventTimeout { .. } => UpstreamSendState::Ambiguous,
@@ -2360,7 +2388,7 @@ const fn websocket_error_kind(error: &CodexWebSocketExchangeError) -> ProviderEr
CodexWebSocketExchangeError::Transport(_)
| CodexWebSocketExchangeError::Connect(_)
| CodexWebSocketExchangeError::PostSendAmbiguous { .. }
| CodexWebSocketExchangeError::ClosedBeforeTerminal
| CodexWebSocketExchangeError::ClosedBeforeTerminal(_)
| CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent { .. } => {
ProviderErrorKind::Transport
}
@@ -2619,3 +2647,72 @@ impl DaemonTask for OpenAiCatalogEtagTask {
})
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn capacity_selection_error_preserves_classification_and_retry_after() {
let retry_after = Duration::from_secs(7);
let error = map_selection_error(CredentialSelectionError::CapacityUnavailable {
retry_after: Some(retry_after),
});
assert_eq!(
(error.kind(), error.send_state(), error.retry_after()),
(
ProviderErrorKind::AccountCapacityUnavailable,
UpstreamSendState::NotSent,
Some(retry_after),
)
);
}
#[test]
fn selection_infrastructure_errors_have_a_distinct_classification() {
for selection_error in [
CredentialSelectionError::InvalidCredential,
CredentialSelectionError::Store,
CredentialSelectionError::Coordinator,
CredentialSelectionError::CookiePolicy,
] {
let error = map_selection_error(selection_error);
assert_eq!(
(error.kind(), error.send_state()),
(
ProviderErrorKind::ProviderInfrastructureUnavailable,
UpstreamSendState::NotSent,
)
);
}
}
#[test]
fn websocket_close_details_are_only_exposed_through_the_client_error() {
let close = CodexWebSocketExchangeError::closed_before_terminal(
Some(1009),
Some("message too big".to_owned()),
);
assert!(!format!("{close:?}").contains("message too big"));
assert!(!close.to_string().contains("message too big"));
let failure = map_stream_error(CodexClientError::WebSocket(
CodexWebSocketExchangeError::PostSendAmbiguous {
message: close.to_string(),
source: Some(Box::new(close)),
},
));
assert_eq!(failure.error.kind(), ProviderErrorKind::Transport);
assert_eq!(failure.error.send_state(), UpstreamSendState::Ambiguous);
assert!(failure.error.allows_pre_delivery_retry());
let detail = failure
.error
.client_visible_upstream_error()
.expect("WebSocket close detail");
assert_eq!(detail.message(), "message too big");
assert_eq!(detail.code(), Some("1009"));
assert_eq!(detail.error_type(), Some("websocket_close_error"));
}
}
backend/crates/providers/openai/src/transport/client.rs
+2
View File
@@ -477,6 +477,7 @@ pub enum CodexTransportDecision {
Http2BreakerOpen,
Http2PoolUnavailable,
Http2PreSendFailure,
Http2PreDeliveryFailure,
}
impl CodexTransportDecision {
@@ -491,6 +492,7 @@ impl CodexTransportDecision {
Self::Http2BreakerOpen => "http2_breaker_open",
Self::Http2PoolUnavailable => "http2_pool_unavailable",
Self::Http2PreSendFailure => "http2_ws_pre_send_failure",
Self::Http2PreDeliveryFailure => "http2_ws_pre_delivery_failure",
}
}
}
backend/crates/providers/openai/src/transport/client_sse.rs
+144 -39
View File
@@ -32,10 +32,11 @@ use crate::transport::{
},
response_meta,
websocket::{
CodexWebSocketConnection, CodexWebSocketPool, CodexWebSocketPoolKey,
DEFAULT_INITIAL_EVENT_TIMEOUT, WEBSOCKET_FAST_PATH_BUDGET, WebSocketOriginBreaker,
WebSocketPoolDecision, execute_prepared_response_create_request_stream,
post_send_ambiguous, prepare_response_create_request_with_pool, websocket_audit_dir,
CodexWebSocketConnection, CodexWebSocketExchangeError, CodexWebSocketPool,
CodexWebSocketPoolKey, CodexWebSocketStreamingExchange, DEFAULT_INITIAL_EVENT_TIMEOUT,
WEBSOCKET_FAST_PATH_BUDGET, WebSocketOriginBreaker, WebSocketPoolDecision,
execute_prepared_response_create_request_stream, post_send_ambiguous,
prepare_response_create_request_with_pool, websocket_audit_dir,
write_websocket_audit_artifact_from_env,
},
};
@@ -308,7 +309,7 @@ impl CodexBackendClient {
route,
metrics,
} = prepared;
let result = match route {
match route {
PreparedResponseRoute::Http => self
.create_response_stream_http_sse(request, context)
.await
@@ -317,50 +318,124 @@ impl CodexBackendClient {
response
}),
PreparedResponseRoute::WebSocket(route) => {
let PreparedWebSocketRoute { request, prepared } = *route;
execute_prepared_response_create_request_stream(&request, prepared)
.await
.map_err(websocket_exchange_error_to_client_error)
.map(|exchange| {
tracing::info!(
request_id = %context.request_id,
websocket_connection_id = %exchange.websocket_connection_id,
ws_pool = exchange.pool_decision.map_or("unpooled", WebSocketPoolDecision::kind),
"WebSocket response stream established"
);
CodexBackendStreamingResponse {
body: Box::pin(
exchange
.body
.map_err(post_send_ambiguous)
.map_err(websocket_exchange_error_to_client_error),
),
transport: CodexBackendTransport::WebSocket,
turn_state: exchange.turn_state,
set_cookie_headers: exchange.set_cookie_headers,
rate_limit_headers: exchange.rate_limit_headers,
rate_limit_header_updates: Some(exchange.rate_limit_header_updates),
turn_state_update: Some(exchange.turn_state_update),
websocket_pool_decision: exchange.pool_decision,
diagnostics: exchange.diagnostics,
response_metadata: exchange.response_metadata,
transport_metrics: metrics,
connection_local_continuation: exchange.connection_local_continuation,
}
})
let PreparedWebSocketRoute {
request: websocket_request,
prepared,
} = *route;
let delivery_wait_started_at = Instant::now();
let mut exchange = match execute_prepared_response_create_request_stream(
&websocket_request,
prepared,
)
.await
{
Ok(exchange) => exchange,
Err(error)
if requirement.allows_pre_delivery_http_fallback()
&& error.allows_pre_delivery_http_fallback() =>
{
return self
.fallback_to_http_before_websocket_delivery(
request,
context,
requirement,
metrics,
delivery_wait_started_at,
error,
)
.await;
}
Err(error) => return Err(websocket_exchange_error_to_client_error(error)),
};
if requirement.allows_pre_delivery_http_fallback()
&& let Err(error) = await_websocket_delivery_boundary(&mut exchange).await
{
if error.allows_pre_delivery_http_fallback() {
return self
.fallback_to_http_before_websocket_delivery(
request,
context,
requirement,
metrics,
delivery_wait_started_at,
error,
)
.await;
}
return Err(websocket_exchange_error_to_client_error(
post_send_ambiguous(error),
));
}
tracing::info!(
request_id = %context.request_id,
websocket_connection_id = %exchange.websocket_connection_id,
ws_pool = exchange.pool_decision.map_or("unpooled", WebSocketPoolDecision::kind),
"WebSocket response stream established"
);
Ok(CodexBackendStreamingResponse {
body: Box::pin(
exchange
.body
.map_err(post_send_ambiguous)
.map_err(websocket_exchange_error_to_client_error),
),
transport: CodexBackendTransport::WebSocket,
turn_state: exchange.turn_state,
set_cookie_headers: exchange.set_cookie_headers,
rate_limit_headers: exchange.rate_limit_headers,
rate_limit_header_updates: Some(exchange.rate_limit_header_updates),
turn_state_update: Some(exchange.turn_state_update),
websocket_pool_decision: exchange.pool_decision,
diagnostics: exchange.diagnostics,
response_metadata: exchange.response_metadata,
transport_metrics: metrics,
connection_local_continuation: exchange.connection_local_continuation,
})
}
};
result.inspect_err(|error| {
}
.inspect_err(|error| {
tracing::warn!(
request_id = %context.request_id,
transport_requirement = requirement.as_str(),
failure_phase = "post_send_or_explicit_response",
error = %error,
"Responses stream transport failed after preparation; automatic fallback is disabled"
"Responses stream transport failed after preparation"
);
})
}
async fn fallback_to_http_before_websocket_delivery(
&self,
request: &CodexResponsesRequest,
context: CodexRequestContext<'_>,
requirement: TransportRequirement,
mut metrics: CodexTransportMetrics,
delivery_wait_started_at: Instant,
error: CodexWebSocketExchangeError,
) -> CodexClientResult<CodexBackendStreamingResponse> {
let decision_wait_ms = metrics
.transport_decision_wait_ms
.unwrap_or_default()
.saturating_add(elapsed_duration_millis(delivery_wait_started_at.elapsed()));
metrics.decision = Some(CodexTransportDecision::Http2PreDeliveryFailure);
metrics.transport_decision_wait_ms = Some(decision_wait_ms);
tracing::warn!(
request_id = %context.request_id,
account_id = context.account_id.unwrap_or_default(),
transport_requirement = requirement.as_str(),
transport_decision = CodexTransportDecision::Http2PreDeliveryFailure.as_str(),
transport_decision_wait_ms = decision_wait_ms,
error = %error,
"WebSocket failed before first deliverable event; using same-account HTTP"
);
self.create_response_stream_http_sse(request, context)
.await
.map(|mut response| {
merge_preparation_metrics(&mut response.transport_metrics, metrics);
response
})
}
fn websocket_pool_key(
&self,
request: &CodexResponsesRequest,
@@ -563,6 +638,36 @@ impl CodexBackendClient {
}
}
async fn await_websocket_delivery_boundary(
exchange: &mut CodexWebSocketStreamingExchange,
) -> Result<(), CodexWebSocketExchangeError> {
let mut prelude = Vec::new();
loop {
match exchange.body.next().await {
Some(Ok(frame)) if is_websocket_lifecycle_prelude(&frame) => prelude.push(frame),
Some(Ok(frame)) => {
prelude.push(frame);
let remaining =
std::mem::replace(&mut exchange.body, Box::pin(futures::stream::empty()));
exchange.body =
Box::pin(futures::stream::iter(prelude.into_iter().map(Ok)).chain(remaining));
return Ok(());
}
Some(Err(error)) => return Err(error),
None => {
return Err(CodexWebSocketExchangeError::closed_before_terminal(
None, None,
));
}
}
}
}
fn is_websocket_lifecycle_prelude(frame: &[u8]) -> bool {
frame.starts_with(b"event: response.created\n")
|| frame.starts_with(b"event: response.in_progress\n")
}
async fn read_model_catalog_body(response: ReqwestResponse) -> CodexClientResult<Vec<u8>> {
if response
.content_length()
backend/crates/providers/openai/src/transport/protocol/responses.rs
+5
View File
@@ -139,6 +139,11 @@ impl TransportRequirement {
)
}
/// WebSocket 已发送 payload、但尚未交付业务事件时，是否允许同账号 HTTP 重放。
pub fn allows_pre_delivery_http_fallback(self) -> bool {
matches!(self, Self::NewChain)
}
/// 用于审计与遥测的稳定名称。
pub fn as_str(self) -> &'static str {
match self {
backend/crates/providers/openai/src/transport/websocket/error.rs
+85 -2
View File
@@ -71,8 +71,8 @@ pub enum CodexWebSocketExchangeError {
reason: PreviousResponseUnavailableReason,
},
/// 上游在 terminal 事件前关闭。
#[error("websocket closed before terminal event")]
ClosedBeforeTerminal,
#[error("{0}")]
ClosedBeforeTerminal(CodexWebSocketCloseError),
/// 上游在指定时间内没有发送任何事件。
#[error("websocket receive idle timeout after {timeout:?}")]
ReceiveIdleTimeout {
@@ -87,6 +87,9 @@ pub enum CodexWebSocketExchangeError {
ReusedConnectionDiedBeforeFirstEvent {
/// 底层失效原因。
message: String,
/// 原始 typed transport failure。
#[source]
source: Option<Box<CodexWebSocketExchangeError>>,
},
/// 建连并发送后，上游在配置时间内没有产生任何事件。
#[error("websocket first upstream event not received within {timeout:?}")]
@@ -96,6 +99,55 @@ pub enum CodexWebSocketExchangeError {
},
}
/// 上游在 terminal 事件前发送的 WebSocket close 信息。
///
/// close reason 只供原请求的客户端错误响应使用；`Debug` 与 `Display` 均不会输出它。
#[derive(Clone, PartialEq, Eq)]
pub struct CodexWebSocketCloseError {
code: Option<u16>,
reason: Option<String>,
}
impl CodexWebSocketCloseError {
pub(crate) fn new(code: Option<u16>, reason: Option<String>) -> Self {
Self { code, reason }
}
/// 返回上游 close code；没有 close frame 时为 `None`。
#[must_use]
pub const fn code(&self) -> Option<u16> {
self.code
}
/// 返回上游 close reason；仅可用于当前请求的客户端协议响应。
#[must_use]
pub fn reason(&self) -> Option<&str> {
self.reason.as_deref()
}
}
impl fmt::Debug for CodexWebSocketCloseError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("CodexWebSocketCloseError")
.field("code", &self.code)
.field("reason", &self.reason.as_ref().map(|_| "<redacted>"))
.finish()
}
}
impl fmt::Display for CodexWebSocketCloseError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self.code {
Some(code) => write!(
formatter,
"websocket closed before terminal event (code {code})"
),
None => formatter.write_str("websocket closed before terminal event"),
}
}
}
/// WebSocket 上游错误帧载荷。
#[derive(Clone, PartialEq, Eq)]
pub struct CodexWebSocketUpstreamError {
@@ -140,6 +192,25 @@ impl fmt::Display for CodexWebSocketUpstreamError {
}
impl CodexWebSocketExchangeError {
pub(crate) fn closed_before_terminal(code: Option<u16>, reason: Option<String>) -> Self {
Self::ClosedBeforeTerminal(CodexWebSocketCloseError::new(code, reason))
}
pub(crate) fn close_before_terminal(&self) -> Option<&CodexWebSocketCloseError> {
match self {
Self::ClosedBeforeTerminal(close) => Some(close),
Self::PostSendAmbiguous {
source: Some(source),
..
}
| Self::ReusedConnectionDiedBeforeFirstEvent {
source: Some(source),
..
} => source.close_before_terminal(),
_ => None,
}
}
pub(super) fn upstream(
status_code: u16,
retry_after_seconds: Option<u64>,
@@ -173,4 +244,16 @@ impl CodexWebSocketExchangeError {
| Self::ContinuationUnavailable { .. }
)
}
/// 首个业务事件交付前，普通 WebSocket 失败可切到同账号 HTTP。
pub(in crate::transport) fn allows_pre_delivery_http_fallback(&self) -> bool {
match self {
Self::Upstream(_) => false,
Self::PostSendAmbiguous {
source: Some(source),
..
} => source.allows_pre_delivery_http_fallback(),
_ => true,
}
}
}
backend/crates/providers/openai/src/transport/websocket/exchange/io.rs
+4 -3
View File
@@ -40,19 +40,20 @@ pub(super) fn reused_stream_receive_error(
error: CodexWebSocketExchangeError,
) -> CodexWebSocketExchangeError {
match error {
CodexWebSocketExchangeError::ClosedBeforeTerminal
CodexWebSocketExchangeError::ClosedBeforeTerminal(_)
| CodexWebSocketExchangeError::ReceiveIdleTimeout { .. }
| CodexWebSocketExchangeError::Transport(_) => {
reused_connection_died_before_first_event(&error)
reused_connection_died_before_first_event(error)
}
error => error,
}
}
pub(super) fn reused_connection_died_before_first_event(
error: &CodexWebSocketExchangeError,
error: CodexWebSocketExchangeError,
) -> CodexWebSocketExchangeError {
CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstEvent {
message: error.to_string(),
source: Some(Box::new(error)),
}
}
backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs
+12 -3
View File
@@ -208,14 +208,17 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
.await;
return;
}
tungstenite::Message::Close(_) => {
tungstenite::Message::Close(frame) => {
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::UpstreamClosed,
)
.await;
let error = CodexWebSocketExchangeError::ClosedBeforeTerminal;
let error = CodexWebSocketExchangeError::closed_before_terminal(
frame.as_ref().map(|frame| u16::from(frame.code)),
frame.map(|frame| frame.reason.to_string()),
);
let error = if reused_connection {
reused_stream_receive_error(error)
} else {
@@ -285,13 +288,19 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
}
}
let upstream_close = websocket
.exit_reason()
.and_then(|reason| reason.upstream_close().cloned());
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::UpstreamClosed,
)
.await;
let error = CodexWebSocketExchangeError::ClosedBeforeTerminal;
let error = upstream_close.map_or_else(
|| CodexWebSocketExchangeError::closed_before_terminal(None, None),
CodexWebSocketExchangeError::ClosedBeforeTerminal,
);
let error = if reused_connection {
reused_stream_receive_error(error)
} else {
backend/crates/providers/openai/src/transport/websocket/mod.rs
+1 -1
View File
@@ -19,7 +19,7 @@ pub use self::{
WebSocketOriginBreaker, WebSocketOriginBreakerConfig, WebSocketOriginBreakerDecision,
WebSocketOriginBreakerPermit,
},
error::{CodexWebSocketExchangeError, CodexWebSocketUpstreamError},
error::{CodexWebSocketCloseError, CodexWebSocketExchangeError, CodexWebSocketUpstreamError},
exchange::{
CodexWebSocketRateLimitHeaderUpdates, CodexWebSocketSseStream,
CodexWebSocketStreamingExchange, CodexWebSocketTurnStateUpdate,
backend/crates/providers/openai/src/transport/websocket/pump.rs
+34 -8
View File
@@ -29,6 +29,8 @@ use tokio::{
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, tungstenite::Message};
use uuid::Uuid;
use super::CodexWebSocketCloseError;
/// 底层 tungstenite WebSocket 流。
pub(crate) type RawWsStream = WebSocketStream<MaybeTlsStream<TcpStream>>;
@@ -69,13 +71,25 @@ const PUMP_MESSAGE_BUFFER: usize = 64;
pub(crate) enum PumpExitReason {
CommandChannelClosed,
LocalClose,
OutboundTransportError { message: String },
UpstreamCloseFrame { frame: Option<String> },
OutboundTransportError {
message: String,
},
UpstreamCloseFrame {
close: Option<CodexWebSocketCloseError>,
},
UpstreamEof,
InboundTransportError { message: String },
PongTimeout { timeout: Duration },
LivenessTimeout { timeout: Duration },
KeepaliveTransportError { message: String },
InboundTransportError {
message: String,
},
PongTimeout {
timeout: Duration,
},
LivenessTimeout {
timeout: Duration,
},
KeepaliveTransportError {
message: String,
},
MessageReceiverClosed,
}
@@ -100,7 +114,7 @@ impl PumpExitReason {
Self::OutboundTransportError { message }
| Self::InboundTransportError { message }
| Self::KeepaliveTransportError { message } => Some(message.clone()),
Self::UpstreamCloseFrame { frame } => frame.clone(),
Self::UpstreamCloseFrame { close } => close.as_ref().map(ToString::to_string),
Self::PongTimeout { timeout } | Self::LivenessTimeout { timeout } => {
Some(format!("{timeout:?}"))
}
@@ -111,6 +125,13 @@ impl PumpExitReason {
}
}
pub(crate) fn upstream_close(&self) -> Option<&CodexWebSocketCloseError> {
match self {
Self::UpstreamCloseFrame { close: Some(close) } => Some(close),
_ => None,
}
}
fn is_unexpected(&self) -> bool {
matches!(
self,
@@ -336,7 +357,12 @@ async fn pump_loop(
message => {
let terminal_reason = match &message {
Message::Close(frame) => Some(PumpExitReason::UpstreamCloseFrame {
frame: frame.as_ref().map(|frame| format!("{frame:?}")),
close: frame.as_ref().map(|frame| {
CodexWebSocketCloseError::new(
Some(u16::from(frame.code)),
Some(frame.reason.to_string()),
)
}),
}),
_ => None,
};
backend/crates/providers/openai/tests/transport/latency.rs
+137 -9
View File
@@ -5,6 +5,7 @@ use provider_openai::transport::{
WebSocketOriginBreaker, WebSocketOriginBreakerConfig, WebSocketOriginBreakerDecision,
},
};
use tokio_tungstenite::tungstenite::protocol::{CloseFrame, frame::coding::CloseCode};
use super::*;
@@ -913,14 +914,133 @@ async fn concurrent_same_key_should_singleflight_websocket_opening() {
}
#[tokio::test]
async fn payload_send_failure_should_not_open_http_fallback() {
async fn websocket_failure_before_first_delivery_should_use_same_account_http() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut websocket = accept_codex_test_websocket_with(stream, |request, response| {
assert_eq!(
request
.headers()
.get("authorization")
.and_then(|value| value.to_str().ok()),
Some("Bearer access-token")
);
assert_eq!(
request
.headers()
.get("chatgpt-account-id")
.and_then(|value| value.to_str().ok()),
Some("chatgpt-account")
);
response.headers_mut().insert(
"sec-websocket-extensions",
"permessage-deflate".parse().unwrap(),
);
})
.await;
let _payload = websocket.next().await.unwrap().unwrap();
websocket
.send(Message::Text(
json!({
"type": "response.created",
"response": {
"id": "resp_abandoned_websocket",
"model": "gpt-test"
}
})
.to_string()
.into(),
))
.await
.unwrap();
websocket
.send(Message::Text(
json!({
"type": "response.in_progress",
"response": {
"id": "resp_abandoned_websocket",
"model": "gpt-test"
}
})
.to_string()
.into(),
))
.await
.unwrap();
websocket
.close(Some(CloseFrame {
code: CloseCode::Size,
reason: "message too big".into(),
}))
.await
.unwrap();
let (mut http, _) = listener.accept().await.unwrap();
let request = read_http_request(&mut http).await;
assert!(request.starts_with("POST /codex/responses HTTP/1.1"));
assert_eq!(
read_header_value(&request, "authorization"),
Some("Bearer access-token")
);
assert_eq!(
read_header_value(&request, "chatgpt-account-id"),
Some("chatgpt-account")
);
write_completed_sse_response(&mut http).await;
});
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
format!("http://{addr}"),
test_wire_profile(),
)
.with_websocket_pool(Arc::new(CodexWebSocketPool::default()));
let response = backend
.create_response(
&new_chain_request("conversation-pre-delivery-fallback"),
request_context("req_pre_delivery_fallback", Some("chatgpt-account")),
)
.await
.expect("pre-delivery WebSocket failure should use same-account HTTP");
server.await.unwrap();
assert_eq!(response.transport, CodexBackendTransport::HttpSse);
assert_eq!(
response.transport_metrics.decision,
Some(CodexTransportDecision::Http2PreDeliveryFailure)
);
assert!(response.body.contains("resp_http_fallback"));
assert!(!response.body.contains("resp_abandoned_websocket"));
}
#[tokio::test]
async fn websocket_failure_after_first_delivery_should_not_open_http_fallback() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut websocket = accept_codex_test_websocket(stream).await;
let _payload = websocket.next().await.unwrap().unwrap();
websocket.close(None).await.unwrap();
websocket
.send(Message::Text(
json!({
"type": "response.output_text.delta",
"delta": "delivered"
})
.to_string()
.into(),
))
.await
.unwrap();
websocket
.close(Some(CloseFrame {
code: CloseCode::Size,
reason: "message too big after output".into(),
}))
.await
.unwrap();
assert!(
timeout(Duration::from_millis(100), listener.accept())
.await
@@ -936,17 +1056,25 @@ async fn payload_send_failure_should_not_open_http_fallback() {
let error = backend
.create_response(
&new_chain_request("conversation-post-send"),
request_context("req_post_send", Some("chatgpt-account")),
&new_chain_request("conversation-post-delivery"),
request_context("req_post_delivery", Some("chatgpt-account")),
)
.await
.expect_err("post-send close must not be replayed");
.expect_err("post-delivery close must not be replayed");
server.await.unwrap();
std::assert_matches!(
error,
CodexClientError::WebSocket(CodexWebSocketExchangeError::PostSendAmbiguous { .. })
);
let CodexClientError::WebSocket(CodexWebSocketExchangeError::PostSendAmbiguous {
source: Some(source),
..
}) = error
else {
panic!("expected typed post-delivery WebSocket failure");
};
let CodexWebSocketExchangeError::ClosedBeforeTerminal(close) = *source else {
panic!("expected upstream close frame source");
};
assert_eq!(close.code(), Some(1009));
assert_eq!(close.reason(), Some("message too big after output"));
}
#[tokio::test]
backend/crates/providers/openai/tests/transport/protocol/responses.rs
+22
View File
@@ -218,6 +218,17 @@ fn transport_should_prefer_websocket_when_requested_without_history() {
assert!(transport_requirement(&request).allows_pre_send_http_fallback());
}
#[test]
fn new_chain_should_allow_pre_delivery_http_fallback() {
let mut request = codex_request("gpt-test", "be brief", Vec::new());
request.use_websocket = true;
assert!(
transport_requirement(&request).allows_pre_delivery_http_fallback(),
"new chains may follow the TS same-account fallback behavior"
);
}
#[test]
fn transport_should_mark_unknown_previous_response_as_external() {
let mut request = codex_request("gpt-test", "be brief", Vec::new());
@@ -230,6 +241,17 @@ fn transport_should_mark_unknown_previous_response_as_external() {
assert!(transport_requirement(&request).allows_pre_send_http_fallback());
}
#[test]
fn previous_response_should_forbid_pre_delivery_http_fallback() {
let mut request = codex_request("gpt-test", "be brief", Vec::new());
request.set_previous_response_id(Some("resp_previous".to_owned()));
assert!(
!transport_requirement(&request).allows_pre_delivery_http_fallback(),
"a sent continuation must not be replayed on another transport"
);
}
#[test]
fn transport_should_allow_forced_http_sse() {
let mut request = codex_request("gpt-test", "be brief", Vec::new());
backend/crates/providers/openai/tests/transport/websocket.rs
+9 -1
View File
@@ -649,7 +649,15 @@ async fn websocket_execute_response_create_request_should_reject_binary_event()
.unwrap();
websocket.close(None).await.unwrap();
});
let prepared = prepared_websocket_request(&format!("http://{addr}"));
let mut request = codex_request("gpt-test", "be brief", Vec::new());
request.set_previous_response_id(Some("resp_binary_previous".to_owned()));
let prepared = CodexWebSocketConnection::responses_create_request(
&format!("http://{addr}"),
"dGhlIHNhbXBsZSBub25jZQ==",
vec![("authorization".to_owned(), "Bearer access-token".to_owned())],
&request,
)
.expect("prepare continuation WebSocket request");
let error = execute_response_create_request(&prepared)
.await