diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index 9f14ad7cf..b4f2f6ccc 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/backend/crates/providers/openai/src/provider/failure.rs @@ -506,6 +506,27 @@ pub(super) fn apply_websocket_recovery_policy( failure: &mut MappedProviderFailure, context: WebSocketRecoveryContext<'_>, ) { + // 账号级上游拒绝(额度、限流、封禁等)没有可恢复的传输语义:同账号重试只会 + // 命中同一拒绝,而上游 retry-after 通常是小时级的额度窗口。保留账号失败记录 + // 让健康度立即生效,并把回放安全证据随错误交给 Core 换号恢复,与 HTTP 路径 + // 的调度语义保持一致。 + if failure.account_failure.is_some() && failure.error.replay_is_safe() { + tracing::warn!( + request_id = context.request_id, + attempt_index = context.attempt_index, + account_id = context.account_id, + websocket_failure_kind = failure.error.kind().as_str(), + websocket_failure_code = failure + .error + .upstream_code() + .map_or("", OpaqueUpstreamValue::as_str), + upstream_status_code = failure.error.upstream_status().unwrap_or_default(), + upstream_status_code_present = failure.error.upstream_status().is_some(), + transport_requirement = context.requirement.as_str(), + "OpenAI upstream WebSocket rejected by an account-level failure; deferring to account rotation" + ); + return; + } let session_budget_exhausted = context.session_affinity_key.is_some_and(|key| { context .session_transport_recovery diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index 62a0afc9a..5c286b97a 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -465,6 +465,8 @@ fn contract_account_scope() -> Arc { "acct_websocket_busy_replay", "acct_websocket_metadata_close", "acct_websocket_turn_state", + "acct_ws_quota_a", + "acct_ws_quota_b", ] .into_iter() .map(|id| { @@ -3124,6 +3126,60 @@ async fn downstream_websocket_new_chain_should_override_session_and_attempt_http server.await.unwrap(); } +#[tokio::test] +async fn websocket_opening_account_rejection_keeps_replay_safe_without_transport_retry() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_ws_quota_a").await; + create_account(&store, "acct_ws_quota_b").await; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (mut opening, _) = listener.accept().await.unwrap(); + let request = capture_http_request(&mut opening).await; + assert!(String::from_utf8_lossy(&request).starts_with("GET /codex/responses")); + // 额度耗尽拒绝通常携带小时级的 retry-after;同账号重试注定再次命中。 + let body = r#"{"error":{"message":"You have reached your usage limit.","type":"rate_limit_error"}}"#; + opening + .write_all( + format!( + "HTTP/1.1 429 Too Many Requests\r\nretry-after: 129600\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ) + .as_bytes(), + ) + .await + .unwrap(); + }); + + let provider = provider_with_base_url(&store, base_url); + let mut stream = provider + .execute( + planned_request("openai", generate_operation()), + context("req_ws_quota_rotation", CancellationToken::new()), + ) + .await + .expect("prepare quota-rejected stream"); + let error = loop { + match stream.next().await { + Some(Ok(_)) => {} + Some(Err(error)) => break error, + None => panic!("quota rejection must surface as a terminal error"), + } + }; + server.await.expect("upstream server"); + + assert_eq!(error.kind(), ProviderErrorKind::RateLimited); + assert_eq!(error.upstream_status(), Some(429)); + assert_eq!(error.send_state(), UpstreamSendState::NotSent); + assert!( + error.replay_is_safe(), + "before-payload rejection is replay safe" + ); + // 回放安全的账号级拒绝必须把换号决策留给 Core,不得钉死同账号传输重试 + // (旧行为会携带小时级 retry-after 的同账号重试标记,请求必然超时)。 + assert_eq!(error.pre_delivery_retry(), None); +} + #[tokio::test] async fn websocket_turn_state_metadata_is_exposed_through_response_observation() { let store = Arc::new(MemoryAccountStore::default());