Compare commits

...
11 Commits
Author SHA1 Message Date
ZyyandCodex 1294e50b76 chore: bump version to 3.1.10
Co-authored-by: Codex <noreply@openai.com>
2026-08-05 12:22:46 +08:00
ZyyandCodex a0e540642b fix(oauth): bound refresh recovery and persist failure reasons
Co-authored-by: Codex <noreply@openai.com>
2026-08-05 12:21:16 +08:00
ZyyandCodex 14d6a7de31 fix(dashboard): align summary with daily trend
Co-authored-by: Codex <noreply@openai.com>
2026-08-05 10:46:14 +08:00
ZyyandCodex cd7b60a3ba chore: bump version to 3.1.9
Co-authored-by: Codex <noreply@openai.com>
2026-08-05 00:03:39 +08:00
ZyyandCodex ebbed22f36 fix(openai): keep Codex wire OS exact
Co-authored-by: Codex <noreply@openai.com>
2026-08-05 00:03:21 +08:00
ZyyandCodex f6e5d8f673 fix(openai): drop null secondary quota windows
Co-authored-by: Codex <noreply@openai.com>
2026-08-05 00:03:11 +08:00
ZyyandCodex bf1c6211b8 chore: bump version to 3.1.8
Co-authored-by: Codex <noreply@openai.com>
2026-08-04 22:30:40 +08:00
ZyyandCodex d388fa344a fix(openai): drop placeholder secondary quota windows
Co-authored-by: Codex <noreply@openai.com>
2026-08-04 22:30:04 +08:00
ZyyandCodex eabb61b08c fix(openai): restore Codex Core app-server identity
Co-authored-by: Codex <noreply@openai.com>
2026-08-04 22:30:01 +08:00
ZyyandCodex a5c3438927 fix(openai): isolate thread-spawn session affinity
Co-authored-by: Codex <noreply@openai.com>
2026-08-04 22:29:53 +08:00
ZyyandCodex b6948542bb fix: align Codex Desktop HTTP identity
- send the official Desktop HTTP user agent across OpenAI upstream paths
- keep the audited model catalog version separate from the wire identity
- remove CLI release drift and clarify the dashboard client identity

Co-authored-by: Codex <noreply@openai.com>
2026-08-04 18:36:03 +08:00
41 changed files with 1656 additions and 308 deletions
+1 -1
View File
@@ -134,7 +134,7 @@ fn config_loader_should_reject_removed_fingerprint_section() {
}
#[test]
fn config_loader_should_reject_invalid_codex_core_version() {
fn config_loader_should_reject_invalid_codex_cli_version() {
assert_rejected(valid_config().replace("codex_version: '0.146.0'", "codex_version: 'latest'"));
}
@@ -310,16 +310,6 @@ pub(crate) fn usage_range(
domain::TimeRange::new(start, end).map_err(|_| WireValidationError::new("timeRange"))
}
pub(crate) fn dashboard_range(
start: Option<&str>,
end: Option<&str>,
) -> Result<domain::TimeRange, WireValidationError> {
let end = parse_datetime(end)?.unwrap_or_else(Utc::now);
let start =
parse_datetime(start)?.unwrap_or_else(|| domain::china_day_start(end) - Duration::days(1));
domain::TimeRange::new(start, end).map_err(|_| WireValidationError::new("timeRange"))
}
pub(crate) fn dashboard_today_range(
start: Option<&str>,
end: Option<&str>,
@@ -38,7 +38,8 @@ where
S: AdminSessionState + Send + Sync,
{
let kind = query.trend_kind().map_err(map_wire_error)?;
let range = dashboard_range(query.start_time.as_deref(), query.end_time.as_deref())
// 概览默认按中国时区当日统计,与单独趋势接口保持同一口径。
let range = dashboard_today_range(query.start_time.as_deref(), query.end_time.as_deref())
.map_err(map_wire_error)?;
let result = state
.admin_services()
+10 -1
View File
@@ -83,6 +83,7 @@ pub(super) struct AdminTestFixture {
pub diagnostics: Arc<Mutex<Vec<DiagnosticObservation>>>,
pub ops_errors: Arc<Mutex<Vec<OpsError>>>,
pub dashboard_observation: Arc<Mutex<Option<DashboardObservation>>>,
pub dashboard_summary_range: Arc<Mutex<Option<TimeRange>>>,
}
impl AdminTestFixture {
@@ -100,12 +101,14 @@ impl AdminTestFixture {
let diagnostics = Arc::new(Mutex::new(Vec::new()));
let ops_errors = Arc::new(Mutex::new(Vec::new()));
let dashboard_observation = Arc::new(Mutex::new(None));
let dashboard_summary_range = Arc::new(Mutex::new(None));
let unused = Arc::new(UnusedStore {
usage_records: Arc::clone(&usage_records),
usage_detail: Arc::clone(&usage_detail),
diagnostics: Arc::clone(&diagnostics),
ops_errors: Arc::clone(&ops_errors),
dashboard_observation: Arc::clone(&dashboard_observation),
dashboard_summary_range: Arc::clone(&dashboard_summary_range),
});
let stores = AdminStorePorts::new(
unused.clone(),
@@ -142,6 +145,7 @@ impl AdminTestFixture {
diagnostics,
ops_errors,
dashboard_observation,
dashboard_summary_range,
}
}
@@ -413,6 +417,7 @@ struct UnusedStore {
diagnostics: Arc<Mutex<Vec<DiagnosticObservation>>>,
ops_errors: Arc<Mutex<Vec<OpsError>>>,
dashboard_observation: Arc<Mutex<Option<DashboardObservation>>>,
dashboard_summary_range: Arc<Mutex<Option<TimeRange>>>,
}
#[async_trait]
@@ -523,7 +528,11 @@ impl AccountStore for UnusedStore {
#[async_trait]
impl ObservabilityStore for UnusedStore {
async fn dashboard_summary(&self, _: TimeRange) -> AdminStoreResult<DashboardObservation> {
async fn dashboard_summary(&self, range: TimeRange) -> AdminStoreResult<DashboardObservation> {
*self
.dashboard_summary_range
.lock()
.expect("dashboard summary range") = Some(range);
self.dashboard_observation
.lock()
.expect("dashboard observation")
@@ -270,6 +270,61 @@ async fn dashboard_summary_should_include_quota_exhaustion_in_unavailable_headli
);
}
#[tokio::test]
async fn dashboard_summary_should_default_to_current_china_day() {
use axum::{
body::Body,
http::{Request, StatusCode, header},
};
use chrono::{Duration, Utc};
use gateway_admin::model::observability::{
AccountPoolMetrics, AttemptMetrics, DashboardObservation, RequestMetrics, TimeRange,
china_day_start,
};
use gateway_api::admin::observability;
use tower::ServiceExt as _;
use crate::admin::{AdminTestFixture, AdminTestState};
let fixture = AdminTestFixture::new().await;
fixture.auth.insert_session("valid-session");
let now = Utc::now();
*fixture
.dashboard_observation
.lock()
.expect("dashboard observation") = Some(DashboardObservation {
range: TimeRange::new(now - Duration::hours(1), now).expect("dashboard range"),
requests: RequestMetrics::default(),
attempts: AttemptMetrics::default(),
provider_accounts: AccountPoolMetrics::default(),
trend: Vec::new(),
account_usage: Vec::new(),
recent_requests: Vec::new(),
});
let response = observability::router::<AdminTestState>()
.with_state(fixture.state())
.oneshot(
Request::builder()
.uri("/api/admin/dashboard/summary")
.header(header::COOKIE, "cpr_admin_session=valid-session")
.header("x-request-id", "req_dashboard_today_range")
.body(Body::empty())
.expect("dashboard request"),
)
.await
.expect("dashboard response");
let range = fixture
.dashboard_summary_range
.lock()
.expect("dashboard summary range")
.take()
.expect("recorded dashboard summary range");
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(range.start, china_day_start(range.end));
}
#[tokio::test]
async fn usage_detail_should_keep_attempt_snapshot_contract() {
use axum::{
@@ -215,6 +215,20 @@ impl AccountAvailability {
}
}
/// 返回账号进入不可恢复错误态、但 Provider 未给出更具体上游原因时的稳定兜底。
///
/// Provider 应优先持久化已脱敏的上游 message 或原因码;该值只避免错误态沿用
/// 上一次无关的错误信息。
#[must_use]
pub const fn fallback_error_reason(self) -> Option<&'static str> {
match self {
Self::Expired => Some("credential_expired"),
Self::Banned => Some("account_banned"),
Self::Invalid => Some("credential_invalid"),
Self::Unknown | Self::Ready | Self::QuotaExhausted => None,
}
}
/// 解析数据库稳定值。
#[must_use]
pub fn parse(value: &str) -> Option<Self> {
+5 -4
View File
@@ -194,12 +194,13 @@ impl ProviderAdmin for OpenAiAdminProvider {
let profile = self.profile.snapshot();
let release = self.desktop_release.snapshot();
let user_agent = profile.user_agent();
let client_identity = format!("{}; {}", profile.originator, profile.desktop_version);
let release = dashboard_desktop_release(&profile, release);
Some(DashboardWireProfile {
provider: self.provider_kind.as_str().to_owned(),
product: profile.originator,
version: profile.desktop_version,
build: Some(profile.desktop_build),
version: profile.codex_version,
build: None,
target: DashboardWireTarget {
os_type: profile.os_type,
os_version: profile.os_version,
@@ -208,8 +209,8 @@ impl ProviderAdmin for OpenAiAdminProvider {
},
user_agent,
attributes: vec![DashboardWireAttribute {
label: "Codex Core".to_owned(),
value: profile.codex_version,
label: "客户端标识".to_owned(),
value: client_identity,
}],
verified_at: Some(profile.verified_at),
release: Some(release),
@@ -278,6 +278,7 @@ impl CodexAuthSettings {
#[serde(deny_unknown_fields)]
pub struct CodexWireProfileConfig {
pub originator: String,
/// npm `@openai/codex` 启动版本;运行时发布任务会更新此画像字段。
pub codex_version: String,
pub desktop_version: String,
pub desktop_build: String,
@@ -292,10 +293,10 @@ impl Default for CodexWireProfileConfig {
fn default() -> Self {
Self {
originator: "Codex Desktop".to_owned(),
codex_version: "0.145.0".to_owned(),
desktop_version: "26.721.31836".to_owned(),
desktop_build: "5828".to_owned(),
os_type: "macOS".to_owned(),
codex_version: "0.146.0".to_owned(),
desktop_version: "26.727.51351".to_owned(),
desktop_build: "6119".to_owned(),
os_type: "Mac OS".to_owned(),
os_version: "15.7.1".to_owned(),
arch: "arm64".to_owned(),
terminal: "unknown".to_owned(),
@@ -36,7 +36,7 @@ use crate::transport::{CodexBackendClient, CodexClientError, CodexRequestContext
use super::agent_identity::{CodexAgentIdentityTaskService, PreparedCodexRuntimeCredential};
use super::repository::{CodexCredentialRepository, CredentialRepositoryError};
use recovery::{quota_state_transition, quota_success_state};
use recovery::{quota_state_reason, quota_state_transition, quota_success_state};
use snapshot::{
parse_account_quota_snapshot, quota_projection_ttl, quota_snapshot_from_observation,
scheduling_signals_from_snapshot,
@@ -430,7 +430,10 @@ impl CodexCredentialQuotaService {
.and_then(|observation| observation.quota)
.map(OpaqueProviderData::into_inner)
.unwrap_or_default();
let quota = confirmed_exhaustion_projection(existing, reset_at);
let quota = confirmed_exhaustion_projection(
normalize_quota_window_placeholders(existing),
reset_at,
);
let observed_at = SystemTime::now();
let outcome = self
.store
@@ -545,21 +548,23 @@ impl CodexCredentialQuotaService {
// 402:真额度耗尽,写入 QuotaExhausted(worker 按 reset_at 自动恢复)。
Some(AccountAvailability::QuotaExhausted) => {
summary.exhausted += 1;
let _ = self
.repository
.apply_state(
&account,
AccountAvailability::QuotaExhausted,
observed_at,
)
.await;
self.persist_upstream_state(
&account,
AccountAvailability::QuotaExhausted,
&error,
observed_at,
)
.await;
}
Some(AccountAvailability::Banned) => {
summary.banned += 1;
let _ = self
.repository
.apply_state(&account, AccountAvailability::Banned, observed_at)
.await;
self.persist_upstream_state(
&account,
AccountAvailability::Banned,
&error,
observed_at,
)
.await;
}
// 429/503 等临时拒绝:不写 availability(限流不改变账号可用性),仅记失败待重试。
None | Some(_) => {
@@ -594,10 +599,12 @@ impl CodexCredentialQuotaService {
let refreshed_fact = refreshed_snapshot.fact();
let previous_reset_at = self.previous_confirmed_reset(account).await;
let now = SystemTime::now();
let object = value
.as_object()
.cloned()
.ok_or(CodexCredentialQuotaError::InvalidCredentialData)?;
let object = normalize_quota_window_placeholders(
value
.as_object()
.cloned()
.ok_or(CodexCredentialQuotaError::InvalidCredentialData)?,
);
let object = if account.availability() == AccountAvailability::QuotaExhausted
&& quota_success_state(
account.availability(),
@@ -908,9 +915,7 @@ impl CodexCredentialQuotaService {
Err(CodexQuotaFetchError::Upstream { account, error }) => {
// 402/deactivated → 写终态;429/503 等临时拒绝不写 availability(限流不改变账号可用性)。
if let Some(availability) = quota_state_transition(&error) {
let _ = self
.repository
.apply_state(&account, availability, observed_at)
self.persist_upstream_state(&account, availability, &error, observed_at)
.await;
}
return Err(CodexCredentialQuotaError::Upstream {
@@ -927,10 +932,12 @@ impl CodexCredentialQuotaService {
let refreshed_fact = refreshed_snapshot.fact();
let previous_reset_at = self.previous_confirmed_reset(&account).await;
let now = SystemTime::now();
let object = value
.as_object()
.cloned()
.ok_or(CodexCredentialQuotaError::InvalidCredentialData)?;
let object = normalize_quota_window_placeholders(
value
.as_object()
.cloned()
.ok_or(CodexCredentialQuotaError::InvalidCredentialData)?,
);
let object = if account.availability() == AccountAvailability::QuotaExhausted
&& quota_success_state(
account.availability(),
@@ -989,6 +996,29 @@ impl CodexCredentialQuotaService {
Ok(snapshot)
}
async fn persist_upstream_state(
&self,
account: &ProviderAccount,
availability: AccountAvailability,
upstream_error: &CodexClientError,
observed_at: SystemTime,
) {
let reason = quota_state_reason(upstream_error).map(str::to_owned);
if let Err(error) = self
.repository
.apply_state_with_message(account, availability, observed_at, reason.clone())
.await
{
tracing::warn!(
account_id = %account.id(),
?availability,
reason = reason.as_deref().unwrap_or_default(),
error = %error,
"OpenAI quota upstream state write failed"
);
}
}
async fn fetch_usage_with_recovery(
&self,
client: &CodexBackendClient,
@@ -1181,6 +1211,43 @@ fn merge_passive_quota(
quota
}
/// 丢弃 core 限额中上游给出的无事实 `secondary_window` 占位。
///
/// 生产响应头可能携带 `used_percent=0`、零时长和空 reset 的占位,`/usage` 也会
/// 返回 `secondary_window: null`。正常被动同步保留该响应事实,Admin 展示会将其
/// 隐藏;但主动 `/usage` 刷新或 402 确认投影会把一个存在但无事实的字段写成
/// 100%,从而显示并不存在的“次级额度”。因此这些非被动写入口在落库前移除
/// 无事实值。带 reset、时长、正用量、触顶、未知或非法字段的次级窗口均完全按
/// 原有额度逻辑保留。
fn normalize_quota_window_placeholders(mut quota: Map<String, Value>) -> Map<String, Value> {
if let Some(rate_limit) = quota.get_mut("rate_limit").and_then(Value::as_object_mut) {
drop_secondary_window_placeholder(rate_limit);
}
quota
}
fn drop_secondary_window_placeholder(rate_limit: &mut Map<String, Value>) {
let placeholder = rate_limit.get("secondary_window").is_some_and(|window| {
window.is_null()
|| window
.as_object()
.is_some_and(secondary_window_is_placeholder)
});
if placeholder {
rate_limit.remove("secondary_window");
}
}
fn secondary_window_is_placeholder(window: &Map<String, Value>) -> bool {
window.iter().all(|(field, value)| match field.as_str() {
"used_percent" => value
.as_f64()
.is_some_and(|used_percent| used_percent.is_finite() && used_percent == 0.0),
"limit_reached" => value.as_bool() == Some(false),
_ => false,
})
}
/// 用已确认的上游额度拒绝覆盖 core 限额投影。
///
/// 上游 usage 快照可能在结算期间短暂滞后为 98/99% 或 `allowed=true`;保留其他
@@ -50,6 +50,18 @@ pub(crate) fn quota_state_transition(error: &CodexClientError) -> Option<Account
}
}
/// 配额接口将账号变为非 ready 状态时可安全持久化的上游原因码。
pub(crate) fn quota_state_reason(error: &CodexClientError) -> Option<&'static str> {
match quota_state_transition(error)? {
AccountAvailability::Banned => Some("deactivated_workspace"),
AccountAvailability::QuotaExhausted => Some("payment_required"),
AccountAvailability::Unknown
| AccountAvailability::Ready
| AccountAvailability::Expired
| AccountAvailability::Invalid => None,
}
}
fn is_deactivated_workspace(body: &str) -> bool {
serde_json::from_str::<Value>(body)
.ok()
@@ -11,6 +11,7 @@ use gateway_core::provider_ports::{
ProviderCredentialStatePort, ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest,
ProviderRefreshCapacityRequest, ProviderRefreshLeaseRequest, ProviderRefreshPolicy,
ProviderRuntimePolicyPort, ProviderStoreError, provider_refresh_backoff_at,
provider_refresh_retry_at,
};
use secrecy::{ExposeSecret, SecretString};
use thiserror::Error;
@@ -27,13 +28,88 @@ const PROVIDER_NAME: &str = "openai";
const MAX_REFRESH_BATCH: u32 = 1_000;
/// 连续失败计数窗口;每次瞬态失败刷新该 TTL,静默满窗后计数过期归零。
const REFRESH_BACKOFF_WINDOW: Duration = Duration::from_secs(30 * 60);
/// 窗口内先执行五次指数退避,再进入固定恢复周期。
const REFRESH_BACKOFF_MAX_ATTEMPTS: u32 = 5;
/// 耗尽指数退避后的 OAuth 恢复周期。
const REFRESH_RECOVERY_DELAY: Duration = Duration::from_secs(10 * 60);
/// 过期 AT 仍允许 RT 恢复的最长窗口。
const REFRESH_RECOVERY_WINDOW: Duration = Duration::from_secs(24 * 60 * 60);
pub(crate) fn refresh_recovery_deadline(
access_token_expires_at: Option<SystemTime>,
) -> Option<SystemTime> {
access_token_expires_at.and_then(|expires_at| expires_at.checked_add(REFRESH_RECOVERY_WINDOW))
}
fn refresh_recovery_window_exhausted(
access_token_expires_at: Option<SystemTime>,
now: SystemTime,
) -> bool {
refresh_recovery_deadline(access_token_expires_at).is_some_and(|deadline| deadline <= now)
}
fn bounded_refresh_retry_at(
retry_at: SystemTime,
access_token_expires_at: Option<SystemTime>,
observed_at: SystemTime,
) -> SystemTime {
let mut bounded = retry_at;
if let Some(expires_at) = access_token_expires_at
&& expires_at > observed_at
&& bounded > expires_at
{
bounded = expires_at;
}
if let Some(deadline) = refresh_recovery_deadline(access_token_expires_at)
&& bounded > deadline
{
bounded = deadline;
}
bounded
}
fn oauth_refresh_retry_at(
account_id: &ProviderAccountId,
access_token_expires_at: Option<SystemTime>,
observed_at: SystemTime,
attempt: u32,
reason: &'static str,
) -> Result<SystemTime, ProviderStoreError> {
let retry_at = if attempt <= REFRESH_BACKOFF_MAX_ATTEMPTS {
provider_refresh_backoff_at(account_id, observed_at, attempt, reason)?
} else {
provider_refresh_retry_at(account_id, observed_at, REFRESH_RECOVERY_DELAY, reason)?
};
Ok(bounded_refresh_retry_at(
retry_at,
access_token_expires_at,
observed_at,
))
}
fn log_refresh_deferred(
account_id: &ProviderAccountId,
access_token_expires_at: Option<SystemTime>,
attempt: u32,
reason: &'static str,
retry_at: SystemTime,
) {
tracing::warn!(
account_id = %account_id,
attempt,
reason,
retry_at = %DateTime::<Utc>::from(retry_at),
access_token_expires_at = ?access_token_expires_at.map(DateTime::<Utc>::from),
recovery_deadline = ?refresh_recovery_deadline(access_token_expires_at)
.map(DateTime::<Utc>::from),
"OpenAI OAuth refresh deferred"
);
}
fn refresh_due_at(account: &ProviderAccount, now: SystemTime) -> Option<SystemTime> {
if let Some(expires_at) = account.access_token_expires_at()
&& expires_at <= now
{
// AT 已过期时不能继续等待持久退避;后续额度/请求探测都必须先用 RT 续出可用 AT。
return Some(expires_at);
if refresh_recovery_window_exhausted(account.access_token_expires_at(), now) {
// 即使旧 next_refresh_at 异常地落在 deadline 之后,也必须进入终态化路径。
return refresh_recovery_deadline(account.access_token_expires_at());
}
account.next_refresh_at().filter(|next| *next <= now)
}
@@ -145,7 +221,14 @@ impl CodexCredentialRefreshService {
for (account_id, result) in refreshed {
match result {
Ok(outcome) => outcomes.push(outcome),
Err(_) => outcomes.push(CodexCredentialRefreshOutcome::Failed { account_id }),
Err(error) => {
tracing::warn!(
account_id = %account_id,
error = %error,
"OpenAI OAuth refresh attempt failed"
);
outcomes.push(CodexCredentialRefreshOutcome::Failed { account_id });
}
}
}
Ok(outcomes)
@@ -157,6 +240,19 @@ impl CodexCredentialRefreshService {
policy: ProviderRefreshPolicy,
) -> Result<CodexCredentialRefreshOutcome, CodexCredentialRefreshError> {
let account_id = due.account.id().to_string();
if refresh_recovery_window_exhausted(
due.account.access_token_expires_at(),
SystemTime::now(),
) {
return self
.persist_terminal(
&due.account,
AccountAvailability::Expired,
"refresh_recovery_window_exhausted",
CodexCredentialRefreshOutcome::Invalidated { account_id },
)
.await;
}
let capacity = self
.leases
.try_acquire(ProviderLeaseRequest::RefreshCapacity(
@@ -192,6 +288,7 @@ impl CodexCredentialRefreshService {
self.persist_terminal(
&due.account,
AccountAvailability::Expired,
"refresh_invalid_grant",
CodexCredentialRefreshOutcome::Invalidated { account_id },
)
.await
@@ -200,6 +297,7 @@ impl CodexCredentialRefreshService {
self.persist_terminal(
&due.account,
AccountAvailability::Banned,
"account_banned",
CodexCredentialRefreshOutcome::Banned { account_id },
)
.await
@@ -328,6 +426,7 @@ impl CodexCredentialRefreshService {
.persist_terminal(
&due.account,
AccountAvailability::Invalid,
"identity_rejected",
CodexCredentialRefreshOutcome::Invalidated {
account_id: due.account.id().to_string(),
},
@@ -359,8 +458,11 @@ impl CodexCredentialRefreshService {
.record_refresh_backoff(due.account.id(), REFRESH_BACKOFF_WINDOW)
.await
.unwrap_or(1);
let retry_at = provider_refresh_backoff_at(
let access_token_expires_at =
Some(SystemTime::from(signed.access_token_expires_at()));
let retry_at = oauth_refresh_retry_at(
due.account.id(),
access_token_expires_at,
SystemTime::now(),
attempt,
"identity-completion",
@@ -371,6 +473,13 @@ impl CodexCredentialRefreshService {
.await
{
Ok(_) => {
log_refresh_deferred(
due.account.id(),
access_token_expires_at,
attempt,
"identity-completion",
retry_at,
);
return Ok(CodexCredentialRefreshOutcome::Transient {
account_id: due.account.id().to_string(),
});
@@ -384,6 +493,7 @@ impl CodexCredentialRefreshService {
}
}
};
let access_token_expires_at = profile.access_token_expires_at.map(SystemTime::from);
let next_refresh_at = policy
.next_attempt_at(
due.account.id(),
@@ -412,6 +522,13 @@ impl CodexCredentialRefreshService {
.credential_state
.clear_refresh_backoff(due.account.id())
.await;
tracing::info!(
account_id = %due.account.id(),
credential_revision = revision.get(),
access_token_expires_at = ?access_token_expires_at.map(DateTime::<Utc>::from),
next_refresh_at = %next_refresh_at,
"OpenAI OAuth refresh succeeded"
);
Ok(CodexCredentialRefreshOutcome::Refreshed {
account_id: due.account.id().to_string(),
credential_revision: revision.get(),
@@ -430,6 +547,7 @@ impl CodexCredentialRefreshService {
&self,
account: &ProviderAccount,
availability: AccountAvailability,
reason: &'static str,
outcome: CodexCredentialRefreshOutcome,
) -> Result<CodexCredentialRefreshOutcome, CodexCredentialRefreshError> {
match self.repository.load_runtime_credential(account).await {
@@ -443,10 +561,27 @@ impl CodexCredentialRefreshService {
}
match self
.repository
.apply_state(account, availability, SystemTime::now())
.apply_state_with_message(
account,
availability,
SystemTime::now(),
Some(reason.to_owned()),
)
.await
{
Ok(()) => Ok(outcome),
Ok(()) => {
tracing::warn!(
account_id = %account.id(),
?availability,
reason,
access_token_expires_at = ?account.access_token_expires_at()
.map(DateTime::<Utc>::from),
recovery_deadline = ?refresh_recovery_deadline(account.access_token_expires_at())
.map(DateTime::<Utc>::from),
"OpenAI OAuth refresh marked account terminal"
);
Ok(outcome)
}
Err(CredentialRepositoryError::RevisionConflict) => {
Ok(CodexCredentialRefreshOutcome::Stale {
account_id: account.id().to_string(),
@@ -466,10 +601,24 @@ impl CodexCredentialRefreshService {
.record_refresh_backoff(account.id(), REFRESH_BACKOFF_WINDOW)
.await
.unwrap_or(1);
let retry_at =
provider_refresh_backoff_at(account.id(), SystemTime::now(), attempt, reason)?;
let retry_at = oauth_refresh_retry_at(
account.id(),
account.access_token_expires_at(),
SystemTime::now(),
attempt,
reason,
)?;
match self.repository.defer_refresh(account, retry_at).await {
Ok(_) => Ok(true),
Ok(_) => {
log_refresh_deferred(
account.id(),
account.access_token_expires_at(),
attempt,
reason,
retry_at,
);
Ok(true)
}
Err(CredentialRepositoryError::RevisionConflict) => Ok(false),
Err(error) => Err(error.into()),
}
@@ -239,7 +239,8 @@ impl CodexCredentialRepository {
.await
}
/// 带上游错误描述的状态写入;`message` 为 `Some` 时持久化最近错误,恢复 `Ready` 时清空。
/// 带上游错误描述的状态写入;上游 message 优先,终态缺失 message 时写稳定兜底,
/// 恢复 `Ready` 时清空。
pub async fn apply_state_with_message(
&self,
account: &ProviderAccount,
@@ -250,6 +251,9 @@ impl CodexCredentialRepository {
if !account.enabled() {
return Ok(());
}
let message = message
.filter(|value| !value.trim().is_empty())
.or_else(|| availability.fallback_error_reason().map(str::to_owned));
self.store
.apply_state_change(AccountStateChange {
account_id: account.id().clone(),
@@ -25,11 +25,12 @@ use super::agent_identity::CodexAgentIdentityTaskService;
use super::catalog::CodexCredentialCatalogService;
use super::cookie::CodexCookiePolicy;
use super::quota::CodexCredentialQuotaService;
use super::refresh::refresh_recovery_deadline;
use super::repository::{CodexCredentialRepository, CredentialRepositoryError};
use super::security::CodexRuntimeAuthentication;
use super::types::{
CODEX_AUTHENTICATION_KIND_AGENT_IDENTITY, CodexCookie, CodexCookieCaptureOutcome,
RuntimeCodexCookie,
CODEX_AUTHENTICATION_KIND_AGENT_IDENTITY, CODEX_AUTHENTICATION_KIND_OAUTH, CodexCookie,
CodexCookieCaptureOutcome, RuntimeCodexCookie,
};
const CLOUDFLARE_RECOVERY_STALE_AFTER: Duration = Duration::from_secs(60 * 60);
@@ -73,6 +74,21 @@ pub enum CodexAccountFailure {
CloudflarePathBlocked,
}
impl CodexAccountFailure {
/// 上游未提供可展示错误文本时,保留已分类的稳定原因。
const fn fallback_error_reason(self) -> Option<&'static str> {
match self {
Self::CredentialExpired => Some("upstream_credential_expired"),
Self::IdentityVerificationRequired => Some("upstream_identity_verification_required"),
Self::Banned => Some("upstream_account_banned"),
Self::QuotaExhausted => Some("upstream_quota_exhausted"),
Self::UsageLimitExhausted { .. } => Some("upstream_usage_limit_exhausted"),
Self::CloudflarePathBlocked => Some("upstream_cloudflare_path_blocked"),
Self::RateLimited { .. } | Self::CloudflareChallenge { .. } => None,
}
}
}
#[derive(Debug, Clone, Copy)]
struct RiskRecoveryState {
challenge_count: u32,
@@ -733,8 +749,29 @@ impl CodexCredentialSelector {
message: Option<String>,
) -> Result<(), CredentialSelectionError> {
let now = SystemTime::now();
let message = message
.filter(|value| !value.trim().is_empty())
.or_else(|| failure.fallback_error_reason().map(str::to_owned));
match failure {
CodexAccountFailure::CredentialExpired => {
if account.authentication_kind() == CODEX_AUTHENTICATION_KIND_OAUTH
&& account.has_refresh_token()
&& account
.access_token_expires_at()
.is_some_and(|expires_at| expires_at <= now)
&& refresh_recovery_deadline(account.access_token_expires_at())
.is_some_and(|deadline| deadline > now)
{
tracing::info!(
account_id = %account.id(),
access_token_expires_at = ?account.access_token_expires_at()
.map(chrono::DateTime::<chrono::Utc>::from),
recovery_deadline = ?refresh_recovery_deadline(account.access_token_expires_at())
.map(chrono::DateTime::<chrono::Utc>::from),
"OpenAI access token expired; retaining account for bounded OAuth refresh recovery"
);
return Ok(());
}
self.apply_state_with_message(account, AccountAvailability::Expired, now, message)
.await
}
+3 -1
View File
@@ -82,7 +82,7 @@ pub async fn initialize(
profile.clone(),
Arc::new(
OfficialCodexCliReleaseTransport::new()
.map_err(|_| OpenAiInitializeError::DesktopRelease)?,
.map_err(|_| OpenAiInitializeError::CliRelease)?,
),
));
let repository = CodexCredentialRepository::new(Arc::clone(&accounts));
@@ -268,6 +268,8 @@ pub enum OpenAiInitializeError {
Refresh,
#[error("OpenAI Desktop release service could not initialize")]
DesktopRelease,
#[error("OpenAI CLI release service could not initialize")]
CliRelease,
#[error("OpenAI worker plan is invalid")]
Worker,
}
@@ -2435,11 +2435,10 @@ impl ScheduledTask for OpenAiOAuthRefreshTask {
if context.cancellation().is_cancelled() {
return Ok(());
}
let outcomes = self
.service
.refresh_due()
.await
.map_err(|_| WorkerTaskError::safe("OpenAI OAuth refresh failed"))?;
let outcomes = self.service.refresh_due().await.map_err(|error| {
tracing::error!(error = %error, "OpenAI OAuth refresh cycle failed");
WorkerTaskError::safe("OpenAI OAuth refresh failed")
})?;
let mut refreshed = 0_u64;
let mut invalidated = 0_u64;
let mut banned = 0_u64;
@@ -2468,6 +2467,18 @@ impl ScheduledTask for OpenAiOAuthRefreshTask {
}
}
}
if !outcomes.is_empty() {
tracing::info!(
refreshed,
invalidated,
banned,
transient,
lease_unavailable,
stale,
failed,
"OpenAI OAuth refresh cycle completed"
);
}
if transient > 0 || failed > 0 {
tracing::warn!(
refreshed,
@@ -14,21 +14,22 @@ use url::Url;
pub const CODEX_DESKTOP_APPCAST_URL: &str =
"https://persistent.oaistatic.com/codex-app-prod/appcast.xml";
pub const CODEX_CLI_RELEASE_URL: &str = "https://registry.npmjs.org/@openai%2Fcodex/latest";
/// 官方制品画像检查周期。
/// 官方发布元数据检查周期。
pub const APPCAST_POLL_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60);
const APPCAST_TIMEOUT: Duration = Duration::from_secs(30);
const MAX_APPCAST_BYTES: usize = 1024 * 1024;
const MAX_CLI_RELEASE_BYTES: usize = 64 * 1024;
/// Codex Desktop 上游请求身份。
///
/// 启动配置提供经源码审计的 Core、运行环境和初始 Desktop 版本。官方 appcast
/// 检查成功后,会仅同步 Desktop 版本和构建号。
/// 启动配置提供经源码审计的 Core、运行环境和 Desktop 启动版本。npm CLI metadata
/// 与官方 appcast 检查成功后,分别同步 Core 版本和 Desktop 版本及构建号。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CodexWireProfile {
/// `originator` 请求头及 User-Agent 产品名。
pub originator: String,
/// bundled Codex Core 版本;同时用于 `/codex/models?client_version=`。
/// npm `@openai/codex` 启动版本;用于 `/codex/models?client_version=` 与 UA。
pub codex_version: String,
/// Desktop 应用版本,用于 app-server `clientInfo.version` 对应的 UA 后缀。
pub desktop_version: String,
@@ -47,16 +48,18 @@ pub struct CodexWireProfile {
}
impl CodexWireProfile {
/// 按 Codex Core 的官方格式生成最终 User-Agent。
/// 按 bundled Core app-server 的官方格式生成最终 User-Agent。
pub fn user_agent(&self) -> String {
format!(
"{}/{} ({} {}; {}) {}",
"{}/{} ({} {}; {}) {} ({}; {})",
self.originator,
self.codex_version,
self.os_type,
self.os_version,
self.arch,
self.terminal,
self.originator,
self.desktop_version,
)
}
}
@@ -96,7 +99,7 @@ impl CodexWireProfileState {
profile.desktop_build = desktop_build.to_owned();
}
/// 使用官方 npm metadata 发布的 Codex CLI 版本更新运行时 Core 画像。
/// 使用官方 npm metadata 发布的 Codex CLI 版本更新运行时请求画像。
pub fn update_cli_release(&self, codex_version: &str) {
let mut profile = self
.profile
@@ -143,12 +146,12 @@ impl CodexCliReleaseTransport for OfficialCodexCliReleaseTransport {
}
if response
.content_length()
.is_some_and(|size| size > 64 * 1024)
.is_some_and(|size| size > MAX_CLI_RELEASE_BYTES as u64)
{
return Err(CodexCliReleaseError::ResponseTooLarge);
}
let bytes = response.bytes().await?;
if bytes.len() > 64 * 1024 {
if bytes.len() > MAX_CLI_RELEASE_BYTES {
return Err(CodexCliReleaseError::ResponseTooLarge);
}
let document: serde_json::Value = serde_json::from_slice(&bytes)
@@ -10,6 +10,8 @@ use sha2::{Digest, Sha256};
use crate::transport::protocol::responses::CodexResponsesRequest;
const PASSTHROUGH_HEADERS_CONTEXT_KEY: &str = "opaque_request_headers";
const THREAD_SPAWN_SUBAGENT_KIND: &str = "thread_spawn";
const THREAD_SPAWN_CONVERSATION_PREFIX: &str = "thread-spawn:";
const CROSS_ACCOUNT_IDENTITY_KEYS: &[&str] = &[
"authorization",
@@ -160,28 +162,52 @@ fn string_value(value: Option<&Value>) -> Option<String> {
pub(crate) fn derive_conversation_anchor(
request: &CodexResponsesRequest,
) -> Option<(&'static str, String)> {
thread_spawn_conversation_anchor(request).or_else(|| {
request
.client_session_id
.as_deref()
.map(|value| ("session", value.to_owned()))
.or_else(|| {
request
.client_conversation_id
.as_deref()
.map(|value| ("conversation", value.to_owned()))
})
.or_else(|| {
request
.client_thread_id
.as_deref()
.map(|value| ("thread", value.to_owned()))
})
.or_else(|| {
request
.prompt_cache_key()
.map(|value| ("prompt-cache", value.to_owned()))
})
.or_else(|| derive_stable_conversation_key(request).map(|value| ("request", value)))
})
}
/// A `thread_spawn` shares its parent session but is a separate child execution.
/// Its child thread must therefore own the affinity and WebSocket pool identity.
fn thread_spawn_conversation_anchor(
request: &CodexResponsesRequest,
) -> Option<(&'static str, String)> {
if request.subagent_kind().as_deref() != Some(THREAD_SPAWN_SUBAGENT_KIND) {
return None;
}
request
.client_session_id
.client_thread_id
.as_deref()
.map(|value| ("session", value.to_owned()))
.or_else(|| {
request
.client_conversation_id
.as_deref()
.map(|value| ("conversation", value.to_owned()))
.map(str::trim)
.filter(|thread_id| !thread_id.is_empty())
.map(|thread_id| {
(
"subagent-thread",
format!("{THREAD_SPAWN_CONVERSATION_PREFIX}{thread_id}"),
)
})
.or_else(|| {
request
.client_thread_id
.as_deref()
.map(|value| ("thread", value.to_owned()))
})
.or_else(|| {
request
.prompt_cache_key()
.map(|value| ("prompt-cache", value.to_owned()))
})
.or_else(|| derive_stable_conversation_key(request).map(|value| ("request", value)))
}
const LEADING_SYSTEM_REMINDER_OPEN: &str = "<system-reminder>";
+27 -3
View File
@@ -82,6 +82,22 @@ async fn openai_bundle_exposes_one_core_provider_and_drains_worker_contributions
panic!("Desktop release worker must be scheduled");
};
assert_eq!(schedule.interval(), APPCAST_POLL_INTERVAL);
let cli_release_worker = contributions
.iter()
.find_map(|contribution| match contribution {
WorkerContribution::Registration(registration)
if registration.id.owner() == "openai-cli-release" =>
{
Some(registration)
}
WorkerContribution::Registration(_) | WorkerContribution::Disabled { .. } => None,
})
.expect("CLI release worker");
assert_eq!(cli_release_worker.id.kind(), WorkerKind::QuotaCatalogHealth);
let WorkerRunnable::Scheduled { schedule, .. } = &cli_release_worker.runnable else {
panic!("CLI release worker must be scheduled");
};
assert_eq!(schedule.interval(), APPCAST_POLL_INTERVAL);
assert!(contributions.iter().any(|contribution| {
matches!(
contribution,
@@ -101,13 +117,21 @@ async fn openai_admin_provider_exposes_live_wire_profile_and_validated_billing()
.expect("OpenAI bundle");
let admin = bundle.admin_provider();
let profile = admin.dashboard_wire_profile().expect("wire profile");
assert_eq!(profile.version, "0.102.0");
assert_eq!(profile.build, None);
assert_eq!(profile.target.os_type, "Mac OS");
assert_eq!(profile.target.os_version, "15.5.0");
assert_eq!(
profile.user_agent,
"Codex Desktop/0.102.0 (Mac OS 15.5.0; arm64) xterm-256color (Codex Desktop; 1.2026.190)"
);
assert_eq!(
profile
.attributes
.iter()
.find(|attribute| attribute.label == "Codex Core")
.find(|attribute| attribute.label == "客户端标识")
.map(|attribute| attribute.value.as_str()),
Some("0.102.0")
Some("Codex Desktop; 1.2026.190")
);
assert_eq!(
profile.release.as_ref().map(|release| release.status),
@@ -771,7 +795,7 @@ fn valid_config() -> TestOpenAiConfig {
codex_version: "0.102.0".to_owned(),
desktop_version: "1.2026.190".to_owned(),
desktop_build: "19012345678".to_owned(),
os_type: "macOS".to_owned(),
os_type: "Mac OS".to_owned(),
os_version: "15.5.0".to_owned(),
arch: "arm64".to_owned(),
terminal: "xterm-256color".to_owned(),
@@ -13,7 +13,7 @@ fn openai_config_builds_the_audited_wire_profile() {
assert_eq!(
profile.user_agent(),
"Codex Desktop/0.102.0 (macOS 15.5.0; arm64) xterm-256color"
"Codex Desktop/0.102.0 (Mac OS 15.5.0; arm64) xterm-256color (Codex Desktop; 1.2026.190)"
);
assert_eq!(profile.desktop_build, "19012345678");
}
@@ -124,6 +124,10 @@ fn openai_config_defaults_to_the_provider_owned_operating_values() {
"https://auth.openai.com/oauth/token",
)
);
assert_eq!(
config.wire_profile_state().snapshot().user_agent(),
"Codex Desktop/0.146.0 (Mac OS 15.7.1; arm64) unknown (Codex Desktop; 26.727.51351)"
);
}
fn valid_config() -> OpenAiConfig {
@@ -133,7 +137,7 @@ fn valid_config() -> OpenAiConfig {
codex_version: "0.102.0".to_owned(),
desktop_version: "1.2026.190".to_owned(),
desktop_build: "19012345678".to_owned(),
os_type: "macOS".to_owned(),
os_type: "Mac OS".to_owned(),
os_version: "15.5.0".to_owned(),
arch: "arm64".to_owned(),
terminal: "xterm-256color".to_owned(),
@@ -688,6 +688,43 @@ fn credential_expired_failure_marks_unified_account_expired() {
.availability(),
AccountAvailability::Expired
);
assert_eq!(
store.last_error_message("acct_primary").as_deref(),
Some("upstream_credential_expired")
);
}
#[test]
fn credential_expired_failure_keeps_expired_oauth_for_bounded_refresh_recovery() {
let store = Arc::new(MemoryAccountStore::default());
let mut expired_profile = profile("chatgpt-acct_primary");
expired_profile.access_token_expires_at =
Some(chrono::Utc::now() - chrono::Duration::minutes(1));
block_on(store.seed_oauth_credential(ImportCodexOAuthCredential {
account_id: "acct_primary".to_owned(),
name: "acct_primary".to_owned(),
secret: secret("at-primary"),
verified_account: expired_profile,
next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(10)),
enabled: true,
}));
let account = store.account("acct_primary").expect("account");
let selector = selector(&store, Arc::new(TestLeaseCoordinator::default()));
block_on(selector.record_failure(
&account,
CodexAccountFailure::CredentialExpired,
Some("token_expired".to_owned()),
))
.expect("record credential expiry");
assert_eq!(
store
.account("acct_primary")
.expect("account retained for refresh")
.availability(),
AccountAvailability::Ready
);
}
#[test]
@@ -20,7 +20,7 @@ use provider_openai::credential::{
CodexQuotaWindowKind, ImportCodexOAuthCredential,
};
use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState};
use serde_json::json;
use serde_json::{Value, json};
use wiremock::matchers::{header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -424,14 +424,182 @@ async fn passive_rate_limit_headers_replace_stale_secondary_window() {
.await
.expect("read quota")
.expect("quota snapshot");
let raw = store
.quota_json("acct_passive_quota_snapshot")
.expect("stored passive quota");
let monthly = snapshot
.windows()
.iter()
.filter(|window| window.kind() == CodexQuotaWindowKind::Monthly)
.collect::<Vec<_>>();
assert_eq!(snapshot.windows().len(), 1);
assert_eq!(monthly.len(), 1);
assert_eq!(monthly[0].used_percent(), Some(92.0));
assert!(
raw.pointer("/rate_limit/secondary_window").is_none(),
"a normal response without a secondary window must delete stale raw state"
);
}
#[tokio::test]
async fn manual_quota_refresh_drops_empty_secondary_window() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_manual_secondary_placeholder";
create_account(&store, account_id).await;
let account = store.account(account_id).expect("created account");
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.and(header(
"authorization",
format!("Bearer token-{account_id}"),
))
.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,
"limit_window_seconds": 2_592_000
},
"secondary_window": {
"used_percent": 0,
"limit_reached": false
}
}
})))
.expect(1)
.mount(&server)
.await;
let service = quota_service_with_base_url(
&store,
reqwest::Client::builder()
.no_proxy()
.build()
.expect("client"),
server.uri(),
);
let snapshot = service
.refresh_account(account.id())
.await
.expect("manual quota refresh");
assert_eq!(snapshot.windows().len(), 1);
assert_eq!(snapshot.windows()[0].used_percent(), Some(100.0));
}
#[tokio::test]
async fn manual_quota_refresh_drops_null_secondary_window_during_confirmed_exhaustion() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_manual_null_secondary_placeholder";
create_account(&store, account_id).await;
let account = store.account(account_id).expect("created account");
let reset_at = 1_900_000_000;
store
.compare_and_swap_quota(QuotaObservation {
account_id: account.id().clone(),
expected_revision: account.revision(),
quota: Some(OpaqueProviderData::new(
json!({
"rate_limit": {
"allowed": false,
"limit_reached": true,
"primary_window": {
"used_percent": 100,
"reset_at": reset_at,
"limit_window_seconds": 2_592_000
}
}
})
.as_object()
.expect("quota object")
.clone(),
)),
observed_at: Some(SystemTime::now()),
limit_reached: None,
})
.await
.expect("seed confirmed exhaustion");
store
.apply_state_change(AccountStateChange {
message: None,
account_id: account.id().clone(),
expected_revision: account.revision(),
availability: AccountAvailability::QuotaExhausted,
observed_at: SystemTime::now(),
})
.await
.expect("mark account exhausted");
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.and(header(
"authorization",
format!("Bearer token-{account_id}"),
))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"rate_limit": {
"allowed": true,
"limit_reached": false,
"primary_window": {
"used_percent": 86,
"reset_at": reset_at,
"reset_after_seconds": 2_133_000,
"limit_window_seconds": 2_592_000
},
"secondary_window": null
}
})))
.expect(1)
.mount(&server)
.await;
let service = quota_service_with_base_url(
&store,
reqwest::Client::builder()
.no_proxy()
.build()
.expect("client"),
server.uri(),
);
let snapshot = service
.refresh_account(account.id())
.await
.expect("manual quota refresh");
let persisted = store
.get_quotas(&[account.id().clone()])
.await
.expect("read persisted quota")
.into_iter()
.next()
.and_then(|observation| observation.quota)
.map(OpaqueProviderData::into_inner)
.map(Value::Object)
.expect("persisted quota");
assert_eq!(snapshot.windows().len(), 1);
assert_eq!(snapshot.windows()[0].used_percent(), Some(100.0));
assert_eq!(
snapshot.windows()[0]
.reset_at()
.map(|value| value.timestamp()),
Some(reset_at)
);
assert!(snapshot.fact().exhausted());
assert_eq!(
store
.account(account_id)
.expect("account after refresh")
.availability(),
AccountAvailability::QuotaExhausted
);
assert!(
persisted.pointer("/rate_limit/secondary_window").is_none(),
"a null secondary window must not be projected as 100%"
);
}
#[tokio::test]
@@ -719,6 +887,183 @@ async fn periodic_quota_synchronization_does_not_downgrade_exhaustion_on_rate_li
assert_eq!(account.availability(), AccountAvailability::QuotaExhausted);
}
#[tokio::test]
async fn periodic_quota_refresh_drops_empty_secondary_window() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_periodic_secondary_placeholder";
create_account(&store, account_id).await;
let account = store.account(account_id).expect("created account");
store
.apply_state_change(AccountStateChange {
message: None,
account_id: account.id().clone(),
expected_revision: account.revision(),
availability: AccountAvailability::QuotaExhausted,
observed_at: SystemTime::now(),
})
.await
.expect("mark account exhausted");
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": true,
"limit_reached": false,
"primary_window": {
"used_percent": 99,
"reset_at": 1_900_000_000,
"limit_window_seconds": 2_592_000
},
"secondary_window": {}
}
})))
.expect(1)
.mount(&server)
.await;
let service = quota_service_with_base_url(
&store,
reqwest::Client::builder()
.no_proxy()
.build()
.expect("client"),
server.uri(),
);
let summary = service.synchronize().await.expect("periodic quota refresh");
let snapshot = service
.read_account(account.id())
.await
.expect("read quota")
.expect("quota snapshot");
assert_eq!(summary.exhausted, 1);
assert_eq!(snapshot.windows().len(), 1);
assert_eq!(snapshot.windows()[0].used_percent(), Some(100.0));
}
#[tokio::test]
async fn periodic_quota_refresh_drops_null_secondary_window() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_periodic_null_secondary_placeholder";
create_account(&store, account_id).await;
let account = store.account(account_id).expect("created account");
store
.apply_state_change(AccountStateChange {
message: None,
account_id: account.id().clone(),
expected_revision: account.revision(),
availability: AccountAvailability::QuotaExhausted,
observed_at: SystemTime::now(),
})
.await
.expect("mark account exhausted");
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": true,
"limit_reached": false,
"primary_window": {
"used_percent": 86,
"reset_at": 1_900_000_000,
"reset_after_seconds": 2_133_000,
"limit_window_seconds": 2_592_000
},
"secondary_window": null
}
})))
.expect(1)
.mount(&server)
.await;
let service = quota_service_with_base_url(
&store,
reqwest::Client::builder()
.no_proxy()
.build()
.expect("client"),
server.uri(),
);
let summary = service.synchronize().await.expect("periodic quota refresh");
let snapshot = service
.read_account(account.id())
.await
.expect("read quota")
.expect("quota snapshot");
assert_eq!(summary.exhausted, 1);
assert_eq!(snapshot.windows().len(), 1);
assert_eq!(snapshot.windows()[0].used_percent(), Some(100.0));
}
#[tokio::test]
async fn periodic_quota_refresh_keeps_nonempty_secondary_window_during_confirmed_exhaustion() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_periodic_real_secondary";
create_account(&store, account_id).await;
let account = store.account(account_id).expect("created account");
store
.apply_state_change(AccountStateChange {
message: None,
account_id: account.id().clone(),
expected_revision: account.revision(),
availability: AccountAvailability::QuotaExhausted,
observed_at: SystemTime::now(),
})
.await
.expect("mark account exhausted");
let secondary_reset_at = 1_900_604_800;
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": true,
"limit_reached": false,
"primary_window": {
"used_percent": 99,
"reset_at": 1_900_000_000,
"limit_window_seconds": 2_592_000
},
"secondary_window": {
"used_percent": 0,
"reset_at": secondary_reset_at,
"limit_window_seconds": 604_800
}
}
})))
.expect(1)
.mount(&server)
.await;
let service = quota_service_with_base_url(
&store,
reqwest::Client::builder()
.no_proxy()
.build()
.expect("client"),
server.uri(),
);
let summary = service.synchronize().await.expect("periodic quota refresh");
let snapshot = service
.read_account(account.id())
.await
.expect("read quota")
.expect("quota snapshot");
assert_eq!(summary.exhausted, 1);
assert_eq!(snapshot.windows().len(), 2);
assert!(snapshot.windows().iter().any(|window| {
window.window_seconds() == Some(604_800)
&& window.used_percent() == Some(100.0)
&& window
.reset_at()
.is_some_and(|reset_at| reset_at.timestamp() == secondary_reset_at)
}));
}
#[tokio::test]
async fn periodic_quota_synchronization_respects_retry_after_on_service_unavailable() {
let store = Arc::new(MemoryAccountStore::default());
@@ -211,3 +211,47 @@ async fn expired_old_reset_with_fresh_allowed_snapshot_recovers_ready() {
"old reset elapsed + fresh allowed snapshot must recover QuotaExhausted"
);
}
#[tokio::test]
async fn deactivated_workspace_quota_response_persists_the_upstream_reason() {
let store = Arc::new(MemoryAccountStore::default());
let account_id = "acct_deactivated_workspace";
create_account(&store, account_id).await;
let account = store.account(account_id).expect("created account");
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/codex/usage"))
.and(header(
"authorization",
format!("Bearer token-{account_id}"),
))
.respond_with(ResponseTemplate::new(402).set_body_json(json!({
"detail": {
"code": "deactivated_workspace",
"message": "workspace disabled"
}
})))
.mount(&server)
.await;
let service = quota_service_with_base_url(
&store,
reqwest::Client::builder()
.no_proxy()
.build()
.expect("client"),
server.uri(),
);
assert!(service.refresh_account(account.id()).await.is_err());
assert_eq!(
store
.account(account_id)
.expect("account after deactivation")
.availability(),
AccountAvailability::Banned
);
assert_eq!(
store.last_error_message(account_id).as_deref(),
Some("deactivated_workspace")
);
}
@@ -19,6 +19,8 @@ use secrecy::{ExposeSecret, SecretString};
use crate::support::{MemoryAccountStore, profile, runtime_policy, secret};
const OAUTH_BACKOFF_ATTEMPTS: u32 = 5;
struct Refresher {
outcomes: Mutex<VecDeque<Result<TokenPair, RefreshFailure>>>,
seen: Mutex<Vec<String>>,
@@ -553,6 +555,10 @@ async fn upstream_ban_marks_unified_account_banned() {
.availability(),
AccountAvailability::Banned
);
assert_eq!(
store.last_error_message("acct_refresh").as_deref(),
Some("account_banned")
);
}
#[tokio::test]
@@ -597,7 +603,7 @@ async fn transient_refresh_defers_without_invalidating_account() {
}
#[tokio::test]
async fn expired_access_token_refreshes_even_when_next_refresh_is_deferred() {
async fn expired_access_token_respects_persisted_refresh_backoff() {
let store = Arc::new(MemoryAccountStore::default());
seed_due_account(&store).await;
let original = store.account("acct_refresh").expect("seeded account");
@@ -615,16 +621,6 @@ async fn expired_access_token_refreshes_even_when_next_refresh_is_deferred() {
})
.await
.expect("seed expired access token with deferred refresh");
let deferred = store.account("acct_refresh").expect("deferred account");
store
.repository()
.apply_state(
&deferred,
AccountAvailability::QuotaExhausted,
SystemTime::now(),
)
.await
.expect("mark account quota exhausted");
let refresher = Arc::new(Refresher::new(Ok(success_tokens())));
let service = CodexCredentialRefreshService::new(
store.repository(),
@@ -640,31 +636,19 @@ async fn expired_access_token_refreshes_even_when_next_refresh_is_deferred() {
let outcomes = service.refresh_due().await.expect("refresh due");
assert!(matches!(
outcomes.as_slice(),
[CodexCredentialRefreshOutcome::Refreshed {
account_id,
credential_revision: 3,
}] if account_id == "acct_refresh"
));
assert_eq!(
refresher.seen.lock().expect("seen tokens lock").as_slice(),
["rt-old-access"]
);
let refreshed = store.account("acct_refresh").expect("refreshed account");
assert_eq!(
refreshed.availability(),
AccountAvailability::QuotaExhausted
);
assert!(outcomes.is_empty());
assert!(refresher.seen.lock().expect("seen tokens lock").is_empty());
assert!(
refreshed
.access_token_expires_at()
.is_some_and(|expires_at| expires_at > SystemTime::now())
store
.account("acct_refresh")
.expect("deferred account")
.next_refresh_at()
.is_some_and(|next_refresh_at| next_refresh_at > SystemTime::now())
);
}
#[tokio::test]
async fn expired_access_token_refreshes_during_active_cooldown() {
async fn expired_access_token_refreshes_when_persisted_refresh_is_due() {
let store = Arc::new(MemoryAccountStore::default());
seed_due_account(&store).await;
let original = store.account("acct_refresh").expect("seeded account");
@@ -678,7 +662,7 @@ async fn expired_access_token_refreshes_during_active_cooldown() {
expected_credential_revision: original.revision().get(),
secret: secret("old-access"),
verified_account: expired_profile,
next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::hours(12)),
next_refresh_at: Some(chrono::Utc::now() - chrono::Duration::seconds(1)),
})
.await
.expect("seed expired access token with deferred refresh");
@@ -825,6 +809,15 @@ async fn seed_due_account(store: &MemoryAccountStore) {
.await;
}
async fn force_due_account(store: &Arc<MemoryAccountStore>) {
let account = store.account("acct_refresh").expect("account to reset");
store
.repository()
.defer_refresh(&account, SystemTime::now() - Duration::from_secs(1))
.await
.expect("reset next_refresh_at to the past");
}
#[tokio::test]
async fn refresh_backoff_grows_exponentially_across_attempts() {
let store = Arc::new(MemoryAccountStore::default());
@@ -891,6 +884,175 @@ async fn refresh_backoff_grows_exponentially_across_attempts() {
);
}
#[tokio::test]
async fn exhausted_oauth_backoff_continues_after_access_token_expiry() {
let store = Arc::new(MemoryAccountStore::default());
let mut expired_profile = profile("chatgpt-acct_refresh");
expired_profile.access_token_expires_at =
Some(chrono::Utc::now() - chrono::Duration::minutes(1));
store
.seed_oauth_credential(ImportCodexOAuthCredential {
account_id: "acct_refresh".to_owned(),
name: "refresh".to_owned(),
secret: secret("old-access"),
verified_account: expired_profile,
next_refresh_at: Some(chrono::Utc::now() - chrono::Duration::seconds(1)),
enabled: true,
})
.await;
let credential_state = Arc::new(CountingCredentialState::default());
let state_port: Arc<dyn ProviderCredentialStatePort> = credential_state.clone();
let refresher = Arc::new(Refresher::scripted([
Err(RefreshFailure::Transport),
Err(RefreshFailure::Transport),
Err(RefreshFailure::Transport),
Err(RefreshFailure::Transport),
Err(RefreshFailure::Transport),
Err(RefreshFailure::Transport),
]));
let service = CodexCredentialRefreshService::new(
store.repository(),
refresher.clone(),
Arc::new(VerifiedIdentity),
Arc::new(RefreshLeases {
available: true,
requests: Mutex::new(Vec::new()),
}),
state_port,
runtime_policy(),
);
for _ in 0..OAUTH_BACKOFF_ATTEMPTS {
service.refresh_due().await.expect("backoff attempt");
force_due_account(&store).await;
}
let observed_at = SystemTime::now();
service.refresh_due().await.expect("recovery attempt");
let recovery_delay = store
.account("acct_refresh")
.expect("account after recovery defer")
.next_refresh_at()
.expect("recovery scheduled")
.duration_since(observed_at)
.expect("recovery is in the future");
assert!(
(Duration::from_secs(8 * 60)..=Duration::from_secs(12 * 60 + 1)).contains(&recovery_delay),
"recovery delay should be about ten minutes, got {recovery_delay:?}"
);
assert_eq!(
credential_state.count("acct_refresh"),
OAUTH_BACKOFF_ATTEMPTS + 1
);
assert_eq!(refresher.seen.lock().expect("seen tokens lock").len(), 6);
}
#[tokio::test]
async fn oauth_backoff_never_passes_an_unexpired_access_token_expiry() {
let store = Arc::new(MemoryAccountStore::default());
let mut account_profile = profile("chatgpt-acct_refresh");
account_profile.access_token_expires_at =
Some(chrono::Utc::now() + chrono::Duration::seconds(30));
let access_token_expires_at: SystemTime = account_profile
.access_token_expires_at
.expect("access expiry")
.into();
store
.seed_oauth_credential(ImportCodexOAuthCredential {
account_id: "acct_refresh".to_owned(),
name: "refresh".to_owned(),
secret: secret("old-access"),
verified_account: account_profile,
next_refresh_at: Some(chrono::Utc::now() - chrono::Duration::seconds(1)),
enabled: true,
})
.await;
let credential_state = Arc::new(CountingCredentialState::default());
let state_port: Arc<dyn ProviderCredentialStatePort> = credential_state.clone();
let service = CodexCredentialRefreshService::new(
store.repository(),
Arc::new(Refresher::scripted([
Err(RefreshFailure::Transport),
Err(RefreshFailure::Transport),
Err(RefreshFailure::Transport),
Err(RefreshFailure::Transport),
Err(RefreshFailure::Transport),
Err(RefreshFailure::Transport),
])),
Arc::new(VerifiedIdentity),
Arc::new(RefreshLeases {
available: true,
requests: Mutex::new(Vec::new()),
}),
state_port,
runtime_policy(),
);
for attempt in 0..=OAUTH_BACKOFF_ATTEMPTS {
service.refresh_due().await.expect("backoff attempt");
if attempt < OAUTH_BACKOFF_ATTEMPTS {
force_due_account(&store).await;
}
}
assert_eq!(
store
.account("acct_refresh")
.expect("account after capped retry")
.next_refresh_at(),
Some(access_token_expires_at)
);
assert_eq!(
credential_state.count("acct_refresh"),
OAUTH_BACKOFF_ATTEMPTS + 1
);
}
#[tokio::test]
async fn recovery_window_exhaustion_marks_account_expired_without_token_exchange() {
let store = Arc::new(MemoryAccountStore::default());
let mut expired_profile = profile("chatgpt-acct_refresh");
expired_profile.access_token_expires_at =
Some(chrono::Utc::now() - chrono::Duration::hours(25));
store
.seed_oauth_credential(ImportCodexOAuthCredential {
account_id: "acct_refresh".to_owned(),
name: "refresh".to_owned(),
secret: secret("old-access"),
verified_account: expired_profile,
next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::hours(12)),
enabled: true,
})
.await;
let refresher = Arc::new(Refresher::new(Ok(success_tokens())));
let service = CodexCredentialRefreshService::new(
store.repository(),
refresher.clone(),
Arc::new(VerifiedIdentity),
Arc::new(RefreshLeases {
available: true,
requests: Mutex::new(Vec::new()),
}),
Arc::new(CountingCredentialState::default()),
runtime_policy(),
);
let outcomes = service.refresh_due().await.expect("refresh due");
assert!(matches!(
outcomes.as_slice(),
[CodexCredentialRefreshOutcome::Invalidated { account_id }] if account_id == "acct_refresh"
));
assert!(refresher.seen.lock().expect("seen tokens lock").is_empty());
assert_eq!(
store
.account("acct_refresh")
.expect("expired account")
.availability(),
AccountAvailability::Expired
);
}
#[tokio::test]
async fn successful_refresh_clears_backoff_counter() {
let store = Arc::new(MemoryAccountStore::default());
@@ -214,6 +214,27 @@ fn generate_operation() -> Operation {
))
}
fn generate_with_session_context(
session_id: &str,
thread_id: Option<&str>,
turn_metadata: Option<&str>,
) -> GenerateRequest {
let mut body = Map::from_iter([
("model".to_owned(), json!("gpt-5.4")),
("input".to_owned(), json!("hello")),
("session_id".to_owned(), json!(session_id)),
]);
if let Some(thread_id) = thread_id {
body.insert("thread_id".to_owned(), json!(thread_id));
}
if let Some(turn_metadata) = turn_metadata {
body.insert("turnMetadata".to_owned(), json!(turn_metadata));
}
GenerateRequest::from_protocol_payload(
ProtocolPayload::json_object("openai", body).expect("OpenAI payload"),
)
}
fn http_generate_operation() -> Operation {
let payload = ProtocolPayload::json_object(
"openai",
@@ -1173,6 +1194,58 @@ async fn explicit_session_id_should_override_turn_specific_prompt_cache_keys_for
assert_eq!(keys[0], keys[1]);
}
#[tokio::test]
async fn thread_spawn_children_should_have_independent_session_affinity_keys() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_thread_spawn_affinity").await;
let affinity = Arc::new(MemorySessionAffinity::default());
let provider = provider_with_affinity(&store, Arc::clone(&affinity));
let thread_spawn = r#"{"subagent_kind":"thread_spawn"}"#;
for (request_id, thread_id, turn_metadata) in [
(
"req_thread_spawn_first",
Some("child-one"),
Some(thread_spawn),
),
(
"req_thread_spawn_second",
Some("child-two"),
Some(thread_spawn),
),
(
"req_thread_spawn_repeat",
Some("child-one"),
Some(thread_spawn),
),
("req_thread_spawn_fallback", None, Some(thread_spawn)),
("req_thread_spawn_parent", None, None),
] {
let stream = provider
.execute(
planned_request(
"openai",
Operation::Generate(generate_with_session_context(
"parent-session",
thread_id,
turn_metadata,
)),
),
context(request_id, CancellationToken::new()),
)
.await
.expect("prepare provider stream");
drop(stream);
}
let keys = affinity.lookup_keys();
assert_eq!(keys.len(), 5);
assert_ne!(keys[0], keys[1]);
assert_eq!(keys[0], keys[2]);
assert_ne!(keys[3], keys[4]);
assert_ne!(keys[0], keys[3]);
}
#[tokio::test]
async fn invalid_local_conversation_id_should_fall_back_to_an_opaque_affinity_key() {
let store = Arc::new(MemoryAccountStore::default());
@@ -1364,7 +1437,8 @@ async fn official_usage_limit_failure_projects_100_percent_and_holds_until_reset
"rate_limit": {
"allowed": true,
"limit_reached": false,
"primary_window": {"used_percent": 99, "reset_at": reset_at}
"primary_window": {"used_percent": 99, "reset_at": reset_at},
"secondary_window": {"used_percent": 0}
}
})
.as_object()
@@ -1390,7 +1464,8 @@ async fn official_usage_limit_failure_projects_100_percent_and_holds_until_reset
"rate_limit": {
"allowed": true,
"limit_reached": false,
"primary_window": {"used_percent": 99, "reset_at": reset_at}
"primary_window": {"used_percent": 99, "reset_at": reset_at},
"secondary_window": {"used_percent": 0}
}
})),
)
@@ -1439,16 +1514,22 @@ async fn official_usage_limit_failure_projects_100_percent_and_holds_until_reset
.next()
.expect("confirmed quota observation");
let projected_observed_at = projected.observed_at.expect("projection observed at");
let projected = projected
.quota
.map(OpaqueProviderData::into_inner)
.expect("confirmed quota projection");
let projected = Value::Object(
projected
.quota
.map(OpaqueProviderData::into_inner)
.expect("confirmed quota projection"),
);
assert_eq!(
Value::Object(projected)
projected
.pointer("/rate_limit/primary_window/used_percent")
.and_then(Value::as_u64),
Some(100)
);
assert!(
projected.pointer("/rate_limit/secondary_window").is_none(),
"request failure must not manufacture a secondary quota"
);
tokio::time::sleep(Duration::from_millis(200)).await;
let requests = server.received_requests().await.expect("received requests");
assert_eq!(
@@ -1499,6 +1580,10 @@ async fn official_usage_limit_failure_projects_100_percent_and_holds_until_reset
.and_then(Value::as_i64),
Some(reset_at)
);
assert!(
quota.pointer("/rate_limit/secondary_window").is_none(),
"background quota refresh must retain the same window projection"
);
let requests = server.received_requests().await.expect("received requests");
assert_eq!(
requests
@@ -38,6 +38,7 @@ struct StoredAccount {
credential: gateway_core::engine::credential::PlaintextCredential,
quota: Option<QuotaObservation>,
state_observed_at: Option<SystemTime>,
last_error_message: Option<String>,
}
#[derive(Default)]
@@ -70,6 +71,15 @@ impl MemoryAccountStore {
.map(|stored| stored.account.clone())
}
pub(crate) fn last_error_message(&self, id: &str) -> Option<String> {
let id = ProviderAccountId::new(id).ok()?;
self.accounts
.lock()
.expect("account store lock")
.get(&id)
.and_then(|stored| stored.last_error_message.clone())
}
pub(crate) async fn advance_credential_revision(&self, id: &str) -> CredentialRevision {
let account_id = ProviderAccountId::new(id).expect("test account id");
let current = self
@@ -148,6 +158,7 @@ impl ProviderAccountStore for MemoryAccountStore {
credential: input.credential,
quota: None,
state_observed_at: None,
last_error_message: None,
},
);
Ok(())
@@ -321,6 +332,11 @@ impl ProviderAccountStore for MemoryAccountStore {
},
);
stored.state_observed_at = Some(change.observed_at);
if change.availability == AccountAvailability::Ready {
stored.last_error_message = None;
} else if let Some(message) = change.message {
stored.last_error_message = Some(message);
}
Ok(())
}
@@ -219,6 +219,10 @@ async fn fetch_should_send_official_catalog_headers_and_query() {
.and(header("authorization", "Bearer oauth-access"))
.and(header("chatgpt-account-id", "acct_123"))
.and(header("originator", "codex_cli_rs"))
.and(header(
"user-agent",
"codex_cli_rs/0.144.0 (linux 6.8; x86_64) xterm (codex_cli_rs; 1.0.0)",
))
.and(header("accept", "application/json"))
.and(header("x-codex-installation-id", "installation-123"))
.respond_with(
@@ -60,13 +60,13 @@ impl CodexCliReleaseTransport for CliReleaseTransport {
}
#[test]
fn wire_profile_should_generate_codex_core_user_agent() {
fn wire_profile_should_generate_bundled_core_app_server_user_agent() {
let profile = CodexWireProfile {
originator: "Codex Desktop".to_owned(),
codex_version: "0.144.2".to_owned(),
desktop_version: "26.707.72221".to_owned(),
desktop_build: "72221".to_owned(),
os_type: "macOS".to_owned(),
os_type: "Mac OS".to_owned(),
os_version: "15.7.1".to_owned(),
arch: "arm64".to_owned(),
terminal: "unknown".to_owned(),
@@ -78,7 +78,7 @@ fn wire_profile_should_generate_codex_core_user_agent() {
assert_eq!(
profile.user_agent(),
"Codex Desktop/0.144.2 (macOS 15.7.1; arm64) unknown"
"Codex Desktop/0.144.2 (Mac OS 15.7.1; arm64) unknown (Codex Desktop; 26.707.72221)"
);
}
@@ -137,6 +137,7 @@ fn desktop_appcast_should_reject_invalid_version_and_build() {
#[tokio::test]
async fn wire_profile_release_update_should_change_only_desktop_identity() {
let original = wire_profile();
let original_user_agent = original.user_agent();
let state = CodexWireProfileState::new(original.clone());
let service = CodexDesktopReleaseService::new(
state.clone(),
@@ -155,6 +156,11 @@ async fn wire_profile_release_update_should_change_only_desktop_identity() {
assert_eq!(updated.arch, original.arch);
assert_eq!(updated.terminal, original.terminal);
assert_eq!(updated.verified_at, original.verified_at);
assert_ne!(updated.user_agent(), original_user_agent);
assert_eq!(
updated.user_agent(),
"Codex Desktop/0.144.2 (Mac OS 15.7.1; arm64) unknown (Codex Desktop; 26.720.1)"
);
let status = service.status().snapshot();
assert_eq!(status.latest, Some(release("26.720.1", "72001")));
assert!(status.checked_at.is_some());
@@ -164,6 +170,7 @@ async fn wire_profile_release_update_should_change_only_desktop_identity() {
#[tokio::test]
async fn cli_release_update_should_change_only_codex_core_identity() {
let original = wire_profile();
let original_user_agent = original.user_agent();
let state = CodexWireProfileState::new(original.clone());
let service = CodexCliReleaseService::new(
state.clone(),
@@ -182,6 +189,11 @@ async fn cli_release_update_should_change_only_codex_core_identity() {
assert_eq!(updated.arch, original.arch);
assert_eq!(updated.terminal, original.terminal);
assert_eq!(updated.verified_at, original.verified_at);
assert_ne!(updated.user_agent(), original_user_agent);
assert_eq!(
updated.user_agent(),
"Codex Desktop/0.145.0 (Mac OS 15.7.1; arm64) unknown (Codex Desktop; 26.707.72221)"
);
}
#[tokio::test]
@@ -219,7 +231,7 @@ fn wire_profile() -> CodexWireProfile {
codex_version: "0.144.2".to_owned(),
desktop_version: "26.707.72221".to_owned(),
desktop_build: "72221".to_owned(),
os_type: "macOS".to_owned(),
os_type: "Mac OS".to_owned(),
os_version: "15.7.1".to_owned(),
arch: "arm64".to_owned(),
terminal: "unknown".to_owned(),
@@ -163,7 +163,12 @@ async fn fetch_should_use_wham_usage_headers_only() {
"unexpected {name} header"
);
}
assert!(headers.get("user-agent").is_some());
assert_eq!(
headers
.get("user-agent")
.and_then(|value| value.to_str().ok()),
Some("codex_cli_rs/0.144.0 (linux 6.8; x86_64) xterm (codex_cli_rs; 1.0.0)")
);
let quota_header_names = headers
.keys()
.map(|name| name.as_str())
@@ -221,7 +221,7 @@ impl GrokOAuthClient {
/// # Errors
///
/// 返回已分类的 OAuth 错误。refresh token 可能已轮换,Ambiguous 类
/// transport 失败不得自动重试。
/// transport 失败不在本次 exchange 内重试,后续交由 refresh scheduler 退避协调。
pub async fn refresh(
&self,
discovery: &DiscoveryDocument,
@@ -37,7 +37,7 @@ pub enum OAuthErrorCode {
pub enum FailureClass {
/// 瞬时失败,允许后续独立协调的重试。
Transient,
/// 发送状态不确定,一次性凭据不得重放。
/// 发送状态不确定,不应在同一 exchange 内重放一次性凭据。
Ambiguous,
/// 当前 revision 的凭据被永久拒绝。
CredentialPermanent,
@@ -12,10 +12,10 @@ use gateway_core::engine::credential::{
AccountAvailability, CredentialRevision, LoadedCredential, ProviderAccountId,
};
use gateway_core::provider_ports::{
ProviderCooldown, ProviderCooldownPort, ProviderCredentialStatePort, ProviderLeaseAcquisition,
ProviderLeasePort, ProviderLeaseRequest, ProviderRefreshCapacityRequest,
ProviderRefreshLeaseRequest, ProviderRefreshPolicy, ProviderRuntimePolicyPort,
ProviderStoreError, provider_refresh_backoff_at,
ProviderCredentialStatePort, ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest,
ProviderRefreshCapacityRequest, ProviderRefreshLeaseRequest, ProviderRefreshPolicy,
ProviderRuntimePolicyPort, ProviderStoreError, provider_refresh_backoff_at,
provider_refresh_retry_at,
};
use super::catalog::GrokCredentialCatalogService;
@@ -38,12 +38,49 @@ const MAX_SECRET_BYTES: usize = 64 * 1_024;
const DISCOVERY_NEGATIVE_CACHE_TTL: Duration = Duration::from_secs(5);
/// 连续失败计数窗口;静默满窗后计数过期归零。
const REFRESH_BACKOFF_WINDOW: Duration = Duration::from_secs(30 * 60);
const REFRESH_AMBIGUITY_COOLDOWN: Duration = Duration::from_secs(24 * 60 * 60);
/// 窗口内先执行五次指数退避,再进入固定恢复周期。
const REFRESH_BACKOFF_MAX_ATTEMPTS: u32 = 5;
/// 耗尽指数退避后的 OAuth 恢复周期。
const REFRESH_RECOVERY_DELAY: Duration = Duration::from_secs(10 * 60);
/// 过期 AT 仍允许 RT 恢复的最长窗口。
const REFRESH_RECOVERY_WINDOW: Duration = Duration::from_secs(24 * 60 * 60);
fn refresh_recovery_deadline(access_token_expires_at: Option<SystemTime>) -> Option<SystemTime> {
access_token_expires_at.and_then(|expires_at| expires_at.checked_add(REFRESH_RECOVERY_WINDOW))
}
fn refresh_recovery_window_exhausted(
access_token_expires_at: Option<SystemTime>,
now: SystemTime,
) -> bool {
refresh_recovery_deadline(access_token_expires_at).is_some_and(|deadline| deadline <= now)
}
fn bounded_refresh_retry_at(
retry_at: SystemTime,
access_token_expires_at: Option<SystemTime>,
observed_at: SystemTime,
) -> SystemTime {
let mut bounded = retry_at;
if let Some(expires_at) = access_token_expires_at
&& expires_at > observed_at
&& bounded > expires_at
{
bounded = expires_at;
}
if let Some(deadline) = refresh_recovery_deadline(access_token_expires_at)
&& bounded > deadline
{
bounded = deadline;
}
bounded
}
/// 一个到期且已按 revision 读取明文 RT 的 xAI account。
pub struct DueGrokCredential {
account_id: ProviderAccountId,
credential_revision: CredentialRevision,
access_token_expires_at: Option<SystemTime>,
refresh_token: SecretValue,
id_token: Option<SecretValue>,
scope: String,
@@ -77,6 +114,7 @@ impl fmt::Debug for DueGrokCredential {
.debug_struct("DueGrokCredential")
.field("account_id", &self.account_id)
.field("credential_revision", &self.credential_revision)
.field("access_token_expires_at", &self.access_token_expires_at)
.field("refresh_token", &"[REDACTED]")
.field("id_token", &self.id_token.as_ref().map(|_| "[REDACTED]"))
.field("scope", &"[REDACTED]")
@@ -277,7 +315,6 @@ pub struct GrokCredentialRefreshService {
refresher: Arc<dyn GrokCredentialRefresher>,
catalog: Arc<GrokCredentialCatalogService>,
leases: Arc<dyn ProviderLeasePort>,
cooldowns: Arc<dyn ProviderCooldownPort>,
credential_state: Arc<dyn ProviderCredentialStatePort>,
runtime_policy: Arc<dyn ProviderRuntimePolicyPort>,
}
@@ -288,7 +325,6 @@ impl GrokCredentialRefreshService {
refresher: Arc<dyn GrokCredentialRefresher>,
catalog: Arc<GrokCredentialCatalogService>,
leases: Arc<dyn ProviderLeasePort>,
cooldowns: Arc<dyn ProviderCooldownPort>,
credential_state: Arc<dyn ProviderCredentialStatePort>,
runtime_policy: Arc<dyn ProviderRuntimePolicyPort>,
) -> Self {
@@ -297,7 +333,6 @@ impl GrokCredentialRefreshService {
refresher,
catalog,
leases,
cooldowns,
credential_state,
runtime_policy,
}
@@ -331,6 +366,7 @@ impl GrokCredentialRefreshService {
let credential = DueGrokCredential {
account_id: account_id.clone(),
credential_revision,
access_token_expires_at: loaded.account.access_token_expires_at(),
refresh_token: loaded.refresh_token,
id_token: loaded.id_token,
scope: loaded.scope,
@@ -447,12 +483,7 @@ impl GrokCredentialRefreshService {
.into_iter()
.map(|account_id| GrokCredentialRefreshOutcome::Failed { account_id })
.collect::<Vec<_>>();
let mut credentials = Vec::with_capacity(batch.credentials.len());
for credential in batch.credentials {
if !self.runtime_cooldown_active(&credential).await {
credentials.push(credential);
}
}
let credentials = batch.credentials;
if credentials.is_empty() {
return Ok(outcomes);
}
@@ -474,7 +505,14 @@ impl GrokCredentialRefreshService {
for (account_id, result) in refreshed {
match result {
Ok(outcome) => outcomes.push(outcome),
Err(_) => outcomes.push(GrokCredentialRefreshOutcome::Failed { account_id }),
Err(error) => {
tracing::warn!(
account_id = %account_id,
error = %error,
"xAI OAuth refresh attempt failed"
);
outcomes.push(GrokCredentialRefreshOutcome::Failed { account_id });
}
}
}
Ok(outcomes)
@@ -486,9 +524,10 @@ impl GrokCredentialRefreshService {
policy: ProviderRefreshPolicy,
) -> Result<GrokCredentialRefreshOutcome, GrokCredentialRefreshError> {
let account_id = credential.account_id.clone();
let observed_at = SystemTime::now();
if credential
.refresh_token_expires_at
.is_some_and(|expires_at| expires_at <= Utc::now())
.is_some_and(|expires_at| expires_at <= DateTime::<Utc>::from(observed_at))
{
return self
.persist_terminal_failure(
@@ -498,6 +537,15 @@ impl GrokCredentialRefreshService {
)
.await;
}
if refresh_recovery_window_exhausted(credential.access_token_expires_at, observed_at) {
return self
.persist_terminal_failure(
credential,
GrokCredentialAvailability::Expired,
"refresh_recovery_window_exhausted",
)
.await;
}
let capacity = self
.leases
.try_acquire(ProviderLeaseRequest::RefreshCapacity(
@@ -545,9 +593,14 @@ impl GrokCredentialRefreshService {
.await
}
Err(GrokRefreshFailure::Ambiguous) => {
self.record_runtime_cooldown(&credential, REFRESH_AMBIGUITY_COOLDOWN)
.await;
Ok(GrokCredentialRefreshOutcome::Ambiguous { account_id })
if self
.persist_backoff(&credential, "refresh-ambiguous")
.await?
{
Ok(GrokCredentialRefreshOutcome::Ambiguous { account_id })
} else {
Ok(GrokCredentialRefreshOutcome::Stale { account_id })
}
}
Err(GrokRefreshFailure::Transient) => {
if self
@@ -644,35 +697,19 @@ impl GrokCredentialRefreshService {
.credential_state
.clear_refresh_backoff(&account_id)
.await;
tracing::info!(
account_id = %account_id,
credential_revision = record.credential_revision.get(),
access_token_expires_at = %access_expires_at,
next_refresh_at = %next_refresh_at,
"xAI OAuth refresh succeeded"
);
Ok(GrokCredentialRefreshOutcome::Refreshed {
account_id,
credential_revision: record.credential_revision,
})
}
async fn runtime_cooldown_active(&self, credential: &DueGrokCredential) -> bool {
// runtime cooldown 是账号/model 级生命周期,与 credential revision 无关:
// 轮换不清除、不因 revision 变化判失效(与 selector 的读取语义一致)。
let Ok(Some(cooldown)) = self.cooldowns.read(credential.account_id()).await else {
return false;
};
cooldown.until() > SystemTime::now()
}
async fn record_runtime_cooldown(&self, credential: &DueGrokCredential, duration: Duration) {
let cooldown_until = SystemTime::now()
.checked_add(duration)
.unwrap_or_else(SystemTime::now);
let _ = self
.cooldowns
.put_if_later(ProviderCooldown::new(
credential.account_id.clone(),
credential.credential_revision,
cooldown_until,
))
.await;
}
async fn persist_terminal_failure(
&self,
credential: DueGrokCredential,
@@ -702,7 +739,19 @@ impl GrokCredentialRefreshService {
})
.await
{
Ok(()) => Ok(GrokCredentialRefreshOutcome::Invalidated { account_id }),
Ok(()) => {
tracing::warn!(
account_id = %account_id,
?availability,
reason,
access_token_expires_at = ?credential.access_token_expires_at
.map(DateTime::<Utc>::from),
recovery_deadline = ?refresh_recovery_deadline(credential.access_token_expires_at)
.map(DateTime::<Utc>::from),
"xAI OAuth refresh marked account terminal"
);
Ok(GrokCredentialRefreshOutcome::Invalidated { account_id })
}
Err(error) if stale_repository_error(&error) => {
Ok(GrokCredentialRefreshOutcome::Stale { account_id })
}
@@ -720,12 +769,19 @@ impl GrokCredentialRefreshService {
.record_refresh_backoff(&credential.account_id, REFRESH_BACKOFF_WINDOW)
.await
.unwrap_or(1);
let retry_at = provider_refresh_backoff_at(
&credential.account_id,
SystemTime::now(),
attempt,
reason,
)?;
let observed_at = SystemTime::now();
let retry_at = if attempt <= REFRESH_BACKOFF_MAX_ATTEMPTS {
provider_refresh_backoff_at(&credential.account_id, observed_at, attempt, reason)?
} else {
provider_refresh_retry_at(
&credential.account_id,
observed_at,
REFRESH_RECOVERY_DELAY,
reason,
)?
};
let retry_at =
bounded_refresh_retry_at(retry_at, credential.access_token_expires_at, observed_at);
match self
.repository
.defer_refresh(
@@ -735,7 +791,20 @@ impl GrokCredentialRefreshService {
)
.await
{
Ok(()) => Ok(true),
Ok(()) => {
tracing::warn!(
account_id = %credential.account_id,
attempt,
reason,
retry_at = %DateTime::<Utc>::from(retry_at),
access_token_expires_at = ?credential.access_token_expires_at
.map(DateTime::<Utc>::from),
recovery_deadline = ?refresh_recovery_deadline(credential.access_token_expires_at)
.map(DateTime::<Utc>::from),
"xAI OAuth refresh deferred"
);
Ok(true)
}
Err(error) if stale_repository_error(&error) => Ok(false),
Err(error) => Err(error.into()),
}
@@ -814,6 +883,7 @@ impl GrokCredentialRepository {
due.push(DueGrokCredential {
account_id: account.id().clone(),
credential_revision: account.revision(),
access_token_expires_at: account.access_token_expires_at(),
refresh_token: loaded.refresh_token,
id_token: loaded.id_token,
scope: loaded.scope,
@@ -844,7 +914,8 @@ fn account_due(
| AccountAvailability::Banned
| AccountAvailability::Invalid
)
&& account.next_refresh_at().is_some_and(|next| next <= now)
&& (refresh_recovery_window_exhausted(account.access_token_expires_at(), now)
|| account.next_refresh_at().is_some_and(|next| next <= now))
}
fn refreshed_tokens(tokens: RefreshedTokenSet) -> Result<GrokRefreshTokens, GrokRefreshFailure> {
@@ -6,10 +6,10 @@ use std::time::SystemTime;
use chrono::{DateTime, Utc};
use gateway_core::engine::credential::{
AccountStateChange, CredentialCasOutcome, CredentialCasUpdate, CredentialRevision,
LoadedCredential, NewProviderAccount, OpaqueProviderData, PlaintextCredential, ProviderAccount,
ProviderAccountId, ProviderAccountStore, ProviderAccountUpdate, QuotaObservation,
QuotaWriteOutcome,
AccountAvailability, AccountStateChange, CredentialCasOutcome, CredentialCasUpdate,
CredentialRevision, LoadedCredential, NewProviderAccount, OpaqueProviderData,
PlaintextCredential, ProviderAccount, ProviderAccountId, ProviderAccountStore,
ProviderAccountUpdate, QuotaObservation, QuotaWriteOutcome,
};
use gateway_core::error::StoreErrorKind;
use gateway_core::provider_ports::ProviderRefreshPolicy;
@@ -428,15 +428,21 @@ impl GrokCredentialRepository {
&self,
input: &UpdateGrokCredentialState,
) -> Result<(), GrokCredentialRepositoryError> {
validate_reason(input.availability_reason.as_deref())?;
let availability = AccountAvailability::from(input.availability);
let availability_reason = input
.availability_reason
.clone()
.filter(|value| !value.trim().is_empty())
.or_else(|| availability.fallback_error_reason().map(str::to_owned));
validate_reason(availability_reason.as_deref())?;
self.ensure_xai_account(&input.account_id).await?;
self.store
.apply_state_change(AccountStateChange {
account_id: input.account_id.clone(),
expected_revision: input.expected_revision,
availability: input.availability.into(),
availability,
observed_at: to_system_time(input.observed_at),
message: input.availability_reason.clone(),
message: availability_reason,
})
.await
.map_err(map_store_error)
-1
View File
@@ -171,7 +171,6 @@ pub async fn initialize(
refresher,
Arc::clone(&catalog),
Arc::clone(&leases),
cooldowns.clone(),
ports.credential_state(),
Arc::clone(&runtime_policy),
));
+11 -5
View File
@@ -2040,11 +2040,10 @@ impl ScheduledTask for XaiOAuthRefreshTask {
if context.cancellation().is_cancelled() {
return Ok(());
}
let outcomes = self
.service
.refresh_due()
.await
.map_err(|_| WorkerTaskError::safe("xAI OAuth refresh failed"))?;
let outcomes = self.service.refresh_due().await.map_err(|error| {
tracing::error!(error = %error, "xAI OAuth refresh cycle failed");
WorkerTaskError::safe("xAI OAuth refresh failed")
})?;
let failures = outcomes
.iter()
.filter(|outcome| {
@@ -2056,6 +2055,13 @@ impl ScheduledTask for XaiOAuthRefreshTask {
)
})
.count();
if !outcomes.is_empty() {
tracing::info!(
accounts = outcomes.len(),
failures,
"xAI OAuth refresh cycle completed"
);
}
if failures > 0 {
tracing::warn!(failures, "xAI OAuth refresh cycle contained failures");
}
@@ -233,6 +233,29 @@ async fn state_update_uses_credential_revision_fence() {
assert_eq!(account.availability(), AccountAvailability::Ready);
}
#[tokio::test]
async fn terminal_state_without_upstream_message_uses_stable_error_reason() {
let (store, repository) = repository();
let input = create_input("state-fallback", "subject-state-fallback");
seed_input(&store, &input).await.expect("create");
repository
.update_state(&UpdateGrokCredentialState {
account_id: input.account_id.clone(),
expected_revision: CredentialRevision::new(1).expect("revision"),
availability: GrokCredentialAvailability::Banned,
availability_reason: None,
observed_at: Utc::now(),
})
.await
.expect("state update");
assert_eq!(
store.last_error_message(&input.account_id).as_deref(),
Some("account_banned")
);
}
#[tokio::test]
async fn admin_prepare_does_not_mutate_provider_account_store() {
let (store, _) = repository();
@@ -10,8 +10,8 @@ use gateway_core::engine::credential::{
ProviderAccountId, ProviderAccountStore, ProviderAccountUpdate,
};
use gateway_core::provider_ports::{
ProviderCooldown, ProviderCooldownPort, ProviderCredentialState, ProviderCredentialStatePort,
ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest, ProviderStoreError,
ProviderCredentialState, ProviderCredentialStatePort, ProviderLeaseAcquisition,
ProviderLeasePort, ProviderLeaseRequest, ProviderStoreError,
};
use provider_xai::{
GrokCredentialCatalogCache, GrokCredentialRecovery, GrokCredentialRecoveryOutcome,
@@ -22,12 +22,13 @@ use provider_xai::{
};
use crate::support::{
MemoryCooldownPort, MemoryGrokCatalogCache, MemoryProviderAccountStore, create_input,
credential_object, runtime_policy, seed_input,
MemoryGrokCatalogCache, MemoryProviderAccountStore, create_input, credential_object,
runtime_policy, seed_input,
};
const OFFICIAL_FIXTURE: &[u8] =
include_bytes!("../transport/catalog/fixtures/official_grok_models_snapshot.json");
const OAUTH_BACKOFF_ATTEMPTS: u32 = 5;
struct StaticCatalogTransport;
@@ -143,48 +144,6 @@ async fn fixture(
fixture_many([input], responses, lease_available).await
}
/// 带外部注入 cooldown 端口的 fixture(AUD-05 组合测试用)。
#[allow(clippy::type_complexity)]
async fn fixture_with_cooldowns(
input: provider_xai::CreateGrokCredential,
responses: impl IntoIterator<Item = Result<GrokRefreshTokens, GrokRefreshFailure>>,
lease_available: bool,
cooldowns: Arc<MemoryCooldownPort>,
) -> (
Arc<MemoryProviderAccountStore>,
GrokCredentialRepository,
Arc<QueueRefresher>,
GrokCredentialRefreshService,
Arc<MemoryCooldownPort>,
) {
let store = MemoryProviderAccountStore::shared();
let account_store: Arc<dyn ProviderAccountStore> = store.clone();
let repository = GrokCredentialRepository::new(account_store);
seed_input(&store, &input).await.expect("create account");
let refresher = QueueRefresher::new(responses);
let refresher_port: Arc<dyn GrokCredentialRefresher> = refresher.clone();
let cache: Arc<dyn GrokCredentialCatalogCache> = MemoryGrokCatalogCache::shared();
let catalog = Arc::new(crate::support::grok_catalog_service(
repository.clone(),
Arc::new(StaticCatalogTransport),
cache,
));
let leases = Arc::new(TestRefreshLeases {
available: lease_available,
calls: AtomicUsize::new(0),
});
let service = GrokCredentialRefreshService::new(
repository.clone(),
refresher_port,
catalog,
leases,
cooldowns.clone(),
Arc::new(CountingCredentialState::default()),
runtime_policy(),
);
(store, repository, refresher, service, cooldowns)
}
/// 真实累加连续失败计数的测试 double,用于断言退避的指数增长与成功清零。
#[derive(Default)]
struct CountingCredentialState {
@@ -313,13 +272,11 @@ async fn fixture_many_with_state(
available: lease_available,
calls: AtomicUsize::new(0),
});
let cooldowns = Arc::new(MemoryCooldownPort::default());
let service = GrokCredentialRefreshService::new(
repository.clone(),
refresher_port,
catalog,
leases,
cooldowns,
credential_state,
runtime_policy(),
);
@@ -354,6 +311,38 @@ async fn force_due(store: &MemoryProviderAccountStore, id: &ProviderAccountId) {
));
}
async fn set_token_deadlines(
store: &MemoryProviderAccountStore,
id: &ProviderAccountId,
access_token_expires_at: SystemTime,
next_refresh_at: SystemTime,
) {
let account = store.account(id).expect("account to update");
let credential = store.credential(id).expect("credential to update");
let update = CredentialCasUpdate::new(
id.clone(),
account.revision(),
ProviderAccountUpdate {
account_id: id.clone(),
name: account.name().to_owned(),
email: account.email().map(str::to_owned),
plan_type: account.plan_type().map(str::to_owned),
},
credential,
true,
Some(access_token_expires_at),
Some(next_refresh_at),
)
.expect("valid token deadline update");
assert!(matches!(
store
.compare_and_swap_credential(update)
.await
.expect("update token deadlines"),
CredentialCasOutcome::Updated(_)
));
}
#[tokio::test]
async fn successful_refresh_rotates_plaintext_tokens_once() {
let input = due_input("success");
@@ -415,6 +404,10 @@ async fn unauthorized_recovery_marks_a_permanently_rejected_refresh_expired() {
store.account(&id).expect("account").availability(),
AccountAvailability::Expired
);
assert_eq!(
store.last_error_message(&id).as_deref(),
Some("refresh_invalid_grant")
);
}
#[tokio::test]
@@ -573,14 +566,25 @@ async fn banned_failure_marks_account_banned() {
store.account(&id).expect("account").availability(),
AccountAvailability::Banned
);
assert_eq!(
store.last_error_message(&id).as_deref(),
Some("account_banned")
);
}
#[tokio::test]
async fn ambiguous_refresh_uses_runtime_cooldown_without_persisting_account_state() {
async fn ambiguous_refresh_applies_bounded_oauth_backoff() {
let input = due_input("ambiguous");
let id = input.account_id.clone();
let (store, _, refresher, service) =
fixture(input, [Err(GrokRefreshFailure::Ambiguous)], true).await;
let counting = Arc::new(CountingCredentialState::default());
let (store, _, refresher, service) = fixture_many_with_state(
[input],
[Err(GrokRefreshFailure::Ambiguous)],
true,
counting.clone(),
)
.await;
let started_at = SystemTime::now();
let outcomes = service.refresh_due().await.expect("refresh");
assert!(matches!(
outcomes.as_slice(),
@@ -590,7 +594,18 @@ async fn ambiguous_refresh_uses_runtime_cooldown_without_persisting_account_stat
store.account(&id).expect("account").availability(),
AccountAvailability::Unknown
);
assert_eq!(store.account(&id).expect("account").revision().get(), 1);
let account = store.account(&id).expect("account");
assert_eq!(account.revision().get(), 2);
let retry_delay = account
.next_refresh_at()
.expect("OAuth retry scheduled")
.duration_since(started_at)
.expect("OAuth retry is in the future");
assert!(
retry_delay <= Duration::from_secs(7),
"first retry must use bounded OAuth backoff, got {retry_delay:?}"
);
assert_eq!(counting.count(&id), 1);
assert!(
service
.refresh_due()
@@ -602,7 +617,7 @@ async fn ambiguous_refresh_uses_runtime_cooldown_without_persisting_account_stat
}
#[tokio::test]
async fn pre_send_transient_failure_applies_bounded_cooldown() {
async fn pre_send_transient_failure_applies_bounded_oauth_backoff() {
let input = due_input("transient");
let id = input.account_id.clone();
let (store, _, _, service) = fixture(input, [Err(GrokRefreshFailure::Transient)], true).await;
@@ -672,6 +687,121 @@ async fn refresh_backoff_grows_exponentially_across_attempts() {
);
}
#[tokio::test]
async fn exhausted_oauth_backoff_continues_after_access_token_expiry() {
let input = due_input("backoff-recovery");
let id = input.account_id.clone();
let counting = Arc::new(CountingCredentialState::default());
let (store, _, refresher, service) = fixture_many_with_state(
[input],
[
Err(GrokRefreshFailure::Transient),
Err(GrokRefreshFailure::Transient),
Err(GrokRefreshFailure::Transient),
Err(GrokRefreshFailure::Transient),
Err(GrokRefreshFailure::Transient),
Err(GrokRefreshFailure::Transient),
],
true,
counting.clone(),
)
.await;
set_token_deadlines(
&store,
&id,
(Utc::now() - chrono::Duration::minutes(1)).into(),
SystemTime::now() - Duration::from_secs(1),
)
.await;
for _ in 0..OAUTH_BACKOFF_ATTEMPTS {
service.refresh_due().await.expect("backoff attempt");
force_due(&store, &id).await;
}
let observed_at = SystemTime::now();
service.refresh_due().await.expect("recovery attempt");
let recovery_delay = store
.account(&id)
.expect("account after recovery defer")
.next_refresh_at()
.expect("recovery scheduled")
.duration_since(observed_at)
.expect("recovery is in the future");
assert!(
(Duration::from_secs(8 * 60)..=Duration::from_secs(12 * 60 + 1)).contains(&recovery_delay),
"recovery delay should be about ten minutes, got {recovery_delay:?}"
);
assert_eq!(counting.count(&id), OAUTH_BACKOFF_ATTEMPTS + 1);
assert_eq!(refresher.prepare_calls.load(Ordering::SeqCst), 6);
}
#[tokio::test]
async fn oauth_backoff_never_passes_an_unexpired_access_token_expiry() {
let mut input = due_input("backoff-expiry-cap");
input.account.access_token_expires_at = Utc::now() + chrono::Duration::seconds(30);
let access_expires_at: SystemTime = input.account.access_token_expires_at.into();
let id = input.account_id.clone();
let counting = Arc::new(CountingCredentialState::default());
let (store, _, _, service) = fixture_many_with_state(
[input],
[
Err(GrokRefreshFailure::Transient),
Err(GrokRefreshFailure::Transient),
Err(GrokRefreshFailure::Transient),
Err(GrokRefreshFailure::Transient),
Err(GrokRefreshFailure::Transient),
Err(GrokRefreshFailure::Transient),
],
true,
counting.clone(),
)
.await;
for attempt in 0..=OAUTH_BACKOFF_ATTEMPTS {
service.refresh_due().await.expect("backoff attempt");
if attempt < OAUTH_BACKOFF_ATTEMPTS {
force_due(&store, &id).await;
}
}
assert_eq!(
store
.account(&id)
.expect("account after capped retry")
.next_refresh_at(),
Some(access_expires_at)
);
assert_eq!(counting.count(&id), OAUTH_BACKOFF_ATTEMPTS + 1);
}
#[tokio::test]
async fn recovery_window_exhaustion_marks_account_expired_without_token_exchange() {
let input = due_input("recovery-window-exhausted");
let id = input.account_id.clone();
let (store, _, refresher, service) = fixture(input, [Ok(success_tokens(None))], true).await;
set_token_deadlines(
&store,
&id,
(Utc::now() - chrono::Duration::hours(25)).into(),
(Utc::now() + chrono::Duration::hours(12)).into(),
)
.await;
let outcomes = service.refresh_due().await.expect("refresh due");
assert!(matches!(
outcomes.as_slice(),
[GrokCredentialRefreshOutcome::Invalidated { account_id }] if account_id == &id
));
assert_eq!(
store.account(&id).expect("expired account").availability(),
AccountAvailability::Expired
);
assert_eq!(refresher.prepare_calls.load(Ordering::SeqCst), 1);
assert_eq!(refresher.responses.lock().expect("queue").len(), 1);
}
#[tokio::test]
async fn successful_refresh_clears_backoff_counter() {
let input = due_input("backoff-clear");
@@ -776,52 +906,33 @@ async fn malformed_account_refresh_does_not_stop_later_accounts() {
}
#[tokio::test]
async fn runtime_cooldown_survives_credential_rotation_and_blocks_refresh() {
// cooldown 写于 revision N,轮换到 N+1 后 refresh worker 不得清除
// 或失效它;cooldown 活跃期间账号被跳过刷新。
let input = create_input(
"cooldown-survives-rotation",
"subject-cooldown-survives-rotation",
);
async fn scheduled_oauth_refresh_runs_after_credential_rotation() {
let input = due_input("refresh-after-rotation");
let id = input.account_id.clone();
let cooldowns = Arc::new(MemoryCooldownPort::default());
let (store, _, _, service, cooldowns) = fixture_with_cooldowns(
let (store, _, refresher, service) = fixture(
input,
[Ok(success_tokens(Some("rotated-refresh")))],
[
Ok(success_tokens(Some("recovered-refresh"))),
Ok(success_tokens(Some("scheduled-refresh"))),
],
true,
cooldowns,
)
.await;
// 先写 account cooldown(revision 1,未来到期)。
let until = SystemTime::now() + Duration::from_secs(3600);
cooldowns
.put_if_later(ProviderCooldown::new(
id.clone(),
CredentialRevision::new(1).expect("revision"),
until,
))
.await
.expect("write cooldown");
// 轮换到 revision 2。
let outcome = service
.recover_unauthorized(&id, CredentialRevision::new(1).expect("revision"))
.await;
assert_eq!(outcome, GrokCredentialRecoveryOutcome::Recovered);
assert_eq!(store.account(&id).expect("account").revision().get(), 2);
// cooldown 仍有效(轮换不清除)。
assert!(
cooldowns
.cooldown(&id)
.is_some_and(|cooldown| cooldown.until() > SystemTime::now()),
"rotation must not clear account runtime cooldown"
);
// cooldown 活跃期间 refresh worker 跳过该账号(不清除、不刷新)。
force_due(&store, &id).await;
let outcomes = service.refresh_due().await.expect("refresh due");
assert!(
outcomes.is_empty(),
"cooldown-active account must be skipped by refresh worker"
);
assert_eq!(store.account(&id).expect("account").revision().get(), 2);
assert!(matches!(
outcomes.as_slice(),
[GrokCredentialRefreshOutcome::Refreshed {
account_id,
credential_revision,
}] if account_id == &id && credential_revision.get() == 4
));
assert_eq!(refresher.prepare_calls.load(Ordering::SeqCst), 2);
}
@@ -194,6 +194,7 @@ struct StoredAccount {
account: ProviderAccount,
credential: PlaintextCredential,
quota: Option<QuotaObservation>,
last_error_message: Option<String>,
}
#[derive(Default)]
@@ -214,6 +215,12 @@ impl MemoryProviderAccountStore {
.map(|stored| stored.account.clone())
}
pub fn last_error_message(&self, id: &ProviderAccountId) -> Option<String> {
lock(&self.accounts)
.get(id)
.and_then(|stored| stored.last_error_message.clone())
}
pub fn credential(&self, id: &ProviderAccountId) -> Option<PlaintextCredential> {
lock(&self.accounts)
.get(id)
@@ -246,6 +253,7 @@ impl ProviderAccountStore for MemoryProviderAccountStore {
account: account.account,
credential: account.credential,
quota: None,
last_error_message: None,
},
);
Ok(())
@@ -388,6 +396,11 @@ impl ProviderAccountStore for MemoryProviderAccountStore {
plan_type: stored.account.plan_type().map(str::to_owned),
},
);
if change.availability == AccountAvailability::Ready {
stored.last_error_message = None;
} else if let Some(message) = change.message {
stored.last_error_message = Some(message);
}
Ok(())
}
+4 -3
View File
@@ -57,16 +57,17 @@ api:
openai:
# 上游地址、WebSocket 池、额度刷新和 OAuth 使用 Provider 内置默认值;模板只保留画像启动基线。
wire_profile:
# Provider 每 24 小时读取官方 npm metadata 与 Desktop appcast,仅在进程内更新版本字段。
# codex_version 是 npm CLI 的启动基线,并用于模型目录与请求 UA;Provider 每 24 小时更新它。
# Desktop appcast 同期只更新 Desktop 版本与构建号。
originator: 'Codex Desktop'
codex_version: '0.146.0'
desktop_version: '26.727.51351'
desktop_build: '6119'
os_type: 'macOS'
os_type: 'Mac OS'
os_version: '15.7.1'
arch: 'arm64'
terminal: 'unknown'
verified_at: '2026-07-25T00:00:00+08:00'
verified_at: '2026-08-04T00:00:00+08:00'
xai:
# OAuth 端点、额度与模型目录策略由 xAI Provider 固定管理;这里只声明请求画像启动基线。
+1 -1
View File
@@ -116,7 +116,7 @@ Provider 独占以下职责:
- OpenAI 是逐字节透明代理:请求 body 的未知字段和字段顺序保持不变(受控模型映射除外),上游 wire 是客户端可见的事实来源,HTTP SSE 帧以原始字节原样下发,WebSocket 上游 JSON 文本原样下发;response ID 以 opaque bytes 存储,不假设 UUID 或固定长度。canonical facts 只从同一帧旁路解析,用于观测、亲和与计费。仅网关内部帧(rate limits、metadata)被就地消费,不向客户端转发。
- xAI 是翻译层:把 Grok wire 解码后重新合成 Responses wire 事件,不承诺字节透明;上游错误按结构化 message/code/type 透出,message 中的 UUID 账号指纹替换为占位符。
credential 更新使用 `credential_revision` CAS。认证永久失败、封禁、额度耗尽和带截止时间的 cooldown 是账号运行时事实。cooldown 到期后,Core 的有效调度谓词会自然允许账号重新参与选择;Redis cooldown 只是热缓存。成功调用或成功额度观测才可把账号状态重新观测为 `ready`。自动 refresh 的瞬态失败(限流、5xx、超时、畸形响应)保留现有凭据并按指数退避推迟下一次刷新,不把账号标记为失效;只有上游明确的永久错误(`invalid_grant`、封禁)才写入终态失效。
credential 更新使用 `credential_revision` CAS。认证永久失败、封禁、额度耗尽和带截止时间的 cooldown 是账号运行时事实。cooldown 到期后,Core 的有效调度谓词会自然允许账号重新参与选择;Redis cooldown 只是热缓存。成功调用或成功额度观测才可把账号状态重新观测为 `ready`。推理请求的账号/model cooldown 只约束数据面调度,不参与 OAuth refresh;OpenAI 与 xAI 的 OAuth refresh 请求状态不明确时只按 OAuth 自身的有界退避推进该 credential 的 `next_refresh_at`:前五次约为 5/15/45/135/300 秒,之后约每 10 分钟恢复;仍有效 AT 时不晚于其到期时刻,AT 到期后仍按该节奏恢复,至 `AT 到期 + 24h` 才终态化。自动 refresh 的瞬态失败(限流、5xx、超时、畸形响应)保留现有凭据并按指数退避推迟下一次刷新,不把账号标记为失效;上游明确的永久错误(`invalid_grant`、封禁)仍立即写入终态失效。
`refresh_token_expires_at` 不是公共 SQL 列或 Core 权威状态。xAI 可在 `provider_credentials_json` 内保存它作为 Provider 私有提示;真正失效以 refresh endpoint 返回的永久错误为准。
@@ -124,8 +124,7 @@ const clientIdentity = computed(() => {
const current = profile.value
if (!current)
return '—'
const label = current.provider === 'openai' ? 'Codex Core' : '客户端标识'
const value = current.attributes.find(attribute => attribute.label === label)?.value ?? '—'
const value = current.attributes.find(attribute => attribute.label === '客户端标识')?.value ?? '—'
return current.provider === 'xai' ? toPascalCase(value) : value
})
@@ -144,7 +143,6 @@ const runtimeEnvironment = computed(() => {
const present = (value: string) => value !== '—' && value.toLowerCase() !== 'unknown'
const primary = [target.osType, target.osVersion]
.filter(present)
.map(value => value.toLowerCase() === 'linux' ? 'Linux' : value)
.join(' ')
const details = [target.arch, target.terminal].filter(present)
return {
@@ -286,7 +284,7 @@ function providerLabel(provider: string) {
class="flex items-center gap-1.5 text-[10px] leading-none font-bold text-(--cp-text-muted)"
>
<Terminal aria-hidden="true" class="size-3.25 text-(--cp-info)" />
{{ profile.provider === 'openai' ? 'Codex Core' : 'Grok Client' }}
客户端标识
</dt>
<dd
class="mt-2 mb-0 truncate font-mono text-[16px] leading-none font-heavy tabular-nums text-(--cp-text-primary)"
+1 -1
View File
@@ -1 +1 @@
version: "3.1.7"
version: "3.1.10"