From 0f2e0b18a9d902dd1d94f2c0c832f9747ff9844f Mon Sep 17 00:00:00 2001 From: Zyy <39424122+zyycn@users.noreply.github.com> Date: Thu, 17 Sep 2026 15:07:31 +0800 Subject: [PATCH] feat(openai): recover quota-exhausted continuations through client replay (#140) Project replay-safe quota rejections to the existing client history recovery contract while preserving upstream failure facts and account isolation. Co-authored-by: Zyy Co-authored-by: Codex --- .../crates/gateway-api/tests/openai/error.rs | 30 ++ .../openai/responses/websocket/forward.rs | 26 ++ backend/crates/gateway-core/src/error.rs | 7 +- .../gateway-core/tests/engine/coordinator.rs | 117 ++--- .../openai/src/provider/execution.rs | 24 +- .../providers/openai/src/provider/failure.rs | 73 ++- .../openai/tests/provider/contract.rs | 422 ++++++++++++++++++ docs/api.md | 15 +- docs/architecture.md | 9 +- 9 files changed, 657 insertions(+), 66 deletions(-) diff --git a/backend/crates/gateway-api/tests/openai/error.rs b/backend/crates/gateway-api/tests/openai/error.rs index e4647fa41..7fc046455 100644 --- a/backend/crates/gateway-api/tests/openai/error.rs +++ b/backend/crates/gateway-api/tests/openai/error.rs @@ -615,3 +615,33 @@ mod model_routing { ); } } + +#[tokio::test] +async fn quota_recovery_http_response_uses_projection_without_replacing_upstream_facts() { + use bytes::Bytes; + use gateway_core::error::{ClientVisibleUpstreamResponse, OpaqueUpstreamValue}; + let body = Bytes::from_static(br#"{"error":{"code":"previous_response_not_found","type":"invalid_request_error","message":"Previous response was not found. Retrying the full request."}}"#); + let provider = ProviderError::new(ProviderErrorKind::QuotaExhausted, UpstreamSendState::Sent) + .with_status(429) + .with_upstream_code(OpaqueUpstreamValue::new("usage_limit_reached")) + .with_retry_after(std::time::Duration::from_secs(129_600)) + .with_client_visible_upstream_response(ClientVisibleUpstreamResponse::new( + 400, + Some(b"application/json".to_vec()), + body.clone(), + )); + let error = EngineError::Provider(provider); + let response = gateway_api::openai::error::engine_error_response(&error); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!(response.headers()["content-type"], "application/json"); + assert!(!response.headers().contains_key("retry-after")); + assert_eq!(to_bytes(response.into_body(), 4096).await.unwrap(), body); + let EngineError::Provider(provider) = error else { + unreachable!() + }; + assert_eq!(provider.upstream_status(), Some(429)); + assert_eq!( + provider.upstream_code().unwrap().as_str(), + "usage_limit_reached" + ); +} diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/forward.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/forward.rs index 5101cb11d..4e7d7d400 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/forward.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/forward.rs @@ -447,3 +447,29 @@ async fn deliverable_business_failures_are_not_rewritten_or_followed_by_a_second socket.close(None).await.unwrap(); } } + +#[tokio::test] +async fn initial_quota_recovery_delivers_client_projection_instead_of_upstream_status() { + let detail = json!({"message":"Previous response was not found. Retrying the full request.","code":"previous_response_not_found","type":"invalid_request_error"}); + let provider = ProviderError::new(ProviderErrorKind::QuotaExhausted, UpstreamSendState::Sent) + .with_status(429) + .with_retry_after(Duration::from_secs(129_600)) + .with_client_visible_upstream_error(ClientVisibleUpstreamError::new( + detail["message"].as_str().unwrap(), + Some("previous_response_not_found".to_owned()), + Some("invalid_request_error".to_owned()), + )) + .with_client_visible_upstream_response( + ClientVisibleUpstreamResponse::new( + 400, + Some(b"application/json".to_vec()), + Bytes::from(json!({"error":detail}).to_string()), + ) + .with_headers(vec![header("x-request-id", b"req-quota-rejected")]), + ); + let error = initial_error(EngineError::Provider(provider), Vec::new()).await; + assert_eq!(error["status"], 400); + assert_eq!(error["error"], detail); + assert_eq!(error["headers"]["x-request-id"], "req-quota-rejected"); + assert!(error["headers"].get("retry-after").is_none()); +} diff --git a/backend/crates/gateway-core/src/error.rs b/backend/crates/gateway-core/src/error.rs index dcaa9b197..7c8eb8876 100644 --- a/backend/crates/gateway-core/src/error.rs +++ b/backend/crates/gateway-core/src/error.rs @@ -338,7 +338,10 @@ impl ProviderConnectionObservation { } } -/// 只供当前客户端请求使用的原始上游 HTTP 失败响应。 +/// 只供当前客户端请求使用的 HTTP 失败响应。 +/// +/// 通常保留原始上游响应；Provider 也可为明确的恢复流程准备客户端投影。 +/// 投影不能覆盖 [`ProviderError`] 中用于诊断和记账的真实上游事实。 /// /// 该值不属于稳定诊断事实，不能进入日志或持久化。它刻意不实现 [`Clone`]； /// [`ProviderError`] 的普通 clone 会丢弃它，只有最终失败的原对象才能把响应交给 @@ -352,7 +355,7 @@ pub struct ClientVisibleUpstreamResponse { } impl ClientVisibleUpstreamResponse { - /// 保存 transport 实际收到的状态码、Content-Type 原值和正文。 + /// 保存客户端响应的状态码、Content-Type 和正文。 #[must_use] pub fn new(status: u16, content_type: Option>, body: Bytes) -> Self { Self { diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 2e789d0d5..27038e481 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -1719,61 +1719,72 @@ fn native_continuation_uses_provider_defined_owner_then_cross_account_recovery() #[test] fn native_continuation_client_replay_required_is_terminal() { - let Operation::Generate(generate) = generate_operation() else { - panic!("generate operation"); - }; - let operation = Operation::Generate(generate.with_provider_session_state( - ProviderSessionState::new("openai", Map::new()).expect("provider session state"), - )); - let route_plan = plan(&operation); - let (coordinator, store, provider) = coordinator(vec![ - Script::Stream { - account_id: "acct_one", - items: vec![Err(ProviderError::new( - ProviderErrorKind::ContinuationRecoveryRequired, - UpstreamSendState::NotSent, - ) - .with_continuation_failure(ContinuationFailure::HistoryUnavailable) - .with_continuation_recovery_disposition( - ContinuationRecoveryDisposition::ClientReplayRequired, - ))], - }, - Script::Stream { - account_id: "acct_one", - items: complete_stream(None), - }, - ]); - let continuation = NativeContinuationPin::new( - PreviousResponseId::new("previous-secret-id"), - PreviousResponseId::new("provider-native-id"), - ClientApiKeyId::new("key_client_1").expect("client key"), - ProviderKind::new("openai").expect("provider"), - ProviderAccountId::new("acct_one").expect("account"), - ); + for (kind, send_state) in [ + ( + ProviderErrorKind::ContinuationRecoveryRequired, + UpstreamSendState::NotSent, + ), + ( + ProviderErrorKind::QuotaExhausted, + UpstreamSendState::NotSent, + ), + (ProviderErrorKind::QuotaExhausted, UpstreamSendState::Sent), + ] { + let Operation::Generate(generate) = generate_operation() else { + panic!("generate operation"); + }; + let operation = Operation::Generate(generate.with_provider_session_state( + ProviderSessionState::new("openai", Map::new()).expect("provider session state"), + )); + let route_plan = plan(&operation); + let (coordinator, store, provider) = coordinator(vec![ + Script::Stream { + account_id: "acct_one", + items: vec![Err(ProviderError::new(kind, send_state) + .with_replay_safe() + .with_continuation_failure(ContinuationFailure::HistoryUnavailable) + .with_continuation_recovery_disposition( + ContinuationRecoveryDisposition::ClientReplayRequired, + ))], + }, + Script::Stream { + account_id: "acct_one", + items: complete_stream(None), + }, + ]); + let continuation = NativeContinuationPin::new( + PreviousResponseId::new("previous-secret-id"), + PreviousResponseId::new("provider-native-id"), + ClientApiKeyId::new("key_client_1").expect("client key"), + ProviderKind::new("openai").expect("provider"), + ProviderAccountId::new("acct_one").expect("account"), + ); - let mut session = block_on(coordinator.start( - model_request(&operation, SystemTime::now() + Duration::from_secs(30)), - operation, - route_plan, - None, - Some(ContinuationBinding::Pinned(continuation)), - CancellationToken::new(), - )) - .expect("start execution"); - let error = block_on(session.collect_uncommitted()) - .expect_err("client replay requirement must stop proxy recovery"); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + Some(ContinuationBinding::Pinned(continuation)), + CancellationToken::new(), + )) + .expect("start execution"); + let error = block_on(session.collect_uncommitted()) + .expect_err("client replay requirement must stop proxy recovery"); - assert!(matches!(error, EngineError::Provider(_))); - let contexts = provider.contexts.lock().expect("contexts lock"); - assert_eq!(contexts.len(), 1); - assert_eq!( - contexts[0].continuation_attempt(), - ContinuationAttempt::Native - ); - let state = store.state.lock().expect("store lock"); - assert_eq!(state.intermediate_failures, 0); - assert_eq!(state.finalizations[0].attempt_count, 1); - assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed); + assert!(matches!(error, EngineError::Provider(_))); + let contexts = provider.contexts.lock().expect("contexts lock"); + assert_eq!(contexts.len(), 1); + assert_eq!( + contexts[0].continuation_attempt(), + ContinuationAttempt::Native + ); + let state = store.state.lock().expect("store lock"); + assert_eq!(state.intermediate_failures, 0); + assert_eq!(state.finalizations[0].attempt_count, 1); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed); + assert_eq!(state.finalizations[0].send_state, send_state); + } } #[test] diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index 977fcfc07..dad32518f 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/backend/crates/providers/openai/src/provider/execution.rs @@ -661,7 +661,11 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { } apply_failure(&failure_context, &active_account, &failure) .await; - Err(failure.error)?; + Err(quota_continuation_replay_error( + failure.error, + &request, + ReplayBoundary::BeforeSemanticOutput, + ))?; return; } }; @@ -847,7 +851,11 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { } apply_failure(&failure_context, &active_account, &failure) .await; - Err(failure.error)?; + Err(quota_continuation_replay_error( + failure.error, + &request, + ReplayBoundary::from_semantic_output(pre_commit_events.is_committed()), + ))?; return; } }; @@ -974,7 +982,11 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { .error .with_atomic_client_events(pre_commit_events.take_for_failure(events)); } - Err(failure.error)?; + Err(quota_continuation_replay_error( + failure.error, + &request, + ReplayBoundary::from_semantic_output(failure_after_commit), + ))?; return; } let events = pre_commit_events.stage(events, timing_signals, completed); @@ -1101,7 +1113,11 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { .error .with_atomic_client_events(pre_commit_events.take_for_failure(events)); } - Err(failure.error)?; + Err(quota_continuation_replay_error( + failure.error, + &request, + ReplayBoundary::from_semantic_output(failure_after_commit), + ))?; return; } let events = pre_commit_events.finish(events, timing_signals, completed); diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index 75de8acab..51662b455 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/backend/crates/providers/openai/src/provider/failure.rs @@ -677,11 +677,80 @@ pub(super) fn continuation_replay_required_error(reason: &'static str) -> Provid .with_diagnostic(ProviderDiagnostic::new( "previous response is unavailable in the selected upstream scope; client replay is required", )) - .with_client_visible_upstream_error(ClientVisibleUpstreamError::new( + .with_client_visible_upstream_error(continuation_replay_error_detail()) +} + +/// 额度拒绝使当前账号无法继续原生历史，由客户端携带完整输入创建新链。 +/// 只投影客户端响应；真实上游分类、发送状态和错误正文继续用于隔离与记账。 +pub(super) fn quota_continuation_replay_error( + mut error: ProviderError, + request: &CodexResponsesRequest, + replay_boundary: ReplayBoundary, +) -> ProviderError { + if request.previous_response_id().is_none() + || error.kind() != ProviderErrorKind::QuotaExhausted + || !error.replay_is_safe() + || error.send_state() == UpstreamSendState::Ambiguous + || !replay_boundary.permits_provider_proof() + { + return error; + } + + let mut headers: Vec = error + .client_visible_upstream_response() + .map(|response| { + response + .headers() + .iter() + // 额度窗口的等待时间不适用于客户端重建历史。 + .filter(|header| !header.name().eq_ignore_ascii_case("retry-after")) + .cloned() + .collect() + }) + .unwrap_or_default(); + // 流内拒绝没有原始 HTTP 失败响应，仍需把该次失败的关联 ID 交给 HTTP 客户端。 + if !headers.iter().any(|header| { + header.name().eq_ignore_ascii_case("x-request-id") + || header.name().eq_ignore_ascii_case("x-oai-request-id") + }) && let Some(request_id) = error.upstream_request_id() + { + headers.push(ProviderResponseHeader::new( + "x-request-id", + Bytes::copy_from_slice(request_id.as_str().as_bytes()), + )); + } + let detail = continuation_replay_error_detail(); + let body = json!({"error": { + "message": detail.message(), + "code": detail.code(), + "type": detail.error_type(), + }}); + // 未交付的 response.created 与额度错误帧必须一起丢弃，否则原错误帧会 + // 覆盖恢复投影。该拒绝发生在语义输出和 commit 之前，不能带走已交付事件。 + drop(error.take_atomic_client_events()); + error + .with_continuation_failure(ContinuationFailure::HistoryUnavailable) + .with_continuation_recovery_disposition( + ContinuationRecoveryDisposition::ClientReplayRequired, + ) + .with_continuation_unavailable_reason("quota_exhausted") + .with_client_visible_upstream_error(detail) + .with_client_visible_upstream_response( + ClientVisibleUpstreamResponse::new( + reqwest::StatusCode::BAD_REQUEST.as_u16(), + Some(b"application/json".to_vec()), + Bytes::from(body.to_string()), + ) + .with_headers(headers), + ) +} + +fn continuation_replay_error_detail() -> ClientVisibleUpstreamError { + ClientVisibleUpstreamError::new( PREVIOUS_RESPONSE_NOT_FOUND_MESSAGE, Some(PREVIOUS_RESPONSE_NOT_FOUND_CODE.to_owned()), Some("invalid_request_error".to_owned()), - )) + ) } pub(super) fn map_stream_error(error: CodexClientError) -> MappedProviderFailure { diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index 256ce7e54..a0246886e 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -8512,3 +8512,425 @@ async fn api_key_websocket_uses_api_path_and_bearer_without_oauth_identity() { .unwrap() .unwrap(); } + +fn quota_continuation_operation(use_websocket: bool) -> Operation { + Operation::Generate( + GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", json!({ + "model": "gpt-5.4", "input": "continue", "previous_response_id": "resp_previous", + "session_id": "quota-replay", "thread_id": "turn", + }).as_object().unwrap().clone()) + .unwrap() + .with_context(Map::from_iter([("use_websocket".to_owned(), json!(use_websocket))])), + ) + .with_provider_session_state( + generate_with_persisted_session_context("acct_provider_contract", "conversation-quota", "quota-replay", "turn") + .provider_session_state("openai").unwrap().clone(), + ), + ) +} + +#[tokio::test] +async fn quota_continuation_opening_rejection_projects_replay_without_losing_upstream_facts() { + for use_websocket in [false, true] { + for (status, code, projects_replay) in [ + (429, "usage_limit_reached", true), + (402, "insufficient_quota", true), + (429, "quota_exceeded", false), + (429, "rate_limit_exceeded", false), + (429, "slow_down", false), + (429, "unknown_error", false), + ] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + let original = + json!({"error": {"type": code, "code": code, "message": "upstream rejection"}}); + Mock::given(method(if use_websocket { "GET" } else { "POST" })) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(status) + .insert_header("retry-after", "129600") + .insert_header("x-request-id", "req-upstream-quota") + .set_body_json(&original), + ) + .expect(1) + .mount(&server) + .await; + let mut stream = provider_with_base_url_and_retry_budget(&store, server.uri(), 0) + .execute( + planned_request("openai", quota_continuation_operation(use_websocket)), + context("req_quota_replay", CancellationToken::new()) + .with_continuation_attempt(ContinuationAttempt::Native), + ) + .await + .expect("prepare continuation"); + let mut error = loop { + match stream.next().await { + Some(Ok(event)) => assert!(!event.has_client_event()), + Some(Err(error)) => break error, + None => panic!("expected rejection"), + } + }; + assert_eq!(error.upstream_status(), Some(status)); + assert_eq!(error.retry_after(), Some(Duration::from_secs(129_600))); + assert_eq!( + serde_json::from_str::(error.raw_upstream_error().unwrap().as_str()) + .unwrap(), + original + ); + assert!(error.take_atomic_client_events().is_empty()); + let response = error.client_visible_upstream_response().unwrap(); + let client: Value = serde_json::from_slice(response.body()).unwrap(); + if projects_replay { + assert_eq!(error.upstream_code().map(|code| code.as_str()), Some(code)); + assert_eq!(error.kind(), ProviderErrorKind::QuotaExhausted); + assert_eq!( + error.continuation_recovery_disposition(), + Some(ContinuationRecoveryDisposition::ClientReplayRequired) + ); + assert_eq!(response.status(), 400); + assert_eq!(client["error"]["code"], "previous_response_not_found"); + assert!( + !response + .headers() + .iter() + .any(|header| header.name() == "retry-after") + ); + assert!( + response + .headers() + .iter() + .any(|header| header.name() == "x-request-id" + && header.value().as_ref() == b"req-upstream-quota") + ); + assert_eq!( + store + .account("acct_provider_contract") + .unwrap() + .quota() + .access(), + QuotaAccessState::Exhausted + ); + } else { + assert_ne!( + error.continuation_recovery_disposition(), + Some(ContinuationRecoveryDisposition::ClientReplayRequired) + ); + assert_eq!(response.status(), 429); + assert_eq!(client, original); + } + } + } +} + +#[tokio::test] +async fn quota_continuation_stream_rejection_only_projects_before_delivery() { + for use_websocket in [false, true] { + for semantic_output in [false, true] { + for native_continuation in [false, true] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let mut events = vec![ + json!({"type":"response.created","response":{"id":"resp_quota","model":"gpt-5.4"}}), + ]; + if semantic_output { + events.push(json!({"type":"response.output_text.delta","output_index":0,"content_index":0,"delta":"hello"})); + } + let original = json!({"type":"response.failed","response":{"id":"resp_quota","error":{"code":"usage_limit_reached","message":"You have reached your usage limit."}}}); + events.push(original.clone()); + let (base_url, _http_server, websocket_server) = if use_websocket { + 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 (socket, _) = listener.accept().await.unwrap(); + let mut ws = accept_codex_test_websocket(socket).await; + ws.next().await.unwrap().unwrap(); + for event in events { + ws.send(Message::Text(event.to_string().into())) + .await + .unwrap(); + } + }); + (base_url, None, Some(server)) + } else { + let server = MockServer::start().await; + let frames = events + .iter() + .map(|event| { + format!( + "event: {}\ndata: {event}\n\n", + event["type"].as_str().unwrap() + ) + }) + .collect::(); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("x-request-id", "req-quota-stream") + .set_body_raw(frames, "text/event-stream"), + ) + .expect(1) + .mount(&server) + .await; + (server.uri(), Some(server), None) + }; + let operation = if native_continuation { + quota_continuation_operation(use_websocket) + } else if use_websocket { + generate_operation() + } else { + http_generate_operation() + }; + let attempt = context("req_quota_stream", CancellationToken::new()) + .with_continuation_attempt(if native_continuation { + ContinuationAttempt::Native + } else { + ContinuationAttempt::None + }); + let mut stream = provider_with_base_url(&store, base_url) + .execute(planned_request("openai", operation), attempt) + .await + .unwrap(); + let mut client_events = Vec::new(); + let mut error = loop { + match stream.next().await { + Some(Ok(event)) => { + if event.has_client_event() { + client_events.push(event); + } + } + Some(Err(error)) => break error, + None => panic!("expected quota failure"), + } + }; + assert_eq!(error.kind(), ProviderErrorKind::QuotaExhausted); + assert_eq!(error.send_state(), UpstreamSendState::Sent); + assert_eq!(error.replay_is_safe(), !semantic_output); + assert_eq!( + serde_json::from_str::(error.raw_upstream_error().unwrap().as_str()) + .unwrap(), + original + ); + client_events.extend(error.take_atomic_client_events()); + if native_continuation && !semantic_output { + assert!( + client_events.is_empty(), + "raw quota error must not override replay projection" + ); + assert_eq!( + error.continuation_recovery_disposition(), + Some(ContinuationRecoveryDisposition::ClientReplayRequired) + ); + assert_eq!( + error.client_visible_upstream_error().unwrap().code(), + Some("previous_response_not_found") + ); + if !use_websocket { + assert_eq!( + error.upstream_request_id().unwrap().as_str(), + "req-quota-stream" + ); + assert!( + error + .client_visible_upstream_response() + .unwrap() + .headers() + .iter() + .any(|header| header.name() == "x-request-id" + && header.value().as_ref() == b"req-quota-stream") + ); + } + } else { + assert_ne!( + error.continuation_recovery_disposition(), + Some(ContinuationRecoveryDisposition::ClientReplayRequired) + ); + assert_eq!( + client_events.last().unwrap().wire_event().unwrap().data(), + &original + ); + } + if let Some(server) = websocket_server { + server.await.unwrap(); + } + } + } + } +} + +#[tokio::test] +async fn quota_continuation_after_structural_commit_preserves_original_failure() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let original = json!({"type":"response.failed","response":{"id":"resp_quota","error":{"code":"usage_limit_reached","message":"limit reached"}}}); + let (base_url, release, _, server) = paused_chunked_sse_server( + "event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_quota\",\"model\":\"gpt-5.4\"}}\n\n".to_owned(), + format!("event: response.failed\ndata: {original}\n\n"), + ).await; + let mut stream = provider_with_base_url(&store, base_url) + .execute( + planned_request("openai", quota_continuation_operation(false)), + context("req_quota_commit", CancellationToken::new()) + .with_continuation_attempt(ContinuationAttempt::Native), + ) + .await + .unwrap(); + timeout(Duration::from_secs(3), async { + while let Some(event) = stream.next().await { + if event.unwrap().has_client_event() { + return; + } + } + panic!("expected grace commit"); + }) + .await + .expect("bounded grace"); + release.send(()).unwrap(); + let mut delivered_failure = false; + let error = loop { + match stream.next().await { + Some(Ok(event)) => { + delivered_failure |= event + .wire_event() + .is_some_and(|wire| wire.data() == &original); + } + Some(Err(error)) => break error, + None => panic!("expected failure"), + } + }; + assert!(delivered_failure); + assert!(!error.replay_is_safe()); + assert_ne!( + error.continuation_recovery_disposition(), + Some(ContinuationRecoveryDisposition::ClientReplayRequired) + ); + server.await.unwrap(); +} + +#[tokio::test] +async fn quota_continuation_full_client_replay_selects_another_account() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let full_input = json!([ + {"role":"user","content":"hello"}, + {"role":"assistant","content":"first answer"}, + {"role":"user","content":"continue"}, + ]); + let expected_input = full_input.clone(); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = accept_codex_test_websocket(socket).await; + ws.next().await.unwrap().unwrap(); + for event in [ + json!({"type":"response.created","response":{"id":"resp_previous","model":"gpt-5.4"}}), + json!({"type":"response.completed","response":{"id":"resp_previous","model":"gpt-5.4","status":"completed","output":[]}}), + ] { + ws.send(Message::Text(event.to_string().into())) + .await + .unwrap(); + } + let delta: Value = + serde_json::from_str(ws.next().await.unwrap().unwrap().to_text().unwrap()).unwrap(); + assert_eq!(delta["previous_response_id"], "resp_previous"); + assert_eq!(delta["input"], "continue"); + ws.send(Message::Text(json!({"type":"error","status":429,"error":{"type":"usage_limit_reached","code":"usage_limit_reached","message":"You have reached your usage limit."}}).to_string().into())).await.unwrap(); + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = accept_codex_test_websocket(socket).await; + let replay: Value = + serde_json::from_str(ws.next().await.unwrap().unwrap().to_text().unwrap()).unwrap(); + assert!(replay.get("previous_response_id").is_none()); + assert_eq!(replay["input"], expected_input); + for event in [ + json!({"type":"response.created","response":{"id":"resp_replayed","model":"gpt-5.4"}}), + json!({"type":"response.completed","response":{"id":"resp_replayed","model":"gpt-5.4","status":"completed","output":[]}}), + ] { + ws.send(Message::Text(event.to_string().into())) + .await + .unwrap(); + } + }); + let provider = provider_with_base_url(&store, base_url); + let first = Operation::Generate(generate_with_persisted_session_context( + "acct_provider_contract", + "conversation-quota", + "quota-replay", + "turn", + )); + let mut stream = provider + .execute( + planned_request("openai", first), + context("req_quota_first", CancellationToken::new()), + ) + .await + .unwrap(); + assert_eq!( + stream.metadata().provider_account_id().as_str(), + "acct_provider_contract" + ); + while let Some(event) = stream.next().await { + event.unwrap(); + } + drop(stream); + let mut stream = provider + .execute( + planned_request("openai", quota_continuation_operation(true)), + context("req_quota_delta", CancellationToken::new()) + .with_continuation_attempt(ContinuationAttempt::Native), + ) + .await + .unwrap(); + let mut error = loop { + match stream.next().await { + Some(Ok(event)) => assert!(!event.has_client_event()), + Some(Err(error)) => break error, + None => panic!("expected quota rejection"), + } + }; + drop(stream); + assert_eq!( + error.client_visible_upstream_error().unwrap().code(), + Some("previous_response_not_found") + ); + assert!(error.take_atomic_client_events().is_empty()); + assert_eq!( + store + .account("acct_provider_contract") + .unwrap() + .quota() + .access(), + QuotaAccessState::Exhausted + ); + create_account(&store, "acct_affinity_switch_b").await; + // 客户端重建完整历史，保留同一会话标识；选号必须跳过刚刚耗尽的原账号。 + let replay = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object("openai", json!({"model":"gpt-5.4","session_id":"quota-replay","thread_id":"turn","input":full_input}).as_object().unwrap().clone()).unwrap(), + )); + let mut stream = provider + .execute( + planned_request("openai", replay), + context("req_quota_full", CancellationToken::new()), + ) + .await + .unwrap(); + assert_eq!( + stream.metadata().provider_account_id().as_str(), + "acct_affinity_switch_b" + ); + let mut completed = false; + while let Some(event) = stream.next().await { + completed |= event + .unwrap() + .canonical_facts() + .iter() + .any(|fact| matches!(fact, GatewayEvent::Completed(_))); + } + assert!(completed); + timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(); +} diff --git a/docs/api.md b/docs/api.md index 38a704326..b72bb3948 100644 --- a/docs/api.md +++ b/docs/api.md @@ -215,7 +215,8 @@ OAuth 账号在客户端使用 HTTP/SSE 时仍可能选择上游 WebSocket。API WebSocket 在尚未交付上游业务事件时合成的错误保留已确认的失败状态，以及 Provider 提取的结构化 message/type/code；没有结构化错误时使用稳定安全文案，不把原始 HTML 或截断正文当作 message。 合成错误自身的 `headers` 携带允许下发的响应头：优先保留实际失败的上游 request ID，无上游 ID 时 -提供网关关联 ID，并用 `x-gateway-request-id` 独立标识网关请求。已经取得的原始上游错误帧不重写。 +提供网关关联 ID，并用 `x-gateway-request-id` 独立标识网关请求。除下述原生续写额度恢复外， +已经取得的原始上游错误帧不重写。 客户端可能对特定状态另行统一展示；这不构成网关改写真实状态码的理由。 `GET /v1/models` 默认返回 OpenAI 兼容列表 `{"object": "list", "data": [...]}`；请求携带非空 @@ -249,8 +250,9 @@ Codex 专用目录中的 `context_window` 与 `max_context_window` 分别表示 OpenAI 路径保留客户端 Responses wire 语义：请求 body 的未知字段和字段顺序保持不变（受控模型 映射除外），HTTP SSE 与 WebSocket 的上游业务事件字节原样转发，response ID 按 opaque 值处理而不 -假设 UUID 或固定长度；OpenAI 上游错误 envelope 和允许下发的 opaque header 值也不由 canonical -观测结果重写。Images 请求不读取或重建 JSON，也不要求或映射模型字段；它固定使用 OpenAI Provider， +假设 UUID 或固定长度；除下述原生续写额度恢复外，OpenAI 上游错误 envelope 和允许下发的 opaque +header 值也不由 canonical 观测结果重写。Images 请求不读取或重建 JSON，也不要求或映射模型字段； +它固定使用 OpenAI Provider， 只在原始字节之外完成账号选择、鉴权头替换和端点路由，成功与失败响应正文同样保持原始字节。 `/v1/alpha/search` 使用相同的 OpenAI Provider 原生端点边界：body（包括 `model`）不解析、不映射， `x-codex-turn-metadata` 在移除客户端账号身份并按当前 lease 重写 installation ID 后转发；上游账号 @@ -270,6 +272,13 @@ OpenAI 明确返回 `server_is_overloaded`、`slow_down` 或模型容量不足 明确额度耗尽触发账号隔离与安全换号， 包括 WebSocket 握手返回的 429；不会因其长 `Retry-After` 而转入同账号传输恢复等待。 +带 `previous_response_id` 的 OpenAI 原生续写仍绑定原账号。若该账号明确拒绝请求且额度已耗尽， +并且请求可安全重放、尚无语义输出且未提交下游，网关隔离该账号，对客户端返回 HTTP `400` +（WebSocket 为 `status: 400`）及 `previous_response_not_found`，不附带额度窗口的 `Retry-After`。 +支持该恢复协议的客户端应去掉 `previous_response_id`、携带完整历史重试，由正常调度选择可用账号； +官方 Codex 的 WebSocket 客户端支持这一流程。其他客户端需要自行处理，网关不会跨账号发送原增量输入。 +普通限流、容量不足、发送结果不明以及已经交付输出的失败不触发此转换。 + ## 4. 浏览器认证 ### 统一登录与会话 diff --git a/docs/architecture.md b/docs/architecture.md index 03500aaf7..45a356a86 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -202,7 +202,8 @@ OAuth 与 API Key 共用业务请求、响应和能力透传链路，差异限 OpenAI 模型目录用于发现，不因目录缺项拒绝请求；管理员配置的模型权限仍由 Core 与选号链路执行。 - OpenAI 是透明边界。Responses 请求保留未知字段和字段顺序；SSE、WebSocket、Images 与 standalone - Search 的业务正文按原始字节转发。canonical facts 从同一数据旁路提取，只用于路由、观测和计费。 + Search 的业务正文按原始字节转发，原生续写额度恢复遵循下述 continuation 例外。 + canonical facts 从同一数据旁路提取，用于路由、恢复判断、观测和计费。 - Responses 的业务扩展头保留原始多值字节。API 负责剥离鉴权、账号身份和 HTTP 传输字段， 并提取会话语义；`gateway-protocol` 共享 HTTP 传输与网关链路字段分类。客户端兼容规则集中在 `providers/openai/src/transport/downstream/`：`headers.rs` 管理下游环境头和已提取语义的头部别名， @@ -281,7 +282,8 @@ client，OIDC 的 JWKS 缓存与单飞归属对应出口状态。自动刷新提 错误信息按用途分成三层，不能用同一个 `message` 同时承担协议、界面和诊断职责： 1. **数据面协议错误**：`/v1/*` 继续遵守 OpenAI/xAI wire 合同。可交付原始上游响应时保留其状态、headers、 - content type 和 body；本地 fallback 使用数据面稳定机器码与安全英文，不受管理端中文化影响。 + content type 和 body；原生续写额度恢复由 Provider 单独投影客户端响应，真实上游事实仍用于诊断和记账。 + 本地 fallback 使用数据面稳定机器码与安全英文，不受管理端中文化影响。 2. **控制面展示错误**：`/api/admin/*` 由 API 统一 HTTP 状态与数值业务码，由 Admin/API owner 提供安全中文 文案。extractor rejection、namespace 404 和 method 405 也使用同一 JSON 信封；任意 Store、Serde、 Provider `Display` 不得直接跨越 HTTP 边界。 @@ -340,6 +342,9 @@ Client Key 与账号分组形成授权范围： Continuation 仍受原请求的 Client Key、账号范围、Provider 和发送/交付边界约束： - native continuation 固定创建它的 Provider 与账号； +- OpenAI 在交付前收到可安全重放的明确额度拒绝时，先隔离账号，再投影 `ClientReplayRequired`； + 丢弃未交付的原错误帧，由客户端提交完整历史开启新链，不把原增量输入交给其他账号。 + 真实错误分类、状态码、发送状态和上游诊断保持不变，客户端合同见 [Responses API](api.md#3-openai-数据面与模型目录)； - OpenAI 按 native → replay owner → replay any 推进，并保留官方 `previous_response_id` 语义； - xAI 使用客户端提交的完整历史作为重放输入； - scope 外账号、跨 Key 复用或不明确发送结果均 fail closed。