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 <noreply@openai.com>
2026-11-06 08:06:19 +00:00 · 2026-11-01 19:24:17 +08:00
parent c57af9b0dd
commit db849c1251
12 changed files with 592 additions and 338 deletions
@@ -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),
@@ -13,24 +13,26 @@ pub(super) fn observe(
    started_at: SystemTime,
    now: SystemTime,
 ) -> Option<QuotaState> {
+    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)
@@ -70,22 +70,23 @@ pub(super) struct SchedulingObservation {
 impl SchedulingObservation {
    pub(super) fn new(peer: &PeerLink, now: SystemTime) -> Option<Self> {
        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<usize>>::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 = &quota_accounts[index];
                                let validity = Validity::component(
                                    account.observed_at.as_deref(),
                                    account.expires_at.as_deref(),
@@ -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> {
-    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<OpaqueProviderData, UpstreamInfoFailure> {
    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::<usize>()
            > 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::<u64>();
-        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>() > 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::<u64>();
+    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>() > 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 &quota.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<f64> {
    if value.is_empty()
        || value.len() > 64
@@ -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())
    }
 }
@@ -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<SystemTime>,
@@ -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::<Utc>::from(time).to_rfc3339()),
+        reset_at: resets_at.map(|time| DateTime::<Utc>::from(time).to_rfc3339()),
    }
 }
@@ -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());
@@ -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<String>,
    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<PeerUsage>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub pool: Option<PeerPool>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub quota: Option<PeerQuota>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub models: Vec<PeerModelAvailability>,
 }
@@ -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<String>,
    /// 上游账本给出的访问结论，缺失时不能根据展示金额推断恢复
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub allowed: Option<bool>,
-    pub daily: PeerBudgetWindow,
-    pub weekly: PeerBudgetWindow,
+    /// 只返回实际提供的账本窗口，缺失周期不表示零用量或不限额
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub windows: Vec<PeerBudgetWindow>,
 }

 /// 金额保留十进制字符串，不限额时 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<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub window_seconds: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit_reached: Option<bool>,
    pub total: Option<String>,
    pub used: String,
    pub remaining: Option<String>,
-    #[serde(rename = "resetsAt")]
-    pub resets_at: Option<String>,
-}
-
-#[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<String>,
 }

 #[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<String>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub account_count: Option<u32>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub states: Option<PeerPoolStates>,
+    pub account_count: u32,
+    pub states: PeerPoolStates,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub plans: Option<Vec<PeerPlanCount>>,
 }
@@ -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<PeerQuotaAccount>,
 }
@@ -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<String>,
+    /// 缺失时以稳定的窗口 key 展示
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub label: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -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::<Utc>::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),
@@ -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-互联只读协议)

 ### 批量维护代理

@@ -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 账号仍按实际周/月额度窗口统计

@@ -1004,7 +1004,8 @@ Admin 仅为开启自动使用的账号保存卡片 ID、到期时间、下一

 共享端由 Admin 汇聚已授权分组中的本地账号事实、Provider 只读额度和当前 Key 费用账本，API 负责版本化线协议输出。
 导出同时核对请求冻结权限和当前共享设置、分组及认证方式，不回源查询，不输出账号凭据或本地账号标识。
-协议按版本和能力识别，第三方实现与内置节点使用相同合同；共享端不转发接入端的远端快照。
+协议按版本和能力识别，第三方实现与内置节点使用相同合同；数据块按授权独立提供，预算周期由上游账本决定。
+Admin 集中校验与裁剪公开能力，展示和调度消费同一份观测；缺失预算或窗口保持未知，共享端不转发接入端的远端快照。
 Admin 为账号聚合设置独立并发上限，满额立即返回可重试的限流响应；模型展开按响应字节预算提前拒绝必然超限的组合，批量投影定期让出执行时间，避免重查询占满共享数据库与执行资源

 Admin 的 `peer_link` 分别拥有共享导出与上游消费用例；上游消费统一处理发现、协议校验、同步、展示和调度投影。