From e390417821f676e02a705aec3754b8d458c0da4b Mon Sep 17 00:00:00 2001 From: Zyy Date: Fri, 6 Nov 2026 20:01:01 +0800 Subject: [PATCH] fix(openai): prefer the same account for sessionless requests Use the client cache key or the opening messages as a tenant-scoped scheduling preference. Keep explicit sessions and continuation ownership separate, and allow spillover when the preferred account is busy. Refs #622 Co-authored-by: Codex --- .../openai/src/credential/affinity.rs | 89 +++++++++--- .../providers/openai/src/credential/mod.rs | 2 +- .../openai/src/credential/selector.rs | 73 +++++++--- .../providers/openai/src/provider/mod.rs | 33 +++-- .../providers/openai/src/transport/request.rs | 39 +++-- .../provider/contract/cache_preference.rs | 133 ++++++++++++++++++ .../openai/tests/provider/contract/mod.rs | 20 ++- .../provider/contract/session_binding.rs | 85 +++++++++-- docs/api.md | 5 + docs/architecture.md | 8 +- 10 files changed, 402 insertions(+), 85 deletions(-) create mode 100644 backend/crates/providers/openai/tests/provider/contract/cache_preference.rs diff --git a/backend/crates/providers/openai/src/credential/affinity.rs b/backend/crates/providers/openai/src/credential/affinity.rs index 4095487e7..8b24409ea 100644 --- a/backend/crates/providers/openai/src/credential/affinity.rs +++ b/backend/crates/providers/openai/src/credential/affinity.rs @@ -8,8 +8,10 @@ use sha2::{Digest, Sha256}; use crate::request_identity::{account_session_with_headers, follows_session_with_headers}; use crate::transport::protocol::responses::CodexResponsesRequest; -use crate::transport::request::derive_conversation_anchor; +use crate::transport::request::{derive_conversation_anchor, message_text_parts}; +const CLIENT_CACHE_PREFERENCE: &str = "client-cache-preference"; +const CONTENT_PREFERENCE: &str = "content-preference"; const AFFINITY_KEY_HASH_LENGTH: usize = 12; /// 一次请求派生出的账号亲和键及其结构化日志上下文 @@ -18,7 +20,6 @@ pub(crate) struct CodexSessionAffinity { turn_alias: Option<(ProviderSessionAffinityKey, ProviderSessionAlias)>, key_hash: String, anchor_source: &'static str, - anchor: String, session_id: Option, follow_only: bool, } @@ -35,7 +36,6 @@ impl CodexSessionAffinity { key_hash: short_key_hash(&key), key, anchor_source: "turn-session", - anchor: String::new(), session_id: None, follow_only: root.is_some() || alias.follow_only, turn_alias: Some((turn, alias)), @@ -88,9 +88,20 @@ impl CodexSessionAffinity { self.anchor_source } + #[must_use] + pub(crate) fn is_cache_preference(&self) -> bool { + matches!( + self.anchor_source, + CLIENT_CACHE_PREFERENCE | CONTENT_PREFERENCE + ) + } + #[must_use] pub(crate) fn anchor(&self) -> &str { - &self.anchor + if self.is_cache_preference() { + return self.key_hash(); + } + self.session_id.as_deref().unwrap_or_default() } #[must_use] @@ -130,13 +141,60 @@ pub(crate) fn derive_codex_session_affinity( let session_id = non_empty(request.client_account_session_id.as_deref())?; session_affinity( "root-session", - session_id.to_owned(), - Some(session_id.to_owned()), + session_id, + Some(session_id), client_api_key_id, ) .map(|affinity| affinity.with_follow_only(request.client_account_follow_only)) } +/// 仅供选号使用。缓存键与输入正文均不作为会话、轮次或传输身份。 +pub(crate) fn derive_cache_preference( + request: &CodexResponsesRequest, + client_key: &ClientApiKeyId, +) -> Option { + if request.client_account_follow_only || request.previous_response_id().is_some() { + return None; + } + if request.explicit_prompt_cache_key + && let Some(key) = non_empty(request.prompt_cache_key()) + { + return session_affinity(CLIENT_CACHE_PREFERENCE, key, None, client_key); + } + let mut hash = Sha256::new(); + hash_text_field(&mut hash, std::iter::once(request.model())); + hash_text_field(&mut hash, std::iter::once(request.instructions())); + let mut items = request.input().iter().peekable(); + while let Some(item) = + items.next_if(|item| matches!(item["role"].as_str(), Some("system" | "developer"))) + { + hash_text_field(&mut hash, message_text_parts(item)); + } + let first_user = items.next().filter(|item| item["role"] == "user")?; + if !hash_text_field(&mut hash, message_text_parts(first_user)) { + return None; + } + session_affinity( + CONTENT_PREFERENCE, + &hex::encode(hash.finalize()), + None, + client_key, + ) +} + +/// 长度前缀区分消息边界；同一消息的文本块按顺序流式计算，避免复制长提示词。 +fn hash_text_field<'a>(hash: &mut Sha256, parts: impl Iterator + Clone) -> bool { + if !parts.clone().any(|part| !part.trim().is_empty()) { + return false; + } + let length: usize = parts.clone().map(str::len).sum(); + hash.update((length as u64).to_be_bytes()); + for part in parts { + hash.update(part.as_bytes()); + } + true +} + /// 原生 JSON 端点与 Responses 共用绑定命名空间，Search id 是逻辑会话身份 pub(crate) fn derive_codex_endpoint_session_affinity( payload: &RawJsonPayload, @@ -159,8 +217,8 @@ pub(crate) fn derive_endpoint_affinity_with_headers( let session_id = account_session_with_headers(&body, payload.context(), headers)?; session_affinity( "root-session", - session_id.clone(), - Some(session_id), + &session_id, + Some(&session_id), client_api_key_id, ) .map(|affinity| { @@ -195,8 +253,8 @@ pub(crate) fn derive_live_session_affinity( let session_id = account_session_with_headers(&Map::new(), &Map::new(), &effective)?; session_affinity( "root-session", - session_id.clone(), - Some(session_id), + &session_id, + Some(&session_id), client_api_key_id, ) .map(|affinity| { @@ -216,16 +274,16 @@ pub(crate) fn derive_codex_transport_key( let (source, anchor) = non_empty(request.client_thread_id.as_deref()) .map(|thread| ("thread-transport", thread.to_owned())) .or_else(|| derive_conversation_anchor(request))?; - session_affinity(source, anchor, None, client_api_key_id).map(CodexSessionAffinity::into_key) + session_affinity(source, &anchor, None, client_api_key_id).map(CodexSessionAffinity::into_key) } fn session_affinity( anchor_source: &'static str, - anchor: String, - session_id: Option, + anchor: &str, + session_id: Option<&str>, client_api_key_id: &ClientApiKeyId, ) -> Option { - let session_key = opaque_affinity_key(anchor_source, &anchor)?; + let session_key = opaque_affinity_key(anchor_source, anchor)?; let key = opaque_affinity_key( "client-session", &format!( @@ -240,8 +298,7 @@ fn session_affinity( turn_alias: None, key_hash, anchor_source, - anchor, - session_id, + session_id: session_id.map(str::to_owned), follow_only: false, }) } diff --git a/backend/crates/providers/openai/src/credential/mod.rs b/backend/crates/providers/openai/src/credential/mod.rs index 8651cbb06..3ee195b54 100644 --- a/backend/crates/providers/openai/src/credential/mod.rs +++ b/backend/crates/providers/openai/src/credential/mod.rs @@ -19,7 +19,7 @@ pub mod token_client; mod types; pub(crate) use affinity::{ - CodexSessionAffinity, derive_codex_cyber_policy_session_key, + CodexSessionAffinity, derive_cache_preference, derive_codex_cyber_policy_session_key, derive_codex_endpoint_session_affinity, derive_codex_session_affinity, derive_codex_transport_key, derive_endpoint_affinity_with_headers, derive_live_session_affinity, derive_previous_response_id_hash, derive_turn_alias, diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index 8a6480129..31d12f064 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -249,15 +249,21 @@ impl AffinitySelection { } } - fn telemetry(&self, selected_account: &ProviderAccountId) -> AffinityTelemetry { + fn telemetry( + &self, + selected_account: &ProviderAccountId, + cache_preference: bool, + ) -> AffinityTelemetry { AffinityTelemetry { affinity_hit: self.bound_account.as_ref() == Some(selected_account) && self.escape_reason.is_none(), escape_reason: self.escape_reason, - account_switch: self - .bound_account - .as_ref() - .is_some_and(|bound| bound != selected_account), + // 仅因缓存偏好而分流时，客户端的轮次状态仍没有已知归属。 + account_switch: !cache_preference + && self + .bound_account + .as_ref() + .is_some_and(|bound| bound != selected_account), } } } @@ -514,11 +520,18 @@ impl CodexCredentialSelector { }; let eligibility = request.attempt.account_eligibility(); let diagnostic = eligibility.bypasses_local_eligibility(); - let binding_key = request.session_affinity_key.filter(|_| !diagnostic); - let affinity_mode = request - .attempt - .account_selection_policy() - .openai_account_affinity(); + let mut binding_key = request.session_affinity_key.filter(|_| !diagnostic); + let cache_preference = request + .session_affinity_observation + .is_some_and(CodexSessionAffinity::is_cache_preference); + let affinity_mode = if cache_preference { + AccountAffinity::Preferred + } else { + request + .attempt + .account_selection_policy() + .openai_account_affinity() + }; let mut queue_policy = request.attempt.account_selection_policy().queue_policy(); if follow_only { // 关闭普通账号排队不能放开后代线程换号，沿用队列时间设置并保留有界人数 @@ -608,7 +621,16 @@ impl CodexCredentialSelector { let mut snapshot_retries = 0; 'capacity: loop { let binding = if let Some(key) = binding_key { - self.lookup_session_affinity(key).await? + match self.lookup_session_affinity(key).await { + Ok(binding) => binding, + Err(error) if cache_preference => { + // 本次跳过缓存偏好读写，账号资格、策略和租约仍正常检查。 + tracing::warn!(%error, "OpenAI cache preference lookup unavailable"); + binding_key = None; + None + } + Err(error) => return Err(error), + } } else { None }; @@ -1266,23 +1288,28 @@ impl CodexCredentialSelector { { binding_account = primary.account_id(); } - if let Some(key) = binding_key - && !self + if let Some(key) = binding_key { + match self .admit_session( key, binding.as_ref(), binding_account, - request - .attempt - .account_selection_policy() - .openai_session_affinity_ttl(), + policy.openai_session_affinity_ttl(), ) - .await? - { - drop(guard); - continue 'capacity; + .await + { + Ok(false) => { + drop(guard); + continue 'capacity; + } + Err(error) if cache_preference => { + tracing::warn!(%error, "OpenAI cache preference update unavailable"); + } + Err(error) => return Err(error), + Ok(true) => {} + } } - let affinity_telemetry = affinity.telemetry(account.id()); + let affinity_telemetry = affinity.telemetry(account.id(), cache_preference); let affinity_observation = request.session_affinity_observation; tracing::info!( request_id = %request.attempt.request_id(), @@ -1338,7 +1365,7 @@ impl CodexCredentialSelector { cyber_policy_scope, allows_account_state_mutation, affinity_telemetry, - admitted_session: binding_key.cloned(), + admitted_session: binding_key.filter(|_| !cache_preference).cloned(), capacity: capacity.map(AccountCapacitySnapshot::with_acquired_request), _guard: guard, }); diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 6f1fed660..9daa21820 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -9,7 +9,7 @@ use std::time::{Duration, Instant, SystemTime}; use async_trait::async_trait; use bytes::Bytes; use futures::{StreamExt, future::BoxFuture}; -use gateway_core::account::{AccountFeedbackStats, ProviderAccount}; +use gateway_core::account::{AccountAffinity, AccountFeedbackStats, ProviderAccount}; use gateway_core::engine::continuation::{ContinuationBinding, NativeContinuationScope}; use gateway_core::engine::middleware::MiddlewareHeader; use gateway_core::engine::provider::{ @@ -60,10 +60,10 @@ use crate::credential::{ CodexCredentialRefreshService, CodexCredentialRepository, CodexCredentialSelector, CodexCyberPolicyScope, CodexQuotaRefreshPolicy, CodexSessionAffinity, CredentialSelectionError, RuntimeCodexCookie, SelectCodexCredential, SelectCodexProviderEndpointCredential, - derive_codex_cyber_policy_session_key, derive_codex_endpoint_session_affinity, - derive_codex_session_affinity, derive_codex_transport_key, - derive_endpoint_affinity_with_headers, derive_live_session_affinity, - derive_previous_response_id_hash, derive_turn_alias, + derive_cache_preference, derive_codex_cyber_policy_session_key, + derive_codex_endpoint_session_affinity, derive_codex_session_affinity, + derive_codex_transport_key, derive_endpoint_affinity_with_headers, + derive_live_session_affinity, derive_previous_response_id_hash, derive_turn_alias, }; use crate::request_identity::{account_session_with_headers, turn_id_with_headers}; use crate::session_transport::{CodexSessionTransportRecovery, WS_PAYLOAD_LIMIT}; @@ -680,6 +680,19 @@ impl Provider for CodexProvider { if let Some(prepared) = preselection.as_ref() { prepared.checked_native_continuation_scope(&context)?; } + let cache_preference = preselection + .as_ref() + .filter(|prepared| { + prepared.session_affinity.is_none() + && prepared.previous_session.is_none() + && context.account_selection_policy().openai_account_affinity() + != AccountAffinity::Relaxed + && context.continuation_attempt() == ContinuationAttempt::None + && !guardian + }) + .and_then(|prepared| { + derive_cache_preference(&prepared.upstream, context.client_api_key_ref()) + }); let (selection_session_affinity, selection_cyber_policy_key, requires_websocket) = preselection.map_or(adapter_affinity.unwrap_or((None, None, false)), |prepared| { let requires_websocket = @@ -696,6 +709,10 @@ impl Provider for CodexProvider { requires_websocket, ) }); + // 弱偏好只交给选号器；中间件复验、轮次关联和原生续接仍只认真实会话。 + let account_affinity = selection_session_affinity + .as_ref() + .or(cache_preference.as_ref()); let selection_started_at = Instant::now(); let selection = async { self.selector @@ -704,12 +721,10 @@ impl Provider for CodexProvider { upstream_model: upstream_model.as_str(), request_url: &self.responses_url, attempt: &context, - session_affinity_key: selection_session_affinity - .as_ref() - .map(|affinity| affinity.key()), + session_affinity_key: account_affinity.map(CodexSessionAffinity::key), }, selection_cyber_policy_key.as_ref(), - selection_session_affinity.as_ref(), + account_affinity, requires_websocket, guardian, ) diff --git a/backend/crates/providers/openai/src/transport/request.rs b/backend/crates/providers/openai/src/transport/request.rs index 2086f579b..9580055e5 100644 --- a/backend/crates/providers/openai/src/transport/request.rs +++ b/backend/crates/providers/openai/src/transport/request.rs @@ -415,26 +415,25 @@ fn derive_stable_conversation_key(request: &CodexResponsesRequest) -> Option String { - for item in input { - if item.get("role").and_then(Value::as_str) != Some("user") { - continue; - } - let Some(content) = item.get("content") else { - return String::new(); - }; - if let Some(text) = content.as_str() { - return text.to_owned(); - } - if let Some(parts) = content.as_array() { - return parts - .iter() - .filter(|part| part.get("type").and_then(Value::as_str) == Some("input_text")) - .filter_map(|part| part.get("text").and_then(Value::as_str)) - .collect(); - } - return String::new(); - } - String::new() + input + .iter() + .find(|item| item.get("role").and_then(Value::as_str) == Some("user")) + .into_iter() + .flat_map(message_text_parts) + .collect() +} + +/// 同时接受简写字符串与 Responses 文本块，忽略图片等非文本部分。 +pub(crate) fn message_text_parts(item: &Value) -> impl Iterator + Clone { + let content = &item["content"]; + content.as_str().into_iter().chain( + content + .as_array() + .into_iter() + .flatten() + .filter(|part| part.get("type").and_then(Value::as_str) == Some("input_text")) + .filter_map(|part| part.get("text").and_then(Value::as_str)), + ) } fn normalize_conversation_anchor_text(text: &str) -> String { diff --git a/backend/crates/providers/openai/tests/provider/contract/cache_preference.rs b/backend/crates/providers/openai/tests/provider/contract/cache_preference.rs new file mode 100644 index 000000000..de3fde702 --- /dev/null +++ b/backend/crates/providers/openai/tests/provider/contract/cache_preference.rs @@ -0,0 +1,133 @@ +//! 无显式会话时，选号提示的优先级、隔离边界与完整历史要求 + +use super::*; + +#[tokio::test] +async fn cache_preference_keys_follow_client_hints_or_the_opening_messages() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_affinity").await; + let affinity = Arc::new(MemorySessionAffinity::default()); + let provider = provider_with_affinity(&store, affinity.clone()); + + let select = async |client: &str, model: &str, body: serde_json::Value| { + let before = affinity.lookup_keys().len(); + let attempt = AttemptContext::new( + RequestAttemptContext::new( + ModelRequestId::new("req_cache_key").unwrap(), + ClientApiKeyId::new(client).unwrap(), + ), + NonZeroU32::MIN, + SystemTime::now() + Duration::from_secs(5), + account_policy(), + AccountAttemptContext::new(BTreeSet::new(), None, None) + .with_account_scope(contract_account_scope()), + None, + CancellationToken::new(), + ); + let payload = + ProtocolPayload::json_object("openai", body.as_object().unwrap().clone()).unwrap(); + drop( + provider + .clone() + .execute( + planned_request_for_model( + "openai", + Operation::Generate(GenerateRequest::from_protocol_payload(payload)), + model, + ), + attempt, + ) + .await + .unwrap(), + ); + affinity.lookup_keys().get(before).cloned() + }; + + let model = "gpt-5.4"; + let opening = json!({"instructions":"be brief", "input":[{"role":"developer","content":"review"}, {"role":"user","content":"hello"}]}); + let key = select("client_a", model, opening.clone()).await.unwrap(); + let mut next = opening.clone(); + next["tools"] = json!([{"type":"function","name":"new_tool"}]); + next["input"][1]["content"] = + json!([{"type":"input_text","text":"hel"},{"type":"input_text","text":"lo"}]); + next["input"] + .as_array_mut() + .unwrap() + .push(json!({"role":"user","content":"next turn"})); + assert_eq!(select("client_a", model, next).await.as_ref(), Some(&key)); + assert_ne!( + select("client_b", model, opening.clone()).await.unwrap(), + key + ); + assert_ne!( + select("client_a", "gpt-5.5", opening.clone()) + .await + .unwrap(), + key + ); + for path in ["/instructions", "/input/0/content", "/input/1/content"] { + let mut changed = opening.clone(); + *changed.pointer_mut(path).unwrap() = json!("different"); + assert_ne!(select("client_a", model, changed).await.unwrap(), key); + } + + let chat_opening = json!({"input":[{"role":"system","content":"be brief"},{"role":"developer","content":"review"},{"role":"user","content":"hello"}]}); + assert_eq!( + select("client_a", model, chat_opening).await.as_ref(), + Some(&key) + ); + + let hinted = json!({"prompt_cache_key":"client-hint","input":"hello"}); + let cache = select("client_a", model, hinted.clone()).await.unwrap(); + assert_ne!(key, cache); + assert_ne!(select("client_b", model, hinted).await.unwrap(), cache); + assert_eq!( + select( + "client_a", + model, + json!({"prompt_cache_key":"client-hint", "input":"different"}) + ) + .await, + Some(cache) + ); + for input in [ + json!(" "), + json!([{"role":"assistant","content":"partial history"},{"role":"user","content":"next"}]), + json!([{"role":"user","content":[{"type":"input_image","image_url":"https://invalid.test/x"}]}]), + ] { + assert_eq!( + select( + "client_a", + model, + json!({"instructions":"shared", "input":input}) + ) + .await, + None + ); + } +} + +#[tokio::test] +async fn cache_preference_storage_timeouts_allow_normal_admission() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_affinity").await; + for affinity in [ + MemorySessionAffinity::with_operation_delay(Duration::from_secs(1)), + MemorySessionAffinity::with_initial_claim_barrier(2), + ] { + let affinity = Arc::new(affinity); + let provider = provider_with_affinity(&store, affinity.clone()); + let stream = provider + .execute( + planned_request("openai", generate_operation()), + context("req_cache_store_timeout", CancellationToken::new()), + ) + .await + .unwrap(); + assert_eq!( + stream.metadata().provider_account_id().as_str(), + "acct_affinity" + ); + assert_eq!(affinity.binding_count(), 0); + } +} diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index d0f86c49c..374fc2e8f 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -2,6 +2,7 @@ mod account_eligibility; mod account_isolation; +mod cache_preference; mod capacity; mod cookies; mod correlation; @@ -7095,8 +7096,14 @@ async fn account_selection_log_should_include_affinity_observation_fields() { let conversation = selected_account_log_fields(&events, "req_affinity_observation_conversation"); - assert_eq!(conversation["affinity_anchor_source"], ""); - assert_eq!(conversation["affinity_anchor"], ""); + assert_eq!( + conversation["affinity_anchor_source"], + "client-cache-preference" + ); + assert_eq!( + conversation["affinity_anchor"], + conversation["affinity_key_hash"] + ); assert_eq!(conversation["session_id"], ""); assert_eq!(conversation["session_id_present"], false); assert_ne!( @@ -7106,7 +7113,7 @@ async fn account_selection_log_should_include_affinity_observation_fields() { } #[tokio::test] -async fn prompt_cache_key_alone_does_not_create_an_account_binding() { +async fn relaxed_mode_skips_cache_preferences() { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_affinity").await; let affinity = Arc::new(MemorySessionAffinity::default()); @@ -7125,7 +7132,12 @@ async fn prompt_cache_key_alone_does_not_create_an_account_binding() { let stream = provider_with_affinity(&store, Arc::clone(&affinity)) .execute( planned_request("openai", Operation::Generate(generation)), - context("req_affinity_key", CancellationToken::new()), + session_binding::context_with_affinity_policy( + "req_affinity_key", + contract_account_scope(), + account_policy() + .with_openai_account_affinity(gateway_core::account::AccountAffinity::Relaxed), + ), ) .await .expect("prepare provider stream"); diff --git a/backend/crates/providers/openai/tests/provider/contract/session_binding.rs b/backend/crates/providers/openai/tests/provider/contract/session_binding.rs index 7ee49ceeb..fc14e70de 100644 --- a/backend/crates/providers/openai/tests/provider/contract/session_binding.rs +++ b/backend/crates/providers/openai/tests/provider/contract/session_binding.rs @@ -489,7 +489,23 @@ async fn final_middleware_child_identity_can_claim_a_missing_binding() { stream.metadata().provider_account_id().as_str(), "acct_subagent_a" ); - assert_eq!(affinity.binding_count(), 1); + drop(stream); + // 初始正文的缓存偏好与中间件最终声明的会话分别准入；后续显式会话直接复用后者。 + let claimed = affinity.lookup_keys(); + assert_eq!(claimed.iter().collect::>().len(), 2); + let follow = provider_with_affinity_and_base_url(&store, affinity.clone(), server.uri()) + .execute( + planned_request("openai", turn_request("new-session", "next")), + context("req_final_child_follow", CancellationToken::new()), + ) + .await + .unwrap(); + assert_eq!( + follow.metadata().provider_account_id().as_str(), + "acct_subagent_a" + ); + assert_eq!(affinity.lookup_keys().last(), claimed.last()); + assert_eq!(affinity.binding_count(), 2); assert!(server.received_requests().await.unwrap().is_empty()); } @@ -1227,7 +1243,7 @@ fn preferred_context_with_ttl( scope: Arc, ttl: Duration, ) -> AttemptContext { - preferred_context_with_policy( + context_with_affinity_policy( request_id, scope, account_policy() @@ -1236,7 +1252,7 @@ fn preferred_context_with_ttl( ) } -fn preferred_context_with_policy( +pub(super) fn context_with_affinity_policy( request_id: &str, scope: Arc, policy: gateway_core::account::AccountSelectionPolicy, @@ -1315,7 +1331,7 @@ async fn relaxed_and_preferred_session_requests_apply_distinct_scheduling_polici }; provider.clone().execute( request, - preferred_context_with_policy( + context_with_affinity_policy( "req_preferred_session", contract_account_scope(), policy, @@ -1699,6 +1715,15 @@ async fn preferred_final_thread_rewrite_within_the_session_preserves_the_primary #[tokio::test] async fn preferred_spillover_uses_selected_credentials_on_the_wire() { + assert_spillover_on_wire(false).await; +} + +#[tokio::test] +async fn content_preference_spills_over_even_in_strict_mode_and_preserves_turn_state() { + assert_spillover_on_wire(true).await; +} + +async fn assert_spillover_on_wire(anonymous: bool) { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_subagent_a").await; let affinity = Arc::new(MemorySessionAffinity::default()); @@ -1731,12 +1756,38 @@ async fn preferred_spillover_uses_selected_credentials_on_the_wire() { } else { leases.busy_accounts.lock().unwrap().clear(); } - let mut stream = provider - .clone() - .execute( - planned_request("openai", turn_request("root", turn)), + let (operation, attempt) = if anonymous { + let mut input = vec![ + json!({"role":"developer", "content":"review"}), + json!({"role":"user","content":"check this patch"}), + ]; + if turn != "first" { + input.push(json!({"role":"assistant","content":"ready"})); + input.push(json!({"role":"user","content":turn})); + } + let payload = ProtocolPayload::json_object( + "openai", + Map::from_iter([("input".into(), json!(input))]), + ) + .unwrap() + .with_context(Map::from_iter([ + ("use_websocket".to_owned(), json!(false)), + ("turn_id".to_owned(), json!(turn)), + ("turn_state".to_owned(), json!("unowned-turn-state")), + ])); + ( + Operation::Generate(GenerateRequest::from_protocol_payload(payload)), + context("req_preference_wire", CancellationToken::new()), + ) + } else { + ( + turn_request("root", turn), preferred_context("req_preferred_wire", contract_account_scope()), ) + }; + let mut stream = provider + .clone() + .execute(planned_request("openai", operation), attempt) .await .unwrap(); while let Some(event) = stream.next().await { @@ -1757,6 +1808,22 @@ async fn preferred_spillover_uses_selected_credentials_on_the_wire() { ] ); assert_eq!(affinity.binding_count(), 1); + if anonymous { + assert!(affinity.alias_ttls().is_empty()); + for request in &sent { + assert!( + captured_request_body(request) + .get("prompt_cache_key") + .is_none() + ); + assert!(!request.headers.contains_key("session-id")); + assert!(!request.headers.contains_key("session_id")); + assert_eq!( + captured_header_values(request, "x-codex-turn-state"), + vec![b"unowned-turn-state".to_vec()] + ); + } + } server.verify().await; } @@ -1801,7 +1868,7 @@ async fn preferred_primary_migration_distinguishes_temporary_and_persistent_unav let select = || { provider.clone().execute( planned_request("openai", turn_request("root", "availability-turn")), - preferred_context_with_policy( + context_with_affinity_policy( "req_primary_availability", contract_account_scope(), policy, diff --git a/docs/api.md b/docs/api.md index acf84d0a3..cd77059d0 100644 --- a/docs/api.md +++ b/docs/api.md @@ -2115,6 +2115,11 @@ Guardian 以 `subagent_kind` 或 `client_metadata.x-openai-subagent` 值 `guardi 插件显式选号及原生续写的状态归属约束在三种模式下仍然生效 +Chat Completions 和 Responses 通过原生 OpenAI Provider 选号时，缺少显式会话 ID 的请求默认在同一 Client Key 下优先复用账号。 +网关先匹配客户端传入的 `prompt_cache_key`，否则按上游模型、开头的指令和首条用户消息的文本匹配。 +后续消息及工具列表不参与匹配。即使在严格模式下，首选账号繁忙时也可以临时换号。 +宽松模式、没有缓存键又缺少对话开头文本的请求，仍按普通调度选号。此偏好只用于选号 + `openaiSessionAffinityTtlHours` 是账号绑定及其会话关联的滑动保留时长，单位小时，默认 24，取值 1～720。 各类请求统一在发送前成功准入时按请求冻结值续期。 响应完成不回写或续期。调整时长不扫描已有 Redis 键，已有记录保留原到期时间，下一次成功准入时使用新值； diff --git a/docs/architecture.md b/docs/architecture.md index 50f102e52..ee5a2c48b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -796,7 +796,7 @@ Continuation 受原请求的 Client Key、账号范围、Provider 和发送/交 - scope 外账号与跨 Key 复用均拒绝；发送结果不明时，仅接受 Provider 明确给出的有界传输恢复意图 OpenAI Provider 解释官方逻辑 `session_id` 与 `thread_id`，按 Client Key 隔离账号绑定， -不将缓存路由键或请求内容推断为账号身份。账号亲和模式随请求策略冻结： +账号亲和模式随请求策略冻结： - 宽松：会话内请求直接按调度策略选号，不使用主账号偏好，成功准入后更新当前账号 - 优先模式在首次成功准入时认领主账号，短时容量或单模型限制触发临时分流 @@ -817,8 +817,10 @@ OpenAI 固定归属请求直接读取目标账号，其余请求按冻结的账 目标失效、租约争用或凭据版本冲突触发完整刷新，策略插件和受管适配器使用完整候选评估 Store 只保存带版本的绑定事实，Provider 在取得租约后、发送前原子认领、续期或按模式迁移；快照冲突释放租约并重新选择。 -首次认领不依赖根线程先发请求，独立后台任务也使用自身会话绑定；没有可靠会话身份时正常选号，不推测会话关系。 -响应完成不写回账号，迁移前已准入的请求可以完成，不能把绑定改回旧账号。绑定存储不可用时不绕过绑定发送 +首次认领不依赖根线程先发请求，独立后台任务也使用自身会话绑定。 +无显式会话的 OpenAI 生成请求可按客户端缓存键或对话开头形成账号偏好，复用优先模式的准入与过期机制。 +偏好键按 Client Key 隔离，与会话键使用不同命名空间。偏好存储不可用时按普通调度继续选号。 +响应完成不写回账号，迁移前已准入的请求可以完成，不能把绑定改回旧账号。显式会话的绑定存储不可用时不绕过绑定发送 官方独立 Images 仅携带 `turn_id`，通过已观测 Responses 轮次关联到会话，关联同时保留根线程或后代线程的调度权限。 轮次关联保存不透明会话键，不复制账号；显式会话身份只用于校验轮次归属。