From df3e07edbb89bcfcd9a36c32a40b8311f2db5acc Mon Sep 17 00:00:00 2001 From: Jason Date: Mon, 20 Jul 2026 12:16:48 +0800 Subject: [PATCH] fix(codex): strip forked history via parent-rollout token-prefix alignment Rewrite the Codex session importer: identify forks by forked_from_id or subagent thread_spawn.parent_thread_id (conflicts defer), anchor thread identity to the rollout filename UUID, and align child token_count signatures against the parent rollout as a strict prefix so replayed parent history only establishes the cumulative baseline and is never imported as child usage. Ambiguous alignments defer without advancing the cursor and recover once the parent appears; parent signatures are cached per (path, cutoff). Fixes the 3x-20x double counting behind #5335/#5433 while keeping reference-type subagents fully counted (#5381). --- src-tauri/src/services/session_usage_codex.rs | 1475 +++++++++++++---- 1 file changed, 1166 insertions(+), 309 deletions(-) diff --git a/src-tauri/src/services/session_usage_codex.rs b/src-tauri/src/services/session_usage_codex.rs index 92e5b6935..55f836af5 100644 --- a/src-tauri/src/services/session_usage_codex.rs +++ b/src-tauri/src/services/session_usage_codex.rs @@ -21,11 +21,16 @@ use crate::proxy::usage::parser::TokenUsage; use crate::services::session_usage::{ get_sync_state, metadata_modified_nanos, update_sync_state, SessionSyncResult, }; -use crate::services::usage_stats::{find_model_pricing, should_skip_session_insert, DedupKey}; +use crate::services::usage_stats::{ + find_model_pricing, has_suspected_codex_session_duplicate, should_skip_session_insert, DedupKey, +}; +use chrono::{DateTime, Utc}; use rust_decimal::Decimal; +use std::collections::HashMap; use std::fs; use std::io::{BufRead, BufReader}; use std::path::{Path, PathBuf}; +use std::sync::{Mutex, OnceLock}; use std::time::SystemTime; const CODEX_THREAD_REQUEST_ID_PREFIX: &str = "codex_session:thread-v1"; @@ -52,120 +57,280 @@ impl DeltaTokens { } } -/// 单文件解析时的运行状态 -struct FileParseState { - thread_id: Option, - current_model: String, - prev_total: Option, - event_index: u32, - history_replay_boundary: Option, +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct TokenCountersSignature { + input: Option, + cached_input: Option, + output: Option, + reasoning_output: Option, + total: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct TokenUsageSignature { + total: Option, + last: Option, +} + +#[derive(Debug)] +struct ParsedTokenEvent { + line_offset: i64, + signature: TokenUsageSignature, + delta: DeltaTokens, + event_index: Option, + model: String, + timestamp: Option, +} + +#[derive(Debug)] +enum ParentResolution { + None, + Parent(String), + Deferred(String), +} + +#[derive(Debug)] +struct ParsedCodexFile { + root_thread_id: Option, + root_meta_seen: bool, + root_timestamp: Option>, + parent: ParentResolution, + token_events: Vec, + line_offset: i64, + has_billable_tokens: bool, } -/// Codex 子代理日志中的 `id` 是当前线程的唯一 ID，`session_id` 则指向父线程。 #[derive(Debug, Clone, PartialEq, Eq)] -struct CodexSessionIdentity { - thread_id: String, - carries_history_snapshot: bool, +enum PendingReason { + MissingParent(String), + Stable(String), + Retryable(String), } -fn parse_codex_session_identity(payload: &serde_json::Value) -> Option { - let thread_id = payload - .get("id") - .or_else(|| payload.get("thread_id")) - .or_else(|| payload.get("threadId")) - .or_else(|| payload.get("session_id")) - .or_else(|| payload.get("sessionId")) - .and_then(|value| value.as_str())? - .to_string(); - let session_id = payload - .get("session_id") - .or_else(|| payload.get("sessionId")) - .and_then(|value| value.as_str()); - let carries_history_snapshot = payload - .get("forked_from_id") - .and_then(|value| value.as_str()) - .is_some_and(|value| !value.is_empty()) - || payload - .get("source") - .and_then(|source| source.get("subagent")) - .is_some() - || session_id.is_some_and(|session_id| session_id != thread_id); +#[derive(Debug, Clone, PartialEq, Eq)] +struct PendingEntry { + modified: i64, + size: u64, + reason: PendingReason, +} - Some(CodexSessionIdentity { - thread_id, - carries_history_snapshot, +#[derive(Debug, Default)] +struct CodexReplayCaches { + parent_signatures: HashMap<(PathBuf, i64), Vec>, + replay_prefixes: HashMap<(PathBuf, i64, u64), usize>, + pending: HashMap, +} + +static CODEX_REPLAY_CACHES: OnceLock> = OnceLock::new(); + +fn replay_caches() -> &'static Mutex { + CODEX_REPLAY_CACHES.get_or_init(|| Mutex::new(CodexReplayCaches::default())) +} + +pub(crate) fn clear_codex_replay_caches() { + if let Ok(mut caches) = replay_caches().lock() { + *caches = CodexReplayCaches::default(); + } +} + +fn is_rollout_filename(file_name: &str) -> bool { + if !file_name.starts_with("rollout-") || !file_name.ends_with(".jsonl") { + return false; + } + let stem = file_name.trim_end_matches(".jsonl"); + stem.get(stem.len().saturating_sub(36)..) + .is_some_and(|candidate| uuid::Uuid::parse_str(candidate).is_ok()) +} + +fn is_codex_cursor_path(file_path: &str, codex_dir: &Path) -> bool { + let path = Path::new(file_path); + let file_name = file_path.rsplit(['/', '\\']).next().unwrap_or_default(); + if !is_rollout_filename(file_name) { + return false; + } + + if path.starts_with(codex_dir.join("sessions")) + || path.starts_with(codex_dir.join("archived_sessions")) + { + return true; + } + + // 兼容用户改过 CODEX_HOME 后遗留、且源文件已不存在的 cursor。只接受 + // 明确目录段 + Codex rollout UUID 文件名，避免宽 codex_dir 误删其他 importer。 + file_path + .replace('\\', "/") + .split('/') + .any(|segment| matches!(segment, "sessions" | "archived_sessions")) +} + +fn sqlite_table_exists(conn: &rusqlite::Connection, table: &str) -> Result { + conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1)", + [table], + |row| row.get(0), + ) + .map_err(|error| AppError::Database(format!("查询表 {table} 失败: {error}"))) +} + +fn sqlite_column_exists( + conn: &rusqlite::Connection, + table: &str, + column: &str, +) -> Result { + conn.query_row( + "SELECT EXISTS(SELECT 1 FROM pragma_table_info(?1) WHERE name = ?2)", + rusqlite::params![table, column], + |row| row.get(0), + ) + .map_err(|error| AppError::Database(format!("查询列 {table}.{column} 失败: {error}"))) +} + +pub(crate) fn reset_codex_usage_on_conn( + conn: &rusqlite::Connection, + codex_dir: &Path, +) -> Result<(), AppError> { + if sqlite_table_exists(conn, "proxy_request_logs")? + && sqlite_column_exists(conn, "proxy_request_logs", "data_source")? + { + conn.execute( + "DELETE FROM proxy_request_logs WHERE data_source = 'codex_session'", + [], + ) + .map_err(|error| AppError::Database(format!("清理 Codex 会话明细失败: {error}")))?; + } + if sqlite_table_exists(conn, "usage_daily_rollups")? + && sqlite_column_exists(conn, "usage_daily_rollups", "provider_id")? + { + conn.execute( + "DELETE FROM usage_daily_rollups WHERE provider_id = '_codex_session'", + [], + ) + .map_err(|error| AppError::Database(format!("清理 Codex 用量汇总失败: {error}")))?; + } + if sqlite_table_exists(conn, "session_log_sync")? + && sqlite_column_exists(conn, "session_log_sync", "file_path")? + { + let paths = { + let mut statement = conn + .prepare("SELECT file_path FROM session_log_sync") + .map_err(|error| { + AppError::Database(format!("读取会话同步 cursor 失败: {error}")) + })?; + let paths = statement + .query_map([], |row| row.get::<_, String>(0)) + .map_err(|error| AppError::Database(format!("查询会话同步 cursor 失败: {error}")))? + .collect::, _>>() + .map_err(|error| { + AppError::Database(format!("解析会话同步 cursor 失败: {error}")) + })?; + paths + }; + for file_path in paths + .into_iter() + .filter(|path| is_codex_cursor_path(path, codex_dir)) + { + conn.execute( + "DELETE FROM session_log_sync WHERE file_path = ?1", + [file_path], + ) + .map_err(|error| AppError::Database(format!("清理 Codex 同步 cursor 失败: {error}")))?; + } + } + Ok(()) +} + +impl Database { + pub(crate) fn reset_codex_usage(&self) -> Result<(), AppError> { + let codex_dir = get_codex_config_dir(); + let conn = lock_conn!(self.conn); + conn.execute("SAVEPOINT reset_codex_usage", []) + .map_err(|error| AppError::Database(format!("开启 Codex 重建事务失败: {error}")))?; + let result = reset_codex_usage_on_conn(&conn, &codex_dir); + match result { + Ok(()) => { + conn.execute("RELEASE reset_codex_usage", []) + .map_err(|error| { + AppError::Database(format!("提交 Codex 重建事务失败: {error}")) + })?; + drop(conn); + clear_codex_replay_caches(); + Ok(()) + } + Err(error) => { + conn.execute("ROLLBACK TO reset_codex_usage", []).ok(); + conn.execute("RELEASE reset_codex_usage", []).ok(); + Err(error) + } + } + } +} + +fn non_empty_string(value: Option<&serde_json::Value>) -> Option { + value + .and_then(serde_json::Value::as_str) + .filter(|value| !value.is_empty()) + .map(str::to_owned) +} + +fn thread_id_from_filename(path: &Path) -> Option { + let stem = path.file_stem()?.to_str()?; + let candidate = stem.get(stem.len().checked_sub(36)?..)?; + uuid::Uuid::parse_str(candidate) + .ok() + .map(|value| value.hyphenated().to_string()) +} + +fn explicit_parent_from_meta(payload: &serde_json::Value) -> ParentResolution { + let forked_from = non_empty_string(payload.get("forked_from_id")); + let spawned_from = payload + .get("source") + .and_then(|source| source.get("subagent")) + .and_then(|subagent| subagent.get("thread_spawn")) + .and_then(|spawn| non_empty_string(spawn.get("parent_thread_id"))); + + match (forked_from, spawned_from) { + (None, None) => ParentResolution::None, + (Some(parent), None) | (None, Some(parent)) => ParentResolution::Parent(parent), + (Some(forked), Some(spawned)) if forked == spawned => ParentResolution::Parent(forked), + (Some(forked), Some(spawned)) => ParentResolution::Deferred(format!( + "forked_from_id ({forked}) 与 thread_spawn.parent_thread_id ({spawned}) 不一致" + )), + } +} + +fn parse_timestamp(value: Option<&serde_json::Value>) -> Option> { + value + .and_then(serde_json::Value::as_str) + .and_then(|value| DateTime::parse_from_rfc3339(value).ok()) + .map(|value| value.with_timezone(&Utc)) +} + +fn parse_signature_counters(value: Option<&serde_json::Value>) -> Option { + let value = value?.as_object()?; + Some(TokenCountersSignature { + input: value + .get("input_tokens") + .and_then(serde_json::Value::as_u64), + cached_input: value + .get("cached_input_tokens") + .or_else(|| value.get("cache_read_input_tokens")) + .and_then(serde_json::Value::as_u64), + output: value + .get("output_tokens") + .and_then(serde_json::Value::as_u64), + reasoning_output: value + .get("reasoning_output_tokens") + .and_then(serde_json::Value::as_u64), + total: value + .get("total_tokens") + .and_then(serde_json::Value::as_u64), }) } -fn read_codex_session_identity(file_path: &Path) -> Option { - let file = fs::File::open(file_path).ok()?; - - for line in BufReader::new(file).lines() { - let Ok(line) = line else { - continue; - }; - if !line.contains("\"session_meta\"") { - continue; - } - let Ok(value) = serde_json::from_str::(&line) else { - continue; - }; - if value.get("type").and_then(|value| value.as_str()) != Some("session_meta") { - continue; - } - if let Some(identity) = value.get("payload").and_then(parse_codex_session_identity) { - return Some(identity); - } - } - - None -} - -/// fork/子代理日志会先重放父线程历史，再以接管事件开始当前线程。 -/// 返回接管事件所在行；此前的 token_count 只用于恢复累计值基线。 -fn codex_history_replay_boundary( - file_path: &Path, - identity: Option<&CodexSessionIdentity>, -) -> Option { - if !identity.is_some_and(|identity| identity.carries_history_snapshot) { - return None; - } - - let file = fs::File::open(file_path).ok()?; - for (index, line) in BufReader::new(file).lines().enumerate() { - let Ok(line) = line else { - continue; - }; - if !line.contains("\"thread_settings_applied\"") - && !line.contains("\"inter_agent_communication") - { - continue; - } - let Ok(value) = serde_json::from_str::(&line) else { - continue; - }; - let Some(event_type) = value.get("type").and_then(|value| value.as_str()) else { - continue; - }; - let is_replay_boundary = event_type.starts_with("inter_agent_communication") - || (event_type == "event_msg" - && value - .get("payload") - .and_then(|payload| payload.get("type")) - .and_then(|value| value.as_str()) - == Some("thread_settings_applied")); - if is_replay_boundary { - return Some(index as i64 + 1); - } - } - - None -} - -fn is_history_snapshot_event(state: &FileParseState, line_offset: i64) -> bool { - state - .history_replay_boundary - .is_some_and(|boundary| line_offset < boundary) +fn parse_token_signature(info: &serde_json::Value) -> Option { + let total = parse_signature_counters(info.get("total_token_usage")); + let last = parse_signature_counters(info.get("last_token_usage")); + (total.is_some() || last.is_some()).then_some(TokenUsageSignature { total, last }) } fn get_codex_sync_state(db: &Database, file_path: &Path) -> Result<(i64, i64), AppError> { @@ -296,30 +461,42 @@ fn parse_cumulative_tokens(total_usage: &serde_json::Value) -> Option>; + +#[derive(Debug, Default)] +struct CodexFileSyncResult { + imported: u32, + skipped: u32, + suspected_duplicates: u32, + deferred: bool, +} + /// 同步 Codex 使用数据（从 JSONL 会话日志） pub fn sync_codex_usage(db: &Database) -> Result { let codex_dir = get_codex_config_dir(); - let files = collect_codex_session_files(&codex_dir); + let rollout_index = build_rollout_index(&files); let mut result = SessionSyncResult { imported: 0, skipped: 0, files_scanned: files.len() as u32, - errors: vec![], suspected_duplicates: 0, deferred_files: 0, + errors: vec![], }; - if files.is_empty() { - return Ok(result); - } - for file_path in &files { - match sync_single_codex_file(db, file_path) { - Ok((imported, skipped)) => { - result.imported += imported; - result.skipped += skipped; + match sync_single_codex_file(db, file_path, &rollout_index) { + Ok(file_result) => { + result.imported = result.imported.saturating_add(file_result.imported); + result.skipped = result.skipped.saturating_add(file_result.skipped); + result.suspected_duplicates = result + .suspected_duplicates + .saturating_add(file_result.suspected_duplicates); + if file_result.deferred { + result.deferred_files = result.deferred_files.saturating_add(1); + } } Err(e) => { let msg = format!("Codex 会话文件解析失败 {}: {e}", file_path.display()); @@ -329,11 +506,12 @@ pub fn sync_codex_usage(db: &Database) -> Result { } } - if result.imported > 0 { + if result.imported > 0 || result.deferred_files > 0 { log::info!( - "[CODEX-SYNC] 同步完成: 导入 {} 条, 跳过 {} 条, 扫描 {} 个文件", + "[CODEX-SYNC] 同步完成: 导入 {} 条, 跳过 {} 条, deferred {} 个, 扫描 {} 个文件", result.imported, result.skipped, + result.deferred_files, result.files_scanned ); } @@ -364,9 +542,23 @@ fn collect_codex_session_files(codex_dir: &Path) -> Vec { } } + files.sort(); files } +fn build_rollout_index(files: &[PathBuf]) -> RolloutIndex { + let mut index = RolloutIndex::new(); + for path in files { + if let Some(thread_id) = thread_id_from_filename(path) { + index.entry(thread_id).or_default().push(path.clone()); + } + } + for paths in index.values_mut() { + paths.sort(); + } + index +} + /// 递归扫描目录下的 .jsonl 文件（限制最大深度） fn collect_jsonl_recursive(dir: &Path, files: &mut Vec, depth: u32, max_depth: u32) { let entries = match fs::read_dir(dir) { @@ -384,60 +576,36 @@ fn collect_jsonl_recursive(dir: &Path, files: &mut Vec, depth: u32, max } } -/// 同步单个 Codex JSONL 文件，返回 (imported, skipped) -fn sync_single_codex_file(db: &Database, file_path: &Path) -> Result<(u32, u32), AppError> { - let file_path_str = file_path.to_string_lossy().to_string(); - - // 获取文件元数据 - let metadata = fs::metadata(file_path) - .map_err(|e| AppError::Config(format!("无法读取文件元数据: {e}")))?; - let file_modified = metadata_modified_nanos(&metadata); - - // 检查同步状态 - let (last_modified, last_offset) = get_codex_sync_state(db, file_path)?; - - // 文件未变化则跳过 - if file_modified <= last_modified { - return Ok((0, 0)); - } - - let identity = read_codex_session_identity(file_path); - - // 打开文件逐行解析 +fn parse_codex_file( + file_path: &Path, + root_thread_id: Option, +) -> Result { let file = fs::File::open(file_path).map_err(|e| AppError::Config(format!("无法打开文件: {e}")))?; let reader = BufReader::new(file); - let history_replay_boundary = codex_history_replay_boundary(file_path, identity.as_ref()); - - let mut state = FileParseState { - thread_id: identity.map(|identity| identity.thread_id), - current_model: "unknown".to_string(), - prev_total: None, - event_index: 0, - history_replay_boundary, - }; - - let mut line_offset: i64 = 0; - let mut imported: u32 = 0; - let mut skipped: u32 = 0; + let mut root_meta_seen = false; + let mut root_timestamp = None; + let mut parent = ParentResolution::None; + let mut current_model = "unknown".to_string(); + let mut prev_total: Option = None; + let mut event_index = 0u32; + let mut token_events = Vec::new(); + let mut line_offset = 0i64; + let mut has_billable_tokens = false; for line_result in reader.lines() { line_offset += 1; - let line = match line_result { - Ok(l) => l, - Err(_) => continue, // 容忍不完整的最后一行 + Ok(line) => line, + Err(_) => continue, }; - if line.trim().is_empty() { continue; } - // 快速过滤：在 JSON 反序列化前跳过无关行 let is_event_msg = line.contains("\"event_msg\""); let is_turn_context = line.contains("\"turn_context\""); let is_session_meta = line.contains("\"session_meta\""); - if !is_event_msg && !is_turn_context && !is_session_meta { continue; } @@ -446,61 +614,85 @@ fn sync_single_codex_file(db: &Database, file_path: &Path) -> Result<(u32, u32), } let value: serde_json::Value = match serde_json::from_str(&line) { - Ok(v) => v, + Ok(value) => value, Err(_) => continue, }; - - let event_type = match value.get("type").and_then(|t| t.as_str()) { - Some(t) => t, - None => continue, + let Some(event_type) = value.get("type").and_then(serde_json::Value::as_str) else { + continue; }; match event_type { - "session_meta" if state.thread_id.is_none() => { - state.thread_id = value - .get("payload") - .and_then(parse_codex_session_identity) - .map(|identity| identity.thread_id); + "session_meta" if !root_meta_seen => { + root_meta_seen = true; + root_timestamp = parse_timestamp(value.get("timestamp")); + let payload = value.get("payload").unwrap_or(&serde_json::Value::Null); + parent = explicit_parent_from_meta(payload); + + let meta_thread_id = non_empty_string( + payload + .get("id") + .or_else(|| payload.get("thread_id")) + .or_else(|| payload.get("threadId")), + ); + if let (Some(filename_id), Some(meta_id)) = (&root_thread_id, meta_thread_id) { + if filename_id != &meta_id { + parent = ParentResolution::Deferred(format!( + "文件名线程 ID ({filename_id}) 与 root meta ID ({meta_id}) 不一致" + )); + } + } + + if let ParentResolution::Parent(parent_id) = &mut parent { + match uuid::Uuid::parse_str(parent_id) { + Ok(value) => *parent_id = value.hyphenated().to_string(), + Err(_) => { + parent = ParentResolution::Deferred(format!( + "显式 parent_thread_id 不是有效 UUID: {parent_id}" + )); + } + } + } + if matches!((&root_thread_id, &parent), (Some(root), ParentResolution::Parent(parent_id)) if root == parent_id) + { + parent = ParentResolution::Deferred( + "parent_thread_id 与 root_thread_id 相同".to_string(), + ); + } } "turn_context" => { if let Some(payload) = value.get("payload") { - // model 可能在 payload.model 或 payload.info.model if let Some(model) = payload .get("model") .or_else(|| payload.get("info").and_then(|info| info.get("model"))) - .and_then(|v| v.as_str()) + .and_then(serde_json::Value::as_str) { - state.current_model = normalize_codex_model(model); + current_model = normalize_codex_model(model); } } } "event_msg" => { - let payload = match value.get("payload") { - Some(p) => p, - None => continue, + let Some(payload) = value.get("payload") else { + continue; }; - - // 只处理 token_count 类型 - if payload.get("type").and_then(|t| t.as_str()) != Some("token_count") { + if payload.get("type").and_then(serde_json::Value::as_str) != Some("token_count") { continue; } - - let info = match payload.get("info") { - Some(i) if !i.is_null() => i, - _ => continue, // 跳过 info 为 null 的首个事件 + let Some(info) = payload.get("info").filter(|info| !info.is_null()) else { + continue; + }; + let Some(signature) = parse_token_signature(info) else { + continue; }; - // 提取模型（token_count 事件也可能携带 model） if let Some(model) = info .get("model") .or_else(|| info.get("model_name")) .or_else(|| payload.get("model")) - .and_then(|v| v.as_str()) + .and_then(serde_json::Value::as_str) { - state.current_model = normalize_codex_model(model); + current_model = normalize_codex_model(model); } - // 优先用 total_token_usage（累计值），fallback 到 last_token_usage（增量值） let (cumulative, is_total) = if let Some(total) = info.get("total_token_usage") { (parse_cumulative_tokens(total), true) } else if let Some(last) = info.get("last_token_usage") { @@ -508,89 +700,376 @@ fn sync_single_codex_file(db: &Database, file_path: &Path) -> Result<(u32, u32), } else { continue; }; - - let cumulative = match cumulative { - Some(c) => c, - None => continue, + let Some(cumulative) = cumulative else { + continue; }; - let delta = if is_total { - // 累计值模式：计算与上次的 delta - let d = compute_delta(&state.prev_total, &cumulative); - state.prev_total = Some(cumulative); - d + let delta = compute_delta(&prev_total, &cumulative); + prev_total = Some(cumulative); + delta } else { - // 增量值模式：直接使用 last_token_usage 的值 DeltaTokens { input: cumulative.input as u32, cached_input: cumulative.cached_input as u32, output: cumulative.output as u32, } }; - - // 钳制：cached 不应超过 input（防护异常数据） let delta = DeltaTokens { cached_input: delta.cached_input.min(delta.input), ..delta }; + let nonzero_index = if delta.is_zero() { + None + } else { + has_billable_tokens = true; + event_index = event_index.saturating_add(1); + Some(event_index) + }; - if delta.is_zero() { - continue; // 跳过 task 边界的零 delta 事件 - } - - // 所有非零事件都占据稳定序号，包括已同步事件与 replay 快照。 - state.event_index += 1; - - // replay 快照更新了 prev_total，但不是当前线程的新用量。 - if is_history_snapshot_event(&state, line_offset) { - if line_offset > last_offset { - skipped += 1; - } - continue; - } - - // 跳过已处理的行（但仍需解析以恢复状态） - if line_offset <= last_offset { - continue; - } - - // 生成唯一 request_id - let thread_id = state.thread_id.as_deref().unwrap_or("unknown"); - let request_id = format!( - "{CODEX_THREAD_REQUEST_ID_PREFIX}:{thread_id}:{}", - state.event_index - ); - - // 提取时间戳 - let timestamp = value - .get("timestamp") - .and_then(|v| v.as_str()) - .map(|s| s.to_string()); - - match insert_codex_session_entry( - db, - &request_id, - &delta, - &state.current_model, - state.thread_id.as_deref(), - timestamp.as_deref(), - ) { - Ok(true) => imported += 1, - Ok(false) => skipped += 1, - Err(e) => { - log::warn!("[CODEX-SYNC] 插入失败 ({}): {e}", request_id); - skipped += 1; - } - } + token_events.push(ParsedTokenEvent { + line_offset, + signature, + delta, + event_index: nonzero_index, + model: current_model.clone(), + timestamp: value + .get("timestamp") + .and_then(serde_json::Value::as_str) + .map(str::to_owned), + }); } _ => {} } } - // 更新同步状态 - update_sync_state(db, &file_path_str, file_modified, line_offset)?; + Ok(ParsedCodexFile { + root_thread_id, + root_meta_seen, + root_timestamp, + parent, + token_events, + line_offset, + has_billable_tokens, + }) +} - Ok((imported, skipped)) +fn parent_signatures_before( + parent_path: &Path, + cutoff: DateTime, +) -> Result, String> { + let cache_key = (parent_path.to_path_buf(), cutoff.timestamp_micros()); + if let Ok(caches) = replay_caches().lock() { + if let Some(signatures) = caches.parent_signatures.get(&cache_key) { + return Ok(signatures.clone()); + } + } + + let file = fs::File::open(parent_path) + .map_err(|error| format!("无法打开父 rollout {}: {error}", parent_path.display()))?; + let mut signatures = Vec::new(); + let mut max_timestamp: Option> = None; + + // 必须扫描完整父文件并逐行应用 cutoff，不能在首个未来时间戳处 break： + // rollout 写入顺序不承诺时间戳严格单调。 + for line in BufReader::new(file).lines() { + let Ok(line) = line else { + continue; + }; + let Ok(value) = serde_json::from_str::(&line) else { + continue; + }; + let timestamp = parse_timestamp(value.get("timestamp")); + if let Some(timestamp) = timestamp { + max_timestamp = Some(max_timestamp.map_or(timestamp, |current| current.max(timestamp))); + } + if value.get("type").and_then(serde_json::Value::as_str) != Some("event_msg") + || value + .get("payload") + .and_then(|payload| payload.get("type")) + .and_then(serde_json::Value::as_str) + != Some("token_count") + { + continue; + } + let Some(info) = value + .get("payload") + .and_then(|payload| payload.get("info")) + .filter(|info| !info.is_null()) + else { + continue; + }; + let Some(signature) = parse_token_signature(info) else { + continue; + }; + let Some(timestamp) = timestamp else { + return Err(format!( + "父 rollout {} 的 token_count 缺少有效 timestamp", + parent_path.display() + )); + }; + if timestamp <= cutoff { + signatures.push(signature); + } + } + + if max_timestamp.is_none_or(|timestamp| timestamp < cutoff) { + return Err(format!( + "父 rollout {} 尚未写到 child fork 时刻", + parent_path.display() + )); + } + + if let Ok(mut caches) = replay_caches().lock() { + caches + .parent_signatures + .insert(cache_key, signatures.clone()); + } + Ok(signatures) +} + +fn resolve_parent_signatures( + parent_id: &str, + cutoff: DateTime, + rollout_index: &RolloutIndex, +) -> Result, String> { + let Some(candidates) = rollout_index.get(parent_id) else { + return Err(format!("找不到父 rollout: {parent_id}")); + }; + + let mut snapshots = Vec::with_capacity(candidates.len()); + for candidate in candidates { + snapshots.push(parent_signatures_before(candidate, cutoff)?); + } + let Some(first) = snapshots.first() else { + return Err(format!("找不到父 rollout: {parent_id}")); + }; + if snapshots.iter().skip(1).any(|snapshot| snapshot != first) { + return Err(format!( + "父 rollout UUID {parent_id} 对应多个内容不一致的文件" + )); + } + Ok(first.clone()) +} + +fn matching_replay_prefix(child: &[ParsedTokenEvent], parent: &[TokenUsageSignature]) -> usize { + let mut parent_offset = 0usize; + let mut matched = 0usize; + for event in child { + let Some(relative_match) = parent[parent_offset..] + .iter() + .position(|signature| signature == &event.signature) + else { + break; + }; + parent_offset += relative_match + 1; + matched += 1; + } + matched +} + +fn mark_deferred( + file_path: &Path, + modified: i64, + size: u64, + reason: PendingReason, +) -> CodexFileSyncResult { + let entry = PendingEntry { + modified, + size, + reason, + }; + let should_warn = replay_caches() + .lock() + .ok() + .and_then(|mut caches| { + caches + .pending + .insert(file_path.to_path_buf(), entry.clone()) + }) + .as_ref() + != Some(&entry); + if should_warn { + let reason = match &entry.reason { + PendingReason::MissingParent(parent) => format!("找不到父 rollout {parent}"), + PendingReason::Stable(reason) | PendingReason::Retryable(reason) => reason.clone(), + }; + log::warn!("[CODEX-SYNC] deferred {}: {reason}", file_path.display()); + } + CodexFileSyncResult { + deferred: true, + ..CodexFileSyncResult::default() + } +} + +/// 同步单个 Codex JSONL 文件。 +fn sync_single_codex_file( + db: &Database, + file_path: &Path, + rollout_index: &RolloutIndex, +) -> Result { + let file_path_str = file_path.to_string_lossy().to_string(); + + // 获取文件元数据 + let metadata = fs::metadata(file_path) + .map_err(|e| AppError::Config(format!("无法读取文件元数据: {e}")))?; + let file_modified = metadata_modified_nanos(&metadata); + let file_size = metadata.len(); + + // 检查同步状态 + let (last_modified, last_offset) = get_codex_sync_state(db, file_path)?; + + // 文件未变化则跳过 + if file_modified <= last_modified { + return Ok(CodexFileSyncResult::default()); + } + + if let Ok(mut caches) = replay_caches().lock() { + if let Some(pending) = caches.pending.get(file_path).cloned() { + if pending.modified == file_modified && pending.size == file_size { + match &pending.reason { + PendingReason::MissingParent(parent) if !rollout_index.contains_key(parent) => { + return Ok(CodexFileSyncResult { + deferred: true, + ..CodexFileSyncResult::default() + }); + } + PendingReason::Stable(_) => { + return Ok(CodexFileSyncResult { + deferred: true, + ..CodexFileSyncResult::default() + }); + } + PendingReason::Retryable(_) => { + caches.pending.remove(file_path); + } + _ => { + caches.pending.remove(file_path); + } + } + } + } + } + + let parsed = parse_codex_file(file_path, thread_id_from_filename(file_path))?; + if !parsed.has_billable_tokens { + update_sync_state(db, &file_path_str, file_modified, parsed.line_offset)?; + return Ok(CodexFileSyncResult::default()); + } + let Some(root_thread_id) = parsed.root_thread_id.as_deref() else { + return Ok(mark_deferred( + file_path, + file_modified, + file_size, + PendingReason::Stable("文件名缺少有效的尾部 UUID".to_string()), + )); + }; + if !parsed.root_meta_seen { + return Ok(mark_deferred( + file_path, + file_modified, + file_size, + PendingReason::Stable("含计费 token 但尚无 session_meta".to_string()), + )); + } + + let replay_prefix = match &parsed.parent { + ParentResolution::None => 0, + ParentResolution::Deferred(reason) => { + return Ok(mark_deferred( + file_path, + file_modified, + file_size, + PendingReason::Stable(reason.clone()), + )); + } + ParentResolution::Parent(parent_id) => { + let Some(cutoff) = parsed.root_timestamp else { + return Ok(mark_deferred( + file_path, + file_modified, + file_size, + PendingReason::Stable( + "parented rollout 的 root meta 缺少有效 timestamp".to_string(), + ), + )); + }; + let cache_key = (file_path.to_path_buf(), file_modified, file_size); + if let Ok(caches) = replay_caches().lock() { + if let Some(prefix) = caches.replay_prefixes.get(&cache_key) { + *prefix + } else { + drop(caches); + let parent_signatures = + match resolve_parent_signatures(parent_id, cutoff, rollout_index) { + Ok(signatures) => signatures, + Err(reason) => { + let pending_reason = if rollout_index.contains_key(parent_id) { + PendingReason::Retryable(reason) + } else { + PendingReason::MissingParent(parent_id.clone()) + }; + return Ok(mark_deferred( + file_path, + file_modified, + file_size, + pending_reason, + )); + } + }; + let prefix = matching_replay_prefix(&parsed.token_events, &parent_signatures); + if let Ok(mut caches) = replay_caches().lock() { + caches.replay_prefixes.insert(cache_key, prefix); + } + prefix + } + } else { + let parent_signatures = resolve_parent_signatures(parent_id, cutoff, rollout_index) + .map_err(AppError::Config)?; + matching_replay_prefix(&parsed.token_events, &parent_signatures) + } + } + }; + + if let Ok(mut caches) = replay_caches().lock() { + caches.pending.remove(file_path); + } + + let mut result = CodexFileSyncResult::default(); + for (token_offset, event) in parsed.token_events.iter().enumerate() { + let Some(event_index) = event.event_index else { + continue; + }; + if token_offset < replay_prefix { + if event.line_offset > last_offset { + result.skipped = result.skipped.saturating_add(1); + } + continue; + } + 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); + } + } + } + + update_sync_state(db, &file_path_str, file_modified, parsed.line_offset)?; + Ok(result) } /// 插入单条 Codex 会话记录到 proxy_request_logs @@ -601,6 +1080,7 @@ fn insert_codex_session_entry( model: &str, session_id: Option<&str>, timestamp: Option<&str>, + suspected_duplicates: &mut u32, ) -> Result { let conn = lock_conn!(db.conn); @@ -629,6 +1109,15 @@ fn insert_codex_session_entry( if should_skip_session_insert(&conn, request_id, &dedup_key)? { return Ok(false); } + 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={}", + delta.input, + delta.output, + delta.cached_input + ); + } // 计算费用 let usage = TokenUsage { @@ -701,11 +1190,7 @@ fn insert_codex_session_entry( ) .map_err(|e| AppError::Database(format!("插入 Codex 会话日志失败: {e}")))?; - if inserted_rows > 0 { - crate::usage_events::notify_log_recorded(); - } - - Ok(true) + Ok(inserted_rows > 0) } /// 查找 Codex 模型定价（带归一化） @@ -718,6 +1203,10 @@ mod tests { use super::*; use tempfile::tempdir; + const PARENT_ID: &str = "00000000-0000-4000-8000-000000000001"; + const CHILD_A_ID: &str = "00000000-0000-4000-8000-000000000002"; + const CHILD_B_ID: &str = "00000000-0000-4000-8000-000000000003"; + fn write_jsonl(path: &Path, values: &[serde_json::Value]) { let contents = values .iter() @@ -728,45 +1217,86 @@ mod tests { fs::write(path, contents).unwrap(); } - fn session_meta(thread_id: &str, session_id: &str) -> serde_json::Value { + fn rollout_path(dir: &Path, thread_id: &str) -> PathBuf { + dir.join(format!("rollout-2026-07-10T03-00-00-{thread_id}.jsonl")) + } + + fn session_meta_at( + thread_id: &str, + forked_from_id: Option<&str>, + spawned_from_id: Option<&str>, + timestamp: &str, + ) -> serde_json::Value { + let source = spawned_from_id.map_or_else( + || serde_json::Value::String("cli".to_string()), + |parent| { + serde_json::json!({ + "subagent": { + "thread_spawn": { "parent_thread_id": parent } + } + }) + }, + ); serde_json::json!({ - "timestamp": "2026-07-10T03:00:00Z", + "timestamp": timestamp, "type": "session_meta", "payload": { "id": thread_id, - "session_id": session_id, - "source": if thread_id == session_id { - serde_json::Value::String("cli".to_string()) - } else { - serde_json::json!({ "subagent": {} }) - } + "forked_from_id": forked_from_id, + "source": source } }) } - fn turn_context() -> serde_json::Value { + fn session_meta(thread_id: &str) -> serde_json::Value { + session_meta_at(thread_id, None, None, "2026-07-10T03:00:00Z") + } + + fn turn_context_at(timestamp: &str) -> serde_json::Value { serde_json::json!({ - "timestamp": "2026-07-10T03:00:01Z", + "timestamp": timestamp, "type": "turn_context", "payload": { "model": "gpt-5.6-sol" } }) } - fn token_count(input: u64, cached: u64, output: u64) -> serde_json::Value { + fn turn_context() -> serde_json::Value { + turn_context_at("2026-07-10T03:00:01Z") + } + + fn token_count_at(input: u64, cached: u64, output: u64, timestamp: &str) -> serde_json::Value { serde_json::json!({ - "timestamp": "2026-07-10T03:00:02Z", + "timestamp": timestamp, "type": "event_msg", "payload": { "type": "token_count", "info": { "total_token_usage": { "input_tokens": input, "cached_input_tokens": cached, - "output_tokens": output + "output_tokens": output, + "reasoning_output_tokens": 0, + "total_tokens": input + output }} } }) } + fn token_count(input: u64, cached: u64, output: u64) -> serde_json::Value { + token_count_at(input, cached, output, "2026-07-10T03:00:02Z") + } + + fn sync_test_file( + db: &Database, + file: &Path, + all_files: &[&Path], + ) -> Result { + let files = all_files + .iter() + .map(|path| path.to_path_buf()) + .collect::>(); + sync_single_codex_file(db, file, &build_rollout_index(&files)) + } + #[test] fn test_delta_first_event() { let prev = None; @@ -877,61 +1407,270 @@ mod tests { } #[test] - fn test_subagent_identity_prefers_unique_thread_id() { - let identity = - parse_codex_session_identity(session_meta("child", "parent").get("payload").unwrap()) - .unwrap(); - - assert_eq!(identity.thread_id, "child"); - assert!(identity.carries_history_snapshot); - } - - #[test] - fn test_subagent_replay_only_establishes_token_baseline() -> Result<(), AppError> { + fn test_thread_spawn_parent_strips_replay_and_keeps_live_usage() -> Result<(), AppError> { + clear_codex_replay_caches(); let db = Database::memory()?; let temp = tempdir().unwrap(); - let child = temp.path().join("child.jsonl"); + let parent = rollout_path(temp.path(), PARENT_ID); + let child = rollout_path(temp.path(), CHILD_A_ID); + write_jsonl( + &parent, + &[ + session_meta(PARENT_ID), + token_count_at(1_000, 900, 100, "2026-07-10T03:00:01Z"), + turn_context_at("2026-07-10T03:00:10Z"), + ], + ); write_jsonl( &child, &[ - session_meta("child", "parent"), + session_meta_at(CHILD_A_ID, None, Some(PARENT_ID), "2026-07-10T03:00:05Z"), turn_context(), - token_count(1_000, 900, 100), - token_count(1_200, 1_000, 120), - serde_json::json!({ - "timestamp": "2026-07-10T03:00:03Z", - "type": "event_msg", - "payload": { "type": "thread_settings_applied" } - }), - token_count(1_300, 1_050, 150), + token_count_at(1_000, 900, 100, "2026-07-10T03:00:06Z"), + token_count_at(1_300, 1_050, 150, "2026-07-10T03:00:07Z"), ], ); - assert_eq!(sync_single_codex_file(&db, &child)?, (1, 2)); + let result = sync_test_file(&db, &child, &[&parent, &child])?; + assert_eq!( + (result.imported, result.skipped, result.deferred), + (1, 1, false) + ); let conn = lock_conn!(db.conn); let usage: (i64, i64, i64) = conn.query_row( "SELECT input_tokens, cache_read_tokens, output_tokens - FROM proxy_request_logs - WHERE request_id = 'codex_session:thread-v1:child:3'", - [], + FROM proxy_request_logs WHERE request_id = ?1", + [format!("{CODEX_THREAD_REQUEST_ID_PREFIX}:{CHILD_A_ID}:2")], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), )?; - assert_eq!(usage, (100, 50, 30)); - + assert_eq!(usage, (300, 150, 50)); Ok(()) } #[test] - fn test_subagents_under_same_parent_use_distinct_request_ids() -> Result<(), AppError> { + fn test_filtered_parent_events_use_subsequence_prefix_alignment() -> Result<(), AppError> { + clear_codex_replay_caches(); let db = Database::memory()?; let temp = tempdir().unwrap(); - let child_a = temp.path().join("child-a.jsonl"); - let child_b = temp.path().join("child-b.jsonl"); + let parent = rollout_path(temp.path(), PARENT_ID); + let child = rollout_path(temp.path(), CHILD_A_ID); + write_jsonl( + &parent, + &[ + session_meta(PARENT_ID), + token_count_at(100, 50, 10, "2026-07-10T03:00:01Z"), + token_count_at(200, 100, 20, "2026-07-10T03:00:02Z"), + token_count_at(300, 150, 30, "2026-07-10T03:00:03Z"), + turn_context_at("2026-07-10T03:00:10Z"), + ], + ); + write_jsonl( + &child, + &[ + session_meta_at(CHILD_A_ID, Some(PARENT_ID), None, "2026-07-10T03:00:05Z"), + token_count_at(100, 50, 10, "2026-07-10T03:00:06Z"), + token_count_at(300, 150, 30, "2026-07-10T03:00:07Z"), + token_count_at(450, 220, 45, "2026-07-10T03:00:08Z"), + ], + ); + + let result = sync_test_file(&db, &child, &[&parent, &child])?; + assert_eq!((result.imported, result.skipped), (1, 2)); + Ok(()) + } + + #[test] + fn test_empty_fork_imports_no_parent_usage() -> Result<(), AppError> { + clear_codex_replay_caches(); + let db = Database::memory()?; + let temp = tempdir().unwrap(); + let parent = rollout_path(temp.path(), PARENT_ID); + let child = rollout_path(temp.path(), CHILD_A_ID); + write_jsonl( + &parent, + &[ + session_meta(PARENT_ID), + token_count_at(100, 50, 10, "2026-07-10T03:00:01Z"), + token_count_at(200, 100, 20, "2026-07-10T03:00:02Z"), + turn_context_at("2026-07-10T03:00:10Z"), + ], + ); + write_jsonl( + &child, + &[ + session_meta_at(CHILD_A_ID, Some(PARENT_ID), None, "2026-07-10T03:00:05Z"), + token_count_at(100, 50, 10, "2026-07-10T03:00:06Z"), + token_count_at(200, 100, 20, "2026-07-10T03:00:07Z"), + serde_json::json!({ + "timestamp": "2026-07-10T03:00:08Z", + "type": "event_msg", + "payload": { "type": "thread_settings_applied" } + }), + ], + ); + + let result = sync_test_file(&db, &child, &[&parent, &child])?; + assert_eq!( + (result.imported, result.skipped, result.deferred), + (0, 2, false) + ); + let conn = lock_conn!(db.conn); + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM proxy_request_logs WHERE data_source = 'codex_session'", + [], + |row| row.get(0), + )?; + assert_eq!(count, 0); + Ok(()) + } + + #[test] + fn test_conflicting_explicit_parents_are_deferred() -> Result<(), AppError> { + clear_codex_replay_caches(); + let db = Database::memory()?; + let temp = tempdir().unwrap(); + let child = rollout_path(temp.path(), CHILD_A_ID); + write_jsonl( + &child, + &[ + session_meta_at( + CHILD_A_ID, + Some(PARENT_ID), + Some(CHILD_B_ID), + "2026-07-10T03:00:05Z", + ), + token_count_at(100, 50, 10, "2026-07-10T03:00:06Z"), + ], + ); + + let result = sync_test_file(&db, &child, &[&child])?; + assert!(result.deferred); + assert_eq!(get_sync_state(&db, &child.to_string_lossy())?, (0, 0)); + Ok(()) + } + + #[test] + fn test_parent_future_signature_cannot_extend_replay_prefix() -> Result<(), AppError> { + clear_codex_replay_caches(); + let db = Database::memory()?; + let temp = tempdir().unwrap(); + let parent = rollout_path(temp.path(), PARENT_ID); + let child = rollout_path(temp.path(), CHILD_A_ID); + write_jsonl( + &parent, + &[ + session_meta(PARENT_ID), + token_count_at(100, 50, 10, "2026-07-10T03:00:01Z"), + token_count_at(200, 100, 20, "2026-07-10T03:00:06Z"), + ], + ); + write_jsonl( + &child, + &[ + session_meta_at(CHILD_A_ID, Some(PARENT_ID), None, "2026-07-10T03:00:05Z"), + token_count_at(200, 100, 20, "2026-07-10T03:00:07Z"), + ], + ); + + let result = sync_test_file(&db, &child, &[&parent, &child])?; + assert_eq!( + (result.imported, result.skipped, result.deferred), + (1, 0, false) + ); + Ok(()) + } + + #[test] + fn test_missing_parent_is_deferred_and_recovered_without_child_change() -> Result<(), AppError> + { + clear_codex_replay_caches(); + let db = Database::memory()?; + let temp = tempdir().unwrap(); + let parent = rollout_path(temp.path(), PARENT_ID); + let child = rollout_path(temp.path(), CHILD_A_ID); + write_jsonl( + &child, + &[ + session_meta_at(CHILD_A_ID, None, Some(PARENT_ID), "2026-07-10T03:00:05Z"), + token_count_at(900, 400, 90, "2026-07-10T03:00:06Z"), + ], + ); + + let deferred = sync_test_file(&db, &child, &[&child])?; + assert!(deferred.deferred); + assert_eq!(get_sync_state(&db, &child.to_string_lossy())?, (0, 0)); + + write_jsonl( + &parent, + &[ + session_meta(PARENT_ID), + token_count_at(100, 50, 10, "2026-07-10T03:00:01Z"), + turn_context_at("2026-07-10T03:00:10Z"), + ], + ); + let recovered = sync_test_file(&db, &child, &[&parent, &child])?; + assert_eq!((recovered.imported, recovered.deferred), (1, false)); + Ok(()) + } + + #[test] + fn test_billable_file_without_meta_is_deferred_without_cursor() -> Result<(), AppError> { + clear_codex_replay_caches(); + let db = Database::memory()?; + let temp = tempdir().unwrap(); + let child = rollout_path(temp.path(), CHILD_A_ID); + write_jsonl(&child, &[turn_context(), token_count(100, 50, 10)]); + + let result = sync_test_file(&db, &child, &[&child])?; + assert!(result.deferred); + assert_eq!(get_sync_state(&db, &child.to_string_lossy())?, (0, 0)); + + std::thread::sleep(std::time::Duration::from_millis(2)); + write_jsonl( + &child, + &[ + turn_context(), + token_count(100, 50, 10), + session_meta_at(CHILD_A_ID, None, None, "2026-07-10T03:00:03Z"), + ], + ); + let recovered = sync_test_file(&db, &child, &[&child])?; + assert_eq!((recovered.imported, recovered.deferred), (1, false)); + Ok(()) + } + + #[test] + fn test_non_billable_file_without_meta_advances_cursor() -> Result<(), AppError> { + clear_codex_replay_caches(); + let db = Database::memory()?; + let temp = tempdir().unwrap(); + let child = rollout_path(temp.path(), CHILD_A_ID); + write_jsonl( + &child, + &[ + turn_context(), + token_count_at(0, 0, 0, "2026-07-10T03:00:02Z"), + ], + ); + + let result = sync_test_file(&db, &child, &[&child])?; + assert!(!result.deferred); + assert_eq!(get_sync_state(&db, &child.to_string_lossy())?.1, 2); + Ok(()) + } + + #[test] + fn test_subagents_use_filename_thread_ids() -> Result<(), AppError> { + clear_codex_replay_caches(); + let db = Database::memory()?; + let temp = tempdir().unwrap(); + let child_a = rollout_path(temp.path(), CHILD_A_ID); + let child_b = rollout_path(temp.path(), CHILD_B_ID); write_jsonl( &child_a, &[ - session_meta("child-a", "parent"), + session_meta(CHILD_A_ID), turn_context(), token_count(100, 50, 10), ], @@ -939,14 +1678,20 @@ mod tests { write_jsonl( &child_b, &[ - session_meta("child-b", "parent"), + session_meta(CHILD_B_ID), turn_context(), token_count(200, 100, 20), ], ); - assert_eq!(sync_single_codex_file(&db, &child_a)?, (1, 0)); - assert_eq!(sync_single_codex_file(&db, &child_b)?, (1, 0)); + assert_eq!( + sync_test_file(&db, &child_a, &[&child_a, &child_b])?.imported, + 1 + ); + assert_eq!( + sync_test_file(&db, &child_b, &[&child_a, &child_b])?.imported, + 1 + ); let conn = lock_conn!(db.conn); let request_ids = conn @@ -959,11 +1704,10 @@ mod tests { assert_eq!( request_ids, vec![ - "codex_session:thread-v1:child-a:1", - "codex_session:thread-v1:child-b:1" + format!("{CODEX_THREAD_REQUEST_ID_PREFIX}:{CHILD_A_ID}:1"), + format!("{CODEX_THREAD_REQUEST_ID_PREFIX}:{CHILD_B_ID}:1") ] ); - Ok(()) } @@ -975,12 +1719,12 @@ mod tests { let archived = temp.path().join("archived_sessions"); fs::create_dir_all(&sessions).unwrap(); fs::create_dir_all(&archived).unwrap(); - let source = sessions.join("rollout-parent.jsonl"); - let archived_file = archived.join("rollout-parent.jsonl"); + let source = rollout_path(&sessions, PARENT_ID); + let archived_file = rollout_path(&archived, PARENT_ID); write_jsonl( &archived_file, &[ - session_meta("parent", "parent"), + session_meta(PARENT_ID), turn_context(), token_count(100, 50, 10), token_count(200, 100, 20), @@ -1004,8 +1748,14 @@ mod tests { let source_path = source.to_string_lossy().to_string(); update_sync_state(&db, &source_path, 1, 3)?; - assert_eq!(sync_single_codex_file(&db, &archived_file)?, (1, 0)); - assert_eq!(sync_single_codex_file(&db, &archived_file)?, (0, 0)); + assert_eq!( + sync_test_file(&db, &archived_file, &[&archived_file])?.imported, + 1 + ); + assert_eq!( + sync_test_file(&db, &archived_file, &[&archived_file])?.imported, + 0 + ); let conn = lock_conn!(db.conn); let old_row_count: i64 = conn.query_row( @@ -1018,8 +1768,8 @@ mod tests { let usage: (i64, i64, i64) = conn.query_row( "SELECT input_tokens, cache_read_tokens, output_tokens FROM proxy_request_logs - WHERE request_id = 'codex_session:thread-v1:parent:2'", - [], + WHERE request_id = ?1", + [format!("{CODEX_THREAD_REQUEST_ID_PREFIX}:{PARENT_ID}:2")], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), )?; assert_eq!(usage, (100, 50, 10)); @@ -1064,6 +1814,7 @@ mod tests { cached_input: 1, output: 2, }; + let mut suspected_duplicates = 0; let inserted = insert_codex_session_entry( &db, "codex-session-dup", @@ -1071,6 +1822,7 @@ mod tests { "gpt-5.4", Some("session-1"), Some("1970-01-01T00:16:45Z"), + &mut suspected_duplicates, )?; assert!(!inserted); @@ -1083,6 +1835,111 @@ mod tests { Ok(()) } + #[test] + fn test_codex_session_duplicate_is_observed_but_still_inserted() -> Result<(), AppError> { + let db = Database::memory()?; + let delta = DeltaTokens { + input: 10, + cached_input: 1, + output: 2, + }; + let mut suspected_duplicates = 0; + assert!(insert_codex_session_entry( + &db, + "codex-session-a", + &delta, + "gpt-5.4", + Some("session-a"), + Some("1970-01-01T00:16:40Z"), + &mut suspected_duplicates, + )?); + assert!(insert_codex_session_entry( + &db, + "codex-session-b", + &delta, + "gpt-5.4", + Some("session-b"), + Some("1970-01-01T00:16:45Z"), + &mut suspected_duplicates, + )?); + assert_eq!(suspected_duplicates, 1); + + let conn = lock_conn!(db.conn); + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM proxy_request_logs WHERE data_source = 'codex_session'", + [], + |row| row.get(0), + )?; + assert_eq!(count, 2); + Ok(()) + } + + #[test] + fn reset_codex_usage_only_removes_codex_rows_and_structural_cursors() -> Result<(), AppError> { + let db = Database::memory()?; + let temp = tempdir().unwrap(); + let wide_dir = temp.path(); + let current_codex = rollout_path(&wide_dir.join("sessions"), CHILD_A_ID); + let legacy_codex = + format!("C:\\old-codex\\archived_sessions\\rollout-old-{CHILD_B_ID}.jsonl"); + let gemini_cursor = wide_dir.join("gemini/sessions/session-123.json"); + let claude_cursor = wide_dir.join(format!("projects/rollout-{PARENT_ID}.jsonl")); + + { + let conn = lock_conn!(db.conn); + conn.execute_batch( + "INSERT INTO proxy_request_logs ( + request_id, provider_id, app_type, model, input_tokens, + output_tokens, cache_read_tokens, latency_ms, status_code, + created_at, data_source + ) VALUES + ('codex-row', '_codex_session', 'codex', 'gpt', 1, 1, 0, 0, 200, 1, 'codex_session'), + ('gemini-row', '_gemini_session', 'gemini', 'gemini', 1, 1, 0, 0, 200, 1, 'gemini_session'); + INSERT INTO usage_daily_rollups (date, app_type, provider_id, model) + VALUES + ('2026-07-10', 'codex', '_codex_session', 'gpt'), + ('2026-07-10', 'gemini', '_gemini_session', 'gemini');", + )?; + for path in [ + current_codex.to_string_lossy().to_string(), + legacy_codex, + gemini_cursor.to_string_lossy().to_string(), + claude_cursor.to_string_lossy().to_string(), + ] { + conn.execute( + "INSERT INTO session_log_sync + (file_path, last_modified, last_line_offset, last_synced_at) + VALUES (?1, 1, 1, 1)", + [path], + )?; + } + + reset_codex_usage_on_conn(&conn, wide_dir)?; + let codex_rows: i64 = conn.query_row( + "SELECT COUNT(*) FROM proxy_request_logs WHERE data_source = 'codex_session'", + [], + |row| row.get(0), + )?; + let gemini_rows: i64 = conn.query_row( + "SELECT COUNT(*) FROM proxy_request_logs WHERE data_source = 'gemini_session'", + [], + |row| row.get(0), + )?; + let codex_rollups: i64 = conn.query_row( + "SELECT COUNT(*) FROM usage_daily_rollups WHERE provider_id = '_codex_session'", + [], + |row| row.get(0), + )?; + let remaining_cursors: i64 = + conn.query_row("SELECT COUNT(*) FROM session_log_sync", [], |row| { + row.get(0) + })?; + assert_eq!((codex_rows, gemini_rows, codex_rollups), (0, 1, 0)); + assert_eq!(remaining_cursors, 2); + } + Ok(()) + } + // ── 模型名归一化测试 ── #[test]