mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 16:06:08 +00:00
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 <noreply@openai.com>
This commit is contained in:
@@ -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<String>,
|
||||
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<CodexSessionAffinity> {
|
||||
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<Item = &'a str> + 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<String>,
|
||||
anchor: &str,
|
||||
session_id: Option<&str>,
|
||||
client_api_key_id: &ClientApiKeyId,
|
||||
) -> Option<CodexSessionAffinity> {
|
||||
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,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
});
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -415,26 +415,25 @@ fn derive_stable_conversation_key(request: &CodexResponsesRequest) -> Option<Str
|
||||
}
|
||||
|
||||
fn first_user_text(input: &[Value]) -> 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<Item = &str> + 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 {
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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");
|
||||
|
||||
@@ -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::<BTreeSet<_>>().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<FrozenAccountScope>,
|
||||
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<FrozenAccountScope>,
|
||||
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,
|
||||
|
||||
@@ -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 键,已有记录保留原到期时间,下一次成功准入时使用新值;
|
||||
|
||||
@@ -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 轮次关联到会话,关联同时保留根线程或后代线程的调度权限。
|
||||
轮次关联保存不透明会话键,不复制账号;显式会话身份只用于校验轮次归属。
|
||||
|
||||
Reference in New Issue
Block a user