From f218e31c72eb2061a021104dc801e737c6a9441d Mon Sep 17 00:00:00 2001 From: zyycn Date: Tue, 14 Jul 2026 23:15:05 +0800 Subject: [PATCH] refactor(dispatch): harden account failover boundaries Invalidate unavailable accounts in memory before asynchronous persistence, evict account WebSockets, and classify explicit auth, quota, ban, and workspace failures for immediate rotation. Separate account-scoped installation identity from client session semantics, preserve logical IDs across failover, clear opaque account state, remove forced HTTP/1.1 configuration, and document the lifecycle rules. Co-authored-by: Codex --- backend/src/api/client/responses/mod.rs | 20 +- backend/src/bootstrap/config.rs | 13 - backend/src/bootstrap/services.rs | 2 +- backend/src/dispatch/affinity/identity.rs | 296 +++++++++++----- backend/src/dispatch/affinity/mod.rs | 3 +- .../dispatch/controllers/account_failure.rs | 250 ++++++++------ .../src/dispatch/controllers/account_state.rs | 38 +++ backend/src/dispatch/controllers/mod.rs | 99 ++++-- backend/src/dispatch/controllers/quota.rs | 95 +++--- backend/src/dispatch/lifecycle/attempt.rs | 25 +- backend/src/dispatch/service.rs | 1 + backend/src/dispatch/stream/lifecycle.rs | 1 + backend/src/dispatch/transport/account.rs | 92 ++--- backend/src/dispatch/transport/observation.rs | 24 +- backend/src/fleet/manage/import.rs | 1 - backend/src/fleet/manage/probe.rs | 1 - backend/src/fleet/manage/quota.rs | 1 - backend/src/fleet/pool/mod.rs | 45 +++ backend/src/fleet/pool/persistence.rs | 94 +++++ backend/src/fleet/quota/runtime.rs | 1 - .../src/upstream/openai/protocol/events.rs | 9 +- .../src/upstream/openai/protocol/responses.rs | 4 +- .../src/upstream/openai/transport/client.rs | 119 +------ .../upstream/openai/transport/client_sse.rs | 4 +- backend/tests/bootstrap/config/mod.rs | 13 + backend/tests/dispatch/lifecycle/mod.rs | 3 +- .../tests/dispatch/service/responses_http.rs | 13 +- .../dispatch/service/responses_recovery.rs | 323 +++++++++++++++++- .../dispatch/service/responses_websocket.rs | 11 +- backend/tests/fleet/pool/runtime.rs | 233 ++++++++++++- backend/tests/support/config.rs | 5 +- .../upstream/openai/transport/headers.rs | 19 +- .../upstream/openai/transport/http_client.rs | 3 - .../tests/upstream/openai/transport/mod.rs | 1 - .../upstream/openai/transport/websocket.rs | 1 - deploy/config.example.yaml | 3 - docs/architecture.md | 23 +- docs/request-lifecycle-controller-design.md | 230 ++++++++++++- 38 files changed, 1571 insertions(+), 548 deletions(-) create mode 100644 backend/src/dispatch/controllers/account_state.rs create mode 100644 backend/src/fleet/pool/persistence.rs diff --git a/backend/src/api/client/responses/mod.rs b/backend/src/api/client/responses/mod.rs index 56a413bed..11c696658 100644 --- a/backend/src/api/client/responses/mod.rs +++ b/backend/src/api/client/responses/mod.rs @@ -103,36 +103,36 @@ pub fn build_codex_request( None => {} } apply_context_header_fields(&mut request, headers); - apply_identity_fields(&mut request, headers); + apply_session_fields(&mut request, headers); request.client_user_agent = user_agent_from_headers(headers); request } -fn apply_identity_fields(request: &mut CodexResponsesRequest, headers: &HeaderMap) { - request.client_conversation_id = identity_context_string(request.body(), "conversation_id") +fn apply_session_fields(request: &mut CodexResponsesRequest, headers: &HeaderMap) { + request.client_conversation_id = session_context_string(request.body(), "conversation_id") .or_else(|| header_string(headers, "conversation-id")) .or_else(|| header_string(headers, "conversation_id")) .or_else(|| request.prompt_cache_key().map(ToString::to_string)); - request.client_session_id = identity_context_string(request.body(), "session_id") + request.client_session_id = session_context_string(request.body(), "session_id") .or_else(|| header_string(headers, "session-id")) .or_else(|| header_string(headers, "session_id")); - request.client_thread_id = identity_context_string(request.body(), "thread_id") + request.client_thread_id = session_context_string(request.body(), "thread_id") .or_else(|| header_string(headers, "thread-id")); - request.client_request_id = identity_context_string(request.body(), "x-client-request-id") + request.client_request_id = session_context_string(request.body(), "x-client-request-id") .or_else(|| header_string(headers, "x-client-request-id")); - request.client_turn_id = identity_context_string(request.body(), "turn_id") + request.client_turn_id = session_context_string(request.body(), "turn_id") .or_else(|| header_string(headers, "x-codex-turn-id")); request.codex_window_id = request.codex_window_id.take().or_else(|| { - identity_context_string(request.body(), "x-codex-window-id") + session_context_string(request.body(), "x-codex-window-id") .or_else(|| header_string(headers, "x-codex-window-id")) }); request.parent_thread_id = request.parent_thread_id.take().or_else(|| { - identity_context_string(request.body(), "x-codex-parent-thread-id") + session_context_string(request.body(), "x-codex-parent-thread-id") .or_else(|| header_string(headers, "x-codex-parent-thread-id")) }); } -fn identity_context_string(body: &Map, key: &str) -> Option { +fn session_context_string(body: &Map, key: &str) -> Option { body_context_string(body, key).or_else(|| { body.get("client_metadata") .and_then(Value::as_object) diff --git a/backend/src/bootstrap/config.rs b/backend/src/bootstrap/config.rs index d36844269..395b9b103 100644 --- a/backend/src/bootstrap/config.rs +++ b/backend/src/bootstrap/config.rs @@ -105,7 +105,6 @@ impl BootstrapConfig { redis, runtime, quota, - tls, ws_pool, fingerprint, admin, @@ -122,7 +121,6 @@ impl BootstrapConfig { database: DatabaseConfig { url: database.url }, redis: RedisConfig { url: redis.url }, runtime, - tls, ws_pool, fingerprint, admin: AdminConfig { @@ -241,8 +239,6 @@ pub struct AppConfig { pub redis: RedisConfig, /// 本地持久运行目录。 pub runtime: RuntimePathsConfig, - /// TLS 偏好配置。 - pub tls: TlsConfig, /// WebSocket 连接池启动设置。 pub ws_pool: WebSocketPoolSettings, /// 上游请求指纹默认配置。 @@ -344,14 +340,6 @@ pub struct RuntimePathsConfig { pub data_directory: PathBuf, } -/// TLS/HTTP 协议偏好配置。 -#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] -#[serde(deny_unknown_fields)] -pub struct TlsConfig { - /// 是否强制 HTTP/1.1。 - pub force_http11: bool, -} - /// WebSocket 连接池启动设置。 #[derive(Debug, Clone, Deserialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] @@ -464,7 +452,6 @@ struct FileAppConfig { redis: FileRedisConfig, runtime: RuntimePathsConfig, quota: QuotaConfig, - tls: TlsConfig, ws_pool: WebSocketPoolSettings, fingerprint: FingerprintConfig, admin: FileAdminConfig, diff --git a/backend/src/bootstrap/services.rs b/backend/src/bootstrap/services.rs index d8e357ce1..713482fb1 100644 --- a/backend/src/bootstrap/services.rs +++ b/backend/src/bootstrap/services.rs @@ -212,7 +212,7 @@ impl Services { let websocket_initial_event_timeout = (config.ws_pool.initial_event_timeout_ms > 0) .then(|| std::time::Duration::from_millis(config.ws_pool.initial_event_timeout_ms)); let client = CodexBackendClient::new( - build_reqwest_client(config.tls.force_http11)?, + build_reqwest_client()?, config.api.base_url.clone(), fingerprint.clone(), ) diff --git a/backend/src/dispatch/affinity/identity.rs b/backend/src/dispatch/affinity/identity.rs index 21906cadd..da2e59fe7 100644 --- a/backend/src/dispatch/affinity/identity.rs +++ b/backend/src/dispatch/affinity/identity.rs @@ -1,10 +1,15 @@ //! 客户端身份、本地会话索引与账号作用域 wire 身份隔离。 +use serde_json::{Map, Value}; + use crate::{ infra::identity::AccountPseudonymizer, upstream::openai::protocol::responses::CodexResponsesRequest, }; +const TURN_STATE_METADATA_KEY: &str = "x-codex-turn-state"; +const TURN_METADATA_KEY: &str = "x-codex-turn-metadata"; + /// 持久化密钥驱动的账号身份隔离服务。 #[derive(Clone)] pub struct AccountIdentityService { @@ -21,25 +26,51 @@ impl std::fmt::Debug for AccountIdentityService { /// 单次账号 attempt 的 wire 身份。 #[derive(Debug, Clone, Default, PartialEq, Eq)] -pub struct AccountScopedIdentity { - pub prompt_cache_key: Option, - pub session_id: Option, - pub thread_id: Option, - pub client_request_id: String, - pub turn_id: Option, - pub window_id: Option, - pub parent_thread_id: Option, +pub(in crate::dispatch) struct AccountScopedIdentity { pub installation_id: String, } -struct IdentityValues<'a> { - prompt_cache_key: Option<&'a str>, - session_id: Option<&'a str>, - thread_id: Option<&'a str>, - client_request_id: Option<&'a str>, - turn_id: Option<&'a str>, - window_id: Option<&'a str>, - parent_thread_id: Option<&'a str>, +/// 单个请求内账号绑定 opaque 状态的 owner。 +pub(in crate::dispatch) struct AccountIdentityScope { + turn_state_account_id: Option, +} + +impl AccountIdentityScope { + pub(in crate::dispatch) fn new(preferred_account_id: Option) -> Self { + Self { + turn_state_account_id: preferred_account_id, + } + } + + fn preserves_turn_state(&mut self, account_id: &str, has_turn_state: bool) -> bool { + if !has_turn_state { + return false; + } + let owner = self + .turn_state_account_id + .get_or_insert_with(|| account_id.to_string()); + owner == account_id + } +} + +/// 已按当前账号重建 wire 身份、可安全交给上游 transport 的请求。 +pub(in crate::dispatch) struct AccountScopedRequest { + request: CodexResponsesRequest, + identity: AccountScopedIdentity, +} + +impl AccountScopedRequest { + pub(in crate::dispatch) fn request(&self) -> &CodexResponsesRequest { + &self.request + } + + pub(in crate::dispatch) fn identity(&self) -> &AccountScopedIdentity { + &self.identity + } + + pub(in crate::dispatch) fn into_request(self) -> CodexResponsesRequest { + self.request + } } impl AccountIdentityService { @@ -63,88 +94,183 @@ impl AccountIdentityService { .map(|anchor| format!("lc_{}", self.pseudonym("local-conversation", None, anchor))); } - pub fn scope( + /// 从请求语义为单次账号 attempt 重建完整 wire 身份。 + pub(in crate::dispatch) fn scope_request( &self, - request: &CodexResponsesRequest, + scope: &mut AccountIdentityScope, + mut request: CodexResponsesRequest, account_id: &str, - proxy_request_id: &str, - ) -> AccountScopedIdentity { - self.scope_values( + ) -> AccountScopedRequest { + let identity = AccountScopedIdentity { + installation_id: self.pseudonymizer.installation_id(account_id), + }; + let client_metadata_turn_state = client_metadata_string(&request, TURN_STATE_METADATA_KEY); + let preserve_turn_state = scope.preserves_turn_state( account_id, - proxy_request_id, - IdentityValues { - prompt_cache_key: request.prompt_cache_key(), - session_id: request.client_session_id.as_deref(), - thread_id: request.client_thread_id.as_deref(), - client_request_id: request.client_request_id.as_deref(), - turn_id: request.client_turn_id.as_deref(), - window_id: request.codex_window_id.as_deref(), - parent_thread_id: request.parent_thread_id.as_deref(), - }, - ) + request.turn_state.is_some() || client_metadata_turn_state.is_some(), + ); + let turn_state = if preserve_turn_state { + request.turn_state.clone() + } else { + None + }; + let client_metadata_turn_state = if preserve_turn_state { + client_metadata_turn_state + } else { + None + }; + let turn_metadata = request + .turn_metadata + .as_deref() + .map(|raw| Self::scope_turn_metadata(raw, &identity)); + let client_metadata_turn_metadata = client_metadata_string(&request, TURN_METADATA_KEY) + .map(|raw| Self::scope_turn_metadata(&raw, &identity)); + + apply_identity_to_request( + &mut request, + &identity, + turn_state.as_deref(), + turn_metadata.as_deref(), + client_metadata_turn_state.as_deref(), + client_metadata_turn_metadata.as_deref(), + ); + request.turn_state = turn_state; + request.turn_metadata = turn_metadata; + + AccountScopedRequest { request, identity } } - pub fn scope_auxiliary( - &self, - account_id: &str, - proxy_request_id: &str, - ) -> AccountScopedIdentity { + pub(in crate::dispatch) fn scope_auxiliary(&self, account_id: &str) -> AccountScopedIdentity { AccountScopedIdentity { - client_request_id: format!( - "cr_{}", - self.pseudonym("client-request-id", Some(account_id), proxy_request_id) - ), - installation_id: self.pseudonymizer.installation_id(account_id), - ..AccountScopedIdentity::default() - } - } - - fn scope_values( - &self, - account_id: &str, - proxy_request_id: &str, - values: IdentityValues<'_>, - ) -> AccountScopedIdentity { - AccountScopedIdentity { - prompt_cache_key: self.scope_optional( - "prompt-cache-key", - account_id, - values.prompt_cache_key, - ), - session_id: self.scope_optional("session-id", account_id, values.session_id), - thread_id: self.scope_optional("thread-id", account_id, values.thread_id), - client_request_id: format!( - "cr_{}", - self.pseudonym( - "client-request-id", - Some(account_id), - values.client_request_id.unwrap_or(proxy_request_id), - ) - ), - turn_id: self.scope_optional("turn-id", account_id, values.turn_id), - window_id: self.scope_optional("window-id", account_id, values.window_id), - parent_thread_id: self.scope_optional( - "parent-thread-id", - account_id, - values.parent_thread_id, - ), installation_id: self.pseudonymizer.installation_id(account_id), } } - fn scope_optional( - &self, - domain: &str, - account_id: &str, - value: Option<&str>, - ) -> Option { - value - .map(str::trim) - .filter(|value| !value.is_empty()) - .map(|value| format!("wi_{}", self.pseudonym(domain, Some(account_id), value))) + fn scope_turn_metadata(raw: &str, identity: &AccountScopedIdentity) -> String { + let Ok(Value::Object(mut metadata)) = serde_json::from_str::(raw) else { + return raw.to_string(); + }; + + let contains_installation_identity = [ + "installation_id", + "installationId", + "x-codex-installation-id", + ] + .into_iter() + .any(|key| metadata.contains_key(key)); + if !contains_installation_identity { + return raw.to_string(); + } + + for key in [ + "installation_id", + "installationId", + "x-codex-installation-id", + ] { + replace_existing_metadata_field(&mut metadata, key, Some(&identity.installation_id)); + } + + serde_json::to_string(&metadata).unwrap_or_else(|_| raw.to_string()) } fn pseudonym(&self, domain: &str, account_id: Option<&str>, value: &str) -> String { self.pseudonymizer.scoped(domain, account_id, value) } } + +fn client_metadata_string(request: &CodexResponsesRequest, key: &str) -> Option { + request + .client_metadata()? + .as_object()? + .get(key)? + .as_str() + .map(ToString::to_string) +} + +fn apply_identity_to_request( + request: &mut CodexResponsesRequest, + identity: &AccountScopedIdentity, + turn_state: Option<&str>, + turn_metadata: Option<&str>, + client_metadata_turn_state: Option<&str>, + client_metadata_turn_metadata: Option<&str>, +) { + for (key, value) in [ + ( + "x-codex-installation-id", + Some(identity.installation_id.as_str()), + ), + ("installation_id", Some(identity.installation_id.as_str())), + ("installationId", Some(identity.installation_id.as_str())), + ("turnState", turn_state), + ("x-codex-turn-state", turn_state), + ("turnMetadata", turn_metadata), + ("x-codex-turn-metadata", turn_metadata), + ] { + request.replace_existing_identity_field(key, value); + } + + let client_metadata = scoped_client_metadata( + request.client_metadata(), + identity, + client_metadata_turn_state, + client_metadata_turn_metadata, + ); + request.set_client_metadata(client_metadata); +} + +fn scoped_client_metadata( + client_metadata: Option<&Value>, + identity: &AccountScopedIdentity, + turn_state: Option<&str>, + turn_metadata: Option<&str>, +) -> Option { + let client_metadata = client_metadata?; + let Value::Object(mut metadata) = client_metadata.clone() else { + return Some(client_metadata.clone()); + }; + + for (key, value) in [ + ( + "x-codex-installation-id", + Some(identity.installation_id.as_str()), + ), + (TURN_STATE_METADATA_KEY, turn_state), + (TURN_METADATA_KEY, turn_metadata), + ] { + replace_metadata_field(&mut metadata, key, value); + } + replace_existing_metadata_field( + &mut metadata, + "installation_id", + Some(&identity.installation_id), + ); + replace_existing_metadata_field( + &mut metadata, + "installationId", + Some(&identity.installation_id), + ); + + (!metadata.is_empty()).then_some(Value::Object(metadata)) +} + +fn replace_existing_metadata_field( + metadata: &mut Map, + key: &str, + value: Option<&str>, +) { + if metadata.contains_key(key) { + replace_metadata_field(metadata, key, value); + } +} + +fn replace_metadata_field(metadata: &mut Map, key: &str, value: Option<&str>) { + match value.filter(|value| !value.trim().is_empty()) { + Some(value) => { + metadata.insert(key.to_string(), Value::String(value.to_string())); + } + None => { + metadata.remove(key); + } + } +} diff --git a/backend/src/dispatch/affinity/mod.rs b/backend/src/dispatch/affinity/mod.rs index 98f13be67..c1127b9b8 100644 --- a/backend/src/dispatch/affinity/mod.rs +++ b/backend/src/dispatch/affinity/mod.rs @@ -6,7 +6,8 @@ pub mod service; pub mod store; pub mod types; -pub use identity::{AccountIdentityService, AccountScopedIdentity}; +pub use identity::AccountIdentityService; +pub(in crate::dispatch) use identity::{AccountIdentityScope, AccountScopedRequest}; pub use resolve::*; pub use service::*; pub use store::*; diff --git a/backend/src/dispatch/controllers/account_failure.rs b/backend/src/dispatch/controllers/account_failure.rs index ff95fb968..598da988b 100644 --- a/backend/src/dispatch/controllers/account_failure.rs +++ b/backend/src/dispatch/controllers/account_failure.rs @@ -2,7 +2,10 @@ use crate::{ dispatch::{ - controllers::ControllerFailureFact, + controllers::{ + ControllerFailureFact, + account_state::{AccountStateEffect, AccountStateEffects}, + }, errors::ClientFailure, failure::exhaustion::{AccountExhaustionRecord, ExhaustedAccountKind}, lifecycle::contract::{ @@ -11,19 +14,14 @@ use crate::{ transport::observation::{UpstreamFailureFacts, UpstreamFailureKind}, }, fleet::{account::AccountStatus, pool::AccountPoolService}, - upstream::openai::protocol::responses::ResponsesSseFailure, + upstream::openai::{protocol::responses::ResponsesSseFailure, transport::CodexBackendClient}, }; pub(super) struct AccountFailureController; pub(super) struct ClassifiedFailure { exhaustion: AccountExhaustionRecord, - effect: AccountEffect, -} - -enum AccountEffect { - None, - SetStatus(AccountStatus), + effect: Option, } impl AccountFailureController { @@ -41,16 +39,14 @@ impl AccountFailureController { pub(super) async fn apply_effect( account_pool: &AccountPoolService, + codex: &CodexBackendClient, classified: &ClassifiedFailure, ) { let Some(account_id) = classified.exhaustion.account_id.as_deref() else { return; }; - match &classified.effect { - AccountEffect::None => {} - AccountEffect::SetStatus(status) => { - account_pool.set_status(account_id, *status).await; - } + if let Some(effect) = &classified.effect { + AccountStateEffects::apply(account_pool, codex, account_id, effect).await; } } @@ -114,6 +110,11 @@ impl AccountFailureController { .as_deref() .unwrap_or_default() .to_ascii_lowercase(); + let error_type = failure + .upstream_type + .as_deref() + .unwrap_or_default() + .to_ascii_lowercase(); let message = failure.message.to_ascii_lowercase(); if is_model_unsupported(&code) || is_model_unsupported(&message) @@ -125,10 +126,11 @@ impl AccountFailureController { { return Some(ClientFailure::new(failure.clone(), 400, false)); } - if is_auth_failure(&code, &message) { + if is_auth_failure(&code, &error_type, &message) { return Some(ClientFailure::new(failure.clone(), 401, false)); } - if code.contains("forbidden") || code.contains("banned") { + if code.contains("forbidden") || code.contains("banned") || error_type == "permission_error" + { return Some(ClientFailure::new(failure.clone(), 403, false)); } if code.contains("server_overloaded") { @@ -147,7 +149,7 @@ fn classify(observation: &AttemptObservation, account_id: &str) -> Option ControllerFailureFact::from_attempt(observation) @@ -168,108 +170,151 @@ fn classify_failure( fn classify_upstream(account_id: &str, facts: &UpstreamFailureFacts) -> Option { let status = facts.status_code; let body = facts.body.as_str(); - let lower = body.to_ascii_lowercase(); - let (kind, effect, message, status_code) = if (status == Some(403) && !is_html(body)) - || (status == Some(402) && is_deactivated_workspace(body)) - { - ( - ExhaustedAccountKind::Banned, - AccountEffect::SetStatus(AccountStatus::Banned), - body.to_string(), + let code = facts.code.as_deref().unwrap_or_default(); + let error_type = facts.error_type.as_deref().unwrap_or_default(); + let message = facts.message.as_str(); + + if is_model_unsupported(code) || is_model_unsupported(message) || is_model_unsupported(body) { + return Some(model_unsupported_failure(account_id, body)); + } + if let Some(account_status) = explicit_account_status(code, error_type, body) { + return Some(account_status_failure( + account_id, + account_status, + body, status, - ) - } else if status == Some(401) { - let account_status = if lower.contains("banned") || lower.contains("deactivated") { - AccountStatus::Banned - } else { - AccountStatus::Expired - }; - ( - if account_status == AccountStatus::Banned { - ExhaustedAccountKind::Banned - } else { - ExhaustedAccountKind::Expired - }, - AccountEffect::SetStatus(account_status), - body.to_string(), + )); + } + if status == Some(403) && !is_html(body) { + return Some(account_status_failure( + account_id, + AccountStatus::Banned, + body, status, - ) - } else if status.is_some_and(|status| { - (400..=499).contains(&status) && !matches!(status, 401 | 402 | 403 | 404 | 429) - }) && is_model_unsupported(body) - { - ( + )); + } + if status == Some(401) { + return Some(account_status_failure( + account_id, + AccountStatus::Expired, + body, + status, + )); + } + None +} + +fn classify_sse(account_id: &str, failure: &ResponsesSseFailure) -> Option { + let code = failure.upstream_code.as_deref().unwrap_or_default(); + let error_type = failure.upstream_type.as_deref().unwrap_or_default(); + if is_model_unsupported(code) || is_model_unsupported(&failure.message) { + return Some(model_unsupported_failure( + account_id, + &crate::dispatch::failure::sse::sse_failure_error_body(failure), + )); + } + let account_status = + explicit_account_status(code, error_type, &failure.message).or_else(|| { + (code.eq_ignore_ascii_case("forbidden") || error_type == "permission_error") + .then_some(AccountStatus::Banned) + })?; + Some(account_status_failure( + account_id, + account_status, + &crate::dispatch::failure::sse::sse_failure_error_body(failure), + failure.explicit_status_code, + )) +} + +fn model_unsupported_failure(account_id: &str, message: &str) -> ClassifiedFailure { + ClassifiedFailure { + exhaustion: AccountExhaustionRecord::new( + account_id, ExhaustedAccountKind::ModelUnsupported, - AccountEffect::None, - body.to_string(), - None, - ) - } else { - return None; + message, + ), + effect: None, + } +} + +fn account_status_failure( + account_id: &str, + status: AccountStatus, + message: &str, + status_code: Option, +) -> ClassifiedFailure { + let kind = match status { + AccountStatus::Expired => ExhaustedAccountKind::Expired, + AccountStatus::Disabled => ExhaustedAccountKind::Disabled, + AccountStatus::Banned => ExhaustedAccountKind::Banned, + AccountStatus::Active | AccountStatus::QuotaExhausted => { + unreachable!("account failure controller only invalidates unavailable statuses") + } }; let mut exhaustion = AccountExhaustionRecord::new(account_id, kind, message); if let Some(status_code) = status_code { exhaustion = exhaustion.with_status_code(status_code); } - Some(ClassifiedFailure { exhaustion, effect }) -} - -fn classify_sse(account_id: &str, failure: &ResponsesSseFailure) -> Option { - let code = failure.upstream_code.as_deref().unwrap_or_default(); - let lower_message = failure.message.to_ascii_lowercase(); - let (kind, effect, status) = if is_model_unsupported(code) - || is_model_unsupported(&failure.message) - { - ( - ExhaustedAccountKind::ModelUnsupported, - AccountEffect::None, - None, - ) - } else if is_auth_failure(code, &lower_message) { - let account_status = if lower_message.contains("banned") || code == "account_deactivated" { - AccountStatus::Banned - } else { - AccountStatus::Expired - }; - ( - if account_status == AccountStatus::Banned { - ExhaustedAccountKind::Banned - } else { - ExhaustedAccountKind::Expired - }, - AccountEffect::SetStatus(account_status), - Some(if account_status == AccountStatus::Banned { - 403 - } else { - 401 - }), - ) - } else { - return None; - }; - let mut exhaustion = AccountExhaustionRecord::new( - account_id, - kind, - crate::dispatch::failure::sse::sse_failure_error_body(failure), - ); - if let Some(status) = status { - exhaustion = exhaustion.with_status_code(status); + ClassifiedFailure { + exhaustion, + effect: Some(AccountStateEffect::SetStatus(status)), } - Some(ClassifiedFailure { exhaustion, effect }) } -fn is_auth_failure(code: &str, message: &str) -> bool { +fn explicit_account_status(code: &str, error_type: &str, message: &str) -> Option { + let code = code.trim().to_ascii_lowercase(); + let error_type = error_type.trim().to_ascii_lowercase(); + let message = message.to_ascii_lowercase(); + if matches!( + code.as_str(), + "identity_verification_required" | "verification_required" + ) || message.contains("identity verification is required") + { + return Some(AccountStatus::Disabled); + } + if matches!( + code.as_str(), + "account_banned" + | "account_deactivated" + | "account_disabled" + | "account_suspended" + | "deactivated_workspace" + | "organization_disabled" + | "workspace_deactivated" + ) || message.contains("account is banned") + || message.contains("account has been banned") + || message.contains("account deactivated") + || message.contains("account has been deactivated") + || message.contains("account disabled") + || message.contains("account has been disabled") + || message.contains("account suspended") + || message.contains("organization has been disabled") + || message.contains("workspace has been deactivated") + || message.contains("deactivated_workspace") + { + return Some(AccountStatus::Banned); + } + is_auth_failure(&code, &error_type, &message).then_some(AccountStatus::Expired) +} + +fn is_auth_failure(code: &str, error_type: &str, message: &str) -> bool { matches!( code, "token_invalid" + | "token_invalidated" | "token_expired" | "token_revoked" - | "account_deactivated" + | "refresh_token_invalidated" | "unauthorized" | "invalid_api_key" - ) || message.contains("token revoked") + | "authentication_error" + ) || error_type == "authentication_error" + || message.contains("token revoked") + || message.contains("token invalidated") || message.contains("token invalid") || message.contains("token expired") + || message.contains("unauthorized") + || message.contains("invalid api key") } fn is_model_unsupported(value: &str) -> bool { @@ -287,12 +332,3 @@ fn is_html(value: &str) -> bool { let value = value.trim_start().to_ascii_lowercase(); value.starts_with(" bool { - serde_json::from_str::(value).is_ok_and(|value| { - value - .pointer("/detail/code") - .and_then(serde_json::Value::as_str) - == Some("deactivated_workspace") - }) -} diff --git a/backend/src/dispatch/controllers/account_state.rs b/backend/src/dispatch/controllers/account_state.rs new file mode 100644 index 000000000..2ad995905 --- /dev/null +++ b/backend/src/dispatch/controllers/account_state.rs @@ -0,0 +1,38 @@ +//! 明确账号事实触发的运行时失效与连接驱逐。 + +use chrono::{DateTime, Utc}; + +use crate::{ + fleet::{account::AccountStatus, pool::AccountPoolService}, + upstream::openai::transport::CodexBackendClient, +}; + +pub(super) enum AccountStateEffect { + SetStatus(AccountStatus), + MarkQuotaLimitedUntil(DateTime), +} + +pub(super) struct AccountStateEffects; + +impl AccountStateEffects { + pub(super) async fn apply( + account_pool: &AccountPoolService, + codex: &CodexBackendClient, + account_id: &str, + effect: &AccountStateEffect, + ) { + match effect { + AccountStateEffect::SetStatus(status) => { + account_pool + .set_status_immediately(account_id, *status) + .await; + } + AccountStateEffect::MarkQuotaLimitedUntil(until) => { + account_pool + .mark_quota_limited_until_immediately(account_id, *until) + .await; + } + } + codex.evict_websocket_account(account_id).await; + } +} diff --git a/backend/src/dispatch/controllers/mod.rs b/backend/src/dispatch/controllers/mod.rs index 7ee8bc322..b4b113484 100644 --- a/backend/src/dispatch/controllers/mod.rs +++ b/backend/src/dispatch/controllers/mod.rs @@ -15,7 +15,10 @@ use tokio::time::timeout; use crate::{ dispatch::{ - affinity::{AccountIdentityService, SessionAffinityService}, + affinity::{ + AccountIdentityScope, AccountIdentityService, AccountScopedRequest, + SessionAffinityService, + }, errors::{ClientFailure, ResponseDispatchError, upstream_error_set_cookie_headers}, lifecycle::trace::{ResponseDispatchAttempt, ResponseDispatchTrace}, }, @@ -53,6 +56,7 @@ use crate::{ }; mod account_failure; +mod account_state; mod affinity; pub(crate) mod cloudflare; pub(crate) mod cyber_policy; @@ -243,6 +247,7 @@ pub(in crate::dispatch) struct ControllerSet { cyber_policy: CyberPolicyController, session_affinity: Arc, account_pool: Arc, + codex: Arc, cloudflare: CloudflareRecovery, recorder: Arc, } @@ -251,6 +256,7 @@ pub(in crate::dispatch) struct ControllerSet { pub(in crate::dispatch) struct ControllerRequestScope { cyber_policy: CyberPolicyScope, history: HistoryState, + identity: AccountIdentityScope, usage: usage::UsageScope, } @@ -277,8 +283,13 @@ pub(in crate::dispatch) enum AttemptAccountPreparation { Rejected, } +pub(in crate::dispatch) struct AttemptRoutePreparationContext<'a> { + pub account_identity: &'a AccountIdentityService, + pub account_id: &'a str, +} + pub(in crate::dispatch) enum AttemptRoutePreparation { - Ready(Box), + Ready(Box), Unavailable { message: String }, } @@ -323,12 +334,21 @@ impl ControllerSet { &self, scope: &mut ControllerRequestScope, request: &CodexResponsesRequest, - account_id: &str, + context: AttemptRoutePreparationContext<'_>, ) -> AttemptRoutePreparation { - match HistoryController::prepare_attempt(&mut scope.history, request, account_id) { - Ok(request) => AttemptRoutePreparation::Ready(Box::new(request)), - Err(message) => AttemptRoutePreparation::Unavailable { message }, - } + let request = match HistoryController::prepare_attempt( + &mut scope.history, + request, + context.account_id, + ) { + Ok(request) => request, + Err(message) => return AttemptRoutePreparation::Unavailable { message }, + }; + AttemptRoutePreparation::Ready(Box::new(context.account_identity.scope_request( + &mut scope.identity, + request, + context.account_id, + ))) } pub(in crate::dispatch) fn prepare_same_account_retry( @@ -473,6 +493,7 @@ impl ControllerSet { pub(in crate::dispatch) fn new( session_affinity: Arc, account_pool: Arc, + codex: Arc, cloudflare: CloudflareRecovery, recorder: Arc, ) -> Self { @@ -480,6 +501,7 @@ impl ControllerSet { cyber_policy: CyberPolicyController::new(Arc::clone(&session_affinity)), session_affinity, account_pool, + codex, cloudflare, recorder, } @@ -508,6 +530,7 @@ impl ControllerSet { scope: ControllerRequestScope { cyber_policy, history, + identity: AccountIdentityScope::new(preferred_account_id.clone()), usage, }, preferred_account_id, @@ -553,33 +576,32 @@ impl ControllerSet { } } - async fn apply_stream_failure( + async fn apply_stream_account_state(&self, classification: &StreamFailureClassification<'_>) { + match classification.owner.as_ref() { + Some(SharedFailureClassification::AccountFailure(classified)) => { + AccountFailureController::apply_effect(&self.account_pool, &self.codex, classified) + .await; + } + Some(SharedFailureClassification::Quota(classified)) => { + QuotaController::apply_effect(&self.account_pool, &self.codex, classified).await; + } + Some(SharedFailureClassification::CyberPolicy(_)) | None => {} + } + } + + async fn apply_stream_policy_failure( &self, scope: &ControllerRequestScope, account_id: &str, classification: &StreamFailureClassification<'_>, ) { - match classification.owner.as_ref() { - Some(SharedFailureClassification::CyberPolicy(_)) => { - self.cyber_policy - .exclude_account(&scope.cyber_policy, account_id) - .await; - } - Some(SharedFailureClassification::AccountFailure(classified)) => { - let _ = best_effort_controller_io( - "account_failure.stream", - AccountFailureController::apply_effect(&self.account_pool, classified), - ) + if matches!( + classification.owner, + Some(SharedFailureClassification::CyberPolicy(_)) + ) { + self.cyber_policy + .exclude_account(&scope.cyber_policy, account_id) .await; - } - Some(SharedFailureClassification::Quota(classified)) => { - let _ = best_effort_controller_io( - "quota.stream", - QuotaController::apply_effect(&self.account_pool, classified), - ) - .await; - } - None => {} } } @@ -611,6 +633,11 @@ impl ControllerSet { StreamTerminal::Cancelled | StreamTerminal::DownstreamClosed => FinalOutcome::Cancelled, StreamTerminal::Shutdown => FinalOutcome::Shutdown, }; + if !matches!(outcome, FinalOutcome::Cancelled | FinalOutcome::Shutdown) + && let Some(classification) = stream_failure.as_ref() + { + self.apply_stream_account_state(classification).await; + } let _ = best_effort_controller_io( "cloudflare", cloudflare::CloudflareController::leave_stream(cloudflare::StreamExit { @@ -684,7 +711,7 @@ impl ControllerSet { if !matches!(outcome, FinalOutcome::Cancelled | FinalOutcome::Shutdown) { if let Some(classification) = stream_failure.as_ref() { - self.apply_stream_failure(&scope, &context.account_id, classification) + self.apply_stream_policy_failure(&scope, &context.account_id, classification) .await; } else { self.finalize_stream_success(&scope, &summary.terminal) @@ -730,18 +757,16 @@ impl ControllerSet { } } SharedFailureClassification::AccountFailure(classified) => { - let _ = best_effort_controller_io( - "account_failure.attempt", - AccountFailureController::apply_effect(&self.account_pool, classified), + AccountFailureController::apply_effect( + &self.account_pool, + &self.codex, + classified, ) .await; } SharedFailureClassification::Quota(classified) => { - let _ = best_effort_controller_io( - "quota.attempt", - QuotaController::apply_effect(&self.account_pool, classified), - ) - .await; + QuotaController::apply_effect(&self.account_pool, &self.codex, classified) + .await; } }, AttemptClassification::Cloudflare(failure) => { diff --git a/backend/src/dispatch/controllers/quota.rs b/backend/src/dispatch/controllers/quota.rs index a572c0220..7d6aa87c1 100644 --- a/backend/src/dispatch/controllers/quota.rs +++ b/backend/src/dispatch/controllers/quota.rs @@ -6,7 +6,10 @@ use serde_json::Value; use crate::{ dispatch::{ affinity::AccountIdentityService, - controllers::ControllerFailureFact, + controllers::{ + ControllerFailureFact, + account_state::{AccountStateEffect, AccountStateEffects}, + }, errors::ClientFailure, failure::exhaustion::{AccountExhaustionRecord, ExhaustedAccountKind}, lifecycle::contract::{ @@ -55,9 +58,7 @@ impl QuotaController { } let account_id = acquired.account.id.clone(); - let identity = context - .account_identity - .scope_auxiliary(&account_id, context.request_id); + let identity = context.account_identity.scope_auxiliary(&account_id); let usage = context .codex .fetch_usage(CodexRequestContext { @@ -75,8 +76,7 @@ impl QuotaController { installation_id: Some(&identity.installation_id), session_id: None, thread_id: None, - prompt_cache_key: None, - client_request_id: Some(&identity.client_request_id), + client_request_id: None, turn_id: None, }) .await; @@ -132,7 +132,7 @@ impl QuotaController { ExhaustedAccountKind::RateLimited, LIMIT_REACHED_MESSAGE, ), - effect: QuotaEffect::None, + effect: None, }); } classify(observation) @@ -160,26 +160,14 @@ impl QuotaController { pub(super) async fn apply_effect( account_pool: &AccountPoolService, + codex: &CodexBackendClient, classified: &ClassifiedQuotaFailure, ) { let Some(account_id) = classified.exhaustion.account_id.as_deref() else { return; }; - match classified.effect { - QuotaEffect::None => {} - QuotaEffect::SetExhausted => { - account_pool - .set_status( - account_id, - crate::fleet::account::AccountStatus::QuotaExhausted, - ) - .await; - } - QuotaEffect::MarkLimitedUntil(until) => { - account_pool - .mark_quota_limited_until(account_id, until) - .await; - } + if let Some(effect) = &classified.effect { + AccountStateEffects::apply(account_pool, codex, account_id, effect).await; } } @@ -188,15 +176,9 @@ impl QuotaController { } } -enum QuotaEffect { - None, - SetExhausted, - MarkLimitedUntil(chrono::DateTime), -} - pub(super) struct ClassifiedQuotaFailure { exhaustion: AccountExhaustionRecord, - effect: QuotaEffect, + effect: Option, } fn classify(observation: &AttemptObservation) -> Option { @@ -219,29 +201,37 @@ fn classify_upstream( account_id: &str, facts: &UpstreamFailureFacts, ) -> Option { - if facts.status_code == Some(429) { - return Some(rate_limited_failure( + let signaled_kind = facts + .code + .as_deref() + .and_then(classify_quota_signal) + .or_else(|| facts.error_type.as_deref().and_then(classify_quota_signal)) + .or_else(|| classify_quota_message(&facts.message)); + match signaled_kind.or(match facts.status_code { + Some(429) => Some(StreamQuotaKind::RateLimited), + Some(402) => Some(StreamQuotaKind::QuotaExhausted), + _ => None, + })? { + StreamQuotaKind::RateLimited => Some(rate_limited_failure( account_id, facts.body.clone(), facts.retry_after_seconds, - )); + )), + StreamQuotaKind::QuotaExhausted => { + Some(quota_exhausted_failure(account_id, facts.body.clone())) + } } - if facts.status_code == Some(402) { - return Some(quota_exhausted_failure(account_id, facts.body.clone())); - } - None } fn classify_stream( account_id: &str, failure: &ResponsesSseFailure, ) -> Option { - let kind = match failure.explicit_status_code { - Some(429) => StreamQuotaKind::RateLimited, - Some(402) => StreamQuotaKind::QuotaExhausted, - Some(_) => return None, - None => classify_stream_signal(failure)?, - }; + let kind = classify_stream_signal(failure).or(match failure.explicit_status_code { + Some(429) => Some(StreamQuotaKind::RateLimited), + Some(402) => Some(StreamQuotaKind::QuotaExhausted), + _ => None, + })?; let body = crate::dispatch::failure::sse::sse_failure_error_body(failure); Some(match kind { StreamQuotaKind::RateLimited => { @@ -277,10 +267,15 @@ fn classify_quota_signal(signal: &str) -> Option { "usage_limit_reached" | "rate_limit_exceeded" | "rate_limit_reached" - | "rate_limit_error" => Some(StreamQuotaKind::RateLimited), - "quota_exhausted" | "quota_exceeded" | "payment_required" | "insufficient_quota" => { - Some(StreamQuotaKind::QuotaExhausted) - } + | "rate_limit_error" + | "workspace_owner_usage_limit_reached" + | "workspace_member_usage_limit_reached" => Some(StreamQuotaKind::RateLimited), + "quota_exhausted" + | "quota_exceeded" + | "payment_required" + | "insufficient_quota" + | "workspace_owner_credits_depleted" + | "workspace_member_credits_depleted" => Some(StreamQuotaKind::QuotaExhausted), signal if signal.starts_with("billing_limit") => Some(StreamQuotaKind::QuotaExhausted), _ => None, } @@ -314,7 +309,9 @@ fn rate_limited_failure( ExhaustedAccountKind::RateLimited, message, ), - effect: QuotaEffect::MarkLimitedUntil(Utc::now() + Duration::seconds(seconds)), + effect: Some(AccountStateEffect::MarkQuotaLimitedUntil( + Utc::now() + Duration::seconds(seconds), + )), } } @@ -325,7 +322,9 @@ fn quota_exhausted_failure(account_id: &str, message: String) -> ClassifiedQuota ExhaustedAccountKind::QuotaExhausted, message, ), - effect: QuotaEffect::SetExhausted, + effect: Some(AccountStateEffect::SetStatus( + crate::fleet::account::AccountStatus::QuotaExhausted, + )), } } diff --git a/backend/src/dispatch/lifecycle/attempt.rs b/backend/src/dispatch/lifecycle/attempt.rs index 921c056c9..63ec562bf 100644 --- a/backend/src/dispatch/lifecycle/attempt.rs +++ b/backend/src/dispatch/lifecycle/attempt.rs @@ -9,7 +9,8 @@ use crate::{ affinity::AccountIdentityService, controllers::{ AttemptAccountPreparation, AttemptAccountPreparationContext, AttemptRoutePreparation, - AttemptUpstreamErrorContext, ControllerRequestScope, ControllerSet, + AttemptRoutePreparationContext, AttemptUpstreamErrorContext, ControllerRequestScope, + ControllerSet, }, errors::{ClientFailure, ResponseDispatchError}, failure::exhaustion::AccountExhaustionTracker, @@ -218,10 +219,13 @@ impl<'a> AttemptRunner<'a> { let acquired = prepared_account.lease; let cookie_header = prepared_account.cookie_header; - let attempt_request = match self.dependencies.controllers.prepare_attempt_route( + let account_scoped_request = match self.dependencies.controllers.prepare_attempt_route( controller_scope, request, - &account.id, + AttemptRoutePreparationContext { + account_identity: self.dependencies.account_identity, + account_id: &account.id, + }, ) { AttemptRoutePreparation::Ready(request) => *request, AttemptRoutePreparation::Unavailable { message } => { @@ -242,7 +246,6 @@ impl<'a> AttemptRunner<'a> { .start_attempt(&account.id); let upstream_context = AccountUpstreamContext { codex: self.dependencies.codex, - account_identity: self.dependencies.account_identity, request_id: self.dependencies.request_id, account: &account, cookie_header: cookie_header.as_deref(), @@ -253,9 +256,13 @@ impl<'a> AttemptRunner<'a> { started_at, tuple_schema, } => { - let result = - create_response_with_account(upstream_context, &attempt_request, *started_at) - .await; + let result = create_response_with_account( + upstream_context, + &account_scoped_request, + *started_at, + ) + .await; + let attempt_request = account_scoped_request.into_request(); acquired.complete().await; self.pending = Some(match result { Ok(response) => { @@ -317,7 +324,9 @@ impl<'a> AttemptRunner<'a> { } AttemptMode::Stream { tuple_schema } => { let result = - create_response_stream_with_account(upstream_context, &attempt_request).await; + create_response_stream_with_account(upstream_context, &account_scoped_request) + .await; + let attempt_request = account_scoped_request.into_request(); self.pending = Some(match result { Err(error) => { acquired.complete().await; diff --git a/backend/src/dispatch/service.rs b/backend/src/dispatch/service.rs index b742bfe70..58f277359 100644 --- a/backend/src/dispatch/service.rs +++ b/backend/src/dispatch/service.rs @@ -183,6 +183,7 @@ impl ResponseDispatchService { let controllers = ControllerSet::new( Arc::clone(&self.session_affinity), Arc::clone(&self.account_pool), + Arc::clone(&self.codex), self.cloudflare.clone(), Arc::clone(&self.recorder), ); diff --git a/backend/src/dispatch/stream/lifecycle.rs b/backend/src/dispatch/stream/lifecycle.rs index 0da07ca81..72c06e0f9 100644 --- a/backend/src/dispatch/stream/lifecycle.rs +++ b/backend/src/dispatch/stream/lifecycle.rs @@ -39,6 +39,7 @@ impl ResponseDispatchService { let controllers = ControllerSet::new( Arc::clone(&self.session_affinity), Arc::clone(&self.account_pool), + Arc::clone(&self.codex), self.cloudflare.clone(), Arc::clone(&self.recorder), ); diff --git a/backend/src/dispatch/transport/account.rs b/backend/src/dispatch/transport/account.rs index ee4c77cad..cf8b1e65d 100644 --- a/backend/src/dispatch/transport/account.rs +++ b/backend/src/dispatch/transport/account.rs @@ -3,21 +3,17 @@ use std::time::Instant; use crate::{ - dispatch::affinity::AccountIdentityService, + dispatch::affinity::AccountScopedRequest, fleet::account::Account, - upstream::openai::{ - protocol::responses::CodexResponsesRequest, - transport::{ - CodexBackendClient, CodexBackendResponse, CodexBackendStreamingResponse, - CodexClientError, CodexRequestContext, - }, + upstream::openai::transport::{ + CodexBackendClient, CodexBackendResponse, CodexBackendStreamingResponse, CodexClientError, + CodexRequestContext, }, }; #[derive(Clone, Copy)] pub(in crate::dispatch) struct AccountUpstreamContext<'a> { pub codex: &'a CodexBackendClient, - pub account_identity: &'a AccountIdentityService, pub request_id: &'a str, pub account: &'a Account, pub cookie_header: Option<&'a str>, @@ -25,35 +21,14 @@ pub(in crate::dispatch) struct AccountUpstreamContext<'a> { pub(in crate::dispatch) async fn create_response_with_account( context: AccountUpstreamContext<'_>, - request: &CodexResponsesRequest, + request: &AccountScopedRequest, started_at: Instant, ) -> Result { - let identity = context - .account_identity - .scope(request, &context.account.id, context.request_id); context .codex .create_response_with_pool_account_started_at( - request, - CodexRequestContext { - access_token: &context.account.access_token, - account_id: context.account.account_id.as_deref(), - request_id: context.request_id, - turn_state: request.turn_state.as_deref(), - turn_metadata: request.turn_metadata.as_deref(), - beta_features: request.beta_features.as_deref(), - include_timing_metrics: request.include_timing_metrics.as_deref(), - version: request.version.as_deref(), - codex_window_id: identity.window_id.as_deref(), - parent_thread_id: identity.parent_thread_id.as_deref(), - cookie_header: context.cookie_header, - installation_id: Some(&identity.installation_id), - session_id: identity.session_id.as_deref(), - thread_id: identity.thread_id.as_deref(), - prompt_cache_key: identity.prompt_cache_key.as_deref(), - client_request_id: Some(&identity.client_request_id), - turn_id: identity.turn_id.as_deref(), - }, + request.request(), + codex_request_context(context, request), Some(&context.account.id), started_at, ) @@ -62,35 +37,40 @@ pub(in crate::dispatch) async fn create_response_with_account( pub(in crate::dispatch) async fn create_response_stream_with_account( context: AccountUpstreamContext<'_>, - request: &CodexResponsesRequest, + request: &AccountScopedRequest, ) -> Result { - let identity = context - .account_identity - .scope(request, &context.account.id, context.request_id); context .codex .create_response_stream_with_pool_account( - request, - CodexRequestContext { - access_token: &context.account.access_token, - account_id: context.account.account_id.as_deref(), - request_id: context.request_id, - turn_state: request.turn_state.as_deref(), - turn_metadata: request.turn_metadata.as_deref(), - beta_features: request.beta_features.as_deref(), - include_timing_metrics: request.include_timing_metrics.as_deref(), - version: request.version.as_deref(), - codex_window_id: identity.window_id.as_deref(), - parent_thread_id: identity.parent_thread_id.as_deref(), - cookie_header: context.cookie_header, - installation_id: Some(&identity.installation_id), - session_id: identity.session_id.as_deref(), - thread_id: identity.thread_id.as_deref(), - prompt_cache_key: identity.prompt_cache_key.as_deref(), - client_request_id: Some(&identity.client_request_id), - turn_id: identity.turn_id.as_deref(), - }, + request.request(), + codex_request_context(context, request), Some(&context.account.id), ) .await } + +fn codex_request_context<'a>( + context: AccountUpstreamContext<'a>, + request: &'a AccountScopedRequest, +) -> CodexRequestContext<'a> { + let request_body = request.request(); + let identity = request.identity(); + CodexRequestContext { + access_token: &context.account.access_token, + account_id: context.account.account_id.as_deref(), + request_id: context.request_id, + turn_state: request_body.turn_state.as_deref(), + turn_metadata: request_body.turn_metadata.as_deref(), + beta_features: request_body.beta_features.as_deref(), + include_timing_metrics: request_body.include_timing_metrics.as_deref(), + version: request_body.version.as_deref(), + codex_window_id: request_body.codex_window_id.as_deref(), + parent_thread_id: request_body.parent_thread_id.as_deref(), + cookie_header: context.cookie_header, + installation_id: Some(&identity.installation_id), + session_id: request_body.client_session_id.as_deref(), + thread_id: request_body.client_thread_id.as_deref(), + client_request_id: request_body.client_request_id.as_deref(), + turn_id: request_body.client_turn_id.as_deref(), + } +} diff --git a/backend/src/dispatch/transport/observation.rs b/backend/src/dispatch/transport/observation.rs index 4f66978f5..6cf622c6d 100644 --- a/backend/src/dispatch/transport/observation.rs +++ b/backend/src/dispatch/transport/observation.rs @@ -28,6 +28,7 @@ pub(in crate::dispatch) struct UpstreamFailureFacts { pub kind: UpstreamFailureKind, pub status_code: Option, pub code: Option, + pub error_type: Option, pub message: String, pub body: String, pub retry_after_seconds: Option, @@ -47,6 +48,7 @@ pub(in crate::dispatch) fn upstream_failure_facts( }, status_code: error.status().map(|status| status.as_u16()), code: None, + error_type: None, message: error.to_string(), body: error.to_string(), retry_after_seconds: None, @@ -57,11 +59,12 @@ pub(in crate::dispatch) fn upstream_failure_facts( retry_after_seconds, .. } => { - let (code, message) = error_fields(body); + let (code, error_type, message) = error_fields(body); UpstreamFailureFacts { kind: UpstreamFailureKind::HttpStatus, status_code: Some(status.as_u16()), code, + error_type, message: message.unwrap_or_else(|| body.clone()), body: body.clone(), retry_after_seconds: *retry_after_seconds, @@ -71,17 +74,16 @@ pub(in crate::dispatch) fn upstream_failure_facts( simple_facts(UpstreamFailureKind::StreamIdle, error) } CodexClientError::WebSocket(CodexWebSocketExchangeError::Upstream(upstream)) => { + let (body_code, body_type, body_message) = error_fields(&upstream.body); UpstreamFailureFacts { kind: UpstreamFailureKind::WebSocketUpstream, status_code: Some(upstream.status_code), - code: upstream - .code - .clone() - .or_else(|| error_fields(&upstream.body).0), + code: upstream.code.clone().or(body_code), + error_type: upstream.error_type.clone().or(body_type), message: upstream .message .clone() - .or_else(|| error_fields(&upstream.body).1) + .or(body_message) .unwrap_or_else(|| upstream.body.clone()), body: upstream.body.clone(), retry_after_seconds: upstream.retry_after_seconds, @@ -123,15 +125,16 @@ fn simple_facts(kind: UpstreamFailureKind, error: &CodexClientError) -> Upstream kind, status_code: None, code: None, + error_type: None, body: message.clone(), message, retry_after_seconds: None, } } -fn error_fields(body: &str) -> (Option, Option) { +fn error_fields(body: &str) -> (Option, Option, Option) { let Ok(value) = serde_json::from_str::(body) else { - return (None, None); + return (None, None, None); }; let error = value .pointer("/response/error") @@ -143,6 +146,11 @@ fn error_fields(body: &str) -> (Option, Option) { .or_else(|| value.get("code")) .and_then(Value::as_str) .map(ToString::to_string), + error + .get("type") + .or_else(|| value.get("type")) + .and_then(Value::as_str) + .map(ToString::to_string), error .get("message") .or_else(|| value.get("message")) diff --git a/backend/src/fleet/manage/import.rs b/backend/src/fleet/manage/import.rs index e37b542f9..eac665167 100644 --- a/backend/src/fleet/manage/import.rs +++ b/backend/src/fleet/manage/import.rs @@ -208,7 +208,6 @@ impl AccountManageService { installation_id: installation_id.as_deref(), session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }; diff --git a/backend/src/fleet/manage/probe.rs b/backend/src/fleet/manage/probe.rs index 3ed51eec0..e8383cb54 100644 --- a/backend/src/fleet/manage/probe.rs +++ b/backend/src/fleet/manage/probe.rs @@ -171,7 +171,6 @@ impl AccountManageService { installation_id: Some(&installation_id), session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }; diff --git a/backend/src/fleet/manage/quota.rs b/backend/src/fleet/manage/quota.rs index 8fb302447..dd29e8570 100644 --- a/backend/src/fleet/manage/quota.rs +++ b/backend/src/fleet/manage/quota.rs @@ -50,7 +50,6 @@ impl AccountManageService { installation_id: Some(&installation_id), session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }; diff --git a/backend/src/fleet/pool/mod.rs b/backend/src/fleet/pool/mod.rs index ea98e142c..70292391b 100644 --- a/backend/src/fleet/pool/mod.rs +++ b/backend/src/fleet/pool/mod.rs @@ -33,8 +33,11 @@ use crate::upstream::openai::protocol::events::{ pub use crate::fleet::scheduler::RotationStrategy; mod filters; +mod persistence; mod state; +use persistence::AccountStatePersistence; + pub use filters::*; pub use state::*; @@ -71,6 +74,7 @@ pub struct AccountPoolService { usage_store: Arc, request_interval: Arc>, candidate_changed: Arc, + state_persistence: AccountStatePersistence, } /// 一次请求对账号池 slot 的唯一所有权凭据。 @@ -191,6 +195,7 @@ impl AccountPoolService { options: AccountPoolOptions, request_interval_ms: u64, ) -> Self { + let state_persistence = AccountStatePersistence::new(Arc::clone(&store)); Self { pool: Arc::new(tokio::sync::Mutex::new(AccountPool::with_options(options))), store, @@ -199,6 +204,7 @@ impl AccountPoolService { request_interval_ms, ))), candidate_changed: Arc::new(tokio::sync::Notify::new()), + state_persistence, } } @@ -340,9 +346,26 @@ impl AccountPoolService { } }; let in_memory = self.pool.lock().await.set_status(account_id, status); + if in_memory { + self.candidate_changed.notify_waiters(); + } persisted || in_memory } + /// 先在线性化的内存池中更新账号状态，再将持久化写入后台队列。 + pub async fn set_status_immediately(&self, account_id: &str, status: AccountStatus) -> bool { + let in_memory = { + let mut pool = self.pool.lock().await; + let in_memory = pool.set_status(account_id, status); + self.state_persistence.set_status(account_id, status); + in_memory + }; + if in_memory { + self.candidate_changed.notify_waiters(); + } + in_memory + } + /// 标记账号 quota 冷却状态。 pub async fn mark_quota_limited_until(&self, account_id: &str, until: DateTime) -> bool { let persisted = match self.store.mark_quota_limited_until(account_id, until).await { @@ -361,9 +384,31 @@ impl AccountPoolService { .lock() .await .mark_quota_limited_until(account_id, until); + if in_memory { + self.candidate_changed.notify_waiters(); + } persisted || in_memory } + /// 先在线性化的内存池中标记配额冷却，再将持久化写入后台队列。 + pub async fn mark_quota_limited_until_immediately( + &self, + account_id: &str, + until: DateTime, + ) -> bool { + let in_memory = { + let mut pool = self.pool.lock().await; + let in_memory = pool.mark_quota_limited_until(account_id, until); + self.state_persistence + .mark_quota_limited_until(account_id, until); + in_memory + }; + if in_memory { + self.candidate_changed.notify_waiters(); + } + in_memory + } + /// 按请求上下文获取可用账号。 pub async fn acquire_with(&self, request: &AccountAcquireRequest) -> Option { let refresh = self.pool.lock().await.acquire_with_status_refresh(request); diff --git a/backend/src/fleet/pool/persistence.rs b/backend/src/fleet/pool/persistence.rs new file mode 100644 index 000000000..dcdcb6296 --- /dev/null +++ b/backend/src/fleet/pool/persistence.rs @@ -0,0 +1,94 @@ +//! 账号运行状态的有序后台持久化。 + +use std::sync::Arc; + +use chrono::{DateTime, Utc}; +use tokio::sync::mpsc; + +use crate::fleet::{account::AccountStatus, store::AccountStore}; + +#[derive(Clone)] +pub(super) struct AccountStatePersistence { + sender: mpsc::UnboundedSender, +} + +enum AccountStateWrite { + SetStatus { + account_id: String, + status: AccountStatus, + }, + MarkQuotaLimitedUntil { + account_id: String, + until: DateTime, + }, +} + +impl AccountStatePersistence { + pub(super) fn new(store: Arc) -> Self { + let (sender, receiver) = mpsc::unbounded_channel(); + tokio::spawn(run(receiver, store)); + Self { sender } + } + + pub(super) fn set_status(&self, account_id: &str, status: AccountStatus) { + self.enqueue(AccountStateWrite::SetStatus { + account_id: account_id.to_owned(), + status, + }); + } + + pub(super) fn mark_quota_limited_until(&self, account_id: &str, until: DateTime) { + self.enqueue(AccountStateWrite::MarkQuotaLimitedUntil { + account_id: account_id.to_owned(), + until, + }); + } + + fn enqueue(&self, write: AccountStateWrite) { + if self.sender.send(write).is_err() { + tracing::error!("Account state persistence worker stopped unexpectedly"); + } + } +} + +async fn run( + mut receiver: mpsc::UnboundedReceiver, + store: Arc, +) { + while let Some(write) = receiver.recv().await { + match write { + AccountStateWrite::SetStatus { account_id, status } => { + match store.set_status(&account_id, status).await { + Ok(true) => {} + Ok(false) => tracing::warn!( + account_id, + ?status, + "Account disappeared before deferred status persistence" + ), + Err(error) => tracing::error!( + account_id, + ?status, + error = %error, + "Failed to persist deferred account status" + ), + } + } + AccountStateWrite::MarkQuotaLimitedUntil { account_id, until } => { + match store.mark_quota_limited_until(&account_id, until).await { + Ok(true) => {} + Ok(false) => tracing::warn!( + account_id, + %until, + "Account disappeared before deferred quota persistence" + ), + Err(error) => tracing::error!( + account_id, + %until, + error = %error, + "Failed to persist deferred quota state" + ), + } + } + } + } +} diff --git a/backend/src/fleet/quota/runtime.rs b/backend/src/fleet/quota/runtime.rs index c598f3b6a..6970308c5 100644 --- a/backend/src/fleet/quota/runtime.rs +++ b/backend/src/fleet/quota/runtime.rs @@ -176,7 +176,6 @@ impl QuotaRefreshService { installation_id: Some(&installation_id), session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }) diff --git a/backend/src/upstream/openai/protocol/events.rs b/backend/src/upstream/openai/protocol/events.rs index 4b1f27997..335bf4276 100644 --- a/backend/src/upstream/openai/protocol/events.rs +++ b/backend/src/upstream/openai/protocol/events.rs @@ -520,7 +520,14 @@ fn retry_after_seconds_from_rate_limit_message(error: &Value) -> Option { .get("code") .or_else(|| error.get("type")) .and_then(Value::as_str)?; - if code != "rate_limit_exceeded" { + if !matches!( + code, + "rate_limit_exceeded" + | "rate_limit_reached" + | "usage_limit_reached" + | "workspace_owner_usage_limit_reached" + | "workspace_member_usage_limit_reached" + ) { return None; } let message = error.get("message").and_then(Value::as_str)?; diff --git a/backend/src/upstream/openai/protocol/responses.rs b/backend/src/upstream/openai/protocol/responses.rs index 725390444..6978c0b42 100644 --- a/backend/src/upstream/openai/protocol/responses.rs +++ b/backend/src/upstream/openai/protocol/responses.rs @@ -724,9 +724,9 @@ impl CodexResponsesRequest { } } - /// 仅在客户端原本提供字符串字段时替换其账号作用域 wire 值。 + /// 替换客户端原本提供的账号身份字段；无法安全重建时删除该字段。 pub(crate) fn replace_existing_identity_field(&mut self, key: &str, value: Option<&str>) { - if !self.body.get(key).is_some_and(Value::is_string) { + if !self.body.contains_key(key) { return; } match value.filter(|value| !value.trim().is_empty()) { diff --git a/backend/src/upstream/openai/transport/client.rs b/backend/src/upstream/openai/transport/client.rs index f361196b9..525464ad9 100644 --- a/backend/src/upstream/openai/transport/client.rs +++ b/backend/src/upstream/openai/transport/client.rs @@ -57,12 +57,12 @@ const UPSTREAM_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); const UPSTREAM_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60); const X_CODEX_WS_STREAM_REQUEST_START_MS_CLIENT_METADATA_KEY: &str = "x-codex-ws-stream-request-start-ms"; -type ReqwestClientCacheKey = (bool, Option); +type ReqwestClientCacheKey = Option; type ReqwestClientCache = Mutex>; -/// 构建带缓存的 reqwest Client。若 `force_http11` 为 true 则强制 HTTP/1.1。 -pub fn build_reqwest_client(force_http11: bool) -> Result { - let cache_key = (force_http11, custom_ca_env_cache_key()); +/// 构建带缓存、自动协商 HTTP/2 的 reqwest Client。 +pub fn build_reqwest_client() -> Result { + let cache_key = custom_ca_env_cache_key(); static CLIENTS: OnceLock = OnceLock::new(); let cache = CLIENTS.get_or_init(|| Mutex::new(HashMap::new())); if let Some(client) = cache @@ -83,11 +83,6 @@ pub fn build_reqwest_client(force_http11: bool) -> Result .brotli(true) .zstd(true) .deflate(true); - let builder = if force_http11 { - builder.http1_only() - } else { - builder - }; let client = build_reqwest_client_with_custom_ca(builder)?; let mut clients = cache .lock() @@ -232,11 +227,11 @@ pub struct CodexRequestContext<'a> { pub access_token: &'a str, /// ChatGPT account id。 pub account_id: Option<&'a str>, - /// 请求 ID。 + /// 代理请求 ID。 pub request_id: &'a str, - /// x-codex-turn-state。 + /// 当前账号同一 turn 内的 opaque sticky-routing 状态。 pub turn_state: Option<&'a str>, - /// x-codex-turn-metadata。 + /// 客户端 turn metadata；其中 installation ID 已按当前账号处理。 pub turn_metadata: Option<&'a str>, /// x-codex-beta-features。 pub beta_features: Option<&'a str>, @@ -244,23 +239,21 @@ pub struct CodexRequestContext<'a> { pub include_timing_metrics: Option<&'a str>, /// version。 pub version: Option<&'a str>, - /// x-codex-window-id。 + /// 客户端 window ID。 pub codex_window_id: Option<&'a str>, - /// x-codex-parent-thread-id。 + /// 客户端 parent thread ID。 pub parent_thread_id: Option<&'a str>, /// cookie 头。 pub cookie_header: Option<&'a str>, - /// x-codex-installation-id。 + /// 当前账号稳定派生的 installation ID。 pub installation_id: Option<&'a str>, - /// session_id。 + /// 客户端 session ID。 pub session_id: Option<&'a str>, - /// thread_id。 + /// 客户端 thread ID。 pub thread_id: Option<&'a str>, - /// 账号作用域 prompt cache key。 - pub prompt_cache_key: Option<&'a str>, - /// 账号作用域 client request ID。 + /// 客户端逻辑 request ID；缺失时使用代理请求 ID。 pub client_request_id: Option<&'a str>, - /// 账号作用域 turn ID。 + /// 客户端 turn ID。 pub turn_id: Option<&'a str>, } @@ -391,7 +384,6 @@ impl CodexModelCatalogClient for CodexBackendClient { installation_id: request.installation_id, session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }) @@ -517,29 +509,6 @@ fn is_model_entry(value: &Value) -> bool { // Request helpers // --------------------------------------------------------------------------- -fn response_upstream_request( - request: &CodexResponsesRequest, - context: CodexRequestContext<'_>, -) -> CodexResponsesRequest { - let mut upstream = request.clone(); - if request.prompt_cache_key().is_some() { - upstream.set_prompt_cache_key(context.prompt_cache_key.map(ToString::to_string)); - } - for (key, value) in [ - ("session_id", context.session_id), - ("thread_id", context.thread_id), - ("turn_id", context.turn_id), - ("x-client-request-id", context.client_request_id), - ("x-codex-window-id", context.codex_window_id), - ("x-codex-parent-thread-id", context.parent_thread_id), - ("x-codex-installation-id", context.installation_id), - ] { - upstream.replace_existing_identity_field(key, value); - } - upstream.set_client_metadata(response_client_metadata(request.client_metadata(), context)); - upstream -} - fn websocket_upstream_request(request: &CodexResponsesRequest) -> CodexResponsesRequest { let mut request = request.clone(); stamp_ws_stream_request_start_ms(&mut request); @@ -566,66 +535,6 @@ fn now_unix_timestamp_millis() -> u128 { .unwrap_or_default() } -fn response_client_metadata( - client_metadata: Option<&Value>, - context: CodexRequestContext<'_>, -) -> Option { - // 以客户端原始 client_metadata 为基础（保留 number/bool/object 等所有值类型）， - // 在其上追加代理自身的上下文字段。 - let mut metadata = match client_metadata? { - Value::Object(input) => input.clone(), - value => return Some(value.clone()), - }; - - replace_metadata_string( - &mut metadata, - "x-codex-installation-id", - context.installation_id, - ); - replace_metadata_string(&mut metadata, "session_id", context.session_id); - replace_metadata_string(&mut metadata, "thread_id", context.thread_id); - replace_metadata_string( - &mut metadata, - "x-client-request-id", - context.client_request_id, - ); - replace_metadata_string(&mut metadata, "turn_id", context.turn_id); - replace_metadata_string(&mut metadata, "x-codex-window-id", context.codex_window_id); - insert_metadata_string( - &mut metadata, - "x-codex-turn-metadata", - context.turn_metadata, - ); - replace_metadata_string( - &mut metadata, - "x-codex-parent-thread-id", - context.parent_thread_id, - ); - - if metadata.is_empty() { - None - } else { - Some(Value::Object(metadata)) - } -} - -fn insert_metadata_string(metadata: &mut Map, key: &str, value: Option<&str>) { - if let Some(value) = value.filter(|value| !value.trim().is_empty()) { - metadata.insert(key.to_string(), Value::String(value.to_string())); - } -} - -fn replace_metadata_string(metadata: &mut Map, key: &str, value: Option<&str>) { - match value.filter(|value| !value.trim().is_empty()) { - Some(value) => { - metadata.insert(key.to_string(), Value::String(value.to_string())); - } - None => { - metadata.remove(key); - } - } -} - fn openai_subagent_from_metadata(client_metadata: Option<&Value>) -> Option { let value = client_metadata? .as_object()? diff --git a/backend/src/upstream/openai/transport/client_sse.rs b/backend/src/upstream/openai/transport/client_sse.rs index 5f0b3762a..689c36d3b 100644 --- a/backend/src/upstream/openai/transport/client_sse.rs +++ b/backend/src/upstream/openai/transport/client_sse.rs @@ -167,7 +167,7 @@ impl CodexBackendClient { pool_account_id: Option<&str>, started_at: Instant, ) -> CodexClientResult { - let upstream_request = response_upstream_request(request, context); + let upstream_request = request.clone(); match transport_for_request(&upstream_request) { CodexTransport::HttpSse => { self.create_response_http_sse(&upstream_request, context, started_at) @@ -222,7 +222,7 @@ impl CodexBackendClient { context: CodexRequestContext<'_>, pool_account_id: Option<&str>, ) -> CodexClientResult { - let upstream_request = response_upstream_request(request, context); + let upstream_request = request.clone(); match transport_for_request(&upstream_request) { CodexTransport::HttpSse => { self.create_response_stream_http_sse(&upstream_request, context) diff --git a/backend/tests/bootstrap/config/mod.rs b/backend/tests/bootstrap/config/mod.rs index f5a77bab8..9d6d631bd 100644 --- a/backend/tests/bootstrap/config/mod.rs +++ b/backend/tests/bootstrap/config/mod.rs @@ -133,6 +133,19 @@ fn config_loader_should_reject_unknown_fields() { assert!(matches!(error, ConfigError::InvalidDocument { .. })); } +#[test] +fn config_loader_should_reject_removed_tls_section() { + let yaml = complete_config().replace( + " ws_pool:", + " tls:\n force_http11: false\n\n ws_pool:", + ); + let directory = write_config(&yaml); + + let error = BootstrapConfig::load_from_path(directory.path().join("config.yaml")).unwrap_err(); + + assert!(matches!(error, ConfigError::InvalidDocument { .. })); +} + #[test] fn config_loader_should_reject_missing_explicit_fields() { let yaml = complete_config().replace(" enabled: true\n\n\n# Compose", "\n\n# Compose"); diff --git a/backend/tests/dispatch/lifecycle/mod.rs b/backend/tests/dispatch/lifecycle/mod.rs index 0312ef634..ba9f02aac 100644 --- a/backend/tests/dispatch/lifecycle/mod.rs +++ b/backend/tests/dispatch/lifecycle/mod.rs @@ -240,11 +240,12 @@ fn controller_exit_order_should_be_static_and_effect_only() { assert_appears_in_order( stream_exit, &[ + "self.apply_stream_account_state", "CloudflareController::leave_stream", "UsageController::leave_stream", "AffinityController::leave_stream", "TelemetryController::leave_stream", - "self.apply_stream_failure", + "self.apply_stream_policy_failure", ], ); } diff --git a/backend/tests/dispatch/service/responses_http.rs b/backend/tests/dispatch/service/responses_http.rs index 13d45722e..23646c0f3 100644 --- a/backend/tests/dispatch/service/responses_http.rs +++ b/backend/tests/dispatch/service/responses_http.rs @@ -677,10 +677,7 @@ async fn responses_should_forward_parity_fields_context_headers_and_account_scop assert_eq!(upstream_body["service_tier"], "priority"); assert_eq!(upstream_body["reasoning"], json!({"effort": "high"})); assert!(upstream_body.get("include").is_none()); - let prompt_cache_key = upstream_body["prompt_cache_key"] - .as_str() - .expect("prompt cache key should be account scoped"); - assert_ne!(prompt_cache_key, "pcache"); + assert_eq!(upstream_body["prompt_cache_key"], "pcache"); assert_eq!(upstream_body["client_metadata"]["safe"], "yes"); assert_eq!(upstream_body["client_metadata"]["drop"], 42); assert_eq!( @@ -695,16 +692,16 @@ async fn responses_should_forward_parity_fields_context_headers_and_account_scop assert!(upstream_body["client_metadata"].get("thread_id").is_none()); let window_id = upstream_body["client_metadata"]["x-codex-window-id"] .as_str() - .expect("window id should be account scoped"); - assert_ne!(window_id, "window-metadata"); + .expect("window id should be preserved"); + assert_eq!(window_id, "window-metadata"); assert_eq!( upstream_body["client_metadata"]["x-codex-turn-metadata"], "meta-metadata" ); let parent_thread_id = upstream_body["client_metadata"]["x-codex-parent-thread-id"] .as_str() - .expect("parent thread id should be account scoped"); - assert_ne!(parent_thread_id, "parent-metadata"); + .expect("parent thread id should be preserved"); + assert_eq!(parent_thread_id, "parent-metadata"); assert_eq!(upstream_header("session-id"), None); assert_eq!(upstream_header("thread-id"), None); assert_eq!(upstream_header("session_id"), None); diff --git a/backend/tests/dispatch/service/responses_recovery.rs b/backend/tests/dispatch/service/responses_recovery.rs index 6379aed36..3c6020133 100644 --- a/backend/tests/dispatch/service/responses_recovery.rs +++ b/backend/tests/dispatch/service/responses_recovery.rs @@ -544,6 +544,57 @@ async fn responses_should_mark_expired_after_401_and_fallback() { assert_eq!(account_status.0, "expired"); } +#[tokio::test] +async fn responses_should_disable_identity_verification_account_and_fallback() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-primary")) + .respond_with(ResponseTemplate::new(400).set_body_json(json!({ + "error": { + "code": "identity_verification_required", + "message": "Identity verification is required before using this account" + } + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-secondary")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(RESPONSES_AFTER_403_SSE), + ) + .expect(1) + .mount(&server) + .await; + let (app, api_key, pool, _dir) = test_app_with_two_accounts(server.uri()).await; + + let response = app + .oneshot(responses_json_request( + &api_key, + &json!({ + "model": "gpt-5.5", + "input": [{"role": "user", "content": "Say hello"}], + "stream": false, + "use_websocket": false + }), + )) + .await + .unwrap(); + let body = response_json(response).await; + let account_status: String = sqlx::query_scalar("select status from accounts where id = $1") + .bind("acct_primary") + .fetch_one(&pool) + .await + .unwrap(); + + assert_eq!(body["id"], "resp_after_403"); + assert_eq!(account_status, "disabled"); +} + #[tokio::test] async fn responses_should_recover_auth_failure_from_sse_failed_event() { let server = MockServer::start().await; @@ -1168,12 +1219,12 @@ async fn responses_should_disable_account_after_three_cloudflare_path_block_404s } #[tokio::test] -async fn responses_should_fallback_after_http_model_unsupported() { +async fn responses_should_fallback_after_http_404_model_unsupported() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/codex/responses")) .and(header("authorization", "Bearer access-primary")) - .respond_with(ResponseTemplate::new(400).set_body_json(json!({ + .respond_with(ResponseTemplate::new(404).set_body_json(json!({ "error": { "code": "model_not_supported", "message": "Model gpt-5.5 is not supported on this account plan" @@ -2401,6 +2452,274 @@ async fn responses_should_rotate_account_after_5xx_without_same_account_retry() assert_eq!(secondary_requests, 1, "requests: {authorizations:?}"); } +#[tokio::test] +async fn responses_failover_should_replace_account_identity_and_preserve_session_semantics() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-primary")) + .respond_with(ResponseTemplate::new(401).set_body_json(json!({ + "error": { + "type": "authentication_error", + "code": "token_invalidated", + "message": "token is no longer valid" + } + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-secondary")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(RESPONSES_AFTER_5XX_RETRY_SSE), + ) + .expect(1) + .mount(&server) + .await; + + let (app, api_key, pool, _dir) = test_app_with_two_accounts(server.uri()).await; + let cookie_store = PgCookieStore::new(pool); + cookie_store + .capture_set_cookie( + "acct_primary", + "cf_clearance=primary; Domain=.chatgpt.com; Path=/codex", + ) + .await + .unwrap(); + cookie_store + .capture_set_cookie( + "acct_secondary", + "cf_clearance=secondary; Domain=.chatgpt.com; Path=/codex", + ) + .await + .unwrap(); + let turn_metadata = json!({ + "installation_id": "installation-client", + "session_id": "session-client", + "thread_id": "thread-client", + "turn_id": "turn-client", + "window_id": "window-client", + "forked_from_thread_id": "forked-client", + "parent_thread_id": "parent-client", + "request_kind": "turn", + "workspace_kind": "local" + }) + .to_string(); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/responses") + .header("authorization", format!("Bearer {api_key}")) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "model": "gpt-5.5", + "input": [{"role": "user", "content": "Replay without identity pollution"}], + "stream": false, + "use_websocket": false, + "prompt_cache_key": "cache-client", + "conversation_id": "conversation-client", + "session_id": "session-client", + "thread_id": "thread-client", + "turn_id": "turn-client", + "x-client-request-id": "request-client", + "x-codex-window-id": "window-client", + "x-codex-parent-thread-id": "parent-client", + "x-codex-installation-id": "installation-client", + "turnState": "turn-state-primary", + "turnMetadata": turn_metadata, + "codexWindowId": "window-client", + "parentThreadId": "parent-client", + "client_metadata": { + "safe": "preserved", + "conversation_id": "conversation-client", + "session_id": "session-client", + "thread_id": "thread-client", + "turn_id": "turn-client", + "x-client-request-id": "request-client", + "x-codex-window-id": "window-client", + "x-codex-parent-thread-id": "parent-client", + "x-codex-installation-id": "installation-client", + "x-codex-turn-state": "turn-state-primary", + "x-codex-turn-metadata": turn_metadata + } + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let requests = server.received_requests().await.unwrap(); + let responses_requests = requests + .iter() + .filter(|request| request.url.path() == "/codex/responses") + .collect::>(); + let primary = responses_requests + .iter() + .copied() + .find(|request| { + request + .headers + .get("authorization") + .is_some_and(|value| value == "Bearer access-primary") + }) + .unwrap(); + let secondary = responses_requests + .iter() + .copied() + .find(|request| { + request + .headers + .get("authorization") + .is_some_and(|value| value == "Bearer access-secondary") + }) + .unwrap(); + let primary_body: Value = serde_json::from_slice(&primary.body).unwrap(); + let secondary_body: Value = serde_json::from_slice(&secondary.body).unwrap(); + let primary_metadata = &primary_body["client_metadata"]; + let secondary_metadata = &secondary_body["client_metadata"]; + + assert_eq!(responses_requests.len(), 2); + assert_eq!(primary.headers["chatgpt-account-id"], "chatgpt-primary"); + assert_eq!(secondary.headers["chatgpt-account-id"], "chatgpt-secondary"); + assert_eq!(primary.headers["cookie"], "cf_clearance=primary"); + assert_eq!(secondary.headers["cookie"], "cf_clearance=secondary"); + assert_ne!( + primary_metadata["x-codex-installation-id"], + secondary_metadata["x-codex-installation-id"] + ); + for key in [ + "session_id", + "thread_id", + "turn_id", + "x-client-request-id", + "x-codex-window-id", + "x-codex-parent-thread-id", + ] { + assert_eq!(primary_metadata[key], secondary_metadata[key], "key: {key}"); + } + assert_ne!( + primary_body["x-codex-installation-id"], + secondary_body["x-codex-installation-id"] + ); + for key in [ + "prompt_cache_key", + "conversation_id", + "session_id", + "thread_id", + "turn_id", + "x-client-request-id", + "x-codex-window-id", + "x-codex-parent-thread-id", + "codexWindowId", + "parentThreadId", + ] { + assert_eq!(primary_body[key], secondary_body[key], "key: {key}"); + } + let primary_turn_metadata: Value = + serde_json::from_str(primary_metadata["x-codex-turn-metadata"].as_str().unwrap()).unwrap(); + let secondary_turn_metadata: Value = serde_json::from_str( + secondary_metadata["x-codex-turn-metadata"] + .as_str() + .unwrap(), + ) + .unwrap(); + assert_ne!( + primary_turn_metadata["installation_id"], + secondary_turn_metadata["installation_id"] + ); + for key in [ + "session_id", + "thread_id", + "turn_id", + "window_id", + "forked_from_thread_id", + "parent_thread_id", + ] { + assert_eq!( + primary_turn_metadata[key], secondary_turn_metadata[key], + "turn metadata key: {key}" + ); + } + for (request, body, metadata, turn_metadata) in [ + ( + primary, + &primary_body, + primary_metadata, + &primary_turn_metadata, + ), + ( + secondary, + &secondary_body, + secondary_metadata, + &secondary_turn_metadata, + ), + ] { + for (header_key, body_key, metadata_key, turn_metadata_key) in [ + ( + "x-codex-installation-id", + "x-codex-installation-id", + "x-codex-installation-id", + "installation_id", + ), + ("session-id", "session_id", "session_id", "session_id"), + ("thread-id", "thread_id", "thread_id", "thread_id"), + ("x-codex-turn-id", "turn_id", "turn_id", "turn_id"), + ( + "x-codex-window-id", + "x-codex-window-id", + "x-codex-window-id", + "window_id", + ), + ( + "x-codex-parent-thread-id", + "x-codex-parent-thread-id", + "x-codex-parent-thread-id", + "parent_thread_id", + ), + ] { + let header_value = request.headers[header_key].to_str().unwrap(); + assert_eq!(body[body_key].as_str(), Some(header_value)); + assert_eq!(metadata[metadata_key].as_str(), Some(header_value)); + assert_eq!( + turn_metadata[turn_metadata_key].as_str(), + Some(header_value) + ); + } + assert_eq!( + body["x-client-request-id"].as_str(), + request.headers["x-client-request-id"].to_str().ok() + ); + assert_eq!( + metadata["x-client-request-id"].as_str(), + request.headers["x-client-request-id"].to_str().ok() + ); + assert_eq!( + body["turnMetadata"].as_str(), + metadata["x-codex-turn-metadata"].as_str() + ); + assert_eq!( + request.headers["x-codex-turn-metadata"].to_str().ok(), + metadata["x-codex-turn-metadata"].as_str() + ); + } + assert_eq!(primary_body["turnState"], "turn-state-primary"); + assert_eq!(primary_metadata["x-codex-turn-state"], "turn-state-primary"); + assert!(secondary_body.get("turnState").is_none()); + assert!(secondary_metadata.get("x-codex-turn-state").is_none()); + assert!(secondary.headers.get("x-codex-turn-state").is_none()); + assert_eq!(secondary_metadata["safe"], "preserved"); + assert_eq!(secondary_body["input"], primary_body["input"]); + assert!(secondary_body.get("previous_response_id").is_none()); +} + #[tokio::test] async fn responses_should_traverse_all_configured_strategy_candidates_without_fixed_cap() { let server = MockServer::start().await; diff --git a/backend/tests/dispatch/service/responses_websocket.rs b/backend/tests/dispatch/service/responses_websocket.rs index b9118c5e1..b71df56f2 100644 --- a/backend/tests/dispatch/service/responses_websocket.rs +++ b/backend/tests/dispatch/service/responses_websocket.rs @@ -1286,10 +1286,9 @@ async fn responses_websocket_should_prefer_conversation_account_without_previous let (_reused, first_payload, second_payload) = upstream.await.unwrap(); assert!(body.contains("\"id\":\"resp_conversation_affinity\"")); - assert!( - first_payload["prompt_cache_key"] - .as_str() - .is_some_and(|value| value.starts_with("wi_")) + assert_eq!( + first_payload["prompt_cache_key"], + "conv_conversation_affinity" ); assert_eq!( second_payload["prompt_cache_key"], @@ -1646,7 +1645,7 @@ async fn responses_websocket_response_failed_rate_limit_should_preserve_cooldown "status": "failed", "error": { "type": "rate_limit_error", - "code": "rate_limit_exceeded", + "code": "workspace_owner_usage_limit_reached", "message": "Rate limit reached. Please try again in 11.054s." } } @@ -1715,7 +1714,7 @@ async fn responses_websocket_response_failed_quota_should_retry_fallback_account "response": { "id": "resp_ws_quota_failed", "error": { - "code": "insufficient_quota", + "code": "workspace_owner_credits_depleted", "message": "quota exhausted" } } diff --git a/backend/tests/fleet/pool/runtime.rs b/backend/tests/fleet/pool/runtime.rs index 59680e6b8..46ec741af 100644 --- a/backend/tests/fleet/pool/runtime.rs +++ b/backend/tests/fleet/pool/runtime.rs @@ -1,11 +1,15 @@ -use std::sync::Arc; +use std::{collections::HashMap, sync::Arc}; -use chrono::Utc; +use chrono::{DateTime, Utc}; use codex_proxy_rs::fleet::{ + account::{Account, AccountStatus}, pool::{AccountAcquireRequest, AccountPoolOptions, AccountPoolService, RotationStrategy}, - store::{AccountStore, PgAccountStore}, + store::{AccountStore, AccountStoreResult, PgAccountStore}, +}; +use codex_proxy_rs::telemetry::account_usage::store::{ + AccountUsageDelta, AccountUsageSnapshot, AccountUsageStore, AccountUsageStoreError, + AccountUsageWindow, PgAccountUsageStore, }; -use codex_proxy_rs::telemetry::account_usage::store::PgAccountUsageStore; use crate::support::storage::init_test_db; @@ -81,6 +85,73 @@ async fn runtime_account_pool_should_restore_accounts_by_added_at_and_id() { assert_eq!(acquired.account.id, "acct_a"); } +#[tokio::test] +async fn upstream_status_should_exclude_account_before_database_write_completes() { + let store = Arc::new(BlockingAccountStore::new()); + let service = AccountPoolService::new( + Arc::clone(&store) as Arc, + Arc::new(NoopAccountUsageStore), + AccountPoolOptions::default(), + 0, + ); + service.restore_from_store().await.unwrap(); + + let updated = tokio::time::timeout( + std::time::Duration::from_millis(250), + service.set_status_immediately("acct_primary", AccountStatus::Banned), + ) + .await + .expect("runtime invalidation must not wait for PostgreSQL"); + store.wait_until_write_started().await; + let acquired = service + .acquire_with(&AccountAcquireRequest::new("gpt-5.5", Utc::now())) + .await + .expect("secondary account should remain available"); + + assert!(updated); + assert_eq!(acquired.account.id, "acct_secondary"); + + acquired.release_without_usage().await; + store.release_write(); + store + .wait_for_write(PersistedAccountState::Status(AccountStatus::Banned)) + .await; +} + +#[tokio::test] +async fn upstream_quota_should_exclude_account_before_database_write_completes() { + let store = Arc::new(BlockingAccountStore::new()); + let service = AccountPoolService::new( + Arc::clone(&store) as Arc, + Arc::new(NoopAccountUsageStore), + AccountPoolOptions::default(), + 0, + ); + service.restore_from_store().await.unwrap(); + let cooldown_until = Utc::now() + chrono::Duration::minutes(5); + + let updated = tokio::time::timeout( + std::time::Duration::from_millis(250), + service.mark_quota_limited_until_immediately("acct_primary", cooldown_until), + ) + .await + .expect("runtime quota invalidation must not wait for PostgreSQL"); + store.wait_until_write_started().await; + let acquired = service + .acquire_with(&AccountAcquireRequest::new("gpt-5.5", Utc::now())) + .await + .expect("secondary account should remain available"); + + assert!(updated); + assert_eq!(acquired.account.id, "acct_secondary"); + + acquired.release_without_usage().await; + store.release_write(); + store + .wait_for_write(PersistedAccountState::QuotaLimitedUntil(cooldown_until)) + .await; +} + #[tokio::test] async fn runtime_account_lease_drop_should_release_capacity() { let (pool, _dir) = init_test_db("runtime-account-pool-lease-drop").await; @@ -224,3 +295,157 @@ async fn insert_account(pool: &sqlx::PgPool, id: &str) { .await .expect("account should be inserted"); } + +#[derive(Clone, Debug, PartialEq, Eq)] +enum PersistedAccountState { + Status(AccountStatus), + QuotaLimitedUntil(DateTime), +} + +struct BlockingAccountStore { + accounts: Vec, + write_started: tokio::sync::Semaphore, + release_write: tokio::sync::Semaphore, + persisted: tokio::sync::Mutex>, +} + +impl BlockingAccountStore { + fn new() -> Self { + Self { + accounts: vec![ + crate::support::accounts::test_account("acct_primary", AccountStatus::Active), + crate::support::accounts::test_account("acct_secondary", AccountStatus::Active), + ], + write_started: tokio::sync::Semaphore::new(0), + release_write: tokio::sync::Semaphore::new(0), + persisted: tokio::sync::Mutex::new(Vec::new()), + } + } + + async fn persist(&self, state: PersistedAccountState) { + self.write_started.add_permits(1); + self.release_write + .acquire() + .await + .expect("test persistence gate should remain open") + .forget(); + self.persisted.lock().await.push(state); + } + + async fn wait_until_write_started(&self) { + tokio::time::timeout( + std::time::Duration::from_secs(1), + self.write_started.acquire(), + ) + .await + .expect("deferred persistence should start") + .expect("test persistence gate should remain open") + .forget(); + } + + fn release_write(&self) { + self.release_write.add_permits(1); + } + + async fn wait_for_write(&self, expected: PersistedAccountState) { + tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + if self.persisted.lock().await.contains(&expected) { + return; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("deferred account state should be persisted"); + } +} + +#[async_trait::async_trait] +impl AccountStore for BlockingAccountStore { + async fn list_pool_accounts(&self) -> AccountStoreResult> { + Ok(self.accounts.clone()) + } + + async fn mark_quota_limited_until( + &self, + _account_id: &str, + cooldown_until: DateTime, + ) -> AccountStoreResult { + self.persist(PersistedAccountState::QuotaLimitedUntil(cooldown_until)) + .await; + Ok(true) + } + + async fn set_cloudflare_cooldown_until( + &self, + _account_id: &str, + _cooldown_until: DateTime, + ) -> AccountStoreResult { + Ok(false) + } + + async fn set_status( + &self, + _account_id: &str, + status: AccountStatus, + ) -> AccountStoreResult { + self.persist(PersistedAccountState::Status(status)).await; + Ok(true) + } + + async fn get_quota_json(&self, _account_id: &str) -> AccountStoreResult> { + Ok(None) + } + + async fn apply_quota_snapshot( + &self, + _account_id: &str, + _quota_json: &str, + _limit_reached: bool, + _cooldown_until: Option>, + ) -> AccountStoreResult { + Ok(false) + } + + async fn sync_runtime_account_state(&self, _account: &Account) -> AccountStoreResult { + Ok(false) + } +} + +struct NoopAccountUsageStore; + +#[async_trait::async_trait] +impl AccountUsageStore for NoopAccountUsageStore { + async fn snapshots( + &self, + _account_ids: &[String], + ) -> Result, AccountUsageStoreError> { + Ok(HashMap::new()) + } + + async fn record_usage_delta( + &self, + _account_id: &str, + _usage: AccountUsageDelta, + ) -> Result<(), AccountUsageStoreError> { + Ok(()) + } + + async fn sync_runtime_window( + &self, + _account_id: &str, + _window: AccountUsageWindow, + ) -> Result<(), AccountUsageStoreError> { + Ok(()) + } + + async fn sync_rate_limit_window( + &self, + _account_id: &str, + _reset_at: DateTime, + _limit_window_seconds: Option, + ) -> Result<(), AccountUsageStoreError> { + Ok(()) + } +} diff --git a/backend/tests/support/config.rs b/backend/tests/support/config.rs index 2c01fb058..0485bb607 100644 --- a/backend/tests/support/config.rs +++ b/backend/tests/support/config.rs @@ -1,7 +1,7 @@ use codex_proxy_rs::bootstrap::config::{ AdminConfig, ApiConfig, AppConfig, AuthConfig, DatabaseConfig, FileLoggingConfig, LoggingConfig, QuotaConfig, RedisConfig, RuntimePathsConfig, ServerConfig, TelemetryConfig, - TlsConfig, WebSocketPoolSettings, + WebSocketPoolSettings, }; pub(crate) fn test_config(database_url: String) -> AppConfig { @@ -36,9 +36,6 @@ pub(crate) fn test_config(database_url: String) -> AppConfig { runtime: RuntimePathsConfig { data_directory: std::path::PathBuf::from(".runtime/test-data"), }, - tls: TlsConfig { - force_http11: false, - }, ws_pool: WebSocketPoolSettings { enabled: true, max_age_ms: 3_300_000, diff --git a/backend/tests/upstream/openai/transport/headers.rs b/backend/tests/upstream/openai/transport/headers.rs index 8c3c73a31..db0089b47 100644 --- a/backend/tests/upstream/openai/transport/headers.rs +++ b/backend/tests/upstream/openai/transport/headers.rs @@ -223,7 +223,7 @@ async fn websocket_execute_response_create_request_should_capture_handshake_head } #[tokio::test] -async fn codex_backend_client_websocket_should_forward_security_chain_headers_and_payload_fields() { +async fn codex_backend_client_websocket_should_forward_headers_and_preserve_payload_fields() { let received_headers = Arc::new(Mutex::new(Vec::<(String, String)>::new())); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); @@ -304,7 +304,6 @@ async fn codex_backend_client_websocket_should_forward_security_chain_headers_an installation_id: Some("install-123"), session_id: Some("cp_derived"), thread_id: None, - prompt_cache_key: Some("cp_derived"), client_request_id: None, turn_id: None, }, @@ -327,20 +326,14 @@ async fn codex_backend_client_websocket_should_forward_security_chain_headers_an ); let mut stable_metadata = metadata.clone(); stable_metadata.remove("x-codex-ws-stream-request-start-ms"); - assert_eq!(payload["prompt_cache_key"], "cp_derived"); - // 透明代理：客户端 client_metadata 的非字符串值（ignored_non_string: 42）原样保留， - // 代理字段只做追加式合并，不再过滤用户字段。 + assert_eq!(payload["prompt_cache_key"], "client-thread"); + // transport 只生成请求头，不把 context 中的身份或会话字段回填进 payload。 assert_eq!( Value::Object(stable_metadata), json!({ "safe": "yes", "x-openai-subagent": "review", - "ignored_non_string": 42, - "x-codex-installation-id": "install-123", - "session_id": "cp_derived", - "x-codex-window-id": "cw_derived", - "x-codex-turn-metadata": "{\"thread_source\":\"subagent\"}", - "x-codex-parent-thread-id": "parent-456" + "ignored_non_string": 42 }) ); @@ -468,7 +461,6 @@ async fn codex_backend_client_should_send_desktop_headers_and_capture_response_m installation_id: None, session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }, @@ -541,7 +533,6 @@ async fn codex_backend_client_should_use_latest_runtime_fingerprint_for_new_requ installation_id: None, session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }; @@ -620,7 +611,6 @@ async fn codex_backend_client_usage_should_use_wham_usage_headers() { installation_id: Some("install-1"), session_id: Some("session-1"), thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }) @@ -789,7 +779,6 @@ async fn codex_backend_client_should_send_http_sse_headers_in_fingerprint_order( installation_id: Some("install-1"), session_id: Some("session-1"), thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }, diff --git a/backend/tests/upstream/openai/transport/http_client.rs b/backend/tests/upstream/openai/transport/http_client.rs index 79d47123c..e4fef241b 100644 --- a/backend/tests/upstream/openai/transport/http_client.rs +++ b/backend/tests/upstream/openai/transport/http_client.rs @@ -135,7 +135,6 @@ async fn codex_backend_client_should_cap_non_success_error_body_at_one_mib() { installation_id: None, session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }, @@ -193,7 +192,6 @@ async fn codex_backend_client_should_parse_retry_after_from_rate_limit_error_bod installation_id: None, session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }, @@ -309,7 +307,6 @@ async fn codex_backend_client_should_capture_only_safe_response_metadata() { installation_id: None, session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }, diff --git a/backend/tests/upstream/openai/transport/mod.rs b/backend/tests/upstream/openai/transport/mod.rs index 64f9b8fce..0b792ba8e 100644 --- a/backend/tests/upstream/openai/transport/mod.rs +++ b/backend/tests/upstream/openai/transport/mod.rs @@ -178,7 +178,6 @@ pub(crate) fn request_context<'a>( installation_id: None, session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, } diff --git a/backend/tests/upstream/openai/transport/websocket.rs b/backend/tests/upstream/openai/transport/websocket.rs index 93bcebb60..47ea9a422 100644 --- a/backend/tests/upstream/openai/transport/websocket.rs +++ b/backend/tests/upstream/openai/transport/websocket.rs @@ -1690,7 +1690,6 @@ async fn codex_backend_client_should_use_websocket_when_previous_response_id_is_ installation_id: None, session_id: None, thread_id: None, - prompt_cache_key: None, client_request_id: None, turn_id: None, }, diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index e352e3c93..f511a779f 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -40,9 +40,6 @@ x-cpr: refresh_interval_minutes: 5 skip_exhausted: true - tls: - force_http11: false - ws_pool: enabled: true max_age_ms: 3300000 diff --git a/docs/architecture.md b/docs/architecture.md index babae3d6f..9df23f6e7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -457,18 +457,25 @@ previous response 按历史来源处理： Responses 协议没有按事件序号续传未完成 response 的请求字段。官方 Codex 在 turn 层持有完整会话历史和工具执行状态，收到可重试的断流后会建立新请求重新采样。代理保留这一责任边界：新请求仍进入账号池调度和会话恢复，但不会伪造同一个 response 的续传。 -## 账号作用域身份 +## 账号身份、安装身份与会话语义 -`identity_hmac_secret` 是实例级 256-bit 持久密钥。服务端用它对以下值做 HMAC 派生： +账号切换不能把所有带 `id` 的字段都一起改写。官方 Codex `5bed644` 将 `session_id`、`thread_id`、`turn_id`、`window_id`、parent/forked thread 共同建模为客户端 turn metadata；默认 `prompt_cache_key` 直接取 `session_id`，HTTP/WS 的 `x-client-request-id` 也使用 thread ID。它们表达客户端会话拓扑，不表达登录账号。盲目按账号 HMAC 会让一次换号在上游变成另一条会话，并破坏 prompt cache、父子线程和增量 WebSocket 语义。 -- prompt cache key -- session、thread、turn、window 和 parent thread ID -- client request ID -- installation ID +官方客户端把 `installation_id` 持久化在 Codex home 中，登录账号切换时不会自动变化。号池与单客户端不同：多个互不相关的上游账号共享一个代理 installation ID 会形成横向关联。因此工程有意采用更严格的隔离策略，用实例级 `identity_hmac_secret` 和账号 ID 稳定派生 UUID 形态的 installation ID。该偏离是号池风险隔离策略，不是对官方字段语义的误判。 -派生输入包含字段域和账号 ID。同一客户端标识在不同账号上得到不同值；同一账号、同一输入在密钥不变时保持稳定。installation ID 只由账号 ID 和持久密钥派生，因此每个账号有一个稳定值，不在每次请求时随机生成。 +| 字段或状态 | 语义 | 换号规则 | +| --- | --- | --- | +| access token、`chatgpt-account-id`、Cookie | 选中账号的认证身份 | 每个 attempt 必须从当前账号重建，禁止客户端值或上一账号值进入下个 attempt | +| `x-codex-installation-id`、turn metadata 的 `installation_id` | 安装/设备指纹 | 使用当前账号的稳定派生值；header、body、`client_metadata` 和嵌套 turn metadata 中已有投影必须一致 | +| `x-codex-turn-state` | 上游在同一 turn 内签发的 opaque sticky-routing 状态 | 同账号重试可复用；切换账号必须删除，不能伪造新值 | +| `previous_response_id` | 上游历史状态句柄 | 只在拥有该状态的账号上原样使用；代理持有完整 transcript 时可删除 ID 后全量重放，否则禁止跨账号尝试 | +| `session_id`、`thread_id`/`conversation_id`、`turn_id`、window、parent/forked thread | 客户端逻辑会话拓扑 | 原样透传，不做 HMAC、不按账号改写；transport 只可把同一原值投影到协议要求的 header | +| `prompt_cache_key`、`x-client-request-id` | 会话缓存命名空间和客户端请求关联 | 客户端提供时原样透传；缺失的 `x-client-request-id` 才回退代理 request ID | +| response/item `id`、`encrypted_content`、工具 `call_id` | 上游输出引用和工具配对语义 | 跨账号全量重放时删除上游 item `id` 与 `encrypted_content`，保留工具配对所需 `call_id` | -镜像更新或容器重建时必须保留数据目录中的 `identity_hmac_secret`。丢失该文件会为所有账号生成新的身份集合，已有会话亲和也失去连续性。 +`AccountScopedIdentity` 类型只包含 installation ID；会话字段不进入该类型。`AccountScopedRequest` 只有在当前账号 token/account/Cookie、installation ID 和 opaque turn state 边界处理完成后才能交给 transport。这样“账号身份要替换”和“会话语义要透传”不是依赖注释约定，而是由类型边界分开。 + +镜像更新或容器重建时必须保留数据目录中的 `identity_hmac_secret`。丢失该文件会为所有账号生成新的 installation ID，也会改变只在代理内部使用的本地 conversation HMAC 索引；客户端会话 ID 本身不受影响。 ## 数据存储 diff --git a/docs/request-lifecycle-controller-design.md b/docs/request-lifecycle-controller-design.md index 15ceb425b..7547fb5e4 100644 --- a/docs/request-lifecycle-controller-design.md +++ b/docs/request-lifecycle-controller-design.md @@ -1,6 +1,6 @@ # Responses 请求生命周期控制器设计 -状态：已实现。`v1/*` 已统一采用 Request → Attempt → Stream 三层洋葱生命周期；不存在新旧双路径或兼容层。 +状态：Request → Attempt → Stream 三层洋葱生命周期、明确账号事实的运行时秒级失效，以及单一 HTTP 自动协商路径已经实现；不存在新旧双路径或兼容层。本文“Responses WebSocket v2 与 HTTP/2 低延迟传输审计”一节中的预热、快路径预算、熔断与 HTTP/2 PING 仍是尚未落地的传输目标，不能把这些参数误认为当前生产行为。 本文回答一个具体问题：Responses 代理能否像 Koa 洋葱模型一样，让身份、历史恢复、账号恢复、会话策略、用量和遥测等模块只处理自己的入口与出口，并按固定顺序组合，而不把同一规则散落在请求过程中。 @@ -77,6 +77,7 @@ backend/src/dispatch/ │ ├── history.rs │ ├── quota.rs │ ├── account_failure.rs +│ ├── account_state.rs 运行时失效与 WS 驱逐的共享 effect 执行器 │ ├── cloudflare.rs │ ├── affinity.rs │ ├── usage.rs @@ -116,8 +117,8 @@ RequestLifecycle enter: normalize/model -> local identity -> history/session plan -> candidate snapshot AttemptLifecycle (每个候选账号一次) - enter: acquire -> quota verify -> history for account -> interval -> scoped identity -> call - leave: normalize observation -> classify -> account effects -> retry/fail decision -> trace + enter: acquire -> quota verify -> history for account -> interval -> account boundary -> call + leave: normalize observation -> classify -> runtime invalidation -> retry/fail decision -> trace StreamLifecycle (响应提交后) enter: canonical decoder -> observers -> downstream encoder @@ -193,6 +194,8 @@ controller 分为两类： - Critical：无法建立正确请求语义，转为 `DispatchFailure` 并触发逆序退出。 - BestEffort：记录警告并 fail-open，不阻塞正常请求。 +账号可用性 effect 属于第三种更窄的 `CriticalRoutingState`：它不能改写已经观察到的上游 outcome，但必须在返回换号 decision 前完成内存池失效和 WebSocket 池摘除。PostgreSQL 只是该事实的持久副本，必须后台写入，不能位于换号的等待链上。遥测、Cookie 和普通 usage 持久化仍是 BestEffort，不能与这条关键路径混为一谈。 + 同层 controller 不直接调用另一个 controller，也不自行选择下一账号。lifecycle 是唯一解释 decision 的组件。这样模块只拥有规则，编排器只拥有顺序和控制流。 ## 固定顺序 @@ -212,7 +215,7 @@ controller 分为两类： 2. 必要时验证 quota。 3. 为该账号准备 history；不能跨账号时返回 typed failure。 4. 等待账号请求间隔。 -5. 生成账号作用域 identity、Cookie 和上游上下文。 +5. 从当前账号生成 token/account/Cookie 和稳定 installation ID；会话语义 ID 保持客户端原值。 6. 发起上游调用。 7. 统一规范化 HTTP、SSE 首帧和 WebSocket 错误为 `AttemptObservation`。 8. 按唯一优先级分类 history、session policy、账号隔离和 transport retry。 @@ -220,6 +223,52 @@ controller 分为两类： classifier 优先级只能存在一处。complete、prefetch 与 committed stream 共同经过同一个 feature failure owner classifier；`cyber_policy` 必须先于通用 403、quota 或 5xx 分类。composition root 只能把不透明的分类结果交还 owner 产生 decision，其他模块不能再次调用 `is_cyber_policy_*` 或自行拼装 Cyber retry。 +### 账号边界与会话 ID 边界 + +官方 Codex 源码将 identifier 分为三种完全不同的状态，lifecycle 不能用一个“账号作用域 identity”概念把它们混在一起： + +| 类别 | 字段 | Attempt 规则 | +| --- | --- | --- | +| 当前账号认证/指纹 | token、`chatgpt-account-id`、Cookie、installation ID | 每次按当前账号重建；installation ID 按账号稳定派生 | +| 上游 opaque 状态 | `x-codex-turn-state`、`previous_response_id` | turn state 换号清除；previous ID 只留在原 owner，或在有完整 transcript 时删除并全量重放 | +| 客户端会话语义 | session/thread/conversation、turn、window、parent/forked thread、`prompt_cache_key`、`x-client-request-id` | 原样透传；不得 HMAC、随机化或随账号改写 | + +官方客户端将 installation ID 持久化为安装级 UUID，所以它不是登录账号 ID；本工程按账号稳定派生是号池防关联策略。相反，官方 `prompt_cache_key` 默认取 session ID，`x-client-request-id` 使用 thread ID，证明它们属于会话连续性。代理可以把客户端原值投影到 HTTP/WS header，但不能生成第二份账号作用域值。 + +`AccountScopedRequest` 是 attempt 到 transport 的提交能力：只有当前账号认证、installation ID、Cookie 和 opaque state 已处理的请求才能调用上游。`AccountScopedIdentity` 被刻意收窄到 installation ID，不保存任何 session/thread/turn/window 值；换号时编译期可见的身份处理入口因此只有一个。 + +### 明确账号事实与秒切号边界 + +数据库中的 `active` 只是上一次持久化快照，不是当前账号可用性的最终真相。请求已经选中账号后，上游仍可能返回封禁、工作区停用、token 失效、额度耗尽或窗口限流。处理顺序固定为： + +```text +typed upstream fact + -> feature owner 唯一分类 + -> 持有账号池 mutex 时线性化内存失效并移除现有 slots + -> 将同一状态写入有序 PostgreSQL 后台队列 + -> 从 WebSocket 池摘除该账号的 idle/busy entries + -> OpenAttempt 返回 RetryNextCandidate +``` + +任何并发请求即使已经冻结了包含该账号的 candidate snapshot，在真正 `acquire_candidate` 时也必须重读运行时状态，因此内存失效完成后不能再租到旧账号。数据库连接池拥塞、行锁或写失败只能影响持久化收敛，不能让当前请求等待，也不能让其他请求继续看到旧 `Active`。账号运行状态后台写入按产生顺序消费，避免同一进程内的失败事实乱序落库。 + +账号事实矩阵来自当前工程、官方 Codex error/rate-limit 类型以及 `/home/zyy/桌面/Codes/sub2api` 的 OpenAI failover 实现交叉审计；只吸收明确的 OpenAI/Codex 信号，不复制 sub2api 的多平台、自定义状态码或连续次数产品策略： + +| 上游事实 | owner 与运行时 effect | 当前 attempt | +| --- | --- | --- | +| `account_deactivated`、`deactivated_workspace`、明确 banned/suspended/organization disabled | `account_failure` → `Banned` + WS 驱逐 | 立即下一账号 | +| `identity_verification_required` | `account_failure` → `Disabled` + WS 驱逐 | 立即下一账号 | +| 401、`token_invalidated`、`token_revoked`、`refresh_token_invalidated`、`invalid_api_key`、`authentication_error` | `account_failure` → `Expired` + WS 驱逐 | 立即下一账号 | +| `insufficient_quota`、`payment_required`、`workspace_*_credits_depleted` 或普通 402 | `quota` → `QuotaExhausted` + WS 驱逐 | 立即下一账号 | +| 429、`rate_limit_exceeded`、`usage_limit_reached`、`workspace_*_usage_limit_reached` | `quota` → 带 reset/retry 时间的运行时限流 + WS 驱逐 | 立即下一账号 | +| model unavailable/not supported，包括明确的 404 | `account_failure` 只记录 `ModelUnsupported`，不失效整账号 | 立即下一账号 | +| connect/timeout、529、5xx | 通用 transport retry；不写账号状态 | 有候选时下一账号 | +| Cyber、Cloudflare HTML challenge、previous-response failure | 各自 controller/history owner | 不得被通用 403/404 抢占 | + +同一失败在 HTTP 状态响应、SSE/WS `response.failed` 和 `error` 事件中都消费相同的 `code`、`type`、status、message 与 retry typed facts。transport 只能提取这些字段，不能决定账号状态。普通 `403` 在 Cyber、Cloudflare 和 model classifier 都未命中后才作为账号禁止事实处理。 + +`CommittedAttempt` 类型上没有 retry API：如果失败在客户端响应已经提交后才到达，只能在其他 BestEffort finalizer I/O 之前执行同样的内存失效、后台持久化和 WS 驱逐，让后续请求立即避开该账号；绝不能把另一账号的输出拼到当前流。 + ### Stream enter/leave 1. 上游 SSE 或 WebSocket 事件转换为一次性的 `CanonicalResponseChunk` / `CanonicalResponseEvent`。 @@ -249,6 +298,7 @@ classifier 优先级只能存在一处。complete、prefetch 与 committed strea - complete 与 stream 入口共享 `enter_request` 和 `AttemptRunner`，不存在各自维护的账号尝试主循环。 - `ControllerSet` 是静态 composition root；controller 顺序和 typed hook 在编译期可见，不使用动态 middleware 链。 - complete、prefetch 与 committed stream 的公共失败 owner 优先级只在 `ControllerSet::classify_shared_failure` 声明一次；composition root 不实现 Cyber、quota 或账号失败规则。 +- ban、deactivated workspace、认证失效与 quota owner 共享同一 `account_state` effect 执行器：内存状态先线性化，PostgreSQL 有序后台持久化，WS entry 随即驱逐；controller 不直接等待数据库。 - Cyber failure 由 `CyberPolicyController` 返回不透明分类，并由该 owner 唯一构造换号或返回 decision；`ControllerSet` 只分发 effect 与 decision。 - `ControllerRequestScope` 只保存 cyber、history、usage 的真实状态；不存在通用 entered-controller marker、容量错误或空 unwind hook。 - complete 的 accepted response 归一为 `FinalOutcome` 后由 `finalize_complete` 消费 scope；retry 继续进入下一 attempt，rejection 在 controller effect 已完成后直接返回，不经过空 finalize。 @@ -260,10 +310,182 @@ classifier 优先级只能存在一处。complete、prefetch 与 committed strea - 上游 WebSocket pool 保持独立的连接 lease 与回收职责，dispatch lifecycle 只消费其 typed transport outcome。 - 测试全部位于 `backend/tests/`；`backend/src/` 中不存在 `#[cfg(test)]`、`#[test]`、`mod tests` 或 test-only helper。 +## Responses WebSocket v2 与 HTTP/2 低延迟传输审计 + +状态：审计完成，目标方案已确定，尚未实现。 + +本节解决的不是“WebSocket 和 HTTP/SSE 谁在理论上更快”，而是两个直接影响体感的问题：如何让健康连接尽量在请求到来前准备好，以及当 WebSocket 建连异常时，如何避免在换账号或 HTTP fallback 之前白等 15 秒。 + +最终结论是双热通道：精确会话 WebSocket 负责 v2 的连接本地状态和增量续接；按 origin 共享的 HTTP/2 连接负责新链和可持久化续链的低延迟后备。两者不能互相冒充，也不能用一个通用“连接可用”布尔值抹掉 continuation 的所有权差异。 + +### 审计证据与事实边界 + +以下结论来自 2026-07-14 的四类证据，必须区分来源： + +1. 官方 Codex 源码确认：本地官方仓库当时的 `main` 为 `5bed6447998c754d154dbd796517310b8f04d4ce`。OpenAI provider 默认声明支持 WebSocket；旧 `responses_websockets` / `responses_websockets_v2` feature 已标记 Removed，实际传输路径不再依赖这两个开关；握手固定发送 `OpenAI-Beta: responses_websockets=2026-02-06`。 +2. 官方 Codex 源码确认：会话初始化后会在后台构造稳定的 instructions、tools 和空 input，发送 v2 `response.create` 且设置 `generate=false`，等待 `response.completed` 后把连接和 response ID 交给首个真实 turn。首个 turn 只有在请求属性仍匹配时才使用 warmup response ID；模型、service tier、tools 等属性变化时仍可复用 socket，但必须发送新的完整 create。 +3. OpenAI 官方文档确认：WebSocket Mode 在活跃 socket 上只保存最新一个 previous-response 的连接本地内存状态；`store=false` / ZDR 下没有持久化 fallback，ID 不在该 socket 缓存时返回 `previous_response_not_found`；单条连接一次只允许一个 in-flight response，不支持 multiplex，连接最长 60 分钟。 +4. 当前工程源码确认：已经透明透传 `generate=false`，发送 v2 beta header，按 `(base_url, account_id, conversation_id)` 建池，并在 socket 上记录和校验最新 response ID；池内连接默认 25 秒 ping、5 秒探活、55 分钟最大寿命。 +5. 当前工程源码确认：reqwest Client 已按自定义 CA profile 全局缓存，HTTP/2 可以跨账号共享同一 origin 的连接池；强制 HTTP/1.1 的配置和分支已经删除，但尚未配置 HTTP/2 PING，reqwest 0.12.28 默认会在 90 秒后清理 idle pool。 +6. 线上与网络探针只能用于校准，不构成协议保证：此前当日生产失败样本中，59 次 WebSocket 超时有 56 次被同账号 HTTP 恢复；OCI 未鉴权探针中，HTTP/2 冷连接完成 TLS 约 32–35ms，WebSocket upgrade 响应约 70–591ms。该探针不包含真实鉴权和模型 TTFT，只说明健康握手通常不需要 15 秒。 + +官方依据： + +- [Codex `client.rs` 的 prewarm 契约](https://github.com/openai/codex/blob/5bed6447998c754d154dbd796517310b8f04d4ce/codex-rs/core/src/client.rs#L15-L24) +- [Codex 会话启动预热](https://github.com/openai/codex/blob/5bed6447998c754d154dbd796517310b8f04d4ce/codex-rs/core/src/session_startup_prewarm.rs#L183-L224) +- [Codex `generate=false` 请求构造与等待完成](https://github.com/openai/codex/blob/5bed6447998c754d154dbd796517310b8f04d4ce/codex-rs/core/src/session_startup_prewarm.rs#L241-L324) +- [OpenAI Responses API WebSocket Mode](https://developers.openai.com/api/docs/guides/websocket-mode) +- [reqwest 0.12.28 HTTP/2 keep-alive 配置](https://docs.rs/reqwest/0.12.28/reqwest/struct.ClientBuilder.html#method.http2_keep_alive_interval) + +### WebSocket v2 的收益边界 + +WebSocket v2 的主要收益不是帧格式比 HTTP/SSE 更快，而是同时消除三类重复工作： + +- socket 持续存在，避免每轮重新建立 TCP/TLS/HTTP 请求通道；健康 HTTP/2 keep-alive 也能覆盖这一部分。 +- 客户端只发送新增 input 和 `previous_response_id`，不重复上传完整 history、instructions 和 tools。 +- 上游直接复用当前 socket 上的 connection-local previous-response 状态；HTTP/2 连接复用不能提供这份模型请求状态。 + +因此，热 HTTP/2 可以让 fallback 的网络建连接近无感，但不能替代 `store=false` WebSocket 链。OpenAI 官方文档给出的典型收益场景是长工具链，并报告 20 次以上工具调用的 rollout 最高约 40% 端到端加速；这不是所有单轮请求都能获得的固定收益。 + +### 当前实现缺口 + +当前实现已经具备 v2 透传和池化基础，但还有四个会影响正确性或尾延迟的缺口： + +1. `upstream/openai/transport/client_sse.rs` 只根据是否携带 previous ID 判断 fallback。首个 `generate=false` 没有 previous ID，因此上游 WebSocket 失败后仍可能回退 HTTP。 +2. `controllers/affinity.rs` 只根据 `store` 记录 continuation scope。HTTP fallback 产生的 `store=false` response 也会被记为 `ConnectionLocal`，但实际没有任何 WebSocket 持有该 ID。 +3. `WebSocketPreferred` 与 `WebSocketRequired` 的冷建连都可能在前台等待完整 15 秒；池只有 Idle / Busy 等连接结果，没有复用中的单飞 Connecting 状态和短快路径预算。 +4. HTTP Client 虽然全局缓存并默认协商 HTTP/2，但 90 秒 idle pool timeout 会在长期只走 WebSocket 时主动清掉健康后备连接；TCP keepalive 不能阻止连接池自己驱逐 socket。 + +第 1、2 点会形成错误链： + +```text +客户端后台发送 generate=false + -> 上游 WebSocket 失败 + -> 代理降级 HTTP，得到 store=false response_id + -> 客户端下一轮携带 previous_response_id + -> 新 WebSocket 没有该 connection-local 状态 + -> previous_response_not_found 或被迫全量重放 +``` + +目标实现必须禁止这条链，而不是再增加一次重试掩盖它。 + +### 最终传输状态机 + +请求首先被规范化为显式 transport requirement，不继续靠 `use_websocket`、`force_http_sse` 和 `previous_response_id.is_some()` 的组合隐式推断： + +| 请求事实 | 传输要求 | 允许行为 | +| --- | --- | --- | +| 客户端显式 `generate=false` 且 `store=false` | Explicit WebSocket Warmup | 必须使用 WebSocket；允许后台完整建连预算；禁止 HTTP fallback | +| `store=false` connection-local previous ID | Exact WebSocket Continuation | 只能租用持有同一最新 ID 的精确 socket；缺失时立即返回 typed unavailable，由 HistoryController 决定全量重放或失败 | +| 已确认 persisted previous ID | Persisted Continuation | 热 WebSocket 优先；没有热 socket 时允许使用 HTTP/2 或新 WebSocket hydration | +| 外部来源且所有权未知的 previous ID | External Unknown | 只在已选定账号原样尝试一次，不遍历账号碰运气；失败后返回观察到的协议错误 | +| 没有 previous ID 的普通新链 | New Chain | 精确热 WebSocket 优先；冷建连只等快路径预算，超时后同账号走热 HTTP/2 | +| 客户端显式要求 HTTP | HTTP Required | 直接使用 HTTP，不启动 WebSocket 冷建连 | + +response 完成后的 continuation scope 必须同时由 `store` 和实际 transport 推导：`store=true` 是 Persisted；`store=false + WebSocket completed` 才是 ConnectionLocal；`store=false + HTTP` 必须是 ReplayRequired 或 Unavailable，绝不能伪装成 ConnectionLocal。只有入站 WebSocket 已持有完整 canonical transcript 时才能标记 ReplayRequired，普通 HTTP 客户端没有本地 replay snapshot 时只能标记 Unavailable。 + +普通新链的目标时序： + +```text +获取账号 lease + -> controller 完成账号与 route 准备 + -> 生成当前账号凭据、installation ID / Cookie / handshake 输入 + -> 启动精确 WebSocket 单飞建连 + 与 wait_for_request_interval 并行 + -> 已有热 socket 或 800ms 内完成：发送 response.create + -> 800ms 未完成：取消尚未发送 payload 的冷连接，同账号立即走热 HTTP/2 + -> HTTP/2 也失败或返回明确可换号失败：Attempt lifecycle 选择下一候选账号 +``` + +800ms 是初始快路径预算，不是永久硬编码真理。它来自当前 OCI 探针的健康区间并为鉴权波动保留余量，必须根据 `ws_connect_ms` 的生产 p95/p99 调整。无论以后调成多少，该预算都只能覆盖发送 payload 之前的握手阶段。 + +一旦 `response.create` 已可能送达上游，连接断开或首事件超时就是“是否已经开始推理未知”的 ambiguous failure。没有上游幂等保证时，不能为了缩短体感而并发发送同一请求到 HTTP 或其他账号，否则可能重复推理、重复扣费或产生两条不一致输出。现有 20 秒 initial-event guard 与 5 分钟 active-stream idle timeout 必须继续和 pre-send connect budget 分开建模。 + +### 官方客户端预热与代理预热 + +官方 Codex 客户端已经提供标准预热信号。代理收到 `generate=false` 后应当： + +1. 正常进入 Request / Attempt 洋葱并选定账号，不绕过 quota、history 或账号隔离规则。 +2. 在该账号、base URL 和本地 conversation identity 下建立 WebSocket。 +3. 原样发送 `response.create(generate=false)`，等待 `response.completed`。 +4. 将 socket、最新 warmup response ID 和连接元数据作为一个不可拆分的池条目保存。 +5. 首个真实请求只有在 previous ID 和请求属性都满足续接条件时，才在该 socket 上发送增量 input。 + +代理不得给所有账号盲目建立“空 WebSocket”：握手携带账号凭据、session/thread 会话语义、Cookie 和路由信息，空连接不能安全跨账号或跨 conversation 借用；单 socket 不支持 multiplex，批量空连接还会提前消耗上游连接额度。对不发送标准 warmup 的客户端，只允许在候选账号和会话语义已经确定后做 handshake-only 单飞建连，不得由代理擅自补发 `generate=false`，因为后者可能额外处理完整上下文并引入隐藏成本、状态映射和并发竞态。 + +### HTTP/2 热后备 + +HTTP/2 连接按 `(scheme, host, port, proxy/TLS profile)` 共享，不按账号建池；Authorization 和账号头属于单个 HTTP/2 stream，同一 origin 的多个账号可以复用一条 TLS/H2 连接。这正适合作为 WebSocket 冷建连失败后的廉价后备。 + +工程不再暴露 `tls.force_http11`，也不保留 `http1_only()` 兼容分支；旧 `tls` 配置节点会被 `deny_unknown_fields` 直接拒绝。上游 HTTP 只有一条自动 ALPN 协商路径，主动连接保活优化只针对 HTTP/2，不提供运行时切回“强制 HTTP/1.1”的第二套策略。目标 Client 配置为： + +```rust +Client::builder() + .pool_idle_timeout(None::) + .http2_keep_alive_interval(Duration::from_secs(30)) + .http2_keep_alive_timeout(Duration::from_secs(5)) + .http2_keep_alive_while_idle(true) +``` + +有可用账号时，现有启动模型目录刷新已经会在服务启动后建立同 origin HTTP 连接，不需要新增一条“为了预热而预热”的业务请求。目标配置的目的只是让该连接在 WebSocket-only 流量期间不被本地 90 秒 idle timeout 清理，并用 HTTP/2 PING 在后台发现已经失活的链路。 + +### 熔断与账号边界 + +WebSocket 建连失败通常是 origin、出口网络、代理或 Cloudflare 边缘链路事实，不应直接把账号标为失效。此前生产样本中绝大多数超时能被同账号 HTTP 恢复，因此顺序必须是“同账号换 transport，再由 HTTP 结果决定是否换账号”，而不是“换账号后再重复一次同样的 WebSocket 超时”。 + +初始熔断规则： + +- key 为 origin + 出口 route/TLS profile，不按账号 key。 +- 30 秒窗口内连续 3 次 WebSocket 快路径超时，打开 30 秒。 +- 熔断期间已有热 WebSocket 仍可使用；普通新链直接走 HTTP/2，不再创建冷 WebSocket。 +- 30 秒后只允许一个 half-open 单飞探针；成功关闭熔断，失败重新打开。 +- 401、403、quota、账号停用和套餐限制仍由对应 controller 处理，不能污染 transport 熔断。 +- `previous_response_not_found` 只驱逐精确 conversation socket 并交给 HistoryController，不打开 origin 熔断。 + +这些时间和次数属于 transport policy，只能由 `upstream/openai/transport` 单一 owner 定义和观测；lifecycle 只能消费 `ReusedWebSocket`、`ConnectedWebSocket`、`Http2Fallback`、`ContinuationUnavailable`、`PreSendFailure`、`PostSendAmbiguousFailure` 等 typed outcome。 + +### 洋葱边界下的代码归属 + +| 责任 | 唯一归属 | 禁止越界 | +| --- | --- | --- | +| v2 payload、握手、快路径预算、单飞、连接池、origin 熔断 | `upstream/openai/transport/` | lifecycle/controller 不解析 WebSocket 错误字符串，不维护连接状态 | +| `store` + 实际 transport 推导 continuation scope | upstream 产出事实，`controllers/history.rs` 解释所有权与重放 | affinity 不能只看 `store` 猜测 scope | +| 账号选择、retry next、提交边界 | `dispatch/lifecycle/` | transport 不选择下一个账号，不写账号状态 | +| history full replay、previous ID owner、跨账号可恢复性 | `controllers/history.rs` | API 和 WebSocket pool 不自行拼接历史 | +| quota、封禁、Cyber 和账号状态 | 各自 classifier controller；`account_state` 只执行通用失效 effect | transport connect failure 不能直接停用账号，effect 执行器不能识别错误语义 | +| HTTP/WS 入站解析和最终事件编码 | `api/client/responses/` | API 不决定 upstream transport fallback | + +Attempt lifecycle 只增加一个窄的 transport preparation 调用点：在账号、route、identity 和 Cookie 已确定后启动连接准备，并与账号请求间隔并行；它接收准备结果，不拥有 800ms、30 秒或熔断阈值。这样低延迟优化不会把 WebSocket 规则扩散进 `attempt.rs`，也不会形成第二套 retry 主循环。 + +### 可观测性与验收 + +必须分别记录选择账号、连接准备、发送请求、首协议事件和首真实输出，不能只用一个总耗时掩盖慢点： + +- `transport_decision`：`ws_reused`、`ws_connected_fast`、`http2_ws_slow`、`http2_breaker_open`、`ws_exact_required`。 +- `ws_connect_ms`、`transport_decision_wait_ms`、`upstream_headers_ms`、`first_event_ms`、`first_output_ms`。 +- `http_version`、origin breaker state、WebSocket pool decision、connection reused。 +- pre-send failure 与 post-send ambiguous failure 必须是不同枚举和不同指标。 +- `generate=false` warmup 成功率、warmup 被首轮使用率、请求属性变化导致的 warmup miss。 + +完成实现必须同时满足： + +- `generate=false` 的 `store=false` 请求在 WebSocket 失败后绝不回退 HTTP，也不记录虚假的 ConnectionLocal scope。 +- warmup 与首个真实 turn 在属性匹配时使用同一 socket，首轮携带 warmup response ID 和增量 input。 +- 精确 connection-local socket 缺失时不冷建连碰运气，不遍历账号；只走 HistoryController 的 typed replay/failure。 +- 普通 `WebSocketPreferred` 冷建连不会在前台等待 15 秒；健康快路径预算耗尽后同账号 HTTP/2 立即接管。 +- WebSocket transport failure 不更新账号失效状态；只有后续明确的账号事实才能触发账号 controller。 +- 明确账号事实发生后，即使 PostgreSQL 行写入被阻塞，当前 attempt 仍立即切到下一账号，并发候选租用也不能再次获得旧账号；解除阻塞后状态最终落库。 +- 账号失效必须同时驱逐该账号的 WebSocket entries；已提交流只能影响后续请求，不能构造 retry。 +- origin 熔断期间没有冷 WebSocket 建连风暴，已有热连接不受影响。 +- payload 发送后的 ambiguous failure 不自动并发 hedge 到第二 transport 或账号。 +- HTTP/2 连接在超过 90 秒的 WebSocket-only 空闲窗口后仍可复用，并能在 PING timeout 后清理失活连接。 +- 所有测试位于 `backend/tests/`，按 transport、history controller 和 lifecycle contract 边界组织，`backend/src/` 不出现 test-only 代码。 + ## 性能约束 - 独立且都需要的入口查询并行执行，不能把多个 Redis RTT 串行叠加。 - best-effort controller 必须有有界超时和明确 fail-open 行为。 +- 账号可用性内存失效和 WS 池摘除不是数据库 BestEffort I/O；数据库持久化只能进入后台队列，不能占用换号延迟预算。 - 逐事件 observer 只做同步、低成本状态更新；Redis、数据库和遥测 I/O 只在必要的 terminal/finalize 阶段执行。 - canonical event 同时保留 raw payload，避免为透明转发再次序列化。 - 是否引入动态分发以基准和可维护性为依据，不凭感觉优化。