diff --git a/src-tauri/src/database/backup_restore_certification_ext.rs b/src-tauri/src/database/backup_restore_certification_ext.rs new file mode 100644 index 000000000..8e9716195 --- /dev/null +++ b/src-tauri/src/database/backup_restore_certification_ext.rs @@ -0,0 +1,496 @@ +#![cfg(test)] +// 裁决方授权:扩展认证沿用核心套件的纯风格类 lint 豁免。安全/正确性 +// lint 不在豁免范围内。 +#![allow(clippy::type_complexity)] +//! 前置工程 B:历史迁移 fail-closed 扩展认证。 +//! +//! 本文件只扩展、不得替代 `backup_restore_certification.rs`。行为测试必须 +//! 穿过两个公开入口,证明 v1 旧式 proxy 配置中的损坏不会被默认值修复后 +//! 静默发布。 + +use super::Database; +use crate::error::AppError; +use rusqlite::{backup::Backup, Connection}; +use serial_test::serial; +use std::path::Path; +use std::time::Duration; + +#[derive(Clone, Copy, Debug)] +enum RestoreEntry { + Sql, + Binary, +} + +#[derive(Clone, Copy, Debug)] +enum V1Damage { + None, + MissingPrimaryRow, + WrongStorageClass, + FieldDecodeFailure, + MissingCircuitRow, + MissingSettingRow, +} + +struct TestHomeGuard(Option); + +impl TestHomeGuard { + fn set(path: &Path) -> Self { + let previous = std::env::var_os("CC_SWITCH_TEST_HOME"); + std::env::set_var("CC_SWITCH_TEST_HOME", path); + Self(previous) + } +} + +impl Drop for TestHomeGuard { + fn drop(&mut self) { + match self.0.take() { + Some(previous) => std::env::set_var("CC_SWITCH_TEST_HOME", previous), + None => std::env::remove_var("CC_SWITCH_TEST_HOME"), + } + } +} + +fn v1_proxy_fixture(damage: V1Damage) -> String { + let proxy_row = match damage { + V1Damage::MissingPrimaryRow => "(2, 0, '127.0.0.1', 15721, 1, 3, 60, 120, 600)", + V1Damage::WrongStorageClass => "(1, 0, '127.0.0.1', X'00', 1, 3, 60, 120, 600)", + V1Damage::FieldDecodeFailure => "(1, 0, '127.0.0.1', 15721, 1, 3, 2147483648, 120, 600)", + V1Damage::None | V1Damage::MissingCircuitRow | V1Damage::MissingSettingRow => { + "(1, 0, '127.0.0.1', 15721, 1, 3, 60, 120, 600)" + } + }; + let circuit_row = if matches!(damage, V1Damage::MissingCircuitRow) { + "" + } else { + "INSERT INTO circuit_breaker_config VALUES (1, 5, 2, 60, 0.5, 10);" + }; + let settings_rows = if matches!(damage, V1Damage::MissingSettingRow) { + "" + } else { + "INSERT INTO settings (key, value) VALUES + ('proxy_takeover_claude', 'false'), + ('auto_failover_enabled_claude', 'false'), + ('proxy_takeover_codex', 'false'), + ('auto_failover_enabled_codex', 'false'), + ('proxy_takeover_gemini', 'false'), + ('auto_failover_enabled_gemini', 'false');" + }; + + format!( + "-- CC Switch SQLite 导出 +BEGIN TRANSACTION; +CREATE TABLE proxy_config ( + id INTEGER PRIMARY KEY, + proxy_enabled INTEGER NOT NULL DEFAULT 0, + listen_address TEXT NOT NULL, + listen_port INTEGER NOT NULL, + enable_logging INTEGER NOT NULL DEFAULT 1, + max_retries INTEGER NOT NULL DEFAULT 3, + streaming_first_byte_timeout INTEGER NOT NULL DEFAULT 60, + streaming_idle_timeout INTEGER NOT NULL DEFAULT 120, + non_streaming_timeout INTEGER NOT NULL DEFAULT 600 +); +INSERT INTO proxy_config ( + id, proxy_enabled, listen_address, listen_port, enable_logging, + max_retries, streaming_first_byte_timeout, streaming_idle_timeout, + non_streaming_timeout +) VALUES {proxy_row}; +CREATE TABLE circuit_breaker_config ( + id INTEGER PRIMARY KEY, + failure_threshold INTEGER NOT NULL, + success_threshold INTEGER NOT NULL, + timeout_seconds INTEGER NOT NULL, + error_rate_threshold REAL NOT NULL, + min_requests INTEGER NOT NULL +); +{circuit_row} +CREATE TABLE settings ( + key TEXT PRIMARY KEY, + value TEXT +); +{settings_rows} +PRAGMA user_version = 1; +COMMIT;" + ) +} + +fn write_binary_fixture(sql: &str, path: &Path) -> Result<(), AppError> { + let source = Connection::open_in_memory()?; + source.execute_batch(sql)?; + let mut destination = Connection::open(path)?; + { + let backup = Backup::new(&source, &mut destination)?; + backup.run_to_completion(64, Duration::ZERO, None)?; + } + Ok(()) +} + +fn run_restore_fixture( + target: &Database, + fixture: &str, + entry: RestoreEntry, + filename: &str, +) -> Result { + match entry { + RestoreEntry::Sql => target.import_sql_string(fixture), + RestoreEntry::Binary => { + let backup_dir = crate::config::get_app_config_dir().join("backups"); + std::fs::create_dir_all(&backup_dir) + .map_err(|error| AppError::io(&backup_dir, error))?; + write_binary_fixture(fixture, &backup_dir.join(filename))?; + target.restore_from_backup(filename) + } + } +} + +fn seed_live_sentinel(target: &Database) -> Result<(), AppError> { + let conn = super::lock_conn!(target.conn); + conn.execute( + "INSERT INTO providers ( + id, app_type, name, settings_config, meta, is_current + ) VALUES ( + 'migration-cert-live', 'claude', 'live sentinel', '{}', '{}', 0 + )", + [], + )?; + Ok(()) +} + +fn assert_live_sentinel_unchanged(target: &Database) -> Result<(), AppError> { + let conn = super::lock_conn!(target.conn); + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM providers + WHERE id = 'migration-cert-live' + AND app_type = 'claude' + AND name = 'live sentinel'", + [], + |row| row.get(0), + )?; + assert_eq!( + count, 1, + "failed restore must leave the live database intact" + ); + Ok(()) +} + +fn current_schema_fixture( + version: i32, + mutate: impl FnOnce(&Connection) -> Result<(), AppError>, +) -> Result { + let source = Database::memory()?; + { + let conn = super::lock_conn!(source.conn); + mutate(&conn)?; + Database::set_user_version(&conn, version)?; + } + source.export_sql_string() +} + +fn assert_fixture_rejected( + label: &str, + expected_step: &str, + fixture: &str, +) -> Result<(), AppError> { + for entry in [RestoreEntry::Sql, RestoreEntry::Binary] { + let target = Database::memory()?; + seed_live_sentinel(&target)?; + let filename = format!("{label}-{entry:?}.db").to_ascii_lowercase(); + let error = match run_restore_fixture(&target, fixture, entry, &filename) { + Ok(value) => { + panic!("{label} via {entry:?} silently survived repair-requiring input: {value:?}") + } + Err(error) => error, + }; + assert!( + matches!(error, AppError::InvalidInput(_)), + "{label} via {entry:?} must return structured InvalidInput, got {error:?}" + ); + assert!( + error.to_string().contains(expected_step), + "{label} via {entry:?} failed at the wrong boundary: {error}" + ); + assert_live_sentinel_unchanged(&target)?; + } + Ok(()) +} + +#[test] +#[serial] +fn v1_damaged_proxy_rows_fail_closed_through_sql_and_binary_entries() -> Result<(), AppError> { + let home = tempfile::tempdir().map_err(|error| AppError::IoContext { + context: "create migration certification home".to_string(), + source: error, + })?; + let _home_guard = TestHomeGuard::set(home.path()); + + for damage in [ + V1Damage::MissingPrimaryRow, + V1Damage::WrongStorageClass, + V1Damage::FieldDecodeFailure, + V1Damage::MissingCircuitRow, + V1Damage::MissingSettingRow, + ] { + for entry in [RestoreEntry::Sql, RestoreEntry::Binary] { + let target = Database::memory()?; + seed_live_sentinel(&target)?; + let fixture = v1_proxy_fixture(damage); + let filename = format!("v1-{damage:?}-{entry:?}.db").to_ascii_lowercase(); + + let error = run_restore_fixture(&target, &fixture, entry, &filename) + .expect_err("damaged v1 restore must fail closed"); + assert!( + matches!(error, AppError::InvalidInput(_)), + "migration damage must be a structured InvalidInput, got {error:?}" + ); + assert_live_sentinel_unchanged(&target)?; + } + } + + Ok(()) +} + +#[test] +#[serial] +fn valid_v1_proxy_rows_migrate_losslessly_through_sql_and_binary_entries() -> Result<(), AppError> { + let home = tempfile::tempdir().map_err(|error| AppError::IoContext { + context: "create valid v1 migration certification home".to_string(), + source: error, + })?; + let _home_guard = TestHomeGuard::set(home.path()); + let fixture = v1_proxy_fixture(V1Damage::None); + + for entry in [RestoreEntry::Sql, RestoreEntry::Binary] { + let target = Database::memory()?; + let filename = format!("valid-v1-{entry:?}.db").to_ascii_lowercase(); + run_restore_fixture(&target, &fixture, entry, &filename)?; + let conn = super::lock_conn!(target.conn); + let restored: (i64, String, i64) = conn.query_row( + "SELECT + COUNT(*), + MIN(listen_address), + MIN(listen_port) + FROM proxy_config", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + )?; + assert_eq!( + restored, + (4, "127.0.0.1".to_string(), 15721), + "valid authoritative v1 fields must survive {entry:?}" + ); + } + Ok(()) +} + +#[test] +#[serial] +fn every_repair_class_has_a_public_entry_fail_closed_sentinel() -> Result<(), AppError> { + let home = tempfile::tempdir().map_err(|error| AppError::IoContext { + context: "create migration repair certification home".to_string(), + source: error, + })?; + let _home_guard = TestHomeGuard::set(home.path()); + + let fixtures = vec![ + ( + "v1-unprefixed-skill", + "v1->v2 skills", + current_schema_fixture(1, |conn| { + conn.execute_batch( + "DROP TABLE skills; + CREATE TABLE skills ( + key TEXT PRIMARY KEY, + installed BOOLEAN NOT NULL DEFAULT 0, + installed_at INTEGER NOT NULL DEFAULT 0 + ); + INSERT INTO skills (key, installed, installed_at) + VALUES ('missing-prefix', 1, 10);", + )?; + Ok(()) + })?, + ), + ( + "v2-nonempty-skills", + "v2->v3 skills SSOT rebuild", + current_schema_fixture(2, |conn| { + conn.execute_batch( + "DROP TABLE skills; + CREATE TABLE skills ( + directory TEXT NOT NULL, + app_type TEXT NOT NULL, + installed BOOLEAN NOT NULL DEFAULT 0, + installed_at INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (directory, app_type) + ); + INSERT INTO skills + (directory, app_type, installed, installed_at) + VALUES ('legacy-skill', 'claude', 1, 10);", + )?; + Ok(()) + })?, + ), + ( + "v5-invalid-provider-meta", + "v5->v6 provider meta", + current_schema_fixture(5, |conn| { + conn.execute( + "INSERT INTO providers + (id, app_type, name, settings_config, meta) + VALUES ('bad-meta', 'claude', 'bad meta', '{}', '{')", + [], + )?; + Ok(()) + })?, + ), + ( + "v5-provider-meta-normalization", + "v5->v6 provider meta", + current_schema_fixture(5, |conn| { + conn.execute( + "INSERT INTO providers + (id, app_type, name, settings_config, meta) + VALUES ( + 'copilot-meta', 'claude', 'copilot meta', '{}', + '{\"usage_script\":{\"template_type\":\"copilot\"}}' + )", + [], + )?; + Ok(()) + })?, + ), + ( + "v7-pricing-correction", + "v7->v8 model pricing correction", + current_schema_fixture(7, |conn| { + conn.execute( + "UPDATE model_pricing + SET input_cost_per_million = '999' + WHERE model_id = 'deepseek-v3.2'", + [], + )?; + Ok(()) + })?, + ), + ( + "v8-pricing-reseed", + "v8->v9 model-pricing replacement", + current_schema_fixture(8, |conn| { + conn.execute( + "INSERT INTO model_pricing ( + model_id, display_name, input_cost_per_million, + output_cost_per_million, cache_read_cost_per_million, + cache_creation_cost_per_million + ) VALUES ('untrusted-extra', 'extra', '1', '1', '0', '0')", + [], + )?; + Ok(()) + })?, + ), + ( + "v10-rollup-synthesis", + "v10->v11 usage rollup rebuild", + current_schema_fixture(10, |conn| { + conn.execute( + "INSERT INTO usage_daily_rollups + (date, app_type, provider_id, model) + VALUES ('2026-08-01', 'claude', 'legacy', 'legacy-model')", + [], + )?; + Ok(()) + })?, + ), + ( + "v13-grokbuild-synthesis", + "v13->v14 proxy rebuild", + current_schema_fixture(13, |conn| { + conn.execute("DELETE FROM proxy_config WHERE app_type = 'grokbuild'", [])?; + Ok(()) + })?, + ), + ( + "v13-missing-column-fallback", + "v13->v14 proxy rebuild", + current_schema_fixture(13, |conn| { + conn.execute_batch("ALTER TABLE proxy_config DROP COLUMN live_takeover_active;")?; + Ok(()) + })?, + ), + ( + "v13-wrong-storage", + "v13->v14 proxy rebuild", + current_schema_fixture(13, |conn| { + conn.execute( + "UPDATE proxy_config SET listen_port = X'00' + WHERE app_type = 'claude'", + [], + )?; + Ok(()) + })?, + ), + ( + "v15-codex-log-reset", + "v15->v16 Codex usage reset", + current_schema_fixture(15, |conn| { + conn.execute( + "INSERT INTO proxy_request_logs ( + request_id, provider_id, app_type, model, + latency_ms, status_code, created_at, data_source + ) VALUES ( + 'untrusted-codex-log', '_codex_session', 'codex', 'legacy', + 1, 200, 1, 'codex_session' + )", + [], + )?; + Ok(()) + })?, + ), + ]; + + for (label, expected_step, fixture) in fixtures { + assert_fixture_rejected(label, expected_step, &fixture)?; + } + Ok(()) +} + +#[test] +fn migration_classification_and_no_swallowing_guard_cover_v1_through_v17() { + let source = include_str!("schema.rs"); + for from in 1..17 { + let to = from + 1; + let classification_row = format!("| v{from}→v{to}"); + assert!( + source.contains(&classification_row), + "migration classification table is missing v{from}->v{to}" + ); + let dispatch = format!("Self::migrate_v{from}_to_v{to}(conn, context)?;"); + assert!( + source.contains(&dispatch), + "v{from}->v{to} must receive MigrationRunContext at dispatch" + ); + } + + let migration_start = source + .find("fn migrate_v1_to_v2") + .expect("v1 migration exists"); + let migration_end = source + .find("/// 插入默认模型定价数据") + .expect("migration span end exists"); + let migration_span = &source[migration_start..migration_end]; + for forbidden in [ + ".unwrap_or(", + ".unwrap_or_else(", + ".unwrap_or_default(", + ".ok()", + "if let Ok(", + "let _ =", + ] { + assert!( + !migration_span.contains(forbidden), + "historical migration span contains forbidden error swallowing: {forbidden}" + ); + } + assert!( + source.contains("untrusted schema migration v{version}->v{} failed"), + "the migration boundary must map non-structured failures to InvalidInput" + ); +} diff --git a/src-tauri/src/database/mod.rs b/src-tauri/src/database/mod.rs index 9cc55b78b..b156c0ff7 100644 --- a/src-tauri/src/database/mod.rs +++ b/src-tauri/src/database/mod.rs @@ -26,6 +26,8 @@ pub(crate) mod backup; #[cfg(test)] mod backup_restore_certification; +#[cfg(test)] +mod backup_restore_certification_ext; mod dao; mod migration; mod schema; diff --git a/src-tauri/src/database/schema.rs b/src-tauri/src/database/schema.rs index 1fd3234e6..ea1c61c9f 100644 --- a/src-tauri/src/database/schema.rs +++ b/src-tauri/src/database/schema.rs @@ -28,12 +28,48 @@ impl CanonicalStage { /// Selects whether schema migrations may repair locally trusted historical /// data or must fail closed on an input that would require reconciliation. +/// +/// # Historical migration classification (blind-review authority) +/// +/// “Pure shape” means the migration has a deterministic version-defined +/// mapping and never chooses a value because an input read failed. A declared +/// column default or an exact version sentinel belongs to that mapping. +/// “Repair/synthesis” means the old implementation guessed, discarded, +/// reconciled, or regenerated portable data. Under `UntrustedRestore`, that +/// branch may only prove that the postcondition already holds; if it would +/// change or invent portable data, it returns a structured `InvalidInput`. +/// +/// | Step | Class | `UntrustedRestore` decision and basis | +/// | --- | --- | --- | +/// | v1→v2 column/table/index additions | Pure shape | Exact DDL; every SQLite error propagates. | +/// | v1→v2 proxy/settings/circuit conversion | Repair/synthesis | Requires the authoritative `id=1` rows and decodes every field; missing/malformed data aborts instead of selecting defaults. | +/// | v1→v2 skill-key conversion | Repair/synthesis | Prefixed keys map exactly; the legacy “missing prefix means claude” fallback aborts. | +/// | v1→v2 model-pricing replacement | Repair/synthesis | Seed-owned input must be empty or already equal the canonical seed; otherwise reseeding aborts. | +/// | v2→v3 skill SSOT rebuild | Repair/synthesis | Empty old storage is a shape rebuild; discarding non-empty rows for a later filesystem scan aborts. | +/// | v3→v4 | Pure shape | Adds OpenCode enablement columns with version-defined false sentinels. | +/// | v4→v5 | Pure shape | Adds billing/request-model columns with version-defined sentinels. | +/// | v5→v6 rollup DDL | Pure shape | Exact table creation. | +/// | v5→v6 provider-meta normalization | Repair/synthesis | Invalid JSON and `copilot` values requiring normalization abort; valid already-normalized JSON is preserved. | +/// | v6→v7 | Pure shape | Adds skill hash/timestamp columns with version-defined sentinels. | +/// | v7→v8 log/session DDL | Pure shape | Exact column/table/index creation. | +/// | v7→v8 pricing correction | Repair/synthesis | Empty/current-canonical pricing is a no-op; otherwise affected rows must already contain the declared corrected tuple or the overwrite aborts. | +/// | v8→v9 pricing reseed | Repair/synthesis | Seed-owned input must be empty or equal the canonical seed; otherwise replacement aborts. | +/// | v9→v10 | Pure shape | Adds Hermes enablement columns with version-defined false sentinels. | +/// | v10→v11 rollup rebuild | Repair/synthesis | Empty storage may be reshaped; non-empty rows whose request/pricing model would be synthesized as `''` abort. | +/// | v11→v12 | Pure shape | Creates the profiles table without rewriting rows. | +/// | v12→v13 | Pure shape | Adds explicit “unknown” token-semantics sentinel columns. | +/// | v13→v14 proxy rebuild | Repair/synthesis | A non-empty source requires every column, its declared storage class, and an existing Grok Build row; empty-table shape conversion is allowed, while fallback values for rows or row synthesis abort. | +/// | v14→v15 | Pure shape | Adds Grok Build enablement columns with version-defined false sentinels. | +/// | v15→v16 Codex reset | Repair/synthesis | Any runtime row that the reset could discard aborts; an empty runtime set is a no-op. | +/// | v16→v17 endpoint rebuild/ledger DDL | Pure shape + repair/synthesis | Canonical rebuild and empty device-ledger creation are exact; duplicate endpoint reconciliation aborts and is never executed. | #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum MigrationRunContext { LocalUpgrade, UntrustedRestore, } +type ModelPricingSnapshotRow = (String, String, String, String, String, String); + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub(crate) enum CanonicalRestoreClass { @@ -931,7 +967,7 @@ impl Database { if Self::table_exists(conn, "proxy_config")? && !Self::has_column(conn, "proxy_config", "app_type")? { - Self::migrate_proxy_config_to_per_app(conn)?; + Self::migrate_proxy_config_to_per_app(conn, MigrationRunContext::LocalUpgrade)?; } Self::add_column_if_missing( @@ -970,8 +1006,7 @@ impl Database { let mut version = Self::get_user_version(conn)?; if version > SCHEMA_VERSION { - conn.execute("ROLLBACK TO schema_migration;", []).ok(); - conn.execute("RELEASE schema_migration;", []).ok(); + Self::rollback_schema_migration_savepoint(conn); return Err(AppError::Database(format!( "数据库版本过新（{version}），当前应用仅支持 {SCHEMA_VERSION}，请升级应用后再尝试。" ))); @@ -982,84 +1017,84 @@ impl Database { match version { 0 => { log::info!("检测到 user_version=0，迁移到 1（补齐缺失列并设置版本）"); - Self::migrate_v0_to_v1(conn)?; + Self::migrate_v0_to_v1(conn, context)?; Self::set_user_version(conn, 1)?; } 1 => { log::info!( "迁移数据库从 v1 到 v2（添加使用统计表和完整字段，重构 skills 表）" ); - Self::migrate_v1_to_v2(conn)?; + Self::migrate_v1_to_v2(conn, context)?; Self::set_user_version(conn, 2)?; } 2 => { log::info!("迁移数据库从 v2 到 v3（Skills 统一管理架构）"); - Self::migrate_v2_to_v3(conn)?; + Self::migrate_v2_to_v3(conn, context)?; Self::set_user_version(conn, 3)?; } 3 => { log::info!("迁移数据库从 v3 到 v4（OpenCode 支持）"); - Self::migrate_v3_to_v4(conn)?; + Self::migrate_v3_to_v4(conn, context)?; Self::set_user_version(conn, 4)?; } 4 => { log::info!("迁移数据库从 v4 到 v5（计费模式支持）"); - Self::migrate_v4_to_v5(conn)?; + Self::migrate_v4_to_v5(conn, context)?; Self::set_user_version(conn, 5)?; } 5 => { log::info!("迁移数据库从 v5 到 v6（使用量聚合表 + Copilot 模板类型统一）"); - Self::migrate_v5_to_v6(conn)?; + Self::migrate_v5_to_v6(conn, context)?; Self::set_user_version(conn, 6)?; } 6 => { log::info!("迁移数据库从 v6 到 v7（Skills 更新检测支持）"); - Self::migrate_v6_to_v7(conn)?; + Self::migrate_v6_to_v7(conn, context)?; Self::set_user_version(conn, 7)?; } 7 => { log::info!("迁移数据库从 v7 到 v8（会话日志使用追踪 + 修正模型定价）"); - Self::migrate_v7_to_v8(conn)?; + Self::migrate_v7_to_v8(conn, context)?; Self::set_user_version(conn, 8)?; } 8 => { log::info!("迁移数据库从 v8 到 v9（全面补充模型定价）"); - Self::migrate_v8_to_v9(conn)?; + Self::migrate_v8_to_v9(conn, context)?; Self::set_user_version(conn, 9)?; } 9 => { log::info!("迁移数据库从 v9 到 v10（添加 Hermes Agent 支持）"); - Self::migrate_v9_to_v10(conn)?; + Self::migrate_v9_to_v10(conn, context)?; Self::set_user_version(conn, 10)?; } 10 => { log::info!("迁移数据库从 v10 到 v11（usage_daily_rollups 保留 request_model 维度）"); - Self::migrate_v10_to_v11(conn)?; + Self::migrate_v10_to_v11(conn, context)?; Self::set_user_version(conn, 11)?; } 11 => { log::info!("迁移数据库从 v11 到 v12（添加项目 Profiles 表）"); - Self::migrate_v11_to_v12(conn)?; + Self::migrate_v11_to_v12(conn, context)?; Self::set_user_version(conn, 12)?; } 12 => { log::info!("迁移数据库从 v12 到 v13（记录输入 token 缓存语义）"); - Self::migrate_v12_to_v13(conn)?; + Self::migrate_v12_to_v13(conn, context)?; Self::set_user_version(conn, 13)?; } 13 => { log::info!("迁移数据库从 v13 到 v14（添加 Grok Build 代理配置）"); - Self::migrate_v13_to_v14(conn)?; + Self::migrate_v13_to_v14(conn, context)?; Self::set_user_version(conn, 14)?; } 14 => { log::info!("迁移数据库从 v14 到 v15（Skills/MCP 添加 Grok Build 支持）"); - Self::migrate_v14_to_v15(conn)?; + Self::migrate_v14_to_v15(conn, context)?; Self::set_user_version(conn, 15)?; } 15 => { log::info!("迁移数据库从 v15 到 v16（重建 Codex 会话用量）"); - Self::migrate_v15_to_v16(conn)?; + Self::migrate_v15_to_v16(conn, context)?; Self::set_user_version(conn, 16)?; } 16 => { @@ -1087,15 +1122,146 @@ impl Database { Ok(()) } Err(e) => { - conn.execute("ROLLBACK TO schema_migration;", []).ok(); - conn.execute("RELEASE schema_migration;", []).ok(); - Err(e) + Self::rollback_schema_migration_savepoint(conn); + if context == MigrationRunContext::UntrustedRestore { + match e { + AppError::InvalidInput(_) => Err(e), + other => Err(AppError::InvalidInput(format!( + "untrusted schema migration v{version}->v{} failed: {other}", + version + 1 + ))), + } + } else { + Err(e) + } } } } + fn rollback_schema_migration_savepoint(conn: &Connection) { + if let Err(error) = conn.execute("ROLLBACK TO schema_migration;", []) { + log::error!("failed to roll back schema migration savepoint: {error}"); + } + if let Err(error) = conn.execute("RELEASE schema_migration;", []) { + log::error!("failed to release failed schema migration savepoint: {error}"); + } + } + + fn untrusted_repair_error(step: &str, detail: impl std::fmt::Display) -> AppError { + AppError::InvalidInput(format!( + "untrusted migration {step} requires forbidden repair or synthesis: {detail}" + )) + } + + fn legacy_query_or_local_default( + context: MigrationRunContext, + step: &str, + result: rusqlite::Result, + local_default: T, + ) -> Result { + match result { + Ok(value) => Ok(value), + Err(error) if context == MigrationRunContext::UntrustedRestore => { + Err(Self::untrusted_repair_error( + step, + format!("authoritative row is missing or undecodable: {error}"), + )) + } + Err(_) => Ok(local_default), + } + } + + fn legacy_bool_or_local_false( + conn: &Connection, + context: MigrationRunContext, + step: &str, + key: &str, + ) -> Result { + let value = conn + .query_row("SELECT value FROM settings WHERE key = ?1", [key], |row| { + row.get::<_, String>(0) + }) + .optional(); + match value { + Ok(Some(value)) => match value.as_str() { + "true" | "1" => Ok(true), + "false" | "0" => Ok(false), + _ if context == MigrationRunContext::UntrustedRestore => { + Err(Self::untrusted_repair_error( + step, + format!("{key} has unsupported boolean value {value:?}"), + )) + } + _ => Ok(false), + }, + Ok(None) if context == MigrationRunContext::LocalUpgrade => Ok(false), + Ok(None) => Err(Self::untrusted_repair_error( + step, + format!("required key {key:?} is missing"), + )), + Err(error) if context == MigrationRunContext::UntrustedRestore => Err( + Self::untrusted_repair_error(step, format!("failed to decode {key:?}: {error}")), + ), + Err(_) => Ok(false), + } + } + + fn require_untrusted_column_storage( + conn: &Connection, + context: MigrationRunContext, + step: &str, + table: &str, + columns: &[(&str, &str)], + ) -> Result<(), AppError> { + if context == MigrationRunContext::LocalUpgrade { + return Ok(()); + } + for (column, expected_storage) in columns { + if !Self::has_column(conn, table, column)? { + return Err(Self::untrusted_repair_error( + step, + format!("{table}.{column} is missing and would use a fallback expression"), + )); + } + let sql = format!( + "SELECT typeof(\"{column}\") FROM \"{table}\" + WHERE typeof(\"{column}\") <> ?1 LIMIT 1" + ); + let observed = conn + .query_row(&sql, [expected_storage], |row| row.get::<_, String>(0)) + .optional() + .map_err(|error| { + AppError::Database(format!("validate storage for {table}.{column}: {error}")) + })?; + if let Some(observed) = observed { + return Err(Self::untrusted_repair_error( + step, + format!( + "{table}.{column} has storage class {observed}, expected {expected_storage}" + ), + )); + } + } + Ok(()) + } + + fn require_no_untrusted_repair_rows( + context: MigrationRunContext, + step: &str, + source: &str, + row_count: i64, + ) -> Result<(), AppError> { + if context == MigrationRunContext::UntrustedRestore && row_count != 0 { + return Err(Self::untrusted_repair_error( + step, + format!("{source} contains {row_count} rows that the repair could discard"), + )); + } + Ok(()) + } + /// v0 -> v1 迁移：补齐所有缺失列 - fn migrate_v0_to_v1(conn: &Connection) -> Result<(), AppError> { + fn migrate_v0_to_v1(conn: &Connection, _context: MigrationRunContext) -> Result<(), AppError> { // providers 表 Self::add_column_if_missing(conn, "providers", "category", "TEXT")?; Self::add_column_if_missing(conn, "providers", "created_at", "INTEGER")?; @@ -1155,7 +1321,7 @@ impl Database { } /// v1 -> v2 迁移：添加使用统计表和完整字段，重构 skills 表 - fn migrate_v1_to_v2(conn: &Connection) -> Result<(), AppError> { + fn migrate_v1_to_v2(conn: &Connection, context: MigrationRunContext) -> Result<(), AppError> { // providers 表字段 Self::add_column_if_missing( conn, @@ -1267,6 +1433,12 @@ impl Database { Self::add_column_if_missing(conn, "proxy_request_logs", "first_token_ms", "INTEGER")?; Self::add_column_if_missing(conn, "proxy_request_logs", "duration_ms", "INTEGER")?; + // Validate/convert the two legacy structures before any data seed + // replacement. This makes a malformed legacy row the reported cause + // instead of masking it behind a later repair decision. + Self::migrate_skills_table(conn, context)?; + Self::migrate_proxy_config_to_per_app(conn, context)?; + // model_pricing 表 conn.execute( "CREATE TABLE IF NOT EXISTS model_pricing ( @@ -1278,22 +1450,23 @@ impl Database { [], )?; - // 清空并重新插入模型定价 - conn.execute("DELETE FROM model_pricing", []) - .map_err(|e| AppError::Database(format!("清空模型定价失败: {e}")))?; - Self::seed_model_pricing(conn)?; - - // 重构 skills 表（添加 app_type 字段） - Self::migrate_skills_table(conn)?; - - // 重构 proxy_config 为三行结构（每应用独立配置） - Self::migrate_proxy_config_to_per_app(conn)?; + if context == MigrationRunContext::UntrustedRestore { + Self::require_canonical_model_pricing(conn, "v1->v2 model-pricing replacement")?; + } else { + // Local upgrades preserve the historical repair behavior. + conn.execute("DELETE FROM model_pricing", []) + .map_err(|e| AppError::Database(format!("清空模型定价失败: {e}")))?; + Self::seed_model_pricing(conn)?; + } Ok(()) } /// 将 proxy_config 迁移为三行结构（每应用独立配置） - fn migrate_proxy_config_to_per_app(conn: &Connection) -> Result<(), AppError> { + fn migrate_proxy_config_to_per_app( + conn: &Connection, + context: MigrationRunContext, + ) -> Result<(), AppError> { // 检查是否已经是新表结构（幂等性） if !Self::table_exists(conn, "proxy_config")? { // 表不存在，跳过迁移（新安装） @@ -1307,46 +1480,65 @@ impl Database { } // 读取旧配置 - let old_config = conn - .query_row( - "SELECT listen_address, listen_port, max_retries, enable_logging, + let old_config_result = conn.query_row( + "SELECT listen_address, listen_port, max_retries, enable_logging, streaming_first_byte_timeout, streaming_idle_timeout, non_streaming_timeout FROM proxy_config WHERE id = 1", + [], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, i32>(1)?, + row.get::<_, i32>(2)?, + row.get::<_, i32>(3)?, + row.get::<_, i32>(4)?, + row.get::<_, i32>(5)?, + row.get::<_, i32>(6)?, + )) + }, + ); + let old_config = Self::legacy_query_or_local_default( + context, + "v1->v2 proxy_config id=1", + old_config_result, + ("127.0.0.1".to_string(), 5000, 3, 1, 30, 60, 300), + )?; + + let old_cb_result = if Self::table_exists(conn, "circuit_breaker_config")? { + conn.query_row( + "SELECT failure_threshold, success_threshold, timeout_seconds, + error_rate_threshold, min_requests + FROM circuit_breaker_config WHERE id = 1", [], |row| { Ok(( - row.get::<_, String>(0)?, + row.get::<_, i32>(0)?, row.get::<_, i32>(1)?, - row.get::<_, i32>(2)?, - row.get::<_, i32>(3)?, - row.get::<_, i32>(4).unwrap_or(30), - row.get::<_, i32>(5).unwrap_or(60), - row.get::<_, i32>(6).unwrap_or(300), + row.get::<_, i64>(2)?, + row.get::<_, f64>(3)?, + row.get::<_, i32>(4)?, )) }, ) - .unwrap_or_else(|_| ("127.0.0.1".to_string(), 5000, 3, 1, 30, 60, 300)); + } else { + Err(rusqlite::Error::QueryReturnedNoRows) + }; + let old_cb = Self::legacy_query_or_local_default( + context, + "v1->v2 circuit_breaker_config id=1", + old_cb_result, + (5, 2, 60, 0.5, 10), + )?; - let old_cb = conn.query_row( - "SELECT failure_threshold, success_threshold, timeout_seconds, error_rate_threshold, min_requests - FROM circuit_breaker_config WHERE id = 1", [], - |row| Ok((row.get::<_, i32>(0)?, row.get::<_, i32>(1)?, row.get::<_, i64>(2)?, - row.get::<_, f64>(3)?, row.get::<_, i32>(4)?)) - ).unwrap_or((5, 2, 60, 0.5, 10)); - - let get_bool = |key: &str| -> bool { - conn.query_row("SELECT value FROM settings WHERE key = ?", [key], |r| { - r.get::<_, String>(0) - }) - .map(|v| v == "true" || v == "1") - .unwrap_or(false) + let get_bool = |key: &str| { + Self::legacy_bool_or_local_false(conn, context, "v1->v2 proxy ownership settings", key) }; let apps = [ ( "claude", - get_bool("proxy_takeover_claude"), - get_bool("auto_failover_enabled_claude"), + get_bool("proxy_takeover_claude")?, + get_bool("auto_failover_enabled_claude")?, 6, 45, 90, @@ -1358,8 +1550,8 @@ impl Database { ), ( "codex", - get_bool("proxy_takeover_codex"), - get_bool("auto_failover_enabled_codex"), + get_bool("proxy_takeover_codex")?, + get_bool("auto_failover_enabled_codex")?, 3, old_config.4, old_config.5, @@ -1371,8 +1563,8 @@ impl Database { ), ( "gemini", - get_bool("proxy_takeover_gemini"), - get_bool("auto_failover_enabled_gemini"), + get_bool("proxy_takeover_gemini")?, + get_bool("auto_failover_enabled_gemini")?, 5, old_config.4, old_config.5, @@ -1444,7 +1636,10 @@ impl Database { } /// 迁移 skills 表：从单 key 主键改为 (directory, app_type) 复合主键 - fn migrate_skills_table(conn: &Connection) -> Result<(), AppError> { + fn migrate_skills_table( + conn: &Connection, + context: MigrationRunContext, + ) -> Result<(), AppError> { // v3 结构（统一管理架构）已经是更高版本的 skills 表： // - 主键为 id // - 包含 enabled_claude / enabled_codex / enabled_gemini 等列 @@ -1503,9 +1698,19 @@ impl Database { for (key, installed, installed_at) in old_skills { // 解析 key: "app:directory" 或 "directory"（默认 claude） - let (app_type, directory) = if let Some(idx) = key.find(':') { - let (app, dir) = key.split_at(idx); - (app.to_string(), dir[1..].to_string()) // 跳过冒号 + let (app_type, directory) = if let Some((app, directory)) = key.split_once(':') { + if app.is_empty() || directory.is_empty() { + return Err(Self::untrusted_repair_error( + "v1->v2 skills", + format!("legacy skill key {key:?} has an empty component"), + )); + } + (app.to_string(), directory.to_string()) + } else if context == MigrationRunContext::UntrustedRestore { + return Err(Self::untrusted_repair_error( + "v1->v2 skills", + format!("legacy skill key {key:?} requires a synthesized claude prefix"), + )); } else { ("claude".to_string(), key.clone()) }; @@ -1535,7 +1740,7 @@ impl Database { /// 迁移策略： /// 1. 旧数据库只存储安装记录，真正的 skill 文件在文件系统 /// 2. 直接重建新表结构，后续由 SkillService 在首次启动时扫描文件系统重建数据 - fn migrate_v2_to_v3(conn: &Connection) -> Result<(), AppError> { + fn migrate_v2_to_v3(conn: &Connection, context: MigrationRunContext) -> Result<(), AppError> { // 检查是否已经是新结构（通过检查是否有 enabled_claude 列） if Self::has_column(conn, "skills", "enabled_claude")? { log::info!("skills 表已经是 v3 结构，跳过迁移"); @@ -1547,39 +1752,52 @@ impl Database { // 1. 备份旧数据（用于日志和后续启动迁移） let old_count: i64 = conn .query_row("SELECT COUNT(*) FROM skills", [], |row| row.get(0)) - .unwrap_or(0); + .map_err(|error| AppError::Database(format!("统计旧 skills 数据失败: {error}")))?; log::info!("旧 skills 表有 {old_count} 条记录"); - let mut stmt = conn - .prepare( - "SELECT directory, app_type FROM skills - WHERE installed = 1", - ) - .map_err(|e| AppError::Database(format!("查询旧 skills 快照失败: {e}")))?; - let snapshot_rows: Vec = stmt - .query_map([], |row| { - Ok(LegacySkillMigrationRow { - directory: row.get(0)?, - app_type: row.get(1)?, - }) - }) - .map_err(|e| AppError::Database(format!("读取旧 skills 快照失败: {e}")))? - .collect::, _>>() - .map_err(|e| AppError::Database(format!("解析旧 skills 快照失败: {e}")))?; - let snapshot_json = serde_json::to_string(&snapshot_rows) - .map_err(|e| AppError::Database(format!("序列化旧 skills 快照失败: {e}")))?; + Self::require_no_untrusted_repair_rows( + context, + "v2->v3 skills SSOT rebuild", + "legacy skills", + old_count, + )?; - // 标记：需要在启动后从文件系统扫描并重建 Skills 数据 - // 说明：v3 结构将 Skills 的 SSOT 迁移到 ~/.cc-switch/skills/， - // 旧表只存“安装记录”，无法直接无损迁移到新结构，因此改为启动后扫描 app 目录导入。 - let _ = conn.execute( - "INSERT OR REPLACE INTO settings (key, value) VALUES ('skills_ssot_migration_pending', 'true')", - [], - ); - let _ = conn.execute( - "INSERT OR REPLACE INTO settings (key, value) VALUES ('skills_ssot_migration_snapshot', ?1)", - [snapshot_json], - ); + if context == MigrationRunContext::LocalUpgrade { + let mut stmt = conn + .prepare( + "SELECT directory, app_type FROM skills + WHERE installed = 1", + ) + .map_err(|e| AppError::Database(format!("查询旧 skills 快照失败: {e}")))?; + let snapshot_rows: Vec = stmt + .query_map([], |row| { + Ok(LegacySkillMigrationRow { + directory: row.get(0)?, + app_type: row.get(1)?, + }) + }) + .map_err(|e| AppError::Database(format!("读取旧 skills 快照失败: {e}")))? + .collect::, _>>() + .map_err(|e| AppError::Database(format!("解析旧 skills 快照失败: {e}")))?; + let snapshot_json = serde_json::to_string(&snapshot_rows) + .map_err(|e| AppError::Database(format!("序列化旧 skills 快照失败: {e}")))?; + + // 标记：需要在启动后从文件系统扫描并重建 Skills 数据 + // 说明：v3 结构将 Skills 的 SSOT 迁移到 ~/.cc-switch/skills/， + // 旧表只存“安装记录”，无法直接无损迁移到新结构，因此改为启动后扫描 app 目录导入。 + conn.execute( + "INSERT OR REPLACE INTO settings (key, value) + VALUES ('skills_ssot_migration_pending', 'true')", + [], + ) + .map_err(|error| AppError::Database(format!("写入 skills 迁移标记失败: {error}")))?; + conn.execute( + "INSERT OR REPLACE INTO settings (key, value) + VALUES ('skills_ssot_migration_snapshot', ?1)", + [snapshot_json], + ) + .map_err(|error| AppError::Database(format!("写入 skills 迁移快照失败: {error}")))?; + } // 2. 删除旧表 conn.execute("DROP TABLE IF EXISTS skills", []) @@ -1616,7 +1834,7 @@ impl Database { /// v3 -> v4 迁移：添加 OpenCode 支持 /// /// 为 mcp_servers 和 skills 表添加 enabled_opencode 列。 - fn migrate_v3_to_v4(conn: &Connection) -> Result<(), AppError> { + fn migrate_v3_to_v4(conn: &Connection, _context: MigrationRunContext) -> Result<(), AppError> { // 为 mcp_servers 表添加 enabled_opencode 列 Self::add_column_if_missing( conn, @@ -1638,7 +1856,7 @@ impl Database { } /// v4 -> v5 迁移：新增计费模式配置与请求模型字段 - fn migrate_v4_to_v5(conn: &Connection) -> Result<(), AppError> { + fn migrate_v4_to_v5(conn: &Connection, _context: MigrationRunContext) -> Result<(), AppError> { if Self::table_exists(conn, "proxy_config")? { Self::add_column_if_missing( conn, @@ -1662,7 +1880,7 @@ impl Database { } /// v5 -> v6 迁移：添加使用量日聚合表 + 统一 Copilot 模板类型 - fn migrate_v5_to_v6(conn: &Connection) -> Result<(), AppError> { + fn migrate_v5_to_v6(conn: &Connection, context: MigrationRunContext) -> Result<(), AppError> { // 1. 添加使用量日聚合表 conn.execute( "CREATE TABLE IF NOT EXISTS usage_daily_rollups ( @@ -1703,24 +1921,39 @@ impl Database { for row in rows { let (id, app_type, meta_str) = row.map_err(|e| AppError::Database(e.to_string()))?; - if let Ok(mut meta) = serde_json::from_str::(&meta_str) { - let mut updated = false; + let mut meta = match serde_json::from_str::(&meta_str) { + Ok(meta) => meta, + Err(error) if context == MigrationRunContext::UntrustedRestore => { + return Err(Self::untrusted_repair_error( + "v5->v6 provider meta", + format!("provider ({id}, {app_type}) has invalid JSON: {error}"), + )); + } + Err(_) => continue, + }; + let mut updated = false; - if let Some(usage_script) = meta.get_mut("usage_script") { - if let Some(template_type) = usage_script.get_mut("template_type") { - if template_type == "copilot" { - *template_type = - serde_json::Value::String("github_copilot".to_string()); - updated = true; + if let Some(usage_script) = meta.get_mut("usage_script") { + if let Some(template_type) = usage_script.get_mut("template_type") { + if template_type == "copilot" { + if context == MigrationRunContext::UntrustedRestore { + return Err(Self::untrusted_repair_error( + "v5->v6 provider meta", + format!( + "provider ({id}, {app_type}) requires copilot template normalization" + ), + )); } + *template_type = serde_json::Value::String("github_copilot".to_string()); + updated = true; } } + } - if updated { - let new_meta_str = serde_json::to_string(&meta) - .map_err(|e| AppError::Database(e.to_string()))?; - updates.push((id, app_type, new_meta_str)); - } + if updated { + let new_meta_str = + serde_json::to_string(&meta).map_err(|e| AppError::Database(e.to_string()))?; + updates.push((id, app_type, new_meta_str)); } } @@ -1737,7 +1970,7 @@ impl Database { } /// v6 -> v7: Skills 更新检测支持（content_hash + updated_at） - fn migrate_v6_to_v7(conn: &Connection) -> Result<(), AppError> { + fn migrate_v6_to_v7(conn: &Connection, _context: MigrationRunContext) -> Result<(), AppError> { if Self::table_exists(conn, "skills")? { Self::add_column_if_missing(conn, "skills", "content_hash", "TEXT")?; Self::add_column_if_missing( @@ -1752,7 +1985,7 @@ impl Database { } /// v7 -> v8: 会话日志使用追踪（无代理模式统计支持） - fn migrate_v7_to_v8(conn: &Connection) -> Result<(), AppError> { + fn migrate_v7_to_v8(conn: &Connection, context: MigrationRunContext) -> Result<(), AppError> { // 1. 为 proxy_request_logs 添加 data_source 列，区分数据来源 if Self::table_exists(conn, "proxy_request_logs")? { Self::add_column_if_missing( @@ -1778,6 +2011,14 @@ impl Database { // 3. 修正国产模型定价：之前误将 CNY 值存为 USD 字段，统一转换为 USD if Self::table_exists(conn, "model_pricing")? { + if context == MigrationRunContext::UntrustedRestore + && Self::model_pricing_is_empty_or_canonical(conn)? + { + log::info!( + "v7 -> v8 untrusted restore: pricing is empty/canonical; repair is a no-op" + ); + return Ok(()); + } let pricing_fixes: &[(&str, &str, &str, &str, &str)] = &[ ("deepseek-v3.2", "0.28", "0.42", "0.028", "0"), ("deepseek-v3.1", "0.55", "1.67", "0.055", "0"), @@ -1794,6 +2035,43 @@ impl Database { ("mimo-v2-flash", "0.09", "0.29", "0.009", "0"), ]; for (model_id, input, output, cache_read, cache_creation) in pricing_fixes { + if context == MigrationRunContext::UntrustedRestore { + let current = conn + .query_row( + "SELECT input_cost_per_million, output_cost_per_million, + cache_read_cost_per_million, + cache_creation_cost_per_million + FROM model_pricing WHERE model_id = ?1", + [model_id], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + )) + }, + ) + .optional() + .map_err(|error| { + AppError::Database(format!("读取模型 {model_id} 定价失败: {error}")) + })?; + if let Some(current) = current { + let expected = ( + (*input).to_string(), + (*output).to_string(), + (*cache_read).to_string(), + (*cache_creation).to_string(), + ); + if current != expected { + return Err(Self::untrusted_repair_error( + "v7->v8 model pricing correction", + format!("model {model_id} differs from the corrected tuple"), + )); + } + } + continue; + } conn.execute( "UPDATE model_pricing SET input_cost_per_million = ?2, @@ -1812,7 +2090,7 @@ impl Database { } /// v8 → v9: 全面补充模型定价（清空 + 重新 seed） - fn migrate_v8_to_v9(conn: &Connection) -> Result<(), AppError> { + fn migrate_v8_to_v9(conn: &Connection, context: MigrationRunContext) -> Result<(), AppError> { conn.execute( "CREATE TABLE IF NOT EXISTS model_pricing ( model_id TEXT PRIMARY KEY, display_name TEXT NOT NULL, @@ -1823,6 +2101,9 @@ impl Database { [], ) .map_err(|e| AppError::Database(format!("创建 model_pricing 表失败: {e}")))?; + if context == MigrationRunContext::UntrustedRestore { + return Self::require_canonical_model_pricing(conn, "v8->v9 model-pricing replacement"); + } conn.execute("DELETE FROM model_pricing", []) .map_err(|e| AppError::Database(format!("清空模型定价失败: {e}")))?; Self::seed_model_pricing(conn)?; @@ -1831,7 +2112,7 @@ impl Database { } /// v9 -> v10 迁移：添加 Hermes Agent 支持 - fn migrate_v9_to_v10(conn: &Connection) -> Result<(), AppError> { + fn migrate_v9_to_v10(conn: &Connection, _context: MigrationRunContext) -> Result<(), AppError> { Self::add_column_if_missing( conn, "mcp_servers", @@ -1859,7 +2140,7 @@ impl Database { /// 路由接管下 model（真实上游模型）≠ request_model（客户端别名）， /// 旧 rollup 只按 model 聚合，明细 prune 后映射关系永久丢失、计费不可审计。 /// SQLite 改主键必须重建表；历史行的 request_model 已不可知，填 ''。 - fn migrate_v10_to_v11(conn: &Connection) -> Result<(), AppError> { + fn migrate_v10_to_v11(conn: &Connection, context: MigrationRunContext) -> Result<(), AppError> { // proxy_request_logs.pricing_model：NULL = v11 前的历史行（回填走 // model → 占位符回退 request_model 的旧逻辑），'' = 未计价的错误行 if Self::table_exists(conn, "proxy_request_logs")? { @@ -1871,6 +2152,22 @@ impl Database { return Ok(()); } + if context == MigrationRunContext::UntrustedRestore { + let row_count: i64 = conn + .query_row("SELECT COUNT(*) FROM usage_daily_rollups", [], |row| { + row.get(0) + }) + .map_err(|error| { + AppError::Database(format!("统计 v10 usage_daily_rollups 失败: {error}")) + })?; + Self::require_no_untrusted_repair_rows( + context, + "v10->v11 usage rollup rebuild", + "legacy usage_daily_rollups", + row_count, + )?; + } + conn.execute_batch( "ALTER TABLE usage_daily_rollups RENAME TO usage_daily_rollups_v10; CREATE TABLE usage_daily_rollups ( @@ -1912,7 +2209,10 @@ impl Database { /// v11 -> v12 迁移：添加项目 Profiles 表 /// 与 create_tables_on_conn 中的建表语句保持一致（IF NOT EXISTS 保证幂等） - fn migrate_v11_to_v12(conn: &Connection) -> Result<(), AppError> { + fn migrate_v11_to_v12( + conn: &Connection, + _context: MigrationRunContext, + ) -> Result<(), AppError> { conn.execute( "CREATE TABLE IF NOT EXISTS profiles ( id TEXT PRIMARY KEY, @@ -1932,7 +2232,10 @@ impl Database { /// /// 默认 0 表示旧版/未知语义；旧 Codex 行只包含 cache read，不包含 /// cache creation。新代理行会显式写入 1(total-inclusive) 或 2(fresh)。 - fn migrate_v12_to_v13(conn: &Connection) -> Result<(), AppError> { + fn migrate_v12_to_v13( + conn: &Connection, + _context: MigrationRunContext, + ) -> Result<(), AppError> { if Self::table_exists(conn, "proxy_request_logs")? { Self::add_column_if_missing( conn, @@ -1953,11 +2256,68 @@ impl Database { } /// v13 -> v14: allow Grok Build to own an independent proxy configuration row. - fn migrate_v13_to_v14(conn: &Connection) -> Result<(), AppError> { + fn migrate_v13_to_v14(conn: &Connection, context: MigrationRunContext) -> Result<(), AppError> { if !Self::table_exists(conn, "proxy_config")? { return Ok(()); } + const V13_PROXY_STORAGE: &[(&str, &str)] = &[ + ("app_type", "text"), + ("proxy_enabled", "integer"), + ("listen_address", "text"), + ("listen_port", "integer"), + ("enable_logging", "integer"), + ("enabled", "integer"), + ("auto_failover_enabled", "integer"), + ("max_retries", "integer"), + ("streaming_first_byte_timeout", "integer"), + ("streaming_idle_timeout", "integer"), + ("non_streaming_timeout", "integer"), + ("circuit_failure_threshold", "integer"), + ("circuit_success_threshold", "integer"), + ("circuit_timeout_seconds", "integer"), + ("circuit_error_rate_threshold", "real"), + ("circuit_min_requests", "integer"), + ("default_cost_multiplier", "text"), + ("pricing_model_source", "text"), + ("live_takeover_active", "integer"), + ("created_at", "text"), + ("updated_at", "text"), + ]; + let untrusted_source_empty = if context == MigrationRunContext::UntrustedRestore { + let (total_rows, grokbuild_rows): (i64, i64) = conn + .query_row( + "SELECT COUNT(*), + COUNT(CASE WHEN app_type = 'grokbuild' THEN 1 END) + FROM proxy_config", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .map_err(|error| { + AppError::Database(format!("count v13 Grok Build proxy rows failed: {error}")) + })?; + if total_rows != 0 { + Self::require_untrusted_column_storage( + conn, + context, + "v13->v14 proxy rebuild", + "proxy_config", + V13_PROXY_STORAGE, + )?; + } + if total_rows != 0 && grokbuild_rows != 1 { + return Err(Self::untrusted_repair_error( + "v13->v14 proxy rebuild", + format!( + "expected one existing Grok Build row, found {grokbuild_rows}; row synthesis is forbidden" + ), + )); + } + total_rows == 0 + } else { + false + }; + conn.execute("DROP TABLE IF EXISTS proxy_config_v14", []) .map_err(|e| AppError::Database(e.to_string()))?; conn.execute( @@ -2012,14 +2372,17 @@ impl Database { ("updated_at", "datetime('now')"), ] .into_iter() - .map(|(column, fallback)| { - Self::has_column(conn, "proxy_config", column).map(|exists| { - if exists { - format!("\"{column}\"") - } else { - fallback.into() - } - }) + .map(|(column, fallback)| -> Result { + if Self::has_column(conn, "proxy_config", column)? { + Ok(format!("\"{column}\"")) + } else if context == MigrationRunContext::LocalUpgrade || untrusted_source_empty { + Ok(fallback.into()) + } else { + Err(Self::untrusted_repair_error( + "v13->v14 proxy rebuild", + format!("proxy_config.{column} would use fallback {fallback}"), + )) + } }) .collect::, AppError>>()? .join(", "); @@ -2043,17 +2406,22 @@ impl Database { .map_err(|e| AppError::Database(e.to_string()))?; conn.execute("ALTER TABLE proxy_config_v14 RENAME TO proxy_config", []) .map_err(|e| AppError::Database(e.to_string()))?; - conn.execute( - "INSERT OR IGNORE INTO proxy_config (app_type) VALUES ('grokbuild')", - [], - ) - .map_err(|e| AppError::Database(e.to_string()))?; + if context == MigrationRunContext::LocalUpgrade { + conn.execute( + "INSERT OR IGNORE INTO proxy_config (app_type) VALUES ('grokbuild')", + [], + ) + .map_err(|e| AppError::Database(e.to_string()))?; + } Ok(()) } /// v14 -> v15: persist Grok Build enablement for unified Skills and MCP. - fn migrate_v14_to_v15(conn: &Connection) -> Result<(), AppError> { + fn migrate_v14_to_v15( + conn: &Connection, + _context: MigrationRunContext, + ) -> Result<(), AppError> { if Self::table_exists(conn, "mcp_servers")? { Self::add_column_if_missing( conn, @@ -2076,7 +2444,58 @@ impl Database { /// v15 -> v16: remove Codex session rows and cursors so startup sync can /// rebuild them with fork-history alignment. Must stay connection-level: /// schema migration already owns the Database connection mutex. - fn migrate_v15_to_v16(conn: &Connection) -> Result<(), AppError> { + fn migrate_v15_to_v16(conn: &Connection, context: MigrationRunContext) -> Result<(), AppError> { + if context == MigrationRunContext::UntrustedRestore { + let codex_logs: i64 = conn + .query_row( + "SELECT COUNT(*) FROM proxy_request_logs + WHERE data_source = 'codex_session'", + [], + |row| row.get(0), + ) + .map_err(|error| { + AppError::Database(format!("count Codex session logs: {error}")) + })?; + Self::require_no_untrusted_repair_rows( + context, + "v15->v16 Codex usage reset", + "proxy_request_logs[data_source=codex_session]", + codex_logs, + )?; + + let codex_rollups: i64 = conn + .query_row( + "SELECT COUNT(*) FROM usage_daily_rollups + WHERE provider_id = '_codex_session'", + [], + |row| row.get(0), + ) + .map_err(|error| { + AppError::Database(format!("count Codex session rollups: {error}")) + })?; + Self::require_no_untrusted_repair_rows( + context, + "v15->v16 Codex usage reset", + "usage_daily_rollups[provider_id=_codex_session]", + codex_rollups, + )?; + + // Cursor ownership is resolved against this device's filesystem. + // An untrusted backup cannot prove that relationship, so any + // incoming cursor would require a local-path repair decision. + let cursors: i64 = conn + .query_row("SELECT COUNT(*) FROM session_log_sync", [], |row| { + row.get(0) + }) + .map_err(|error| AppError::Database(format!("count session cursors: {error}")))?; + Self::require_no_untrusted_repair_rows( + context, + "v15->v16 Codex usage reset", + "session_log_sync", + cursors, + )?; + return Ok(()); + } let codex_dir = crate::codex_config::get_codex_config_dir(); crate::services::session_usage_codex::reset_codex_usage_on_conn(conn, &codex_dir) } @@ -2118,8 +2537,9 @@ impl Database { // Merge their timestamps before rebuilding from the canonical // definition. The fixed-column copy makes FK/UNIQUE/collation // semantics part of migration rather than an optional index patch. - conn.execute_batch( - "UPDATE provider_endpoints AS kept + if context == MigrationRunContext::LocalUpgrade { + conn.execute_batch( + "UPDATE provider_endpoints AS kept SET added_at = ( SELECT MIN(other.added_at) FROM provider_endpoints AS other @@ -2147,8 +2567,9 @@ impl Database { FROM provider_endpoints GROUP BY provider_id, app_type, url );", - ) - .map_err(|error| AppError::Database(error.to_string()))?; + ) + .map_err(|error| AppError::Database(error.to_string()))?; + } const REBUILT_ENDPOINTS: &str = "provider_endpoints_v17_canonical"; conn.execute(&format!("DROP TABLE IF EXISTS \"{REBUILT_ENDPOINTS}\""), []) .map_err(|error| AppError::Database(error.to_string()))?; @@ -3094,6 +3515,77 @@ impl Database { Ok(()) } + fn model_pricing_snapshot(conn: &Connection) -> Result, AppError> { + let mut statement = conn + .prepare( + "SELECT model_id, display_name, input_cost_per_million, + output_cost_per_million, cache_read_cost_per_million, + cache_creation_cost_per_million + FROM model_pricing ORDER BY model_id", + ) + .map_err(|error| { + AppError::Database(format!("prepare model-pricing snapshot: {error}")) + })?; + let rows = statement + .query_map([], |row| { + Ok(( + row.get(0)?, + row.get(1)?, + row.get(2)?, + row.get(3)?, + row.get(4)?, + row.get(5)?, + )) + }) + .map_err(|error| { + AppError::Database(format!("query model-pricing snapshot: {error}")) + })?; + rows.collect::, _>>() + .map_err(|error| AppError::Database(format!("decode model-pricing snapshot: {error}"))) + } + + fn canonical_model_pricing_snapshot() -> Result, AppError> { + let reference = Connection::open_in_memory() + .map_err(|error| AppError::Database(format!("open pricing oracle: {error}")))?; + reference + .execute( + "CREATE TABLE model_pricing ( + model_id TEXT PRIMARY KEY, + display_name TEXT NOT NULL, + input_cost_per_million TEXT NOT NULL, + output_cost_per_million TEXT NOT NULL, + cache_read_cost_per_million TEXT NOT NULL DEFAULT '0', + cache_creation_cost_per_million TEXT NOT NULL DEFAULT '0' + )", + [], + ) + .map_err(|error| AppError::Database(format!("create pricing oracle table: {error}")))?; + Self::ensure_model_pricing_seeded_on_conn(&reference)?; + Self::model_pricing_snapshot(&reference) + } + + fn model_pricing_is_empty_or_canonical(conn: &Connection) -> Result { + let actual = Self::model_pricing_snapshot(conn)?; + Ok(actual.is_empty() || actual == Self::canonical_model_pricing_snapshot()?) + } + + fn require_canonical_model_pricing(conn: &Connection, step: &str) -> Result<(), AppError> { + let actual = Self::model_pricing_snapshot(conn)?; + let expected = Self::canonical_model_pricing_snapshot()?; + if actual.is_empty() || actual == expected { + Ok(()) + } else { + Err(Self::untrusted_repair_error( + step, + format!( + "incoming model_pricing has {} rows but the canonical seed has {}; replacement is forbidden", + actual.len(), + expected.len() + ), + )) + } + } + fn repair_current_model_pricing(conn: &Connection) -> Result<(), AppError> { let pricing_fixes = [ // 2026-07-12 GPT-5.6 家族 cache write=1.25× 输入价（OpenAI 5.6 起的新规），