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<String>,
-    current_model: String,
-    prev_total: Option<CumulativeTokens>,
-    event_index: u32,
-    history_replay_boundary: Option<i64>,
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct TokenCountersSignature {
+    input: Option<u64>,
+    cached_input: Option<u64>,
+    output: Option<u64>,
+    reasoning_output: Option<u64>,
+    total: Option<u64>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct TokenUsageSignature {
+    total: Option<TokenCountersSignature>,
+    last: Option<TokenCountersSignature>,
+}
+
+#[derive(Debug)]
+struct ParsedTokenEvent {
+    line_offset: i64,
+    signature: TokenUsageSignature,
+    delta: DeltaTokens,
+    event_index: Option<u32>,
+    model: String,
+    timestamp: Option<String>,
+}
+
+#[derive(Debug)]
+enum ParentResolution {
+    None,
+    Parent(String),
+    Deferred(String),
+}
+
+#[derive(Debug)]
+struct ParsedCodexFile {
+    root_thread_id: Option<String>,
+    root_meta_seen: bool,
+    root_timestamp: Option<DateTime<Utc>>,
+    parent: ParentResolution,
+    token_events: Vec<ParsedTokenEvent>,
+    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<CodexSessionIdentity> {
-    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<TokenUsageSignature>>,
+    replay_prefixes: HashMap<(PathBuf, i64, u64), usize>,
+    pending: HashMap<PathBuf, PendingEntry>,
+}
+
+static CODEX_REPLAY_CACHES: OnceLock<Mutex<CodexReplayCaches>> = OnceLock::new();
+
+fn replay_caches() -> &'static Mutex<CodexReplayCaches> {
+    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<bool, AppError> {
+    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<bool, AppError> {
+    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::<Result<Vec<_>, _>>()
+                .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<String> {
+    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<String> {
+    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<DateTime<Utc>> {
+    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<TokenCountersSignature> {
+    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<CodexSessionIdentity> {
-    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::<serde_json::Value>(&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<i64> {
-    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::<serde_json::Value>(&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<TokenUsageSignature> {
+    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<Cumulative
     })
 }
 
+type RolloutIndex = HashMap<String, Vec<PathBuf>>;
+
+#[derive(Debug, Default)]
+struct CodexFileSyncResult {
+    imported: u32,
+    skipped: u32,
+    suspected_duplicates: u32,
+    deferred: bool,
+}
+
 /// 同步 Codex 使用数据（从 JSONL 会话日志）
 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 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<SessionSyncResult, AppError> {
         }
     }
 
-    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<PathBuf> {
         }
     }
 
+    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<PathBuf>, 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<PathBuf>, 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<String>,
+) -> Result<ParsedCodexFile, AppError> {
     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<CumulativeTokens> = 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<Utc>,
+) -> Result<Vec<TokenUsageSignature>, 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<DateTime<Utc>> = None;
+
+    // 必须扫描完整父文件并逐行应用 cutoff，不能在首个未来时间戳处 break：
+    // rollout 写入顺序不承诺时间戳严格单调。
+    for line in BufReader::new(file).lines() {
+        let Ok(line) = line else {
+            continue;
+        };
+        let Ok(value) = serde_json::from_str::<serde_json::Value>(&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<Utc>,
+    rollout_index: &RolloutIndex,
+) -> Result<Vec<TokenUsageSignature>, 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<CodexFileSyncResult, 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 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<bool, AppError> {
     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<CodexFileSyncResult, AppError> {
+        let files = all_files
+            .iter()
+            .map(|path| path.to_path_buf())
+            .collect::<Vec<_>>();
+        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]