feat(oauth): 允许替换已失效的活跃 OAuth 账号并同步 status_snapshot

- 活跃但 token 已过期或刷新失败的重复账号视为可替换
- 清除失效标记、刷新配额时同步更新 status_snapshot.oauth
- oauth_invalid 清除接口同时识别 invalid_at 与 invalid_reason 两种标记
- 批量导入任务状态区分 created_count / replaced_count，前端据此展示新增/替换统计
- 补充重复替换场景的集成测试，用量测试等待超时从 10s 提升到 30s 以适应并行压力

apps/aether-gateway/src/handlers/admin/provider/endpoint_keys/mutations/oauth_invalid.rs

@@ -39,7 +39,13 @@ pub(super) async fn maybe_handle(
     else {
         return Ok(Some(not_found_response(format!("Key {key_id} 不存在"))));
     };
-    if key.oauth_invalid_at_unix_secs.is_none() {
+    let has_invalid_marker = key.oauth_invalid_at_unix_secs.is_some()
+        || key
+            .oauth_invalid_reason
+            .as_deref()
+            .map(str::trim)
+            .is_some_and(|value| !value.is_empty());
+    if !has_invalid_marker {
         return Ok(Some(
             Json(json!({
                 "message": "该 Key 当前无失效标记，无需清除"

apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/batch/parse.rs

@@ -251,6 +251,8 @@ pub(super) fn build_admin_provider_oauth_batch_task_state(
     processed: usize,
     success: usize,
     failed: usize,
+    created_count: usize,
+    replaced_count: usize,
     message: Option<&str>,
     error: Option<&str>,
     error_samples: Vec<serde_json::Value>,
@@ -277,6 +279,8 @@ pub(super) fn build_admin_provider_oauth_batch_task_state(
         "processed": processed,
         "success": success,
         "failed": failed,
+        "created_count": created_count,
+        "replaced_count": replaced_count,
         "progress_percent": progress_percent,
         "message": message,
         "error": error,

apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/batch/task.rs

@@ -95,6 +95,8 @@ pub(in super::super) async fn handle_admin_provider_oauth_start_batch_import_tas
         0,
         0,
         0,
+        0,
+        0,
         Some("任务已提交，等待执行"),
         None,
         Vec::new(),
@@ -134,6 +136,8 @@ pub(in super::super) async fn handle_admin_provider_oauth_start_batch_import_tas
             0,
             0,
             0,
+            0,
+            0,
             Some("任务开始执行"),
             None,
             Vec::new(),
@@ -169,6 +173,16 @@ pub(in super::super) async fn handle_admin_provider_oauth_start_batch_import_tas
                     .take(PROVIDER_OAUTH_BATCH_TASK_MAX_ERROR_SAMPLES)
                     .cloned()
                     .collect::<Vec<_>>();
+                let replaced_count = outcome
+                    .results
+                    .iter()
+                    .filter(|item| {
+                        item.get("status").and_then(serde_json::Value::as_str) == Some("success")
+                            && item.get("replaced").and_then(serde_json::Value::as_bool)
+                                == Some(true)
+                    })
+                    .count();
+                let created_count = outcome.success.saturating_sub(replaced_count);
                 let message = format!(
                     "导入完成：成功 {}，失败 {}",
                     outcome.success, outcome.failed
@@ -182,6 +196,8 @@ pub(in super::super) async fn handle_admin_provider_oauth_start_batch_import_tas
                     outcome.total,
                     outcome.success,
                     outcome.failed,
+                    created_count,
+                    replaced_count,
                     Some(message.as_str()),
                     None,
                     error_samples,
@@ -209,6 +225,8 @@ pub(in super::super) async fn handle_admin_provider_oauth_start_batch_import_tas
                     0,
                     0,
                     0,
+                    0,
+                    0,
                     Some("导入任务执行失败"),
                     Some(error_message.as_str()),
                     Vec::new(),
@@ -238,6 +256,8 @@ pub(in super::super) async fn handle_admin_provider_oauth_start_batch_import_tas
         0,
         0,
         0,
+        0,
+        0,
         Some("任务已提交，等待执行"),
         None,
         Vec::new(),

apps/aether-gateway/src/handlers/admin/provider/oauth/duplicates.rs

@@ -1,6 +1,7 @@
 use crate::handlers::admin::request::AdminAppState;
 use crate::provider_key_auth::provider_key_is_oauth_managed;
 use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
+use std::time::{SystemTime, UNIX_EPOCH};
 
 fn normalize_codex_plan_group_for_provider_oauth(
     plan_type: Option<&serde_json::Value>,
@@ -123,6 +124,40 @@ fn is_codex_cross_plan_group_non_duplicate(
     )
 }
 
+fn provider_oauth_invalid_reason_allows_replace(reason: &str) -> bool {
+    reason.lines().map(str::trim).any(|line| {
+        line.starts_with("[OAUTH_EXPIRED] ")
+            || line.starts_with("[REFRESH_FAILED] ")
+            || line.contains("Token 无效或已过期")
+            || line.contains("refresh_token 无效、已过期或已撤销")
+    })
+}
+
+fn existing_provider_oauth_key_is_replaceable(existing_key: &StoredProviderCatalogKey) -> bool {
+    if !existing_key.is_active {
+        return true;
+    }
+
+    let now_unix_secs = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .ok()
+        .map(|duration| duration.as_secs())
+        .unwrap_or(0);
+    if existing_key
+        .expires_at_unix_secs
+        .is_some_and(|expires_at| expires_at <= now_unix_secs)
+    {
+        return true;
+    }
+
+    existing_key
+        .oauth_invalid_reason
+        .as_deref()
+        .map(str::trim)
+        .filter(|reason| !reason.is_empty())
+        .is_some_and(provider_oauth_invalid_reason_allows_replace)
+}
+
 pub(crate) async fn find_duplicate_provider_oauth_key(
     state: &AdminAppState<'_>,
     provider_id: &str,
@@ -210,7 +245,7 @@ pub(crate) async fn find_duplicate_provider_oauth_key(
         if !is_duplicate {
             continue;
         }
-        if !existing_key.is_active {
+        if existing_provider_oauth_key_is_replaceable(&existing_key) {
             return Ok(Some(existing_key));
         }
         let identifier =

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/shared.rs

@@ -2,7 +2,9 @@ use crate::handlers::admin::provider::shared::payloads::{
     OAUTH_ACCOUNT_BLOCK_PREFIX, OAUTH_REFRESH_FAILED_PREFIX,
 };
 use crate::handlers::admin::request::{AdminAppState, AdminGatewayProviderTransportSnapshot};
-use crate::handlers::shared::sync_provider_key_quota_status_snapshot;
+use crate::handlers::shared::{
+    sync_provider_key_oauth_status_snapshot, sync_provider_key_quota_status_snapshot,
+};
 use crate::GatewayError;
 use aether_admin::provider::quota as admin_provider_quota_pure;
 use aether_contracts::{ExecutionPlan, ExecutionResult, ExecutionTimeouts, ProxySnapshot};
@@ -146,6 +148,8 @@ pub(crate) async fn persist_provider_quota_refresh_state(
             "refresh_api",
         );
     }
+    latest_key.status_snapshot =
+        sync_provider_key_oauth_status_snapshot(latest_key.status_snapshot.as_ref(), &latest_key);
     latest_key.updated_at_unix_secs = SystemTime::now()
         .duration_since(UNIX_EPOCH)
         .ok()

apps/aether-gateway/src/handlers/shared/catalog.rs

@@ -8,6 +8,12 @@ use aether_crypto::DEVELOPMENT_ENCRYPTION_KEY;
 use aether_crypto::{decrypt_python_fernet_ciphertext, encrypt_python_fernet_plaintext};
 use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
 use serde_json::{json, Map, Value};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+const OAUTH_ACCOUNT_BLOCK_PREFIX: &str = "[ACCOUNT_BLOCK] ";
+const OAUTH_EXPIRED_PREFIX: &str = "[OAUTH_EXPIRED] ";
+const OAUTH_REFRESH_FAILED_PREFIX: &str = "[REFRESH_FAILED] ";
+const OAUTH_REQUEST_FAILED_PREFIX: &str = "[REQUEST_FAILED] ";
 
 pub(crate) fn provider_catalog_key_supports_format(
     key: &StoredProviderCatalogKey,
@@ -155,6 +161,159 @@ pub(crate) fn default_provider_key_status_snapshot() -> serde_json::Value {
     })
 }
 
+fn default_oauth_status_snapshot_value() -> Value {
+    default_provider_key_status_snapshot()
+        .get("oauth")
+        .cloned()
+        .unwrap_or_else(|| {
+            json!({
+                "code": "none",
+                "label": Value::Null,
+                "reason": Value::Null,
+                "expires_at": Value::Null,
+                "invalid_at": Value::Null,
+                "source": Value::Null,
+                "requires_reauth": false,
+                "expiring_soon": false,
+            })
+        })
+}
+
+fn trimmed_oauth_invalid_reason(reason: Option<&str>) -> Option<String> {
+    reason
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map(ToOwned::to_owned)
+}
+
+fn tagged_oauth_invalid_reason(reason: Option<&str>, prefix: &str) -> Option<String> {
+    reason.and_then(|value| {
+        value
+            .lines()
+            .map(str::trim)
+            .find_map(|line| line.strip_prefix(prefix))
+            .map(str::trim)
+            .filter(|value| !value.is_empty())
+            .map(ToOwned::to_owned)
+    })
+}
+
+fn build_provider_key_oauth_status_snapshot(key: &StoredProviderCatalogKey) -> Value {
+    if !key.auth_type.trim().eq_ignore_ascii_case("oauth") {
+        return default_oauth_status_snapshot_value();
+    }
+
+    let now_unix_secs = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .ok()
+        .map(|duration| duration.as_secs())
+        .unwrap_or(0);
+    let expires_at_unix_secs = key.expires_at_unix_secs;
+    let invalid_at_unix_secs = key.oauth_invalid_at_unix_secs;
+    let invalid_reason = trimmed_oauth_invalid_reason(key.oauth_invalid_reason.as_deref());
+
+    if let Some(reason) =
+        tagged_oauth_invalid_reason(invalid_reason.as_deref(), OAUTH_EXPIRED_PREFIX)
+    {
+        return json!({
+            "code": "invalid",
+            "label": "已失效",
+            "reason": reason,
+            "expires_at": expires_at_unix_secs,
+            "invalid_at": invalid_at_unix_secs,
+            "source": "oauth_invalid",
+            "requires_reauth": true,
+            "expiring_soon": false,
+        });
+    }
+    if let Some(reason) =
+        tagged_oauth_invalid_reason(invalid_reason.as_deref(), OAUTH_REFRESH_FAILED_PREFIX)
+    {
+        return json!({
+            "code": "invalid",
+            "label": "已失效",
+            "reason": reason,
+            "expires_at": expires_at_unix_secs,
+            "invalid_at": invalid_at_unix_secs,
+            "source": "oauth_refresh",
+            "requires_reauth": true,
+            "expiring_soon": false,
+        });
+    }
+    if let Some(reason) =
+        tagged_oauth_invalid_reason(invalid_reason.as_deref(), OAUTH_REQUEST_FAILED_PREFIX)
+    {
+        return json!({
+            "code": "check_failed",
+            "label": "检查失败",
+            "reason": reason,
+            "expires_at": expires_at_unix_secs,
+            "invalid_at": Value::Null,
+            "source": "oauth_request",
+            "requires_reauth": false,
+            "expiring_soon": false,
+        });
+    }
+    if invalid_reason
+        .as_deref()
+        .is_some_and(|reason| !reason.starts_with(OAUTH_ACCOUNT_BLOCK_PREFIX))
+        || invalid_at_unix_secs.is_some()
+    {
+        return json!({
+            "code": "invalid",
+            "label": "已失效",
+            "reason": invalid_reason,
+            "expires_at": expires_at_unix_secs,
+            "invalid_at": invalid_at_unix_secs,
+            "source": "oauth_invalid",
+            "requires_reauth": true,
+            "expiring_soon": false,
+        });
+    }
+
+    let Some(expires_at_unix_secs) = expires_at_unix_secs else {
+        return default_oauth_status_snapshot_value();
+    };
+    if expires_at_unix_secs <= now_unix_secs {
+        return json!({
+            "code": "expired",
+            "label": "已过期",
+            "reason": "Token 已过期，请重新授权",
+            "expires_at": expires_at_unix_secs,
+            "invalid_at": Value::Null,
+            "source": "expires_at",
+            "requires_reauth": true,
+            "expiring_soon": false,
+        });
+    }
+
+    let expiring_soon = expires_at_unix_secs.saturating_sub(now_unix_secs) < 24 * 60 * 60;
+    json!({
+        "code": if expiring_soon { "expiring" } else { "valid" },
+        "label": if expiring_soon { "即将过期" } else { "有效" },
+        "reason": Value::Null,
+        "expires_at": expires_at_unix_secs,
+        "invalid_at": Value::Null,
+        "source": "expires_at",
+        "requires_reauth": false,
+        "expiring_soon": expiring_soon,
+    })
+}
+
+pub(crate) fn sync_provider_key_oauth_status_snapshot(
+    status_snapshot: Option<&Value>,
+    key: &StoredProviderCatalogKey,
+) -> Option<Value> {
+    let mut snapshot = provider_key_status_snapshot_object(status_snapshot)
+        .or_else(|| default_provider_key_status_snapshot().as_object().cloned())
+        .unwrap_or_default();
+    snapshot.insert(
+        "oauth".to_string(),
+        build_provider_key_oauth_status_snapshot(key),
+    );
+    Some(Value::Object(snapshot))
+}
+
 fn build_provider_key_account_status_snapshot(
     key: &StoredProviderCatalogKey,
     provider_type: &str,

apps/aether-gateway/src/handlers/shared/mod.rs

@@ -23,7 +23,8 @@ pub(crate) use self::catalog::{
     default_provider_key_status_snapshot, effective_catalog_encryption_key,
     encrypt_catalog_secret_with_fallbacks, masked_catalog_api_key, parse_catalog_auth_config_json,
     provider_catalog_key_supports_format, provider_key_health_summary,
-    provider_key_status_snapshot_payload, sync_provider_key_quota_status_snapshot,
+    provider_key_status_snapshot_payload, sync_provider_key_oauth_status_snapshot,
+    sync_provider_key_quota_status_snapshot,
 };
 pub(crate) use self::email_templates::{
     admin_email_template_definition, admin_email_template_html_key,

apps/aether-gateway/src/state/catalog.rs

@@ -1,5 +1,7 @@
 use super::{AppState, GatewayError, LocalMutationOutcome, LocalProviderDeleteTaskState};
+use crate::handlers::shared::sync_provider_key_oauth_status_snapshot;
 use aether_data_contracts::repository::{candidates, global_models, provider_catalog};
+use std::time::{SystemTime, UNIX_EPOCH};
 
 impl AppState {
     pub fn has_provider_catalog_data_reader(&self) -> bool {
@@ -633,10 +635,29 @@ impl AppState {
         &self,
         key_id: &str,
     ) -> Result<bool, GatewayError> {
-        self.data
+        let Some(mut key) = self
-            .clear_provider_catalog_key_oauth_invalid_marker(key_id)
+            .data
+            .list_provider_catalog_keys_by_ids(&[key_id.to_string()])
             .await
-            .map_err(|err| GatewayError::Internal(err.to_string()))
+            .map_err(|err| GatewayError::Internal(err.to_string()))?
+            .into_iter()
+            .next()
+        else {
+            return Ok(false);
+        };
+
+        key.oauth_invalid_at_unix_secs = None;
+        key.oauth_invalid_reason = None;
+        key.status_snapshot =
+            sync_provider_key_oauth_status_snapshot(key.status_snapshot.as_ref(), &key);
+        key.updated_at_unix_secs = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .ok()
+            .map(|duration| duration.as_secs());
+
+        self.update_provider_catalog_key(&key)
+            .await
+            .map(|updated| updated.is_some())
     }
 
     pub(crate) fn put_provider_delete_task(&self, task: LocalProviderDeleteTaskState) {

apps/aether-gateway/src/tests/control/admin/endpoints/keys.rs

@@ -548,8 +548,40 @@ async fn gateway_clears_admin_provider_key_oauth_invalid_locally_with_trusted_ad
         "openai:chat",
         "sk-test-a",
     );
+    key.auth_type = "oauth".to_string();
     key.oauth_invalid_at_unix_secs = Some(1_710_000_000);
     key.oauth_invalid_reason = Some("token expired".to_string());
+    key.expires_at_unix_secs = Some(1_900_000_000);
+    key.status_snapshot = Some(json!({
+        "oauth": {
+            "code": "invalid",
+            "label": "已失效",
+            "reason": "token expired",
+            "expires_at": 1_900_000_000u64,
+            "invalid_at": 1_710_000_000u64,
+            "source": "oauth_invalid",
+            "requires_reauth": true,
+            "expiring_soon": false
+        },
+        "account": {
+            "code": "ok",
+            "label": serde_json::Value::Null,
+            "reason": serde_json::Value::Null,
+            "blocked": false,
+            "source": serde_json::Value::Null,
+            "recoverable": false
+        },
+        "quota": {
+            "code": "unknown",
+            "label": serde_json::Value::Null,
+            "reason": serde_json::Value::Null,
+            "exhausted": false,
+            "usage_ratio": serde_json::Value::Null,
+            "updated_at": serde_json::Value::Null,
+            "reset_seconds": serde_json::Value::Null,
+            "plan_type": serde_json::Value::Null
+        }
+    }));
 
     let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
         vec![sample_provider("provider-openai", "openai", 10)],
@@ -594,6 +626,22 @@ async fn gateway_clears_admin_provider_key_oauth_invalid_locally_with_trusted_ad
     assert_eq!(reloaded.len(), 1);
     assert_eq!(reloaded[0].oauth_invalid_at_unix_secs, None);
     assert_eq!(reloaded[0].oauth_invalid_reason, None);
+    let oauth_snapshot = reloaded[0]
+        .status_snapshot
+        .as_ref()
+        .and_then(serde_json::Value::as_object)
+        .and_then(|snapshot| snapshot.get("oauth"))
+        .and_then(serde_json::Value::as_object)
+        .expect("oauth snapshot should exist");
+    assert_eq!(oauth_snapshot.get("code"), Some(&json!("valid")));
+    assert_eq!(oauth_snapshot.get("label"), Some(&json!("有效")));
+    assert_eq!(oauth_snapshot.get("reason"), Some(&serde_json::Value::Null));
+    assert_eq!(
+        oauth_snapshot.get("invalid_at"),
+        Some(&serde_json::Value::Null)
+    );
+    assert_eq!(oauth_snapshot.get("source"), Some(&json!("expires_at")));
+    assert_eq!(oauth_snapshot.get("requires_reauth"), Some(&json!(false)));
 
     gateway_handle.abort();
     upstream_handle.abort();

apps/aether-gateway/src/tests/control/admin/oauth.rs

@@ -1176,6 +1176,8 @@ async fn gateway_handles_admin_provider_oauth_batch_import_task_status_locally_w
                     "processed": 2,
                     "success": 1,
                     "failed": 1,
+                    "created_count": 0,
+                    "replaced_count": 1,
                     "progress_percent": 100,
                     "message": "导入完成：成功 1，失败 1",
                     "error": null,
@@ -1216,6 +1218,8 @@ async fn gateway_handles_admin_provider_oauth_batch_import_task_status_locally_w
     assert_eq!(payload["processed"], 2);
     assert_eq!(payload["success"], 1);
     assert_eq!(payload["failed"], 1);
+    assert_eq!(payload["created_count"], 0);
+    assert_eq!(payload["replaced_count"], 1);
     assert_eq!(payload["progress_percent"], 100);
     assert_eq!(payload["error_samples"].as_array().map(Vec::len), Some(1));
     assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
@@ -1595,6 +1599,8 @@ async fn gateway_starts_admin_provider_oauth_batch_import_task_locally_with_trus
     assert_eq!(status_payload["processed"], 1);
     assert_eq!(status_payload["success"], 1);
     assert_eq!(status_payload["failed"], 0);
+    assert_eq!(status_payload["created_count"], 1);
+    assert_eq!(status_payload["replaced_count"], 0);
     assert_eq!(status_payload["progress_percent"], 100);
     assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
     assert_eq!(*token_hits.lock().expect("mutex should lock"), 1);
@@ -2123,6 +2129,183 @@ async fn gateway_imports_admin_provider_oauth_refresh_token_locally_with_trusted
     upstream_handle.abort();
 }
 
+#[tokio::test]
+async fn gateway_imports_admin_provider_oauth_refresh_token_over_active_expired_duplicate() {
+    #[derive(Debug, Clone)]
+    struct SeenTokenRequest {
+        content_type: String,
+        body: String,
+    }
+
+    let upstream_hits = Arc::new(Mutex::new(0usize));
+    let upstream_hits_clone = Arc::clone(&upstream_hits);
+    let upstream = Router::new().fallback(any(move |_request: Request| {
+        let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
+        async move {
+            *upstream_hits_inner.lock().expect("mutex should lock") += 1;
+            (StatusCode::OK, Body::from("unexpected upstream hit"))
+        }
+    }));
+
+    let token_hits = Arc::new(Mutex::new(0usize));
+    let token_hits_clone = Arc::clone(&token_hits);
+    let seen_token = Arc::new(Mutex::new(None::<SeenTokenRequest>));
+    let seen_token_clone = Arc::clone(&seen_token);
+    let token_server = Router::new().route(
+        "/oauth/token",
+        post(move |headers: HeaderMap, body: Bytes| {
+            let token_hits_inner = Arc::clone(&token_hits_clone);
+            let seen_token_inner = Arc::clone(&seen_token_clone);
+            async move {
+                *token_hits_inner.lock().expect("mutex should lock") += 1;
+                *seen_token_inner.lock().expect("mutex should lock") = Some(SeenTokenRequest {
+                    content_type: headers
+                        .get(http::header::CONTENT_TYPE)
+                        .and_then(|value| value.to_str().ok())
+                        .unwrap_or_default()
+                        .to_string(),
+                    body: String::from_utf8(body.to_vec()).unwrap_or_default(),
+                });
+                Json(json!({
+                    "access_token": "imported-expired-codex-access-token",
+                    "refresh_token": "imported-expired-codex-refresh-token",
+                    "token_type": "Bearer",
+                    "expires_in": 1800,
+                    "scope": "openid email profile offline_access",
+                    "email": "alice@example.com",
+                    "account_id": "acct-codex-123",
+                    "plan_type": "plus",
+                }))
+            }
+        }),
+    );
+
+    let mut provider = sample_provider("provider-codex", "codex", 10);
+    provider.provider_type = "codex".to_string();
+    let endpoint = sample_endpoint(
+        "endpoint-codex-chat",
+        "provider-codex",
+        "openai:chat",
+        "https://chatgpt.com/backend-api/codex",
+    );
+
+    let mut existing_key = sample_key(
+        "key-codex-import-expired-duplicate",
+        "provider-codex",
+        "openai:chat",
+        "stale-imported-access-token",
+    );
+    existing_key.auth_type = "oauth".to_string();
+    existing_key.is_active = true;
+    existing_key.expires_at_unix_secs = Some(1);
+    existing_key.oauth_invalid_at_unix_secs = Some(1_700_000_000);
+    existing_key.oauth_invalid_reason =
+        Some("[REFRESH_FAILED] refresh_token 无效、已过期或已撤销，请重新登录授权".to_string());
+    existing_key.encrypted_auth_config = Some(
+        encrypt_python_fernet_plaintext(
+            DEVELOPMENT_ENCRYPTION_KEY,
+            r#"{"provider_type":"codex","email":"alice@example.com","account_id":"acct-codex-123","plan_type":"plus","refresh_token":"old-refresh-token","expires_at":1}"#,
+        )
+        .expect("auth config ciphertext should build"),
+    );
+
+    let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
+        vec![provider],
+        vec![endpoint],
+        vec![existing_key],
+    ));
+
+    let (upstream_url, upstream_handle) = start_server(upstream).await;
+    let (token_url, token_handle) = start_server(token_server).await;
+    let gateway = build_router_with_state(
+        AppState::new()
+            .expect("gateway should build")
+            .with_data_state_for_tests(
+                GatewayDataState::with_provider_catalog_repository_for_tests(
+                    provider_catalog_repository.clone(),
+                )
+                .with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
+            )
+            .with_provider_oauth_token_url_for_tests("codex", format!("{token_url}/oauth/token")),
+    );
+    let (gateway_url, gateway_handle) = start_server(gateway).await;
+
+    let response = reqwest::Client::new()
+        .post(format!(
+            "{gateway_url}/api/admin/provider-oauth/providers/provider-codex/import-refresh-token"
+        ))
+        .header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
+        .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
+        .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
+        .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
+        .json(&json!({
+            "refresh_token": "provider-import-refresh-token",
+            "proxy_node_id": "proxy-node-codex-import",
+            "name": "should-not-override-active-expired-name"
+        }))
+        .send()
+        .await
+        .expect("request should succeed");
+
+    let status = response.status();
+    let payload: serde_json::Value = response.json().await.expect("json body should parse");
+    assert_eq!(status, StatusCode::OK, "payload={payload}");
+    assert_eq!(payload["key_id"], "key-codex-import-expired-duplicate");
+    assert_eq!(payload["provider_type"], "codex");
+    assert_eq!(payload["has_refresh_token"], true);
+    assert_eq!(payload["email"], "alice@example.com");
+    assert_eq!(payload["replaced"], true);
+    assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
+    assert_eq!(*token_hits.lock().expect("mutex should lock"), 1);
+
+    let seen_token = seen_token
+        .lock()
+        .expect("mutex should lock")
+        .clone()
+        .expect("token request should be recorded");
+    assert_eq!(seen_token.content_type, "application/x-www-form-urlencoded");
+    assert!(seen_token.body.contains("grant_type=refresh_token"));
+    assert!(seen_token
+        .body
+        .contains("refresh_token=provider-import-refresh-token"));
+
+    let reloaded = provider_catalog_repository
+        .list_keys_by_ids(&["key-codex-import-expired-duplicate".to_string()])
+        .await
+        .expect("keys should load");
+    let persisted = reloaded.first().expect("persisted key should exist");
+    assert!(persisted.is_active);
+    assert_eq!(persisted.proxy, None);
+    assert_eq!(persisted.oauth_invalid_at_unix_secs, None);
+    assert_eq!(persisted.oauth_invalid_reason, None);
+    let decrypted_api_key =
+        decrypt_python_fernet_ciphertext(DEVELOPMENT_ENCRYPTION_KEY, &persisted.encrypted_api_key)
+            .expect("api key should decrypt");
+    assert_eq!(decrypted_api_key, "imported-expired-codex-access-token");
+    let decrypted_auth_config = decrypt_python_fernet_ciphertext(
+        DEVELOPMENT_ENCRYPTION_KEY,
+        persisted
+            .encrypted_auth_config
+            .as_deref()
+            .expect("auth config should be stored"),
+    )
+    .expect("auth config should decrypt");
+    let auth_config: serde_json::Value =
+        serde_json::from_str(&decrypted_auth_config).expect("auth config json should parse");
+    assert_eq!(auth_config["provider_type"], "codex");
+    assert_eq!(
+        auth_config["refresh_token"],
+        "imported-expired-codex-refresh-token"
+    );
+    assert_eq!(auth_config["email"], "alice@example.com");
+    assert_eq!(auth_config["account_id"], "acct-codex-123");
+    assert_eq!(auth_config["plan_type"], "plus");
+
+    gateway_handle.abort();
+    token_handle.abort();
+    upstream_handle.abort();
+}
+
 #[tokio::test]
 async fn gateway_rejects_kiro_single_refresh_token_import_with_clear_error() {
     let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
@@ -2745,6 +2928,152 @@ async fn gateway_batch_imports_admin_provider_oauth_kiro_locally_with_trusted_ad
     upstream_handle.abort();
 }
 
+#[tokio::test]
+async fn gateway_batch_imports_admin_provider_oauth_kiro_over_active_expired_duplicate() {
+    let upstream_hits = Arc::new(Mutex::new(0usize));
+    let upstream_hits_clone = Arc::clone(&upstream_hits);
+    let upstream = Router::new().fallback(any(move |_request: Request| {
+        let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
+        async move {
+            *upstream_hits_inner.lock().expect("mutex should lock") += 1;
+            (StatusCode::OK, Body::from("unexpected upstream hit"))
+        }
+    }));
+
+    let refresh_hits = Arc::new(Mutex::new(0usize));
+    let refresh_hits_clone = Arc::clone(&refresh_hits);
+    let refresh_server = Router::new().route(
+        "/refreshToken",
+        post(move |_request: Request| {
+            let refresh_hits_inner = Arc::clone(&refresh_hits_clone);
+            async move {
+                *refresh_hits_inner.lock().expect("mutex should lock") += 1;
+                Json(json!({
+                    "accessToken": sample_kiro_device_access_token("kiro-batch@example.com"),
+                    "refreshToken": "kiro-batch-refresh-token-replaced",
+                    "expiresIn": 1800,
+                }))
+            }
+        }),
+    );
+
+    let mut provider = sample_provider("provider-kiro", "kiro", 10);
+    provider.provider_type = "kiro".to_string();
+    let endpoint = sample_endpoint(
+        "endpoint-kiro-chat",
+        "provider-kiro",
+        "kiro:generateAssistantResponse",
+        "https://service.kiro.dev",
+    );
+
+    let mut existing_key = sample_key(
+        "key-kiro-batch-expired-duplicate",
+        "provider-kiro",
+        "kiro:generateAssistantResponse",
+        "stale-kiro-access-token",
+    );
+    existing_key.auth_type = "oauth".to_string();
+    existing_key.is_active = true;
+    existing_key.oauth_invalid_at_unix_secs = Some(1_700_000_000);
+    existing_key.oauth_invalid_reason = Some("Kiro Token 无效或已过期".to_string());
+    existing_key.encrypted_auth_config = Some(
+        encrypt_python_fernet_plaintext(
+            DEVELOPMENT_ENCRYPTION_KEY,
+            r#"{"provider_type":"kiro","auth_method":"social","email":"kiro-batch@example.com","refresh_token":"kiro-batch-refresh-old"}"#,
+        )
+        .expect("auth config ciphertext should build"),
+    );
+
+    let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
+        vec![provider],
+        vec![endpoint],
+        vec![existing_key],
+    ));
+
+    let (upstream_url, upstream_handle) = start_server(upstream).await;
+    let (refresh_url, refresh_handle) = start_server(refresh_server).await;
+    let gateway = build_router_with_state(
+        AppState::new()
+            .expect("gateway should build")
+            .with_data_state_for_tests(
+                GatewayDataState::with_provider_catalog_repository_for_tests(
+                    provider_catalog_repository.clone(),
+                )
+                .with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
+            )
+            .with_provider_oauth_token_url_for_tests(
+                "kiro_social_refresh",
+                refresh_url.to_string(),
+            ),
+    );
+    let (gateway_url, gateway_handle) = start_server(gateway).await;
+
+    let response = reqwest::Client::new()
+        .post(format!(
+            "{gateway_url}/api/admin/provider-oauth/providers/provider-kiro/batch-import"
+        ))
+        .header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
+        .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
+        .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
+        .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
+        .json(&json!({
+            "credentials": "kiro-batch-refresh-token",
+            "proxy_node_id": "proxy-node-kiro-batch"
+        }))
+        .send()
+        .await
+        .expect("request should succeed");
+
+    assert_eq!(response.status(), StatusCode::OK);
+    let payload: serde_json::Value = response.json().await.expect("payload should parse");
+    assert_eq!(payload["total"], 1);
+    assert_eq!(payload["success"], 1);
+    assert_eq!(payload["failed"], 0);
+    let results = payload["results"]
+        .as_array()
+        .expect("results should be array");
+    assert_eq!(results.len(), 1);
+    assert_eq!(results[0]["status"], "success");
+    assert_eq!(results[0]["key_id"], "key-kiro-batch-expired-duplicate");
+    assert_eq!(results[0]["auth_method"], "social");
+    assert_eq!(results[0]["replaced"], true);
+    assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
+    assert_eq!(*refresh_hits.lock().expect("mutex should lock"), 1);
+
+    let stored_key = provider_catalog_repository
+        .list_keys_by_ids(&["key-kiro-batch-expired-duplicate".to_string()])
+        .await
+        .expect("keys should load")
+        .into_iter()
+        .next()
+        .expect("persisted key should exist");
+    assert!(stored_key.is_active);
+    assert_eq!(stored_key.proxy, None);
+    assert_eq!(stored_key.oauth_invalid_at_unix_secs, None);
+    assert_eq!(stored_key.oauth_invalid_reason, None);
+    let decrypted_auth_config = decrypt_python_fernet_ciphertext(
+        DEVELOPMENT_ENCRYPTION_KEY,
+        stored_key
+            .encrypted_auth_config
+            .as_deref()
+            .expect("auth config should exist"),
+    )
+    .expect("auth config should decrypt");
+    let auth_config: serde_json::Value =
+        serde_json::from_str(&decrypted_auth_config).expect("auth config should parse");
+    assert_eq!(auth_config["provider_type"], "kiro");
+    assert_eq!(auth_config["auth_method"], "social");
+    assert_eq!(auth_config["email"], "kiro-batch@example.com");
+    assert_eq!(
+        auth_config["refresh_token"],
+        "kiro-batch-refresh-token-replaced"
+    );
+
+    gateway_handle.abort();
+    refresh_handle.abort();
+    upstream_handle.abort();
+}
+
 #[tokio::test]
 async fn gateway_batch_imports_admin_provider_oauth_kiro_via_execution_runtime_proxy_node() {
     let execution_plans = Arc::new(Mutex::new(Vec::<ExecutionPlan>::new()));
@@ -3462,35 +3791,66 @@ async fn gateway_refreshes_admin_provider_oauth_key_locally_with_trusted_admin_p
         .get("oauth")
         .and_then(serde_json::Value::as_object)
         .expect("oauth snapshot should exist");
-    assert_eq!(
-        oauth_snapshot
-            .get("code")
-            .and_then(serde_json::Value::as_str),
-        Some("expiring")
-    );
-    assert_eq!(
-        oauth_snapshot
-            .get("label")
-            .and_then(serde_json::Value::as_str),
-        Some("即将过期")
-    );
     assert_eq!(
         oauth_snapshot.get("expires_at"),
         auth_config.get("expires_at")
     );
-    assert_eq!(oauth_snapshot.get("reason"), Some(&serde_json::Value::Null));
+    if stored_key.oauth_invalid_reason.is_some() {
-    assert_eq!(
+        assert_eq!(
-        oauth_snapshot
+            oauth_snapshot
-            .get("requires_reauth")
+                .get("code")
-            .and_then(serde_json::Value::as_bool),
+                .and_then(serde_json::Value::as_str),
-        Some(false)
+            Some("invalid")
-    );
+        );
-    assert_eq!(
+        assert_eq!(
-        oauth_snapshot
+            oauth_snapshot
-            .get("expiring_soon")
+                .get("label")
-            .and_then(serde_json::Value::as_bool),
+                .and_then(serde_json::Value::as_str),
-        Some(true)
+            Some("已失效")
-    );
+        );
+        assert_eq!(
+            oauth_snapshot.get("reason"),
+            Some(&json!("Codex Token 无效或已过期 (401)"))
+        );
+        assert_eq!(
+            oauth_snapshot
+                .get("requires_reauth")
+                .and_then(serde_json::Value::as_bool),
+            Some(true)
+        );
+        assert_eq!(
+            oauth_snapshot
+                .get("expiring_soon")
+                .and_then(serde_json::Value::as_bool),
+            Some(false)
+        );
+    } else {
+        assert_eq!(
+            oauth_snapshot
+                .get("code")
+                .and_then(serde_json::Value::as_str),
+            Some("expiring")
+        );
+        assert_eq!(
+            oauth_snapshot
+                .get("label")
+                .and_then(serde_json::Value::as_str),
+            Some("即将过期")
+        );
+        assert_eq!(oauth_snapshot.get("reason"), Some(&serde_json::Value::Null));
+        assert_eq!(
+            oauth_snapshot
+                .get("requires_reauth")
+                .and_then(serde_json::Value::as_bool),
+            Some(false)
+        );
+        assert_eq!(
+            oauth_snapshot
+                .get("expiring_soon")
+                .and_then(serde_json::Value::as_bool),
+            Some(true)
+        );
+    }
 
     gateway_handle.abort();
     execution_runtime_handle.abort();
@@ -3498,6 +3858,199 @@ async fn gateway_refreshes_admin_provider_oauth_key_locally_with_trusted_admin_p
     upstream_handle.abort();
 }
 
+#[tokio::test]
+async fn gateway_marks_manual_oauth_refresh_failures_as_invalid_in_pool_payload() {
+    let token_hits = Arc::new(Mutex::new(0usize));
+    let token_hits_clone = Arc::clone(&token_hits);
+    let token_server = Router::new().route(
+        "/oauth/token",
+        post(move |_request: Request| {
+            let token_hits_inner = Arc::clone(&token_hits_clone);
+            async move {
+                *token_hits_inner.lock().expect("mutex should lock") += 1;
+                (
+                    StatusCode::UNAUTHORIZED,
+                    Json(json!({
+                        "error": {
+                            "message": "Your refresh token has already been used to generate a new access token. Please try signing in again.",
+                            "type": "invalid_request_error",
+                            "param": serde_json::Value::Null,
+                            "code": "refresh_token_reused"
+                        }
+                    })),
+                )
+            }
+        }),
+    );
+
+    let mut provider = sample_provider("provider-codex", "codex", 10).with_transport_fields(
+        true,
+        false,
+        true,
+        None,
+        None,
+        None,
+        None,
+        None,
+        Some(json!({
+            "pool_advanced": {
+                "enabled": true
+            }
+        })),
+    );
+    provider.provider_type = "codex".to_string();
+    let endpoint = sample_endpoint(
+        "endpoint-codex-cli",
+        "provider-codex",
+        "openai:cli",
+        "https://chatgpt.com/backend-api/codex",
+    );
+
+    let mut key = sample_key(
+        "key-codex-oauth-refresh-invalid",
+        "provider-codex",
+        "openai:cli",
+        "stale-codex-access-token",
+    );
+    key.auth_type = "oauth".to_string();
+    key.expires_at_unix_secs = Some(1_900_000_000);
+    key.status_snapshot = Some(json!({
+        "oauth": {
+            "code": "valid",
+            "label": "有效",
+            "reason": serde_json::Value::Null,
+            "expires_at": 1_900_000_000u64,
+            "invalid_at": serde_json::Value::Null,
+            "source": "expires_at",
+            "requires_reauth": false,
+            "expiring_soon": false
+        },
+        "account": {
+            "code": "ok",
+            "label": serde_json::Value::Null,
+            "reason": serde_json::Value::Null,
+            "blocked": false,
+            "source": serde_json::Value::Null,
+            "recoverable": false
+        },
+        "quota": {
+            "code": "unknown",
+            "label": serde_json::Value::Null,
+            "reason": serde_json::Value::Null,
+            "exhausted": false,
+            "usage_ratio": serde_json::Value::Null,
+            "updated_at": serde_json::Value::Null,
+            "reset_seconds": serde_json::Value::Null,
+            "plan_type": serde_json::Value::Null
+        }
+    }));
+    key.encrypted_auth_config = Some(
+        encrypt_python_fernet_plaintext(
+            DEVELOPMENT_ENCRYPTION_KEY,
+            r#"{"provider_type":"codex","refresh_token":"used-refresh-token","email":"alice@example.com","account_id":"acct-codex-123","plan_type":"plus","expires_at":1900000000}"#,
+        )
+        .expect("auth config ciphertext should build"),
+    );
+
+    let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
+        vec![provider],
+        vec![endpoint],
+        vec![key],
+    ));
+
+    let (token_url, token_handle) = start_server(token_server).await;
+    let oauth_refresh =
+        crate::provider_transport::LocalOAuthRefreshCoordinator::with_adapters_for_tests(vec![
+            Arc::new(
+                crate::provider_transport::oauth_refresh::GenericOAuthRefreshAdapter::default()
+                    .with_token_url_for_tests("codex", format!("{token_url}/oauth/token")),
+            ),
+        ]);
+    let gateway = build_router_with_state(
+        AppState::new()
+            .expect("gateway should build")
+            .with_data_state_for_tests(
+                GatewayDataState::with_provider_catalog_repository_for_tests(
+                    provider_catalog_repository,
+                )
+                .with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
+            )
+            .with_oauth_refresh_coordinator_for_tests(oauth_refresh),
+    );
+    let (gateway_url, gateway_handle) = start_server(gateway).await;
+
+    let client = reqwest::Client::new();
+    let refresh_response = client
+        .post(format!(
+            "{gateway_url}/api/admin/provider-oauth/keys/key-codex-oauth-refresh-invalid/refresh"
+        ))
+        .header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
+        .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
+        .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
+        .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
+        .send()
+        .await
+        .expect("refresh request should succeed");
+
+    assert_eq!(refresh_response.status(), StatusCode::BAD_REQUEST);
+    let refresh_payload: serde_json::Value = refresh_response
+        .json()
+        .await
+        .expect("refresh payload should parse");
+    assert_eq!(
+        refresh_payload["detail"],
+        json!("Token 刷新失败：refresh_token 已被使用并轮换，请重新登录授权")
+    );
+    assert_eq!(*token_hits.lock().expect("mutex should lock"), 1);
+
+    let pool_response = client
+        .get(format!(
+            "{gateway_url}/api/admin/pool/provider-codex/keys?page=1&page_size=50&status=all"
+        ))
+        .header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
+        .header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
+        .header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
+        .header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
+        .send()
+        .await
+        .expect("pool request should succeed");
+
+    assert_eq!(pool_response.status(), StatusCode::OK);
+    let pool_payload: serde_json::Value = pool_response
+        .json()
+        .await
+        .expect("pool payload should parse");
+    let keys = pool_payload["keys"]
+        .as_array()
+        .expect("keys should be array");
+    assert_eq!(keys.len(), 1);
+    assert_eq!(
+        keys[0]["oauth_invalid_reason"],
+        json!(
+            "[REFRESH_FAILED] Token 续期失败 (401): refresh_token 已被使用并轮换，请重新登录授权"
+        )
+    );
+    assert_eq!(
+        keys[0]["status_snapshot"]["oauth"]["code"],
+        json!("invalid")
+    );
+    assert_eq!(
+        keys[0]["status_snapshot"]["oauth"]["reason"],
+        json!("Token 续期失败 (401): refresh_token 已被使用并轮换，请重新登录授权")
+    );
+    assert_eq!(
+        keys[0]["status_snapshot"]["oauth"]["source"],
+        json!("oauth_refresh")
+    );
+    assert_eq!(
+        keys[0]["status_snapshot"]["oauth"]["requires_reauth"],
+        json!(true)
+    );
+
+    gateway_handle.abort();
+    token_handle.abort();
+}
+
 #[tokio::test]
 async fn gateway_refreshes_admin_provider_oauth_key_locally_via_execution_runtime_provider_proxy_before_system_proxy(
 ) {

apps/aether-gateway/src/tests/usage/local.rs

@@ -34,7 +34,9 @@ where
     T: UsageReadRepository + ?Sized,
 {
     let mut stored = None;
-    let timeout = std::time::Duration::from_secs(10);
+    // Usage terminal events are written on a shared background runtime; under full-suite parallel
+    // load they can lag noticeably behind the request/response assertion path.
+    let timeout = std::time::Duration::from_secs(30);
     let deadline = tokio::time::Instant::now() + timeout;
     loop {
         stored = repository

apps/aether-gateway/src/tests/usage/pricing.rs

@@ -458,7 +458,9 @@ where
     T: UsageReadRepository + ?Sized,
 {
     let mut stored = None;
-    let timeout = std::time::Duration::from_secs(10);
+    // Usage terminal events are written on a shared background runtime; under full-suite parallel
+    // load they can lag noticeably behind the request/response assertion path.
+    let timeout = std::time::Duration::from_secs(30);
     let deadline = tokio::time::Instant::now() + timeout;
     loop {
         stored = repository

crates/aether-data/src/repository/provider_oauth.rs

@@ -87,6 +87,14 @@ pub fn build_provider_oauth_batch_task_status_payload(
         "processed": state.get("processed").and_then(serde_json::Value::as_i64).unwrap_or(0),
         "success": state.get("success").and_then(serde_json::Value::as_i64).unwrap_or(0),
         "failed": state.get("failed").and_then(serde_json::Value::as_i64).unwrap_or(0),
+        "created_count": state
+            .get("created_count")
+            .and_then(serde_json::Value::as_i64)
+            .unwrap_or(0),
+        "replaced_count": state
+            .get("replaced_count")
+            .and_then(serde_json::Value::as_i64)
+            .unwrap_or(0),
         "progress_percent": state
             .get("progress_percent")
             .and_then(serde_json::Value::as_i64)
@@ -154,6 +162,8 @@ mod tests {
             "processed": 2,
             "success": 1,
             "failed": 1,
+            "created_count": 0,
+            "replaced_count": 1,
             "progress_percent": 999,
             "error_samples": [
                 {"detail": "x"},
@@ -180,6 +190,14 @@ mod tests {
             payload.get("progress_percent").and_then(|v| v.as_i64()),
             Some(100)
         );
+        assert_eq!(
+            payload.get("created_count").and_then(|v| v.as_i64()),
+            Some(0)
+        );
+        assert_eq!(
+            payload.get("replaced_count").and_then(|v| v.as_i64()),
+            Some(1)
+        );
         assert_eq!(
             payload
                 .get("error_samples")

frontend/src/api/endpoints/provider_oauth.ts

@@ -28,6 +28,7 @@ export interface ProviderOAuthCompleteResponseWithKey {
   expires_at?: number | null
   has_refresh_token: boolean
   email?: string | null
+  replaced?: boolean
 }
 
 export interface OAuthBatchImportResultItem {
@@ -49,6 +50,8 @@ export interface OAuthBatchImportTaskStartResponse {
   processed: number
   success: number
   failed: number
+  created_count?: number
+  replaced_count?: number
   progress_percent: number
   message?: string | null
 }
@@ -62,6 +65,8 @@ export interface OAuthBatchImportTaskStatusResponse {
   processed: number
   success: number
   failed: number
+  created_count?: number
+  replaced_count?: number
   progress_percent: number
   message?: string | null
   error?: string | null

frontend/src/features/providers/components/OAuthAccountDialog.vue

@@ -717,6 +717,51 @@ function getImportTaskStatusText(status: OAuthBatchImportTaskStatus): string {
   }
 }
 
+function getOAuthSuccessMessage(
+  action: '授权' | '导入',
+  options?: { email?: string | null; replaced?: boolean }
+): string {
+  const email = typeof options?.email === 'string' ? options.email.trim() : ''
+  const replaced = options?.replaced === true
+  if (email) {
+    return replaced
+      ? `${action}成功: ${email}（已替换旧账号）`
+      : `${action}成功: ${email}`
+  }
+  return replaced
+    ? `${action}成功，已替换旧账号`
+    : `${action}成功，账号已添加`
+}
+
+function getBatchImportSuccessMessage(task: OAuthBatchImportTaskStatusResponse): string {
+  const replacedCount = Math.max(task.replaced_count ?? 0, 0)
+  const createdCount = Math.max(task.created_count ?? task.success - replacedCount, 0)
+  const parts: string[] = []
+
+  if (createdCount > 0) {
+    parts.push(`新增 ${createdCount} 个`)
+  }
+  if (replacedCount > 0) {
+    parts.push(`替换 ${replacedCount} 个`)
+  }
+  if (task.failed > 0) {
+    parts.push(`失败 ${task.failed} 个`)
+  }
+
+  if (parts.length === 0) {
+    return task.failed > 0 ? `批量导入完成：失败 ${task.failed} 个` : '批量导入完成'
+  }
+  if (task.failed === 0 && createdCount > 0 && replacedCount === 0) {
+    return `批量导入成功：${createdCount} 个账号已添加`
+  }
+  if (task.failed === 0 && createdCount === 0 && replacedCount > 0) {
+    return `批量导入成功：已替换 ${replacedCount} 个旧账号`
+  }
+
+  const prefix = task.failed > 0 ? '批量导入完成' : '批量导入成功'
+  return `${prefix}：${parts.join('，')}`
+}
+
 function scheduleImportPoll(taskId: string, delayMs = 1200) {
   stopImportPolling()
   importPollTimer = setTimeout(() => {
@@ -736,11 +781,7 @@ async function pollImportTaskStatus(taskId: string) {
       stopImportPolling()
       importing.value = false
       if (task.success > 0) {
-        if (task.failed > 0) {
+        success(getBatchImportSuccessMessage(task))
-          success(`批量导入完成：成功 ${task.success} 个，失败 ${task.failed} 个`)
-        } else {
-          success(`批量导入成功：${task.success} 个账号已添加`)
-        }
         emit('saved')
         handleClose()
       } else {
@@ -863,12 +904,12 @@ async function handleCompleteOAuth() {
   const requestId = ++oauthCompleteRequestId
   oauth.value.completing = true
   try {
-    await completeProviderLevelOAuth(props.providerId, {
+    const result = await completeProviderLevelOAuth(props.providerId, {
       callback_url: oauth.value.callback_url.trim(),
       proxy_node_id: selectedProxyNodeId.value || undefined,
     })
     if (requestId !== oauthCompleteRequestId) return
-    success('授权成功，账号已添加')
+    success(getOAuthSuccessMessage('授权', result))
     emit('saved')
     handleClose()
   } catch (err: unknown) {
@@ -966,6 +1007,8 @@ async function handleImport() {
         processed: task.processed,
         success: task.success,
         failed: task.failed,
+        created_count: task.created_count ?? 0,
+        replaced_count: task.replaced_count ?? 0,
         progress_percent: task.progress_percent,
         message: task.message || null,
         error: null,
@@ -984,11 +1027,11 @@ async function handleImport() {
         showError('无法解析输入内容，请检查格式', '格式错误')
         return
       }
-      await importProviderRefreshToken(props.providerId, {
+      const result = await importProviderRefreshToken(props.providerId, {
         ...parsed,
         proxy_node_id: proxyNodeId,
       })
-      success('导入成功，账号已添加')
+      success(getOAuthSuccessMessage('导入', result))
       emit('saved')
       handleClose()
     }
@@ -1073,7 +1116,7 @@ async function pollDevice() {
         stopDevicePolling()
         totp.stop()
         device.value.status = 'authorized'
-        success(result.email ? `授权成功: ${result.email}` : '授权成功，账号已添加')
+        success(getOAuthSuccessMessage('授权', result))
         emit('saved')
         handleClose()
         return