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 <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>

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

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

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<Vec<u8>>, body: Bytes) -> Self {
         Self {

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]

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

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<ProviderResponseHeader> = 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 {

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::<Value>(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::<String>();
+                    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::<Value>(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();
+}

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. 浏览器认证
 
 ### 统一登录与会话

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。