chore: bump version to 2.5.3

Co-authored-by: Codex <noreply@openai.com>

backend/src/bootstrap/services.rs

@@ -321,7 +321,8 @@ impl Services {
                 refresh_policy.clone(),
                 token_client.clone(),
             )
-            .with_refresh_lease_store(stores.refresh_leases.clone()),
+            .with_refresh_lease_store(stores.refresh_leases.clone())
+            .with_account_pool_sync(account_pool.clone()),
         );
         let upstream_client: Arc<dyn ModelCatalogSource> = codex.clone();
         let store: Arc<dyn ModelSnapshotStore> = Arc::new(stores.model_snapshots.clone());
@@ -546,6 +547,7 @@ pub async fn run(config: BootstrapConfig) -> Result<(), Box<dyn Error + Send + S
     let listener = tokio::net::TcpListener::bind((host.as_str(), port)).await?;
     let (router, task_coordinator, connection_drain) = build_router(config).await?;
     let mut task_coordinator = Some(task_coordinator);
+    let mut restart_path = None;
 
     let serve_result = {
         let (graceful_shutdown_tx, graceful_shutdown_rx) = tokio::sync::oneshot::channel();
@@ -560,46 +562,63 @@ pub async fn run(config: BootstrapConfig) -> Result<(), Box<dyn Error + Send + S
         tokio::pin!(serve);
         tokio::select! {
             result = &mut serve => result,
-            () = crate::bootstrap::shutdown::shutdown_signal() => {
-                let _ = graceful_shutdown_tx.send(());
-                let active_websocket_connections = connection_drain.begin_shutdown();
-                tracing::info!(
-                    active_websocket_connections,
-                    "正在排空入站连接并关闭运行时任务"
-                );
-                let coordinator = task_coordinator.take();
-                let connection_drain = connection_drain.clone();
-                let runtime_shutdown = async move {
-                    let shutdown_tasks = async move {
-                        if let Some(coordinator) = coordinator {
-                            coordinator.shutdown().await;
-                        }
-                    };
-                    tokio::join!(shutdown_tasks, connection_drain.wait());
-                };
-                let drain_runtime = async {
-                    let (serve_result, ()) = tokio::join!(&mut serve, runtime_shutdown);
-                    serve_result
-                };
-                tokio::pin!(drain_runtime);
-                tokio::select! {
-                    result = &mut drain_runtime => result,
-                    () = tokio::time::sleep(HTTP_DRAIN_TIMEOUT) => {
-                        tracing::warn!(
-                            timeout_secs = HTTP_DRAIN_TIMEOUT.as_secs(),
-                            "HTTP graceful shutdown timed out; cancelling remaining connections"
-                        );
+            action = crate::bootstrap::shutdown::shutdown_signal() => {
+                match action {
+                    crate::bootstrap::shutdown::ShutdownAction::Restart(executable_path) => {
+                        restart_path = Some(executable_path);
                         Ok(())
                     }
-                    () = crate::bootstrap::shutdown::shutdown_signal() => {
-                        tracing::warn!("Received a second shutdown signal; cancelling remaining connections");
-                        Ok(())
+                    crate::bootstrap::shutdown::ShutdownAction::Graceful => {
+                        let _ = graceful_shutdown_tx.send(());
+                        let active_websocket_connections = connection_drain.begin_shutdown();
+                        tracing::info!(
+                            active_websocket_connections,
+                            "正在排空入站连接并关闭运行时任务"
+                        );
+                        let coordinator = task_coordinator.take();
+                        let connection_drain = connection_drain.clone();
+                        let runtime_shutdown = async move {
+                            let shutdown_tasks = async move {
+                                if let Some(coordinator) = coordinator {
+                                    coordinator.shutdown().await;
+                                }
+                            };
+                            tokio::join!(shutdown_tasks, connection_drain.wait());
+                        };
+                        let drain_runtime = async {
+                            let (serve_result, ()) = tokio::join!(&mut serve, runtime_shutdown);
+                            serve_result
+                        };
+                        tokio::pin!(drain_runtime);
+                        tokio::select! {
+                            result = &mut drain_runtime => result,
+                            () = tokio::time::sleep(HTTP_DRAIN_TIMEOUT) => {
+                                tracing::warn!(
+                                    timeout_secs = HTTP_DRAIN_TIMEOUT.as_secs(),
+                                    "HTTP graceful shutdown timed out; cancelling remaining connections"
+                                );
+                                Ok(())
+                            }
+                            _ = crate::bootstrap::shutdown::shutdown_signal() => {
+                                tracing::warn!("Received a second shutdown signal; cancelling remaining connections");
+                                Ok(())
+                            }
+                        }
                     }
                 }
             }
         }
     };
 
+    if let Some(executable_path) = restart_path {
+        tracing::info!(
+            executable = %executable_path.display(),
+            "正在以更新后的二进制替换当前进程"
+        );
+        drop(log_guard);
+        return exec_replacement_process(executable_path);
+    }
+
     if let Some(coordinator) = task_coordinator.take() {
         let active_websocket_connections = connection_drain.begin_shutdown();
         tracing::info!(
@@ -610,15 +629,6 @@ pub async fn run(config: BootstrapConfig) -> Result<(), Box<dyn Error + Send + S
     }
     serve_result?;
 
-    if let Some(executable_path) = crate::bootstrap::shutdown::restart_executable_path() {
-        tracing::info!(
-            executable = %executable_path.display(),
-            "正在以更新后的二进制替换当前进程"
-        );
-        drop(log_guard);
-        exec_replacement_process(executable_path)?;
-    }
-
     Ok(())
 }
 

backend/src/bootstrap/shutdown.rs

@@ -1,19 +1,17 @@
-//! 优雅关闭信号处理。
+//! 进程关闭与重启信号处理。
 
-use std::{
-    path::PathBuf,
-    sync::{
-        OnceLock,
-        atomic::{AtomicBool, Ordering},
-    },
-};
+use std::{path::PathBuf, sync::OnceLock};
 
 use tokio::signal;
 use tokio::sync::broadcast;
 
-static SHUTDOWN_REQUESTS: OnceLock<broadcast::Sender<()>> = OnceLock::new();
-static RESTART_REQUESTED: AtomicBool = AtomicBool::new(false);
-static RESTART_EXECUTABLE_PATH: OnceLock<PathBuf> = OnceLock::new();
+static SHUTDOWN_REQUESTS: OnceLock<broadcast::Sender<ShutdownAction>> = OnceLock::new();
+
+#[derive(Clone)]
+pub(crate) enum ShutdownAction {
+    Graceful,
+    Restart(PathBuf),
+}
 
 #[derive(Debug, Clone, Copy, Default)]
 pub struct RuntimeProcessControl;
@@ -24,12 +22,12 @@ impl crate::api::router::ProcessControl for RuntimeProcessControl {
     }
 
     fn request_restart(&self, executable_path: PathBuf) {
-        request_process_restart(executable_path);
+        request_immediate_restart(executable_path);
     }
 }
 
-/// 等待进程关闭信号（Ctrl+C 或 SIGTERM）。
-pub async fn shutdown_signal() {
+/// 等待进程关闭或重启信号。
+pub(crate) async fn shutdown_signal() -> ShutdownAction {
     let mut requested = shutdown_sender().subscribe();
     let ctrl_c = async {
         signal::ctrl_c()
@@ -49,33 +47,25 @@ pub async fn shutdown_signal() {
     let terminate = std::future::pending::<()>();
 
     tokio::select! {
-        _ = requested.recv() => {},
-        () = ctrl_c => {},
-        () = terminate => {},
+        action = requested.recv() => match action {
+            Ok(action) => action,
+            Err(_) => ShutdownAction::Graceful,
+        },
+        () = ctrl_c => ShutdownAction::Graceful,
+        () = terminate => ShutdownAction::Graceful,
     }
 }
 
 /// 请求进程按优雅关闭路径退出。
 pub fn request_shutdown() {
-    let _ = shutdown_sender().send(());
+    let _ = shutdown_sender().send(ShutdownAction::Graceful);
 }
 
-/// 请求进程完成优雅关闭后以新二进制替换当前进程。
-pub fn request_process_restart(executable_path: PathBuf) {
-    let _ = RESTART_EXECUTABLE_PATH.set(executable_path);
-    RESTART_REQUESTED.store(true, Ordering::SeqCst);
-    request_shutdown();
+fn request_immediate_restart(executable_path: PathBuf) {
+    let _ = shutdown_sender().send(ShutdownAction::Restart(executable_path));
 }
 
-/// 获取待 exec 的重启目标。
-pub fn restart_executable_path() -> Option<PathBuf> {
-    RESTART_REQUESTED
-        .load(Ordering::SeqCst)
-        .then(|| RESTART_EXECUTABLE_PATH.get().cloned())
-        .flatten()
-}
-
-fn shutdown_sender() -> &'static broadcast::Sender<()> {
+fn shutdown_sender() -> &'static broadcast::Sender<ShutdownAction> {
     SHUTDOWN_REQUESTS.get_or_init(|| {
         let (sender, _receiver) = broadcast::channel(16);
         sender

backend/src/fleet/import.rs

@@ -586,11 +586,22 @@ pub fn normalized_imported_account_status(
     status: AccountStatus,
     source: AccountImportSource,
     access_token: &str,
+    access_token_expires_at: Option<&DateTime<Utc>>,
 ) -> AccountStatus {
-    if source == AccountImportSource::Sub2Api
-        && status == AccountStatus::Active
-        && jwt_expiry(access_token, Utc::now()) != JwtExpiry::Valid
-    {
+    if source != AccountImportSource::Sub2Api || status != AccountStatus::Active {
+        return status;
+    }
+
+    let now = Utc::now();
+    if let Some(expires_at) = access_token_expires_at {
+        return if now >= *expires_at {
+            AccountStatus::Expired
+        } else {
+            status
+        };
+    }
+
+    if jwt_expiry(access_token, now) != JwtExpiry::Valid {
         AccountStatus::Expired
     } else {
         status

backend/src/fleet/manage/import.rs

@@ -83,16 +83,17 @@ impl AccountManageService {
         let parsed_status =
             crate::fleet::import::parse_account_import_status(entry.status.as_deref())
                 .map_err(|error| AccountManageError::InvalidStatus(error.to_string()))?;
-        let mut status = crate::fleet::import::normalized_imported_account_status(
-            parsed_status,
-            source,
-            &resolved_tokens.access_token,
-        );
         let access_token_expires_at = resolved_tokens
             .claims
             .as_ref()
             .map(|claims| claims.expires_at)
             .or(access_token_expires_at);
+        let mut status = crate::fleet::import::normalized_imported_account_status(
+            parsed_status,
+            source,
+            &resolved_tokens.access_token,
+            access_token_expires_at.as_ref(),
+        );
         let claims = resolved_tokens.claims.as_ref();
         let mut plan_type = claims
             .and_then(|claims| claims.plan_type.clone())

backend/src/fleet/refresh/service.rs

@@ -22,6 +22,7 @@ use uuid::Uuid;
 use super::TokenRefresher;
 use crate::fleet::{
     account::{Account, AccountStatus},
+    pool::AccountPoolService,
     store::{
         AccountClaimsUpdate, AccountStore, PgAccountStore, PgAccountStoreError, StoredAccount,
     },
@@ -48,6 +49,7 @@ where
     C: TokenRefresher,
 {
     store: PgAccountStore,
+    account_pool: Option<Arc<AccountPoolService>>,
     scheduler: Arc<RefreshScheduler<C>>,
     policy: RuntimeRefreshPolicy,
     refresh_leases: Option<RedisRefreshLeaseStore>,
@@ -193,6 +195,7 @@ where
     fn clone(&self) -> Self {
         Self {
             store: self.store.clone(),
+            account_pool: self.account_pool.clone(),
             scheduler: self.scheduler.clone(),
             policy: self.policy.clone(),
             refresh_leases: self.refresh_leases.clone(),
@@ -215,6 +218,7 @@ where
         let policy = policy.into();
         Self {
             store,
+            account_pool: None,
             scheduler: Arc::new(RefreshScheduler::new(policy.clone(), client)),
             policy,
             refresh_leases: None,
@@ -233,6 +237,12 @@ where
         self
     }
 
+    /// 刷新后同步持久化账号到运行时池，避免旧 token 快照覆盖新状态。
+    pub fn with_account_pool_sync(mut self, account_pool: Arc<AccountPoolService>) -> Self {
+        self.account_pool = Some(account_pool);
+        self
+    }
+
     /// 更新刷新策略。
     pub fn update_policy(&self, policy: RefreshPolicy) {
         self.policy.update(policy);
@@ -587,6 +597,7 @@ where
             if attempt.refresh_token.is_none() {
                 persist_status(&self.store, &attempt.id, AccountStatus::Expired).await?;
                 persist_next_refresh_at(&self.store, &attempt.id, None).await?;
+                self.sync_account_pool_after_update(&attempt.id).await;
                 return Ok(TokenRefreshOutcome::StatusUpdated);
             }
 
@@ -601,6 +612,7 @@ where
 
                     persist_status(&self.store, &attempt.id, updated.status).await?;
                     persist_next_refresh_at(&self.store, &attempt.id, None).await?;
+                    self.sync_account_pool_after_update(&attempt.id).await;
                     return Ok(TokenRefreshOutcome::StatusUpdated);
                 }
                 Ok(mut updated) => {
@@ -609,6 +621,7 @@ where
                     updated.next_refresh_at = self.computed_refresh_at(&updated);
                     let refresh_token_rotated = updated.refresh_token != attempt.refresh_token;
                     persist_token_update(&self.store, &updated).await?;
+                    self.sync_account_pool_after_update(&updated.id).await;
                     info!(
                         account_id = %updated.id,
                         attempt = attempt_index + 1,
@@ -643,6 +656,19 @@ where
         Ok(TokenRefreshOutcome::Skipped)
     }
 
+    async fn sync_account_pool_after_update(&self, account_id: &str) {
+        let Some(account_pool) = self.account_pool.as_ref() else {
+            return;
+        };
+        if let Err(error) = account_pool.sync_account_from_store(account_id).await {
+            warn!(
+                account_id,
+                error = %error,
+                "Failed to sync runtime account pool after token refresh"
+            );
+        }
+    }
+
     async fn prepare_refresh_attempt(
         &self,
         account: &Account,

backend/src/fleet/store/queries.rs

@@ -174,6 +174,7 @@ update accounts
 set
   status = case
     when status in ('disabled', 'banned') then status
+    when $4 = 'expired' and access_token_expires_at > $2 then status
     when (
       case
         when not $1 and quota_cooldown_until is not null and quota_cooldown_until > $2 then quota_limit_reached

backend/tests/api/admin/accounts_routes/import_routes.rs

@@ -237,6 +237,89 @@ async fn admin_accounts_import_should_read_sub2api_backup_access_token_only_acco
     assert!(stored.next_refresh_at.is_none());
 }
 
+#[tokio::test]
+async fn admin_accounts_import_should_keep_opaque_sub2api_token_active() {
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/backend-api/wham/usage"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "account_id": "sub2api-opaque-account",
+            "user_id": "sub2api-opaque-user",
+            "email": "sub2api-opaque@example.com",
+            "plan_type": "team",
+            "rate_limit": {
+                "allowed": true,
+                "limit_reached": false,
+                "primary_window": {
+                    "used_percent": 0,
+                    "reset_at": 1_800_000_000,
+                    "limit_window_seconds": 18_000
+                }
+            }
+        })))
+        .mount(&server)
+        .await;
+    let (app, _state, pool, _dir) = admin_accounts_test_app_with_api_base_url(
+        "admin-accounts-import-sub2api-opaque-at",
+        128,
+        format!("{}/backend-api", server.uri()),
+    )
+    .await;
+
+    let response = app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/api/admin/accounts/import")
+                .header("content-type", "application/json")
+                .header("cookie", "cpr_admin_session=session_1")
+                .header("x-request-id", "req_accounts_import_sub2api_opaque_at")
+                .body(Body::from(
+                    json!({
+                        "sourceFormat": "sub2api",
+                        "exported_at": "2026-07-19T18:05:29Z",
+                        "proxies": [],
+                        "accounts": [{
+                            "name": "sub2api-opaque@example.com",
+                            "platform": "openai",
+                            "type": "oauth",
+                            "credentials": {
+                                "access_token": "at-sub2api-opaque-token",
+                                "chatgpt_account_id": "sub2api-opaque-account",
+                                "chatgpt_user_id": "sub2api-opaque-user",
+                                "email": "sub2api-opaque@example.com",
+                                "plan_type": "team",
+                                "refresh_token": "",
+                                "expires_at": "2099-12-01T00:00:00Z"
+                            },
+                            "auto_pause_on_expired": true,
+                            "concurrency": 30,
+                            "priority": 1,
+                            "type": "oauth"
+                        }]
+                    })
+                    .to_string(),
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    let status = response.status();
+    let body = response_json(response).await;
+    let store = PgAccountStore::new(pool);
+    let stored = store
+        .find_by_chatgpt_identity("sub2api-opaque-account", Some("sub2api-opaque-user"))
+        .await
+        .unwrap()
+        .unwrap();
+
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(body["data"]["imported"], 1);
+    assert_eq!(stored.status, AccountStatus::Active);
+    assert!(stored.access_token_expires_at.is_some());
+    assert!(stored.refresh_token.is_none());
+}
+
 #[tokio::test]
 async fn admin_accounts_import_should_read_cliproxyapi_codex_auth_files() {
     let server = MockServer::start().await;

backend/tests/dispatch/service.rs

@@ -212,6 +212,7 @@ const WEBSOCKET_SECOND_ACCOUNT_LIMITED: &str =
 mod responses_http;
 mod responses_recovery;
 mod responses_websocket;
+mod sub2api_import;
 mod usage_logging;
 
 async fn test_app_state_with_pool(config: &AppConfig, pool: PgPool) -> AppState {

backend/tests/dispatch/service/sub2api_import.rs

@@ -0,0 +1,101 @@
+use super::*;
+
+#[tokio::test]
+async fn sub2api_opaque_access_token_with_future_expiry_should_handle_normal_responses_request() {
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/backend-api/wham/usage"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "account_id": "sub2api-opaque-account",
+            "user_id": "sub2api-opaque-user",
+            "email": "sub2api-opaque@example.com",
+            "plan_type": "team",
+            "rate_limit": {
+                "allowed": true,
+                "limit_reached": false,
+                "primary_window": {
+                    "used_percent": 0,
+                    "reset_at": 1_800_000_000,
+                    "limit_window_seconds": 18_000
+                }
+            }
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    Mock::given(method("POST"))
+        .and(path("/backend-api/codex/responses"))
+        .and(header("authorization", "Bearer at-sub2api-opaque-token"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .set_body_string(RESPONSES_SUCCESS_SSE),
+        )
+        .expect(1)
+        .mount(&server)
+        .await;
+
+    let (pool, _dir) = init_test_db("sub2api-opaque-normal-responses").await;
+    let api_key = insert_client_api_key(&pool).await;
+    let state = test_app_state_with_pool(
+        &test_config(test_database_url(), format!("{}/backend-api", server.uri())),
+        pool,
+    )
+    .await;
+    let imported = state
+        .services
+        .admin_accounts
+        .import(json!({
+            "sourceFormat": "sub2api",
+            "exported_at": "2026-07-19T18:05:29Z",
+            "proxies": [],
+            "accounts": [{
+                "name": "sub2api-opaque@example.com",
+                "platform": "openai",
+                "type": "oauth",
+                "credentials": {
+                    "access_token": "at-sub2api-opaque-token",
+                    "chatgpt_account_id": "sub2api-opaque-account",
+                    "chatgpt_user_id": "sub2api-opaque-user",
+                    "email": "sub2api-opaque@example.com",
+                    "plan_type": "team",
+                    "refresh_token": "",
+                    "expires_at": "2099-12-01T00:00:00Z"
+                },
+                "auto_pause_on_expired": true,
+                "concurrency": 30,
+                "priority": 1,
+                "type": "oauth"
+            }]
+        }))
+        .await
+        .unwrap();
+    assert_eq!(imported.imported, 1);
+
+    let app = router::router().with_state(state);
+    let response = app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/v1/responses")
+                .header("authorization", format!("Bearer {api_key}"))
+                .header("content-type", "application/json")
+                .body(Body::from(
+                    json!({
+                        "model": "gpt-5.5",
+                        "input": [],
+                        "stream": false,
+                        "use_websocket": false
+                    })
+                    .to_string(),
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    let status = response.status();
+    let body = response_json(response).await;
+
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(body["id"], "resp_response_1");
+}

backend/tests/fleet/refresh/mod.rs

@@ -9,12 +9,17 @@ use std::{
 
 use async_trait::async_trait;
 use chrono::{Duration, TimeZone, Utc};
-use codex_proxy_rs::fleet::account::AccountStatus;
 use codex_proxy_rs::fleet::refresh::{RedisRefreshLeaseStore, RefreshPolicy};
 use codex_proxy_rs::fleet::refresh::{RefreshFailure, TokenPair, TokenRefresher};
 use codex_proxy_rs::fleet::store::{
     AccountClaimsUpdate, NewAccount, PgAccountStore, StoredAccount,
 };
+use codex_proxy_rs::fleet::{
+    account::AccountStatus,
+    pool::{AccountPoolOptions, AccountPoolService},
+    store::AccountStore,
+};
+use codex_proxy_rs::telemetry::account_usage::store::PgAccountUsageStore;
 use secrecy::{ExposeSecret, SecretString};
 use serde_json::json;
 use tokio::{

backend/tests/fleet/refresh/success.rs

@@ -1,5 +1,85 @@
 use super::*;
 
+#[tokio::test]
+async fn token_refresh_task_should_sync_refreshed_token_into_runtime_account_pool() {
+    let (pool, _guard) = init_test_db("token-refresh-runtime-pool-sync").await;
+    let store = PgAccountStore::new(pool.clone());
+    let now = Utc.with_ymd_and_hms(2026, 6, 18, 7, 0, 0).unwrap();
+    let new_expires_at = now + Duration::hours(1);
+    let new_access_token = test_jwt(new_expires_at.timestamp());
+    store
+        .insert(NewAccount {
+            id: "acct-refresh-runtime-pool".to_string(),
+            email: Some("runtime-pool@example.com".to_string()),
+            account_id: Some("chatgpt-runtime-pool".to_string()),
+            user_id: Some("user-runtime-pool".to_string()),
+            label: None,
+            plan_type: Some("plus".to_string()),
+            access_token: SecretString::new(
+                test_jwt((now + Duration::seconds(30)).timestamp()).into(),
+            ),
+            refresh_token: Some(SecretString::new("refresh-runtime-pool".to_string().into())),
+            added_at: None,
+            access_token_expires_at: Some(now + Duration::seconds(30)),
+            status: AccountStatus::Active,
+        })
+        .await
+        .expect("account should be inserted");
+    let account_pool = Arc::new(AccountPoolService::new(
+        Arc::new(store.clone()) as Arc<dyn AccountStore>,
+        Arc::new(PgAccountUsageStore::new(pool)),
+        AccountPoolOptions {
+            max_concurrent_per_account: 1,
+            ..AccountPoolOptions::default()
+        },
+        0,
+    ));
+    account_pool
+        .restore_from_store()
+        .await
+        .expect("runtime account pool should restore");
+    let refresher = StaticTokenRefresher {
+        response: Arc::new(Mutex::new(Ok(TokenPair {
+            access_token: new_access_token.clone(),
+            refresh_token: None,
+        }))),
+    };
+    let task = codex_proxy_rs::fleet::refresh::TokenRefreshService::new(
+        store.clone(),
+        RefreshPolicy {
+            refresh_margin_seconds: 300,
+            refresh_concurrency: 1,
+        },
+        refresher,
+    )
+    .with_account_pool_sync(Arc::clone(&account_pool));
+
+    task.schedule_account_timers_once_at(now)
+        .await
+        .expect("refresh timer should be scheduled");
+    let stored = wait_for_account(&store, "acct-refresh-runtime-pool", |account| {
+        account.access_token.expose_secret() == new_access_token.as_str()
+    })
+    .await;
+    let capacity = timeout(StdDuration::from_secs(2), async {
+        loop {
+            let capacity = account_pool
+                .capacity_summary(now + Duration::minutes(1))
+                .await;
+            if capacity.available_slots == 1 {
+                return capacity;
+            }
+            sleep(StdDuration::from_millis(20)).await;
+        }
+    })
+    .await
+    .expect("refreshed account should remain available in the runtime pool");
+    task.shutdown().await;
+
+    assert_eq!(stored.status, AccountStatus::Active);
+    assert_eq!(capacity.available_slots, 1);
+}
+
 #[tokio::test]
 async fn token_refresh_task_should_persist_refreshed_access_token_and_keep_refresh_token() {
     let (pool, _guard) = init_test_db("token-refresh").await;

backend/tests/fleet/store/mod.rs

@@ -793,6 +793,66 @@ async fn account_store_runtime_state_sync_should_not_clear_newer_future_quota_co
     assert_eq!(stored.3, Some(future_cooldown));
 }
 
+#[tokio::test]
+async fn account_store_runtime_state_sync_should_not_expire_a_newer_refreshed_token() {
+    let (pool, _dir) = account_store_parts("accounts", 49).await;
+    let repo = PgAccountStore::new(pool.clone());
+    let stale_expires_at = pg_timestamp(Utc::now() - Duration::minutes(1));
+    repo.insert(NewAccount {
+        id: "acct_a".to_string(),
+        email: Some("user@example.com".to_string()),
+        account_id: Some("chatgpt-account".to_string()),
+        user_id: Some("chatgpt-user".to_string()),
+        label: None,
+        plan_type: Some("plus".to_string()),
+        access_token: SecretString::new("stale-access-token".to_string().into()),
+        refresh_token: Some(SecretString::new("refresh-token".to_string().into())),
+        access_token_expires_at: Some(stale_expires_at),
+        status: AccountStatus::Active,
+        added_at: None,
+    })
+    .await
+    .unwrap();
+    let mut stale_account = AccountStore::list_pool_accounts(&repo)
+        .await
+        .unwrap()
+        .into_iter()
+        .find(|account| account.id == "acct_a")
+        .unwrap();
+    let refreshed_expires_at = pg_timestamp(Utc::now() + Duration::hours(1));
+    repo.update_from_claims(
+        "acct_a",
+        AccountClaimsUpdate {
+            email: Some("user@example.com".to_string()),
+            account_id: Some("chatgpt-account".to_string()),
+            user_id: Some("chatgpt-user".to_string()),
+            plan_type: Some("plus".to_string()),
+            access_token: SecretString::new("refreshed-access-token".to_string().into()),
+            refresh_token: None,
+            access_token_expires_at: Some(refreshed_expires_at),
+            next_refresh_at: Some(refreshed_expires_at - Duration::minutes(5)),
+            status: AccountStatus::Active,
+        },
+    )
+    .await
+    .unwrap();
+    stale_account.status = AccountStatus::Expired;
+
+    repo.sync_runtime_account_state(&stale_account)
+        .await
+        .unwrap();
+    let stored = repo.get("acct_a").await.unwrap().unwrap();
+
+    assert_eq!(stored.status, AccountStatus::Active);
+    assert_eq!(
+        (
+            stored.access_token.expose_secret(),
+            stored.access_token_expires_at
+        ),
+        ("refreshed-access-token", Some(refreshed_expires_at))
+    );
+}
+
 #[tokio::test]
 async fn account_store_runtime_state_sync_should_not_regress_newer_usage_window() {
     let (pool, _dir) = account_store_parts("accounts", 43).await;

release/version.yaml

@@ -1 +1 @@
-version: "2.5.1"
+version: "2.5.3"