fix(openai): preserve affinity while waiting for account capacity (#297)

Classify saturated account snapshots as capacity unavailable and prefer the bound account's existing queue when account waiting is enabled. Preserve explicit policy picks, child affinity delegation and queue limits. Co-authored-by: Zyy <zyycn@users.noreply.github.com> Co-authored-by: Codex <noreply@openai.com>
2026-11-06 08:06:19 +00:00 · 2026-09-27 03:30:09 +08:00
parent 7bbea42fba
commit ae8da00459
9 changed files with 628 additions and 44 deletions
@@ -479,6 +479,7 @@ pub enum PreferredAccountSelection {
 pub struct AccountSelection<'a> {
    candidate: &'a AccountCandidate,
    preferred: PreferredAccountSelection,
+    policy_choice: bool,
 }

 impl<'a> AccountSelection<'a> {
@@ -491,6 +492,12 @@ impl<'a> AccountSelection<'a> {
    pub const fn preferred(self) -> PreferredAccountSelection {
        self.preferred
    }
+
+    /// 显式策略选号不受 Provider 的软亲和等待覆盖；委托内置选号不属于显式选择。
+    #[must_use]
+    pub const fn is_policy_choice(self) -> bool {
+        self.policy_choice
+    }
 }

 /// 同一 target 内唯一的账号排序器。
@@ -536,6 +543,7 @@ impl AccountSelector {
        Some(AccountSelection {
            candidate,
            preferred,
+            policy_choice: true,
        })
    }

@@ -609,6 +617,7 @@ impl AccountSelector {
                return Some(AccountSelection {
                    candidate,
                    preferred: PreferredAccountSelection::Hit,
+                    policy_choice: false,
                });
            }
        }
@@ -664,6 +673,7 @@ impl AccountSelector {
        Some(AccountSelection {
            candidate,
            preferred,
+            policy_choice: false,
        })
    }

@@ -609,7 +609,18 @@ impl CodexCredentialSelector {
                    excluded_accounts: base_excluded.clone(),
                    ..context.clone()
                };
-                let wait_candidates = AccountSelector.wait_candidates(&candidates, &wait_context);
+                let mut wait_candidates =
+                    AccountSelector.wait_candidates(&candidates, &wait_context);
+                let preferred_wait = preferred.as_ref().filter(|account_id| {
+                    !diagnostic
+                        && queue_policy.max_waiting > 0
+                        && (pinned_account.is_some() || !affinity.inherited)
+                        && wait_candidates.contains(account_id)
+                });
+                if let Some(preferred) = preferred_wait {
+                    // 原绑定仍合格时只等待它；并发、请求间隔与已有队列均不应造成临时换号。
+                    wait_candidates.retain(|account_id| account_id == preferred);
+                }
                let capacity = AccountSelector.capacity_snapshot(&candidates, &context);
                for candidate in &candidates {
                    if !waiting.can_try(candidate.account.id()) {
@@ -632,6 +643,19 @@ impl CodexCredentialSelector {
                        return Err(CredentialSelectionError::PolicyUnavailable);
                    }
                };
+                // 可等待集合使用原始排除集；这里的 Excluded 只可能来自本轮租约争用或队列让位。
+                let selection = selection.filter(|selection| {
+                    preferred_wait.is_none()
+                        || selection.is_policy_choice()
+                        || !matches!(
+                            selection.preferred(),
+                            PreferredAccountSelection::Blocked(
+                                AccountSchedulingBlocker::ConcurrencyLimit
+                                    | AccountSchedulingBlocker::RequestInterval
+                                    | AccountSchedulingBlocker::Excluded
+                            )
+                        )
+                });
                request.attempt.trace().account_selection(
                    &candidates,
                    &context,
@@ -673,12 +697,14 @@ impl CodexCredentialSelector {
                    let quota_exhausted = !diagnostic
                        && statuses.next() == Some(AccountStatus::QuotaExhausted)
                        && statuses.all(|status| status == AccountStatus::QuotaExhausted);
-                    return match shortest_retry {
-                        Some(retry_after) => Err(CredentialSelectionError::CapacityUnavailable {
-                            retry_after: Some(retry_after),
-                        }),
-                        None if quota_exhausted => Err(CredentialSelectionError::QuotaExhausted),
-                        None => Err(CredentialSelectionError::NoEligibleCredential),
+                    return if !wait_candidates.is_empty() || shortest_retry.is_some() {
+                        Err(CredentialSelectionError::CapacityUnavailable {
+                            retry_after: shortest_retry,
+                        })
+                    } else if quota_exhausted {
+                        Err(CredentialSelectionError::QuotaExhausted)
+                    } else {
+                        Err(CredentialSelectionError::NoEligibleCredential)
                    };
                };
                affinity.observe_preferred_selection(selection.preferred());
@@ -714,7 +740,9 @@ impl CodexCredentialSelector {
                    .await?
                {
                    ProviderLeaseAcquisition::Busy { retry_after } => {
-                        affinity.observe_lease_busy(account.id());
+                        if preferred_wait != Some(account.id()) {
+                            affinity.observe_lease_busy(account.id());
+                        }
                        shortest_retry = minimum_duration(shortest_retry, retry_after);
                        excluded.insert(account.id().clone());
                    }
@@ -0,0 +1,383 @@
+use futures::{FutureExt, future::BoxFuture};
+use gateway_core::account::AccountRuntimeSignals;
+use gateway_core::concurrency::{ConcurrencyQueuePolicy, QueueRejection};
+use gateway_core::engine::policy::{
+    AccountScheduleDecision, AccountScheduleInput, ModelRouteDecision, ModelRouteInput,
+    RequestPolicyContext, RequestPolicyFault, RequestPolicyPlan,
+};
+use gateway_core::runtime::extensions::{ExtensionSetId, ExtensionSetLease, ExtensionSetReference};
+
+use super::*;
+
+fn busy_signal(in_flight: u32, last_started_at: Option<SystemTime>) -> AccountRuntimeSignals {
+    AccountRuntimeSignals {
+        in_flight,
+        last_started_at,
+        quota_reset_at: None,
+        quota_remaining_rank: None,
+        cooldown: None,
+        failure_rate_basis_points: None,
+        first_output_latency_ms: None,
+    }
+}
+
+async fn bind_first(affinity: &MemorySessionAffinity) -> ProviderSessionAffinityKey {
+    let key = ProviderSessionAffinityKey::try_new("capacity-session").unwrap();
+    affinity
+        .bind(
+            &ProviderKind::new("openai").unwrap(),
+            &key,
+            &ProviderAccountId::new("acct_first").unwrap(),
+            Duration::from_secs(60),
+        )
+        .await
+        .unwrap();
+    key
+}
+
+fn capacity_attempt(
+    policy: AccountSelectionPolicy,
+    request_policy: Option<RequestPolicyContext>,
+) -> AttemptContext {
+    AttemptContext::new(
+        RequestAttemptContext::new(
+            ModelRequestId::new("req_capacity_contract").unwrap(),
+            ClientApiKeyId::new("key_codex_contract").unwrap(),
+        )
+        .with_request_policy(request_policy),
+        NonZeroU32::new(1).unwrap(),
+        SystemTime::now() + Duration::from_secs(5),
+        policy,
+        AccountAttemptContext::new(BTreeSet::new(), None, None)
+            .with_account_scope(contract_account_scope()),
+        None,
+        CancellationToken::new(),
+    )
+}
+
+#[test]
+fn saturated_snapshot_is_capacity_unavailable_even_with_exhausted_accounts() {
+    for interval in [false, true] {
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account(&store, "acct_first", "test-first");
+        create_account(&store, "acct_second", "test-second");
+        let exhausted =
+            block_on(store.get_account(&ProviderAccountId::new("acct_second").unwrap()))
+                .unwrap()
+                .unwrap();
+        persist_quota_exhaustion(&store, &exhausted, None);
+        let leases = Arc::new(TestLeaseCoordinator::default());
+        leases.signals.lock().unwrap().insert(
+            ProviderAccountId::new("acct_first").unwrap(),
+            busy_signal(
+                if interval { 0 } else { u32::MAX },
+                interval.then(SystemTime::now),
+            ),
+        );
+        let selector = selector(&store, leases.clone());
+        let request_attempt = capacity_attempt(
+            AccountSelectionPolicy::new(
+                RotationStrategy::Smart,
+                NonZeroU32::new(2).unwrap(),
+                Duration::from_secs(2),
+            ),
+            None,
+        );
+        let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap();
+        let error = block_on(selector.select(&SelectCodexCredential {
+            upstream_model: "gpt-5.4",
+            request_url: &url,
+            attempt: &request_attempt,
+            session_affinity_key: None,
+        }))
+        .unwrap_err();
+        assert!(
+            matches!(error, CredentialSelectionError::CapacityUnavailable { .. }),
+            "{error:?}"
+        );
+        assert!(leases.requests.lock().unwrap().is_empty());
+    }
+}
+
+#[tokio::test]
+async fn affinity_queue_waits_for_the_original_account_for_local_capacity_blockers() {
+    for blocker in ["lease", "concurrency", "interval"] {
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account(&store, "acct_first", "test-first");
+        create_account(&store, "acct_second", "test-second");
+        let leases = Arc::new(TestLeaseCoordinator::default());
+        let first = ProviderAccountId::new("acct_first").unwrap();
+        if blocker == "lease" {
+            leases.busy_accounts.lock().unwrap().insert(first.clone());
+        } else {
+            leases.signals.lock().unwrap().insert(
+                first.clone(),
+                busy_signal(
+                    if blocker == "concurrency" { 2 } else { 0 },
+                    (blocker == "interval").then(SystemTime::now),
+                ),
+            );
+        }
+        let affinity = Arc::new(MemorySessionAffinity::default());
+        let key = bind_first(&affinity).await;
+        let selector = selector_with_affinity(&store, leases.clone(), affinity.clone());
+        let request_attempt = capacity_attempt(
+            AccountSelectionPolicy::new(
+                RotationStrategy::Smart,
+                NonZeroU32::new(2).unwrap(),
+                Duration::from_secs(2),
+            )
+            .with_queue(ConcurrencyQueuePolicy {
+                max_waiting: 1,
+                timeout: Duration::from_secs(2),
+            }),
+            None,
+        );
+        let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap();
+        let request = SelectCodexCredential {
+            upstream_model: "gpt-5.4",
+            request_url: &url,
+            attempt: &request_attempt,
+            session_affinity_key: Some(&key),
+        };
+        let mut pending = Box::pin(selector.select(&request));
+        assert!(
+            pending.as_mut().now_or_never().is_none(),
+            "blocker={blocker}"
+        );
+        leases.signals.lock().unwrap().clear();
+        leases.busy_accounts.lock().unwrap().clear();
+        let selected = pending.await.unwrap();
+        assert_eq!(selected.account_id(), &first);
+        assert!(selected.affinity_hit());
+        assert_eq!(selected.escape_reason(), None);
+        assert!(!selected.account_switch());
+        assert!(
+            leases
+                .requests
+                .lock()
+                .unwrap()
+                .iter()
+                .all(|request| request.account_id() == &first)
+        );
+    }
+}
+
+#[tokio::test]
+async fn affinity_queue_full_and_timeout_preserve_binding_despite_free_fallback() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_first", "test-first");
+    create_account(&store, "acct_second", "test-second");
+    let first = ProviderAccountId::new("acct_first").unwrap();
+    let leases = Arc::new(TestLeaseCoordinator::default());
+    leases.busy_accounts.lock().unwrap().insert(first.clone());
+    let affinity = Arc::new(MemorySessionAffinity::default());
+    let key = bind_first(&affinity).await;
+    let selector = selector_with_affinity(&store, leases.clone(), affinity.clone());
+    let head_attempt = queued_attempt(Duration::from_secs(2));
+    let next_attempt = queued_attempt(Duration::from_secs(2));
+    let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap();
+    let request = SelectCodexCredential {
+        upstream_model: "gpt-5.4",
+        request_url: &url,
+        attempt: &head_attempt,
+        session_affinity_key: Some(&key),
+    };
+    let mut head = Box::pin(selector.select(&request));
+    assert!(head.as_mut().now_or_never().is_none());
+    // 槽位已空，但新请求不能越过旧等待者，也不能转而抢占另一账号。
+    leases.busy_accounts.lock().unwrap().clear();
+    let rejected = selector
+        .select(&SelectCodexCredential {
+            attempt: &next_attempt,
+            ..request
+        })
+        .await
+        .unwrap_err();
+    assert!(matches!(
+        rejected,
+        CredentialSelectionError::QueueRejected(QueueRejection::Full)
+    ));
+    drop(head);
+
+    leases.busy_accounts.lock().unwrap().insert(first.clone());
+    let timeout_attempt = queued_attempt(Duration::from_millis(20));
+    let timeout = selector
+        .select(&SelectCodexCredential {
+            attempt: &timeout_attempt,
+            ..request
+        })
+        .await
+        .unwrap_err();
+    assert!(matches!(
+        timeout,
+        CredentialSelectionError::QueueRejected(QueueRejection::Timeout)
+    ));
+    assert_eq!(
+        affinity
+            .load(&ProviderKind::new("openai").unwrap(), &key)
+            .await
+            .unwrap(),
+        Some(first.clone())
+    );
+    leases.busy_accounts.lock().unwrap().clear();
+    let selected = selector
+        .select(&SelectCodexCredential {
+            attempt: &next_attempt,
+            ..request
+        })
+        .await
+        .unwrap();
+    assert_eq!(selected.account_id(), &first);
+    assert!(
+        leases
+            .requests
+            .lock()
+            .unwrap()
+            .iter()
+            .all(|request| request.account_id() == &first)
+    );
+}
+
+#[tokio::test]
+async fn affinity_queue_rechecks_hard_unavailability_before_switching_accounts() {
+    for quota_exhausted in [false, true] {
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account(&store, "acct_first", "test-first");
+        create_account(&store, "acct_second", "test-second");
+        let first = ProviderAccountId::new("acct_first").unwrap();
+        let leases = Arc::new(TestLeaseCoordinator::default());
+        leases.busy_accounts.lock().unwrap().insert(first.clone());
+        let affinity = Arc::new(MemorySessionAffinity::default());
+        let key = bind_first(&affinity).await;
+        let selector = selector_with_affinity(&store, leases, affinity);
+        let request_attempt = queued_attempt(Duration::from_secs(2));
+        let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap();
+        let request = SelectCodexCredential {
+            upstream_model: "gpt-5.4",
+            request_url: &url,
+            attempt: &request_attempt,
+            session_affinity_key: Some(&key),
+        };
+        let mut pending = Box::pin(selector.select(&request));
+        assert!(pending.as_mut().now_or_never().is_none());
+        let account = store.get_account(&first).await.unwrap().unwrap();
+        if quota_exhausted {
+            persist_quota_exhaustion(&store, &account, None);
+        } else {
+            persist_credential_state(&store, &account, CredentialState::Expired);
+        }
+        let selected = pending.await.unwrap();
+        assert_eq!(selected.account_id().as_str(), "acct_second");
+        assert!(selected.account_switch());
+    }
+}
+
+#[tokio::test]
+async fn queue_enabled_without_affinity_still_uses_a_free_account() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_first", "test-first");
+    create_account(&store, "acct_second", "test-second");
+    let leases = Arc::new(TestLeaseCoordinator::default());
+    leases.signals.lock().unwrap().insert(
+        ProviderAccountId::new("acct_first").unwrap(),
+        busy_signal(2, None),
+    );
+    let selector = selector(&store, leases);
+    let request_attempt = queued_attempt(Duration::from_secs(2));
+    let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap();
+    let selected = selector
+        .select(&SelectCodexCredential {
+            upstream_model: "gpt-5.4",
+            request_url: &url,
+            attempt: &request_attempt,
+            session_affinity_key: None,
+        })
+        .await
+        .unwrap();
+    assert_eq!(selected.account_id().as_str(), "acct_second");
+}
+
+#[derive(Debug)]
+struct Scheduler {
+    explicit: bool,
+}
+
+impl RequestPolicyPlan for Scheduler {
+    fn route_model(
+        &self,
+        _: ModelRouteInput,
+    ) -> BoxFuture<'static, Result<ModelRouteDecision, RequestPolicyFault>> {
+        Box::pin(async { Ok(ModelRouteDecision::Unhandled) })
+    }
+
+    fn schedule_account(
+        &self,
+        _: AccountScheduleInput,
+    ) -> BoxFuture<'static, Result<AccountScheduleDecision, RequestPolicyFault>> {
+        let decision = if self.explicit {
+            AccountScheduleDecision::Pick(ProviderAccountId::new("acct_second").unwrap())
+        } else {
+            AccountScheduleDecision::Delegate
+        };
+        Box::pin(async move { Ok(decision) })
+    }
+}
+
+struct ExtensionLease;
+
+impl ExtensionSetLease for ExtensionLease {
+    fn is_ready(&self) -> bool {
+        true
+    }
+}
+
+#[tokio::test]
+async fn affinity_wait_respects_explicit_policy_choices_but_not_delegated_fallbacks() {
+    for explicit in [false, true] {
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account(&store, "acct_first", "test-first");
+        create_account(&store, "acct_second", "test-second");
+        let leases = Arc::new(TestLeaseCoordinator::default());
+        leases.signals.lock().unwrap().insert(
+            ProviderAccountId::new("acct_first").unwrap(),
+            busy_signal(2, None),
+        );
+        let affinity = Arc::new(MemorySessionAffinity::default());
+        let key = bind_first(&affinity).await;
+        let selector = selector_with_affinity(&store, leases.clone(), affinity);
+        let request_attempt = capacity_attempt(
+            account_policy().with_queue(ConcurrencyQueuePolicy {
+                max_waiting: 1,
+                timeout: Duration::from_secs(2),
+            }),
+            Some(RequestPolicyContext::new(
+                Arc::new(Scheduler { explicit }),
+                ExtensionSetReference::new(
+                    ExtensionSetId::new("capacity-test".into()).unwrap(),
+                    Arc::new(ExtensionLease),
+                ),
+                ModelRequestId::new("req_capacity_contract").unwrap(),
+                ClientApiKeyId::new("key_codex_contract").unwrap(),
+                vec![],
+            )),
+        );
+        let url = Url::parse(OFFICIAL_CODEX_BASE_URL).unwrap();
+        let request = SelectCodexCredential {
+            upstream_model: "gpt-5.4",
+            request_url: &url,
+            attempt: &request_attempt,
+            session_affinity_key: Some(&key),
+        };
+        let mut pending = Box::pin(selector.select(&request));
+        if explicit {
+            let selected = pending.as_mut().now_or_never().unwrap().unwrap();
+            assert_eq!(selected.account_id().as_str(), "acct_second");
+        } else {
+            assert!(pending.as_mut().now_or_never().is_none());
+            leases.signals.lock().unwrap().clear();
+            let selected = pending.await.unwrap();
+            assert_eq!(selected.account_id().as_str(), "acct_first");
+        }
+    }
+}
@@ -1,3 +1,5 @@
+mod capacity;
+
 use std::collections::{BTreeMap, BTreeSet};
 use std::num::NonZeroU32;
 use std::sync::Arc;
@@ -0,0 +1,131 @@
+use futures::FutureExt;
+use gateway_core::concurrency::ConcurrencyQueuePolicy;
+
+use super::*;
+
+fn queued_context(request_id: &str) -> AttemptContext {
+    AttemptContext::new(
+        RequestAttemptContext::new(
+            ModelRequestId::new(request_id).unwrap(),
+            ClientApiKeyId::new("key_openai_contract").unwrap(),
+        ),
+        NonZeroU32::new(1).unwrap(),
+        SystemTime::now() + Duration::from_secs(5),
+        account_policy().with_queue(ConcurrencyQueuePolicy {
+            max_waiting: 1,
+            timeout: Duration::from_secs(2),
+        }),
+        AccountAttemptContext::new(BTreeSet::new(), None, None)
+            .with_account_scope(contract_account_scope()),
+        None,
+        CancellationToken::new(),
+    )
+}
+
+fn operation(thread_id: &str) -> Operation {
+    Operation::Generate(GenerateRequest::from_protocol_payload(
+        ProtocolPayload::json_object(
+            "openai",
+            Map::from_iter([
+                ("model".to_owned(), json!("gpt-5.4")),
+                ("input".to_owned(), json!("capacity queue test")),
+                ("session_id".to_owned(), json!("capacity-root")),
+                ("thread_id".to_owned(), json!(thread_id)),
+            ]),
+        )
+        .unwrap()
+        .with_context(Map::from_iter([("use_websocket".to_owned(), json!(false))])),
+    ))
+}
+
+#[tokio::test]
+async fn queued_session_sends_only_after_capacity_recovers_while_a_new_child_can_use_another_account()
+ {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_subagent_a").await;
+    let leases = Arc::new(TestLeaseCoordinator::default());
+    let affinity = Arc::new(MemorySessionAffinity::default());
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .set_body_string(CAPTURE_COMPLETED_SSE),
+        )
+        .expect(3)
+        .mount(&server)
+        .await;
+    let provider = provider_with_affinity_and_base_url_and_leases(
+        &store,
+        affinity,
+        server.uri(),
+        leases.clone(),
+    );
+    let mut root = provider
+        .clone()
+        .execute(
+            planned_request("openai", operation("capacity-root")),
+            queued_context("req_capacity_root"),
+        )
+        .await
+        .unwrap();
+    while let Some(event) = root.next().await {
+        event.unwrap();
+    }
+    drop(root);
+
+    create_account(&store, "acct_subagent_b").await;
+    leases
+        .busy_accounts
+        .lock()
+        .unwrap()
+        .insert(ProviderAccountId::new("acct_subagent_a").unwrap());
+    let mut waiting = Box::pin(provider.clone().execute(
+        planned_request("openai", operation("capacity-root")),
+        queued_context("req_capacity_wait"),
+    ));
+    assert!(waiting.as_mut().now_or_never().is_none());
+
+    // 子线程还没有自己的绑定，父账号只是默认偏好，不能被强制挤进父账号队列。
+    let mut child = provider
+        .execute(
+            planned_request("openai", operation("capacity-child")),
+            queued_context("req_capacity_child"),
+        )
+        .await
+        .unwrap();
+    assert_eq!(
+        child.metadata().provider_account_id().as_str(),
+        "acct_subagent_b"
+    );
+    while let Some(event) = child.next().await {
+        event.unwrap();
+    }
+    drop(child);
+    assert_eq!(server.received_requests().await.unwrap().len(), 2);
+
+    leases.busy_accounts.lock().unwrap().clear();
+    let mut resumed = waiting.await.unwrap();
+    assert_eq!(
+        resumed.metadata().provider_account_id().as_str(),
+        "acct_subagent_a"
+    );
+    while let Some(event) = resumed.next().await {
+        event.unwrap();
+    }
+    let requests = server.received_requests().await.unwrap();
+    let accounts = requests
+        .iter()
+        .map(|request| request.headers["chatgpt-account-id"].to_str().unwrap())
+        .collect::<Vec<_>>();
+    assert_eq!(
+        accounts,
+        [
+            "chatgpt-acct_subagent_a",
+            "chatgpt-acct_subagent_b",
+            "chatgpt-acct_subagent_a"
+        ]
+    );
+    server.verify().await;
+}
@@ -1,3 +1,5 @@
+mod capacity;
+
 use std::collections::{BTreeMap, BTreeSet};
 use std::io::{self, Write};
 use std::num::NonZeroU32;
@@ -698,7 +700,7 @@ async fn selected_proxy_location_overrides_global_and_reloads_without_mutating_c
 }

 const OFFICIAL_FIXTURE: &[u8] =
-    include_bytes!("../transport/fixtures/official_models_snapshot.json");
+    include_bytes!("../../transport/fixtures/official_models_snapshot.json");

 #[tokio::test]
 async fn replay_compatibility_should_remove_only_reasoning_status_on_both_transports() {
@@ -3411,30 +3413,47 @@ async fn standalone_search_returns_the_exact_upstream_error_response() {

 #[tokio::test]
 async fn capacity_selection_error_preserves_classification_and_retry_after() {
-    let store = Arc::new(MemoryAccountStore::default());
-    create_account(&store, "acct_capacity_busy").await;
-    let leases = Arc::new(TestLeaseCoordinator::default());
-    *leases.busy.lock().expect("lease busy lock") = true;
+    for snapshot in [false, true] {
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account(&store, "acct_capacity_busy").await;
+        let leases = Arc::new(TestLeaseCoordinator::default());
+        if snapshot {
+            leases.signals.lock().unwrap().insert(
+                ProviderAccountId::new("acct_capacity_busy").unwrap(),
+                gateway_core::account::AccountRuntimeSignals {
+                    in_flight: u32::MAX,
+                    last_started_at: None,
+                    quota_reset_at: None,
+                    quota_remaining_rank: None,
+                    cooldown: None,
+                    failure_rate_basis_points: None,
+                    first_output_latency_ms: None,
+                },
+            );
+        } else {
+            *leases.busy.lock().unwrap() = true;
+        }

-    let error = match provider_with_leases(&store, leases)
-        .execute(
-            planned_request("openai", generate_operation()),
-            context("req_capacity_busy", CancellationToken::new()),
-        )
-        .await
-    {
-        Ok(_) => panic!("busy account selection must fail"),
-        Err(error) => error,
-    };
+        let error = match provider_with_leases(&store, leases)
+            .execute(
+                planned_request("openai", generate_operation()),
+                context("req_capacity_busy", CancellationToken::new()),
+            )
+            .await
+        {
+            Ok(_) => panic!("busy account selection must fail"),
+            Err(error) => error,
+        };

-    assert_eq!(
-        (error.kind(), error.send_state(), error.retry_after()),
-        (
-            ProviderErrorKind::AccountCapacityUnavailable,
-            UpstreamSendState::NotSent,
-            Some(Duration::from_millis(25)),
-        )
-    );
+        assert_eq!(
+            (error.kind(), error.send_state(), error.retry_after()),
+            (
+                ProviderErrorKind::AccountCapacityUnavailable,
+                UpstreamSendState::NotSent,
+                (!snapshot).then_some(Duration::from_millis(25)),
+            )
+        );
+    }
 }

 #[tokio::test]
@@ -655,6 +655,7 @@ pub(crate) struct TestLeaseCoordinator {
    pub(crate) requests: Mutex<Vec<ProviderSchedulingLeaseRequest>>,
    pub(crate) busy: Mutex<bool>,
    pub(crate) busy_accounts: Mutex<BTreeSet<ProviderAccountId>>,
+    pub(crate) signals: Mutex<BTreeMap<ProviderAccountId, AccountRuntimeSignals>>,
    round_robin_cursor: Mutex<u64>,
 }

@@ -666,21 +667,24 @@ impl ProviderLeasePort for TestLeaseCoordinator {
        accounts: &'a [ProviderAccountId],
    ) -> BoxFuture<'a, Result<ProviderSchedulingState, ProviderStoreError>> {
        Box::pin(async move {
+            let overrides = self.signals.lock().expect("scheduling signals lock");
            let signals = accounts
                .iter()
-                .cloned()
                .map(|account| {
                    (
-                        account,
-                        AccountRuntimeSignals {
-                            in_flight: 0,
-                            last_started_at: None,
-                            quota_reset_at: None,
-                            quota_remaining_rank: None,
-                            cooldown: None,
-                            failure_rate_basis_points: None,
-                            first_output_latency_ms: None,
-                        },
+                        account.clone(),
+                        overrides
+                            .get(account)
+                            .cloned()
+                            .unwrap_or(AccountRuntimeSignals {
+                                in_flight: 0,
+                                last_started_at: None,
+                                quota_reset_at: None,
+                                quota_remaining_rank: None,
+                                cooldown: None,
+                                failure_rate_basis_points: None,
+                                first_output_latency_ms: None,
+                            }),
                    )
                })
                .collect();
@@ -1116,8 +1116,12 @@ HTTP 请求头及新建 WS 的握手提示按当时的最终出站档位构造
 `response.create`；空闲连接不占名额，内部重试不重复占用。
 修改 Key 策略对既有 WebSocket 连接的下一次请求同样生效，已开始的请求保持原有快照。

-运行设置可以分别启用 Key 与账号的有界排队。Key 并发满时按 Key 等待；账号先使用其它可调度候选，
-适用账号均暂时满载后按账号等待。RPM、金额限额、失效账号和上游冷却不通过排队绕过。
+运行设置可以分别启用 Key 与账号的有界排队。Key 并发满时按 Key 等待；OpenAI 已绑定的会话账号仅因
+本地并发、请求间隔或已有等待者暂忙时，开启账号排队后优先等待原账号，队列满或超时不因此迁移绑定。
+原账号失效、额度耗尽或进入上游冷却时重新选择；显式调度策略选号与智能调度高权重回切仍按各自规则执行。
+没有适用亲和时先使用其它可调度候选，适用账号均暂时满载后按账号等待。
+关闭账号排队时，OpenAI 本地容量不足返回 `503` / `account_capacity_unavailable`，
+与无候选账号的 `no_available_provider` 区分。RPM、金额限额、失效账号和上游冷却不通过排队绕过。
 队列满返回 `429` / `concurrency_queue_full`，排队超时返回 `429` / `concurrency_queue_timeout`；
 WebSocket 使用对应错误事件。等待期间不发送上游请求，取消后释放等待位置，排队重查不重复计入 RPM。
 SSE 在取得有效执行前不发送保活帧，因此此阶段保留 HTTP 错误状态；已开始交付的失败沿用流内错误合同。
@@ -494,6 +494,9 @@ Continuation 仍受原请求的 Client Key、账号范围、Provider 和发送/
 - scope 外账号、跨 Key 复用或不明确发送结果均 fail closed。

 会话亲和是优先选择提示，不是硬账号绑定；native continuation 才携带不可跨越的 owner 约束。
+OpenAI 启用账号排队时，可等待的已有亲和绑定优先保留；子线程首绑前继承的根偏好仍允许分流。
+快照满载、原子租约争用与已有等待者均进入原账号队列，等待重查仍执行完整资格校验。
+显式策略选号和内置高权重回切保留各自裁决；亲和等待不将暂忙升级成失效，也不覆盖原生续写约束。

 ### 并发等待