feat(accounts): make quota credit use opt-in per account (#628)

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
This commit is contained in:
Zyy
2026-11-07 00:20:43 +08:00
committed by GitHub
co-authored by Zyy
parent 4326eb1e4b
commit 90ecac978a
45 changed files with 641 additions and 268 deletions
@@ -21,6 +21,7 @@ pub use gateway_core::account::{
/// 账号可用的管理操作
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct ProviderAccountCapabilities {
pub quota_credits: bool,
pub quota: bool,
pub quota_refresh: bool,
pub profile: bool,
@@ -110,6 +111,7 @@ pub enum AccountGroupFilter {
/// 账号公共存储投影;Provider 专属字段不进入此结构
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AccountRecord {
pub auto_use_quota_credits: bool,
pub id: String,
pub provider_kind: ProviderKind,
pub groups: Vec<AccountGroupRef>,
@@ -249,6 +251,7 @@ pub struct AccountUpdateResult {
/// 仅修改显式字段的批量账号设置命令
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BatchUpdateAccounts {
pub auto_use_quota_credits: Option<bool>,
pub account_ids: Vec<String>,
pub enabled: Option<bool>,
pub concurrency_limit: Option<Option<AccountConcurrencyLimit>>,
@@ -759,6 +759,7 @@ impl AccountsService for DefaultAccountsService {
self.accounts
.batch_update_accounts(
BatchUpdateAccounts {
auto_use_quota_credits: None,
account_ids: vec![account_id.to_string()],
enabled: Some(true),
concurrency_limit: None,
@@ -837,15 +838,28 @@ impl AccountsService for DefaultAccountsService {
(
std::sync::Arc<dyn crate::ports::provider::ProviderAdmin>,
Vec<ProviderAccountId>,
Vec<ProviderAccountId>,
),
>::new();
for account_id in &account_ids {
let (item, provider) = self.provider_for_account(account_id).await?;
providers
if command.auto_use_quota_credits.is_some()
&& !provider
.account_capabilities(account_id, &item.account.authentication_kind)
.quota_credits
{
return Err(AdminError::invalid("该账号不支持使用额度点数"));
}
let (_, selected, exhausted) = providers
.entry(item.account.provider_kind)
.or_insert_with(|| (provider, Vec::new()))
.1
.push(account_id.clone());
.or_insert_with(|| (provider, Vec::new(), Vec::new()));
selected.push(account_id.clone());
if command.auto_use_quota_credits == Some(true)
&& !item.account.auto_use_quota_credits
&& item.account.quota.is_exhausted()
{
exhausted.push(account_id.clone());
}
}
let enabled = command.enabled;
let result = self
@@ -853,7 +867,7 @@ impl AccountsService for DefaultAccountsService {
.batch_update_accounts(command, context)
.await
.map_err(|error| map_store_error(error, "provider accounts"))?;
for (provider, provider_ids) in providers.values() {
for (provider, provider_ids, _) in providers.values() {
if enabled == Some(false) {
for account_id in provider_ids {
super::notify_account_unavailable(
@@ -872,6 +886,18 @@ impl AccountsService for DefaultAccountsService {
.await;
}
publish_committed(self.snapshot.as_ref(), result.config_revision).await?;
for (provider, _, exhausted) in providers.into_values() {
if exhausted.is_empty() {
continue;
}
super::credential_mutation::observe_account_quotas(
self.upstream_info.as_ref(),
&provider,
&self.accounts,
&exhausted,
context.request_id.as_str(),
);
}
Ok(result)
}
@@ -30,7 +30,18 @@ pub(super) async fn publish_credentials_and_observe_quota(
notify_account_facts_changed(peer.map(AsRef::as_ref), provider.as_ref(), account_ids).await;
publish_committed(snapshot, revision).await?;
// 凭据已经提交;额度只是可重建的观察,不能拖住管理请求或回滚提交结果
observe_account_quotas(peer, provider, accounts, account_ids, request_id);
Ok(())
}
pub(super) fn observe_account_quotas(
peer: Option<&std::sync::Arc<dyn crate::ports::peer::PeerUpstreamService>>,
provider: &std::sync::Arc<dyn ProviderAdmin>,
accounts: &std::sync::Arc<dyn AccountStore>,
account_ids: &[ProviderAccountId],
request_id: &str,
) {
// 配置已提交,额度查询失败不回滚配置
let provider = provider.clone();
let peer = peer.cloned();
let accounts = accounts.clone();
@@ -48,14 +59,14 @@ pub(super) async fn publish_credentials_and_observe_quota(
Ok(Some(account)) => account,
Ok(None) => continue,
Err(error) => {
tracing::warn!(request_id, account_id = %account_id, error = %error, "initial quota account lookup failed");
tracing::warn!(request_id, account_id = %account_id, error = %error, "quota observation account lookup failed");
continue;
}
};
if !account.enabled || &account.provider_kind != provider.provider_kind() {
continue;
}
// 首查与手动刷新共用认证能力分派,互联协议由 Admin 解释
// 后台查询与手动刷新共用认证能力分派,互联协议由 Admin 解释
super::peer_link::upstream::read_quota(
Some(peer),
provider.as_ref(),
@@ -72,12 +83,11 @@ pub(super) async fn publish_credentials_and_observe_quota(
provider = %provider.provider_kind().as_str(),
account_id = %account_id.as_str(),
quota_error = ?error.kind(),
"initial quota observation failed"
"account quota observation failed"
);
}
}
});
Ok(())
}
pub(super) async fn required_credential(
@@ -1908,6 +1908,7 @@ async fn accounts_batch_update_should_commit_once_and_notify_each_provider() {
.batch_update(
&context("batch-update-request"),
BatchUpdateAccounts {
auto_use_quota_credits: None,
model_access: Default::default(),
outbound_proxy: None,
account_ids: vec!["acct_openai".to_owned(), "acct_xai".to_owned()],
@@ -3398,6 +3399,7 @@ fn quota_local_usage(account_id: &str, total_tokens: u64) -> AccountUsage {
pub(super) fn account_record(kind: &str) -> AccountRecord {
let now = Utc::now();
AccountRecord {
auto_use_quota_credits: false,
outbound_proxy_id: None,
outbound_proxy_enabled: true,
notes: None,
@@ -116,6 +116,7 @@ pub(super) fn account_view(
let (mut quota, refresh_token_expires_at) = account_quota_view(quota, cooldown, now, time);
quota.protection_rules = protection_rules;
AccountView {
auto_use_quota_credits: account.auto_use_quota_credits,
upstream,
capabilities: capabilities.into(),
id: account.id.clone(),
@@ -61,6 +61,7 @@ pub struct ListQuery {
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct BatchUpdateAccountsRequest {
pub auto_use_quota_credits: Option<bool>,
pub outbound_proxy_id: Option<String>,
pub outbound_proxy_url: Option<AccountProxyUpdate>,
pub account_ids: Vec<String>,
@@ -117,7 +118,8 @@ impl BatchUpdateAccountsRequest {
.map(parse_concurrency_limit)
.transpose()?;
self.weight.map(parse_account_weight).transpose()?;
if self.enabled.is_none()
if self.auto_use_quota_credits.is_none()
&& self.enabled.is_none()
&& self.concurrency_limit.is_none()
&& self.weight.is_none()
&& self.group_ids.is_none()
@@ -133,6 +135,7 @@ impl BatchUpdateAccountsRequest {
pub(super) fn into_command(self) -> Result<BatchUpdateAccounts, WireValidationError> {
self.validate()?;
Ok(BatchUpdateAccounts {
auto_use_quota_credits: self.auto_use_quota_credits,
outbound_proxy: proxy_selection(self.outbound_proxy_id, self.outbound_proxy_url)?,
account_ids: self.account_ids,
enabled: self.enabled,
@@ -266,6 +269,7 @@ pub struct AccountSummaryView {
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct AccountView {
pub auto_use_quota_credits: bool,
pub upstream: Option<super::AccountUpstreamInfoView>,
pub capabilities: AccountCapabilitiesView,
pub outbound_proxy_id: Option<String>,
@@ -318,6 +322,7 @@ pub struct AccountCapacityView {
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct AccountCapabilitiesView {
pub quota_credits: bool,
pub quota: bool,
pub quota_refresh: bool,
pub profile: bool,
@@ -336,6 +341,7 @@ impl From<gateway_admin::model::accounts::ProviderAccountCapabilities> for Accou
subscription: value.subscription,
avatar: value.avatar,
reset_credits: value.reset_credits,
quota_credits: value.quota_credits,
consume_reset_credit: value.consume_reset_credit,
}
}
@@ -406,6 +406,7 @@ fn account_page_item(
AccountPageItem {
protection_rules: Vec::new(),
account: AccountRecord {
auto_use_quota_credits: false,
outbound_proxy_id: None,
outbound_proxy_enabled: true,
id: "acct_cooldown".to_owned(),
@@ -578,6 +578,7 @@ mod provider {
total_slots: None,
},
account: AccountRecord {
auto_use_quota_credits: false,
outbound_proxy_id: None,
outbound_proxy_enabled: true,
notes: None,
@@ -97,6 +97,7 @@ async fn fixture(sharing: ClientPeerSharing, plaintext: &str) -> (AdminTestFixtu
},
projection: resolve_account_status(&facts, now.into()),
account: AccountRecord {
auto_use_quota_credits: false,
id: "acct_private_sentinel".into(),
provider_kind: provider,
groups: vec![group.clone()],
@@ -364,9 +364,11 @@ impl QuotaAccessState {
}
}
/// 权威额度耗尽证据
/// 账号额度访问依据
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum QuotaEvidence {
/// 套餐已耗尽,但上游确认点数可用
CreditsAvailable,
ProviderDenied,
AccountLimitReached,
UsageLimitReached,
@@ -377,6 +379,7 @@ impl QuotaEvidence {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::CreditsAvailable => "credits_available",
Self::ProviderDenied => "provider_denied",
Self::AccountLimitReached => "account_limit_reached",
Self::UsageLimitReached => "usage_limit_reached",
@@ -387,6 +390,7 @@ impl QuotaEvidence {
#[must_use]
pub fn parse(value: &str) -> Option<Self> {
match value {
"credits_available" => Some(Self::CreditsAvailable),
"provider_denied" => Some(Self::ProviderDenied),
"account_limit_reached" => Some(Self::AccountLimitReached),
"usage_limit_reached" => Some(Self::UsageLimitReached),
@@ -456,6 +460,17 @@ impl QuotaState {
}
}
/// 点数是唯一放行依据时保留套餐重置时间,开关变化不伪造新观测
#[must_use]
pub const fn credits_only(observed_at: SystemTime, reset_at: Option<SystemTime>) -> Self {
Self {
access: QuotaAccessState::Allowed,
evidence: Some(QuotaEvidence::CreditsAvailable),
observed_at: Some(observed_at),
reset_at,
}
}
/// 从持久化列恢复额度状态;非法组合返回 `None`
#[must_use]
pub const fn from_persisted(
@@ -464,32 +479,25 @@ impl QuotaState {
observed_at: Option<SystemTime>,
reset_at: Option<SystemTime>,
) -> Option<Self> {
match access {
QuotaAccessState::Unknown if evidence.is_none() && reset_at.is_none() => Some(Self {
let valid = match (access, evidence) {
(QuotaAccessState::Unknown, None) => reset_at.is_none(),
(QuotaAccessState::Allowed, None) => observed_at.is_some() && reset_at.is_none(),
(QuotaAccessState::Allowed, Some(QuotaEvidence::CreditsAvailable)) => {
observed_at.is_some()
}
(QuotaAccessState::Exhausted, Some(QuotaEvidence::CreditsAvailable)) => false,
(QuotaAccessState::Exhausted, Some(_)) => observed_at.is_some(),
_ => false,
};
if valid {
Some(Self {
access,
evidence,
observed_at,
reset_at,
}),
QuotaAccessState::Allowed
if evidence.is_none() && observed_at.is_some() && reset_at.is_none() =>
{
Some(Self {
access,
evidence,
observed_at,
reset_at,
})
}
QuotaAccessState::Exhausted if evidence.is_some() && observed_at.is_some() => {
Some(Self {
access,
evidence,
observed_at,
reset_at,
})
}
_ => None,
})
} else {
None
}
}
@@ -813,6 +821,7 @@ pub struct ProviderAccount {
enabled: bool,
concurrency_limit: Option<AccountConcurrencyLimit>,
weight: AccountWeight,
auto_use_quota_credits: bool,
model_access: super::AccountModelAccess,
credential_state: CredentialState,
quota: QuotaState,
@@ -851,6 +860,7 @@ impl ProviderAccount {
enabled: true,
concurrency_limit: None,
weight: AccountWeight::DEFAULT,
auto_use_quota_credits: false,
model_access: super::AccountModelAccess::all(),
credential_state: CredentialState::Unknown,
quota: QuotaState::unknown(),
@@ -864,6 +874,17 @@ impl ProviderAccount {
}
}
#[must_use]
pub const fn with_auto_use_quota_credits(mut self, enabled: bool) -> Self {
self.auto_use_quota_credits = enabled;
self
}
#[must_use]
pub const fn auto_use_quota_credits(&self) -> bool {
self.auto_use_quota_credits
}
#[must_use]
pub fn with_notes(mut self, notes: Option<String>) -> Self {
self.notes = notes;
@@ -32,6 +32,7 @@ struct Facts {
fn account() -> AccountRecord {
let at = "2026-01-01T00:00:00Z".parse().unwrap();
AccountRecord {
auto_use_quota_credits: false,
outbound_proxy_id: None,
outbound_proxy_enabled: true,
id: "acct_facts".into(),
@@ -242,7 +242,7 @@ impl AccountGroupStore for PgAccountGroupRepository {
account.upstream_user_id, account.upstream_account_id, account.plan_type,
account.authentication_kind, account.credential_revision, account.outbound_proxy_id, account.outbound_proxy_url,
account.has_refresh_token, account.access_token_expires_at,
account.next_refresh_at, account.enabled, account.concurrency_limit,
account.next_refresh_at, account.enabled, account.auto_use_quota_credits, account.concurrency_limit,
account.quota_observed_at, account.weight, account.model_access_json, account.credential_state, account.quota_access_state,
account.quota_evidence, account.quota_access_observed_at,
account.quota_reset_at, account.last_error_reason,
@@ -664,6 +664,7 @@ impl AccountStore for PgAdminAccountStore {
.accounts
.batch_update_provider_accounts_admin(BatchUpdateProviderAccountsAdmin {
account_ids: vec![command.account_id.clone()],
auto_use_quota_credits: None,
name: command.name,
notes: command.notes,
enabled: Some(command.enabled),
@@ -747,6 +748,10 @@ impl AccountStore for PgAdminAccountStore {
};
let mut changed_fields = Vec::new();
for (changed, field) in [
(
command.auto_use_quota_credits.is_some(),
"auto_use_quota_credits",
),
(command.enabled.is_some(), "enabled"),
(command.concurrency_limit.is_some(), "concurrency_limit"),
(command.weight.is_some(), "weight"),
@@ -766,6 +771,7 @@ impl AccountStore for PgAdminAccountStore {
.accounts
.batch_update_provider_accounts_admin(BatchUpdateProviderAccountsAdmin {
account_ids: command.account_ids,
auto_use_quota_credits: command.auto_use_quota_credits,
name: None,
notes: None,
enabled: command.enabled,
@@ -106,7 +106,7 @@ pub(crate) async fn load_admin_account_page(
select coalesce(p.enabled, true) as outbound_proxy_enabled, coalesce(p.max_websocket_connections, 0) as outbound_proxy_max_websocket_connections, p.auto_location, p.detected_location_json, p.location_country, p.location_region, p.location_city, p.location_timezone, a.outbound_proxy_id, a.outbound_proxy_url, a.id, a.provider_kind, a.name, a.notes, a.email, a.upstream_user_id,
a.upstream_account_id, a.plan_type, a.authentication_kind,
a.credential_revision, a.has_refresh_token, a.access_token_expires_at,
a.next_refresh_at, a.enabled, a.concurrency_limit, a.weight, a.model_access_json,
a.next_refresh_at, a.enabled, a.auto_use_quota_credits, a.concurrency_limit, a.weight, a.model_access_json,
a.credential_state, a.quota_access_state, a.quota_evidence,
a.quota_access_observed_at, a.quota_reset_at, a.last_error_reason,
a.last_error_message, a.credential_observed_at, a.created_at, a.updated_at,
@@ -10,6 +10,7 @@ pub(crate) fn admin_account_record(
summary: ProviderAccountSummary,
) -> AdminStoreResult<AccountRecord> {
Ok(AccountRecord {
auto_use_quota_credits: summary.auto_use_quota_credits,
outbound_proxy_id: summary.outbound_proxy_id,
outbound_proxy_enabled: summary.outbound_proxy_enabled,
outbound_proxy: summary.outbound_proxy,
@@ -117,7 +117,7 @@ impl PgProviderAccountRepository {
let mut query = sqlx::QueryBuilder::<Postgres>::new(
"select coalesce(proxy_enabled, true) as outbound_proxy_enabled, coalesce(max_websocket_connections, 0) as outbound_proxy_max_websocket_connections, auto_location, detected_location_json, location_country, location_region, location_city, location_timezone, outbound_proxy_id, outbound_proxy_url, id, provider_kind, name, notes, email, upstream_user_id,
upstream_account_id, plan_type, authentication_kind, credential_revision, has_refresh_token,
access_token_expires_at, next_refresh_at, enabled, concurrency_limit, weight, model_access_json, credential_state,
access_token_expires_at, next_refresh_at, enabled, auto_use_quota_credits, concurrency_limit, weight, model_access_json, credential_state,
credential_observed_at, quota_access_state, quota_evidence,
quota_access_observed_at, quota_reset_at,
quota_observed_at, last_error_reason, last_error_message, created_at, updated_at
@@ -198,7 +198,7 @@ impl ProviderAccountRepository for PgProviderAccountRepository {
let rows = sqlx::query(
"select coalesce(proxy_enabled, true) as outbound_proxy_enabled, coalesce(max_websocket_connections, 0) as outbound_proxy_max_websocket_connections, auto_location, detected_location_json, location_country, location_region, location_city, location_timezone, outbound_proxy_id, outbound_proxy_url, id, provider_kind, name, notes, email, upstream_user_id,
upstream_account_id, plan_type, authentication_kind, credential_revision, has_refresh_token,
access_token_expires_at, next_refresh_at, enabled, concurrency_limit, weight, model_access_json, credential_state,
access_token_expires_at, next_refresh_at, enabled, auto_use_quota_credits, concurrency_limit, weight, model_access_json, credential_state,
credential_observed_at, quota_access_state, quota_evidence,
quota_access_observed_at, quota_reset_at,
quota_observed_at, last_error_reason, last_error_message, created_at, updated_at
@@ -552,6 +552,11 @@ impl ProviderAccountAdminRepository for PgProviderAccountRepository {
command.outbound_proxy.as_ref(),
)
.await?;
if let Some(enabled) = command.auto_use_quota_credits {
sqlx::query("update provider_accounts set auto_use_quota_credits = $2 where id = any($1) and auto_use_quota_credits <> $2")
.bind(&command.account_ids).bind(enabled).execute(&mut *transaction).await
.map_err(|source| postgres_unavailable("set account quota credits policy", source))?;
}
if let Some(group_ids) = &command.group_ids {
replace_account_group_assignments_in_transaction(
&mut transaction,
@@ -1236,7 +1241,7 @@ impl PgProviderAccountRepository {
let rows = sqlx::query(
"select coalesce(proxy_enabled, true) as outbound_proxy_enabled, coalesce(max_websocket_connections, 0) as outbound_proxy_max_websocket_connections, auto_location, detected_location_json, location_country, location_region, location_city, location_timezone, outbound_proxy_id, outbound_proxy_url, id, provider_kind, name, notes, email, upstream_user_id,
upstream_account_id, plan_type, authentication_kind, credential_revision, has_refresh_token,
access_token_expires_at, next_refresh_at, enabled, concurrency_limit, weight, model_access_json, credential_state,
access_token_expires_at, next_refresh_at, enabled, auto_use_quota_credits, concurrency_limit, weight, model_access_json, credential_state,
credential_observed_at, quota_access_state, quota_evidence,
quota_access_observed_at, quota_reset_at,
quota_observed_at, last_error_reason, last_error_message, created_at, updated_at
@@ -51,6 +51,7 @@ pub(crate) fn parse_error_reason(value: Option<String>) -> StoreResult<Option<Ac
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProviderAccountSummary {
pub auto_use_quota_credits: bool,
pub outbound_proxy_id: Option<String>,
pub outbound_proxy_enabled: bool,
pub request_location: Option<gateway_core::account::RequestLocation>,
@@ -249,6 +250,7 @@ impl fmt::Debug for RotateProviderAccount {
#[derive(Debug, Clone)]
pub struct BatchUpdateProviderAccountsAdmin {
pub auto_use_quota_credits: Option<bool>,
pub outbound_proxy: Option<gateway_admin::model::proxies::AccountProxySelection>,
pub account_ids: Vec<String>,
pub name: Option<String>,
@@ -322,7 +324,7 @@ impl ProviderAccountStateUpdate {
pub(crate) const ACCOUNT_SELECT: &str = "select coalesce(proxy_enabled, true) as outbound_proxy_enabled, coalesce(max_websocket_connections, 0) as outbound_proxy_max_websocket_connections, auto_location, detected_location_json, location_country, location_region, location_city, location_timezone, outbound_proxy_id, outbound_proxy_url, id, provider_kind, name, notes, email, upstream_user_id,
upstream_account_id, plan_type, authentication_kind, provider_credentials_json, credential_revision,
has_refresh_token, access_token_expires_at, next_refresh_at, enabled, concurrency_limit, weight, model_access_json, credential_state,
has_refresh_token, access_token_expires_at, next_refresh_at, enabled, auto_use_quota_credits, concurrency_limit, weight, model_access_json, credential_state,
quota_access_state, quota_evidence, quota_access_observed_at, quota_reset_at,
last_error_reason, last_error_message,
credential_observed_at, quota_observed_at, created_at, updated_at
@@ -333,7 +335,7 @@ pub(crate) const ACCOUNT_SELECT: &str = "select coalesce(proxy_enabled, true) as
pub(crate) const ACCOUNT_SELECT_BY_IDS: &str = "select coalesce(proxy_enabled, true) as outbound_proxy_enabled, coalesce(max_websocket_connections, 0) as outbound_proxy_max_websocket_connections, auto_location, detected_location_json, location_country, location_region, location_city, location_timezone, outbound_proxy_id, outbound_proxy_url, id, provider_kind, name, notes, email, upstream_user_id,
upstream_account_id, plan_type, authentication_kind, provider_credentials_json, credential_revision,
has_refresh_token, access_token_expires_at, next_refresh_at, enabled, concurrency_limit, weight, model_access_json, credential_state,
has_refresh_token, access_token_expires_at, next_refresh_at, enabled, auto_use_quota_credits, concurrency_limit, weight, model_access_json, credential_state,
quota_access_state, quota_evidence, quota_access_observed_at, quota_reset_at,
last_error_reason, last_error_message,
credential_observed_at, quota_observed_at, created_at, updated_at
@@ -345,7 +347,7 @@ pub(crate) const ACCOUNT_SELECT_BY_IDS: &str = "select coalesce(proxy_enabled, t
pub(crate) const REFRESH_CANDIDATES_SELECT: &str = "select coalesce(proxy_enabled, true) as outbound_proxy_enabled, coalesce(max_websocket_connections, 0) as outbound_proxy_max_websocket_connections, auto_location, detected_location_json, location_country, location_region, location_city, location_timezone, outbound_proxy_id, outbound_proxy_url, id, provider_kind, name, notes, email, upstream_user_id,
upstream_account_id, plan_type, authentication_kind, provider_credentials_json, credential_revision,
has_refresh_token, access_token_expires_at, next_refresh_at, enabled, concurrency_limit, weight, model_access_json, credential_state,
has_refresh_token, access_token_expires_at, next_refresh_at, enabled, auto_use_quota_credits, concurrency_limit, weight, model_access_json, credential_state,
quota_access_state, quota_evidence, quota_access_observed_at, quota_reset_at,
last_error_reason, last_error_message,
credential_observed_at, quota_observed_at, created_at, updated_at
@@ -415,6 +417,7 @@ pub(crate) fn core_account_from_summary(
)
.with_scheduling(summary.concurrency_limit, summary.weight)
.with_model_access(summary.model_access)
.with_auto_use_quota_credits(summary.auto_use_quota_credits)
.with_outbound_proxy(
summary
.outbound_proxy
@@ -470,6 +473,7 @@ pub(crate) fn account_summary_from_row(
.and_then(AccountWeight::new)
.ok_or_else(|| invalid("invalid weight"))?;
Ok(ProviderAccountSummary {
auto_use_quota_credits: get(&row, "auto_use_quota_credits")?,
outbound_proxy_id: get(&row, "outbound_proxy_id")?,
outbound_proxy_enabled: get(&row, "outbound_proxy_enabled")?,
request_location: super::super::proxies::effective_location_from_row(&row)?,
@@ -1540,6 +1540,7 @@ async fn account_notes_round_trip_and_survive_import_and_scheduling_updates() {
store
.batch_update_accounts(
BatchUpdateAccounts {
auto_use_quota_credits: None,
account_ids: vec!["acct_notes".to_owned()],
enabled: Some(false),
concurrency_limit: Some(None),
@@ -1778,6 +1779,7 @@ async fn account_enable_preserves_facts_and_explicit_recovery_clears_them() {
let enabled = store
.batch_update_accounts(
BatchUpdateAccounts {
auto_use_quota_credits: None,
account_ids: vec!["acct_recovery".to_owned()],
enabled: Some(true),
concurrency_limit: None,
@@ -1902,6 +1904,7 @@ async fn terminal_batch_update_replaces_state_and_groups_once_or_rolls_back_ever
let result = store
.batch_update_accounts(
BatchUpdateAccounts {
auto_use_quota_credits: None,
model_access: Default::default(),
outbound_proxy: None,
account_ids: account_ids.clone(),
@@ -1939,6 +1942,7 @@ async fn terminal_batch_update_replaces_state_and_groups_once_or_rolls_back_ever
store
.batch_update_accounts(
BatchUpdateAccounts {
auto_use_quota_credits: None,
model_access: Default::default(),
outbound_proxy: None,
account_ids: account_ids.clone(),
@@ -2678,6 +2682,7 @@ async fn provider_account_admin_mutations_are_scoped_audited_and_atomic() {
let revision = repository
.batch_update_provider_accounts_admin(BatchUpdateProviderAccountsAdmin {
auto_use_quota_credits: None,
name: None,
notes: None,
model_access: Default::default(),
@@ -3693,6 +3698,7 @@ async fn model_access_only_batch_update_preserves_other_settings_and_survives_re
.expect("policy");
let store = admin_account_store(&database.pool);
let command = BatchUpdateAccounts {
auto_use_quota_credits: None,
account_ids: vec![input.id.clone()],
enabled: None,
concurrency_limit: None,
@@ -3813,3 +3819,83 @@ async fn connection_configuration_update_preserves_credential_health() {
assert_eq!(after["credential_revision"], 2);
database.close().await;
}
#[tokio::test]
async fn disabling_quota_credits_revokes_credit_only_access_even_with_a_late_observation() {
let Some(database) = TestDatabase::create("quota_credit_policy").await else {
return;
};
let repository = PgProviderAccountRepository::new(database.pool.clone());
let input = account("acct_credit_policy", "user-credit-policy");
repository
.insert_provider_account(input.clone())
.await
.expect("insert");
let id = ProviderAccountId::new(input.id.clone()).expect("id");
let before = repository.get_account(&id).await.unwrap().unwrap();
assert!(!before.auto_use_quota_credits());
let admin = admin_account_store(&database.pool);
let context = MutationContext {
actor: MutationActor::System,
request_id: "quota_credits".into(),
};
let set_policy = async |enabled| {
let command = BatchUpdateAccounts {
account_ids: vec![input.id.clone()],
auto_use_quota_credits: Some(enabled),
enabled: None,
concurrency_limit: None,
weight: None,
model_access: None,
group_ids: None,
outbound_proxy: None,
};
admin
.batch_update_accounts(command, &context)
.await
.unwrap();
};
let observe = async |state: QuotaState| {
let observation = QuotaObservation {
expected_connection: None,
plan_type: None,
account_id: id.clone(),
expected_revision: before.revision(),
quota: OpaqueProviderData::new(Default::default()),
observed_at: state.observed_at().unwrap(),
state,
};
assert_eq!(
repository
.compare_and_swap_quota(observation)
.await
.unwrap(),
QuotaWriteOutcome::Updated
);
};
let access = async || {
repository
.get_account(&id)
.await
.unwrap()
.unwrap()
.quota()
.access()
};
set_policy(true).await;
observe(QuotaState::credits_only(SystemTime::now(), None)).await;
assert_eq!(access().await, QuotaAccessState::Allowed);
set_policy(false).await;
assert_eq!(access().await, QuotaAccessState::Exhausted);
observe(QuotaState::credits_only(SystemTime::now(), None)).await;
assert_eq!(access().await, QuotaAccessState::Exhausted);
set_policy(true).await;
assert_eq!(access().await, QuotaAccessState::Exhausted);
observe(QuotaState::allowed(SystemTime::now())).await;
set_policy(false).await;
assert_eq!(access().await, QuotaAccessState::Allowed);
database.close().await;
}
@@ -355,6 +355,7 @@ async fn disabled_proxy_projects_direct_while_preserving_bindings_and_credential
let error = admin
.batch_update_accounts(
BatchUpdateAccounts {
auto_use_quota_credits: None,
account_ids: vec![
"acct_enabled_openai".to_owned(),
"acct_enabled_xai".to_owned(),
@@ -1180,6 +1181,7 @@ async fn legacy_urls_join_one_catalog_entry_and_invalid_batch_rolls_back() {
admin
.batch_update_accounts(
BatchUpdateAccounts {
auto_use_quota_credits: None,
model_access: Default::default(),
account_ids: vec!["acct_one".to_owned(), "acct_missing".to_owned()],
enabled: Some(false),
@@ -205,6 +205,7 @@ impl ProviderAdmin for OpenAiAdminProvider {
) -> gateway_admin::model::accounts::ProviderAccountCapabilities {
let oauth = authentication_kind == crate::credential::CODEX_AUTHENTICATION_KIND_OAUTH;
gateway_admin::model::accounts::ProviderAccountCapabilities {
quota_credits: oauth,
quota: oauth,
quota_refresh: oauth,
profile: oauth,
@@ -1151,6 +1152,7 @@ fn account_from_record(account: &AccountRecord) -> Result<ProviderAccount, Provi
revision,
account.access_token_expires_at.map(SystemTime::from),
)
.with_auto_use_quota_credits(account.auto_use_quota_credits)
.with_outbound_proxy(account.outbound_proxy.clone())
.with_notes(account.notes.clone())
.with_profile(
@@ -57,27 +57,11 @@ impl CodexCredentialQuotaService {
})
.next();
let plan_type = observed_account_plan(account.plan_type(), observed_plan);
// 套餐、credits 等元数据可以更新,但没有额度窗口事实时必须保留旧观察时刻,
// 也不能借旧快照重新推导 quota state
if !has_quota_facts {
let Some(state) = existing_state else {
return Ok(false);
};
let outcome = self
.store
.compare_and_swap_quota(QuotaObservation {
expected_connection: None,
plan_type,
account_id: account.id().clone(),
expected_revision: account.revision(),
quota: OpaqueProviderData::new(merge_passive_quota(existing, rate_limits)),
observed_at: SystemTime::now(),
state,
})
.await?;
return Ok(outcome != QuotaWriteOutcome::Conflict);
if !has_quota_facts && existing_state.is_none() {
return Ok(false);
}
let quota = merge_passive_quota(existing, rate_limits);
let prior = existing_state.unwrap_or_else(|| account.quota());
let (quota, updated_default) = merge_passive_quota(existing, rate_limits);
let observed_at = SystemTime::now();
let snapshot = parse_account_quota_snapshot(
account.id().clone(),
@@ -85,8 +69,33 @@ impl CodexCredentialQuotaService {
observed_at,
&Value::Object(quota.clone()),
)?;
// 这些 headers 来自一次成功推理,访问结论优先于可能滞后的百分比
let state = QuotaState::allowed(observed_at);
let credit_access = prior.evidence() == Some(QuotaEvidence::CreditsAvailable);
let fresh_credits = rate_limits.iter().any(|limits| limits.credits.is_some());
// 具名桶和元数据不改变套餐结论;点数不能再用时,仍须撤销原先仅凭点数的放行
let subscription = if !updated_default
|| (snapshot.quota().access() == QuotaAccessState::Unknown
&& (credit_access || prior.is_exhausted()))
{
if credit_access {
QuotaState::exhausted(
QuotaEvidence::AccountLimitReached,
observed_at,
prior.reset_at(),
)
} else {
prior
}
} else if snapshot.quota().is_exhausted() {
snapshot.quota()
} else {
QuotaState::allowed(observed_at)
};
// 套餐刚耗尽时不能拿合并快照里的旧余额作为首次放行依据
let state = snapshot.access_with_credits(
subscription,
account.auto_use_quota_credits()
&& (credit_access || (updated_default && fresh_credits)),
);
let outcome = self
.store
.compare_and_swap_quota(QuotaObservation {
@@ -102,7 +111,9 @@ impl CodexCredentialQuotaService {
if outcome == QuotaWriteOutcome::Conflict {
return Ok(false);
}
self.scheduling.observe(&snapshot);
if has_quota_facts {
self.scheduling.observe(&snapshot);
}
Ok(true)
}
}
@@ -110,7 +121,8 @@ impl CodexCredentialQuotaService {
fn merge_passive_quota(
mut quota: Map<String, Value>,
observations: &[ParsedRateLimits],
) -> Map<String, Value> {
) -> (Map<String, Value>, bool) {
let mut updated_default = false;
let mut snapshots = RateLimitSnapshotsByLimitId::take_from_document(&mut quota);
for rate_limits in observations {
let default_is_named_alias = default_limit_is_named_alias(rate_limits);
@@ -127,6 +139,7 @@ fn merge_passive_quota(
let Some(rate_limit) = passive_rate_limit_snapshot(details) else {
continue;
};
updated_default |= limit_id == DEFAULT_CODEX_LIMIT_ID;
snapshots.upsert(&limit_id, details, rate_limit);
resolved_limit_ids.insert(wire_limit_id.as_str(), limit_id);
}
@@ -147,7 +160,7 @@ fn merge_passive_quota(
merge_passive_metadata(&mut quota, rate_limits);
}
snapshots.write_to_document(&mut quota);
quota
(quota, updated_default)
}
fn merge_passive_metadata(quota: &mut Map<String, Value>, rate_limits: &ParsedRateLimits) {
@@ -131,17 +131,21 @@ fn window_reset_recovered(window: &CodexQuotaWindow, previous: Option<DateTime<U
/// 已记录的耗尽窗口全部到期后,允许新的明确放行观察解除,兼容用量百分比滞后
pub(super) fn reconcile_refresh(
current: QuotaState,
auto_use_credits: bool,
refreshed: &mut CodexAccountQuotaSnapshot,
previous: Option<&CodexAccountQuotaSnapshot>,
document: &mut Map<String, Value>,
) -> Result<QuotaState, CodexCredentialQuotaError> {
document.remove(RECOVERY_FIELD);
if refreshed.credits_usable() {
// 点数独立于套餐窗口,恢复后不能继续携带已失效的窗口恢复进度
refreshed.recovery = None;
return Ok(QuotaState::allowed(refreshed.observed_at()));
}
let state = if current.is_exhausted() {
// 点数已不再可用时,旧点数依据不能随下一次开关切换复活
let state = if current.evidence() == Some(QuotaEvidence::CreditsAvailable) {
QuotaState::exhausted(
QuotaEvidence::AccountLimitReached,
current.observed_at().unwrap_or(refreshed.observed_at()),
current.reset_at(),
)
.merge_observation(refreshed.quota())
} else if current.is_exhausted() {
current
} else {
current.merge_observation(refreshed.quota())
@@ -178,17 +182,25 @@ pub(super) fn reconcile_refresh(
.min()
.map(SystemTime::from)
.or(state.reset_at());
let state = refreshed.access_with_credits(
QuotaState::exhausted(
state
.evidence()
.unwrap_or(QuotaEvidence::AccountLimitReached),
state.observed_at().unwrap_or(refreshed.observed_at()),
reset_at,
),
auto_use_credits,
);
if !state.is_exhausted() {
refreshed.recovery = None;
return Ok(state);
}
document.insert(
RECOVERY_FIELD.to_owned(),
serde_json::to_value(&recovery)
.map_err(|_| CodexCredentialQuotaError::InvalidCredentialData)?,
);
refreshed.recovery = Some(recovery);
Ok(QuotaState::exhausted(
state
.evidence()
.unwrap_or(QuotaEvidence::AccountLimitReached),
state.observed_at().unwrap_or(refreshed.observed_at()),
reset_at,
))
Ok(state)
}
@@ -116,15 +116,15 @@ impl CodexCredentialQuotaService {
Ok(())
}
/// 成功推理是额度可访问的权威证据,同时解除账号级 429 冷却
/// 成功推理解除账号级 429 冷却;OAuth 额度由额度观测维护
pub async fn record_successful_inference(
&self,
account: &ProviderAccount,
observed_at: SystemTime,
) -> Result<(), CodexCredentialQuotaError> {
// 已经发出的并发请求可能晚于耗尽事实成功返回;真实成功仍是独立的
// Allowed 权威证据,但调度不会为了试探恢复而放行耗尽账号
if account.quota().access() == QuotaAccessState::Exhausted {
if account.authentication_kind() != crate::credential::CODEX_AUTHENTICATION_KIND_OAUTH
&& account.quota().is_exhausted()
{
let outcome = self
.store
.apply_quota_access(QuotaAccessChange {
@@ -304,6 +304,7 @@ impl CodexCredentialQuotaService {
};
let state = reconcile_refresh(
account.quota(),
account.auto_use_quota_credits(),
&mut snapshot,
previous.as_ref(),
&mut object,
@@ -503,8 +504,8 @@ impl CodexCredentialQuotaService {
None
};
// 失败后回读只允许明确可用的点数恢复访问,套餐 allowed 仍可能滞后
let observe_access =
matches!(authority, QuotaRefreshAuthority::ObserveAccess) || snapshot.credits_usable();
let observe_access = matches!(authority, QuotaRefreshAuthority::ObserveAccess)
|| (account.auto_use_quota_credits() && snapshot.credits_usable());
if account.quota().is_exhausted() && !observe_access {
if self
.store
@@ -529,15 +530,16 @@ impl CodexCredentialQuotaService {
.with_observed_at(observed_at),
});
}
let state = if observe_access {
let state = if matches!(authority, QuotaRefreshAuthority::PreserveAccess) {
snapshot.access_with_credits(account.quota(), account.auto_use_quota_credits())
} else {
reconcile_refresh(
account.quota(),
account.auto_use_quota_credits(),
&mut snapshot,
previous.as_ref(),
&mut object,
)?
} else {
account.quota()
};
let snapshot = snapshot.with_quota_state(state);
if self
@@ -151,6 +151,18 @@ impl CodexAccountQuotaSnapshot {
self.credits.as_ref()
}
pub(super) fn access_with_credits(
&self,
subscription: QuotaState,
enabled: bool,
) -> QuotaState {
if enabled && subscription.is_exhausted() && self.credits_usable {
QuotaState::credits_only(self.observed_at, subscription.reset_at())
} else {
subscription
}
}
pub(super) const fn credits_usable(&self) -> bool {
self.credits_usable
}
@@ -351,11 +363,6 @@ pub(crate) fn parse_account_quota_snapshot(
});
let quota = authoritative_quota_state(&object, &windows, observed_at)?;
let credits_usable = has_usable_credits(&object);
let quota = if credits_usable {
QuotaState::allowed(observed_at)
} else {
quota
};
let recovery = object
.get(RECOVERY_FIELD)
.map(|value| serde_json::from_value(value.clone()))
@@ -400,10 +407,13 @@ fn has_usable_credits(usage: &Map<String, Value>) -> bool {
== Some(true);
let other_limit_reached = usage
.get("rate_limit_reached_type")
.and_then(|reason| reason.get("type"))
.and_then(Value::as_str)
.and_then(|reason| reason.as_str().or_else(|| reason.get("type")?.as_str()))
.is_some_and(|reason| reason != "rate_limit_reached");
available && !spend_reached && !other_limit_reached
let overage_reached = credits
.get("overage_limit_reached")
.and_then(Value::as_bool)
== Some(true);
available && !spend_reached && !other_limit_reached && !overage_reached
}
/// 只有默认 `codex` 桶能提供账号级套餐访问结论,具名桶只保留展示事实
@@ -427,9 +437,14 @@ fn authoritative_quota_state(
.ok_or(CodexCredentialQuotaError::InvalidCredentialData)?;
let allowed = optional_bool(rate_limit, "allowed")?;
let top_level_reached = optional_bool(rate_limit, "limit_reached")?.unwrap_or(false);
let window_reached = windows
.iter()
.any(|w| w.is_account_wide() && w.limit_reached);
let window_reached = windows.iter().any(|window| {
window.is_account_wide()
&& (window.limit_reached
|| (window.limit_reached()
&& window
.reset_at()
.is_none_or(|reset| SystemTime::from(reset) > observed_at)))
});
let reset_at = windows
.iter()
.filter(|w| w.is_account_wide() && w.limit_reached())
@@ -444,16 +459,6 @@ fn authoritative_quota_state(
})
.map(SystemTime::from);
if allowed == Some(true) {
if top_level_reached || window_reached {
tracing::warn!(
top_level_reached,
window_reached,
"OpenAI quota returned contradictory access facts; explicit allowed=true wins"
);
}
return Ok(QuotaState::allowed(observed_at));
}
if allowed == Some(false) {
return Ok(QuotaState::exhausted(
QuotaEvidence::ProviderDenied,
@@ -468,6 +473,9 @@ fn authoritative_quota_state(
reset_at,
));
}
if allowed == Some(true) {
return Ok(QuotaState::allowed(observed_at));
}
Ok(QuotaState::observed_unknown(observed_at))
}
@@ -110,6 +110,10 @@ impl CodexCredentialQuotaService {
let now_utc = chrono::Utc::now();
for account in accounts {
if account.quota().is_exhausted() {
summary.skipped_exhausted += 1;
continue;
}
if let Some(snapshot) = observed_snapshots.get(account.id()) {
// 1. 周线已触顶或耗尽跳过
let weekly_exhausted = snapshot
@@ -391,8 +391,11 @@ pub(super) fn schedule_authoritative_quota_refresh_after_failure(
quota: &Arc<CodexCredentialQuotaService>,
account: &ProviderAccount,
) {
if !account.auto_use_quota_credits() {
return;
}
// 限额错误可以确认账号状态,但 Responses 事件中的 used_percent 可能仍停在上一结算点
// usage 快照在后台补齐展示基线,不得撤销同一轮真实失败,也不能阻塞原始响应
// 后台查询可为后续请求确认点数,但不阻塞或重放原始请求
// 随机等待 1–3 秒让上游结算再查询,避免多账号同时受限时补查完全同步;
// 仍保持单次语义(一次 sleep + 一次查询),5 秒仅限制实际查询自身
let quota = Arc::clone(quota);
@@ -2178,6 +2178,7 @@ fn provider_ports_with_catalog(
fn account_record(account: &ProviderAccount) -> AccountRecord {
let now = Utc::now();
AccountRecord {
auto_use_quota_credits: false,
outbound_proxy_id: None,
outbound_proxy_enabled: true,
notes: None,
@@ -22,62 +22,72 @@ async fn mount_usage(server: &MockServer, usage: serde_json::Value) {
#[tokio::test]
async fn credits_allow_exhausted_subscription_without_bypassing_spend_restrictions() {
for (metadata, allowed) in [
(
json!({"credits": {"has_credits": true, "unlimited": false}}),
true,
),
(
json!({"credits": {"has_credits": false, "unlimited": true}}),
true,
),
(json!({"credits": {"has_credits": true}}), true),
(json!({"credits": {"unlimited": true}}), true),
(
json!({"credits": {"has_credits": false, "unlimited": false, "balance": "500"}}),
false,
),
(json!({"credits": null}), false),
(
json!({"credits": {"has_credits": true}, "spend_control": {"reached": true}}),
false,
),
(
json!({"credits": {"has_credits": true}, "rate_limit_reached_type": {"type": "spend_limit_reached"}}),
false,
),
(
json!({"credits": {"has_credits": true}, "rate_limit_reached_type": {"type": "rate_limit_reached"}}),
true,
),
] {
let store = Arc::new(MemoryAccountStore::default());
create_account_with_enabled(&store, "acct_credits", false).await;
let account = store.account("acct_credits").expect("account");
persist_credential_state(&store, &account, CredentialState::Expired).await;
let server = MockServer::start().await;
let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
let mut usage = exhausted_usage();
usage
.as_object_mut()
.expect("usage")
.extend(metadata.as_object().expect("metadata").clone());
mount_usage(&server, usage).await;
for auto_use in [false, true] {
for (metadata, usable) in [
(
json!({"credits": {"has_credits": true, "overage_limit_reached": true}}),
false,
),
(
json!({"credits": {"has_credits": true, "unlimited": false}}),
true,
),
(
json!({"credits": {"has_credits": false, "unlimited": true}}),
true,
),
(json!({"credits": {"has_credits": true}}), true),
(json!({"credits": {"unlimited": true}}), true),
(
json!({"credits": {"has_credits": false, "unlimited": false, "balance": "500"}}),
false,
),
(json!({"credits": null}), false),
(
json!({"credits": {"has_credits": true}, "spend_control": {"reached": true}}),
false,
),
(
json!({"credits": {"has_credits": true}, "rate_limit_reached_type": {"type": "spend_limit_reached"}}),
false,
),
(
json!({"credits": {"has_credits": true}, "rate_limit_reached_type": {"type": "rate_limit_reached"}}),
true,
),
] {
let store = Arc::new(MemoryAccountStore::default());
create_account_with_enabled(&store, "acct_credits", false).await;
if auto_use {
store.enable_quota_credits("acct_credits");
}
let allowed = auto_use && usable;
let account = store.account("acct_credits").expect("account");
persist_credential_state(&store, &account, CredentialState::Expired).await;
let server = MockServer::start().await;
let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
let mut usage = exhausted_usage();
usage
.as_object_mut()
.expect("usage")
.extend(metadata.as_object().expect("metadata").clone());
mount_usage(&server, usage).await;
let snapshot = service
.refresh_account(account.id())
.await
.expect("refresh credits");
assert_eq!(
snapshot.quota().access() == QuotaAccessState::Allowed,
allowed,
"{metadata}"
);
assert_eq!(snapshot.fact().remaining_percent(), Some(0));
let current = store.account("acct_credits").expect("current account");
assert_eq!(current.quota().is_exhausted(), !allowed, "{metadata}");
assert!(!current.enabled());
assert_eq!(current.credential_state(), CredentialState::Expired);
let snapshot = service
.refresh_account(account.id())
.await
.expect("refresh credits");
assert_eq!(
snapshot.quota().access() == QuotaAccessState::Allowed,
allowed,
"{metadata}"
);
assert_eq!(snapshot.fact().remaining_percent(), Some(0));
let current = store.account("acct_credits").expect("current account");
assert_eq!(current.quota().is_exhausted(), !allowed, "{metadata}");
assert!(!current.enabled());
assert_eq!(current.credential_state(), CredentialState::Expired);
}
}
}
@@ -86,6 +96,7 @@ async fn credits_refresh_clears_old_recovery_and_locks_again_only_when_spent() {
for periodic in [false, true] {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_credits_recovery").await;
store.enable_quota_credits("acct_credits_recovery");
let account = store.account("acct_credits_recovery").expect("account");
let server = MockServer::start().await;
let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
@@ -161,6 +172,7 @@ async fn credits_refresh_clears_old_recovery_and_locks_again_only_when_spent() {
async fn credits_query_started_before_new_exhaustion_cannot_restore_access() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_stale_credits").await;
store.enable_quota_credits("acct_stale_credits");
let account = store.account("acct_stale_credits").expect("account");
let server = MockServer::start().await;
let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
@@ -213,3 +225,45 @@ async fn credits_query_started_before_new_exhaustion_cannot_restore_access() {
.is_exhausted()
);
}
#[tokio::test]
async fn passive_exhaustion_requires_fresh_credits_and_revokes_access_when_spent() {
let store = Arc::new(MemoryAccountStore::default());
let id = "acct_passive_credit_policy";
create_account(&store, id).await;
store.enable_quota_credits(id);
let service = quota_service(&store);
let window = |used| json!({"primary": {"used_percent": used, "window_minutes": 300}});
let credits = |available| json!({"has_credits": available, "unlimited": false});
for (facts, expected) in [
(
json!({"rate_limits": window(20), "credits": credits(true)}),
QuotaAccessState::Allowed,
),
(
json!({"rate_limits": window(100)}),
QuotaAccessState::Exhausted,
),
(
json!({"rate_limits": window(100), "credits": credits(true)}),
QuotaAccessState::Allowed,
),
(
json!({"credits": credits(false)}),
QuotaAccessState::Exhausted,
),
] {
let mut wire = facts;
wire["type"] = json!("codex.rate_limits");
let update = parse_rate_limits_event(&wire).expect("rate limit event");
service
.synchronize_passive_rate_limits(&store.account(id).unwrap(), &[update])
.await
.unwrap();
assert_eq!(
store.account(id).unwrap().quota().access(),
expected,
"{wire}"
);
}
}
@@ -155,7 +155,7 @@ fn exhausted_quota(reset_at: Option<SystemTime>) -> QuotaState {
}
#[tokio::test]
async fn successful_inference_headers_are_authoritative_even_at_full_display_usage() {
async fn successful_inference_at_the_subscription_limit_stops_further_scheduling() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_passive_allowed";
create_account(&store, account_id).await;
@@ -182,11 +182,11 @@ async fn successful_inference_headers_are_authoritative_even_at_full_display_usa
.expect("quota snapshot");
assert_eq!(snapshot.fact().remaining_percent(), Some(0));
assert_eq!(snapshot.quota().access(), QuotaAccessState::Allowed);
assert_eq!(current.quota().access(), QuotaAccessState::Allowed);
assert_eq!(snapshot.quota().access(), QuotaAccessState::Exhausted);
assert_eq!(current.quota().access(), QuotaAccessState::Exhausted);
assert_eq!(
current.status_projection(SystemTime::now(), None).status,
AccountStatus::Normal
AccountStatus::QuotaExhausted
);
}
@@ -8326,6 +8326,7 @@ async fn official_usage_limit_failure_persists_fact_without_fabricating_usage()
let account_id = "acct_usage_limit_request_path";
let reset_at = 1_900_000_000;
create_account(&store, account_id).await;
store.enable_quota_credits(account_id);
let account = store.account(account_id).expect("created account");
let observed_at = SystemTime::now();
store
@@ -8511,24 +8512,28 @@ async fn official_usage_limit_failure_persists_fact_without_fabricating_usage()
#[tokio::test]
async fn usage_limit_failure_recovers_with_credits_without_replaying_the_failed_request() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_provider_contract";
create_account(&store, account_id).await;
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"rate_limit": {
"allowed": false,
"limit_reached": true,
"primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
},
"credits": {"has_credits": true, "unlimited": false, "balance": "20"}
})))
.expect(1)
.mount(&server)
.await;
Mock::given(method("POST"))
for enabled in [false, true] {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_provider_contract";
create_account(&store, account_id).await;
if enabled {
store.enable_quota_credits(account_id);
}
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"rate_limit": {
"allowed": false,
"limit_reached": true,
"primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
},
"credits": {"has_credits": true, "unlimited": false, "balance": "20"}
})))
.expect(if enabled { 1 } else { 0 })
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/codex/responses"))
.respond_with(ResponseTemplate::new(429).set_body_json(json!({
"error": {"type": "usage_limit_reached", "message": "usage limit reached", "resets_at": 1_900_000_000}
@@ -8536,53 +8541,61 @@ async fn usage_limit_failure_recovers_with_credits_without_replaying_the_failed_
.expect(1)
.mount(&server)
.await;
let mut stream = provider_with_base_url(&store, server.uri())
.execute(
planned_request("openai", http_generate_operation()),
context("req_usage_limit_credits", CancellationToken::new()),
)
.await
.expect("provider stream");
let failure = loop {
match stream.next().await {
Some(Ok(_)) => {}
Some(Err(error)) => break error,
None => panic!("quota failure must reach the caller"),
}
};
assert_eq!(failure.kind(), ProviderErrorKind::QuotaExhausted);
assert!(
store
.account(account_id)
.expect("account")
.quota()
.is_exhausted()
);
let mut stream = provider_with_base_url(&store, server.uri())
.execute(
planned_request("openai", http_generate_operation()),
context("req_usage_limit_credits", CancellationToken::new()),
)
.await
.expect("provider stream");
let failure = loop {
match stream.next().await {
Some(Ok(_)) => {}
Some(Err(error)) => break error,
None => panic!("quota failure must reach the caller"),
}
};
assert_eq!(failure.kind(), ProviderErrorKind::QuotaExhausted);
assert!(
store
.account(account_id)
.expect("account")
.quota()
.is_exhausted()
);
timeout(Duration::from_secs(5), async {
while store
.account(account_id)
.expect("account")
.quota()
.is_exhausted()
{
tokio::time::sleep(Duration::from_millis(10)).await;
if !enabled {
tokio::time::sleep(Duration::from_millis(3100)).await;
assert!(store.account(account_id).unwrap().quota().is_exhausted());
assert!(stream.next().await.is_none());
server.verify().await;
continue;
}
})
.await
.expect("credits restore access after background refresh");
assert!(stream.next().await.is_none());
let account = store.account(account_id).expect("recovered account");
assert_eq!(account.quota().access(), QuotaAccessState::Allowed);
assert_eq!(account.credential_state(), CredentialState::Ready);
let quota = store.quota_json(account_id).expect("persisted credits");
assert_eq!(quota.pointer("/credits/has_credits"), Some(&json!(true)));
assert_eq!(
quota.pointer("/rate_limits_by_limit_id/codex/primary_window/used_percent"),
Some(&json!(100))
);
assert!(quota.get("_quota_recovery").is_none());
server.verify().await;
timeout(Duration::from_secs(5), async {
while store
.account(account_id)
.expect("account")
.quota()
.is_exhausted()
{
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("credits restore access after background refresh");
assert!(stream.next().await.is_none());
let account = store.account(account_id).expect("recovered account");
assert_eq!(account.quota().access(), QuotaAccessState::Allowed);
assert_eq!(account.credential_state(), CredentialState::Ready);
let quota = store.quota_json(account_id).expect("persisted credits");
assert_eq!(quota.pointer("/credits/has_credits"), Some(&json!(true)));
assert_eq!(
quota.pointer("/rate_limits_by_limit_id/codex/primary_window/used_percent"),
Some(&json!(100))
);
assert!(quota.get("_quota_recovery").is_none());
server.verify().await;
}
}
#[tokio::test]
@@ -9989,7 +10002,7 @@ async fn quota_limited_account_diagnostic_uses_upstream() {
}
#[tokio::test]
async fn successful_response_treats_inference_success_as_authoritative_allowance() {
async fn successful_response_with_full_subscription_blocks_subsequent_requests() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_success_exhausted";
create_account(&store, account_id).await;
@@ -10037,7 +10050,7 @@ async fn successful_response_treats_inference_success_as_authoritative_allowance
.expect("account after successful response")
.quota()
.access(),
QuotaAccessState::Allowed
QuotaAccessState::Exhausted
);
assert!(store.has_quota(account_id));
}
@@ -10099,7 +10112,7 @@ async fn successful_http_sse_rate_limit_event_persists_structured_exhaustion() {
.account(account_id)
.expect("account after HTTP SSE response");
assert_eq!(account.credential_state(), CredentialState::Ready);
assert_eq!(account.quota().access(), QuotaAccessState::Allowed);
assert_eq!(account.quota().access(), QuotaAccessState::Exhausted);
assert!(store.has_quota(account_id));
}
@@ -150,6 +150,14 @@ impl MemoryAccountStore {
.map(|stored| stored.account.clone())
}
pub(crate) fn enable_quota_credits(&self, id: &str) {
let mut accounts = self.accounts.lock().unwrap();
let stored = accounts
.get_mut(&ProviderAccountId::new(id).unwrap())
.unwrap();
stored.account = stored.account.clone().with_auto_use_quota_credits(true);
}
pub(crate) fn set_model_access(
&self,
id: &str,
@@ -764,6 +772,7 @@ fn rebuild_account(current: &ProviderAccount, rebuild: AccountRebuild) -> Provid
rebuild.last_error_message,
)
.with_scheduling(current.concurrency_limit(), current.weight())
.with_auto_use_quota_credits(current.auto_use_quota_credits())
.with_refresh_schedule(rebuild.has_refresh_token, rebuild.next_refresh_at)
.with_outbound_proxy(current.outbound_proxy().cloned())
.with_request_location(current.request_location().cloned())
@@ -768,6 +768,7 @@ fn provider_ports_with_catalog(
fn account_record(account: &ProviderAccount) -> AccountRecord {
let now = Utc::now();
AccountRecord {
auto_use_quota_credits: false,
outbound_proxy_id: None,
outbound_proxy_enabled: true,
notes: None,
+1
View File
@@ -37,3 +37,4 @@ d05442ee4cb85db8131e015dbed9c6ad988f93306abcda34647317674857ee44 0036_account_d
b215ea90d5c8753aeb8192b17ffe65f7a95c1d620aaeaf9fc4337d8cf6f2634e 0037_retry_after_wait_limit.sql
321f4a09367717eb13c3cf5dbfd1f4b014e31c615a74bf6e3b09b8e810d87198 0038_preserve_client_key_ledger.sql
80ba91cad23667ace80a43daf905ca78fd0f6de5fb57f444c9ebbfa2fab90c34 0039_queue_timeouts.sql
2438e107835cd69effdb37e4d008b3b631a287da9e41185b04dd239449cb6733 0040_quota_credit_policy.sql
@@ -0,0 +1,26 @@
alter table provider_accounts
add column auto_use_quota_credits boolean not null default false,
drop constraint provider_accounts_quota_evidence_ck,
drop constraint provider_accounts_quota_fact_ck,
add constraint provider_accounts_quota_evidence_ck check (
quota_evidence in ('provider_denied', 'account_limit_reached', 'usage_limit_reached', 'payment_required', 'credits_available')
),
add constraint provider_accounts_quota_fact_ck check (
(quota_access_state = 'unknown' and quota_evidence is null and quota_reset_at is null)
or (quota_access_state = 'allowed' and quota_access_observed_at is not null
and ((quota_evidence is null and quota_reset_at is null) or quota_evidence is not distinct from 'credits_available'))
or (quota_access_state = 'exhausted' and quota_evidence is not null and quota_evidence <> 'credits_available' and quota_access_observed_at is not null)
);
-- 在行锁内按当前配置投影仅点数可用的观测,晚返回的查询不能覆盖关闭开关的决定
create function apply_quota_credit_policy() returns trigger language plpgsql as $$
begin
new.quota_access_state := 'exhausted';
new.quota_evidence := 'account_limit_reached';
return new;
end
$$;
create trigger provider_accounts_quota_credit_policy
before insert or update of auto_use_quota_credits, quota_evidence, quota_access_state on provider_accounts
for each row when (not new.auto_use_quota_credits and new.quota_evidence = 'credits_available') execute function apply_quota_credit_policy();
+7 -5
View File
@@ -1133,11 +1133,13 @@ config 返回 `{ name, plaintextKey }`,仅读取服务端会话绑定的当前
`quota.credits` 返回上游点数余额,未提供点数信息时为 `null`。对象包含 `hasCredits`、`unlimited` 和
`balance`,余额为保留上游精度的十进制字符串,未提供余额时为 `null`,明确的零余额为 `"0"`。
OpenAI 点数随现有额度刷新和正常请求的额度信息同步,不与主动重置卡库存或百分比限额合并,
不根据 `balance` 单独推导可用性。主动查询明确 `hasCredits=true` 或 `unlimited=true`,且无阻止
点数支出的上限或工作区限制时,即使套餐窗口耗尽也可使用点数,并解除已有额度耗尽状态。
真实额度拒绝后的后台回读同样支持此恢复路径,但不会自动重发本次失败请求;点数不能清除
凭据错误、停用或冷却状态。上游点数信息也可能延迟,后续真实拒绝仍会记录额度耗尽
点数余额与重置卡库存独立。支持点数的账号返回 `capabilities.quotaCredits=true`,通过
`POST /api/admin/accounts/batch-update` 提交 `{ "accountIds": ["acct_..."], "autoUseQuotaCredits": true }`
设置,账号字段 `autoUseQuotaCredits` 默认 `false`。关闭时,订阅额度耗尽即停止调度,即使还有点数。
开启后,只有上游确认 `hasCredits=true` 或 `unlimited=true` 且没有支出限制,才允许靠点数继续请求;
余额数字本身不能证明点数可用。对已耗尽的账号开启时会复查额度,关闭时立即停止仅靠点数的放行
点数不能清除凭据错误、停用或冷却状态。实际请求被上游拒绝后不会为了使用点数重发同一次请求
账号名称是本地显示信息,与账号 ID、上游邮箱和凭据独立。编辑接口的可选 `name` 省略或为 `null` 时保留原值;
提供字符串时去除首尾空白,空字符串清除名称并回退显示邮箱,原始输入最多 512 个 UTF-8 字节且不能包含控制字符。
+1 -1
View File
@@ -1026,7 +1026,7 @@ credential 与 quota 是两组独立事实:credential refresh 不等于 quota
空值及 `unknown` 不覆盖套餐,同族泛化值保留具体子类型。Token 刷新保留提交时的账号资料
OpenAI Provider 的额度快照统一解释点数可用性,主动查询、耗尽恢复与失败后回读复用同一规则。
可用点数独立于套餐窗口;通过点数恢复时原子保存新快照、访问事实并清理旧窗口恢复进度。
套餐耗尽后,仅对开启自动使用点数的账号以可用点数放行。点数放行依据与快照原子保存,关闭开关在同一账号事务内撤销该依据,迟到的点数观测不能绕过已关闭的开关。
无可用点数的失败回读保留已确认的拒绝;常规刷新在到期前仍按窗口证据恢复,本次耗尽之后的
已记录窗口 reset 全部到期后可依据新的明确放行观察恢复。查询沿用 credential revision 与
观察时间隔离,不允许旧查询撤销较新的失败;恢复条件和后台复核节奏见 [账号 API](api.md#账号能力导入与-oauth)
@@ -66,6 +66,7 @@ interface AccountUpdateParam {
}
interface AccountBatchUpdateParam {
autoUseQuotaCredits?: boolean
outboundProxyUrl?: string
outboundProxyId?: string
accountIds: string[]
@@ -138,6 +138,7 @@ export interface AccountModelAccess {
}
export interface Account {
autoUseQuotaCredits: boolean
upstream: AccountUpstream | null
capabilities: AccountCapabilities
outboundProxyId: string | null
@@ -199,6 +200,7 @@ export interface AccountUpstream {
}
export interface AccountCapabilities {
quotaCredits: boolean
quota: boolean
quotaRefresh: boolean
profile: boolean
+2 -1
View File
@@ -13,6 +13,8 @@
"s": "{duration} s",
"unlimited": "Unlimited",
"noPointsAvailableYet": "No credits available",
"autoUsePoints": "Automatically use credits",
"autoUsePointsHelp": "Use available credits when the subscription limit is reached",
"quotaPoints": "Credits",
"point": "credits",
"requestsInThisWindow": "Requests in this window: {requestDisplay}",
@@ -515,7 +517,6 @@
"label": "Automatically use expiring reset credits"
},
"automaticallyUseResetCard": "Automatically use resets",
"automaticallyUsed": "Auto use",
"automaticUsageRules": "Automatic usage rules",
"setGlobalLeadTime": "Set global lead time",
"minutesInAdvance": "{leadMinutes} min in advance",
@@ -13,6 +13,8 @@
"s": "{duration} 秒",
"unlimited": "无限",
"noPointsAvailableYet": "暂无可用点数",
"autoUsePoints": "自动使用额度点数",
"autoUsePointsHelp": "订阅额度耗尽后,使用可用点数继续请求",
"quotaPoints": "额度点数",
"point": "点",
"requestsInThisWindow": "窗口请求:{requestDisplay} 次",
@@ -515,7 +517,6 @@
"label": "自动使用临期重置卡"
},
"automaticallyUseResetCard": "自动使用重置卡",
"automaticallyUsed": "自动使用",
"automaticUsageRules": "自动使用规则",
"setGlobalLeadTime": "设置全局提前时间",
"minutesInAdvance": "提前 {leadMinutes} 分钟",
@@ -13,6 +13,8 @@
"s": "{duration} 秒",
"unlimited": "無限",
"noPointsAvailableYet": "暫無可用點數",
"autoUsePoints": "自動使用額度點數",
"autoUsePointsHelp": "訂閱額度耗盡後,使用可用點數繼續請求",
"quotaPoints": "額度點數",
"point": "點",
"requestsInThisWindow": "窗口請求:{requestDisplay} 次",
@@ -515,7 +517,6 @@
"label": "自動使用臨期重置卡"
},
"automaticallyUseResetCard": "自動使用重置卡",
"automaticallyUsed": "自動使用",
"automaticUsageRules": "自動使用規則",
"setGlobalLeadTime": "設置全局提前時間",
"minutesInAdvance": "提前 {leadMinutes} 分鐘",
@@ -1,18 +1,40 @@
<script setup lang="ts">
import type { AccountQuotaCredits } from '@/api'
import type { Account } from '@/api'
import { computed } from 'vue'
import { ZSwitch } from '@codex-proxy/ui'
import { computed, shallowRef } from 'vue'
import { useI18n } from 'vue-i18n'
import { batchUpdateAccounts } from '@/api'
import { useAsyncAction } from '@/composables/useAsyncAction'
import { numberLocale } from '@/utils/format'
const props = defineProps<{
credits: AccountQuotaCredits | null
account: Account
}>()
const emit = defineEmits<{ accountUpdated: [account: Account] }>()
const { loading, run } = useAsyncAction()
const pending = shallowRef<boolean | null>(null)
const credits = computed(() => props.account.quota.credits)
async function toggleCredits(enabled: boolean) {
const accountId = props.account.id
await run(async () => {
pending.value = enabled
await batchUpdateAccounts({
accountIds: [accountId],
autoUseQuotaCredits: enabled,
})
if (props.account.id === accountId)
emit('accountUpdated', { ...props.account, autoUseQuotaCredits: enabled })
})
pending.value = null
}
const { t } = useI18n({ useScope: 'global' })
const balanceDisplay = computed(() => {
const balance = props.credits?.balance?.trim()
const balance = credits.value?.balance?.trim()
const decimal = balance?.match(/^(-?)(\d+)(?:\.(\d+))?$/)
if (!decimal)
return null
@@ -24,28 +46,42 @@ const balanceDisplay = computed(() => {
: grouped
})
const value = computed(() => {
if (props.credits?.unlimited)
if (credits.value?.unlimited)
return t('accounts.quota.unlimited')
if (balanceDisplay.value !== null)
return balanceDisplay.value
if (props.credits && !props.credits.hasCredits)
if (credits.value && !credits.value.hasCredits)
return t('accounts.quota.noPointsAvailableYet')
return t('usage.balanceUnavailable')
})
const hasBalance = computed(() => !props.credits?.unlimited && balanceDisplay.value !== null)
const hasBalance = computed(() => !credits.value?.unlimited && balanceDisplay.value !== null)
</script>
<template>
<div class="flex min-w-0 flex-wrap items-baseline justify-between gap-x-4 gap-y-1 px-4 pb-4" :aria-label="t('accounts.quota.quotaPoints')">
<span class="text-z-sm font-heavy text-z-text">{{ t('accounts.quota.quotaPoints') }}</span>
<span class="flex min-w-0 items-baseline gap-1.5">
<strong
class="min-w-0 text-right text-z-sm font-heavy wrap-anywhere"
:class="hasBalance ? 'font-mono tabular-nums text-z-text' : 'text-z-text-secondary'"
>
{{ value }}
</strong>
<span v-if="hasBalance" class="shrink-0 text-z-xs text-z-text-tertiary">{{ t('accounts.quota.point') }}</span>
</span>
</div>
<section class="grid gap-3 rounded-z bg-z-fill-quaternary px-4 py-3.5" :aria-label="t('accounts.quota.quotaPoints')">
<div class="flex min-w-0 flex-wrap items-baseline justify-between gap-x-4 gap-y-1">
<span class="text-z-sm font-heavy text-z-text">{{ t('accounts.quota.quotaPoints') }}</span>
<span class="flex min-w-0 items-baseline gap-1.5">
<strong
class="min-w-0 text-right text-z-sm font-heavy wrap-anywhere"
:class="hasBalance ? 'font-mono tabular-nums text-z-text' : 'text-z-text-secondary'"
>
{{ value }}
</strong>
<span v-if="hasBalance" class="shrink-0 text-z-xs text-z-text-tertiary">{{ t('accounts.quota.point') }}</span>
</span>
</div>
<div v-if="account.capabilities.quotaCredits" class="flex items-center justify-between gap-4">
<span class="min-w-0">
<span class="block text-z-sm font-heavy text-z-text">{{ t('accounts.quota.autoUsePoints') }}</span>
<span class="mt-1 block text-z-xs text-z-text-secondary">{{ t('accounts.quota.autoUsePointsHelp') }}</span>
</span>
<ZSwitch
:model-value="pending ?? account.autoUseQuotaCredits"
:aria-label="t('accounts.quota.autoUsePoints')"
:disabled="loading"
@update:model-value="toggleCredits"
/>
</div>
</section>
</template>
@@ -69,11 +69,11 @@ watch(() => props.accountId, () => {
<div class="flex flex-wrap items-center justify-between gap-2">
<div class="flex items-center gap-1">
<h3 class="m-0 text-z-sm font-heavy text-z-text">
{{ t('accounts.quotaReset.automaticallyUsed') }}
{{ t('accounts.quotaReset.automaticallyUseResetCard') }}
</h3>
<ZPopover trigger="click" placement="bottom-start" role="dialog">
<ZPopover trigger="hover" placement="bottom-start" role="tooltip">
<template #reference="{ open, id }">
<ZIconButton variant="link" :title="undefined" size="small" class="size-6! p-0!" :aria-label="t('accounts.quotaReset.automaticUsageRules')" :aria-expanded="open" :aria-controls="id">
<ZIconButton variant="link" :title="undefined" size="small" class="size-6! p-0!" :aria-label="t('accounts.quotaReset.automaticUsageRules')" :aria-describedby="open ? id : undefined">
<CircleHelp class="size-3.5" />
</ZIconButton>
</template>
@@ -16,6 +16,7 @@ const props = defineProps<{
}>()
const emit = defineEmits<{
accountUpdated: [account: Account]
consumed: [accountId: string]
}>()
@@ -164,9 +165,10 @@ function handleRequestConsume(creditId: string) {
<div v-else class="grid gap-4">
<section class="overflow-hidden rounded-z bg-z-fill-quaternary" :aria-label="t('accounts.quotaReset.currentQuota')">
<UsageLimits :windows="account.quota.windows" />
<AccountQuotaCredits :credits="account.quota.credits" />
</section>
<AccountQuotaCredits :key="account.id" :account="account" @account-updated="emit('accountUpdated', $event)" />
<ResetCreditAutomation v-if="account.capabilities.consumeResetCredit" :key="account.id" :account-id="account.id" />
<section v-if="ambiguous" class="flex items-start gap-3 rounded-z bg-z-warning-container px-4 py-3.5" role="status">
@@ -93,6 +93,7 @@ watch(hasPersonalInfo, (available) => {
<AccountResetCredits
v-if="account.capabilities.resetCredits"
:account="account"
@account-updated="emit('accountUpdated', $event)"
@consumed="emit('quotaReset', $event)"
/>
<ZIconButton