perf(usage): batch Codex session inserts and preload sync cursors

Full Codex usage reimport (triggered by SQL backup import of pre-v16
backups, cursor mismatch after cross-machine restore, or manual rebuild)
pegged one CPU core for minutes on large corpora. Measured on a real
corpus (1920 rollout files / ~1.7GB / 82k billable events, macOS,
release build):

- full reimport on a disk-backed database: 36.3s -> 11.1s (~3.3x)
- the write path accounts for most of the win: per-event autocommit
  under journal_mode=delete paid one full journal create/fsync/delete
  cycle per row; Windows pays ~4.6ms per fsync (per #6122
  measurements), so the absolute win there is an order of magnitude
  larger

Changes:

1. sync_single_codex_file writes token events in batched transactions
   (1000 events per commit), releasing the connection lock between
   batches so UI queries can interleave. The sync cursor advances
   inside the same transaction as the final batch, so a crash can
   never leave the cursor ahead of the data; batches committed before
   a crash are covered on rescan by the existing request_id PK +
   fingerprint dedup.
2. CodexSyncPass preloads all session_log_sync cursors once per pass
   (the archived-sessions suffix inheritance query used substr, could
   not use an index, and ran a full table scan per archived file) and
   caches model pricing lookups per pass instead of one SELECT per
   event.
3. The three dedup queries and the insert statement use
   prepare_cached; the two dynamically assembled dedup SQL strings are
   built once via LazyLock.

Behavioral equivalence verified by replaying the real corpus before
and after the change: all 82k imported rows byte-identical across all
exported columns, with identical import/skip/suspected-duplicate
counts. The replay harness is kept as an ignored test
(replay_real_codex_corpus) for future changes on this path.

src-tauri/src/services/session_usage.rs

@@ -461,18 +461,29 @@ pub(crate) fn update_sync_state(
     file_path: &str,
     last_modified: i64,
     last_offset: i64,
+) -> Result<(), AppError> {
+    let conn = lock_conn!(db.conn);
+    update_sync_state_on_conn(&conn, file_path, last_modified, last_offset)
+}
+
+/// [`update_sync_state`] 的免锁版本，供调用方在已持锁的事务内把游标推进
+/// 与数据插入绑成原子提交。
+pub(crate) fn update_sync_state_on_conn(
+    conn: &rusqlite::Connection,
+    file_path: &str,
+    last_modified: i64,
+    last_offset: i64,
 ) -> Result<(), AppError> {
     let now = SystemTime::now()
         .duration_since(SystemTime::UNIX_EPOCH)
         .map(|d| d.as_secs() as i64)
         .unwrap_or(0);
 
-    let conn = lock_conn!(db.conn);
-    conn.execute(
+    conn.prepare_cached(
         "INSERT OR REPLACE INTO session_log_sync (file_path, last_modified, last_line_offset, last_synced_at)
          VALUES (?1, ?2, ?3, ?4)",
-        rusqlite::params![file_path, last_modified, last_offset, now],
     )
+    .and_then(|mut stmt| stmt.execute(rusqlite::params![file_path, last_modified, last_offset, now]))
     .map_err(|e| AppError::Database(format!("更新同步状态失败: {e}")))?;
     Ok(())
 }

src-tauri/src/services/session_usage_codex.rs

@@ -19,7 +19,7 @@ use crate::error::AppError;
 use crate::proxy::usage::calculator::{CostCalculator, ModelPricing};
 use crate::proxy::usage::parser::TokenUsage;
 use crate::services::session_usage::{
-    get_sync_state, metadata_modified_nanos, update_sync_state, SessionSyncResult,
+    metadata_modified_nanos, update_sync_state, update_sync_state_on_conn, SessionSyncResult,
 };
 use crate::services::usage_stats::{
     find_model_pricing, has_suspected_codex_session_duplicate, should_skip_session_insert, DedupKey,
@@ -459,9 +459,49 @@ fn token_snapshot_source(payload: &serde_json::Value) -> Option<String> {
         .map(str::to_owned)
 }
 
-fn get_codex_sync_state(db: &Database, file_path: &Path) -> Result<(i64, i64), AppError> {
+/// 单个同步 pass 的共享状态。
+///
+/// - `cursors`：pass 开始时一次性预载的 `session_log_sync` 快照，替代逐文件
+///   SELECT（尤其是 archived 继承的 `substr` 后缀匹配无法走索引，逐文件跑等于
+///   每 pass 全表扫 N 次）。快照语义：同 pass 内其他文件刚写入的游标对后续
+///   archived 继承不可见——影响仅是多一轮由 request_id 去重兜底的重扫，
+///   不丢数据、不双算。
+/// - `pricing`：模型定价 pass 级缓存。定价表在 pass 进行中被修改时本 pass
+///   仍用旧价，下一个同步 pass 生效。
+struct CodexSyncPass {
+    cursors: HashMap<String, (i64, i64)>,
+    pricing: HashMap<String, Option<ModelPricing>>,
+}
+
+impl CodexSyncPass {
+    fn load(db: &Database) -> Result<Self, AppError> {
+        let conn = lock_conn!(db.conn);
+        let mut stmt = conn
+            .prepare("SELECT file_path, last_modified, last_line_offset FROM session_log_sync")
+            .map_err(|e| AppError::Database(format!("预载同步游标失败: {e}")))?;
+        let cursors = stmt
+            .query_map([], |row| {
+                Ok((
+                    row.get::<_, String>(0)?,
+                    (row.get::<_, i64>(1)?, row.get::<_, i64>(2)?),
+                ))
+            })
+            .and_then(|rows| rows.collect::<Result<HashMap<_, _>, _>>())
+            .map_err(|e| AppError::Database(format!("预载同步游标失败: {e}")))?;
+        Ok(Self {
+            cursors,
+            pricing: HashMap::new(),
+        })
+    }
+}
+
+fn get_codex_sync_state(
+    db: &Database,
+    file_path: &Path,
+    cursors: &HashMap<String, (i64, i64)>,
+) -> Result<(i64, i64), AppError> {
     let file_path_str = file_path.to_string_lossy().to_string();
-    let state = get_sync_state(db, &file_path_str)?;
+    let state = cursors.get(&file_path_str).copied().unwrap_or((0, 0));
     if state != (0, 0)
         || file_path
             .parent()
@@ -477,29 +517,23 @@ fn get_codex_sync_state(db: &Database, file_path: &Path) -> Result<(i64, i64), A
     };
     let slash_suffix = format!("/{file_name}");
     let backslash_suffix = format!("\\{file_name}");
-    let conn = lock_conn!(db.conn);
-    let inherited = conn.query_row(
-        "SELECT last_modified, last_line_offset
-         FROM session_log_sync
-         WHERE file_path <> ?1
-           AND (substr(file_path, -length(?2)) = ?2
-                OR substr(file_path, -length(?3)) = ?3)
-         ORDER BY last_line_offset DESC, last_modified DESC
-         LIMIT 1",
-        rusqlite::params![file_path_str, slash_suffix, backslash_suffix],
-        |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
-    );
-    drop(conn);
+    // 与原 SQL 等价：ORDER BY last_line_offset DESC, last_modified DESC LIMIT 1
+    // → 在快照上按 (offset, modified) 取最大。
+    let inherited = cursors
+        .iter()
+        .filter(|(path, _)| {
+            path.as_str() != file_path_str
+                && (path.ends_with(&slash_suffix) || path.ends_with(&backslash_suffix))
+        })
+        .map(|(_, &(modified, offset))| (offset, modified))
+        .max();
 
     match inherited {
-        Ok(inherited) => {
-            update_sync_state(db, &file_path_str, inherited.0, inherited.1)?;
-            Ok(inherited)
+        Some((offset, modified)) => {
+            update_sync_state(db, &file_path_str, modified, offset)?;
+            Ok((modified, offset))
         }
-        Err(rusqlite::Error::QueryReturnedNoRows) => Ok(state),
-        Err(error) => Err(AppError::Database(format!(
-            "查询 Codex 归档文件同步状态失败: {error}"
-        ))),
+        None => Ok(state),
     }
 }
 
@@ -619,6 +653,7 @@ pub fn sync_codex_usage(db: &Database) -> Result<SessionSyncResult, AppError> {
     let codex_dir = get_codex_config_dir();
     let files = collect_codex_session_files(&codex_dir);
     let rollout_index = build_rollout_index(&files);
+    let mut pass = CodexSyncPass::load(db)?;
 
     let mut result = SessionSyncResult {
         imported: 0,
@@ -630,7 +665,7 @@ pub fn sync_codex_usage(db: &Database) -> Result<SessionSyncResult, AppError> {
     };
 
     for file_path in &files {
-        match sync_single_codex_file(db, file_path, &rollout_index) {
+        match sync_single_codex_file(db, file_path, &rollout_index, &mut pass) {
             Ok(file_result) => {
                 result.imported = result.imported.saturating_add(file_result.imported);
                 result.skipped = result.skipped.saturating_add(file_result.skipped);
@@ -1089,11 +1124,17 @@ fn mark_deferred(
     }
 }
 
+/// 单文件批量插入的事务粒度。批内 UI 查询会被连接互斥锁挡住约几毫秒，
+/// 批间释放锁让读侧插队——兼顾吞吐（避免逐行 autocommit 的每行 fsync）
+/// 与大文件重导期间面板的响应性。
+const CODEX_INSERT_BATCH_SIZE: usize = 1000;
+
 /// 同步单个 Codex JSONL 文件。
 fn sync_single_codex_file(
     db: &Database,
     file_path: &Path,
     rollout_index: &RolloutIndex,
+    pass: &mut CodexSyncPass,
 ) -> Result<CodexFileSyncResult, AppError> {
     let file_path_str = file_path.to_string_lossy().to_string();
 
@@ -1104,7 +1145,7 @@ fn sync_single_codex_file(
     let file_size = metadata.len();
 
     // 检查同步状态
-    let (last_modified, last_offset) = get_codex_sync_state(db, file_path)?;
+    let (last_modified, last_offset) = get_codex_sync_state(db, file_path, &pass.cursors)?;
 
     // 文件未变化则跳过
     if file_modified <= last_modified {
@@ -1234,6 +1275,7 @@ fn sync_single_codex_file(
     }
 
     let mut result = CodexFileSyncResult::default();
+    let mut to_insert: Vec<(&ParsedTokenEvent, u32)> = Vec::new();
     for (token_offset, event) in parsed.token_events.iter().enumerate() {
         let Some(event_index) = event.event_index else {
             continue;
@@ -1247,31 +1289,67 @@ fn sync_single_codex_file(
         if event.line_offset <= last_offset {
             continue;
         }
-
-        let request_id = format!("{CODEX_THREAD_REQUEST_ID_PREFIX}:{root_thread_id}:{event_index}");
-        match insert_codex_session_entry(
-            db,
-            &request_id,
-            &event.delta,
-            &event.model,
-            Some(root_thread_id),
-            event.timestamp.as_deref(),
-            &mut result.suspected_duplicates,
-        ) {
-            Ok(true) => result.imported = result.imported.saturating_add(1),
-            Ok(false) => result.skipped = result.skipped.saturating_add(1),
-            Err(e) => {
-                log::warn!("[CODEX-SYNC] 插入失败 ({request_id}): {e}");
-                result.skipped = result.skipped.saturating_add(1);
-            }
-        }
+        to_insert.push((event, event_index));
     }
 
-    update_sync_state(db, &file_path_str, file_modified, parsed.line_offset)?;
+    // 分批事务写库：逐行 autocommit（journal_mode=delete 下每行一整套
+    // journal 建立/fsync/删除）是全量重导的最大耗时项。批内单条插入失败
+    // 沿用旧行为跳过该条继续；某批 commit 失败则该批整体回滚且游标不推进，
+    // 下一 pass 重扫时由 request_id 主键 + 指纹去重兜底，不会双算。
+    let batch_count = to_insert.len().div_ceil(CODEX_INSERT_BATCH_SIZE);
+    for (batch_index, batch) in to_insert.chunks(CODEX_INSERT_BATCH_SIZE).enumerate() {
+        let is_last_batch = batch_index + 1 == batch_count;
+        let conn = lock_conn!(db.conn);
+        let tx = conn
+            .unchecked_transaction()
+            .map_err(|e| AppError::Database(format!("开启 Codex 会话写入事务失败: {e}")))?;
+
+        let mut batch_imported = 0u32;
+        let mut batch_skipped = 0u32;
+        let mut batch_suspected = 0u32;
+        for (event, event_index) in batch {
+            let request_id =
+                format!("{CODEX_THREAD_REQUEST_ID_PREFIX}:{root_thread_id}:{event_index}");
+            match insert_codex_session_entry_on_conn(
+                &tx,
+                &request_id,
+                &event.delta,
+                &event.model,
+                Some(root_thread_id),
+                event.timestamp.as_deref(),
+                &mut batch_suspected,
+                &mut pass.pricing,
+            ) {
+                Ok(true) => batch_imported += 1,
+                Ok(false) => batch_skipped += 1,
+                Err(e) => {
+                    log::warn!("[CODEX-SYNC] 插入失败 ({request_id}): {e}");
+                    batch_skipped += 1;
+                }
+            }
+        }
+        if is_last_batch {
+            // 游标推进与最后一批数据同事务提交：中途崩溃时两者一起回滚，
+            // 不会出现"游标已推进但数据缺失"的丢数据窗口。
+            update_sync_state_on_conn(&tx, &file_path_str, file_modified, parsed.line_offset)?;
+        }
+        tx.commit()
+            .map_err(|e| AppError::Database(format!("提交 Codex 会话写入事务失败: {e}")))?;
+
+        result.imported = result.imported.saturating_add(batch_imported);
+        result.skipped = result.skipped.saturating_add(batch_skipped);
+        result.suspected_duplicates = result.suspected_duplicates.saturating_add(batch_suspected);
+    }
+
+    if to_insert.is_empty() {
+        update_sync_state(db, &file_path_str, file_modified, parsed.line_offset)?;
+    }
     Ok(result)
 }
 
-/// 插入单条 Codex 会话记录到 proxy_request_logs
+/// 插入单条 Codex 会话记录到 proxy_request_logs（自取锁的便捷包装，测试专用；
+/// 生产路径走 [`insert_codex_session_entry_on_conn`] 以复用批量事务与定价缓存）
+#[cfg(test)]
 fn insert_codex_session_entry(
     db: &Database,
     request_id: &str,
@@ -1282,7 +1360,34 @@ fn insert_codex_session_entry(
     suspected_duplicates: &mut u32,
 ) -> Result<bool, AppError> {
     let conn = lock_conn!(db.conn);
+    insert_codex_session_entry_on_conn(
+        &conn,
+        request_id,
+        delta,
+        model,
+        session_id,
+        timestamp,
+        suspected_duplicates,
+        &mut HashMap::new(),
+    )
+}
 
+/// 插入单条 Codex 会话记录到 proxy_request_logs。
+///
+/// 调用方负责持锁/事务；`pricing_cache` 按原始 model 字符串键控（
+/// `find_codex_pricing` 是纯函数式查找，同串必同结果），全量重导时把
+/// 每事件一次的定价 SELECT 降为每模型一次。
+#[allow(clippy::too_many_arguments)]
+fn insert_codex_session_entry_on_conn(
+    conn: &rusqlite::Connection,
+    request_id: &str,
+    delta: &DeltaTokens,
+    model: &str,
+    session_id: Option<&str>,
+    timestamp: Option<&str>,
+    suspected_duplicates: &mut u32,
+    pricing_cache: &mut HashMap<String, Option<ModelPricing>>,
+) -> Result<bool, AppError> {
     let created_at = timestamp
         .and_then(|ts| {
             chrono::DateTime::parse_from_rfc3339(ts)
@@ -1305,10 +1410,10 @@ fn insert_codex_session_entry(
         cache_creation_tokens: 0,
         created_at,
     };
-    if should_skip_session_insert(&conn, request_id, &dedup_key)? {
+    if should_skip_session_insert(conn, request_id, &dedup_key)? {
         return Ok(false);
     }
-    if has_suspected_codex_session_duplicate(&conn, request_id, &dedup_key)? {
+    if has_suspected_codex_session_duplicate(conn, request_id, &dedup_key)? {
         *suspected_duplicates = suspected_duplicates.saturating_add(1);
         log::warn!(
             "[CODEX-SYNC] 疑似重复会话用量: request_id={request_id}, model={model}, input={}, output={}, cache_read={}",
@@ -1328,12 +1433,14 @@ fn insert_codex_session_entry(
         message_id: None,
     };
 
-    let pricing = find_codex_pricing(&conn, model);
+    let pricing = pricing_cache
+        .entry(model.to_string())
+        .or_insert_with(|| find_codex_pricing(conn, model));
     let multiplier = Decimal::from(1);
     let (input_cost, output_cost, cache_read_cost, cache_creation_cost, total_cost) = match pricing
     {
         Some(p) => {
-            let cost = CostCalculator::calculate_for_app("codex", &usage, &p, multiplier);
+            let cost = CostCalculator::calculate_for_app("codex", &usage, p, multiplier);
             (
                 cost.input_cost.to_string(),
                 cost.output_cost.to_string(),
@@ -1352,7 +1459,7 @@ fn insert_codex_session_entry(
     };
 
     let inserted_rows = conn
-        .execute(
+        .prepare_cached(
             "INSERT OR IGNORE INTO proxy_request_logs (
             request_id, provider_id, app_type, model, request_model,
             input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens,
@@ -1360,7 +1467,8 @@ fn insert_codex_session_entry(
             latency_ms, first_token_ms, status_code, error_message, session_id,
             provider_type, is_streaming, cost_multiplier, created_at, data_source
         ) 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)",
-            rusqlite::params![
+        )
+        .and_then(|mut stmt| stmt.execute(rusqlite::params![
                 request_id,
                 "_codex_session",    // provider_id
                 "codex",             // app_type
@@ -1385,8 +1493,7 @@ fn insert_codex_session_entry(
                 "1.0",               // cost_multiplier
                 created_at,
                 "codex_session",     // data_source
-            ],
-        )
+            ]))
         .map_err(|e| AppError::Database(format!("插入 Codex 会话日志失败: {e}")))?;
 
     Ok(inserted_rows > 0)
@@ -1400,6 +1507,7 @@ fn find_codex_pricing(conn: &rusqlite::Connection, model_id: &str) -> Option<Mod
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::services::session_usage::get_sync_state;
     use tempfile::tempdir;
 
     const PARENT_ID: &str = "00000000-0000-4000-8000-000000000001";
@@ -1543,7 +1651,8 @@ mod tests {
             .iter()
             .map(|path| path.to_path_buf())
             .collect::<Vec<_>>();
-        sync_single_codex_file(db, file, &build_rollout_index(&files))
+        let mut pass = CodexSyncPass::load(db)?;
+        sync_single_codex_file(db, file, &build_rollout_index(&files), &mut pass)
     }
 
     #[test]
@@ -2861,4 +2970,117 @@ mod tests {
         let clamped = delta.cached_input.min(delta.input);
         assert_eq!(clamped, 10);
     }
+
+    /// 真实语料回放验收 harness（仅手动运行，勿在 CI 跑）。
+    ///
+    /// 把真实 `~/.codex/sessions` 语料在内存库上做一次全量重导，输出计时与
+    /// 结果快照。用于性能改动的行为等价验证：改动前后各跑一次，两侧
+    /// `CODEX_REPLAY_OUT` 文件必须逐字节相同。
+    ///
+    /// ```bash
+    /// CODEX_REPLAY_OUT=/tmp/replay.tsv \
+    ///   cargo test --release replay_real_codex_corpus -- --ignored --nocapture
+    /// ```
+    #[test]
+    #[ignore]
+    fn replay_real_codex_corpus() -> Result<(), AppError> {
+        let Some(real_home) = dirs::home_dir() else {
+            eprintln!("[REPLAY] no home dir, skipping");
+            return Ok(());
+        };
+        let real_sessions = real_home.join(".codex").join("sessions");
+        if !real_sessions.is_dir() {
+            eprintln!("[REPLAY] {} not found, skipping", real_sessions.display());
+            return Ok(());
+        }
+
+        // 临时 HOME 里只放一个指向真实语料的只读 symlink，避免测试
+        // 触碰真实 ~/.cc-switch / ~/.codex 下的任何其他内容。
+        let temp = tempfile::tempdir().expect("create temp home");
+        fs::create_dir_all(temp.path().join(".codex")).expect("mkdir .codex");
+        #[cfg(unix)]
+        std::os::unix::fs::symlink(&real_sessions, temp.path().join(".codex").join("sessions"))
+            .expect("symlink sessions");
+        let previous_home = std::env::var_os("CC_SWITCH_TEST_HOME");
+        std::env::set_var("CC_SWITCH_TEST_HOME", temp.path());
+
+        clear_codex_replay_caches();
+        // CODEX_REPLAY_DISK=1 时用临时 HOME 下的磁盘库：逐行 autocommit 的
+        // 主要成本是磁盘 journal fsync，内存库测不出真实写库开销。
+        let db = if std::env::var("CODEX_REPLAY_DISK").is_ok() {
+            Database::init()?
+        } else {
+            Database::memory()?
+        };
+        let start = std::time::Instant::now();
+        let result = sync_codex_usage(&db)?;
+        let full_elapsed = start.elapsed();
+        eprintln!(
+            "[REPLAY] full reimport: imported={} skipped={} suspected_dup={} deferred={} files={} errors={} elapsed={:.2?}",
+            result.imported,
+            result.skipped,
+            result.suspected_duplicates,
+            result.deferred_files,
+            result.files_scanned,
+            result.errors.len(),
+            full_elapsed
+        );
+
+        let start = std::time::Instant::now();
+        let steady = sync_codex_usage(&db)?;
+        eprintln!(
+            "[REPLAY] steady pass: imported={} deferred={} elapsed={:.2?}",
+            steady.imported,
+            steady.deferred_files,
+            start.elapsed()
+        );
+
+        if let Ok(out_path) = std::env::var("CODEX_REPLAY_OUT") {
+            use std::io::Write;
+            let conn = lock_conn!(db.conn);
+            let mut stmt = conn
+                .prepare(
+                    "SELECT request_id, model, request_model, input_tokens, output_tokens,
+                            cache_read_tokens, cache_creation_tokens,
+                            input_cost_usd, output_cost_usd, cache_read_cost_usd,
+                            cache_creation_cost_usd, total_cost_usd,
+                            session_id, provider_id, provider_type, status_code,
+                            is_streaming, cost_multiplier, created_at, data_source
+                     FROM proxy_request_logs
+                     WHERE data_source = 'codex_session'
+                     ORDER BY request_id",
+                )
+                .map_err(|e| AppError::Database(e.to_string()))?;
+            let rows = stmt
+                .query_map([], |row| {
+                    let mut fields = Vec::with_capacity(20);
+                    for idx in 0..20 {
+                        fields.push(match row.get_ref(idx)? {
+                            rusqlite::types::ValueRef::Null => "NULL".to_string(),
+                            rusqlite::types::ValueRef::Integer(v) => v.to_string(),
+                            rusqlite::types::ValueRef::Real(v) => v.to_string(),
+                            rusqlite::types::ValueRef::Text(v) => {
+                                String::from_utf8_lossy(v).into_owned()
+                            }
+                            rusqlite::types::ValueRef::Blob(v) => format!("blob:{}", v.len()),
+                        });
+                    }
+                    Ok(fields.join("\t"))
+                })
+                .map_err(|e| AppError::Database(e.to_string()))?
+                .collect::<Result<Vec<_>, _>>()
+                .map_err(|e| AppError::Database(e.to_string()))?;
+            let mut out = fs::File::create(&out_path).expect("create replay out file");
+            for line in &rows {
+                writeln!(out, "{line}").expect("write replay row");
+            }
+            eprintln!("[REPLAY] wrote {} rows to {out_path}", rows.len());
+        }
+
+        match previous_home {
+            Some(value) => std::env::set_var("CC_SWITCH_TEST_HOME", value),
+            None => std::env::remove_var("CC_SWITCH_TEST_HOME"),
+        }
+        Ok(())
+    }
 }

src-tauri/src/services/usage_stats.rs

@@ -13,6 +13,7 @@ use rusqlite::{params, Connection, OptionalExtension};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::str::FromStr;
+use std::sync::LazyLock;
 
 /// 使用量汇总
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -370,24 +371,17 @@ pub(crate) fn should_skip_session_insert(
 }
 
 fn proxy_request_id_exists(conn: &Connection, request_id: &str) -> Result<bool, AppError> {
-    conn.query_row(
-        "SELECT EXISTS(SELECT 1 FROM proxy_request_logs WHERE request_id = ?1)",
-        params![request_id],
-        |row| row.get::<_, bool>(0),
-    )
-    .map_err(|e| AppError::Database(format!("查询 request_id 失败: {e}")))
+    conn.prepare_cached("SELECT EXISTS(SELECT 1 FROM proxy_request_logs WHERE request_id = ?1)")
+        .and_then(|mut stmt| stmt.query_row(params![request_id], |row| row.get::<_, bool>(0)))
+        .map_err(|e| AppError::Database(format!("查询 request_id 失败: {e}")))
 }
 
-pub(crate) fn has_matching_proxy_usage_log(
-    conn: &Connection,
-    key: &DedupKey,
-) -> Result<bool, AppError> {
-    let allow_missing_cache_creation =
-        matches!(key.app_type, "codex" | "gemini" | "opencode") && key.cache_creation_tokens == 0;
-
+// 会话重导每个 token 事件都要跑一次这条查询；SQL 文本静态化让
+// prepare_cached 稳定命中，也省掉每行的 format! 分配。
+static MATCHING_PROXY_USAGE_LOG_SQL: LazyLock<String> = LazyLock::new(|| {
     let l_data_source = data_source_expr("l");
     let app_type_match = dedup_app_type_match_sql("l.app_type", "?1");
-    let sql = format!(
+    format!(
         "SELECT EXISTS (
             SELECT 1
             FROM proxy_request_logs l
@@ -406,24 +400,34 @@ pub(crate) fn has_matching_proxy_usage_log(
                   OR LOWER(?2) = 'unknown'
               )
         )"
-    );
-
-    conn.query_row(
-        &sql,
-        params![
-            key.app_type,
-            key.model,
-            key.input_tokens as i64,
-            key.output_tokens as i64,
-            key.cache_read_tokens as i64,
-            key.cache_creation_tokens as i64,
-            key.created_at,
-            SESSION_PROXY_DEDUP_WINDOW_SECONDS,
-            allow_missing_cache_creation as i64,
-        ],
-        |row| row.get::<_, bool>(0),
     )
-    .map_err(|e| AppError::Database(format!("查询重复代理用量日志失败: {e}")))
+});
+
+pub(crate) fn has_matching_proxy_usage_log(
+    conn: &Connection,
+    key: &DedupKey,
+) -> Result<bool, AppError> {
+    let allow_missing_cache_creation =
+        matches!(key.app_type, "codex" | "gemini" | "opencode") && key.cache_creation_tokens == 0;
+
+    conn.prepare_cached(&MATCHING_PROXY_USAGE_LOG_SQL)
+        .and_then(|mut stmt| {
+            stmt.query_row(
+                params![
+                    key.app_type,
+                    key.model,
+                    key.input_tokens as i64,
+                    key.output_tokens as i64,
+                    key.cache_read_tokens as i64,
+                    key.cache_creation_tokens as i64,
+                    key.created_at,
+                    SESSION_PROXY_DEDUP_WINDOW_SECONDS,
+                    allow_missing_cache_creation as i64,
+                ],
+                |row| row.get::<_, bool>(0),
+            )
+        })
+        .map_err(|e| AppError::Database(format!("查询重复代理用量日志失败: {e}")))
 }
 
 /// grokbuild 会话导入的接管活动守卫：给定时刻 ±窗口内存在任何 grokbuild
@@ -460,13 +464,9 @@ pub(crate) fn has_recent_grokbuild_proxy_activity(
     .map_err(|e| AppError::Database(format!("查询 Grok 接管活动失败: {e}")))
 }
 
-pub(crate) fn has_suspected_codex_session_duplicate(
-    conn: &Connection,
-    request_id: &str,
-    key: &DedupKey,
-) -> Result<bool, AppError> {
+static SUSPECTED_CODEX_DUPLICATE_SQL: LazyLock<String> = LazyLock::new(|| {
     let data_source = data_source_expr("l");
-    let sql = format!(
+    format!(
         "SELECT EXISTS (
             SELECT 1
             FROM proxy_request_logs l
@@ -479,21 +479,30 @@ pub(crate) fn has_suspected_codex_session_duplicate(
               AND l.cache_read_tokens = ?5
               AND l.created_at BETWEEN ?6 - ?7 AND ?6 + ?7
         )"
-    );
-    conn.query_row(
-        &sql,
-        params![
-            request_id,
-            key.model,
-            key.input_tokens as i64,
-            key.output_tokens as i64,
-            key.cache_read_tokens as i64,
-            key.created_at,
-            SESSION_PROXY_DEDUP_WINDOW_SECONDS,
-        ],
-        |row| row.get::<_, bool>(0),
     )
-    .map_err(|error| AppError::Database(format!("查询疑似重复 Codex 会话用量失败: {error}")))
+});
+
+pub(crate) fn has_suspected_codex_session_duplicate(
+    conn: &Connection,
+    request_id: &str,
+    key: &DedupKey,
+) -> Result<bool, AppError> {
+    conn.prepare_cached(&SUSPECTED_CODEX_DUPLICATE_SQL)
+        .and_then(|mut stmt| {
+            stmt.query_row(
+                params![
+                    request_id,
+                    key.model,
+                    key.input_tokens as i64,
+                    key.output_tokens as i64,
+                    key.cache_read_tokens as i64,
+                    key.created_at,
+                    SESSION_PROXY_DEDUP_WINDOW_SECONDS,
+                ],
+                |row| row.get::<_, bool>(0),
+            )
+        })
+        .map_err(|error| AppError::Database(format!("查询疑似重复 Codex 会话用量失败: {error}")))
 }
 
 #[derive(Debug, Clone, Default)]