From db849c1251005c527e23165ee45a159c3da68237 Mon Sep 17 00:00:00 2001 From: Zyy Date: Sun, 1 Nov 2026 19:24:17 +0800 Subject: [PATCH] feat(peer-link)!: simplify the CPR Link protocol Make capability blocks optional and represent key budgets as named windows. Centralize validation and preserve unknown states for missing observations. BREAKING CHANGE: CPR Link requires schema_version 2; v1 snapshots are no longer accepted. Co-authored-by: Codex --- .../use_case/peer_link/upstream/projection.rs | 72 +++-- .../src/use_case/peer_link/upstream/quota.rs | 23 +- .../use_case/peer_link/upstream/scheduling.rs | 21 +- .../use_case/peer_link/upstream/snapshot.rs | 253 ++++++++++-------- .../tests/use_case/peer_link/upstream/mod.rs | 7 +- .../gateway-api/src/openai/peer_link/wire.rs | 78 +++--- .../gateway-api/tests/openai/peer_link/mod.rs | 10 +- backend/crates/gateway-protocol/src/peer.rs | 63 ++--- .../providers/openai/tests/admin/peer_info.rs | 217 ++++++++++++--- deploy/README.md | 6 +- docs/api.md | 177 ++++++++---- docs/architecture.md | 3 +- 12 files changed, 592 insertions(+), 338 deletions(-) diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs index 9663e6b2b..3eb1f04a7 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs @@ -12,9 +12,7 @@ use crate::model::{ }, }; use gateway_core::account::{QuotaAccessState, UpstreamConnection, UpstreamInfoFailure}; -use gateway_protocol::peer::{ - PeerComponentStatus, PeerLink, PeerPoolDisclosure, PeerQuotaDisclosure, -}; +use gateway_protocol::peer::{PeerComponentStatus, PeerLink}; use super::{PeerInfoService, quota, snapshot}; @@ -39,11 +37,11 @@ impl PeerInfoService { last_error: last_error.map(|error| error.as_str().to_owned()), pool: peer .as_ref() - .filter(|peer| peer.pool.disclosure == PeerPoolDisclosure::Summary) - .and_then(|peer| { - let states = peer.pool.states.as_ref()?; - Some(UpstreamPool { - account_count: peer.pool.account_count?, + .and_then(|peer| peer.pool.as_ref()) + .map(|pool| { + let states = &pool.states; + UpstreamPool { + account_count: pool.account_count, states: UpstreamPoolStates { ready: states.ready, cooldown: states.cooldown, @@ -52,8 +50,7 @@ impl PeerInfoService { error: states.error, unknown: states.unknown, }, - plans: peer - .pool + plans: pool .plans .as_deref() .unwrap_or_default() @@ -63,19 +60,18 @@ impl PeerInfoService { count: plan.count, }) .collect(), - }) + } }), }; let mut windows = Vec::new(); if let Some(peer) = &peer { - if peer.usage.status == PeerComponentStatus::Available - && let Some(usage) = &peer.usage.data + if let Some(component) = &peer.usage + && component.status == PeerComponentStatus::Available + && let Some(usage) = &component.data + && let Some(unit) = &usage.unit { - for (key, label, seconds, window) in [ - ("daily", "日额度", 86400, &usage.daily), - ("weekly", "周额度", 604800, &usage.weekly), - ] { - // 不限额仍有日、周账本，只是不计算占用百分比 + for window in &usage.windows { + // 不限额仍展示已提供的账本窗口，只是不计算占用百分比 let total = window .total .as_deref() @@ -84,19 +80,17 @@ impl PeerInfoService { let used_percent = snapshot::decimal(&window.used) .zip(total) .map(|(used, total)| (used / total * 100.0).clamp(0.0, 100.0)); - let window_observed = peer - .usage + let window_observed = component .observed_at .as_deref() .and_then(snapshot::datetime); let window_stale = stale - || !peer - .usage + || !component .observed_at .as_deref() .is_some_and(|value| snapshot::fresh(value, now)) || window - .resets_at + .reset_at .as_deref() .and_then(snapshot::timestamp) .is_some_and(|reset| reset <= now); @@ -106,27 +100,28 @@ impl PeerInfoService { observed_at: window_observed, stale: window_stale, amount: Some(UpstreamQuotaAmount { - unit: usage.unit.clone(), + unit: unit.clone(), used: window.used.clone(), total: window.total.clone(), remaining: window.remaining.clone(), }), }), - key: format!("peer:budget:{key}"), + key: format!("peer:budget:{}", window.key), group: "general".to_owned(), - label: label.to_owned(), + label: window.label.as_deref().unwrap_or(&window.key).to_owned(), limit_id: None, limit_name: None, - role: Some(if key == "weekly" { - ProviderQuotaWindowRole::Secondary - } else { - ProviderQuotaWindowRole::Primary - }), + role: match window.key.as_str() { + "daily" => Some(ProviderQuotaWindowRole::Primary), + "weekly" => Some(ProviderQuotaWindowRole::Secondary), + "monthly" => Some(ProviderQuotaWindowRole::Monthly), + _ => None, + }, // Key 消费窗口覆盖此连接的全部请求，本机用量按同一时间边界聚合 local_usage_attribution: QuotaLocalUsageAttribution::AccountWide, - window_seconds: Some(seconds), + window_seconds: window.window_seconds, used_percent, - reset_at: window.resets_at.as_deref().and_then(snapshot::datetime), + reset_at: window.reset_at.as_deref().and_then(snapshot::datetime), limit_reached: !window_stale && quota::window_access(window, now) == QuotaAccessState::Exhausted, local_usage: None, @@ -135,8 +130,8 @@ impl PeerInfoService { } } // 池内窗口各自保留账号归属，不能将不同套餐相加为统一消费额度 - if peer.quota.disclosure == PeerQuotaDisclosure::Accounts { - for (index, account) in peer.quota.accounts.iter().enumerate() { + if let Some(quota) = &peer.quota { + for (index, account) in quota.accounts.iter().enumerate() { let window_observed = account.observed_at.as_deref().and_then(snapshot::datetime); let window_stale = stale @@ -146,6 +141,7 @@ impl PeerInfoService { now, ); for window in &account.windows { + let group = window.group.as_deref().unwrap_or("general"); let window_stale = window_stale || window .reset_at @@ -156,7 +152,7 @@ impl PeerInfoService { "peer:{}:{}:{}", account.id.len(), account.id, - window.limit_id.as_deref().unwrap_or(&window.group) + window.limit_id.as_deref().unwrap_or(group) ); windows.push(ProviderQuotaWindow { upstream: Some(UpstreamQuotaWindow { @@ -166,8 +162,8 @@ impl PeerInfoService { amount: None, }), key: format!("{scope}:{}", window.key), - group: window.group.clone(), - label: window.label.clone(), + group: group.to_owned(), + label: window.label.as_deref().unwrap_or(&window.key).to_owned(), limit_id: Some(scope), limit_name: Some(match window.limit_name.as_deref() { Some(name) => format!("账号 {} · {name}", index + 1), diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/quota.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/quota.rs index 1f3ba9472..5c58d8ca0 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/quota.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/quota.rs @@ -13,24 +13,26 @@ pub(super) fn observe( started_at: SystemTime, now: SystemTime, ) -> Option { + let usage = peer.usage.as_ref()?; if !fresh(&peer.observed_at, now) || !component_fresh(Some(&peer.observed_at), Some(&peer.expires_at), now) - || peer.usage.status != PeerComponentStatus::Available - || !component_fresh(peer.usage.observed_at.as_deref(), None, now) + || usage.status != PeerComponentStatus::Available + || !component_fresh(usage.observed_at.as_deref(), None, now) { return None; } let observed_at = started_at .min(timestamp(&peer.observed_at)?) - .min(timestamp(peer.usage.observed_at.as_deref()?)?); - let usage = peer.usage.data.as_ref()?; + .min(timestamp(usage.observed_at.as_deref()?)?); + let usage = usage.data.as_ref()?; match access(usage, now) { QuotaAccessState::Allowed => Some(QuotaState::allowed(observed_at)), QuotaAccessState::Exhausted => { - let reset_at = [&usage.daily, &usage.weekly] - .into_iter() + let reset_at = usage + .windows + .iter() .filter(|window| window_access(window, now) == QuotaAccessState::Exhausted) - .filter_map(|window| window.resets_at.as_deref().and_then(timestamp)) + .filter_map(|window| window.reset_at.as_deref().and_then(timestamp)) .min(); Some(QuotaState::exhausted( if usage.allowed == Some(false) { @@ -51,8 +53,9 @@ pub(super) fn access(usage: &PeerKeyUsage, now: SystemTime) -> QuotaAccessState match usage.allowed { Some(true) => QuotaAccessState::Allowed, Some(false) => QuotaAccessState::Exhausted, - None if [&usage.daily, &usage.weekly] - .into_iter() + None if usage + .windows + .iter() .any(|window| window_access(window, now) == QuotaAccessState::Exhausted) => { QuotaAccessState::Exhausted @@ -64,7 +67,7 @@ pub(super) fn access(usage: &PeerKeyUsage, now: SystemTime) -> QuotaAccessState /// 过期窗口保持未知，不从重置时间或金额推断恢复 pub(super) fn window_access(window: &PeerBudgetWindow, now: SystemTime) -> QuotaAccessState { if window - .resets_at + .reset_at .as_deref() .and_then(timestamp) .is_some_and(|reset| reset <= now) diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/scheduling.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/scheduling.rs index 92b22e018..11b769146 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/scheduling.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/scheduling.rs @@ -70,22 +70,23 @@ pub(super) struct SchedulingObservation { impl SchedulingObservation { pub(super) fn new(peer: &PeerLink, now: SystemTime) -> Option { let validity = Validity::component(Some(&peer.observed_at), Some(&peer.expires_at))?; - let mut exhausted_budgets = Vec::with_capacity(2); - if peer.usage.status == PeerComponentStatus::Available - && let Some(usage) = &peer.usage.data + let mut exhausted_budgets = Vec::new(); + if let Some(component) = &peer.usage + && component.status == PeerComponentStatus::Available + && let Some(usage) = &component.data && let Some(usage_validity) = - Validity::component(peer.usage.observed_at.as_deref(), None) + Validity::component(component.observed_at.as_deref(), None) && quota::access(usage, now) == QuotaAccessState::Exhausted { if usage.allowed.is_some() { exhausted_budgets.push(usage_validity); } else { // 窗口提供的补充证据只在各自重置边界之前影响调度评分 - for window in [&usage.daily, &usage.weekly] { + for window in &usage.windows { if quota::window_access(window, now) == QuotaAccessState::Exhausted { let mut validity = usage_validity; if let Some(reset) = - window.resets_at.as_deref().and_then(snapshot::timestamp) + window.reset_at.as_deref().and_then(snapshot::timestamp) { validity.expires_at = validity.expires_at.min(reset); } @@ -97,7 +98,11 @@ impl SchedulingObservation { // 同一账号重复列出模型不增加候选数，多账号覆盖不计算池平均百分比 let mut model_accounts = HashMap::<&str, Option>::new(); - for (index, account) in peer.quota.accounts.iter().enumerate() { + let quota_accounts = peer + .quota + .as_ref() + .map_or(&[][..], |quota| quota.accounts.as_slice()); + for (index, account) in quota_accounts.iter().enumerate() { for model in &account.models { model_accounts .entry(model) @@ -128,7 +133,7 @@ impl SchedulingObservation { accounts .entry(index) .or_insert_with(|| { - let account = &peer.quota.accounts[index]; + let account = "a_accounts[index]; let validity = Validity::component( account.observed_at.as_deref(), account.expires_at.as_deref(), diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs index fa1117540..06a395d3d 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs @@ -1,12 +1,15 @@ //! 互联响应的有界校验、能力授权与时间有效性解释 -use std::time::{Duration, SystemTime}; +use std::{ + collections::HashSet, + time::{Duration, SystemTime}, +}; use chrono::{DateTime, Utc}; use gateway_core::account::{OpaqueProviderData, UpstreamInfoFailure}; use gateway_protocol::peer::{ - MAX_ACCOUNTS, MAX_MODELS, MAX_QUOTA_WINDOWS, OBJECT, PeerLink, PeerPoolDisclosure, - SCHEMA_VERSION, + MAX_ACCOUNTS, MAX_MODELS, MAX_QUOTA_WINDOWS, OBJECT, PeerKeyUsage, PeerLink, PeerPool, + PeerQuota, SCHEMA_VERSION, }; use serde::Deserialize as _; @@ -51,14 +54,15 @@ pub(super) fn component_fresh( } pub(super) fn decode(data: &OpaqueProviderData) -> Option { - PeerLink::deserialize( + let peer = PeerLink::deserialize( serde::de::value::MapDeserializer::<_, serde_json::Error>::new( data.expose_to_provider() .iter() .map(|(key, value)| (key.as_str(), value)), ), ) - .ok() + .ok()?; + (peer.object == OBJECT && peer.schema_version == SCHEMA_VERSION).then_some(peer) } pub(super) fn parse( @@ -67,11 +71,17 @@ pub(super) fn parse( ) -> Result { let mut peer: PeerLink = serde_json::from_slice(bytes).map_err(|_| UpstreamInfoFailure::InvalidResponse)?; - let valid_text = |value: &str, max: usize| { - !value.is_empty() && value.len() <= max && !value.chars().any(char::is_control) - }; let observed = timestamp(&peer.observed_at).ok_or(UpstreamInfoFailure::InvalidResponse)?; let expires = timestamp(&peer.expires_at).ok_or(UpstreamInfoFailure::InvalidResponse)?; + let accounts = peer + .quota + .as_ref() + .map_or(&[][..], |quota| quota.accounts.as_slice()); + let budget_windows = peer + .usage + .as_ref() + .and_then(|usage| usage.data.as_ref()) + .map_or(0, |usage| usage.windows.len()); if peer.object != OBJECT || peer.schema_version != SCHEMA_VERSION || observed > now + Duration::from_secs(60) @@ -79,125 +89,39 @@ pub(super) fn parse( || expires <= observed || expires > observed + MAX_AGE || peer.capabilities.len() > 32 - || peer - .quota - .accounts + || accounts .iter() .map(|account| account.windows.len()) .sum::() > MAX_QUOTA_WINDOWS - || peer.quota.accounts.len() > MAX_ACCOUNTS + || budget_windows > MAX_QUOTA_WINDOWS + || accounts.len() > MAX_ACCOUNTS || peer.models.len() > MAX_MODELS { return Err(UpstreamInfoFailure::InvalidResponse); } // 未授权能力不能借同名数据块越过逐层公开边界 - if !peer.capabilities.iter().any(|value| value == "key_usage") { - peer.usage.data = None; + let disclosed = |capability| peer.capabilities.iter().any(|value| value == capability); + if !disclosed("key_usage") { + peer.usage = None; } - if !peer - .capabilities - .iter() - .any(|value| value == "pool_summary") - { - peer.pool = Default::default(); + if !disclosed("pool_summary") { + peer.pool = None; } - if !peer - .capabilities - .iter() - .any(|value| value == "quota_windows") - { - peer.quota = Default::default(); + if !disclosed("quota_windows") { + peer.quota = None; } - if !peer - .capabilities - .iter() - .any(|value| value == "model_availability") - { + if !disclosed("model_availability") { peer.models.clear(); } - if peer.pool.disclosure == PeerPoolDisclosure::Summary { - let count = peer - .pool - .account_count - .ok_or(UpstreamInfoFailure::InvalidResponse)?; - let states = peer - .pool - .states - .as_ref() - .ok_or(UpstreamInfoFailure::InvalidResponse)?; - let state_total = [ - states.ready, - states.cooldown, - states.exhausted, - states.disabled, - states.error, - states.unknown, - ] - .into_iter() - .map(u64::from) - .sum::(); - if u64::from(count) > MAX_ACCOUNTS as u64 || state_total != u64::from(count) { - return Err(UpstreamInfoFailure::InvalidResponse); - } - if let Some(plans) = &peer.pool.plans - && (plans.len() > MAX_ACCOUNTS - || plans - .iter() - .any(|plan| !valid_text(&plan.name, 256) || plan.count > count) - || plans.iter().map(|plan| u64::from(plan.count)).sum::() > u64::from(count)) - { - return Err(UpstreamInfoFailure::InvalidResponse); - } + if let Some(pool) = &peer.pool { + validate_pool(pool)?; } - if let Some(usage) = &peer.usage.data { - if !valid_text(&usage.unit, 16) { - return Err(UpstreamInfoFailure::InvalidResponse); - } - for window in [&usage.daily, &usage.weekly] { - if [ - Some(&window.used), - window.total.as_ref(), - window.remaining.as_ref(), - ] - .into_iter() - .flatten() - .any(|value| decimal(value).is_none()) - || window - .resets_at - .as_ref() - .is_some_and(|value| timestamp(value).is_none()) - { - return Err(UpstreamInfoFailure::InvalidResponse); - } - } + if let Some(usage) = peer.usage.as_ref().and_then(|usage| usage.data.as_ref()) { + validate_usage(usage)?; } - for account in &peer.quota.accounts { - if !valid_text(&account.id, 256) - || account.models.len() > MAX_MODELS - || account.windows.iter().any(|window| { - !valid_text(&window.key, 256) - || !valid_text(&window.label, 256) - || !valid_text(&window.group, 128) - || window - .limit_id - .as_deref() - .is_some_and(|value| !valid_text(value, 256)) - || window - .limit_name - .as_deref() - .is_some_and(|value| !valid_text(value, 256)) - || window - .used_percent - .is_some_and(|value| !value.is_finite() || !(0.0..=100.0).contains(&value)) - || window - .reset_at - .as_ref() - .is_some_and(|value| timestamp(value).is_none()) - }) - { - return Err(UpstreamInfoFailure::InvalidResponse); - } + if let Some(quota) = &peer.quota { + validate_quota(quota)?; } for model in &peer.models { if !valid_text(&model.model, 256) @@ -219,6 +143,113 @@ pub(super) fn parse( } } +fn valid_text(value: &str, max: usize) -> bool { + !value.is_empty() && value.len() <= max && !value.chars().any(char::is_control) +} + +fn validate_pool(pool: &PeerPool) -> Result<(), UpstreamInfoFailure> { + let count = pool.account_count; + let states = &pool.states; + let state_total = [ + states.ready, + states.cooldown, + states.exhausted, + states.disabled, + states.error, + states.unknown, + ] + .into_iter() + .map(u64::from) + .sum::(); + if u64::from(count) > MAX_ACCOUNTS as u64 || state_total != u64::from(count) { + return Err(UpstreamInfoFailure::InvalidResponse); + } + if let Some(plans) = &pool.plans + && (plans.len() > MAX_ACCOUNTS + || plans + .iter() + .any(|plan| !valid_text(&plan.name, 256) || plan.count > count) + || plans.iter().map(|plan| u64::from(plan.count)).sum::() > u64::from(count)) + { + return Err(UpstreamInfoFailure::InvalidResponse); + } + Ok(()) +} + +fn validate_usage(usage: &PeerKeyUsage) -> Result<(), UpstreamInfoFailure> { + if usage + .unit + .as_deref() + .is_some_and(|unit| !valid_text(unit, 16)) + || !usage.windows.is_empty() && usage.unit.is_none() + { + return Err(UpstreamInfoFailure::InvalidResponse); + } + let mut keys = HashSet::new(); + for window in &usage.windows { + if !valid_text(&window.key, 256) + || !keys.insert(&window.key) + || window + .label + .as_deref() + .is_some_and(|label| !valid_text(label, 256)) + || window.window_seconds == Some(0) + || [ + Some(&window.used), + window.total.as_ref(), + window.remaining.as_ref(), + ] + .into_iter() + .flatten() + .any(|value| decimal(value).is_none()) + || window + .reset_at + .as_ref() + .is_some_and(|value| timestamp(value).is_none()) + { + return Err(UpstreamInfoFailure::InvalidResponse); + } + } + Ok(()) +} + +fn validate_quota(quota: &PeerQuota) -> Result<(), UpstreamInfoFailure> { + for account in "a.accounts { + if !valid_text(&account.id, 256) + || account.models.len() > MAX_MODELS + || account.windows.iter().any(|window| { + !valid_text(&window.key, 256) + || window + .label + .as_deref() + .is_some_and(|label| !valid_text(label, 256)) + || window + .group + .as_deref() + .is_some_and(|group| !valid_text(group, 128)) + || window + .limit_id + .as_deref() + .is_some_and(|value| !valid_text(value, 256)) + || window + .limit_name + .as_deref() + .is_some_and(|value| !valid_text(value, 256)) + || window + .used_percent + .is_some_and(|value| !value.is_finite() || !(0.0..=100.0).contains(&value)) + || window + .reset_at + .as_ref() + .is_some_and(|value| timestamp(value).is_none()) + }) + { + return Err(UpstreamInfoFailure::InvalidResponse); + } + } + Ok(()) +} + pub(super) fn decimal(value: &str) -> Option { if value.is_empty() || value.len() > 64 diff --git a/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs b/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs index c75ac961d..8fc867c04 100644 --- a/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs @@ -390,14 +390,13 @@ impl PeerTransport for TestTransport { .push(account.id().as_str().to_owned()); let now = Utc::now(); Ok(serde_json::to_vec(&serde_json::json!({ - "object":"cpr.peer", "schema_version":1, + "object":"cpr.peer", "schema_version":2, "observed_at":now.to_rfc3339(), "expires_at":(now + chrono::Duration::minutes(5)).to_rfc3339(), "capabilities":["key_usage"], "usage":{"status":"available","observed_at":now.to_rfc3339(),"data":{"unit":"USD","allowed":false, - "daily":{"total":"1","used":"1","remaining":"0","resetsAt":(now + chrono::Duration::days(1)).to_rfc3339()}, - "weekly":{"total":"0","used":"0","remaining":"0"}}}, - "pool":{"disclosure":"none"}, "quota":{"disclosure":"none"}, "models":[] + "windows":[{"key":"daily","total":"1","used":"1","remaining":"0","reset_at":(now + chrono::Duration::days(1)).to_rfc3339()}]}}, + "models":[] })).unwrap()) } } diff --git a/backend/crates/gateway-api/src/openai/peer_link/wire.rs b/backend/crates/gateway-api/src/openai/peer_link/wire.rs index b2c7ac0e2..e953c68ad 100644 --- a/backend/crates/gateway-api/src/openai/peer_link/wire.rs +++ b/backend/crates/gateway-api/src/openai/peer_link/wire.rs @@ -5,34 +5,31 @@ use gateway_admin::model::peer_link::{PeerLinkExport, PeerModelState as ModelSta use gateway_core::metering::Decimal; use gateway_protocol::peer::{ OBJECT, PeerBudgetWindow, PeerComponentStatus, PeerKeyUsage, PeerLink, PeerModelAvailability, - PeerModelCoverage, PeerModelState, PeerPlanCount, PeerPool, PeerPoolDisclosure, PeerPoolStates, - PeerQuota, PeerQuotaAccount, PeerQuotaDisclosure, PeerQuotaWindow, PeerUsage, SCHEMA_VERSION, + PeerModelCoverage, PeerModelState, PeerPlanCount, PeerPool, PeerPoolStates, PeerQuota, + PeerQuotaAccount, PeerQuotaWindow, PeerUsage, SCHEMA_VERSION, }; use std::time::SystemTime; pub(super) fn response(export: PeerLinkExport) -> PeerLink { let mut capabilities = vec!["key_usage".to_owned()]; - let pool = export.pool.map_or_else(PeerPool::default, |pool| { + let pool = export.pool.map(|pool| { capabilities.push("pool_summary".to_owned()); PeerPool { - disclosure: PeerPoolDisclosure::Summary, observed_at: Some(export.observed_at.to_rfc3339()), - account_count: Some( - pool.ready - + pool.cooldown - + pool.exhausted - + pool.disabled - + pool.error - + pool.unknown, - ), - states: Some(PeerPoolStates { + account_count: pool.ready + + pool.cooldown + + pool.exhausted + + pool.disabled + + pool.error + + pool.unknown, + states: PeerPoolStates { ready: pool.ready, cooldown: pool.cooldown, exhausted: pool.exhausted, disabled: pool.disabled, error: pool.error, unknown: pool.unknown, - }), + }, plans: pool.plans.map(|plans| { plans .into_iter() @@ -41,10 +38,9 @@ pub(super) fn response(export: PeerLinkExport) -> PeerLink { }), } }); - let quota = export.quota.map_or_else(PeerQuota::default, |accounts| { + let quota = export.quota.map(|accounts| { capabilities.extend(["quota_windows".to_owned(), "model_availability".to_owned()]); PeerQuota { - disclosure: PeerQuotaDisclosure::Accounts, accounts: accounts .into_iter() .map(|account| PeerQuotaAccount { @@ -57,8 +53,8 @@ pub(super) fn response(export: PeerLinkExport) -> PeerLink { .into_iter() .map(|window| PeerQuotaWindow { key: window.key, - group: window.group, - label: window.label, + group: Some(window.group), + label: Some(window.label), limit_id: window.limit_id, limit_name: window.limit_name, role: window.role.map(|role| role.as_str().to_owned()), @@ -78,26 +74,34 @@ pub(super) fn response(export: PeerLinkExport) -> PeerLink { capabilities, observed_at: export.observed_at.to_rfc3339(), expires_at: export.expires_at.to_rfc3339(), - usage: PeerUsage { + usage: Some(PeerUsage { status: PeerComponentStatus::Available, observed_at: Some(export.budget_observed_at.to_rfc3339()), data: Some(PeerKeyUsage { - unit: "USD".to_owned(), + unit: Some("USD".to_owned()), allowed: Some(export.budget_access.allowed()), - daily: budget_window( - export.budget.limits.daily_usd, - export.budget.daily_used_usd, - export.budget.daily_resets_at, - export.budget_access.daily_limit_reached, - ), - weekly: budget_window( - export.budget.limits.weekly_usd, - export.budget.weekly_used_usd, - export.budget.weekly_resets_at, - export.budget_access.weekly_limit_reached, - ), + windows: vec![ + budget_window( + "daily", + "日额度", + 86400, + export.budget.limits.daily_usd, + export.budget.daily_used_usd, + export.budget.daily_resets_at, + export.budget_access.daily_limit_reached, + ), + budget_window( + "weekly", + "周额度", + 604800, + export.budget.limits.weekly_usd, + export.budget.weekly_used_usd, + export.budget.weekly_resets_at, + export.budget_access.weekly_limit_reached, + ), + ], }), - }, + }), pool, quota, models: export @@ -123,6 +127,9 @@ pub(super) fn response(export: PeerLinkExport) -> PeerLink { } fn budget_window( + key: &str, + label: &str, + window_seconds: u64, total: Decimal, used: Decimal, resets_at: Option, @@ -130,10 +137,13 @@ fn budget_window( ) -> PeerBudgetWindow { let limited = total != Decimal::ZERO; PeerBudgetWindow { + key: key.to_owned(), + label: Some(label.to_owned()), + window_seconds: Some(window_seconds), limit_reached: Some(limit_reached), total: limited.then(|| total.canonical()), used: used.canonical(), remaining: limited.then(|| total.saturating_sub(used).canonical()), - resets_at: resets_at.map(|time| DateTime::::from(time).to_rfc3339()), + reset_at: resets_at.map(|time| DateTime::::from(time).to_rfc3339()), } } diff --git a/backend/crates/gateway-api/tests/openai/peer_link/mod.rs b/backend/crates/gateway-api/tests/openai/peer_link/mod.rs index b4f4af567..6dfc541e8 100644 --- a/backend/crates/gateway-api/tests/openai/peer_link/mod.rs +++ b/backend/crates/gateway-api/tests/openai/peer_link/mod.rs @@ -243,10 +243,10 @@ async fn exports_scoped_observations_even_when_the_key_budget_is_exhausted_witho assert_eq!(response.status(), StatusCode::OK); let body = response_json(response).await; assert_eq!(body["object"], "cpr.peer"); - assert_eq!(body["schema_version"], 1); - assert_eq!(body["usage"]["data"]["daily"]["remaining"], "0"); + assert_eq!(body["schema_version"], 2); + assert_eq!(body["usage"]["data"]["windows"][0]["remaining"], "0"); assert_eq!(body["usage"]["data"]["allowed"], false); - assert_eq!(body["usage"]["data"]["daily"]["limit_reached"], true); + assert_eq!(body["usage"]["data"]["windows"][0]["limit_reached"], true); assert_eq!(body["pool"]["account_count"], 1); assert!(body["pool"].get("plans").is_none()); assert_eq!( @@ -338,7 +338,7 @@ async fn current_field_permissions_and_direct_account_facts_narrow_the_frozen_sc .peer_sharing .share_quota = false; let body = response_json(query(&app, PATH, Some(RAW_KEY)).await).await; - assert_eq!(body["quota"], json!({ "disclosure": "none" })); + assert!(body.get("quota").is_none()); assert!(body.get("models").is_none()); fixture .account @@ -511,7 +511,7 @@ async fn heavy_snapshots_reject_excess_concurrency_and_release_slots_on_completi .await .expect("budget-only queries do not consume heavy-query slots"); assert_eq!(budget.status(), StatusCode::OK); - assert_eq!(response_json(budget).await["pool"]["disclosure"], "none"); + assert!(response_json(budget).await.get("pool").is_none()); let cancelled = requests.pop().unwrap(); cancelled.abort(); assert!(cancelled.await.unwrap_err().is_cancelled()); diff --git a/backend/crates/gateway-protocol/src/peer.rs b/backend/crates/gateway-protocol/src/peer.rs index 4a2d69dee..7fe9648fe 100644 --- a/backend/crates/gateway-protocol/src/peer.rs +++ b/backend/crates/gateway-protocol/src/peer.rs @@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize}; pub const OBJECT: &str = "cpr.peer"; -pub const SCHEMA_VERSION: u32 = 1; +pub const SCHEMA_VERSION: u32 = 2; pub const MAX_ACCOUNTS: usize = 3_000; pub const MAX_QUOTA_WINDOWS: usize = 6_000; pub const MAX_MODELS: usize = 2_000; @@ -17,10 +17,12 @@ pub struct PeerLink { pub capabilities: Vec, pub observed_at: String, pub expires_at: String, - pub usage: PeerUsage, - pub pool: PeerPool, - #[serde(default)] - pub quota: PeerQuota, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub usage: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub pool: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub quota: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub models: Vec, } @@ -45,44 +47,39 @@ pub struct PeerUsage { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct PeerKeyUsage { - pub unit: String, + /// 仅在提供金额窗口时必填；访问结论不依赖金额或单位 + #[serde(default, skip_serializing_if = "Option::is_none")] + pub unit: Option, /// 上游账本给出的访问结论，缺失时不能根据展示金额推断恢复 #[serde(default, skip_serializing_if = "Option::is_none")] pub allowed: Option, - pub daily: PeerBudgetWindow, - pub weekly: PeerBudgetWindow, + /// 只返回实际提供的账本窗口，缺失周期不表示零用量或不限额 + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub windows: Vec, } /// 金额保留十进制字符串，不限额时 total 与 remaining 为 null #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct PeerBudgetWindow { + pub key: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub label: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub window_seconds: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub limit_reached: Option, pub total: Option, pub used: String, pub remaining: Option, - #[serde(rename = "resetsAt")] - pub resets_at: Option, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum PeerPoolDisclosure { - Summary, - #[default] - #[serde(other)] - None, + pub reset_at: Option, } #[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] pub struct PeerPool { - pub disclosure: PeerPoolDisclosure, #[serde(default, skip_serializing_if = "Option::is_none")] pub observed_at: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub account_count: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub states: Option, + pub account_count: u32, + pub states: PeerPoolStates, #[serde(default, skip_serializing_if = "Option::is_none")] pub plans: Option>, } @@ -103,18 +100,8 @@ pub struct PeerPlanCount { pub count: u32, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum PeerQuotaDisclosure { - Accounts, - #[default] - #[serde(other)] - None, -} - #[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)] pub struct PeerQuota { - pub disclosure: PeerQuotaDisclosure, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub accounts: Vec, } @@ -134,8 +121,12 @@ pub struct PeerQuotaAccount { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct PeerQuotaWindow { pub key: String, - pub group: String, - pub label: String, + /// 缺失时使用 general 分组；额度桶身份优先由 limit_id 表达 + #[serde(default, skip_serializing_if = "Option::is_none")] + pub group: Option, + /// 缺失时以稳定的窗口 key 展示 + #[serde(default, skip_serializing_if = "Option::is_none")] + pub label: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub limit_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/backend/crates/providers/openai/tests/admin/peer_info.rs b/backend/crates/providers/openai/tests/admin/peer_info.rs index c03b9150a..ccb7d2863 100644 --- a/backend/crates/providers/openai/tests/admin/peer_info.rs +++ b/backend/crates/providers/openai/tests/admin/peer_info.rs @@ -2,18 +2,24 @@ use super::*; use crate::support::MemoryCooldownPort; -use gateway_core::{account::UpstreamConnection, task::WorkerCycleContext}; +use gateway_core::{ + account::{UpstreamConnection, UpstreamInfoSource}, + task::WorkerCycleContext, +}; use provider_openai::credential::ResponsesTransport; fn peer_document() -> Value { let now = Utc::now(); json!({ - "object":"cpr.peer", "schema_version":1, + "object":"cpr.peer", "schema_version":2, "observed_at":now.to_rfc3339(), "expires_at":(now + chrono::Duration::minutes(5)).to_rfc3339(), "capabilities":["key_usage", "pool_summary"], - "usage":{"status":"available","observed_at":now.to_rfc3339(),"data":{"unit":"USD","allowed":true,"daily":{"limit_reached":false,"total":"10","used":"2","remaining":"8","resetsAt":(now + chrono::Duration::days(1)).to_rfc3339()},"weekly":{"limit_reached":false,"total":"50","used":"5","remaining":"45","resetsAt":(now + chrono::Duration::days(7)).to_rfc3339()}}}, - "pool":{"disclosure":"summary","account_count":2,"states":{"ready":2,"cooldown":0,"exhausted":0,"disabled":0,"error":0,"unknown":0},"plans":[{"name":"plus","count":2}]}, - "quota":{"disclosure":"none"},"models":[] + "usage":{"status":"available","observed_at":now.to_rfc3339(),"data":{"unit":"USD","allowed":true,"windows":[ + {"key":"daily","label":"日额度","window_seconds":86400,"limit_reached":false,"total":"10","used":"2","remaining":"8","reset_at":(now + chrono::Duration::days(1)).to_rfc3339()}, + {"key":"weekly","label":"周额度","window_seconds":604800,"limit_reached":false,"total":"50","used":"5","remaining":"45","reset_at":(now + chrono::Duration::days(7)).to_rfc3339()} + ]}}, + "pool":{"account_count":2,"states":{"ready":2,"cooldown":0,"exhausted":0,"disabled":0,"error":0,"unknown":0},"plans":[{"name":"plus","count":2}]}, + "models":[] }) } @@ -44,6 +50,11 @@ async fn peer_sync_restores_exhausted_accounts_only_from_current_key_budget_fact "amounts_disagree", "usage_unknown", "budget_only", + "access_only", + "access_only_denied", + "missing_usage", + "empty_windows", + "custom_window_exhausted", "old_usage", "old_snapshot", "unauthorized", @@ -57,30 +68,31 @@ async fn peer_sync_restores_exhausted_accounts_only_from_current_key_budget_fact let older = DateTime::::from(exhausted_at - Duration::from_secs(1)).to_rfc3339(); match case { "unlimited" => { - for window in ["daily", "weekly"] { - document["usage"]["data"][window] = json!({ - "total":null, "remaining":null, "used":"7", "resetsAt":null + for window in [0, 1] { + document["usage"]["data"]["windows"][window] = json!({ + "key": (["daily", "weekly"][window]), + "total":null, "remaining":null, "used":"7", "reset_at":null }); } } "budget_exhausted" => { document["usage"]["data"]["allowed"] = json!(false); - document["usage"]["data"]["daily"]["limit_reached"] = json!(true); + document["usage"]["data"]["windows"][0]["limit_reached"] = json!(true); } "weekly_exhausted" => { document["usage"]["data"] .as_object_mut() .unwrap() .remove("allowed"); - document["usage"]["data"]["weekly"]["limit_reached"] = json!(true); + document["usage"]["data"]["windows"][1]["limit_reached"] = json!(true); } "expired_budget" => { document["usage"]["data"] .as_object_mut() .unwrap() .remove("allowed"); - document["usage"]["data"]["daily"]["limit_reached"] = json!(true); - document["usage"]["data"]["daily"]["resetsAt"] = json!(older); + document["usage"]["data"]["windows"][0]["limit_reached"] = json!(true); + document["usage"]["data"]["windows"][0]["reset_at"] = json!(older); } "missing_allowed" => { document["usage"]["data"] @@ -94,20 +106,38 @@ async fn peer_sync_restores_exhausted_accounts_only_from_current_key_budget_fact .as_object_mut() .unwrap() .remove("allowed"); - for window in ["daily", "weekly"] { - document["usage"]["data"][window] + for window in [0, 1] { + document["usage"]["data"]["windows"][window] .as_object_mut() .unwrap() .remove("limit_reached"); } } "amounts_disagree" => { - document["usage"]["data"]["daily"]["remaining"] = json!("0"); - document["usage"]["data"]["daily"]["used"] = json!("10"); + document["usage"]["data"]["windows"][0]["remaining"] = json!("0"); + document["usage"]["data"]["windows"][0]["used"] = json!("10"); + } + "missing_remaining" => { + document["usage"]["data"]["windows"][0]["remaining"] = Value::Null } - "missing_remaining" => document["usage"]["data"]["daily"]["remaining"] = Value::Null, "usage_unknown" => document["usage"]["status"] = json!("unknown"), "budget_only" => document["capabilities"] = json!(["key_usage"]), + "access_only" | "access_only_denied" => { + document["usage"]["data"] = json!({"allowed": case == "access_only"}); + document.as_object_mut().unwrap().remove("pool"); + } + "missing_usage" => { + document.as_object_mut().unwrap().remove("usage"); + } + "empty_windows" => document["usage"]["data"] = json!({"windows": []}), + "custom_window_exhausted" => { + document["usage"]["data"] = json!({ + "unit":"USD", "windows":[{ + "key":"monthly", "used":"100", "total":"100", "remaining":"0", + "limit_reached":true, "reset_at":(Utc::now() + chrono::Duration::days(3)).to_rfc3339() + }] + }); + } "old_usage" => document["usage"]["observed_at"] = json!(older), "old_snapshot" => document["observed_at"] = json!(older), _ => {} @@ -132,12 +162,19 @@ async fn peer_sync_restores_exhausted_accounts_only_from_current_key_budget_fact "available" | "unlimited" | "budget_only" + | "access_only" | "missing_remaining" | "amounts_disagree" ), "{case}" ); assert_eq!(account.credential_state(), CredentialState::Ready); + if matches!( + case, + "access_only" | "access_only_denied" | "empty_windows" | "missing_usage" + ) { + assert!(quota(&bundle, &id).await.windows.is_empty(), "{case}"); + } assert!( server .received_requests() @@ -361,11 +398,11 @@ async fn key_usage_retains_unlimited_windows_without_pool_disclosure() { let server = MockServer::start().await; let mut document = peer_document(); document["capabilities"] = json!(["key_usage"]); - document["pool"] = json!({"disclosure":"none"}); - document["usage"]["data"]["daily"] = - json!({"total":null,"used":"0","remaining":null,"resetsAt":null}); - document["usage"]["data"]["weekly"]["total"] = json!(weekly_total); - document["usage"]["data"]["weekly"]["remaining"] = json!(weekly_total.map(|_| "45")); + document.as_object_mut().unwrap().remove("pool"); + document["usage"]["data"]["windows"][0] = + json!({"key":"daily","total":null,"used":"0","remaining":null,"reset_at":null}); + document["usage"]["data"]["windows"][1]["total"] = json!(weekly_total); + document["usage"]["data"]["windows"][1]["remaining"] = json!(weekly_total.map(|_| "45")); Mock::given(method("GET")) .and(path("/custom/v1/cpr/link")) .respond_with(ResponseTemplate::new(200).set_body_json(document)) @@ -402,6 +439,79 @@ async fn key_usage_retains_unlimited_windows_without_pool_disclosure() { } } +#[tokio::test] +async fn custom_budget_windows_preserve_their_period_without_inventing_other_windows() { + let server = MockServer::start().await; + let mut document = peer_document(); + document["capabilities"] = json!(["key_usage"]); + document.as_object_mut().unwrap().remove("pool"); + let reset = Utc::now() + chrono::Duration::days(20); + document["usage"]["data"] = json!({ + "unit":"USD", "allowed":true, "windows":[{ + "key":"monthly", "window_seconds":2592000, + "total":"100", "used":"20", "remaining":"80", "reset_at":reset.to_rfc3339() + }] + }); + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(200).set_body_json(document)) + .expect(1) + .mount(&server) + .await; + let (mut bundle, _, _, id, _) = fixture(&server).await; + cycle(&worker(&mut bundle)).await; + let observed = quota(&bundle, &id).await; + assert_eq!(observed.windows.len(), 1); + let window = &observed.windows[0]; + assert_eq!(window.key, "peer:budget:monthly"); + assert_eq!(window.label, "monthly"); + assert_eq!(window.window_seconds, Some(2592000)); + assert_eq!(window.reset_at, Some(reset)); + assert_eq!(window.used_percent, Some(20.0)); + assert!(!window.upstream.as_ref().unwrap().stale); +} + +#[tokio::test] +async fn model_only_snapshots_publish_signals_without_budget_or_account_disclosure() { + for (coverage, state, exhausted) in [ + ("complete", "exhausted", Some(true)), + ("complete", "available", Some(false)), + ("partial", "exhausted", None), + ] { + let server = MockServer::start().await; + let now = Utc::now(); + let document = json!({ + "object":"cpr.peer", "schema_version":2, + "observed_at":now.to_rfc3339(), "expires_at":(now + chrono::Duration::minutes(5)).to_rfc3339(), + "capabilities":["model_availability"], + "models":[{"model":"gpt-5.4", "coverage":coverage, "state":state, + "observed_at":now.to_rfc3339()}] + }); + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(200).set_body_json(document)) + .expect(1) + .mount(&server) + .await; + let (mut bundle, accounts, _, id, _) = fixture(&server).await; + let reader = gateway_core::account::UpstreamInfoReader::default(); + reader.bind(&bundle.peer.scheduling_source()).unwrap(); + cycle(&worker(&mut bundle)).await; + let observed = quota(&bundle, &id).await; + assert!(observed.upstream.unwrap().discovered); + assert!(observed.windows.is_empty()); + assert_eq!( + reader + .signal( + &accounts.account(id.as_str()).unwrap(), + "gpt-5.4", + SystemTime::now() + ) + .map(|signal| signal.exhausted), + exhausted, + "{coverage} {state}" + ); + } +} + #[tokio::test] async fn manual_refresh_queries_upstream_and_merges_concurrent_refreshes() { let server = MockServer::start().await; @@ -459,13 +569,18 @@ async fn native_windows_preserve_account_and_bucket_identity_without_pool_averag "window_seconds": 604800, "limit_reached": false }) }; + let mut minimal_window = window("Codex", 90.0); + minimal_window.as_object_mut().unwrap().remove("group"); + minimal_window.as_object_mut().unwrap().remove("label"); + document.as_object_mut().unwrap().remove("usage"); + document.as_object_mut().unwrap().remove("pool"); document["quota"] = serde_json::json!({ - "disclosure": "accounts", + "accounts": [ { "id": "first", "models": ["gpt-5.4"], "observed_at": observed_at, "expires_at": expires_at, "windows": [window("Codex", 10.0), window("Other", 30.0)] }, { "id": "second", "models": ["gpt-5.4"], "observed_at": observed_at, - "expires_at": expires_at, "windows": [window("Codex", 90.0)] } + "expires_at": expires_at, "windows": [minimal_window] } ] }); Mock::given(method("GET")) @@ -476,6 +591,8 @@ async fn native_windows_preserve_account_and_bucket_identity_without_pool_averag cycle(&worker(&mut bundle)).await; let observed = quota(&bundle, &id).await; assert_eq!(observed.windows.len(), 3); + assert_eq!(observed.windows[2].group, "general"); + assert_eq!(observed.windows[2].label, "weekly"); assert_eq!( observed .windows @@ -659,11 +776,18 @@ async fn invalid_or_redirected_responses_never_replace_the_last_valid_snapshot() cycle(&worker).await; let observed_at = quota(&bundle, &id).await.observed_at; let mut wrong_schema = peer_document(); - wrong_schema["schema_version"] = json!(2); + wrong_schema["schema_version"] = json!(99); let mut wrong_object = peer_document(); wrong_object["object"] = json!("other.peer"); let mut negative = peer_document(); - negative["usage"]["data"]["daily"]["used"] = json!("-20"); + negative["usage"]["data"]["windows"][0]["used"] = json!("-20"); + let mut missing_unit = peer_document(); + missing_unit["usage"]["data"] + .as_object_mut() + .unwrap() + .remove("unit"); + let mut duplicate_windows = peer_document(); + duplicate_windows["usage"]["data"]["windows"][1]["key"] = json!("daily"); let mut future = peer_document(); future["observed_at"] = json!((Utc::now() + chrono::Duration::minutes(10)).to_rfc3339()); let mut huge_pool = peer_document(); @@ -681,6 +805,8 @@ async fn invalid_or_redirected_responses_never_replace_the_last_valid_snapshot() ResponseTemplate::new(200).set_body_json(wrong_schema), ResponseTemplate::new(200).set_body_json(wrong_object), ResponseTemplate::new(200).set_body_json(negative), + ResponseTemplate::new(200).set_body_json(missing_unit), + ResponseTemplate::new(200).set_body_json(duplicate_windows), ResponseTemplate::new(200).set_body_json(future), ResponseTemplate::new(200).set_body_raw( "x".repeat(gateway_protocol::peer::MAX_RESPONSE_BYTES + 1), @@ -717,11 +843,11 @@ async fn peer_sync_excludes_budget_exhausted_accounts_without_pool_disclosure() let server = MockServer::start().await; let mut peer = peer_document(); peer["capabilities"] = json!(["key_usage"]); - peer["pool"] = json!({"disclosure":"none"}); + peer.as_object_mut().unwrap().remove("pool"); peer["usage"]["data"]["allowed"] = json!(false); - peer["usage"]["data"]["daily"]["limit_reached"] = json!(true); - peer["usage"]["data"]["daily"]["used"] = json!("10.0000"); - peer["usage"]["data"]["daily"]["remaining"] = json!("0.0000"); + peer["usage"]["data"]["windows"][0]["limit_reached"] = json!(true); + peer["usage"]["data"]["windows"][0]["used"] = json!("10.0000"); + peer["usage"]["data"]["windows"][0]["remaining"] = json!("0.0000"); Mock::given(method("GET")) .and(path("/custom/v1/cpr/link")) .respond_with(ResponseTemplate::new(200).set_body_json(peer)) @@ -776,7 +902,7 @@ async fn lists_reuse_versioned_summaries_and_details_keep_node_windows() { let server = MockServer::start().await; let mut document = peer_document(); document["capabilities"] = json!(["key_usage", "pool_summary", "quota_windows"]); - document["quota"] = json!({"disclosure":"accounts","accounts":[{ + document["quota"] = json!({"accounts":[{ "id":"node-one", "models":["gpt-5.4"], "observed_at":document["observed_at"], "expires_at":document["expires_at"], "windows":[{"key":"weekly","group":"general","label":"周额度","used_percent":40,"limit_reached":false}] @@ -893,6 +1019,33 @@ async fn lists_reuse_versioned_summaries_and_details_keep_node_windows() { assert_eq!(store.quota_reads(), read_count + 1); } +#[tokio::test] +async fn cached_observations_require_the_current_protocol_version() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(200).set_body_json(peer_document())) + .expect(1) + .mount(&server) + .await; + let (mut bundle, accounts, store, id, config) = fixture(&server).await; + cycle(&worker(&mut bundle)).await; + let mut observation = store + .get_quotas(std::slice::from_ref(&id)) + .await + .unwrap() + .remove(0); + let mut document = observation.quota.expose_to_provider().clone(); + document.insert("schema_version".to_owned(), json!(99)); + observation.quota = OpaqueProviderData::new(document); + observation.observed_at = SystemTime::now(); + store.compare_and_swap_quota(observation).await.unwrap(); + + let restored = peer_fixture(config.config.clone(), accounts).await; + let observed = quota(&restored, &id).await; + assert!(!observed.upstream.unwrap().discovered); + assert!(observed.windows.is_empty()); +} + #[tokio::test] async fn ordinary_api_upstreams_remain_undiscovered_and_keep_the_detection_backoff() { let server = MockServer::start().await; @@ -943,7 +1096,7 @@ async fn cold_lists_bound_full_snapshot_batches_and_reuse_every_hydrated_summary let connection = UpstreamConnection::for_account(&account); let now = SystemTime::now(); let mut document = peer_document(); - document["usage"]["data"]["daily"]["used"] = json!(index.to_string()); + document["usage"]["data"]["windows"][0]["used"] = json!(index.to_string()); store .compare_and_swap_quota(QuotaObservation { expected_connection: Some(connection), diff --git a/deploy/README.md b/deploy/README.md index bdb4d9adf..752a9fa6b 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -244,12 +244,12 @@ PostgreSQL/Redis 启动密码。日常校验使用 `config --quiet` 在下游导入或编辑 OpenAI API Key 账号，填写上游 API 地址和该 Key，后台自动发现 CPR 互联能力。 地址包含自定义前缀时保留完整前缀，后台在该地址下请求 `cpr/link`。 -列表与详情显示已保存的观测，手动刷新提交后台任务；旧观测和查询失败分别提示，不要求上游实时可用 +列表与详情显示已保存的观测，手动刷新等待上游查询完成；旧观测和查询失败分别提示，不要求上游实时可用 第三方网关通过插件实现相同协议也可被识别，不按软件名称判定兼容。 上游明确返回 Key 预算耗尽时，下游暂停该账号；后续同步确认允许使用后自动恢复，普通限流冷却独立保留。 -金额只供展示，缺少访问状态时不根据余额推断恢复。账号池额度观测作为 Smart 选择的参考，不能保证下一次请求成功； -协议和公开边界见 [CPR 互联 API](../docs/api.md#cpr-互联只读协议) +金额只供展示，缺少访问状态时不根据余额推断恢复。账号池额度观测作为 Smart 选择的参考，不能保证下一次请求成功 +互联两端需支持 CPR Link v2；协议、第三方接入与公开边界见 [CPR 互联 API](../docs/api.md#cpr-互联只读协议) ### 批量维护代理 diff --git a/docs/api.md b/docs/api.md index 78c9714bc..c71a70749 100644 --- a/docs/api.md +++ b/docs/api.md @@ -723,91 +723,156 @@ OpenAI 选号阶段确认本次可选账号全部额度耗尽时，HTTP 返回 ` ### CPR 互联只读协议 -`GET {api_base}/cpr/link` 使用同一数据面入口认证；CPR 默认路径为 `/v1/cpr/link`，传入当前下游使用的推理 Key。 -只有该 Key 显式开启 `peerSharing.enabled` 且绑定非空分组时可查询。接口不接受 Cookie、管理 API Key 或查询参数， -不执行推理、占用推理并发/RPM、更新最近使用时间或刷新上游；Key 预算耗尽后仍可查询 +`GET {api_base}/cpr/link` 使用推理 Key 查询当前授权范围内的只读快照，不接受查询参数。 +响应直接返回 JSON，所有状态均带 `Cache-Control: no-store`。第三方项目或插件可独立实现此 HTTP 协议，无需依赖 CPR 插件 SDK -协议识别依赖 `object: "cpr.peer"`、`schema_version: 1` 和 `capabilities`，不发布独立节点身份。 -未知能力可以忽略，不支持的版本按不兼容处理。 -第三方项目或插件可实现同一协议，连接由本地账号及其凭据、出口版本识别。 -响应直接返回 JSON，所有状态均带 `Cache-Control: no-store` +#### 版本与可选能力 -只公开预算的完整响应形态如下： +| 顶层必填字段 | 合同 | +| --- | --- | +| `object` | 固定为 `"cpr.peer"` | +| `schema_version` | 当前为 `2`，不支持的版本按不兼容处理 | +| `capabilities` | 当前 Key 获准公开的能力名称列表，最多 32 项；未知能力可以忽略 | +| `observed_at` | 本次快照组装时间，RFC3339 UTC 时间 | +| `expires_at` | 快照到期时间，必须晚于 `observed_at`，且不超过其后 30 分钟 | + +`usage`、`pool`、`quota`、`models` 按能力独立提供，未提供的数据块直接省略。 +接入端只消费已声明能力的数据，缺失值保持未知，不解释为零余额、空池或额度恢复。 +未知字段可以忽略；没有独立节点 ID，连接由本地账号及其凭据、出口版本识别 + +| 能力 | 对应数据 | +| --- | --- | +| `key_usage` | `usage`：当前认证 Key 的预算访问结论与可选金额窗口 | +| `pool_summary` | `pool`：账号数量、互斥状态计数及可选套餐分布 | +| `quota_windows` | `quota.accounts`：匿名账号及各自额度窗口 | +| `model_availability` | `models`：按当前 Key 公开的模型额度汇总，可独立于账号明细提供 | + +只公开预算访问结论的完整响应： ```json { "object": "cpr.peer", - "schema_version": 1, + "schema_version": 2, "capabilities": ["key_usage"], - "observed_at": "2026-10-29T12:00:00Z", - "expires_at": "2026-10-29T12:05:00Z", + "observed_at": "2026-11-01T12:00:00Z", + "expires_at": "2026-11-01T12:05:00Z", "usage": { "status": "available", - "observed_at": "2026-10-29T12:00:00Z", - "data": { - "unit": "USD", - "allowed": true, - "daily": { "limit_reached": false, "total": "1", "used": "0.6", "remaining": "0.4", "resetsAt": "2026-10-29T16:00:00Z" }, - "weekly": { "limit_reached": false, "total": null, "used": "2", "remaining": null, "resetsAt": null } - } - }, - "pool": { "disclosure": "none" }, - "quota": { "disclosure": "none" } + "observed_at": "2026-11-01T12:00:00Z", + "data": { "allowed": true } + } } ``` -`usage.data` 沿用 [API Key 额度查询](#api-key-额度查询) 的十进制金额、窗口及不限额语义。 -`usage.status` 可为 `available`、`unavailable` 或 `unknown`；没有可用数据时不把缺失值解释为零余额 +#### Key 预算 -`usage.data.allowed` 是上游根据本地预算账本给出的访问结论，日周窗口的 `limit_reached` 表示对应预算是否耗尽。 -接入端以明确的 `allowed` 为准；缺失时，仍有效的 `limit_reached: true` 只能证明耗尽，不能据此推断恢复。 -缺少这些状态字段时金额仅用于展示，接入端不从 `total`、`used` 或 `remaining` 计算账号资格 +`usage.status` 为 `available`、`unavailable` 或 `unknown`，未知状态按 `unknown` 处理。 +`available` 表示数据可用，不表示预算尚有余额；没有可用数据时可省略 `data`。 +`usage.observed_at` 是预算的原始观测时间，缺失或过期时不能据此改变账号资格 -| 能力 | 数据合同 | 公开条件 | -| --- | --- | --- | -| `key_usage` | `usage` 为当前认证 Key 的本地预算账本 | `enabled` | -| `pool_summary` | `pool.disclosure: "summary"`，含 `observed_at`、`account_count` 与 `states`；状态计数为 `ready`、`cooldown`、`exhausted`、`disabled`、`error`、`unknown` | `sharePool` | -| 套餐分布 | `pool.plans: [{ name, count }]`；未授权时省略，不返回账号与套餐的逐项对应关系 | `sharePlans`，且已公开账号池 | -| `quota_windows` | `quota.disclosure: "accounts"`，`accounts` 保留匿名账号及各自窗口；未授权时只返回 `disclosure: "none"` | `shareQuota` | -| `model_availability` | `models` 按当前 Key 与请求到该节点的公开模型汇总，见下 | `shareQuota` | +`usage.data.allowed` 是上游预算账本给出的访问结论，`false` 表示预算耗尽，不代表认证失败或临时限流。 +接入端以明确的 `allowed` 为准；缺失时，有效窗口中的 `limit_reached: true` 只能证明耗尽，不能据此推断恢复。 +金额只供展示，不从 `total`、`used` 或 `remaining` 计算账号资格 -账号额度示例： +需要公开金额时，在 `usage.data` 中提供 `unit` 和 `windows`： + +```json +{ + "unit": "USD", + "allowed": true, + "windows": [ + { + "key": "monthly", + "label": "月额度", + "window_seconds": 2592000, + "limit_reached": false, + "total": "100", + "used": "20", + "remaining": "80", + "reset_at": "2026-11-30T12:00:00Z" + } + ] +} +``` + +- `windows` 只包含实际提供的周期，可以省略或为空；不要求提供日、周窗口 +- 每个窗口的 `key` 必填且在该预算中唯一，使用稳定的机器标识；`label` 可选，省略时展示 `key` +- 提供金额窗口时 `unit` 必填，表示统一计费单位，如 `USD`；仅提供 `allowed` 时可以省略 +- `used` 是必填的非负十进制字符串，`total` 与 `remaining` 使用相同格式；不限额时两者为 `null` +- `window_seconds` 是可选的正整数，表示本次周期的秒数；`reset_at` 是可选 RFC3339 时间。仅当两者均有效时，接入端可按窗口边界关联本机用量 +- `limit_reached` 可省略，省略表示没有明确的窗口耗尽结论；重置时间已过意味着观测过期，不意味着预算已恢复 + +#### 账号池、额度与模型 + +`pool` 必须包含 `account_count` 与 `states`。`states` 包含 `ready`、`cooldown`、`exhausted`、`disabled`、`error`、`unknown` 六项非负整数， +各项互斥且总和等于 `account_count`。可选 `observed_at` 表示池观测时间；可选 `plans: [{ name, count }]` 表示套餐分布， +套餐总数不能超过账号数，不返回账号与套餐的逐项对应关系 + +`quota.accounts` 保留匿名账号及各自窗口。账号示例： ```json { "id": "df306c3718c99d3615352e41d32751d3", "models": ["gpt-5.4"], - "observed_at": "2026-10-29T11:58:00Z", - "expires_at": "2026-10-29T12:28:00Z", - "windows": [{ "key": "weekly", "group": "shortTerm", "label": "周额度", "role": "secondary", "window_seconds": 604800, "used_percent": 47, "reset_at": "2026-10-31T16:00:00Z", "limit_reached": false }] + "observed_at": "2026-11-01T11:58:00Z", + "expires_at": "2026-11-01T12:28:00Z", + "windows": [{ "key": "weekly", "window_seconds": 604800, "used_percent": 47, "reset_at": "2026-11-07T16:00:00Z", "limit_reached": false }] } ``` -匿名 `id` 在同一 Key 下稳定，不能当作本地账号 ID。返回结果不含 Key、邮箱、账号名称、代理或凭据正文。 -窗口的 `key`、`group`、`label`、可选 `limit_id`／`limit_name` 与可选 `role`（`primary`、`secondary`、`monthly`）保留原生展示口径； -`window_seconds`、`used_percent`、`reset_at` 均可缺失。不同账号或不同周期的百分比不能相加， -不能从百分比推算 Token 总额，也不能用本地请求用量填充上游额度 +账号的 `id`、`models`、`windows` 必填，匿名 `id` 在同一 Key 下稳定，不能当作接入端的本地账号 ID。 +观测时间可省略，但缺失或过期时不能参与调度判断。返回结果不含 Key、邮箱、账号名称、代理或凭据正文 -模型汇总条目形如 `{ "model": "gpt-5.4", "coverage": "complete", "state": "available", "observed_at": "...", "expires_at": "..." }`。 +| 账号窗口字段 | 合同 | +| --- | --- | +| `key` | 必填，额度桶内稳定的窗口标识 | +| `group` | 可选的分组标识，省略时为 `general` | +| `limit_id` | 可选的额度桶标识；同账号的窗口按 `limit_id` 分桶，缺失时按 `group` 分桶 | +| `label`、`limit_name` | 可选的窗口与额度桶展示名；窗口名省略时展示 `key` | +| `role` | 可选展示角色：`primary`、`secondary`、`monthly`，未知值不指定角色 | +| `window_seconds`、`used_percent`、`reset_at` | 可选的周期秒数、已用百分比（0～100）与 RFC3339 重置时间 | +| `limit_reached` | 是否明确触顶，省略时不标记触顶 | + +不同账号或周期的百分比不能相加，不能推算 Token 总额，也不能用本地请求用量填充上游额度 + +`models` 条目形如 `{ "model": "gpt-5.4", "coverage": "complete", "state": "available", "observed_at": "...", "expires_at": "..." }`。 `coverage` 为 `complete` 或 `partial`，`state` 为 `available`、`exhausted` 或 `unknown`。 只有该模型所有候选的资格与额度都明确、有效且全部耗尽，才能给出 `complete + exhausted`。 -它是通用模型额度参考，不承诺 Images、Live 等各操作的额外资格或即时容量；过期、缺失或部分覆盖保持中性， +它是通用模型额度参考，不承诺 Images、Live 等操作的额外资格或即时容量；缺失、过期或部分覆盖保持中性， 请求仍需执行当前节点的正常授权与调度 -顶层 `observed_at` 是此次本地快照组装时间，默认有效期 5 分钟；账号窗口与模型汇总保留各自的原始观测时间， -不因查询或转发而更新。各组件有效期最多为原始观测后 30 分钟，并受重置边界约束；过去的重置时间意味着旧窗口过期， -不意味着额度已经恢复。接入端应按各组件时间判断，不能用新的顶层时间延长旧观测 +#### 有效期与数量限制 + +顶层快照默认有效期 5 分钟，CPR 接入端仅将最近 5 分钟组装且未到期的快照用于访问与调度判断。 +预算观测有效期为 5 分钟；账号与模型可提供各自的 `expires_at`，省略时按观测后 5 分钟处理，最长不得超过 30 分钟。 +各组件保留原始观测时间，并受重置边界约束，不因查询或转发而更新；新的顶层时间不能延长旧观测 + +接口不分页，不截断账号集后声称完整。账号最多 3000 个，预算窗口和账号额度窗口分别最多 6000 个， +模型最多 2000 个，响应最大 4 MiB。窗口标识、展示名和额度桶标识不得为空或包含控制字符，最多 256 字节； +`group` 最多 128 字节，`unit` 最多 16 字节。可选字段没有有效值时省略 + +#### CPR 内置共享规则 + +CPR 默认路径为 `/v1/cpr/link`，使用同一数据面入口认证，不接受 Cookie 或管理 API Key。 +只有当前 Key 显式开启 `peerSharing.enabled` 且绑定非空分组时可查询。 +查询不执行推理、占用推理并发/RPM、更新最近使用时间或刷新上游；预算耗尽后仍可查询 + +| 共享配置 | 导出内容 | +| --- | --- | +| `enabled` | `key_usage`，包含明确访问结论及本地日、周美元账本；金额与周期口径见 [API Key 额度查询](#api-key-额度查询) | +| `sharePool` | `pool_summary` | +| `sharePlans` | 已公开账号池中的 `pool.plans` | +| `shareQuota` | `quota_windows` 与 `model_availability` | 共享范围同时受当前 Key 的公开开关、分组及冻结权限限制，只包含 Provider 明确声明的直连 OAuth 账号。 -互联 Key 的实际推理也使用该直连范围，不把 API Key 转接账号再次作为共享池，协议不递归查询其他节点。 -接口不分页、不截断账号集后声称完整；共享范围最多 3000 个账号、额度窗口总计最多 6000 个、模型最多 2000 个。 -任一数量超限、响应超过 4 MiB 或本地读取失败返回 `503 peer_snapshot_unavailable` +互联 Key 的推理也使用该直连范围，不把 API Key 转接账号再次作为共享池，不递归查询其他节点。 +这些开关和账号组织方式属于 CPR 内置实现；第三方通过自身授权规则生成协议数据 缺失、非法、禁用或删除的 Key 返回 `401`；未开启共享或没有显式分组返回 `403 peer_sharing_disabled`； -未知参数返回 `400 invalid_peer_query`。每个进程对每个认证 Key 独立限制为每分钟 30 次。 -涉及账号池、额度或模型聚合的查询每个进程最多同时执行 8 个，满额时立即返回 -`429 peer_query_rate_limited` 和 `Retry-After: 1`；仅查询 Key 预算不占用聚合槽位。 -每 Key 限流同样返回 `429 peer_query_rate_limited`，`Retry-After` 表示剩余等待秒数；两类限制均不消耗推理请求配额 +未知参数返回 `400 invalid_peer_query`；数量或响应大小超限、本地读取失败返回 `503 peer_snapshot_unavailable`。 +每个进程对每个认证 Key 独立限制为每分钟 30 次，涉及账号池、额度或模型聚合的查询最多同时执行 8 个。 +满额时返回 `429 peer_query_rate_limited` 和 `Retry-After`，聚合槽位满时等待秒数为 `1`；仅查询 Key 预算不占用聚合槽位。 +上述限制均不消耗推理请求配额 ### Codex Live 语音通话 @@ -1067,7 +1132,7 @@ OpenAI 主动额度刷新和正常响应携带的明确套餐会同步到账号 | `lastError` | 当前实例最近同步失败分类：`unauthorized`、`unsupported`、`rate_limited`、`invalid_response`、`unavailable`，成功时为 `null` | | `pool` | 上游授权公开的账号数量、状态数量及套餐数量，未公开时为 `null` | -互联额度复用 `quota.windows`。可用的上游 Key 账本始终保留日、周窗口，不限额也保留已用金额。 +互联额度复用 `quota.windows`。上游 Key 账本按实际提供的周期展示，不补造缺失窗口；不限额也保留已用金额。 窗口返回 `amount: { usedDisplay, remainingDisplay, totalDisplay }`，字符串包含单位，美元使用 `$` 前缀； 不限额的总额与剩余额显示为 `∞`，`usedPercent` 为 `null`，金额不换算成 Token。 Key 消费窗口的 `localUsage` 按各自 `[resetAt - windowSeconds, resetAt)` 边界聚合本机请求， @@ -1328,8 +1393,8 @@ API Key 账号与 OAuth 共用现有 Responses、Images 和 standalone Search 未识别为 CPR 互联的 API Key 账号，其列表和详情 `usage` 汇总账号创建后仍保留的本地请求记录， `usage.windowKind` 为 `general`，管理端显示累计用量。日志清理会影响累计范围，不代表上游余额 -互联账号按上游 Key 最长的消费窗口统计本机 Token、费用和模型用量，有周额度时使用周窗口，只有日额度时使用日窗口， -`usage.windowKind` 分别为 `weekly` 或 `daily`。查询边界同账号额度窗口，为 `[resetAt - windowSeconds, resetAt)`； +互联账号按上游 Key 已提供的最长消费窗口统计本机 Token、费用和模型用量。 +日、周周期的 `usage.windowKind` 分别为 `daily`、`weekly`，其他周期为 `quota`。查询边界同账号额度窗口，为 `[resetAt - windowSeconds, resetAt)`； 未开启或缺少重置边界时显示无数据，不回退到历史累计，也不采用池内其他账号的周期。 OAuth 账号仍按实际周/月额度窗口统计 diff --git a/docs/architecture.md b/docs/architecture.md index ce42e9cf4..3653b965a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1004,7 +1004,8 @@ Admin 仅为开启自动使用的账号保存卡片 ID、到期时间、下一 共享端由 Admin 汇聚已授权分组中的本地账号事实、Provider 只读额度和当前 Key 费用账本，API 负责版本化线协议输出。 导出同时核对请求冻结权限和当前共享设置、分组及认证方式，不回源查询，不输出账号凭据或本地账号标识。 -协议按版本和能力识别，第三方实现与内置节点使用相同合同；共享端不转发接入端的远端快照。 +协议按版本和能力识别，第三方实现与内置节点使用相同合同；数据块按授权独立提供，预算周期由上游账本决定。 +Admin 集中校验与裁剪公开能力，展示和调度消费同一份观测；缺失预算或窗口保持未知，共享端不转发接入端的远端快照。 Admin 为账号聚合设置独立并发上限，满额立即返回可重试的限流响应；模型展开按响应字节预算提前拒绝必然超限的组合，批量投影定期让出执行时间，避免重查询占满共享数据库与执行资源 Admin 的 `peer_link` 分别拥有共享导出与上游消费用例；上游消费统一处理发现、协议校验、同步、展示和调度投影。