fix(openai): tolerate session affinity storage jitter

Raise the session affinity storage budget to 500ms while keeping cyber policy operations on a separate 100ms budget. Preserve fail-closed binding behavior and cover delayed admission and timeout rejection before upstream sending.

Refs #549

Co-authored-by: Codex <noreply@openai.com>

backend/crates/providers/openai/src/credential/selector.rs

@@ -45,7 +45,9 @@ const CLOUDFLARE_CHALLENGE_BACKOFF: [Duration; 4] = [
     Duration::from_secs(120),
 ];
 const CLOUDFLARE_PATH_BLOCK_THRESHOLD: u32 = 3;
-const SESSION_AFFINITY_TIMEOUT: Duration = Duration::from_millis(100);
+// 绑定失败会阻止发送，预算容纳短暂调度抖动；可丢失的策略排除记录保持独立预算
+const SESSION_AFFINITY_TIMEOUT: Duration = Duration::from_millis(500);
+const CYBER_POLICY_TIMEOUT: Duration = Duration::from_millis(100);
 const CYBER_POLICY_SESSION_TTL: Duration = Duration::from_secs(60 * 60);
 const MAX_ACCOUNT_SNAPSHOT_RETRIES: u32 = 3;
 const COOLDOWN_READ_CONCURRENCY: usize = 128;
@@ -1141,7 +1143,7 @@ impl CodexCredentialSelector {
     ) -> Option<CodexCyberPolicyScope> {
         let key = key?.clone();
         let state = match tokio::time::timeout(
-            SESSION_AFFINITY_TIMEOUT,
+            CYBER_POLICY_TIMEOUT,
             self.session_exclusions.load(&self.provider_kind, &key),
         )
         .await
@@ -1153,7 +1155,7 @@ impl CodexCredentialSelector {
             }
             Err(_) => {
                 tracing::warn!(
-                    timeout_ms = SESSION_AFFINITY_TIMEOUT.as_millis(),
+                    timeout_ms = CYBER_POLICY_TIMEOUT.as_millis(),
                     "OpenAI cyber policy state read timed out"
                 );
                 None
@@ -1171,7 +1173,7 @@ impl CodexCredentialSelector {
             return;
         };
         match tokio::time::timeout(
-            SESSION_AFFINITY_TIMEOUT,
+            CYBER_POLICY_TIMEOUT,
             self.session_exclusions.record_failure(
                 &self.provider_kind,
                 &scope.key,
@@ -1192,7 +1194,7 @@ impl CodexCredentialSelector {
             Err(_) => {
                 tracing::warn!(
                     account_id = %account.id(),
-                    timeout_ms = SESSION_AFFINITY_TIMEOUT.as_millis(),
+                    timeout_ms = CYBER_POLICY_TIMEOUT.as_millis(),
                     "OpenAI cyber policy exclusion write timed out"
                 );
             }
@@ -1207,7 +1209,7 @@ impl CodexCredentialSelector {
             return;
         };
         match tokio::time::timeout(
-            SESSION_AFFINITY_TIMEOUT,
+            CYBER_POLICY_TIMEOUT,
             self.session_exclusions
                 .clear(&self.provider_kind, &scope.key, state.revision()),
         )
@@ -1219,7 +1221,7 @@ impl CodexCredentialSelector {
             }
             Err(_) => {
                 tracing::warn!(
-                    timeout_ms = SESSION_AFFINITY_TIMEOUT.as_millis(),
+                    timeout_ms = CYBER_POLICY_TIMEOUT.as_millis(),
                     "OpenAI cyber policy exclusion clear timed out"
                 );
             }

backend/crates/providers/openai/tests/provider/contract/session_binding.rs

@@ -2,6 +2,71 @@
 
 use super::*;
 
+#[tokio::test]
+async fn session_affinity_storage_tolerates_delays_but_rejects_expired_requests_before_sending() {
+    for (delay_ms, should_succeed) in [(250, true), (750, false)] {
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account(&store, "acct_subagent_a").await;
+        let affinity = Arc::new(MemorySessionAffinity::with_operation_delay(
+            Duration::from_millis(delay_ms),
+        ));
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path("/codex/responses"))
+            .respond_with(
+                ResponseTemplate::new(200).set_body_raw(
+                    format!("event: response.created\ndata: {{\"type\":\"response.created\",\"response\":{{\"id\":\"resp_scope_capture\",\"status\":\"in_progress\"}}}}\n\n{CAPTURE_COMPLETED_SSE}"),
+                    "text/event-stream",
+                ),
+            )
+            .expect(if should_succeed { 1 } else { 0 })
+            .mount(&server)
+            .await;
+        let provider = provider_with_affinity_and_base_url(&store, affinity.clone(), server.uri());
+
+        // 只推进存储等待的虚拟时间，网络交付恢复真实时钟，避免依赖宿主调度速度
+        tokio::time::pause();
+        let result = provider
+            .execute(
+                planned_request("openai", turn_request("delayed-session", "delayed-turn")),
+                context("req_affinity_delay", CancellationToken::new()),
+            )
+            .await;
+        tokio::time::resume();
+
+        if should_succeed {
+            let mut stream = result.expect("delayed binding and turn association must succeed");
+            assert_eq!(
+                stream.metadata().provider_account_id().as_str(),
+                "acct_subagent_a"
+            );
+            assert_eq!(affinity.binding_count(), 1);
+            assert_eq!(affinity.alias_ttls().len(), 1);
+            let mut completed = false;
+            while let Some(event) = stream.next().await {
+                completed |= event
+                    .unwrap()
+                    .canonical_facts()
+                    .iter()
+                    .any(|event| matches!(event, GatewayEvent::Completed(_)));
+            }
+            assert!(completed, "admitted request must complete through upstream");
+        } else {
+            let error = result
+                .err()
+                .expect("expired storage wait must reject the request");
+            assert_eq!(
+                error.kind(),
+                ProviderErrorKind::ProviderInfrastructureUnavailable
+            );
+            assert_eq!(error.send_state(), UpstreamSendState::NotSent);
+            assert!(error.retry_is_prohibited());
+            assert_eq!(affinity.binding_count(), 0);
+        }
+        server.verify().await;
+    }
+}
+
 #[tokio::test]
 async fn cyber_policy_session_exclusion_has_actionable_non_retryable_error() {
     let store = Arc::new(MemoryAccountStore::default());

backend/crates/providers/openai/tests/support.rs

@@ -785,6 +785,7 @@ impl ProviderLeasePort for TestLeaseCoordinator {
 
 #[derive(Default)]
 pub(crate) struct MemorySessionAffinity {
+    operation_delay: Duration,
     initial_claim_barrier: Option<Arc<tokio::sync::Barrier>>,
     aliases: Mutex<BTreeMap<(String, String), gateway_core::provider_ports::ProviderSessionAlias>>,
     bindings:
@@ -795,6 +796,19 @@ pub(crate) struct MemorySessionAffinity {
 }
 
 impl MemorySessionAffinity {
+    pub(crate) fn with_operation_delay(operation_delay: Duration) -> Self {
+        Self {
+            operation_delay,
+            ..Self::default()
+        }
+    }
+
+    async fn delay_operation(&self) {
+        if !self.operation_delay.is_zero() {
+            tokio::time::sleep(self.operation_delay).await;
+        }
+    }
+
     pub(crate) fn with_initial_claim_barrier(participants: usize) -> Self {
         Self {
             initial_claim_barrier: Some(Arc::new(tokio::sync::Barrier::new(participants))),
@@ -865,6 +879,7 @@ impl ProviderSessionAffinityPort for MemorySessionAffinity {
         Result<Option<gateway_core::provider_ports::ProviderSessionBinding>, ProviderStoreError>,
     > {
         Box::pin(async move {
+            self.delay_operation().await;
             self.lookups
                 .lock()
                 .expect("lookups")
@@ -893,6 +908,7 @@ impl ProviderSessionAffinityPort for MemorySessionAffinity {
         Result<Option<gateway_core::provider_ports::ProviderSessionBinding>, ProviderStoreError>,
     > {
         Box::pin(async move {
+            self.delay_operation().await;
             // 让并发首请求都读到空绑定，再验证发送前的原子认领冲突
             if expected.is_none()
                 && let Some(barrier) = &self.initial_claim_barrier
@@ -931,6 +947,7 @@ impl ProviderSessionAffinityPort for MemorySessionAffinity {
         Result<Option<gateway_core::provider_ports::ProviderSessionAlias>, ProviderStoreError>,
     > {
         Box::pin(async move {
+            self.delay_operation().await;
             Ok(self
                 .aliases
                 .lock()
@@ -950,6 +967,7 @@ impl ProviderSessionAffinityPort for MemorySessionAffinity {
         ttl: Duration,
     ) -> BoxFuture<'a, Result<bool, ProviderStoreError>> {
         Box::pin(async move {
+            self.delay_operation().await;
             let mut aliases = self.aliases.lock().unwrap();
             let current = aliases
                 .entry((