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 <noreply@openai.com>

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<String, Value>, key: &str) -> Option<String> {
+fn session_context_string(body: &Map<String, Value>, key: &str) -> Option<String> {
     body_context_string(body, key).or_else(|| {
         body.get("client_metadata")
             .and_then(Value::as_object)

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,

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(),
             )

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<String>,
-    pub session_id: Option<String>,
-    pub thread_id: Option<String>,
-    pub client_request_id: String,
-    pub turn_id: Option<String>,
-    pub window_id: Option<String>,
-    pub parent_thread_id: Option<String>,
+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<String>,
+}
+
+impl AccountIdentityScope {
+    pub(in crate::dispatch) fn new(preferred_account_id: Option<String>) -> 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<String> {
-        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::<Value>(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<String> {
+    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<Value> {
+    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<String, Value>,
+    key: &str,
+    value: Option<&str>,
+) {
+    if metadata.contains_key(key) {
+        replace_metadata_field(metadata, key, value);
+    }
+}
+
+fn replace_metadata_field(metadata: &mut Map<String, Value>, 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);
+        }
+    }
+}

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::*;

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<AccountStateEffect>,
 }
 
 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<Classi
                     ExhaustedAccountKind::UpstreamUnavailable,
                     "upstream response did not include visible output",
                 ),
-                effect: AccountEffect::None,
+                effect: None,
             })
         }
         _ => ControllerFailureFact::from_attempt(observation)
@@ -168,108 +170,151 @@ fn classify_failure(
 fn classify_upstream(account_id: &str, facts: &UpstreamFailureFacts) -> Option<ClassifiedFailure> {
     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<ClassifiedFailure> {
+    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<u16>,
+) -> 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<ClassifiedFailure> {
-    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<AccountStatus> {
+    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("<!doctype") || value.starts_with("<html") || value.contains("<html")
 }
-
-fn is_deactivated_workspace(value: &str) -> bool {
-    serde_json::from_str::<serde_json::Value>(value).is_ok_and(|value| {
-        value
-            .pointer("/detail/code")
-            .and_then(serde_json::Value::as_str)
-            == Some("deactivated_workspace")
-    })
-}

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

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<SessionAffinityService>,
     account_pool: Arc<AccountPoolService>,
+    codex: Arc<CodexBackendClient>,
     cloudflare: CloudflareRecovery,
     recorder: Arc<Recorder>,
 }
@@ -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<CodexResponsesRequest>),
+    Ready(Box<AccountScopedRequest>),
     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<crate::dispatch::affinity::SessionAffinityService>,
         account_pool: Arc<AccountPoolService>,
+        codex: Arc<CodexBackendClient>,
         cloudflare: CloudflareRecovery,
         recorder: Arc<Recorder>,
     ) -> 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) => {

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<Utc>),
-}
-
 pub(super) struct ClassifiedQuotaFailure {
     exhaustion: AccountExhaustionRecord,
-    effect: QuotaEffect,
+    effect: Option<AccountStateEffect>,
 }
 
 fn classify(observation: &AttemptObservation) -> Option<ClassifiedQuotaFailure> {
@@ -219,29 +201,37 @@ fn classify_upstream(
     account_id: &str,
     facts: &UpstreamFailureFacts,
 ) -> Option<ClassifiedQuotaFailure> {
-    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<ClassifiedQuotaFailure> {
-    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<StreamQuotaKind> {
         "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,
+        )),
     }
 }
 

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;

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

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

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<CodexBackendResponse, CodexClientError> {
-    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<CodexBackendStreamingResponse, CodexClientError> {
-    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(),
+    }
+}

backend/src/dispatch/transport/observation.rs

@@ -28,6 +28,7 @@ pub(in crate::dispatch) struct UpstreamFailureFacts {
     pub kind: UpstreamFailureKind,
     pub status_code: Option<u16>,
     pub code: Option<String>,
+    pub error_type: Option<String>,
     pub message: String,
     pub body: String,
     pub retry_after_seconds: Option<u64>,
@@ -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<String>, Option<String>) {
+fn error_fields(body: &str) -> (Option<String>, Option<String>, Option<String>) {
     let Ok(value) = serde_json::from_str::<Value>(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<String>, Option<String>) {
             .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"))

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,
         };

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,
             };

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,
         };

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<dyn AccountUsageStore>,
     request_interval: Arc<std::sync::RwLock<StdDuration>>,
     candidate_changed: Arc<tokio::sync::Notify>,
+    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<Utc>) -> 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<Utc>,
+    ) -> 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<AccountLease> {
         let refresh = self.pool.lock().await.acquire_with_status_refresh(request);

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<AccountStateWrite>,
+}
+
+enum AccountStateWrite {
+    SetStatus {
+        account_id: String,
+        status: AccountStatus,
+    },
+    MarkQuotaLimitedUntil {
+        account_id: String,
+        until: DateTime<Utc>,
+    },
+}
+
+impl AccountStatePersistence {
+    pub(super) fn new(store: Arc<dyn AccountStore>) -> 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<Utc>) {
+        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<AccountStateWrite>,
+    store: Arc<dyn AccountStore>,
+) {
+    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"
+                    ),
+                }
+            }
+        }
+    }
+}

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,
                 })

backend/src/upstream/openai/protocol/events.rs

@@ -520,7 +520,14 @@ fn retry_after_seconds_from_rate_limit_message(error: &Value) -> Option<u64> {
         .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)?;

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()) {

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<String>);
+type ReqwestClientCacheKey = Option<String>;
 type ReqwestClientCache = Mutex<HashMap<ReqwestClientCacheKey, Client>>;
 
-/// 构建带缓存的 reqwest Client。若 `force_http11` 为 true 则强制 HTTP/1.1。
-pub fn build_reqwest_client(force_http11: bool) -> Result<Client, CustomCaError> {
-    let cache_key = (force_http11, custom_ca_env_cache_key());
+/// 构建带缓存、自动协商 HTTP/2 的 reqwest Client。
+pub fn build_reqwest_client() -> Result<Client, CustomCaError> {
+    let cache_key = custom_ca_env_cache_key();
     static CLIENTS: OnceLock<ReqwestClientCache> = 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<Client, CustomCaError>
         .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<Value> {
-    // 以客户端原始 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<String, Value>, 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<String, Value>, 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<String> {
     let value = client_metadata?
         .as_object()?

backend/src/upstream/openai/transport/client_sse.rs

@@ -167,7 +167,7 @@ impl CodexBackendClient {
         pool_account_id: Option<&str>,
         started_at: Instant,
     ) -> CodexClientResult<CodexBackendResponse> {
-        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<CodexBackendStreamingResponse> {
-        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)

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

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

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

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::<Vec<_>>();
+    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;

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"
                     }
                 }

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<dyn AccountStore>,
+        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<dyn AccountStore>,
+        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<Utc>),
+}
+
+struct BlockingAccountStore {
+    accounts: Vec<Account>,
+    write_started: tokio::sync::Semaphore,
+    release_write: tokio::sync::Semaphore,
+    persisted: tokio::sync::Mutex<Vec<PersistedAccountState>>,
+}
+
+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<Vec<Account>> {
+        Ok(self.accounts.clone())
+    }
+
+    async fn mark_quota_limited_until(
+        &self,
+        _account_id: &str,
+        cooldown_until: DateTime<Utc>,
+    ) -> AccountStoreResult<bool> {
+        self.persist(PersistedAccountState::QuotaLimitedUntil(cooldown_until))
+            .await;
+        Ok(true)
+    }
+
+    async fn set_cloudflare_cooldown_until(
+        &self,
+        _account_id: &str,
+        _cooldown_until: DateTime<Utc>,
+    ) -> AccountStoreResult<bool> {
+        Ok(false)
+    }
+
+    async fn set_status(
+        &self,
+        _account_id: &str,
+        status: AccountStatus,
+    ) -> AccountStoreResult<bool> {
+        self.persist(PersistedAccountState::Status(status)).await;
+        Ok(true)
+    }
+
+    async fn get_quota_json(&self, _account_id: &str) -> AccountStoreResult<Option<String>> {
+        Ok(None)
+    }
+
+    async fn apply_quota_snapshot(
+        &self,
+        _account_id: &str,
+        _quota_json: &str,
+        _limit_reached: bool,
+        _cooldown_until: Option<DateTime<Utc>>,
+    ) -> AccountStoreResult<bool> {
+        Ok(false)
+    }
+
+    async fn sync_runtime_account_state(&self, _account: &Account) -> AccountStoreResult<bool> {
+        Ok(false)
+    }
+}
+
+struct NoopAccountUsageStore;
+
+#[async_trait::async_trait]
+impl AccountUsageStore for NoopAccountUsageStore {
+    async fn snapshots(
+        &self,
+        _account_ids: &[String],
+    ) -> Result<HashMap<String, AccountUsageSnapshot>, 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<Utc>,
+        _limit_window_seconds: Option<u64>,
+    ) -> Result<(), AccountUsageStoreError> {
+        Ok(())
+    }
+}

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,

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,
             },

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,
             },

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,
     }

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,
             },

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

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 本身不受影响。
 
 ## 数据存储
 

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::<Duration>)
+    .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，避免为透明转发再次序列化。
 - 是否引入动态分发以基准和可维护性为依据，不凭感觉优化。