From 582fa32755647ae9a18d4bdbd56352a6fcb40411 Mon Sep 17 00:00:00 2001 From: lzt404 <2596933790@qq.com> Date: Tue, 11 Aug 2026 23:41:04 +0800 Subject: [PATCH 1/4] feat(usage): migrate legacy records to durable subscription storage - back up and transactionally migrate legacy usage databases - add durable inbox with retry, recovery, and cleanup - prefer RESP usage subscription with HTTP fallback - preserve existing usage APIs, queries, and deduplication behavior - add migration, inbox, and RESP protocol tests --- README.ja.md | 2 + README.md | 2 + README.zh-CN.md | 2 + src-tauri/Cargo.toml | 2 +- src-tauri/src/usage.rs | 1100 +++++++++++++++++++++++++++++++++-- src-tauri/src/usage/resp.rs | 320 ++++++++++ 6 files changed, 1388 insertions(+), 40 deletions(-) create mode 100644 src-tauri/src/usage/resp.rs diff --git a/README.ja.md b/README.ja.md index 2a910f2..436b0c2 100644 --- a/README.ja.md +++ b/README.ja.md @@ -81,6 +81,8 @@ OpenAI、Claude、Gemini、およびその他の互換形式の間で変換で - 時間、モデル、プロバイダー、ソース、キー、結果でデータを絞り込み。 - リクエストと Token の推移、入力・出力・推論・キャッシュの使用量を確認。 - リクエスト詳細、分析画面、価格統計を閲覧。 +- CPA のリアルタイム使用量購読と永続ローカル inbox で収集し、購読できない場合は HTTP に自動フォールバック。 +- 起動時に旧使用履歴データベースを一度だけ移行し、移行前のバックアップを `usage-records/backups` に保存。 ### エージェントクライアント設定 diff --git a/README.md b/README.md index 0c907e3..30cae80 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,8 @@ The Usage page helps you understand local request activity and token consumption - Filter usage by time, model, provider, source, key, and result. - Inspect request/token trends and input, output, reasoning, and cache usage. - Browse request details, analysis views, and price statistics. +- Collect through CPA's real-time usage subscription with a durable local inbox and automatic HTTP fallback. +- Upgrade legacy usage databases once at startup after saving a backup under `usage-records/backups`. ### Agent client configuration diff --git a/README.zh-CN.md b/README.zh-CN.md index 7f2104d..6c3999f 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -79,6 +79,8 @@ API 接入页面按照协议或 Provider 管理上游 API 凭证和服务地址 - 按时间、模型、Provider、来源、密钥和结果筛选数据。 - 查看请求与 Token 趋势，以及输入、输出、思考和缓存用量构成。 - 浏览请求明细、分析视图和价格统计。 +- 通过 CPA 实时订阅和本地持久化 inbox 采集记录，订阅不可用时自动降级到 HTTP。 +- 启动时一次性迁移旧版使用记录数据库，并在 `usage-records/backups` 下保留迁移前备份。 ### 智能体客户端配置 diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 08f3378..72f0aa2 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -29,7 +29,7 @@ tar = "0.4" tauri = { version = "~2.11", features = ["tray-icon"] } tauri-plugin-autostart = "2" tauri-plugin-opener = "2" -tokio = { version = "1", features = ["macros", "sync", "time"] } +tokio = { version = "1", features = ["io-util", "macros", "net", "sync", "time"] } tokio-util = "0.7" toml = "0.8" toml_edit = "0.20" diff --git a/src-tauri/src/usage.rs b/src-tauri/src/usage.rs index e3786b2..384ccf8 100644 --- a/src-tauri/src/usage.rs +++ b/src-tauri/src/usage.rs @@ -1,3 +1,5 @@ +mod resp; + use super::{ apply_configured_proxy, current_core_status, management_authorization, management_endpoint, management_http_client, CoreProcessState, GuiConfigFile, GuiConfigState, @@ -10,26 +12,32 @@ use rusqlite::{ use serde::{Deserialize, Serialize}; use serde_json::Value; use sha2::{Digest, Sha256}; -#[cfg(test)] -use std::time::{SystemTime, UNIX_EPOCH}; use std::{ - collections::HashMap, + collections::{HashMap, HashSet}, fs, path::{Path, PathBuf}, sync::Mutex, - time::Duration, + time::{Duration, SystemTime, UNIX_EPOCH}, }; use tauri::{Emitter, Manager}; use tokio_util::sync::CancellationToken; +use self::resp::UsageSubscription; + const USAGE_DIR_NAME: &str = "usage-records"; const USAGE_DATABASE_FILE: &str = "usage.db"; +const USAGE_BACKUP_DIR_NAME: &str = "backups"; const LEGACY_USAGE_EVENTS_DIR: &str = "events"; const LEGACY_USAGE_INBOX_DIR: &str = "inbox"; const LEGACY_JSON_MIGRATION_KEY: &str = "legacy_json_v1"; +const USAGE_DATABASE_MIGRATION_KEY: &str = "keeper_v3"; const USAGE_UPDATED_EVENT: &str = "usage-records-updated"; const USAGE_SCHEMA_VERSION: u8 = 1; +const USAGE_DATABASE_SCHEMA_VERSION: i64 = 3; const USAGE_QUEUE_BATCH_SIZE: usize = 500; +const USAGE_INBOX_PROCESS_LIMIT: usize = 500; +const USAGE_INBOX_MAX_ATTEMPTS: i64 = 5; +const USAGE_SUBSCRIBE_RETRY_SECONDS: u64 = 30; const SQLITE_BUSY_TIMEOUT_SECONDS: u64 = 5; const TOKENS_PER_PRICE_UNIT: f64 = 1_000_000.0; const LONG_CONTEXT_INPUT_TOKEN_THRESHOLD: u64 = 272_000; @@ -46,6 +54,36 @@ struct UsageCollectorInner { status: UsageCollectorStatus, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum UsageDatabaseLayout { + Empty, + LegacyV2, + CurrentV3, +} + +#[derive(Debug)] +struct UsageInboxRow { + id: i64, + source: String, + raw_message: String, + attempt_count: i64, +} + +#[derive(Debug, Eq, PartialEq)] +struct UsageMigrationSnapshot { + records: i64, + input_tokens: i64, + output_tokens: i64, + reasoning_tokens: i64, + cache_read_tokens: i64, + cache_creation_tokens: i64, + total_tokens: i64, + successes: i64, + failures: i64, + first_timestamp: Option, + last_timestamp: Option, +} + impl Default for UsageCollectorState { fn default() -> Self { Self { @@ -111,10 +149,18 @@ pub(crate) struct UsageRecord { failed: bool, #[serde(default)] provider: String, + #[serde(default, skip_serializing)] + api_group_key: String, #[serde(default)] model: String, #[serde(default)] alias: String, + #[serde(default, skip_serializing)] + client_ip: Option, + #[serde(default, skip_serializing)] + x_forwarded_for: Option, + #[serde(default, skip_serializing)] + user_agent: Option, #[serde(default)] reasoning_effort: String, #[serde(default)] @@ -135,6 +181,12 @@ pub(crate) struct UsageRecord { api_key_remark: String, #[serde(default)] request_id: String, + #[serde(default = "default_usage_generate", skip_serializing)] + generate: bool, + #[serde(default, skip_serializing)] + cached_tokens: u64, + #[serde(default, skip_serializing)] + collector_source: String, #[serde(default)] tokens: UsageTokenStats, } @@ -393,8 +445,207 @@ fn initialize_usage_storage_at(root: &Path) -> Result<(), String> { connection .pragma_update(None, "journal_mode", "WAL") .map_err(|error| format!("启用 SQLite WAL 失败: {error}"))?; - initialize_usage_schema(&connection)?; - migrate_legacy_json_storage(&mut connection, root) + migrate_usage_database(&mut connection, root)?; + migrate_legacy_json_storage(&mut connection, root)?; + cleanup_usage_inbox(&connection, Local::now()) +} + +fn migrate_usage_database(connection: &mut Connection, root: &Path) -> Result<(), String> { + match detect_usage_database_layout(connection)? { + UsageDatabaseLayout::Empty => initialize_usage_schema(connection), + UsageDatabaseLayout::CurrentV3 => initialize_usage_schema(connection), + UsageDatabaseLayout::LegacyV2 => { + let backup_path = create_usage_migration_backup(connection, root)?; + if let Err(error) = migrate_legacy_v2_usage_schema(connection) { + return Err(format!( + "迁移旧版使用记录数据库失败，备份保留在 {}: {error}", + backup_path.display() + )); + } + initialize_usage_schema(connection) + } + } +} + +fn detect_usage_database_layout(connection: &Connection) -> Result { + if !usage_table_exists(connection, "usage_events")? { + return Ok(UsageDatabaseLayout::Empty); + } + let columns = usage_table_columns(connection, "usage_events")?; + let legacy_columns = [ + "event_key", + "timestamp_ms", + "local_hour", + "api_key_hash", + "api_key_display", + "api_key_remark", + ]; + if !legacy_columns + .iter() + .all(|column| columns.contains(*column)) + { + return Err("无法识别使用记录数据库结构，已拒绝自动迁移".to_string()); + } + let keeper_columns = [ + "api_group_key", + "model_alias", + "client_ip", + "x_forwarded_for", + "user_agent", + "generate", + "cached_tokens", + "collector_source", + ]; + let keeper_column_count = keeper_columns + .iter() + .filter(|column| columns.contains(**column)) + .count(); + if keeper_column_count == 0 && !usage_table_exists(connection, "usage_inbox")? { + return Ok(UsageDatabaseLayout::LegacyV2); + } + if keeper_column_count == keeper_columns.len() + && usage_table_exists(connection, "usage_inbox")? + && usage_table_exists(connection, "usage_aggregation_checkpoints")? + { + return Ok(UsageDatabaseLayout::CurrentV3); + } + Err("检测到不完整的使用记录数据库迁移，已拒绝继续修改".to_string()) +} + +fn usage_table_exists(connection: &Connection, table: &str) -> Result { + connection + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1)", + params![table], + |row| row.get::<_, i64>(0), + ) + .map(|exists| exists != 0) + .map_err(|error| format!("检查 SQLite 表 {table} 失败: {error}")) +} + +fn usage_table_columns(connection: &Connection, table: &str) -> Result, String> { + let mut statement = connection + .prepare("SELECT name FROM pragma_table_info(?1)") + .map_err(|error| format!("准备读取 SQLite 表结构失败 {table}: {error}"))?; + let columns = statement + .query_map(params![table], |row| row.get::<_, String>(0)) + .map_err(|error| format!("读取 SQLite 表结构失败 {table}: {error}"))? + .collect::, _>>() + .map_err(|error| format!("解析 SQLite 表结构失败 {table}: {error}"))?; + Ok(columns) +} + +fn create_usage_migration_backup(connection: &Connection, root: &Path) -> Result { + let backup_dir = root.join(USAGE_BACKUP_DIR_NAME); + fs::create_dir_all(&backup_dir) + .map_err(|error| format!("创建使用记录备份目录失败: {error}"))?; + let backup_path = backup_dir.join(format!("usage-before-keeper-v3-{}.db", unique_file_stamp())); + connection + .execute( + "VACUUM INTO ?1", + params![backup_path.to_string_lossy().to_string()], + ) + .map_err(|error| format!("备份旧版使用记录数据库失败: {error}"))?; + Ok(backup_path) +} + +fn migrate_legacy_v2_usage_schema(connection: &mut Connection) -> Result<(), String> { + let before = load_usage_migration_snapshot(connection)?; + let transaction = connection + .transaction() + .map_err(|error| format!("开始使用记录数据库迁移事务失败: {error}"))?; + transaction + .execute_batch( + r#" + ALTER TABLE usage_events ADD COLUMN api_group_key TEXT NOT NULL DEFAULT ''; + ALTER TABLE usage_events ADD COLUMN model_alias TEXT; + ALTER TABLE usage_events ADD COLUMN client_ip TEXT; + ALTER TABLE usage_events ADD COLUMN x_forwarded_for TEXT; + ALTER TABLE usage_events ADD COLUMN user_agent TEXT; + ALTER TABLE usage_events ADD COLUMN generate INTEGER NOT NULL DEFAULT 1; + ALTER TABLE usage_events ADD COLUMN cached_tokens INTEGER NOT NULL DEFAULT 0; + ALTER TABLE usage_events ADD COLUMN collector_source TEXT NOT NULL DEFAULT 'legacy_migration'; + + UPDATE usage_events + SET api_group_key = CASE + WHEN api_key_hash != '' THEN api_key_hash + WHEN provider != '' THEN provider + WHEN endpoint != '' THEN endpoint + ELSE 'unknown' + END, + model_alias = NULLIF(alias, ''), + generate = CASE + WHEN failed = 0 + AND executor_type = 'CodexWebsocketsExecutor' + AND input_tokens = 0 + AND output_tokens = 0 + AND reasoning_tokens = 0 + AND cache_read_tokens = 0 + AND cache_creation_tokens = 0 + AND total_tokens = 0 + THEN 0 ELSE 1 + END, + cached_tokens = cache_read_tokens + cache_creation_tokens, + collector_source = 'legacy_migration'; + "#, + ) + .map_err(|error| format!("转换旧版使用记录字段失败: {error}"))?; + let after = load_usage_migration_snapshot(&transaction)?; + if before != after { + return Err(format!( + "使用记录迁移校验不一致，迁移前 {before:?}，迁移后 {after:?}" + )); + } + initialize_usage_schema(&transaction)?; + transaction + .execute( + "INSERT INTO usage_metadata (key, value) VALUES (?1, ?2) ON CONFLICT(key) DO UPDATE SET value = excluded.value", + params![USAGE_DATABASE_MIGRATION_KEY, Local::now().to_rfc3339()], + ) + .map_err(|error| format!("写入使用记录迁移标记失败: {error}"))?; + transaction + .commit() + .map_err(|error| format!("提交使用记录数据库迁移失败: {error}")) +} + +fn load_usage_migration_snapshot( + connection: &Connection, +) -> Result { + connection + .query_row( + r#" + SELECT + COUNT(*), + COALESCE(SUM(input_tokens), 0), + COALESCE(SUM(output_tokens), 0), + COALESCE(SUM(reasoning_tokens), 0), + COALESCE(SUM(cache_read_tokens), 0), + COALESCE(SUM(cache_creation_tokens), 0), + COALESCE(SUM(total_tokens), 0), + COALESCE(SUM(CASE WHEN failed = 0 THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN failed != 0 THEN 1 ELSE 0 END), 0), + MIN(timestamp), + MAX(timestamp) + FROM usage_events + "#, + [], + |row| { + Ok(UsageMigrationSnapshot { + records: row.get(0)?, + input_tokens: row.get(1)?, + output_tokens: row.get(2)?, + reasoning_tokens: row.get(3)?, + cache_read_tokens: row.get(4)?, + cache_creation_tokens: row.get(5)?, + total_tokens: row.get(6)?, + successes: row.get(7)?, + failures: row.get(8)?, + first_timestamp: row.get(9)?, + last_timestamp: row.get(10)?, + }) + }, + ) + .map_err(|error| format!("读取使用记录迁移校验快照失败: {error}")) } fn initialize_usage_schema(connection: &Connection) -> Result<(), String> { @@ -430,6 +681,14 @@ fn initialize_usage_schema(connection: &Connection) -> Result<(), String> { api_key_display TEXT NOT NULL DEFAULT '', api_key_remark TEXT NOT NULL DEFAULT '', request_id TEXT NOT NULL DEFAULT '', + api_group_key TEXT NOT NULL DEFAULT '', + model_alias TEXT, + client_ip TEXT, + x_forwarded_for TEXT, + user_agent TEXT, + generate INTEGER NOT NULL DEFAULT 1, + cached_tokens INTEGER NOT NULL DEFAULT 0, + collector_source TEXT NOT NULL DEFAULT '', input_tokens INTEGER NOT NULL DEFAULT 0, output_tokens INTEGER NOT NULL DEFAULT 0, reasoning_tokens INTEGER NOT NULL DEFAULT 0, @@ -453,6 +712,31 @@ fn initialize_usage_schema(connection: &Connection) -> Result<(), String> { ON usage_events(api_key_hash, timestamp_ms DESC); CREATE INDEX IF NOT EXISTS idx_usage_events_failed_timestamp ON usage_events(failed, timestamp_ms DESC); + CREATE INDEX IF NOT EXISTS idx_usage_events_api_group_timestamp + ON usage_events(api_group_key, timestamp_ms DESC); + + CREATE TABLE IF NOT EXISTS usage_inbox ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT NOT NULL, + message_hash TEXT NOT NULL, + raw_message TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + attempt_count INTEGER NOT NULL DEFAULT 0, + last_error TEXT NOT NULL DEFAULT '', + usage_event_key TEXT NOT NULL DEFAULT '', + received_at TEXT NOT NULL, + processed_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_usage_inbox_status_id + ON usage_inbox(status, id); + + CREATE TABLE IF NOT EXISTS usage_aggregation_checkpoints ( + name TEXT PRIMARY KEY NOT NULL, + last_usage_event_id INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL + ); CREATE TABLE IF NOT EXISTS model_prices ( model TEXT PRIMARY KEY NOT NULL, @@ -469,11 +753,12 @@ fn initialize_usage_schema(connection: &Connection) -> Result<(), String> { source_model_id TEXT NOT NULL DEFAULT '', updated_at_ms INTEGER NOT NULL DEFAULT 0 ); - - PRAGMA user_version = 2; "#, ) - .map_err(|error| format!("初始化 SQLite 使用记录结构失败: {error}")) + .map_err(|error| format!("初始化 SQLite 使用记录结构失败: {error}"))?; + connection + .pragma_update(None, "user_version", USAGE_DATABASE_SCHEMA_VERSION) + .map_err(|error| format!("更新 SQLite 使用记录版本失败: {error}")) } fn open_usage_database() -> Result { @@ -604,6 +889,9 @@ async fn usage_collector_loop(app: tauri::AppHandle, token: CancellationToken) { }; let mut retry_seconds = 1_u64; + let mut subscription: Option = None; + let mut subscribe_retry_at = tokio::time::Instant::now(); + let mut next_inbox_cleanup_at = tokio::time::Instant::now() + Duration::from_secs(60 * 60); loop { if token.is_cancelled() { return; @@ -617,17 +905,140 @@ async fn usage_collector_loop(app: tauri::AppHandle, token: CancellationToken) { continue; } }; + if tokio::time::Instant::now() >= next_inbox_cleanup_at { + if let Err(error) = open_usage_database_at(&root) + .and_then(|connection| cleanup_usage_inbox(&connection, Local::now())) + { + eprintln!("清理使用记录 inbox 失败: {error}"); + } + next_inbox_cleanup_at = tokio::time::Instant::now() + Duration::from_secs(60 * 60); + } + let recovered = open_usage_database_at(&root) + .and_then(|mut connection| process_usage_inbox(&mut connection, &config)); + match recovered { + Ok(saved) if saved > 0 => { + let collected_at = Local::now().to_rfc3339(); + app.state::() + .increment_total_records(saved); + set_collector_status( + &app, + "collecting", + &format!("已恢复 {saved} 条待处理记录"), + Some(collected_at.clone()), + ); + let _ = app.emit(USAGE_UPDATED_EVENT, collected_at); + retry_seconds = 1; + continue; + } + Ok(_) => {} + Err(error) => { + set_collector_error(&app, error); + wait_or_cancel(&token, retry_seconds).await; + retry_seconds = (retry_seconds * 2).min(10); + continue; + } + } let process_state = app.state::(); let core_running = current_core_status(Some(process_state.inner()), Some(config.port)) .map(|status| status.running) .unwrap_or(false); if !core_running { + subscription = None; set_collector_status(&app, "waiting-core", "等待内核启动", None); retry_seconds = 1; wait_or_cancel(&token, 1).await; continue; } + if subscription.is_none() && tokio::time::Instant::now() >= subscribe_retry_at { + match UsageSubscription::connect(config.port, &config.management_secret_key).await { + Ok(next_subscription) => { + subscription = Some(next_subscription); + set_collector_status(&app, "collecting", "已连接 CPA usage 实时订阅", None); + match backfill_usage_queue(&root, &config).await { + Ok(saved) => { + publish_collected_records( + &app, + saved, + &format!("实时订阅已连接，补录 {saved} 条队列记录"), + ); + retry_seconds = 1; + } + Err(error) => { + set_collector_status( + &app, + "collecting", + &format!("实时订阅已连接，历史队列补录失败: {error}"), + None, + ); + } + } + continue; + } + Err(error) => { + subscribe_retry_at = tokio::time::Instant::now() + + Duration::from_secs(USAGE_SUBSCRIBE_RETRY_SECONDS); + set_collector_status( + &app, + "collecting", + &format!("使用 HTTP 兼容模式采集；实时订阅不可用: {error}"), + None, + ); + } + } + } + + if let Some(active_subscription) = subscription.as_mut() { + let message = tokio::select! { + _ = token.cancelled() => return, + result = tokio::time::timeout( + Duration::from_secs(2), + active_subscription.next_message(), + ) => result, + }; + match message { + Ok(Ok(payload)) => { + match persist_raw_usage_message_from_source( + &root, + "redis_subscribe:usage", + payload, + &config, + ) { + Ok(saved) => { + publish_collected_records( + &app, + saved, + &format!("实时订阅已保存 {saved} 条新记录"), + ); + retry_seconds = 1; + } + Err(error) => { + set_collector_error(&app, error); + wait_or_cancel(&token, retry_seconds).await; + retry_seconds = (retry_seconds * 2).min(10); + } + } + continue; + } + Err(_) => { + set_collector_status(&app, "collecting", "CPA usage 实时订阅采集中", None); + retry_seconds = 1; + continue; + } + Ok(Err(error)) => { + subscription = None; + subscribe_retry_at = tokio::time::Instant::now() + + Duration::from_secs(USAGE_SUBSCRIBE_RETRY_SECONDS); + set_collector_status( + &app, + "collecting", + &format!("实时订阅断开，已切换 HTTP 兼容模式: {error}"), + None, + ); + } + } + } + match fetch_usage_queue(&config).await { Ok(items) if items.is_empty() => { set_collector_status(&app, "collecting", "使用记录采集中", None); @@ -636,20 +1047,11 @@ async fn usage_collector_loop(app: tauri::AppHandle, token: CancellationToken) { } Ok(items) => match persist_queue_items(&root, items, &config) { Ok(saved) => { - let collected_at = Local::now().to_rfc3339(); - if saved > 0 { - app.state::() - .increment_total_records(saved); - } - set_collector_status( + publish_collected_records( &app, - "collecting", - &format!("已保存 {saved} 条新记录"), - Some(collected_at.clone()), + saved, + &format!("HTTP 兼容模式已保存 {saved} 条新记录"), ); - if saved > 0 { - let _ = app.emit(USAGE_UPDATED_EVENT, collected_at); - } retry_seconds = 1; } Err(error) => { @@ -667,6 +1069,43 @@ async fn usage_collector_loop(app: tauri::AppHandle, token: CancellationToken) { } } +async fn backfill_usage_queue(root: &Path, config: &GuiConfigFile) -> Result { + let mut saved_total = 0_usize; + loop { + let items = fetch_usage_queue(config).await?; + let fetched = items.len(); + if fetched == 0 { + return Ok(saved_total); + } + saved_total = saved_total.saturating_add(persist_queue_items_from_source( + root, + "http_backfill", + items, + config, + )?); + if fetched < USAGE_QUEUE_BATCH_SIZE { + return Ok(saved_total); + } + } +} + +fn publish_collected_records(app: &tauri::AppHandle, saved: usize, message: &str) { + let collected_at = Local::now().to_rfc3339(); + if saved > 0 { + app.state::() + .increment_total_records(saved); + } + set_collector_status( + app, + "collecting", + message, + (saved > 0).then_some(collected_at.clone()), + ); + if saved > 0 { + let _ = app.emit(USAGE_UPDATED_EVENT, collected_at); + } +} + async fn wait_or_cancel(token: &CancellationToken, seconds: u64) { tokio::select! { _ = token.cancelled() => {}, @@ -729,18 +1168,268 @@ fn persist_queue_items( items: Vec, config: &GuiConfigFile, ) -> Result { - let records = items + persist_queue_items_from_source(root, "http_pull", items, config) +} + +fn persist_queue_items_from_source( + root: &Path, + source: &str, + items: Vec, + config: &GuiConfigFile, +) -> Result { + let mut connection = open_usage_database_at(root)?; + enqueue_usage_queue_items(&mut connection, source, items)?; + process_usage_inbox(&mut connection, config) +} + +fn enqueue_usage_queue_items( + connection: &mut Connection, + source: &str, + items: Vec, +) -> Result { + let messages = items .into_iter() - .filter(Value::is_object) - .map(|item| normalize_usage_record(item, config)) + .map(|item| { + serde_json::to_string(&item) + .map_err(|error| format!("序列化 CPA 使用记录失败: {error}")) + }) .collect::, _>>()?; + enqueue_usage_raw_messages(connection, source, messages) +} + +fn enqueue_usage_raw_messages( + connection: &mut Connection, + source: &str, + messages: Vec, +) -> Result { + if messages.is_empty() { + return Ok(0); + } + let transaction = connection + .transaction() + .map_err(|error| format!("开始 SQLite 使用记录 inbox 事务失败: {error}"))?; + let mut statement = transaction + .prepare( + r#" + INSERT INTO usage_inbox ( + source, message_hash, raw_message, status, attempt_count, + received_at, created_at, updated_at + ) VALUES (?1, ?2, ?3, 'pending', 0, ?4, ?4, ?4) + "#, + ) + .map_err(|error| format!("准备 SQLite 使用记录 inbox 写入失败: {error}"))?; + let received_at = Local::now().to_rfc3339(); + let mut inserted = 0_usize; + for raw_message in messages { + inserted = inserted.saturating_add( + statement + .execute(params![ + source, + hash_text(&raw_message), + raw_message, + received_at, + ]) + .map_err(|error| format!("写入 SQLite 使用记录 inbox 失败: {error}"))?, + ); + } + drop(statement); + transaction + .commit() + .map_err(|error| format!("提交 SQLite 使用记录 inbox 失败: {error}"))?; + Ok(inserted) +} + +fn persist_raw_usage_message_from_source( + root: &Path, + source: &str, + raw_message: String, + config: &GuiConfigFile, +) -> Result { + let mut connection = open_usage_database_at(root)?; + enqueue_usage_raw_messages(&mut connection, source, vec![raw_message])?; + process_usage_inbox(&mut connection, config) +} + +fn process_usage_inbox( + connection: &mut Connection, + config: &GuiConfigFile, +) -> Result { + let rows = list_processable_usage_inbox(connection, USAGE_INBOX_PROCESS_LIMIT)?; + if rows.is_empty() { + return Ok(0); + } + let mut valid_rows = Vec::with_capacity(rows.len()); + let mut records = Vec::with_capacity(rows.len()); + for row in rows { + let parsed = serde_json::from_str::(&row.raw_message) + .map_err(|error| format!("解析 inbox JSON 失败: {error}")) + .and_then(|value| normalize_usage_record(value, config)); + match parsed { + Ok(mut record) => { + record.collector_source = row.source.clone(); + valid_rows.push(row); + records.push(record); + } + Err(error) => mark_usage_inbox_decode_failed(connection, row.id, &error)?, + } + } if records.is_empty() { return Ok(0); } - let mut connection = open_usage_database_at(root)?; - insert_usage_records(&mut connection, &records) + + let persist_result = (|| -> Result { + let transaction = connection + .transaction() + .map_err(|error| format!("开始 SQLite inbox 处理事务失败: {error}"))?; + let inserted = insert_usage_records_in_transaction(&transaction, &records)?; + let processed_at = Local::now().to_rfc3339(); + for (row, record) in valid_rows.iter().zip(records.iter()) { + transaction + .execute( + r#" + UPDATE usage_inbox + SET status = 'processed', attempt_count = attempt_count + 1, + last_error = '', usage_event_key = ?1, + processed_at = ?2, updated_at = ?2 + WHERE id = ?3 + "#, + params![record.id, processed_at, row.id], + ) + .map_err(|error| format!("标记 SQLite 使用记录 inbox 已处理失败: {error}"))?; + } + transaction + .commit() + .map_err(|error| format!("提交 SQLite inbox 处理事务失败: {error}"))?; + Ok(inserted) + })(); + match persist_result { + Ok(inserted) => Ok(inserted), + Err(error) => { + mark_usage_inbox_process_failed(connection, &valid_rows, &error)?; + Err(error) + } + } } +fn list_processable_usage_inbox( + connection: &Connection, + limit: usize, +) -> Result, String> { + let mut statement = connection + .prepare( + r#" + SELECT id, source, raw_message, attempt_count + FROM usage_inbox + WHERE status IN ('pending', 'process_failed') + AND attempt_count < ?1 + ORDER BY id ASC + LIMIT ?2 + "#, + ) + .map_err(|error| format!("准备读取 SQLite 使用记录 inbox 失败: {error}"))?; + let rows = statement + .query_map( + params![ + USAGE_INBOX_MAX_ATTEMPTS, + limit.min(i64::MAX as usize) as i64 + ], + |row| { + Ok(UsageInboxRow { + id: row.get(0)?, + source: row.get(1)?, + raw_message: row.get(2)?, + attempt_count: row.get(3)?, + }) + }, + ) + .map_err(|error| format!("查询 SQLite 使用记录 inbox 失败: {error}"))? + .collect::, _>>() + .map_err(|error| format!("读取 SQLite 使用记录 inbox 失败: {error}"))?; + Ok(rows) +} + +fn mark_usage_inbox_decode_failed( + connection: &Connection, + id: i64, + error: &str, +) -> Result<(), String> { + let now = Local::now().to_rfc3339(); + connection + .execute( + r#" + UPDATE usage_inbox + SET status = 'decode_failed', attempt_count = attempt_count + 1, + last_error = ?1, processed_at = ?2, updated_at = ?2 + WHERE id = ?3 + "#, + params![bounded_usage_inbox_error(error), now, id], + ) + .map(|_| ()) + .map_err(|update_error| { + format!("标记 SQLite 使用记录 inbox 解码失败时出错: {update_error}") + }) +} + +fn mark_usage_inbox_process_failed( + connection: &Connection, + rows: &[UsageInboxRow], + error: &str, +) -> Result<(), String> { + let now = Local::now().to_rfc3339(); + let error = bounded_usage_inbox_error(error); + for row in rows { + let next_attempt = row.attempt_count.saturating_add(1); + let status = if next_attempt >= USAGE_INBOX_MAX_ATTEMPTS { + "discarded" + } else { + "process_failed" + }; + connection + .execute( + r#" + UPDATE usage_inbox + SET status = ?1, attempt_count = ?2, last_error = ?3, + processed_at = CASE WHEN ?1 = 'discarded' THEN ?4 ELSE NULL END, + updated_at = ?4 + WHERE id = ?5 + "#, + params![status, next_attempt, error, now, row.id], + ) + .map_err(|update_error| { + format!("标记 SQLite 使用记录 inbox 处理失败时出错: {update_error}") + })?; + } + Ok(()) +} + +fn bounded_usage_inbox_error(error: &str) -> String { + error.chars().take(1_000).collect() +} + +fn cleanup_usage_inbox(connection: &Connection, now: DateTime) -> Result<(), String> { + let processed_cutoff = now + .date_naive() + .and_hms_opt(0, 0, 0) + .and_then(|value| value.and_local_timezone(Local).single()) + .unwrap_or(now) + .to_rfc3339(); + let failed_cutoff = (now - chrono::Duration::days(7)).to_rfc3339(); + connection + .execute( + "DELETE FROM usage_inbox WHERE status = 'processed' AND processed_at < ?1", + params![processed_cutoff], + ) + .map_err(|error| format!("清理已处理使用记录 inbox 失败: {error}"))?; + connection + .execute( + "DELETE FROM usage_inbox WHERE status IN ('decode_failed', 'discarded') AND updated_at < ?1", + params![failed_cutoff], + ) + .map_err(|error| format!("清理失败使用记录 inbox 失败: {error}"))?; + Ok(()) +} + +#[cfg(test)] fn insert_usage_records( connection: &mut Connection, records: &[UsageRecord], @@ -770,12 +1459,15 @@ fn insert_usage_records_in_transaction( source, auth_index, failed, provider, model, alias, reasoning_effort, service_tier, response_service_tier, executor_type, endpoint, auth_type, api_key_hash, api_key_display, api_key_remark, request_id, + api_group_key, model_alias, client_ip, x_forwarded_for, user_agent, + generate, cached_tokens, collector_source, input_tokens, output_tokens, reasoning_tokens, cache_read_tokens, cache_creation_tokens, total_tokens, created_at ) VALUES ( ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, - ?21, ?22, ?23, ?24, ?25, ?26, ?27, ?28, ?29 + ?21, ?22, ?23, ?24, ?25, ?26, ?27, ?28, ?29, ?30, + ?31, ?32, ?33, ?34, ?35, ?36, ?37 ) "#, ) @@ -783,6 +1475,28 @@ fn insert_usage_records_in_transaction( let created_at = Local::now().to_rfc3339(); let mut inserted = 0_usize; for record in records { + let api_group_key = if !record.api_group_key.trim().is_empty() { + record.api_group_key.as_str() + } else if !record.api_key_hash.trim().is_empty() { + record.api_key_hash.as_str() + } else if !record.provider.trim().is_empty() { + record.provider.as_str() + } else if !record.endpoint.trim().is_empty() { + record.endpoint.as_str() + } else { + "unknown" + }; + let cached_tokens = record.cached_tokens.max( + record + .tokens + .cache_read_tokens + .saturating_add(record.tokens.cache_creation_tokens), + ); + let collector_source = if record.collector_source.trim().is_empty() { + "legacy_json" + } else { + record.collector_source.as_str() + }; inserted = inserted.saturating_add( statement .execute(params![ @@ -808,6 +1522,18 @@ fn insert_usage_records_in_transaction( record.api_key_display, record.api_key_remark, record.request_id, + api_group_key, + if record.alias.trim().is_empty() { + None::<&str> + } else { + Some(record.alias.as_str()) + }, + record.client_ip, + record.x_forwarded_for, + record.user_agent, + record.generate, + to_sql_i64(cached_tokens), + collector_source, to_sql_i64(record.tokens.input_tokens), to_sql_i64(record.tokens.output_tokens), to_sql_i64(record.tokens.reasoning_tokens), @@ -841,8 +1567,9 @@ fn normalize_usage_record(value: Value, config: &GuiConfigFile) -> Result Result Result bool { + true +} + fn build_usage_filter(query: &UsageQuery) -> UsageSqlFilter { let mut clauses = Vec::::new(); let mut params = Vec::::new(); @@ -1844,6 +2616,8 @@ fn load_usage_events( provider, model, alias, reasoning_effort, service_tier, response_service_tier, executor_type, endpoint, auth_type, api_key_hash, api_key_display, api_key_remark, request_id, + api_group_key, client_ip, x_forwarded_for, user_agent, generate, + cached_tokens, collector_source, input_tokens, output_tokens, reasoning_tokens, cache_read_tokens, cache_creation_tokens, total_tokens FROM usage_events{} @@ -1886,8 +2660,12 @@ fn usage_record_from_row(row: &Row<'_>) -> rusqlite::Result { auth_index: row.get(5)?, failed: row.get::<_, i64>(6)? != 0, provider: row.get(7)?, + api_group_key: row.get(20)?, model: row.get(8)?, alias: row.get(9)?, + client_ip: row.get(21)?, + x_forwarded_for: row.get(22)?, + user_agent: row.get(23)?, reasoning_effort: row.get(10)?, service_tier: row.get(11)?, response_service_tier: row.get(12)?, @@ -1898,13 +2676,16 @@ fn usage_record_from_row(row: &Row<'_>) -> rusqlite::Result { api_key_display: row.get(17)?, api_key_remark: row.get(18)?, request_id: row.get(19)?, + generate: row.get::<_, i64>(24)? != 0, + cached_tokens: from_sql_i64(row.get(25)?), + collector_source: row.get(26)?, tokens: UsageTokenStats { - input_tokens: from_sql_i64(row.get(20)?), - output_tokens: from_sql_i64(row.get(21)?), - reasoning_tokens: from_sql_i64(row.get(22)?), - cache_read_tokens: from_sql_i64(row.get(23)?), - cache_creation_tokens: from_sql_i64(row.get(24)?), - total_tokens: from_sql_i64(row.get(25)?), + input_tokens: from_sql_i64(row.get(27)?), + output_tokens: from_sql_i64(row.get(28)?), + reasoning_tokens: from_sql_i64(row.get(29)?), + cache_read_tokens: from_sql_i64(row.get(30)?), + cache_creation_tokens: from_sql_i64(row.get(31)?), + total_tokens: from_sql_i64(row.get(32)?), }, }) } @@ -1976,7 +2757,6 @@ fn from_sql_i64(value: i64) -> u64 { value.max(0) as u64 } -#[cfg(test)] fn unique_file_stamp() -> u128 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -2101,6 +2881,55 @@ mod tests { open_usage_database_at(root).unwrap() } + fn create_legacy_v2_database(root: &Path) -> Connection { + fs::create_dir_all(root).unwrap(); + let connection = Connection::open(root.join(USAGE_DATABASE_FILE)).unwrap(); + connection + .execute_batch( + r#" + CREATE TABLE usage_metadata ( + key TEXT PRIMARY KEY NOT NULL, + value TEXT NOT NULL + ); + CREATE TABLE usage_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_key TEXT NOT NULL UNIQUE, + timestamp TEXT NOT NULL, + timestamp_ms INTEGER NOT NULL, + local_hour TEXT NOT NULL, + latency_ms INTEGER NOT NULL DEFAULT 0, + ttft_ms INTEGER, + source TEXT NOT NULL DEFAULT '', + auth_index TEXT NOT NULL DEFAULT '', + failed INTEGER NOT NULL DEFAULT 0, + provider TEXT NOT NULL DEFAULT '', + model TEXT NOT NULL DEFAULT '', + alias TEXT NOT NULL DEFAULT '', + reasoning_effort TEXT NOT NULL DEFAULT '', + service_tier TEXT NOT NULL DEFAULT '', + response_service_tier TEXT NOT NULL DEFAULT '', + executor_type TEXT NOT NULL DEFAULT '', + endpoint TEXT NOT NULL DEFAULT '', + auth_type TEXT NOT NULL DEFAULT '', + api_key_hash TEXT NOT NULL DEFAULT '', + api_key_display TEXT NOT NULL DEFAULT '', + api_key_remark TEXT NOT NULL DEFAULT '', + request_id TEXT NOT NULL DEFAULT '', + input_tokens INTEGER NOT NULL DEFAULT 0, + output_tokens INTEGER NOT NULL DEFAULT 0, + reasoning_tokens INTEGER NOT NULL DEFAULT 0, + cache_read_tokens INTEGER NOT NULL DEFAULT 0, + cache_creation_tokens INTEGER NOT NULL DEFAULT 0, + total_tokens INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL + ); + PRAGMA user_version = 2; + "#, + ) + .unwrap(); + connection + } + fn sample_record(id: &str, timestamp: &str, model: &str) -> UsageRecord { UsageRecord { id: id.to_string(), @@ -2112,8 +2941,12 @@ mod tests { auth_index: "auth".to_string(), failed: false, provider: "openai".to_string(), + api_group_key: "hash".to_string(), model: model.to_string(), alias: String::new(), + client_ip: None, + x_forwarded_for: None, + user_agent: None, reasoning_effort: "high".to_string(), service_tier: String::new(), response_service_tier: String::new(), @@ -2124,6 +2957,9 @@ mod tests { api_key_display: "12••••".to_string(), api_key_remark: "内置密钥".to_string(), request_id: id.to_string(), + generate: true, + cached_tokens: 2, + collector_source: "test".to_string(), tokens: UsageTokenStats { input_tokens: 10, output_tokens: 20, @@ -2317,6 +3153,192 @@ mod tests { fs::remove_dir_all(root).unwrap(); } + #[test] + fn legacy_v2_database_is_backed_up_and_migrated_once() { + let root = test_root("keeper-v3-migration"); + let connection = create_legacy_v2_database(&root); + connection + .execute( + r#" + INSERT INTO usage_events ( + event_key, timestamp, timestamp_ms, local_hour, latency_ms, ttft_ms, + source, auth_index, failed, provider, model, alias, reasoning_effort, + service_tier, response_service_tier, executor_type, endpoint, auth_type, + api_key_hash, api_key_display, api_key_remark, request_id, + input_tokens, output_tokens, reasoning_tokens, cache_read_tokens, + cache_creation_tokens, total_tokens, created_at + ) VALUES ( + 'request-1', '2026-07-17T20:30:00+08:00', 1784291400000, + '2026-07-17-20', 100, 20, 'source', 'auth', 0, 'openai', + 'gpt-test', 'alias-test', 'high', '', '', '', + 'POST /v1/responses', 'oauth', 'legacy-hash', '12••••', + '旧密钥', 'request-1', 10, 20, 5, 2, 3, 30, + '2026-07-17T20:30:01+08:00' + ) + "#, + [], + ) + .unwrap(); + drop(connection); + + initialize_usage_storage_at(&root).unwrap(); + initialize_usage_storage_at(&root).unwrap(); + let connection = open_usage_database_at(&root).unwrap(); + let user_version = connection + .query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0)) + .unwrap(); + let migrated = connection + .query_row( + r#" + SELECT COUNT(*), SUM(total_tokens), api_group_key, model_alias, + cached_tokens, collector_source + FROM usage_events + "#, + [], + |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, i64>(1)?, + row.get::<_, String>(2)?, + row.get::<_, Option>(3)?, + row.get::<_, i64>(4)?, + row.get::<_, String>(5)?, + )) + }, + ) + .unwrap(); + let marker = connection + .query_row( + "SELECT value FROM usage_metadata WHERE key = ?1", + params![USAGE_DATABASE_MIGRATION_KEY], + |row| row.get::<_, String>(0), + ) + .unwrap(); + let backups = fs::read_dir(root.join(USAGE_BACKUP_DIR_NAME)) + .unwrap() + .collect::, _>>() + .unwrap(); + + assert_eq!(user_version, USAGE_DATABASE_SCHEMA_VERSION); + assert_eq!(migrated.0, 1); + assert_eq!(migrated.1, 30); + assert_eq!(migrated.2, "legacy-hash"); + assert_eq!(migrated.3.as_deref(), Some("alias-test")); + assert_eq!(migrated.4, 5); + assert_eq!(migrated.5, "legacy_migration"); + assert!(!marker.is_empty()); + assert_eq!(backups.len(), 1); + drop(connection); + fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn durable_inbox_recovers_valid_rows_and_isolates_malformed_rows() { + let root = test_root("durable-inbox"); + initialize_usage_storage_at(&root).unwrap(); + let mut connection = open_usage_database_at(&root).unwrap(); + enqueue_usage_queue_items( + &mut connection, + "http_pull", + vec![ + serde_json::json!({ + "timestamp": "2026-07-17T20:30:00+08:00", + "request_id": "request-1", + "model": "gpt-test", + "tokens": { "input_tokens": 10, "output_tokens": 20 } + }), + serde_json::json!("not-an-object"), + ], + ) + .unwrap(); + drop(connection); + + let mut connection = open_usage_database_at(&root).unwrap(); + let inserted = process_usage_inbox(&mut connection, &GuiConfigFile::default()).unwrap(); + let event_count = connection + .query_row("SELECT COUNT(*) FROM usage_events", [], |row| { + row.get::<_, i64>(0) + }) + .unwrap(); + let processed_count = connection + .query_row( + "SELECT COUNT(*) FROM usage_inbox WHERE status = 'processed'", + [], + |row| row.get::<_, i64>(0), + ) + .unwrap(); + let failed_count = connection + .query_row( + "SELECT COUNT(*) FROM usage_inbox WHERE status = 'decode_failed'", + [], + |row| row.get::<_, i64>(0), + ) + .unwrap(); + + assert_eq!(inserted, 1); + assert_eq!(event_count, 1); + assert_eq!(processed_count, 1); + assert_eq!(failed_count, 1); + drop(connection); + fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn inbox_cleanup_preserves_pending_and_recent_failure_rows() { + let root = test_root("inbox-cleanup"); + let connection = open_test_database(&root); + let now = Local::now(); + let old_processed = (now - chrono::Duration::days(1)).to_rfc3339(); + let recent_failure = (now - chrono::Duration::days(1)).to_rfc3339(); + let old_failure = (now - chrono::Duration::days(8)).to_rfc3339(); + for (status, processed_at, updated_at) in [ + ( + "processed", + Some(old_processed.as_str()), + old_processed.as_str(), + ), + ("pending", None, old_failure.as_str()), + ( + "decode_failed", + Some(recent_failure.as_str()), + recent_failure.as_str(), + ), + ( + "discarded", + Some(old_failure.as_str()), + old_failure.as_str(), + ), + ] { + connection + .execute( + r#" + INSERT INTO usage_inbox ( + source, message_hash, raw_message, status, attempt_count, + received_at, processed_at, created_at, updated_at + ) VALUES ('test', 'hash', '{}', ?1, 0, ?3, ?2, ?3, ?3) + "#, + params![status, processed_at, updated_at], + ) + .unwrap(); + } + + cleanup_usage_inbox(&connection, now).unwrap(); + let statuses = { + let mut statement = connection + .prepare("SELECT status FROM usage_inbox ORDER BY id") + .unwrap(); + statement + .query_map([], |row| row.get::<_, String>(0)) + .unwrap() + .collect::, _>>() + .unwrap() + }; + + assert_eq!(statuses, vec!["pending", "decode_failed"]); + drop(connection); + fs::remove_dir_all(root).unwrap(); + } + #[test] fn sqlite_storage_deduplicates_event_keys_transactionally() { let root = test_root("sqlite-deduplicate"); diff --git a/src-tauri/src/usage/resp.rs b/src-tauri/src/usage/resp.rs new file mode 100644 index 0000000..83b39bc --- /dev/null +++ b/src-tauri/src/usage/resp.rs @@ -0,0 +1,320 @@ +use std::time::Duration; + +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpStream, + time::timeout, +}; + +const USAGE_CHANNEL: &str = "usage"; +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); +const IO_TIMEOUT: Duration = Duration::from_secs(10); +const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024; +const MAX_ARRAY_LENGTH: usize = 10_000; +const MAX_NESTING_DEPTH: usize = 8; + +pub(super) struct UsageSubscription { + stream: TcpStream, + read_buffer: Vec, +} + +impl UsageSubscription { + pub(super) async fn connect(port: u16, management_key: &str) -> Result { + let address = format!("127.0.0.1:{port}"); + let stream = timeout(CONNECT_TIMEOUT, TcpStream::connect(&address)) + .await + .map_err(|_| format!("连接 CPA usage 订阅超时: {address}"))? + .map_err(|error| format!("连接 CPA usage 订阅失败 {address}: {error}"))?; + let mut subscription = Self { + stream, + read_buffer: Vec::new(), + }; + subscription.send_command(&["AUTH", management_key]).await?; + match subscription.read_frame().await? { + RespValue::Simple(value) if value.eq_ignore_ascii_case("OK") => {} + RespValue::Error(error) => return Err(format!("CPA usage 订阅认证失败: {error}")), + value => return Err(format!("CPA usage 订阅认证响应无效: {}", value.kind())), + } + subscription + .send_command(&["SUBSCRIBE", USAGE_CHANNEL]) + .await?; + let acknowledgement = subscription.read_frame().await?; + if !is_subscription_ack(&acknowledgement) { + return Err("CPA 未确认 usage 订阅".to_string()); + } + Ok(subscription) + } + + pub(super) async fn next_message(&mut self) -> Result { + loop { + let frame = self.read_frame().await?; + if let Some(payload) = subscription_payload(&frame) { + return Ok(payload); + } + if let RespValue::Error(error) = frame { + return Err(format!("CPA usage 订阅返回错误: {error}")); + } + } + } + + async fn send_command(&mut self, parts: &[&str]) -> Result<(), String> { + let mut command = format!("*{}\r\n", parts.len()).into_bytes(); + for part in parts { + command.extend_from_slice(format!("${}\r\n", part.len()).as_bytes()); + command.extend_from_slice(part.as_bytes()); + command.extend_from_slice(b"\r\n"); + } + timeout(IO_TIMEOUT, self.stream.write_all(&command)) + .await + .map_err(|_| "写入 CPA usage 订阅命令超时".to_string())? + .map_err(|error| format!("写入 CPA usage 订阅命令失败: {error}")) + } + + async fn read_frame(&mut self) -> Result { + loop { + match parse_resp_frame(&self.read_buffer, 0, 0)? { + ParseResult::Complete(value, consumed) => { + self.read_buffer.drain(..consumed); + return Ok(value); + } + ParseResult::Incomplete => {} + } + if self.read_buffer.len() >= MAX_FRAME_BYTES { + return Err("CPA usage 订阅响应超过大小限制".to_string()); + } + let mut chunk = [0_u8; 8192]; + let read = timeout(IO_TIMEOUT, self.stream.read(&mut chunk)) + .await + .map_err(|_| "读取 CPA usage 订阅响应超时".to_string())? + .map_err(|error| format!("读取 CPA usage 订阅响应失败: {error}"))?; + if read == 0 { + return Err("CPA usage 订阅连接已关闭".to_string()); + } + self.read_buffer.extend_from_slice(&chunk[..read]); + } + } +} + +#[derive(Debug)] +enum RespValue { + Simple(String), + Error(String), + Integer(i64), + Bulk(Option>), + Array(Option>), +} + +impl RespValue { + fn kind(&self) -> &'static str { + match self { + Self::Simple(_) => "simple string", + Self::Error(_) => "error", + Self::Integer(_) => "integer", + Self::Bulk(_) => "bulk string", + Self::Array(_) => "array", + } + } + + fn text(&self) -> Option { + match self { + Self::Simple(value) | Self::Error(value) => Some(value.clone()), + Self::Bulk(Some(value)) => String::from_utf8(value.clone()).ok(), + Self::Integer(value) => Some(value.to_string()), + Self::Bulk(None) | Self::Array(_) => None, + } + } +} + +enum ParseResult { + Complete(RespValue, usize), + Incomplete, +} + +fn parse_resp_frame(input: &[u8], offset: usize, depth: usize) -> Result { + if depth > MAX_NESTING_DEPTH { + return Err("CPA usage 订阅响应嵌套过深".to_string()); + } + let Some(prefix) = input.get(offset).copied() else { + return Ok(ParseResult::Incomplete); + }; + match prefix { + b'+' | b'-' | b':' => { + let Some((line, next)) = resp_line(input, offset + 1) else { + return Ok(ParseResult::Incomplete); + }; + let text = String::from_utf8(line.to_vec()) + .map_err(|_| "CPA usage 订阅文本响应不是 UTF-8".to_string())?; + let value = match prefix { + b'+' => RespValue::Simple(text), + b'-' => RespValue::Error(text), + _ => RespValue::Integer( + text.parse::() + .map_err(|_| "CPA usage 订阅整数响应无效".to_string())?, + ), + }; + Ok(ParseResult::Complete(value, next - offset)) + } + b'$' => { + let Some((line, data_start)) = resp_line(input, offset + 1) else { + return Ok(ParseResult::Incomplete); + }; + let length = parse_resp_length(line)?; + if length < 0 { + return Ok(ParseResult::Complete( + RespValue::Bulk(None), + data_start - offset, + )); + } + let length = + usize::try_from(length).map_err(|_| "CPA usage 订阅字符串长度无效".to_string())?; + if length > MAX_FRAME_BYTES { + return Err("CPA usage 订阅字符串超过大小限制".to_string()); + } + let data_end = data_start.saturating_add(length); + let frame_end = data_end.saturating_add(2); + if input.len() < frame_end { + return Ok(ParseResult::Incomplete); + } + if input.get(data_end..frame_end) != Some(b"\r\n") { + return Err("CPA usage 订阅字符串结尾无效".to_string()); + } + Ok(ParseResult::Complete( + RespValue::Bulk(Some(input[data_start..data_end].to_vec())), + frame_end - offset, + )) + } + b'*' => { + let Some((line, mut next)) = resp_line(input, offset + 1) else { + return Ok(ParseResult::Incomplete); + }; + let length = parse_resp_length(line)?; + if length < 0 { + return Ok(ParseResult::Complete(RespValue::Array(None), next - offset)); + } + let length = + usize::try_from(length).map_err(|_| "CPA usage 订阅数组长度无效".to_string())?; + if length > MAX_ARRAY_LENGTH { + return Err("CPA usage 订阅数组超过大小限制".to_string()); + } + let mut values = Vec::with_capacity(length); + for _ in 0..length { + match parse_resp_frame(input, next, depth + 1)? { + ParseResult::Complete(value, consumed) => { + values.push(value); + next = next.saturating_add(consumed); + } + ParseResult::Incomplete => return Ok(ParseResult::Incomplete), + } + } + Ok(ParseResult::Complete( + RespValue::Array(Some(values)), + next - offset, + )) + } + _ => Err("CPA usage 订阅响应类型无效".to_string()), + } +} + +fn resp_line(input: &[u8], start: usize) -> Option<(&[u8], usize)> { + let relative_end = input + .get(start..)? + .windows(2) + .position(|pair| pair == b"\r\n")?; + let end = start + relative_end; + Some((&input[start..end], end + 2)) +} + +fn parse_resp_length(value: &[u8]) -> Result { + std::str::from_utf8(value) + .map_err(|_| "CPA usage 订阅长度不是 UTF-8".to_string())? + .parse::() + .map_err(|_| "CPA usage 订阅长度无效".to_string()) +} + +fn is_subscription_ack(value: &RespValue) -> bool { + let RespValue::Array(Some(values)) = value else { + return false; + }; + values.len() >= 2 + && values[0] + .text() + .is_some_and(|value| value.eq_ignore_ascii_case("subscribe")) + && values[1].text().as_deref() == Some(USAGE_CHANNEL) +} + +fn subscription_payload(value: &RespValue) -> Option { + let RespValue::Array(Some(values)) = value else { + return None; + }; + if values.len() < 3 + || !values[0] + .text() + .is_some_and(|value| value.eq_ignore_ascii_case("message")) + || values[1].text().as_deref() != Some(USAGE_CHANNEL) + { + return None; + } + values[2].text() +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::net::TcpListener; + + #[test] + fn parses_usage_subscription_message() { + let input = b"*3\r\n$7\r\nmessage\r\n$5\r\nusage\r\n$16\r\n{\"request_id\":1}\r\n"; + let ParseResult::Complete(value, consumed) = parse_resp_frame(input, 0, 0).unwrap() else { + panic!("expected complete RESP frame"); + }; + assert_eq!(consumed, input.len()); + assert_eq!( + subscription_payload(&value).as_deref(), + Some("{\"request_id\":1}") + ); + } + + #[test] + fn rejects_oversized_arrays() { + let error = match parse_resp_frame(b"*10001\r\n", 0, 0) { + Err(error) => error, + _ => panic!("expected oversized RESP array to fail"), + }; + assert!(error.contains("数组超过大小限制")); + } + + #[tokio::test] + async fn authenticates_subscribes_and_receives_usage_payload() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let expected_auth = b"*2\r\n$4\r\nAUTH\r\n$6\r\nsecret\r\n"; + let mut auth = vec![0_u8; expected_auth.len()]; + stream.read_exact(&mut auth).await.unwrap(); + assert_eq!(auth, expected_auth); + stream.write_all(b"+OK\r\n").await.unwrap(); + + let expected_subscribe = b"*2\r\n$9\r\nSUBSCRIBE\r\n$5\r\nusage\r\n"; + let mut subscribe = vec![0_u8; expected_subscribe.len()]; + stream.read_exact(&mut subscribe).await.unwrap(); + assert_eq!(subscribe, expected_subscribe); + stream + .write_all(b"*3\r\n$9\r\nsubscribe\r\n$5\r\nusage\r\n:1\r\n") + .await + .unwrap(); + stream + .write_all(b"*3\r\n$7\r\nmessage\r\n$5\r\nusage\r\n$16\r\n{\"request_id\":1}\r\n") + .await + .unwrap(); + }); + + let mut subscription = UsageSubscription::connect(port, "secret").await.unwrap(); + assert_eq!( + subscription.next_message().await.unwrap(), + "{\"request_id\":1}" + ); + server.await.unwrap(); + } +} From 74be9e85a1b8d757bb053e69d9d748c490f9b5a8 Mon Sep 17 00:00:00 2001 From: lzt404 <2596933790@qq.com> Date: Wed, 12 Aug 2026 00:31:20 +0800 Subject: [PATCH 2/4] feat(config): add retry controls for long-running requests --- src-tauri/src/core_config/commands.rs | 12 +- src-tauri/src/core_config/settings.rs | 85 +++++++++++- src-tauri/src/core_config/yaml.rs | 40 ++++++ src-tauri/src/main.rs | 65 ++++++++-- src-tauri/src/tests/app_settings.rs | 4 + src-tauri/src/tests/core_config.rs | 58 +++++++++ src-tauri/src/usage.rs | 4 + src/i18n/ja.ts | 10 ++ src/i18n/locales/en.ts | 10 ++ src/i18n/locales/zh-CN.ts | 10 ++ src/pages/ConfigPanel.tsx | 180 +++++++++++++++++++++++++- 11 files changed, 455 insertions(+), 23 deletions(-) diff --git a/src-tauri/src/core_config/commands.rs b/src-tauri/src/core_config/commands.rs index 2034662..4820b8d 100644 --- a/src-tauri/src/core_config/commands.rs +++ b/src-tauri/src/core_config/commands.rs @@ -103,16 +103,14 @@ pub(crate) fn save_network_routing_settings( next.proxy_url = proxy_url.clone(); next.routing_session_affinity = settings.routing_session_affinity; next.routing_session_affinity_ttl = routing_session_affinity_ttl.clone(); + next.request_retry = settings.request_retry; + next.max_retry_credentials = settings.max_retry_credentials; + next.max_retry_interval = settings.max_retry_interval; + next.streaming_bootstrap_retries = settings.streaming_bootstrap_retries; validate_gui_config(&next)?; patch_core_network_routing_settings(&next)?; - let config = match gui_config_state.update_network_routing( - settings.port, - settings.allow_lan, - &proxy_url, - settings.routing_session_affinity, - &routing_session_affinity_ttl, - ) { + let config = match gui_config_state.update_network_routing(&next) { Ok(config) => config, Err(error) => { let rollback_error = patch_core_network_routing_settings(&previous).err(); diff --git a/src-tauri/src/core_config/settings.rs b/src-tauri/src/core_config/settings.rs index ad7a879..7ffb6a2 100644 --- a/src-tauri/src/core_config/settings.rs +++ b/src-tauri/src/core_config/settings.rs @@ -341,6 +341,31 @@ pub(crate) fn patch_core_network_routing_yaml( "session-affinity-ttl", serde_norway::Value::String(config.routing_session_affinity_ttl.clone()), )?; + set_core_yaml_top_level_value( + document, + "request-retry", + serde_norway::to_value(config.request_retry) + .map_err(|err| format!("序列化请求重试次数失败: {err}"))?, + )?; + set_core_yaml_top_level_value( + document, + "max-retry-credentials", + serde_norway::to_value(config.max_retry_credentials) + .map_err(|err| format!("序列化最大重试凭据数失败: {err}"))?, + )?; + set_core_yaml_top_level_value( + document, + "max-retry-interval", + serde_norway::to_value(config.max_retry_interval) + .map_err(|err| format!("序列化最大重试等待时间失败: {err}"))?, + )?; + set_core_yaml_nested_value( + document, + "streaming", + "bootstrap-retries", + serde_norway::to_value(config.streaming_bootstrap_retries) + .map_err(|err| format!("序列化流式启动重试次数失败: {err}"))?, + )?; Ok(*document != original) }) } @@ -446,6 +471,31 @@ pub(crate) fn apply_gui_managed_settings( "session-affinity-ttl", serde_norway::Value::String(config.routing_session_affinity_ttl.clone()), )?; + changed |= set_core_yaml_top_level_value( + document, + "request-retry", + serde_norway::to_value(config.request_retry) + .map_err(|err| format!("序列化请求重试次数失败: {err}"))?, + )?; + changed |= set_core_yaml_top_level_value( + document, + "max-retry-credentials", + serde_norway::to_value(config.max_retry_credentials) + .map_err(|err| format!("序列化最大重试凭据数失败: {err}"))?, + )?; + changed |= set_core_yaml_top_level_value( + document, + "max-retry-interval", + serde_norway::to_value(config.max_retry_interval) + .map_err(|err| format!("序列化最大重试等待时间失败: {err}"))?, + )?; + changed |= set_core_yaml_nested_value( + document, + "streaming", + "bootstrap-retries", + serde_norway::to_value(config.streaming_bootstrap_retries) + .map_err(|err| format!("序列化流式启动重试次数失败: {err}"))?, + )?; Ok(changed) })? .unwrap_or_else(|| content.to_string()); @@ -711,7 +761,11 @@ pub(crate) fn load_or_create_gui_config() -> Result { || presence.routing_strategy.is_none() || presence.proxy_url.is_none() || presence.routing_session_affinity.is_none() - || presence.routing_session_affinity_ttl.is_none(); + || presence.routing_session_affinity_ttl.is_none() + || presence.request_retry.is_none() + || presence.max_retry_credentials.is_none() + || presence.max_retry_interval.is_none() + || presence.streaming_bootstrap_retries.is_none(); if missing_core_settings && !gui_parse_failed { if let Ok(core_settings) = read_installed_core_config_settings() { if presence.host.is_none() { @@ -759,6 +813,18 @@ pub(crate) fn load_or_create_gui_config() -> Result { if presence.routing_session_affinity_ttl.is_none() { config.routing_session_affinity_ttl = core_settings.routing_session_affinity_ttl; } + if presence.request_retry.is_none() { + config.request_retry = core_settings.request_retry; + } + if presence.max_retry_credentials.is_none() { + config.max_retry_credentials = core_settings.max_retry_credentials; + } + if presence.max_retry_interval.is_none() { + config.max_retry_interval = core_settings.max_retry_interval; + } + if presence.streaming_bootstrap_retries.is_none() { + config.streaming_bootstrap_retries = core_settings.streaming_bootstrap_retries; + } } changed = true; } @@ -828,6 +894,10 @@ pub(crate) fn apply_core_settings_to_gui_config( config.proxy_url = core_settings.proxy_url.clone(); config.routing_session_affinity = core_settings.routing_session_affinity; config.routing_session_affinity_ttl = core_settings.routing_session_affinity_ttl.clone(); + config.request_retry = core_settings.request_retry; + config.max_retry_credentials = core_settings.max_retry_credentials; + config.max_retry_interval = core_settings.max_retry_interval; + config.streaming_bootstrap_retries = core_settings.streaming_bootstrap_retries; } pub(crate) fn default_api_key_entry() -> GuiApiKeyEntry { @@ -1130,6 +1200,19 @@ pub(crate) fn write_gui_config_to_path( "routing-session-affinity-ttl", value(config.routing_session_affinity_ttl.as_str()), ), + ("request-retry", value(i64::from(config.request_retry))), + ( + "max-retry-credentials", + value(i64::from(config.max_retry_credentials)), + ), + ( + "max-retry-interval", + value(i64::from(config.max_retry_interval)), + ), + ( + "streaming-bootstrap-retries", + value(i64::from(config.streaming_bootstrap_retries)), + ), ] { set_codex_table_item(root, key, item); } diff --git a/src-tauri/src/core_config/yaml.rs b/src-tauri/src/core_config/yaml.rs index 9f0974f..48934d6 100644 --- a/src-tauri/src/core_config/yaml.rs +++ b/src-tauri/src/core_config/yaml.rs @@ -1056,6 +1056,42 @@ pub(crate) fn core_config_settings_from_value( }) .transpose()? .unwrap_or_default(); + let request_retry = yaml_mapping_value(root, "request-retry") + .map(|value| { + value + .as_u64() + .and_then(|value| u32::try_from(value).ok()) + .ok_or_else(|| "request-retry 必须是非负整数".to_string()) + }) + .transpose()? + .unwrap_or(DEFAULT_REQUEST_RETRY); + let max_retry_credentials = yaml_mapping_value(root, "max-retry-credentials") + .map(|value| { + value + .as_u64() + .and_then(|value| u32::try_from(value).ok()) + .ok_or_else(|| "max-retry-credentials 必须是非负整数".to_string()) + }) + .transpose()? + .unwrap_or(DEFAULT_MAX_RETRY_CREDENTIALS); + let max_retry_interval = yaml_mapping_value(root, "max-retry-interval") + .map(|value| { + value + .as_u64() + .and_then(|value| u32::try_from(value).ok()) + .ok_or_else(|| "max-retry-interval 必须是非负整数".to_string()) + }) + .transpose()? + .unwrap_or(DEFAULT_MAX_RETRY_INTERVAL); + let streaming_bootstrap_retries = nested_yaml_value(root, &["streaming", "bootstrap-retries"]) + .map(|value| { + value + .as_u64() + .and_then(|value| u32::try_from(value).ok()) + .ok_or_else(|| "streaming.bootstrap-retries 必须是非负整数".to_string()) + }) + .transpose()? + .unwrap_or(DEFAULT_STREAMING_BOOTSTRAP_RETRIES); Ok(CoreConfigSettings { host, @@ -1071,6 +1107,10 @@ pub(crate) fn core_config_settings_from_value( proxy_url, routing_session_affinity, routing_session_affinity_ttl, + request_retry, + max_retry_credentials, + max_retry_interval, + streaming_bootstrap_retries, management_secret_key, }) } diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs index ff8b9a5..774e9e8 100644 --- a/src-tauri/src/main.rs +++ b/src-tauri/src/main.rs @@ -120,6 +120,10 @@ const OAUTH_DIR_NAME: &str = "oauth"; const DEFAULT_AUTH_DIR: &str = "../oauth"; const DEFAULT_API_KEY: &str = "123456"; const DEFAULT_API_KEY_INITIAL_REMARK: &str = "默认密钥"; +const DEFAULT_REQUEST_RETRY: u32 = 3; +const DEFAULT_MAX_RETRY_CREDENTIALS: u32 = 0; +const DEFAULT_MAX_RETRY_INTERVAL: u32 = 30; +const DEFAULT_STREAMING_BOOTSTRAP_RETRIES: u32 = 0; const LEGACY_DEFAULT_MANAGEMENT_SECRET_KEY: &str = "123456"; const MANAGED_AGENT_PROVIDER_ID: &str = "cpa-gui"; const PI_AGENT_ID: &str = "pi"; @@ -561,6 +565,10 @@ struct GuiConfigFile { proxy_url: String, routing_session_affinity: bool, routing_session_affinity_ttl: String, + request_retry: u32, + max_retry_credentials: u32, + max_retry_interval: u32, + streaming_bootstrap_retries: u32, } #[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize)] @@ -668,6 +676,10 @@ impl Default for GuiConfigFile { proxy_url: String::new(), routing_session_affinity: false, routing_session_affinity_ttl: String::new(), + request_retry: DEFAULT_REQUEST_RETRY, + max_retry_credentials: DEFAULT_MAX_RETRY_CREDENTIALS, + max_retry_interval: DEFAULT_MAX_RETRY_INTERVAL, + streaming_bootstrap_retries: DEFAULT_STREAMING_BOOTSTRAP_RETRIES, } } } @@ -691,6 +703,10 @@ struct GuiConfigPresence { proxy_url: Option, routing_session_affinity: Option, routing_session_affinity_ttl: Option, + request_retry: Option, + max_retry_credentials: Option, + max_retry_interval: Option, + streaming_bootstrap_retries: Option, } #[derive(Clone, Serialize)] @@ -1043,6 +1059,10 @@ struct GuiNetworkRoutingSettings { proxy_url: String, routing_session_affinity: bool, routing_session_affinity_ttl: String, + request_retry: u32, + max_retry_credentials: u32, + max_retry_interval: u32, + streaming_bootstrap_retries: u32, } #[derive(Clone, Serialize)] @@ -1063,6 +1083,10 @@ struct CoreConfigSettings { proxy_url: String, routing_session_affinity: bool, routing_session_affinity_ttl: String, + request_retry: u32, + max_retry_credentials: u32, + max_retry_interval: u32, + streaming_bootstrap_retries: u32, // Kept for internal config migration/tests; never exposed to the WebView. #[allow(dead_code)] #[serde(skip_serializing)] @@ -1088,6 +1112,10 @@ struct CoreConfigView { proxy_url: String, routing_session_affinity: bool, routing_session_affinity_ttl: String, + request_retry: u32, + max_retry_credentials: u32, + max_retry_interval: u32, + streaming_bootstrap_retries: u32, } impl Default for CoreInstallTask { @@ -1439,21 +1467,18 @@ impl GuiConfigState { }) } - fn update_network_routing( - &self, - port: u16, - allow_lan: bool, - proxy_url: &str, - routing_session_affinity: bool, - routing_session_affinity_ttl: &str, - ) -> Result { + fn update_network_routing(&self, settings: &GuiConfigFile) -> Result { self.update(|config| { - config.port = port; - config.allow_lan = allow_lan; - config.host = if allow_lan { "0.0.0.0" } else { "127.0.0.1" }.to_string(); - config.proxy_url = proxy_url.to_string(); - config.routing_session_affinity = routing_session_affinity; - config.routing_session_affinity_ttl = routing_session_affinity_ttl.to_string(); + config.port = settings.port; + config.allow_lan = settings.allow_lan; + config.host = settings.host.clone(); + config.proxy_url = settings.proxy_url.clone(); + config.routing_session_affinity = settings.routing_session_affinity; + config.routing_session_affinity_ttl = settings.routing_session_affinity_ttl.clone(); + config.request_retry = settings.request_retry; + config.max_retry_credentials = settings.max_retry_credentials; + config.max_retry_interval = settings.max_retry_interval; + config.streaming_bootstrap_retries = settings.streaming_bootstrap_retries; Ok(()) }) } @@ -1542,6 +1567,10 @@ impl GuiConfigState { config.proxy_url = settings.proxy_url.clone(); config.routing_session_affinity = settings.routing_session_affinity; config.routing_session_affinity_ttl = settings.routing_session_affinity_ttl.clone(); + config.request_retry = settings.request_retry; + config.max_retry_credentials = settings.max_retry_credentials; + config.max_retry_interval = settings.max_retry_interval; + config.streaming_bootstrap_retries = settings.streaming_bootstrap_retries; Ok(()) }) } @@ -1587,6 +1616,10 @@ impl From<&GuiConfigFile> for CoreConfigSettings { proxy_url: config.proxy_url.clone(), routing_session_affinity: config.routing_session_affinity, routing_session_affinity_ttl: config.routing_session_affinity_ttl.clone(), + request_retry: config.request_retry, + max_retry_credentials: config.max_retry_credentials, + max_retry_interval: config.max_retry_interval, + streaming_bootstrap_retries: config.streaming_bootstrap_retries, management_secret_key: Some(config.management_secret_key.clone()), } } @@ -1611,6 +1644,10 @@ impl From<&GuiConfigFile> for CoreConfigView { proxy_url: config.proxy_url.clone(), routing_session_affinity: config.routing_session_affinity, routing_session_affinity_ttl: config.routing_session_affinity_ttl.clone(), + request_retry: config.request_retry, + max_retry_credentials: config.max_retry_credentials, + max_retry_interval: config.max_retry_interval, + streaming_bootstrap_retries: config.streaming_bootstrap_retries, } } } diff --git a/src-tauri/src/tests/app_settings.rs b/src-tauri/src/tests/app_settings.rs index 67c5667..8149a67 100644 --- a/src-tauri/src/tests/app_settings.rs +++ b/src-tauri/src/tests/app_settings.rs @@ -88,6 +88,10 @@ fn gui_config_defaults_are_stable() { assert!(content.contains("management-secret-key = \"\"")); assert!(content.contains("plugins-enabled = false")); assert!(content.contains("routing-strategy = \"round-robin\"")); + assert!(content.contains("request-retry = 3")); + assert!(content.contains("max-retry-credentials = 0")); + assert!(content.contains("max-retry-interval = 30")); + assert!(content.contains("streaming-bootstrap-retries = 0")); } #[test] diff --git a/src-tauri/src/tests/core_config.rs b/src-tauri/src/tests/core_config.rs index 04745c6..ee0db50 100644 --- a/src-tauri/src/tests/core_config.rs +++ b/src-tauri/src/tests/core_config.rs @@ -213,6 +213,10 @@ fn legacy_gui_config_can_seed_managed_core_settings() { proxy_url: "http://127.0.0.1:8080".to_string(), routing_session_affinity: true, routing_session_affinity_ttl: "45m".to_string(), + request_retry: 1, + max_retry_credentials: 2, + max_retry_interval: 5, + streaming_bootstrap_retries: 1, management_secret_key: Some("management-secret".to_string()), }; @@ -233,6 +237,10 @@ fn legacy_gui_config_can_seed_managed_core_settings() { assert_eq!(config.proxy_url, "http://127.0.0.1:8080"); assert!(config.routing_session_affinity); assert_eq!(config.routing_session_affinity_ttl, "45m"); + assert_eq!(config.request_retry, 1); + assert_eq!(config.max_retry_credentials, 2); + assert_eq!(config.max_retry_interval, 5); + assert_eq!(config.streaming_bootstrap_retries, 1); assert!(config.run_on_startup); } @@ -343,6 +351,10 @@ fn confirmed_network_routing_patch_updates_all_fields_together() { proxy_url: "socks5://127.0.0.1:7890".to_string(), routing_session_affinity: true, routing_session_affinity_ttl: "2h".to_string(), + request_retry: 1, + max_retry_credentials: 2, + max_retry_interval: 5, + streaming_bootstrap_retries: 1, ..GuiConfigFile::default() }; let input = "# network\nhost: 127.0.0.1\nport: 8317\nproxy-url: \"\"\n# routing\nrouting:\n strategy: round-robin\n# unrelated\ndebug: true\n"; @@ -356,6 +368,10 @@ fn confirmed_network_routing_patch_updates_all_fields_together() { assert_eq!(document["proxy-url"], "socks5://127.0.0.1:7890"); assert_eq!(document["routing"]["session-affinity"], true); assert_eq!(document["routing"]["session-affinity-ttl"], "2h"); + assert_eq!(document["request-retry"], 1); + assert_eq!(document["max-retry-credentials"], 2); + assert_eq!(document["max-retry-interval"], 5); + assert_eq!(document["streaming"]["bootstrap-retries"], 1); assert_eq!(document["routing"]["strategy"], "round-robin"); assert_eq!(document["debug"], true); assert!(updated.contains("# network")); @@ -634,6 +650,32 @@ fn core_config_reads_proxy_and_session_affinity_fields() { assert_eq!(defaults.routing_session_affinity_ttl, ""); } +#[test] +fn core_config_reads_retry_fields_and_uses_core_defaults() { + let document = serde_norway::from_str::( + "request-retry: 1\nmax-retry-credentials: 2\nmax-retry-interval: 5\nstreaming:\n bootstrap-retries: 4\n", + ) + .unwrap(); + let settings = core_config_settings_from_value(&document).unwrap(); + + assert_eq!(settings.request_retry, 1); + assert_eq!(settings.max_retry_credentials, 2); + assert_eq!(settings.max_retry_interval, 5); + assert_eq!(settings.streaming_bootstrap_retries, 4); + + let defaults = core_config_settings_from_value(&serde_norway::from_str("{}").unwrap()).unwrap(); + assert_eq!(defaults.request_retry, DEFAULT_REQUEST_RETRY); + assert_eq!( + defaults.max_retry_credentials, + DEFAULT_MAX_RETRY_CREDENTIALS + ); + assert_eq!(defaults.max_retry_interval, DEFAULT_MAX_RETRY_INTERVAL); + assert_eq!( + defaults.streaming_bootstrap_retries, + DEFAULT_STREAMING_BOOTSTRAP_RETRIES + ); +} + #[test] fn managed_session_settings_use_canonical_yaml_and_preserve_unrelated_content() { let input = "# global proxy\nproxy-url: old\n# routing options\nrouting:\n strategy: round-robin\n sessionAffinity: false\n sessionAffinityTTL: 10m\n# unrelated option\ndebug: true\n"; @@ -641,6 +683,10 @@ fn managed_session_settings_use_canonical_yaml_and_preserve_unrelated_content() proxy_url: "http://127.0.0.1:8080".to_string(), routing_session_affinity: true, routing_session_affinity_ttl: "1h".to_string(), + request_retry: 1, + max_retry_credentials: 2, + max_retry_interval: 5, + streaming_bootstrap_retries: 1, ..GuiConfigFile::default() }; let rendered = apply_gui_managed_settings(input, &config).unwrap(); @@ -649,6 +695,10 @@ fn managed_session_settings_use_canonical_yaml_and_preserve_unrelated_content() assert_eq!(document["proxy-url"], "http://127.0.0.1:8080"); assert_eq!(document["routing"]["session-affinity"], true); assert_eq!(document["routing"]["session-affinity-ttl"], "1h"); + assert_eq!(document["request-retry"], 1); + assert_eq!(document["max-retry-credentials"], 2); + assert_eq!(document["max-retry-interval"], 5); + assert_eq!(document["streaming"]["bootstrap-retries"], 1); assert!(rendered.contains("# global proxy")); assert!(rendered.contains("# unrelated option")); assert_eq!(document["debug"], true); @@ -737,6 +787,10 @@ fn startup_preserves_all_user_owned_yaml_and_only_applies_gui_managed_values() { proxy_url: "socks5://127.0.0.1:7890".to_string(), routing_session_affinity: true, routing_session_affinity_ttl: "1h".to_string(), + request_retry: 1, + max_retry_credentials: 2, + max_retry_interval: 5, + streaming_bootstrap_retries: 1, }; let merged = merge_core_config_yaml(template, Some(current), &config).unwrap(); let current_document = serde_norway::from_str::(current).unwrap(); @@ -759,6 +813,10 @@ fn startup_preserves_all_user_owned_yaml_and_only_applies_gui_managed_values() { assert_eq!(document["proxy-url"], "socks5://127.0.0.1:7890"); assert_eq!(document["routing"]["session-affinity"], true); assert_eq!(document["routing"]["session-affinity-ttl"], "1h"); + assert_eq!(document["request-retry"], 1); + assert_eq!(document["max-retry-credentials"], 2); + assert_eq!(document["max-retry-interval"], 5); + assert_eq!(document["streaming"]["bootstrap-retries"], 1); assert_eq!(document["usage-statistics-enabled"], false); assert!(document.get("new-option").is_none()); assert_eq!(document["nested"], current_document["nested"]); diff --git a/src-tauri/src/usage.rs b/src-tauri/src/usage.rs index 384ccf8..a225ec5 100644 --- a/src-tauri/src/usage.rs +++ b/src-tauri/src/usage.rs @@ -3038,6 +3038,10 @@ mod tests { proxy_url: String::new(), routing_session_affinity: false, routing_session_affinity_ttl: String::new(), + request_retry: crate::DEFAULT_REQUEST_RETRY, + max_retry_credentials: crate::DEFAULT_MAX_RETRY_CREDENTIALS, + max_retry_interval: crate::DEFAULT_MAX_RETRY_INTERVAL, + streaming_bootstrap_retries: crate::DEFAULT_STREAMING_BOOTSTRAP_RETRIES, }; let record = normalize_usage_record( serde_json::json!({ diff --git a/src/i18n/ja.ts b/src/i18n/ja.ts index 0fc0a57..a57608e 100644 --- a/src/i18n/ja.ts +++ b/src/i18n/ja.ts @@ -274,6 +274,7 @@ export const jaOverrides = { 'config.error.duplicateKey': '同じ認証キーがすでに存在します', 'config.error.remarkTooLong': 'メモは 80 文字以内で入力してください', 'config.error.portRange': 'ポートは 1〜65535 の整数で入力してください', + 'config.error.retryRange': '再試行設定は 0〜4294967295 の整数で入力してください', 'config.error.networkRestartFailed': '設定は保存されましたが、コアの再起動に失敗しました：{error}', 'config.error.saveFailed': '保存に失敗しました：{error}', 'config.notice.keyAdded': '認証キーを追加しました', @@ -330,6 +331,7 @@ export const jaOverrides = { 'config.network.title': 'ネットワークとセッションルーティング', 'config.network.networkSection': 'ネットワーク設定', 'config.network.routingSection': 'セッションルーティング', + 'config.network.retrySection': '失敗時の再試行', 'config.network.saving': '保存中', 'config.network.saved': '保存済み', 'config.network.unsaved': '未保存の変更あり', @@ -345,6 +347,14 @@ export const jaOverrides = { 'config.network.sessionAffinityHint': '同じセッションでは以前選択した認証情報を優先します', 'config.network.sessionTtl': 'セッションアフィニティ TTL', 'config.network.sessionTtlHint': '例：1h。空欄の場合はコアのデフォルト値を使用します', + 'config.network.requestRetry': 'リクエスト再試行回数', + 'config.network.requestRetryHint': '403、408、500、502、503、504 応答の再試行回数です。0 で無効になります', + 'config.network.maxRetryCredentials': '試行する認証情報の上限', + 'config.network.maxRetryCredentialsHint': '失敗した1リクエストで試す認証情報の上限です。0 ですべてを試します', + 'config.network.maxRetryInterval': '再試行の最大待機時間（秒）', + 'config.network.maxRetryIntervalHint': 'クールダウン中の認証情報を待つ上限です。小さくすると累積待機時間を短縮できますが、リクエストの強制タイムアウトではありません', + 'config.network.streamingBootstrapRetries': 'ストリーミング開始時の再試行回数', + 'config.network.streamingBootstrapRetriesHint': '最初の応答を送信する前に許可する安全な再試行回数です。0 で無効になります', 'config.notice.softwareUpdated': 'ソフトウェア設定を保存しました', 'config.software.title': 'ソフトウェア設定', 'config.software.autostart': 'システム起動時に開始', diff --git a/src/i18n/locales/en.ts b/src/i18n/locales/en.ts index 8505ded..85654b9 100644 --- a/src/i18n/locales/en.ts +++ b/src/i18n/locales/en.ts @@ -274,6 +274,7 @@ export const en: Record = { 'config.error.duplicateKey': 'This authentication key already exists', 'config.error.remarkTooLong': 'The remark cannot exceed 80 characters', 'config.error.portRange': 'The port must be an integer from 1 to 65535', + 'config.error.retryRange': 'Retry settings must be integers from 0 to 4294967295', 'config.error.networkRestartFailed': 'Settings were saved, but the core failed to restart: {error}', 'config.error.saveFailed': 'Failed to save: {error}', 'config.notice.keyAdded': 'Authentication key added', @@ -341,6 +342,7 @@ export const en: Record = { 'config.network.title': 'Network & Session Routing', 'config.network.networkSection': 'Network Settings', 'config.network.routingSection': 'Session Routing', + 'config.network.retrySection': 'Failure Retries', 'config.network.saving': 'Saving', 'config.network.saved': 'Saved', 'config.network.unsaved': 'Unsaved changes', @@ -356,6 +358,14 @@ export const en: Record = { 'config.network.sessionAffinityHint': 'Prefer the previously selected credential for the same session', 'config.network.sessionTtl': 'Session Affinity TTL', 'config.network.sessionTtlHint': 'For example, 1h; leave blank to use the core default', + 'config.network.requestRetry': 'Request Retries', + 'config.network.requestRetryHint': 'Retries responses with status 403, 408, 500, 502, 503, or 504; 0 disables retries', + 'config.network.maxRetryCredentials': 'Maximum Credentials Tried', + 'config.network.maxRetryCredentialsHint': 'Maximum credentials tried for one failed request; 0 tries all available credentials', + 'config.network.maxRetryInterval': 'Maximum Retry Wait (seconds)', + 'config.network.maxRetryIntervalHint': 'Limits how long to wait for cooled-down credentials; lowering it reduces cumulative wait, but is not a hard request timeout', + 'config.network.streamingBootstrapRetries': 'Streaming Bootstrap Retries', + 'config.network.streamingBootstrapRetriesHint': 'Safe retries allowed before the first response is sent; 0 disables them', 'config.keys.addTitle': 'Add Authentication Key', 'config.keys.editTitle': 'Edit Authentication Key', 'common.close': 'Close', diff --git a/src/i18n/locales/zh-CN.ts b/src/i18n/locales/zh-CN.ts index a656c6c..081d2d2 100644 --- a/src/i18n/locales/zh-CN.ts +++ b/src/i18n/locales/zh-CN.ts @@ -272,6 +272,7 @@ export const zhCN = { 'config.error.duplicateKey': '该鉴权密钥已经存在', 'config.error.remarkTooLong': '备注不能超过 80 个字符', 'config.error.portRange': '端口必须是 1–65535 之间的整数', + 'config.error.retryRange': '重试设置必须是 0–4294967295 之间的整数', 'config.error.networkRestartFailed': '设置已保存，但内核重启失败：{error}', 'config.error.saveFailed': '保存失败：{error}', 'config.notice.keyAdded': '鉴权密钥已新增', @@ -339,6 +340,7 @@ export const zhCN = { 'config.network.title': '网络与会话路由', 'config.network.networkSection': '网络设置', 'config.network.routingSection': '会话路由', + 'config.network.retrySection': '失败重试', 'config.network.saving': '正在保存', 'config.network.saved': '已保存', 'config.network.unsaved': '有未保存修改', @@ -354,6 +356,14 @@ export const zhCN = { 'config.network.sessionAffinityHint': '同一会话优先复用已选择的凭据', 'config.network.sessionTtl': '会话粘性 TTL', 'config.network.sessionTtlHint': '例如 1h；留空使用内核默认值', + 'config.network.requestRetry': '请求重试次数', + 'config.network.requestRetryHint': '对 403、408、500、502、503、504 响应的重试次数；0 表示不重试', + 'config.network.maxRetryCredentials': '最大尝试凭据数', + 'config.network.maxRetryCredentialsHint': '单次失败请求最多尝试的凭据数；0 表示尝试全部可用凭据', + 'config.network.maxRetryInterval': '最大重试等待（秒）', + 'config.network.maxRetryIntervalHint': '等待冷却凭据的最长时间；减小可缩短累计等待，但不是单次请求硬超时', + 'config.network.streamingBootstrapRetries': '流式启动重试次数', + 'config.network.streamingBootstrapRetriesHint': '发送首个响应前允许的安全重试次数；0 表示禁用', 'config.keys.addTitle': '新增鉴权密钥', 'config.keys.editTitle': '编辑鉴权密钥', 'common.close': '关闭', diff --git a/src/pages/ConfigPanel.tsx b/src/pages/ConfigPanel.tsx index b356228..c5b1515 100644 --- a/src/pages/ConfigPanel.tsx +++ b/src/pages/ConfigPanel.tsx @@ -35,6 +35,10 @@ type CoreConfigSettings = { proxyUrl: string; routingSessionAffinity: boolean; routingSessionAffinityTtl: string; + requestRetry: number; + maxRetryCredentials: number; + maxRetryInterval: number; + streamingBootstrapRetries: number; }; type CoreApiKey = { @@ -54,7 +58,16 @@ type ConfigAction = type NoticeTone = 'success' | 'error'; type ConfigSubpage = 'general' | 'network' | 'software' | 'aliases'; type CloseBehavior = 'ask' | 'exit' | 'minimize-to-tray'; -type NetworkDraftField = 'port' | 'allowLan' | 'proxyUrl' | 'sessionAffinity' | 'sessionTtl'; +type NetworkDraftField = + | 'port' + | 'allowLan' + | 'proxyUrl' + | 'sessionAffinity' + | 'sessionTtl' + | 'requestRetry' + | 'maxRetryCredentials' + | 'maxRetryInterval' + | 'streamingBootstrapRetries'; type DraftRefreshMode = 'replace' | 'preserve'; type NetworkDraftDirty = Record; @@ -71,6 +84,10 @@ const cleanNetworkDraft = (): NetworkDraftDirty => ({ proxyUrl: false, sessionAffinity: false, sessionTtl: false, + requestRetry: false, + maxRetryCredentials: false, + maxRetryInterval: false, + streamingBootstrapRetries: false, }); const ROUTING_OPTIONS = [ @@ -110,7 +127,12 @@ export function ConfigPanelPage() { const [proxyUrlDraft, setProxyUrlDraft] = useState(''); const [sessionAffinityDraft, setSessionAffinityDraft] = useState(false); const [sessionTtlDraft, setSessionTtlDraft] = useState(''); + const [requestRetryDraft, setRequestRetryDraft] = useState('3'); + const [maxRetryCredentialsDraft, setMaxRetryCredentialsDraft] = useState('0'); + const [maxRetryIntervalDraft, setMaxRetryIntervalDraft] = useState('30'); + const [streamingBootstrapRetriesDraft, setStreamingBootstrapRetriesDraft] = useState('0'); const [portError, setPortError] = useState(''); + const [retryError, setRetryError] = useState(''); const [savedStatusVisible, setSavedStatusVisible] = useState(false); const networkDraftDirtyRef = useRef(cleanNetworkDraft()); const noticeTimerRef = useRef(null); @@ -162,6 +184,12 @@ export function ConfigPanelPage() { if (!dirty.proxyUrl) setProxyUrlDraft(result.proxyUrl); if (!dirty.sessionAffinity) setSessionAffinityDraft(result.routingSessionAffinity); if (!dirty.sessionTtl) setSessionTtlDraft(result.routingSessionAffinityTtl); + if (!dirty.requestRetry) setRequestRetryDraft(String(result.requestRetry)); + if (!dirty.maxRetryCredentials) setMaxRetryCredentialsDraft(String(result.maxRetryCredentials)); + if (!dirty.maxRetryInterval) setMaxRetryIntervalDraft(String(result.maxRetryInterval)); + if (!dirty.streamingBootstrapRetries) { + setStreamingBootstrapRetriesDraft(String(result.streamingBootstrapRetries)); + } return; } networkDraftDirtyRef.current = cleanNetworkDraft(); @@ -170,7 +198,12 @@ export function ConfigPanelPage() { setProxyUrlDraft(result.proxyUrl); setSessionAffinityDraft(result.routingSessionAffinity); setSessionTtlDraft(result.routingSessionAffinityTtl); + setRequestRetryDraft(String(result.requestRetry)); + setMaxRetryCredentialsDraft(String(result.maxRetryCredentials)); + setMaxRetryIntervalDraft(String(result.maxRetryInterval)); + setStreamingBootstrapRetriesDraft(String(result.streamingBootstrapRetries)); setPortError(''); + setRetryError(''); }; const markDraftDirty = (field: NetworkDraftField) => { @@ -426,8 +459,25 @@ export function ConfigPanelPage() { const proxyUrl = proxyUrlDraft.trim(); const routingSessionAffinityTtl = sessionTtlDraft.trim(); + const retryDrafts = [ + requestRetryDraft, + maxRetryCredentialsDraft, + maxRetryIntervalDraft, + streamingBootstrapRetriesDraft, + ]; + const retryValues = retryDrafts.map(Number); + if ( + retryDrafts.some((value) => value.length === 0) + || retryValues.some((value) => !Number.isInteger(value) || value < 0 || value > 4294967295) + ) { + setRetryError(t('config.error.retryRange')); + showNotice(t('config.error.retryRange'), 'error'); + return; + } + const [requestRetry, maxRetryCredentials, maxRetryInterval, streamingBootstrapRetries] = retryValues; const networkChanged = port !== settings.port || allowLanDraft !== settings.allowLan; setPortError(''); + setRetryError(''); setSavedStatusVisible(false); setBusyAction('network'); try { @@ -438,6 +488,10 @@ export function ConfigPanelPage() { proxyUrl, routingSessionAffinity: sessionAffinityDraft, routingSessionAffinityTtl, + requestRetry, + maxRetryCredentials, + maxRetryInterval, + streamingBootstrapRetries, }, }); applySettings(result); @@ -509,6 +563,10 @@ export function ConfigPanelPage() { || proxyUrlDraft.trim() !== settings?.proxyUrl || sessionAffinityDraft !== settings?.routingSessionAffinity || sessionTtlDraft.trim() !== settings?.routingSessionAffinityTtl + || requestRetryDraft !== String(settings?.requestRetry) + || maxRetryCredentialsDraft !== String(settings?.maxRetryCredentials) + || maxRetryIntervalDraft !== String(settings?.maxRetryInterval) + || streamingBootstrapRetriesDraft !== String(settings?.streamingBootstrapRetries) ); const networkStatusLabel = loading ? t('common.loading') @@ -1034,6 +1092,126 @@ export function ConfigPanelPage() { + +
+
+
+
+ + + + + + + +
+
From 2ad85880d495b835a1833bc708f3ed9d96543871 Mon Sep 17 00:00:00 2001 From: lzt404 <2596933790@qq.com> Date: Wed, 12 Aug 2026 00:53:10 +0800 Subject: [PATCH 3/4] fix(api-access): show model discovery errors inside provider dialog --- src/pages/ApiAccessPage.tsx | 126 +++++++++++++++++++++++++----------- src/styles.css | 23 +++++++ 2 files changed, 110 insertions(+), 39 deletions(-) diff --git a/src/pages/ApiAccessPage.tsx b/src/pages/ApiAccessPage.tsx index a5532fa..70f07b3 100644 --- a/src/pages/ApiAccessPage.tsx +++ b/src/pages/ApiAccessPage.tsx @@ -1,4 +1,4 @@ -import { FormEvent, useCallback, useEffect, useMemo, useState } from 'react'; +import { FormEvent, useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { invoke } from '@tauri-apps/api/core'; import { Check, @@ -79,6 +79,13 @@ type ProviderRow = { remark: string; }; +type ProviderSaveResult = + | { saved: true } + | { saved: false; error: string; target: 'form' | 'models' }; + +const requestErrorMessage = (error: unknown) => + error instanceof Error ? error.message : String(error); + export type ProviderDraft = { name: string; apiKey: string; @@ -758,12 +765,14 @@ export function ApiAccessPage() { ); const openCreate = () => { + setError(''); setEditingRow(null); setDialogDraft(createProviderDraft(activeCategory)); setDialogOpen(true); }; const openEdit = (row: ProviderRow) => { + setError(''); setEditingRow(row); const draft = draftFromRow(row); setDialogDraft(activeCategory === 'deepseek' @@ -772,7 +781,7 @@ export function ApiAccessPage() { setDialogOpen(true); }; - const saveProvider = async (nextDraft: ProviderDraft): Promise => { + const saveProvider = async (nextDraft: ProviderDraft): Promise => { const definition = activeDefinition; const preparedDraft = applyProviderRemarkIdentity( activeCategory, @@ -790,18 +799,18 @@ export function ApiAccessPage() { || (remarkRequired && !preparedDraft.remark.trim()) || (baseUrlRequired && !preparedDraft.baseUrl.trim()) ) { - setError( - definition.openAi + return { + saved: false, + target: 'form', + error: definition.openAi ? t('apiAccess.error.requiredAll') : baseUrlRequired ? t('apiAccess.error.requiredBaseKey') : t('apiAccess.error.requiredKey'), - ); - return false; + }; } if (Array.from(preparedDraft.remark.trim()).length > 80 || /[\u0000-\u001f\u007f]/.test(preparedDraft.remark)) { - setError(t('apiAccess.error.remarkInvalid')); - return false; + return { saved: false, target: 'form', error: t('apiAccess.error.remarkInvalid') }; } let baseUrl = preparedDraft.baseUrl.trim(); let providerHeaders: Record = {}; @@ -810,23 +819,31 @@ export function ApiAccessPage() { if (baseUrlRequired && !baseUrl) throw new Error(t('apiAccess.error.baseRequired', { provider: definition.label })); providerHeaders = parseProviderHeaders(preparedDraft.headersText ?? ''); } catch (requestError) { - setError(String(requestError)); - return false; + return { saved: false, target: 'form', error: requestErrorMessage(requestError) }; } setBusy(true); setError(''); try { let draftToSave = { ...preparedDraftForSave, baseUrl }; if (definition.openAi && draftToSave.models.length === 0) { - const fetchedModels = await fetchModels( - 'openai', - baseUrl, - parsedApiKeys[0], - editingRow?.authIndex, - providerHeaders, - ); - if (fetchedModels.length === 0) { - throw new Error(t('apiAccess.error.noModels')); + let fetchedModels: ModelOption[]; + try { + fetchedModels = await fetchModels( + 'openai', + baseUrl, + parsedApiKeys[0], + editingRow?.authIndex, + providerHeaders, + ); + if (fetchedModels.length === 0) { + throw new Error(t('apiAccess.error.noModels')); + } + } catch (requestError) { + return { + saved: false, + target: 'models', + error: requestErrorMessage(requestError), + }; } draftToSave = applyProviderPreset(activeCategory, { ...draftToSave, @@ -883,10 +900,9 @@ export function ApiAccessPage() { }); setNotice(editingRow ? t('apiAccess.notice.updated') : t('apiAccess.notice.added')); await loadProviders(); - return true; + return { saved: true }; } catch (requestError) { - setError(String(requestError)); - return false; + return { saved: false, target: 'form', error: requestErrorMessage(requestError) }; } finally { setBusy(false); } @@ -1312,7 +1328,7 @@ type ApiProviderDialogProps = { initialDraft: ProviderDraft; busy: boolean; onClose: () => void; - onSave: (draft: ProviderDraft) => Promise; + onSave: (draft: ProviderDraft) => Promise; }; function ApiProviderDialog({ @@ -1329,6 +1345,7 @@ function ApiProviderDialog({ const [draft, setDraft] = useState(initialDraft); const [modelLoading, setModelLoading] = useState(false); const [modelError, setModelError] = useState(''); + const [formError, setFormError] = useState(''); const [discoveredModels, setDiscoveredModels] = useState([]); const [modelDiscoveryOpen, setModelDiscoveryOpen] = useState(false); const [modelSearch, setModelSearch] = useState(''); @@ -1336,6 +1353,7 @@ function ApiProviderDialog({ const [selectedModelNames, setSelectedModelNames] = useState>( () => new Set(initialDraft.models.map((model) => model.name.toLowerCase())), ); + const modelCardRef = useRef(null); const modelOptions = useMemo(() => { const options = new Map(); @@ -1366,6 +1384,10 @@ function ApiProviderDialog({ field: 'apiKey' | 'remark' | 'baseUrl' | 'priority' | 'prefix' | 'headersText' | 'excludedModelsText' | 'testModel' | 'cloakMode' | 'cloakSensitiveWordsText', value: string, ) => { + setFormError(''); + if (field === 'apiKey' || field === 'baseUrl' || field === 'headersText') { + setModelError(''); + } setDraft((current) => ({ ...current, [field]: value })); }; @@ -1373,6 +1395,7 @@ function ApiProviderDialog({ field: 'disableCooling' | 'websockets' | 'cloakStrictMode' | 'cloakCacheUserId', value: boolean, ) => { + setFormError(''); setDraft((current) => ({ ...current, [field]: value })); }; @@ -1457,8 +1480,7 @@ function ApiProviderDialog({ )); if (!models.length) setModelError(t('apiAccess.error.noAvailableModels')); } catch (requestError) { - setDiscoveredModels([]); - setModelError(String(requestError)); + setModelError(requestErrorMessage(requestError)); } finally { setModelLoading(false); } @@ -1519,7 +1541,20 @@ function ApiProviderDialog({ const submit = async (event: FormEvent) => { event.preventDefault(); - if (await onSave(draft)) onClose(); + setFormError(''); + const result = await onSave(draft); + if (result.saved) { + onClose(); + return; + } + if (result.target === 'models') { + setModelError(result.error); + window.requestAnimationFrame(() => { + modelCardRef.current?.scrollIntoView({ behavior: 'smooth', block: 'center' }); + }); + return; + } + setFormError(result.error); }; const hasModelExclusions = activeSection !== 'openai-compatibility' @@ -1619,7 +1654,7 @@ function ApiProviderDialog({ ) : {t('apiAccess.thinking.empty')}} ) : null} -
+
{t('apiAccess.models.title')}{t('apiAccess.models.description')}
+ {formError ? ( +
+ {formError} +
+ ) : null}
@@ -1754,22 +1794,30 @@ function ApiProviderDialog({
{modelLoading ? (
{t('apiAccess.modelDialog.fetching')}
- ) : modelError ? ( + ) : modelError && modelOptions.length === 0 ? (
{t('apiAccess.modelDialog.fetchFailed')}{modelError}
) : visibleModelOptions.length === 0 ? (
{modelOptions.length ? t('apiAccess.modelDialog.noMatch') : t('apiAccess.modelDialog.none')}{modelOptions.length ? t('apiAccess.modelDialog.tryKeyword') : t('apiAccess.modelDialog.checkCredentials')}
) : ( -
- {visibleModelOptions.map((model) => { - const checked = selectedModelNames.has(model.name.toLowerCase()); - return ( - - ); - })} +
+ {modelError ? ( +
+ {t('apiAccess.modelDialog.fetchFailed')} + {modelError} +
+ ) : null} +
+ {visibleModelOptions.map((model) => { + const checked = selectedModelNames.has(model.name.toLowerCase()); + return ( + + ); + })} +
)}
diff --git a/src/styles.css b/src/styles.css index ac4d64f..fb3ab1f 100644 --- a/src/styles.css +++ b/src/styles.css @@ -3104,6 +3104,29 @@ textarea:focus-visible { overflow: auto; } +.model-discovery-results { + display: grid; + grid-template-rows: auto minmax(0, 1fr); + height: 100%; + min-height: 0; +} + +.model-discovery-inline-error { + display: grid; + gap: 2px; + padding: 8px 11px; + border-bottom: 1px solid var(--theme-dfb7b0); + background: var(--theme-fff1ee); + color: var(--theme-9f2d20); + font-size: var(--font-size-meta); +} + +.model-discovery-inline-error span { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + .model-discovery-row { display: grid; grid-template-columns: 18px minmax(0, 1fr) 18px; From 6ac2d8d4af5a186bedd8b0b99b39028642388ee3 Mon Sep 17 00:00:00 2001 From: lzt404 <2596933790@qq.com> Date: Thu, 13 Aug 2026 23:03:04 +0800 Subject: [PATCH 4/4] chore: stop tracking agent docs --- agent_docs/Instructions for running CLIProxyAPI.md | 8 -------- 1 file changed, 8 deletions(-) delete mode 100644 agent_docs/Instructions for running CLIProxyAPI.md diff --git a/agent_docs/Instructions for running CLIProxyAPI.md b/agent_docs/Instructions for running CLIProxyAPI.md deleted file mode 100644 index 089656e..0000000 --- a/agent_docs/Instructions for running CLIProxyAPI.md +++ /dev/null @@ -1,8 +0,0 @@ -CLIProxyAPI is the core file of this project. After downloading it from the GitHub repository, you can simply extract it locally and use it directly. - -Latest release page: https://github.com/router-for-me/CLIProxyAPI/releases/latest - -After extraction, you will find the cli-proxy-api binary file and the config.example.yaml configuration template. - -You need to copy config.example.yaml to config.yaml and modify the DIY configuration before running it. Of course, the default configuration can also run directly. -