diff --git a/apps/codex-plus-launcher/src/main.rs b/apps/codex-plus-launcher/src/main.rs
index e2a400a3b..db388d2d0 100644
--- a/apps/codex-plus-launcher/src/main.rs
+++ b/apps/codex-plus-launcher/src/main.rs
@@ -104,10 +104,51 @@ async fn launcher_main(args: Vec<String>, helper_only: bool, options: LaunchOpti
     });
     let hooks = LauncherHooks::default();
     let handle = launch_and_inject_with_hooks(options, &hooks).await?;
-    handle.wait_for_codex_exit().await?;
+    run_periodic_until_exit(
+        handle.wait_for_codex_exit(),
+        std::time::Duration::from_secs(60),
+        || repair_session_index_automatically(true),
+    )
+    .await?;
     Ok(())
 }
 
+// 退出时不再安排下一次检查；已开始的数据库事务先完成，避免脱离启动器生命周期。
+async fn run_periodic_until_exit<F, T, C, W>(exit: F, interval: std::time::Duration, mut check: C) -> T
+where
+    F: std::future::Future<Output = T>,
+    C: FnMut() -> W,
+    W: std::future::Future<Output = ()>,
+{
+    tokio::pin!(exit);
+    loop {
+        tokio::select! {
+            biased;
+            result = &mut exit => return result,
+            _ = tokio::time::sleep(interval) => check().await,
+        }
+    }
+}
+
+async fn repair_session_index_automatically(check_setting: bool) {
+    let result = tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
+        if check_setting && !codex_plus_core::settings::SettingsStore::default().load()?.provider_sync_enabled {
+            return Ok(());
+        }
+        codex_plus_data::repair_session_index(None)?;
+        Ok(())
+    })
+    .await
+    .map_err(anyhow::Error::from)
+    .and_then(|result| result);
+    if let Err(error) = result {
+        let _ = codex_plus_core::diagnostic_log::append_diagnostic_log(
+            "launcher.session_index_repair.failed",
+            json!({ "message": error.to_string() }),
+        );
+    }
+}
+
 fn current_timestamp_ms() -> u64 {
     std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -376,7 +417,9 @@ impl LaunchHooks for LauncherHooks {
         let result = tokio::task::spawn_blocking(|| codex_plus_data::run_provider_sync(None))
             .await
             .map_err(|error| anyhow::anyhow!("provider sync task failed: {error}"))?;
-        require_completed_provider_sync(&result.status, &result.message)
+        require_completed_provider_sync(&result.status, &result.message)?;
+        repair_session_index_automatically(false).await;
+        Ok(())
     }
 
     fn has_pending_remote_control_session_recoveries(&self) -> bool {
@@ -1140,6 +1183,33 @@ fn default_user_scripts_config_dir() -> PathBuf {
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn session_index_monitor_does_not_start_after_exit() {
+        let checks = std::cell::Cell::new(0);
+        let result = run_periodic_until_exit(async { 42 }, std::time::Duration::ZERO, || {
+            checks.set(checks.get() + 1);
+            std::future::ready(())
+        }).await;
+        assert_eq!(result, 42);
+        assert_eq!(checks.get(), 0);
+    }
+
+    #[tokio::test]
+    async fn session_index_monitor_finishes_current_check_before_exit() {
+        let (done, exit) = tokio::sync::oneshot::channel::<()>();
+        let mut done = Some(done);
+        let finished = std::cell::Cell::new(false);
+        run_periodic_until_exit(exit, std::time::Duration::from_millis(1), || {
+            done.take().expect("only one check").send(()).unwrap();
+            let finished = &finished;
+            async move {
+                tokio::task::yield_now().await;
+                finished.set(true);
+            }
+        }).await.unwrap();
+        assert!(finished.get());
+    }
+
     #[test]
     fn parse_launch_options_accepts_manager_forwarded_ports_and_app_path() {
         let options = parse_launch_options([
diff --git a/apps/codex-plus-manager/src-tauri/src/commands.rs b/apps/codex-plus-manager/src-tauri/src/commands.rs
index 1f17a6456..8e7e50e4c 100644
--- a/apps/codex-plus-manager/src-tauri/src/commands.rs
+++ b/apps/codex-plus-manager/src-tauri/src/commands.rs
@@ -3226,6 +3226,35 @@ fn merge_manual_provider_sync_targets(
     });
 }
 
+#[tauri::command]
+pub async fn repair_session_index() -> CommandResult<Value> {
+    let result = tauri::async_runtime::spawn_blocking(|| codex_plus_data::repair_session_index(None))
+        .await
+        .map_err(|error| anyhow::anyhow!("session index repair task failed: {error}"))
+        .and_then(|result| result);
+    match result {
+        Ok(report) => ok(
+            &format!("会话索引检查完成，恢复 {} 条消息。", report.repaired_items),
+            json!(report),
+        ),
+        Err(error) => failed(&format!("修复会话索引失败：{error}"), json!({})),
+    }
+}
+
+#[tauri::command]
+pub async fn load_session_index_repair_report() -> CommandResult<Value> {
+    let result = tauri::async_runtime::spawn_blocking(|| {
+        codex_plus_data::load_session_index_repair_report(None)
+    })
+    .await
+    .map_err(|error| anyhow::anyhow!("session index report task failed: {error}"))
+    .and_then(|result| result);
+    match result {
+        Ok(report) => ok("已读取会话索引修复报告。", json!({ "report": report })),
+        Err(error) => failed(&format!("读取会话索引修复报告失败：{error}"), json!({})),
+    }
+}
+
 #[tauri::command]
 pub async fn preview_session_index_cleanup() -> CommandResult<Value> {
     let result = tauri::async_runtime::spawn_blocking(|| {
diff --git a/apps/codex-plus-manager/src-tauri/src/lib.rs b/apps/codex-plus-manager/src-tauri/src/lib.rs
index d7deb13dd..10c18abf1 100644
--- a/apps/codex-plus-manager/src-tauri/src/lib.rs
+++ b/apps/codex-plus-manager/src-tauri/src/lib.rs
@@ -121,6 +121,8 @@ pub fn run() {
             commands::forget_zed_remote_project,
             commands::delete_local_session,
             commands::load_provider_sync_targets,
+            commands::repair_session_index,
+            commands::load_session_index_repair_report,
             commands::preview_session_index_cleanup,
             commands::apply_session_index_cleanup,
             commands::sync_providers_now,
diff --git a/apps/codex-plus-manager/src/App.tsx b/apps/codex-plus-manager/src/App.tsx
index 9b7a42f8a..feeb7b361 100644
--- a/apps/codex-plus-manager/src/App.tsx
+++ b/apps/codex-plus-manager/src/App.tsx
@@ -785,6 +785,28 @@ type TaskProgress = {
   message: string;
 };
 
+type SessionIndexRepairReport = {
+  scannedFiles: number;
+  cachedFiles: number;
+  repairedItems: number;
+  alreadyPresent: number;
+  skippedItems: number;
+  deferredItems?: number;
+  issues: string[];
+  backupPath: string | null;
+  elapsedMs: number;
+  checkedAtMs?: number;
+  pendingDetails?: {
+    threadId: string | null;
+    turnId: string | null;
+    reason: string;
+    state: "waiting" | "blocked";
+    firstSeenAtMs: number;
+    lastCheckedAtMs: number;
+    checks: number;
+  }[];
+};
+
 type LogsResult = CommandResult<{
   path: string;
   text: string;
@@ -1186,6 +1208,10 @@ export function App() {
     message: t("尚未检查官方远端插件缓存。"),
   });
   const [providerSyncTargets, setProviderSyncTargets] = useState<ProviderSyncTargetsResult | null>(null);
+  const [sessionIndexRepairActive, setSessionIndexRepairActive] = useState(false);
+  const sessionIndexRepairRunning = useRef(false);
+  const sessionIndexReportLoading = useRef(false);
+  const [sessionIndexRepairReport, setSessionIndexRepairReport] = useState<SessionIndexRepairReport | null>(null);
   const [selectedProviderSyncTarget, setSelectedProviderSyncTarget] = useState("");
   const [removeOwnedData, setRemoveOwnedData] = useState(false);
   const [relaySwitching, setRelaySwitching] = useState(false);
@@ -2074,6 +2100,7 @@ export function App() {
       await refreshSettings(true);
       await refreshLocalSessions(true);
       await refreshProviderSyncTargets(true);
+      await refreshSessionIndexRepairReport();
     }
     if (next === "zedRemote") {
       await refreshSettings(true);
@@ -2537,7 +2564,48 @@ export function App() {
     return result;
   };
 
+  const refreshSessionIndexRepairReport = async (isCurrent = () => true) => {
+    if (sessionIndexReportLoading.current || sessionIndexRepairRunning.current) return;
+    sessionIndexReportLoading.current = true;
+    try {
+      // 持久报告读取失败时保留现有结果，后台刷新不触发全局通知或忙碌状态。
+      const result = await call<CommandResult<{ report: SessionIndexRepairReport | null }>>(
+        "load_session_index_repair_report",
+      );
+      if (isCurrent() && !sessionIndexRepairRunning.current && isSuccessStatus(result.status)) {
+        setSessionIndexRepairReport((previous) => {
+          if ((previous?.checkedAtMs ?? 0) > (result.report?.checkedAtMs ?? 0)) return previous;
+          return result.report;
+        });
+      }
+    } catch {
+      // 下次页面刷新再读；读取报告本身不会启动修复。
+    } finally {
+      sessionIndexReportLoading.current = false;
+    }
+  };
+
+  const repairSessionIndex = async () => {
+    if (sessionIndexRepairRunning.current || providerSyncProgress.active) return;
+    sessionIndexRepairRunning.current = true;
+    setSessionIndexRepairActive(true);
+    try {
+      const result = await run(() => call<CommandResult<SessionIndexRepairReport>>("repair_session_index"));
+      if (result) {
+        if (isSuccessStatus(result.status)) {
+          setSessionIndexRepairReport(result);
+          await refreshLocalSessions(true);
+        }
+        showNotice(t("修复会话索引"), result.message, result.status);
+      }
+    } finally {
+      sessionIndexRepairRunning.current = false;
+      setSessionIndexRepairActive(false);
+    }
+  };
+
   const syncProvidersNow = async () => {
+    if (sessionIndexRepairRunning.current) return;
     if (providerSyncProgress.active) return;
     setProviderSyncProgress({
       active: true,
@@ -3048,6 +3116,18 @@ export function App() {
     };
   }, []);
 
+  useEffect(() => {
+    if (route !== "sessions") return;
+    let disposed = false;
+    const refresh = () => void refreshSessionIndexRepairReport(() => !disposed);
+    refresh();
+    const timer = window.setInterval(refresh, 15_000);
+    return () => {
+      disposed = true;
+      window.clearInterval(timer);
+    };
+  }, [route]);
+
   useEffect(() => {
     if (route !== "settings" || pendingSettingsSection !== "stepwise") return;
     let secondFrame = 0;
@@ -3337,6 +3417,7 @@ export function App() {
         }
       },
       syncProvidersNow,
+      repairSessionIndex,
       refreshProviderSyncTargets,
       setProviderSyncTarget: (provider: string) => {
         setSelectedProviderSyncTarget(provider);
@@ -3545,6 +3626,8 @@ export function App() {
               form={settingsForm}
               sessions={localSessions}
               providerSyncProgress={providerSyncProgress}
+              sessionIndexRepairActive={sessionIndexRepairActive}
+              sessionIndexRepairReport={sessionIndexRepairReport}
               providerSyncTargets={providerSyncTargets}
               selectedProviderSyncTarget={selectedProviderSyncTarget}
               onFormChange={setSettingsForm}
@@ -3762,6 +3845,7 @@ type Actions = {
   saveDreamSkinScreenshot: () => Promise<void>;
   saveManualCodexAppPath: () => Promise<void>;
   syncProvidersNow: () => Promise<void>;
+  repairSessionIndex: () => Promise<void>;
   refreshProviderSyncTargets: (silent?: boolean) => Promise<ProviderSyncTargetsResult | null>;
   setProviderSyncTarget: (provider: string) => void;
   setLaunchMode: (launchMode: LaunchMode) => Promise<void>;
@@ -6140,6 +6224,8 @@ function SessionsScreen({
   form,
   sessions,
   providerSyncProgress,
+  sessionIndexRepairActive,
+  sessionIndexRepairReport,
   providerSyncTargets,
   selectedProviderSyncTarget,
   onFormChange,
@@ -6149,6 +6235,8 @@ function SessionsScreen({
   form: BackendSettings;
   sessions: LocalSessionsResult | null;
   providerSyncProgress: ProviderSyncProgress;
+  sessionIndexRepairActive: boolean;
+  sessionIndexRepairReport: SessionIndexRepairReport | null;
   providerSyncTargets: ProviderSyncTargetsResult | null;
   selectedProviderSyncTarget: string;
   onFormChange: (value: BackendSettings) => void;
@@ -6277,7 +6365,7 @@ function SessionsScreen({
               />
               <span>
                 <strong>{t("启动前自动修复历史会话")}</strong>
-                <small>{t("启动 Codex 前整理旧对话的归属标记。")}</small>
+                <small>{t("启动前整理会话归属并检查缺失消息；运行期间自动检查索引。保存设置后生效。")}</small>
               </span>
               <ToggleVisual />
             </label>
@@ -6292,13 +6380,21 @@ function SessionsScreen({
                 {t("导入文件")}
               </Button>
               <Button
-                disabled={providerSyncProgress.active || !canRepairProviderSessions}
+                disabled={providerSyncProgress.active || sessionIndexRepairActive || !canRepairProviderSessions}
                 onClick={() => void actions.syncProvidersNow()}
                 variant="outline"
               >
                 <Wrench className="h-4 w-4" />
                 {providerSyncProgress.active ? t("正在修复…") : t("修复历史会话")}
               </Button>
+              <Button
+                disabled={sessionIndexRepairActive || providerSyncProgress.active}
+                onClick={() => void actions.repairSessionIndex()}
+                variant="outline"
+              >
+                <Wrench className="h-4 w-4" />
+                {sessionIndexRepairActive ? t("正在检查索引…") : t("修复会话索引")}
+              </Button>
               <Button onClick={() => void actions.saveSettings()}>
                 <Save className="h-4 w-4" />
                 {t("保存设置")}
@@ -6337,6 +6433,47 @@ function SessionsScreen({
             </div>
           ) : null}
 
+          {sessionIndexRepairActive ? (
+            <p role="status">{t("正在检查全部会话并恢复高可信缺失消息，首次检查可能需要较长时间…")}</p>
+          ) : null}
+          {sessionIndexRepairReport ? (
+            <div className="provider-sync-progress session-repair-progress" aria-live="polite">
+              <strong>{t("最近一次会话索引修复报告")}</strong>
+              <p>{t("最后检查：")}{sessionIndexRepairReport.checkedAtMs ? formatTime(sessionIndexRepairReport.checkedAtMs) : t("旧版报告未记录时间")}</p>
+              <p>
+                {t("读取文件")} {sessionIndexRepairReport.scannedFiles} · {t("复用缓存")} {sessionIndexRepairReport.cachedFiles} · {t("耗时")} {(sessionIndexRepairReport.elapsedMs / 1000).toFixed(1)} s
+              </p>
+              <p>
+                {t("恢复消息")} {sessionIndexRepairReport.repairedItems} · {t("已存在")} {sessionIndexRepairReport.alreadyPresent} · {t("短暂等待")} {sessionIndexRepairReport.deferredItems ?? 0} · {t("需核查")} {sessionIndexRepairReport.skippedItems}
+              </p>
+              <small>{t("仅恢复有本地原文且可确认位置的消息；已打开的会话可能需要重新打开才能显示。")}</small>
+              <p><small>{t("自动检查需要 Codex++ 启动器运行，且自动修复开关已开启并保存。此页面每 15 秒刷新报告，不会单独启动修复；再次检查不保证恢复。")}</small></p>
+              <p><small>{t("短暂等待最长 30 分钟；原文和记录文件都已超过 24 小时未更新的项目直接转入需核查。缺少对应轮次或结束状态，当前证据不足以安全补回；后续检查仍会核验。")}</small></p>
+              {sessionIndexRepairReport.backupPath ? <p className="break-all">{t("修复前备份：")}{sessionIndexRepairReport.backupPath}</p> : null}
+              {sessionIndexRepairReport.pendingDetails?.length ? (
+                <details>
+                  <summary>{t("等待与持续无法恢复详情")} ({sessionIndexRepairReport.pendingDetails.length})</summary>
+                  <ul>
+                    {sessionIndexRepairReport.pendingDetails.map((item, index) => (
+                      <li key={`${item.threadId}-${item.turnId}-${index}`} className="break-all my-3">
+                        <strong>{item.state === "waiting" ? t("短暂等待") : t("持续无法恢复")}</strong>
+                        <p>{t("任务 ID：")}{item.threadId ?? "—"} · {t("轮次 ID：")}{item.turnId ?? "—"}</p>
+                        <p>{t("原因：")}{item.reason}</p>
+                        <small>{t("首次发现：")}{formatTime(item.firstSeenAtMs)} · {t("最后检查：")}{formatTime(item.lastCheckedAtMs)} · {t("检查次数：")}{item.checks}</small>
+                      </li>
+                    ))}
+                  </ul>
+                </details>
+              ) : null}
+              {sessionIndexRepairReport.issues.length ? (
+                <details>
+                  <summary>{t("查看检查详情")} ({sessionIndexRepairReport.issues.length})</summary>
+                  <ul>{sessionIndexRepairReport.issues.map((issue, index) => <li key={index} className="break-all">{issue}</li>)}</ul>
+                </details>
+              ) : null}
+            </div>
+          ) : null}
+
           <div className="hint-line session-delete-hint">
             <Info className="h-4 w-4" />
             <span>{t("删除会创建本地备份；如果 Codex App 正在使用该会话，建议先关闭对应会话窗口再操作。")}</span>
diff --git a/apps/codex-plus-manager/src/i18n-en.ts b/apps/codex-plus-manager/src/i18n-en.ts
index 60f1037eb..453a1b80b 100644
--- a/apps/codex-plus-manager/src/i18n-en.ts
+++ b/apps/codex-plus-manager/src/i18n-en.ts
@@ -5,6 +5,20 @@
 
 // Plain strings: t("中文") -> EN_PLAIN["中文"].
 export const EN_PLAIN: Record<string, string> = {
+  "最后检查：": "Last checked: ",
+  "旧版报告未记录时间": "Timestamp unavailable in this older report",
+  "短暂等待": "Waiting for records to settle",
+  "持续无法恢复": "Still unable to restore",
+  "等待与持续无法恢复详情": "Waiting and unresolved items",
+  "任务 ID：": "Task ID: ",
+  "轮次 ID：": "Turn ID: ",
+  "原因：": "Reason: ",
+  "首次发现：": "First seen: ",
+  "检查次数：": "Checks: ",
+  "自动检查需要 Codex++ 启动器运行，且自动修复开关已开启并保存。此页面每 15 秒刷新报告，不会单独启动修复；再次检查不保证恢复。":
+    "Automatic checks require the Codex++ launcher to be running and automatic repair to be enabled and saved. This page refreshes the report every 15 seconds without starting a repair; another check does not guarantee recovery.",
+  "短暂等待最长 30 分钟；原文和记录文件都已超过 24 小时未更新的项目直接转入需核查。缺少对应轮次或结束状态，当前证据不足以安全补回；后续检查仍会核验。":
+    "Items wait for up to 30 minutes; when both the original message and its record file have been unchanged for more than 24 hours, the item goes directly to review. A missing native turn or completion state means there is insufficient evidence to safely restore it; later checks will still verify it.",
   "主导航": "Main navigation",
   "工作区": "Workspace",
   "扩展": "Extensions",
diff --git a/crates/codex-plus-data/examples/repair_session_index.rs b/crates/codex-plus-data/examples/repair_session_index.rs
new file mode 100644
index 000000000..d50febc59
--- /dev/null
+++ b/crates/codex-plus-data/examples/repair_session_index.rs
@@ -0,0 +1,11 @@
+//! 显式指定目录后执行，与管理器及启动器共用同一修复实现。
+fn main() -> anyhow::Result<()> {
+    let args: Vec<_> = std::env::args_os().skip(1).collect();
+    if args.len() != 2 || args[0] != "--home" {
+        anyhow::bail!("usage: repair_session_index --home <CODEX_HOME>");
+    }
+    let path = std::path::Path::new(&args[1]);
+    let report = codex_plus_data::repair_session_index(Some(path))?;
+    println!("{}", serde_json::to_string_pretty(&report)?);
+    Ok(())
+}
diff --git a/crates/codex-plus-data/src/lib.rs b/crates/codex-plus-data/src/lib.rs
index 903a36d2a..9435c1702 100644
--- a/crates/codex-plus-data/src/lib.rs
+++ b/crates/codex-plus-data/src/lib.rs
@@ -2,6 +2,9 @@ pub mod backup;
 pub mod markdown;
 pub mod provider_sync;
 pub mod storage;
+mod session_index_scan;
+pub mod session_index_repair;
+pub use session_index_repair::{SessionIndexRepairReport, repair_session_index, load_session_index_repair_report};
 
 pub use backup::BackupStore;
 pub use markdown::{MarkdownExportService, export_markdown_from_paths};
diff --git a/crates/codex-plus-data/src/session_index_repair.rs b/crates/codex-plus-data/src/session_index_repair.rs
new file mode 100644
index 000000000..4a4133041
--- /dev/null
+++ b/crates/codex-plus-data/src/session_index_repair.rs
@@ -0,0 +1,1624 @@
+//! 仅修复可从 rollout 交叉验证的原生用户消息投影，不发送消息或修改原始记录。
+use crate::session_index_scan::{Candidate, scan};
+use anyhow::{Context, bail};
+use fs2::FileExt;
+use rusqlite::{Connection, OpenFlags, OptionalExtension, params};
+use serde::{Deserialize, Serialize};
+use serde_json::{Value, json};
+use sha2::{Digest, Sha256};
+use std::{
+    collections::{HashMap, HashSet},
+    fs,
+    path::{Path, PathBuf},
+    time::{Duration, Instant, UNIX_EPOCH},
+};
+
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SessionIndexRepairReport {
+    pub scanned_files: usize,
+    pub cached_files: usize,
+    pub repaired_items: usize,
+    pub already_present: usize,
+    pub skipped_items: usize,
+    #[serde(default)]
+    pub deferred_items: usize,
+    #[serde(default)]
+    pub checked_at_ms: i64,
+    #[serde(default)]
+    pub pending_details: Vec<PendingDetail>,
+    pub issues: Vec<String>,
+    pub backup_path: Option<PathBuf>,
+    pub elapsed_ms: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PendingDetail {
+    pub thread_id: Option<String>,
+    pub turn_id: Option<String>,
+    pub reason: String,
+    pub state: String,
+    pub first_seen_at_ms: i64,
+    pub last_checked_at_ms: i64,
+    pub checks: i64,
+}
+
+// 不改写原生轮次状态；连续等待 30 分钟后明确告知需核查，后续仍重新核验。
+const WAIT_LIMIT_MS: i64 = 30 * 60 * 1000;
+const OLD_SOURCE_MS: i64 = 24 * 60 * 60 * 1000;
+
+fn pending_issue(
+    report: &mut SessionIndexRepairReport,
+    cache: &Connection,
+    c: Option<&Candidate>,
+    reason: &str,
+    source: &str,
+    modified_ms: i64,
+) -> anyhow::Result<()> {
+    let now = report.checked_at_ms;
+    let key = match c {
+        Some(c) => format!(
+            "{}:{}:{}:{:x}:{reason}",
+            c.thread,
+            c.turn,
+            c.ordinal,
+            Sha256::digest(c.text.as_bytes())
+        ),
+        None => format!("file:{source}:{reason}"),
+    };
+    cache.execute("INSERT INTO pending(key,first_seen,last_seen,checks) VALUES(?1,?2,?2,1) ON CONFLICT(key) DO UPDATE SET last_seen=?2,checks=checks+1", params![key,now])?;
+    let (first, checks): (i64, i64) = cache.query_row(
+        "SELECT first_seen,checks FROM pending WHERE key=?1",
+        [&key],
+        |r| Ok((r.get(0)?, r.get(1)?)),
+    )?;
+    let old_source = c.is_some_and(|c| now.saturating_sub(c.created) >= OLD_SOURCE_MS)
+        && modified_ms > 0
+        && now.saturating_sub(modified_ms) >= OLD_SOURCE_MS;
+    let blocked = old_source || now.saturating_sub(first) >= WAIT_LIMIT_MS;
+    let explanation = if blocked {
+        format!(
+            "{reason}；{}，仍会自动核验，不会猜测补建轮次或修改执行状态",
+            if old_source {
+                "原文及文件已超过 24 小时未更新"
+            } else {
+                "已连续等待至少 30 分钟"
+            }
+        )
+    } else {
+        reason.to_owned()
+    };
+    if blocked {
+        issue(
+            report,
+            match c {
+                Some(c) => format!("{} / {}：{explanation}", c.thread, c.turn),
+                None => format!("{source}：{explanation}"),
+            },
+        );
+    } else {
+        report.deferred_items += 1;
+    }
+    report.pending_details.push(PendingDetail {
+        thread_id: c.map(|c| c.thread.clone()),
+        turn_id: c.map(|c| c.turn.clone()),
+        reason: explanation,
+        state: if blocked { "blocked" } else { "waiting" }.into(),
+        first_seen_at_ms: first,
+        last_checked_at_ms: now,
+        checks,
+    });
+    Ok(())
+}
+
+fn home(path: Option<&Path>) -> PathBuf {
+    path.map(Path::to_path_buf)
+        .unwrap_or_else(codex_plus_core::relay_config::default_codex_home_dir)
+}
+
+pub fn load_session_index_repair_report(
+    path: Option<&Path>,
+) -> anyhow::Result<Option<SessionIndexRepairReport>> {
+    let path = home(path).join("session-index-repair/report.json");
+    if !path.exists() {
+        return Ok(None);
+    }
+    Ok(Some(serde_json::from_slice(&fs::read(path)?)?))
+}
+
+fn collect_files(dir: &Path, files: &mut Vec<PathBuf>) -> anyhow::Result<()> {
+    if !dir.exists() {
+        return Ok(());
+    }
+    for entry in fs::read_dir(dir)? {
+        let entry = entry?;
+        let kind = entry.file_type()?;
+        if kind.is_dir() {
+            collect_files(&entry.path(), files)?;
+        } else if kind.is_file() && entry.path().extension().is_some_and(|s| s == "jsonl") {
+            files.push(entry.path());
+        }
+    }
+    Ok(())
+}
+
+// 原生分片可能保留旧 session_meta.id；只信任目录数据库中的确切路径映射。
+fn rollout_owners(home: &Path) -> anyhow::Result<std::collections::HashMap<PathBuf, String>> {
+    let mut owners = std::collections::HashMap::new();
+    let path = home.join("state_5.sqlite");
+    if !path.is_file() {
+        return Ok(owners);
+    }
+    let db = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)?;
+    let mut stmt =
+        db.prepare("SELECT id,rollout_path FROM threads WHERE rollout_path IS NOT NULL")?;
+    let mut ambiguous = std::collections::HashSet::new();
+    for row in stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))? {
+        let (id, raw) = row?;
+        let path = PathBuf::from(raw);
+        let path = if path.is_absolute() {
+            path
+        } else {
+            home.join(path)
+        };
+        if let Ok(path) = fs::canonicalize(path) {
+            if owners.get(&path).is_some_and(|old| old != &id) {
+                ambiguous.insert(path.clone());
+            }
+            owners.insert(path, id);
+        }
+    }
+    for path in ambiguous {
+        owners.remove(&path);
+    }
+    Ok(owners)
+}
+
+fn item_text(raw: &str) -> Option<String> {
+    let item: Value = serde_json::from_str(raw).ok()?;
+    let parts = item["content"].as_array()?;
+    let mut text = String::new();
+    for part in parts {
+        if !matches!(part["type"].as_str(), Some("text" | "Text" | "input_text")) {
+            return None;
+        }
+        text.push_str(part["text"].as_str()?);
+    }
+    Some(text)
+}
+
+// 续写分片可能已不在目录表中。文件名只作约束，必须有原生完成事件的
+// turn + item ID + ordinal + 完整正文证据，并排除其他归属，才整份映射。
+fn anchored_split_owner(
+    db: &Connection,
+    path: &Path,
+    items: &[Candidate],
+    turns: &HashMap<String, HashSet<String>>,
+) -> anyhow::Result<Option<String>> {
+    let Some((_, suffix)) = path
+        .file_stem()
+        .and_then(|v| v.to_str())
+        .and_then(|v| v.rsplit_once('_'))
+    else {
+        return Ok(None);
+    };
+    if uuid::Uuid::parse_str(suffix).is_err() {
+        return Ok(None);
+    }
+    let mut anchored = false;
+    for c in items {
+        if let Some(owners) = turns.get(&c.turn) {
+            if owners.len() != 1 || !owners.contains(suffix) {
+                return Ok(None);
+            }
+            if let Some((id, ordinal)) = &c.projection {
+                let existing: Option<(String,i64)> = db.query_row("SELECT item_json,rollout_ordinal FROM thread_items WHERE thread_id=?1 AND turn_id=?2 AND item_id=?3 AND item_type='userMessage'",params![suffix,c.turn,id],|r|Ok((r.get(0)?,r.get(1)?))).optional()?;
+                if let Some((raw, n)) = existing {
+                    if n != *ordinal || item_text(&raw).as_deref() != Some(c.text.as_str()) {
+                        return Ok(None);
+                    }
+                    anchored = true;
+                }
+            }
+        }
+    }
+    Ok(anchored.then(|| suffix.to_owned()))
+}
+
+fn resolve_owner(
+    db: &Connection,
+    path: &Path,
+    items: &[Candidate],
+    directory_owner: Option<String>,
+    turns: &HashMap<String, HashSet<String>>,
+) -> anyhow::Result<Option<String>> {
+    let owner = anchored_split_owner(db, path, items, turns)?.or(directory_owner);
+    if let Some(owner) = &owner {
+        if items
+            .iter()
+            .any(|c| turns.get(&c.turn).is_some_and(|ids| !ids.contains(owner)))
+        {
+            bail!("目录归属与原生轮次身份冲突，未写入");
+        }
+    }
+    Ok(owner)
+}
+
+fn issue(report: &mut SessionIndexRepairReport, message: String) {
+    report.skipped_items += 1;
+    // 报告不保存消息正文，避免 UI 加载大报告或泄露原文。
+    if report.issues.len() < 200 {
+        report.issues.push(message);
+    }
+}
+
+enum Action {
+    Present,
+    Insert(String),
+    Link(String),
+    Prepend(String, String),
+    RestoreTurn,
+    Wait(&'static str),
+    Skip(&'static str),
+}
+
+fn recovery_id(c: &Candidate) -> String {
+    if c.goal {
+        if let Some(key) = &c.goal_key {
+            return format!(
+                "recovered-goal-{:x}",
+                Sha256::digest(
+                    serde_json::to_vec(&(&c.thread, key, &c.text))
+                        .expect("serializable goal identity")
+                )
+            );
+        }
+    }
+    format!("recovered-user-{}-{}", c.turn, c.ordinal)
+}
+
+fn is_original_first(c: &Candidate) -> bool {
+    if !c.first_user_evidence_valid {
+        return false;
+    }
+    if c.goal {
+        c.first_user_response_ordinal.is_none_or(|n| c.ordinal < n)
+            && c.first_user_projection
+                .as_ref()
+                .is_none_or(|(_, n)| c.ordinal < *n)
+    } else {
+        c.first_user_response_ordinal == Some(c.ordinal)
+            && c.first_user_projection
+                .as_ref()
+                .is_none_or(|p| c.projection.as_ref() == Some(p))
+    }
+}
+
+fn deduplicate_goals(
+    db: &Connection,
+    mut candidates: Vec<Candidate>,
+) -> anyhow::Result<Vec<Candidate>> {
+    candidates.sort_by_key(|c| (c.created, c.ordinal));
+    let mut goals = HashMap::new();
+    let mut unique: Vec<Candidate> = Vec::new();
+    for c in candidates {
+        if let Some(key) = c.goal_key.as_ref().filter(|_| c.goal) {
+            let identity = (c.thread.clone(), key.clone(), c.text.clone());
+            if let Some(&index) = goals.get(&identity) {
+                // 旧版恢复ID不含goal身份。先用每个分片自己的轮次/位置核验，
+                // 保留已经恢复的那个分片，不能因更早分片后来出现而再次插入。
+                if matches!(inspect(db, &c)?, Action::Present | Action::Link(_))
+                    && !matches!(
+                        inspect(db, &unique[index])?,
+                        Action::Present | Action::Link(_)
+                    )
+                {
+                    unique[index] = c;
+                }
+                continue;
+            }
+            goals.insert(identity, unique.len());
+        }
+        unique.push(c);
+    }
+    Ok(unique)
+}
+
+fn inspect(db: &Connection, c: &Candidate) -> anyhow::Result<Action> {
+    let position = c
+        .projection
+        .as_ref()
+        .map_or(c.ordinal, |(_, ordinal)| *ordinal);
+    if c.goal && c.goal_key.is_some() {
+        let exists: bool = db.query_row(
+            "SELECT EXISTS(SELECT 1 FROM thread_items i JOIN thread_turns t ON t.thread_id=i.thread_id AND t.turn_id=i.turn_id AND t.first_user_item_id=i.item_id WHERE i.thread_id=?1 AND i.item_id=?2 AND i.item_type='userMessage' AND json_valid(i.item_json) AND json_extract(i.item_json,'$.content[0].text')=?3)",
+            params![c.thread,recovery_id(c),c.text], |r| r.get(0))?;
+        if exists {
+            return Ok(Action::Present);
+        }
+    }
+    let turn: Option<(Option<String>, String)> = db
+        .query_row(
+            "SELECT first_user_item_id,status FROM thread_turns WHERE thread_id=?1 AND turn_id=?2",
+            params![c.thread, c.turn],
+            |r| Ok((r.get(0)?, r.get(1)?)),
+        )
+        .optional()?;
+    let Some((first, status)) = turn else {
+        return Ok(Action::Wait("原生数据库中尚无对应轮次"));
+    };
+    let mut finished = matches!(status.as_str(), "completed" | "interrupted" | "failed");
+    if !finished && status == "inProgress" {
+        if let Some((next, ordinal)) = &c.following_turn {
+            // 旧投影状态可能没结束。只有原始日志后继边界与原生后继轮次一致时
+            // 才允许补用户消息；不改写原生 status，也不把时间长当作结束证据。
+            if next != &c.turn && *ordinal > position {
+                finished = db.query_row("SELECT EXISTS(SELECT 1 FROM thread_turns WHERE thread_id=?1 AND turn_id=?2 AND rollout_ordinal=?3) AND EXISTS(SELECT 1 FROM thread_turns WHERE thread_id=?1 AND turn_id=?4 AND rollout_ordinal=?5)",params![c.thread,next,ordinal,c.turn,c.start_ordinal],|r|r.get(0))?;
+            }
+        }
+    }
+    let mut stmt = db.prepare_cached("SELECT item_id,item_json,rollout_ordinal FROM thread_items WHERE thread_id=?1 AND turn_id=?2 AND item_type='userMessage'")?;
+    let rows = stmt
+        .query_map(params![c.thread, c.turn], |r| {
+            Ok((
+                r.get::<_, String>(0)?,
+                r.get::<_, String>(1)?,
+                r.get::<_, i64>(2)?,
+            ))
+        })?
+        .collect::<rusqlite::Result<Vec<_>>>()?;
+    let valid_first = rows.iter().find(|(id, _, _)| first.as_ref() == Some(id));
+    for (id, raw, ordinal) in &rows {
+        let text = item_text(raw);
+        let same_source = c
+            .projection
+            .as_ref()
+            .map_or(*ordinal == c.ordinal, |(source, n)| {
+                source == id && *n == *ordinal
+            })
+            || (id.starts_with("recovered-") && *ordinal == c.ordinal);
+        if same_source && text.as_deref() == Some(c.text.as_str()) {
+            let conflict: bool = db.query_row(
+                "SELECT EXISTS(SELECT 1 FROM thread_items WHERE thread_id=?1 AND rollout_ordinal=?3 AND NOT(turn_id=?2 AND item_id=?4))",
+                params![c.thread, c.turn, position, id], |r| r.get(0))?;
+            if conflict {
+                return Ok(Action::Skip("相同记录位置已有不同内容"));
+            }
+            return Ok(if first.as_ref() == Some(id) {
+                Action::Present
+            } else if first.is_none() && rows.len() == 1 && is_original_first(c) {
+                if finished {
+                    Action::Link(id.clone())
+                } else {
+                    Action::Wait("原生轮次仍在执行或状态未知，暂不修改首条消息指针")
+                }
+            } else if !c.goal && valid_first.is_some_and(|(_, _, n)| *n < *ordinal) {
+                Action::Present
+            } else {
+                Action::Skip("已有用户消息顺序存在冲突")
+            });
+        }
+    }
+    // 已存在消息只读核验不受轮次状态影响；只有真正需要写入时才等待结束。
+    if !finished {
+        return Ok(Action::Wait("原生轮次仍在执行或状态未知，暂不补入消息"));
+    }
+    let occupied: bool = db.query_row(
+        "SELECT EXISTS(SELECT 1 FROM thread_items WHERE thread_id=?1 AND (rollout_ordinal=?3 OR (turn_id=?2 AND item_id=?4)))",
+        params![c.thread, c.turn, position, c.projection.as_ref().map(|(id, _)| id)], |r| r.get(0))?;
+    if occupied {
+        return Ok(Action::Skip("相同记录位置已有不同内容"));
+    }
+    if !rows.is_empty() {
+        // 目标有目标事件与内部 objective 双重证据，位置又早于全部用户消息。
+        // 补回目标并调整首条指针，完整保留其后的选项回复和 steering。
+        if c.goal && is_original_first(c) && rows.iter().all(|(_, _, n)| *n > position) {
+            if let Some((first_id, _, first_ordinal)) = valid_first {
+                if rows.iter().all(|(_, _, n)| n >= first_ordinal) {
+                    return Ok(Action::Prepend(recovery_id(c), first_id.clone()));
+                }
+            }
+        }
+        // steering 是同轮次的后续用户消息；只在首条指针有效且顺序明确时补入。
+        if c.goal || c.projection.is_none() || !valid_first.is_some_and(|(_, _, n)| *n < position) {
+            return Ok(Action::Skip("该轮次用户消息顺序无法可靠确认"));
+        }
+        return Ok(Action::Insert(c.projection.as_ref().unwrap().0.clone()));
+    }
+    if !is_original_first(c) {
+        // 同一批次可能先扫描到后续 steering，再扫描到首条消息；暂缓到写入阶段
+        // 重新核验，避免把后续内容提升为首条，同时保留首条出现后可恢复的机会。
+        return Ok(Action::Wait("等待同轮次已确认的首条用户消息"));
+    }
+    if let Some(id) = first {
+        let exists: bool = db.query_row("SELECT EXISTS(SELECT 1 FROM thread_items WHERE thread_id=?1 AND turn_id=?2 AND item_id=?3)",params![c.thread,c.turn,id],|r|r.get(0))?;
+        return Ok(if exists {
+            Action::Skip("首条消息指针指向不同类型的记录")
+        } else if c
+            .projection
+            .as_ref()
+            .is_some_and(|(native, _)| native == &id)
+        {
+            Action::Insert(id)
+        } else {
+            Action::Skip("悬空首条指针与候选消息身份不一致")
+        });
+    }
+    Ok(Action::Insert(
+        c.projection
+            .as_ref()
+            .map(|(id, _)| id.clone())
+            .unwrap_or_else(|| recovery_id(c)),
+    ))
+}
+
+fn apply(db: &Connection, c: &Candidate, action: Action) -> anyhow::Result<bool> {
+    if matches!(action, Action::RestoreTurn) {
+        let boundary = c.turn_boundary.as_ref().context("缺少完整轮次边界")?;
+        db.execute("INSERT INTO thread_turns(thread_id,turn_id,rollout_ordinal,status,error_json,started_at,completed_at,duration_ms,rollout_end_ordinal) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9)",params![c.thread,c.turn,boundary.start_ordinal,boundary.status,boundary.error_json,boundary.started_at,boundary.completed_at,boundary.duration_ms,boundary.end_ordinal])?;
+        return apply(db, c, inspect(db, c)?);
+    }
+    let previous_first = match &action {
+        Action::Prepend(_, old) => Some(old.clone()),
+        _ => None,
+    };
+    let id = match action {
+        Action::Insert(id) | Action::Prepend(id, _) => {
+            let ordinal = c
+                .projection
+                .as_ref()
+                .map_or(c.ordinal, |(_, ordinal)| *ordinal);
+            let item = json!({"type":"userMessage","id":id,"content":[{"type":"text","text":c.text,"text_elements":[]}]}).to_string();
+            db.execute("INSERT INTO thread_items(thread_id,turn_id,item_id,rollout_ordinal,created_at_ms,item_json,item_type,updated_at_ordinal) VALUES(?1,?2,?3,?4,?5,?6,'userMessage',?4)",params![c.thread,c.turn,id,ordinal,c.created,item])?;
+            id
+        }
+        Action::Link(id) => id,
+        _ => return Ok(false),
+    };
+    if let Some(previous_first) = previous_first {
+        let changed = db.execute("UPDATE thread_turns SET first_user_item_id=?3 WHERE thread_id=?1 AND turn_id=?2 AND first_user_item_id=?4", params![c.thread,c.turn,id,previous_first])?;
+        if changed != 1 {
+            bail!("首条消息指针发生变化，取消本次事务");
+        }
+        return Ok(true);
+    }
+    db.execute("UPDATE thread_turns SET first_user_item_id=?3 WHERE thread_id=?1 AND turn_id=?2 AND first_user_item_id IS NULL",params![c.thread,c.turn,id])?;
+    Ok(true)
+}
+
+fn inspect_with_source(
+    db: &Connection,
+    c: &Candidate,
+    verified_source: bool,
+) -> anyhow::Result<Action> {
+    let action = inspect(db, c)?;
+    if matches!(action, Action::Wait("原生数据库中尚无对应轮次")) && verified_source {
+        if let Some(b) = &c.turn_boundary {
+            let position = c.projection.as_ref().map_or(c.ordinal, |(_, n)| *n);
+            let original_first = if c.goal {
+                b.first_user_response_ordinal.is_none_or(|n| c.ordinal < n)
+                    && b.first_user_projection
+                        .as_ref()
+                        .is_none_or(|(_, n)| position < *n)
+            } else {
+                c.projection.is_some()
+                    && c.projection == b.first_user_projection
+                    && b.first_user_response_ordinal == Some(c.ordinal)
+            };
+            if b.start_ordinal < c.ordinal
+                && c.ordinal <= position
+                && position < b.end_ordinal
+                && original_first
+                && b.started_at <= b.completed_at
+                && matches!(b.status.as_str(), "completed" | "failed" | "interrupted")
+            {
+                let conflicting: bool = db.query_row("SELECT EXISTS(SELECT 1 FROM thread_turns WHERE turn_id=?2 OR (thread_id=?1 AND rollout_ordinal=?3)) OR EXISTS(SELECT 1 FROM thread_items WHERE thread_id=?1 AND (turn_id=?2 OR rollout_ordinal=?4 OR item_id=?5))",params![c.thread,c.turn,b.start_ordinal,position,c.projection.as_ref().map(|(id,_)|id)],|r|r.get(0))?;
+                if !conflicting {
+                    return Ok(Action::RestoreTurn);
+                }
+            }
+        }
+    }
+    Ok(action)
+}
+
+pub fn repair_session_index(path: Option<&Path>) -> anyhow::Result<SessionIndexRepairReport> {
+    let started = Instant::now();
+    let home = home(path);
+    let work = home.join("session-index-repair");
+    fs::create_dir_all(&work)?;
+    let lock = fs::OpenOptions::new()
+        .create(true)
+        .truncate(false)
+        .read(true)
+        .write(true)
+        .open(work.join("repair.lock"))?;
+    lock.try_lock_exclusive()
+        .context("会话索引修复正在运行，请稍后查看报告")?;
+    let _lifecycle = crate::try_acquire_provider_sync_lifecycle_guard(Some(&home))?;
+    let mut report = SessionIndexRepairReport {
+        checked_at_ms: chrono::Utc::now().timestamp_millis(),
+        ..Default::default()
+    };
+    let db_path = home.join("thread_history_1.sqlite");
+    if !db_path.is_file() {
+        bail!("未找到原生会话历史数据库，尚不能修复消息索引");
+    }
+    let mut db = Connection::open_with_flags(db_path, OpenFlags::SQLITE_OPEN_READ_WRITE)?;
+    db.busy_timeout(Duration::from_secs(5))?;
+    // 未知数据库版本直接失败，不创建或改造原生 schema。
+    db.prepare("SELECT thread_id,turn_id,first_user_item_id,status FROM thread_turns LIMIT 0")?;
+    db.prepare("SELECT thread_id,turn_id,item_id,rollout_ordinal,created_at_ms,item_json,item_type,updated_at_ordinal FROM thread_items LIMIT 0")?;
+    let mut cache = Connection::open(work.join("scan-cache.sqlite"))?;
+    cache.execute_batch("CREATE TABLE IF NOT EXISTS files(path TEXT PRIMARY KEY,stamp TEXT NOT NULL,candidates TEXT NOT NULL);")?;
+    cache.execute_batch("CREATE TABLE IF NOT EXISTS pending(key TEXT PRIMARY KEY,first_seen INTEGER NOT NULL,last_seen INTEGER NOT NULL,checks INTEGER NOT NULL);")?;
+    let mut files = Vec::new();
+    for dir in ["sessions", "archived_sessions"] {
+        collect_files(&home.join(dir), &mut files)?;
+    }
+    files.sort();
+    let mut candidates = Vec::new();
+    let owners = rollout_owners(&home)?;
+    let mut turn_owners: HashMap<String, HashSet<String>> = HashMap::new();
+    {
+        let mut stmt = db.prepare("SELECT thread_id,turn_id FROM thread_turns")?;
+        for row in stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))? {
+            let (thread, turn) = row?;
+            turn_owners.entry(turn).or_default().insert(thread);
+        }
+    }
+    let mut source_times: HashMap<(String, String, i64), i64> = HashMap::new();
+    let mut verified_sources = HashSet::new();
+    let cache_tx = cache.transaction()?;
+    for path in files {
+        report.scanned_files += 1;
+        let result: anyhow::Result<Vec<Candidate>> = (|| {
+            let meta = fs::metadata(&path)?;
+            let stamp = format!(
+                "v7:{}:{}",
+                meta.len(),
+                meta.modified()?.duration_since(UNIX_EPOCH)?.as_nanos()
+            );
+            let key = path.to_string_lossy();
+            let cached: Option<String> = cache_tx
+                .query_row(
+                    "SELECT candidates FROM files WHERE path=?1 AND stamp=?2",
+                    params![key, stamp],
+                    |r| r.get(0),
+                )
+                .optional()?;
+            if let Some(raw) = cached
+                && let Ok(items) = serde_json::from_str(&raw)
+            {
+                report.cached_files += 1;
+                return Ok(items);
+            }
+            let items = scan(&path)?;
+            let after = fs::metadata(&path)?;
+            if after.len() == meta.len() && after.modified()? == meta.modified()? {
+                cache_tx.execute(
+                    "INSERT OR REPLACE INTO files VALUES(?1,?2,?3)",
+                    params![key, stamp, serde_json::to_string(&items)?],
+                )?;
+            } else {
+                bail!("文件扫描期间仍在写入，将在下次检查");
+            }
+            Ok(items)
+        })();
+        match result {
+            Ok(mut items) => {
+                let owner = fs::canonicalize(&path)
+                    .ok()
+                    .and_then(|p| owners.get(&p).cloned());
+                // 目录表可能仅指向当前分片而保留父任务 ID，原生完成事件证据优先。
+                // 如证据仍指向别的 owner，整文件保留待核查，禁止重复补到目录 owner。
+                let owner = match resolve_owner(&db, &path, &items, owner, &turn_owners) {
+                    Ok(owner) => owner,
+                    Err(error) => {
+                        issue(&mut report, format!("{}：{error}", path.display()));
+                        continue;
+                    }
+                };
+                let modified = fs::metadata(&path)?
+                    .modified()?
+                    .duration_since(UNIX_EPOCH)?
+                    .as_millis() as i64;
+                for item in &mut items {
+                    if let Some(owner) = &owner {
+                        item.thread.clone_from(owner);
+                        verified_sources.insert((
+                            item.thread.clone(),
+                            item.turn.clone(),
+                            item.ordinal,
+                        ));
+                    }
+                    source_times
+                        .entry((item.thread.clone(), item.turn.clone(), item.ordinal))
+                        .and_modify(|t| *t = (*t).max(modified))
+                        .or_insert(modified);
+                }
+                candidates.extend(items);
+            }
+            Err(error)
+                if error.to_string().contains("文件扫描期间仍在写入")
+                    || error.to_string().contains("会话记录尾行不完整") =>
+            {
+                let modified = fs::metadata(&path)
+                    .ok()
+                    .and_then(|m| m.modified().ok())
+                    .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
+                    .map_or(0, |d| d.as_millis() as i64);
+                let reason = if error.to_string().contains("会话记录尾行不完整") {
+                    "会话记录尾行不完整"
+                } else {
+                    "文件扫描期间仍在写入"
+                };
+                if reason == "会话记录尾行不完整"
+                    && modified > 0
+                    && report.checked_at_ms.saturating_sub(modified) >= OLD_SOURCE_MS
+                {
+                    issue(
+                        &mut report,
+                        format!(
+                            "{}：尾行截断且超过24小时未更新，需核查原始日志",
+                            path.display()
+                        ),
+                    );
+                    continue;
+                }
+                pending_issue(
+                    &mut report,
+                    &cache_tx,
+                    None,
+                    reason,
+                    &path.to_string_lossy(),
+                    modified,
+                )?
+            }
+            Err(error) => issue(&mut report, format!("{}：{}", path.display(), error)),
+        }
+    }
+    cache_tx.commit()?;
+    // 同目标优先保留已有原生投影；尚未恢复时才选择最早分片。
+    candidates = deduplicate_goals(&db, candidates)?;
+    candidates
+        .sort_by(|a, b| (&a.thread, &a.turn, a.ordinal).cmp(&(&b.thread, &b.turn, b.ordinal)));
+    candidates.dedup_by(|a, b| {
+        a.thread == b.thread && a.turn == b.turn && a.ordinal == b.ordinal && a.text == b.text
+    });
+    let mut conflicts = std::collections::HashSet::new();
+    for pair in candidates.windows(2) {
+        if pair[0].thread == pair[1].thread
+            && pair[0].turn == pair[1].turn
+            && pair[0].ordinal == pair[1].ordinal
+            && pair[0].text != pair[1].text
+        {
+            conflicts.insert((
+                pair[0].thread.clone(),
+                pair[0].turn.clone(),
+                pair[0].ordinal,
+            ));
+        }
+    }
+    let mut pending = Vec::new();
+    let mut waiting = Vec::new();
+    for c in candidates {
+        if conflicts.contains(&(c.thread.clone(), c.turn.clone(), c.ordinal)) {
+            issue(
+                &mut report,
+                format!(
+                    "{} / {}：相同记录位置有多个候选原文，需人工核对",
+                    c.thread, c.turn
+                ),
+            );
+            continue;
+        }
+        match inspect_with_source(
+            &db,
+            &c,
+            verified_sources.contains(&(c.thread.clone(), c.turn.clone(), c.ordinal)),
+        )? {
+            Action::Present => report.already_present += 1,
+            Action::Skip(reason) => {
+                issue(&mut report, format!("{} / {}：{reason}", c.thread, c.turn))
+            }
+            Action::Wait(_) => waiting.push(c),
+            _ => pending.push(c),
+        }
+    }
+    if !pending.is_empty() {
+        pending.append(&mut waiting);
+        pending
+            .sort_by(|a, b| (&a.thread, &a.turn, a.ordinal).cmp(&(&b.thread, &b.turn, b.ordinal)));
+        let backup = work.join(format!("before-repair-{}.sqlite", uuid::Uuid::new_v4()));
+        db.execute("VACUUM INTO ?1", [backup.to_string_lossy().as_ref()])?;
+        report.backup_path = Some(backup);
+        let tx = db.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        for c in pending {
+            // 获取写锁后重新核验，避免扫描与写入之间原生进程已补齐消息。
+            let action = inspect_with_source(
+                &tx,
+                &c,
+                verified_sources.contains(&(c.thread.clone(), c.turn.clone(), c.ordinal)),
+            )?;
+            match action {
+                Action::Present => report.already_present += 1,
+                Action::Skip(reason) => {
+                    issue(&mut report, format!("{} / {}：{reason}", c.thread, c.turn))
+                }
+                Action::Wait(reason) => pending_issue(
+                    &mut report,
+                    &cache,
+                    Some(&c),
+                    reason,
+                    "",
+                    *source_times
+                        .get(&(c.thread.clone(), c.turn.clone(), c.ordinal))
+                        .unwrap_or(&0),
+                )?,
+                action => {
+                    if apply(&tx, &c, action)? {
+                        report.repaired_items += 1;
+                    }
+                }
+            }
+        }
+        tx.commit()?;
+    }
+    for c in waiting {
+        match inspect_with_source(
+            &db,
+            &c,
+            verified_sources.contains(&(c.thread.clone(), c.turn.clone(), c.ordinal)),
+        )? {
+            Action::Present => report.already_present += 1,
+            Action::Skip(reason) => {
+                issue(&mut report, format!("{} / {}：{reason}", c.thread, c.turn))
+            }
+            action => {
+                let reason = if let Action::Wait(reason) = action {
+                    reason
+                } else {
+                    "原生状态刚发生变化，将在下次检查重新核验"
+                };
+                pending_issue(
+                    &mut report,
+                    &cache,
+                    Some(&c),
+                    reason,
+                    "",
+                    *source_times
+                        .get(&(c.thread.clone(), c.turn.clone(), c.ordinal))
+                        .unwrap_or(&0),
+                )?;
+            }
+        }
+    }
+    report.elapsed_ms = started.elapsed().as_millis() as u64;
+    // 不再待处理的候选删除观察状态；以后真正再次缺失时重新计时。
+    cache.execute(
+        "DELETE FROM pending WHERE last_seen<>?1",
+        [report.checked_at_ms],
+    )?;
+    let report_temp = work.join("report.tmp");
+    fs::write(&report_temp, serde_json::to_vec_pretty(&report)?)?;
+    fs::rename(report_temp, work.join("report.json"))?;
+    Ok(report)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn session_index_uses_catalog_path_for_split_thread_identity() {
+        let home = home_with_rollout();
+        let native = Connection::open(home.path().join("thread_history_1.sqlite")).unwrap();
+        native
+            .execute("UPDATE thread_turns SET thread_id='split'", [])
+            .unwrap();
+        let state = Connection::open(home.path().join("state_5.sqlite")).unwrap();
+        state
+            .execute_batch("CREATE TABLE threads(id TEXT,rollout_path TEXT)")
+            .unwrap();
+        state
+            .execute(
+                "INSERT INTO threads VALUES('split',?1)",
+                [home
+                    .path()
+                    .join("sessions/rollout.jsonl")
+                    .to_string_lossy()
+                    .as_ref()],
+            )
+            .unwrap();
+        let report = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!(report.repaired_items, 1);
+        let owner: String = native
+            .query_row("SELECT thread_id FROM thread_items", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(owner, "split");
+        assert_eq!(
+            repair_session_index(Some(home.path()))
+                .unwrap()
+                .already_present,
+            1
+        );
+    }
+
+    #[test]
+    fn session_index_waits_are_tracked_and_eventually_actionable() {
+        let cache = Connection::open_in_memory().unwrap();
+        cache.execute_batch("CREATE TABLE pending(key TEXT PRIMARY KEY,first_seen INTEGER,last_seen INTEGER,checks INTEGER)").unwrap();
+        let c = candidate();
+        let mut first = SessionIndexRepairReport {
+            checked_at_ms: 200_000,
+            ..Default::default()
+        };
+        pending_issue(&mut first, &cache, Some(&c), "没有原生轮次", "", 200_000).unwrap();
+        assert_eq!((first.deferred_items, first.skipped_items), (1, 0));
+        let mut later = SessionIndexRepairReport {
+            checked_at_ms: 200_000 + WAIT_LIMIT_MS,
+            ..Default::default()
+        };
+        let later_time = later.checked_at_ms;
+        pending_issue(&mut later, &cache, Some(&c), "没有原生轮次", "", later_time).unwrap();
+        assert_eq!((later.deferred_items, later.skipped_items), (0, 1));
+        assert_eq!(later.pending_details[0].checks, 2);
+        assert_eq!(later.pending_details[0].first_seen_at_ms, 200_000);
+        let mut old = SessionIndexRepairReport {
+            checked_at_ms: 200_000 + OLD_SOURCE_MS,
+            ..Default::default()
+        };
+        pending_issue(&mut old, &cache, Some(&c), "原生轮次仍在执行", "", 100_000).unwrap();
+        assert_eq!((old.deferred_items, old.skipped_items), (0, 1));
+    }
+
+    #[test]
+    fn session_index_goal_dedup_preserves_old_repair_but_relinks_orphans() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        let mut c = candidate();
+        c.goal = true;
+        c.goal_key = Some("goal-identity".into());
+        c.first_user_response_ordinal = None;
+        apply(&db, &c, inspect(&db, &c).unwrap()).unwrap();
+        db.execute("UPDATE thread_turns SET first_user_item_id=NULL", [])
+            .unwrap();
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Link(_)));
+        apply(&db, &c, inspect(&db, &c).unwrap()).unwrap();
+        c.turn = "continuation".into();
+        db.execute(
+            "INSERT INTO thread_turns VALUES('thread','continuation','completed',NULL)",
+            [],
+        )
+        .unwrap();
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Present));
+        assert_eq!(counts(&db).0, 1);
+    }
+
+    #[test]
+    fn session_index_existing_active_messages_are_not_deferred() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        let first = candidate();
+        apply(&db, &first, inspect(&db, &first).unwrap()).unwrap();
+        let mut later = first.clone();
+        later.ordinal = 8;
+        later.projection = Some(("steering".into(), 9));
+        later.text = "补充".into();
+        apply(&db, &later, inspect(&db, &later).unwrap()).unwrap();
+        db.execute("UPDATE thread_turns SET status='inProgress'", [])
+            .unwrap();
+        for c in [&first, &later] {
+            assert!(matches!(inspect(&db, c).unwrap(), Action::Present));
+        }
+        later.text = "缺失".into();
+        assert!(matches!(inspect(&db, &later).unwrap(), Action::Wait(_)));
+    }
+
+    #[test]
+    fn session_index_goal_prepend_preserves_question_reply_and_steering() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        let mut reply = candidate();
+        reply.ordinal = 10;
+        reply.first_user_response_ordinal = Some(10);
+        reply.text="<send_user_message_question_reply>model question + user answer</send_user_message_question_reply>".into();
+        apply(&db, &reply, inspect(&db, &reply).unwrap()).unwrap();
+        let before: String = db
+            .query_row("SELECT item_json FROM thread_items", [], |r| r.get(0))
+            .unwrap();
+        let mut goal = candidate();
+        goal.goal = true;
+        goal.first_user_response_ordinal = None;
+        goal.ordinal = 3;
+        assert!(matches!(
+            inspect(&db, &goal).unwrap(),
+            Action::Prepend(_, _)
+        ));
+        apply(&db, &goal, inspect(&db, &goal).unwrap()).unwrap();
+        let after: String = db
+            .query_row(
+                "SELECT item_json FROM thread_items WHERE rollout_ordinal=10",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(before, after);
+        assert_eq!(counts(&db), (2, Some("recovered-user-turn-3".into())));
+        assert!(matches!(inspect(&db, &goal).unwrap(), Action::Present));
+        assert!(matches!(inspect(&db, &reply).unwrap(), Action::Present));
+    }
+
+    #[test]
+    fn session_index_split_mapping_requires_strict_anchor_and_no_opposing_owner() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        let owner = "11111111-1111-4111-8111-111111111111";
+        let mut anchor = candidate();
+        anchor.thread = owner.into();
+        anchor.projection = Some(("native-id".into(), 4));
+        db.execute("UPDATE thread_turns SET thread_id=?1", [owner])
+            .unwrap();
+        apply(&db, &anchor, inspect(&db, &anchor).unwrap()).unwrap();
+        anchor.thread = "legacy".into();
+        let path = PathBuf::from(format!("rollout-legacy_{owner}.jsonl"));
+        let mut owners = HashMap::from([("turn".into(), HashSet::from([owner.into()]))]);
+        assert_eq!(
+            anchored_split_owner(&db, &path, &[anchor.clone()], &owners)
+                .unwrap()
+                .as_deref(),
+            Some(owner)
+        );
+        assert_eq!(
+            resolve_owner(
+                &db,
+                &path,
+                &[anchor.clone()],
+                Some("legacy".into()),
+                &owners
+            )
+            .unwrap()
+            .as_deref(),
+            Some(owner)
+        );
+        anchor.text = "相同ID但不同原文".into();
+        assert!(
+            anchored_split_owner(&db, &path, &[anchor.clone()], &owners)
+                .unwrap()
+                .is_none()
+        );
+        anchor.text = candidate().text;
+        owners.get_mut("turn").unwrap().insert("other".into());
+        assert!(
+            anchored_split_owner(&db, &path, &[anchor], &owners)
+                .unwrap()
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn session_index_old_active_goal_needs_native_following_boundary() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        db.execute_batch("ALTER TABLE thread_turns ADD COLUMN rollout_ordinal INTEGER; UPDATE thread_turns SET status='inProgress'; INSERT INTO thread_turns(thread_id,turn_id,status,rollout_ordinal) VALUES('thread','next','completed',20)").unwrap();
+        db.execute(
+            "UPDATE thread_turns SET rollout_ordinal=1 WHERE turn_id='turn'",
+            [],
+        )
+        .unwrap();
+        let mut c = candidate();
+        c.goal = true;
+        c.following_turn = Some(("next".into(), 20));
+        c.first_user_response_ordinal = None;
+        c.start_ordinal = Some(1);
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Insert(_)));
+        apply(&db, &c, inspect(&db, &c).unwrap()).unwrap();
+        let status: String = db
+            .query_row(
+                "SELECT status FROM thread_turns WHERE turn_id='turn'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(status, "inProgress");
+        c.turn = "absent".into();
+        c.text = "另一个目标".into();
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Wait(_)));
+    }
+
+    #[test]
+    fn session_index_rebuilds_only_verified_missing_turn_with_complete_evidence() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        db.execute_batch("ALTER TABLE thread_turns ADD COLUMN rollout_ordinal INTEGER; ALTER TABLE thread_turns ADD COLUMN error_json TEXT; ALTER TABLE thread_turns ADD COLUMN started_at INTEGER; ALTER TABLE thread_turns ADD COLUMN completed_at INTEGER; ALTER TABLE thread_turns ADD COLUMN duration_ms INTEGER; ALTER TABLE thread_turns ADD COLUMN rollout_end_ordinal INTEGER;").unwrap();
+        let mut c = candidate();
+        c.turn = "missing".into();
+        c.projection = Some(("native-id".into(), 4));
+        c.turn_boundary = Some(crate::session_index_scan::TurnBoundary {
+            start_ordinal: 1,
+            started_at: 100,
+            end_ordinal: 10,
+            completed_at: 105,
+            status: "failed".into(),
+            error_json: Some("{\"message\":\"original error\"}".into()),
+            duration_ms: Some(5123),
+            first_user_projection: Some(("native-id".into(), 4)),
+            first_user_response_ordinal: Some(3),
+        });
+        assert!(matches!(
+            inspect_with_source(&db, &c, false).unwrap(),
+            Action::Wait(_)
+        ));
+        c.turn_boundary
+            .as_mut()
+            .unwrap()
+            .first_user_response_ordinal = Some(2);
+        assert!(matches!(
+            inspect_with_source(&db, &c, true).unwrap(),
+            Action::Wait(_)
+        ));
+        c.turn_boundary
+            .as_mut()
+            .unwrap()
+            .first_user_response_ordinal = Some(3);
+        assert!(matches!(
+            inspect_with_source(&db, &c, true).unwrap(),
+            Action::RestoreTurn
+        ));
+        let tx = db.unchecked_transaction().unwrap();
+        apply(&tx, &c, inspect_with_source(&tx, &c, true).unwrap()).unwrap();
+        tx.commit().unwrap();
+        let actual:(String,i64,i64,String)=db.query_row("SELECT status,started_at,completed_at,first_user_item_id FROM thread_turns WHERE turn_id='missing'",[],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?))).unwrap();
+        assert_eq!(actual, ("failed".into(), 100, 105, "native-id".into()));
+        assert!(matches!(
+            inspect_with_source(&db, &c, true).unwrap(),
+            Action::Present
+        ));
+        let mut wrong_owner = c.clone();
+        wrong_owner.thread = "other".into();
+        assert!(matches!(
+            inspect_with_source(&db, &wrong_owner, true).unwrap(),
+            Action::Wait(_)
+        ));
+        c.turn = "another".into();
+        assert!(matches!(
+            inspect_with_source(&db, &c, true).unwrap(),
+            Action::Wait(_)
+        ));
+    }
+
+    #[test]
+    fn session_index_scans_archives_and_deduplicates_goals_across_files() {
+        let home = home_with_rollout();
+        fs::create_dir(home.path().join("archived_sessions")).unwrap();
+        let db = Connection::open(home.path().join("thread_history_1.sqlite")).unwrap();
+        for (turn, created) in [("goal-a", 100), ("goal-b", 100)] {
+            db.execute(
+                "INSERT INTO thread_turns VALUES('thread',?1,'completed',NULL)",
+                [turn],
+            )
+            .unwrap();
+            let rows = [
+                json!({"type":"session_meta","payload":{"id":"thread"}}),
+                json!({"type":"event_msg","payload":{"type":"thread_goal_updated","threadId":"thread","goal":{"objective":"目标全文","createdAt":created}}}),
+                json!({"type":"event_msg","ordinal":1,"payload":{"type":"task_started","turn_id":turn}}),
+                json!({"type":"response_item","ordinal":4,"payload":{"role":"user","content":[{"type":"input_text","text":"<codex_internal_context source=\"goal\"><objective>\n目标全文\n</objective></codex_internal_context>"}],"internal_chat_message_metadata_passthrough":{"turn_id":turn}}}),
+            ];
+            fs::write(
+                home.path().join(format!("archived_sessions/{turn}.jsonl")),
+                rows.iter().map(|r| format!("{r}\n")).collect::<String>(),
+            )
+            .unwrap();
+        }
+        let report = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!((report.scanned_files, report.repaired_items), (3, 2));
+        let repaired_turn: String = db
+            .query_row(
+                "SELECT turn_id FROM thread_items WHERE turn_id LIKE 'goal-%'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(repaired_turn, "goal-a");
+    }
+
+    #[test]
+    fn session_index_restores_missing_item_behind_existing_pointer() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        db.execute(
+            "UPDATE thread_turns SET first_user_item_id='original-id'",
+            [],
+        )
+        .unwrap();
+        let mut c = candidate();
+        c.projection = Some(("original-id".into(), 3));
+        c.first_user_projection = c.projection.clone();
+        assert!(apply(&db, &c, inspect(&db, &c).unwrap()).unwrap());
+        assert_eq!(counts(&db), (1, Some("original-id".into())));
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Present));
+    }
+
+    #[test]
+    fn legacy_goal_in_later_fragment_prevents_duplicate_repair() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        db.execute_batch("CREATE UNIQUE INDEX idx_thread_items_page ON thread_items(thread_id,rollout_ordinal); INSERT INTO thread_turns VALUES('thread','earlier','completed',NULL)").unwrap();
+        let mut legacy = candidate();
+        legacy.goal = true;
+        legacy.ordinal = 40;
+        legacy.first_user_response_ordinal = None;
+        // 使用旧版没有goal_key时的恢复ID，正文与位置保持原样。
+        apply(&db, &legacy, inspect(&db, &legacy).unwrap()).unwrap();
+        legacy.goal_key = Some("created:100".into());
+        let mut earlier = legacy.clone();
+        earlier.turn = "earlier".into();
+        earlier.ordinal = 4;
+        let selected = deduplicate_goals(&db, vec![earlier.clone(), legacy.clone()]).unwrap();
+        assert_eq!(selected.len(), 1);
+        assert_eq!(selected[0].turn, legacy.turn);
+        assert!(matches!(
+            inspect(&db, &selected[0]).unwrap(),
+            Action::Present
+        ));
+        // 同文但不同身份的新目标仍然独立保留。
+        earlier.goal_key = Some("created:200".into());
+        assert_eq!(
+            deduplicate_goals(&db, vec![earlier, legacy]).unwrap().len(),
+            2
+        );
+        assert_eq!(counts(&db).0, 1);
+    }
+
+    fn schema(db: &Connection) {
+        db.execute_batch("CREATE TABLE thread_turns(thread_id TEXT,turn_id TEXT,status TEXT,first_user_item_id TEXT,PRIMARY KEY(thread_id,turn_id));
+            CREATE TABLE thread_items(thread_id TEXT,turn_id TEXT,item_id TEXT,rollout_ordinal INTEGER,created_at_ms INTEGER,item_json TEXT,item_type TEXT,updated_at_ordinal INTEGER,PRIMARY KEY(thread_id,turn_id,item_id));
+            INSERT INTO thread_turns VALUES('thread','turn','completed',NULL);").unwrap();
+    }
+
+    #[test]
+    fn existing_empty_turn_does_not_promote_steering_or_borrow_dangling_id() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        let mut c = candidate();
+        c.ordinal = 7;
+        c.projection = Some(("steering-native".into(), 8));
+        c.first_user_response_ordinal = Some(2);
+        c.first_user_projection = Some(("image-native".into(), 4));
+        for first in [None, Some("image-native")] {
+            db.execute("UPDATE thread_turns SET first_user_item_id=?1", [first])
+                .unwrap();
+            assert!(matches!(inspect(&db, &c).unwrap(), Action::Wait(_)));
+            assert_eq!(counts(&db).0, 0);
+        }
+        c.ordinal = 2;
+        c.projection = Some(("different-native".into(), 4));
+        c.first_user_projection = c.projection.clone();
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Skip(_)));
+        db.execute(
+            "UPDATE thread_turns SET first_user_item_id='different-native'",
+            [],
+        )
+        .unwrap();
+        assert!(apply(&db, &c, inspect(&db, &c).unwrap()).unwrap());
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Present));
+    }
+
+    #[test]
+    fn thread_wide_ordinal_conflict_does_not_abort_other_repairs() {
+        let home = home_with_rollout();
+        let db = Connection::open(home.path().join("thread_history_1.sqlite")).unwrap();
+        db.execute_batch("CREATE UNIQUE INDEX idx_thread_items_page ON thread_items(thread_id,rollout_ordinal); INSERT INTO thread_turns VALUES('thread','other','completed',NULL);").unwrap();
+        db.execute("INSERT INTO thread_items VALUES('thread','other','occupied',3,1,'{}','agentMessage',3)",[]).unwrap();
+        let original = fs::read_to_string(home.path().join("sessions/rollout.jsonl")).unwrap();
+        let other = original
+            .replace("\"turn\"", "\"good\"")
+            .replace("\"ordinal\":3", "\"ordinal\":9")
+            .replace("完整原文", "独立可修消息");
+        db.execute(
+            "INSERT INTO thread_turns VALUES('thread','good','completed',NULL)",
+            [],
+        )
+        .unwrap();
+        fs::write(home.path().join("sessions/good.jsonl"), other).unwrap();
+        let report = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!((report.repaired_items, report.skipped_items), (1, 1));
+        assert!(
+            load_session_index_repair_report(Some(home.path()))
+                .unwrap()
+                .is_some()
+        );
+        assert_eq!(
+            db.query_row(
+                "SELECT count(*) FROM thread_items WHERE item_type='userMessage'",
+                [],
+                |r| r.get::<_, i64>(0)
+            )
+            .unwrap(),
+            1
+        );
+    }
+
+    #[test]
+    fn goals_with_same_text_but_different_identity_both_survive() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        db.execute_batch("CREATE UNIQUE INDEX idx_thread_items_page ON thread_items(thread_id,rollout_ordinal); INSERT INTO thread_turns VALUES('thread','second','completed',NULL);").unwrap();
+        let mut a = candidate();
+        a.goal = true;
+        a.goal_key = Some("goal-one:100".into());
+        a.first_user_response_ordinal = None;
+        apply(&db, &a, inspect(&db, &a).unwrap()).unwrap();
+        let mut b = a.clone();
+        b.turn = "second".into();
+        b.ordinal = 20;
+        b.goal_key = Some("goal-two:200".into());
+        assert!(apply(&db, &b, inspect(&db, &b).unwrap()).unwrap());
+        assert_ne!(recovery_id(&a), recovery_id(&b));
+        for c in [&a, &b] {
+            assert!(matches!(inspect(&db, c).unwrap(), Action::Present));
+        }
+        b.turn = "continuation".into();
+        assert!(matches!(inspect(&db, &b).unwrap(), Action::Present));
+        assert_eq!(counts(&db).0, 2);
+    }
+
+    #[test]
+    fn incomplete_tail_is_never_success_cached_and_rechecks_when_completed() {
+        let home = home_with_rollout();
+        let path = home.path().join("sessions/rollout.jsonl");
+        let original = fs::read_to_string(&path).unwrap();
+        fs::write(&path, format!("{original}{{\"type\":")).unwrap();
+        for _ in 0..2 {
+            let r = repair_session_index(Some(home.path())).unwrap();
+            assert_eq!(
+                (r.cached_files, r.repaired_items, r.deferred_items),
+                (0, 0, 1)
+            );
+            assert!(r.pending_details[0].reason.contains("尾行不完整"));
+        }
+        let cache =
+            Connection::open(home.path().join("session-index-repair/scan-cache.sqlite")).unwrap();
+        cache
+            .execute("UPDATE pending SET first_seen=0", [])
+            .unwrap();
+        let r = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!((r.deferred_items, r.skipped_items), (0, 1));
+        fs::write(&path,format!("{original}{{\"type\":\"event_msg\",\"payload\":{{\"type\":\"task_complete\",\"turn_id\":\"turn\"}}}}\n")).unwrap();
+        let r = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!(
+            (r.repaired_items, r.skipped_items, r.deferred_items),
+            (1, 0, 0)
+        );
+    }
+
+    #[test]
+    fn single_completion_event_cannot_insert_two_repeated_messages() {
+        let home = home_with_rollout();
+        let rows = [
+            json!({"type":"session_meta","ordinal":0,"payload":{"id":"thread"}}),
+            json!({"type":"event_msg","ordinal":1,"payload":{"type":"task_started","turn_id":"turn"}}),
+            json!({"type":"response_item","ordinal":2,"timestamp":"2026-09-20T01:00:00Z","payload":{"role":"user","content":[{"type":"input_text","text":"重复文本"}],"internal_chat_message_metadata_passthrough":{"turn_id":"turn"}}}),
+            json!({"type":"response_item","ordinal":5,"timestamp":"2026-09-20T01:00:00Z","payload":{"role":"user","content":[{"type":"input_text","text":"重复文本"}],"internal_chat_message_metadata_passthrough":{"turn_id":"turn"}}}),
+            json!({"type":"event_msg","ordinal":6,"payload":{"type":"item_completed","thread_id":"thread","turn_id":"turn","item":{"type":"UserMessage","id":"native","content":[{"type":"text","text":"重复文本"}]}}}),
+        ];
+        fs::write(
+            home.path().join("sessions/rollout.jsonl"),
+            rows.iter().map(|r| format!("{r}\n")).collect::<String>(),
+        )
+        .unwrap();
+        let report = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!((report.repaired_items, report.skipped_items), (0, 0));
+        let db = Connection::open(home.path().join("thread_history_1.sqlite")).unwrap();
+        assert_eq!(counts(&db), (0, None));
+    }
+
+    fn candidate() -> Candidate {
+        Candidate {
+            thread: "thread".into(),
+            turn: "turn".into(),
+            text: "完整原文".into(),
+            ordinal: 3,
+            created: 100_000,
+            goal: false,
+            projection: None,
+            following_turn: None,
+            turn_boundary: None,
+            start_ordinal: None,
+            goal_key: None,
+            first_user_projection: None,
+            first_user_response_ordinal: Some(3),
+            first_user_evidence_valid: true,
+        }
+    }
+
+    #[test]
+    fn session_index_recognizes_multiple_messages_and_restores_missing_steering() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        let first = candidate();
+        apply(&db, &first, inspect(&db, &first).unwrap()).unwrap();
+        let mut last = first.clone();
+        last.ordinal = 9;
+        last.projection = Some(("last".into(), 10));
+        last.text = "后续补充".into();
+        assert!(apply(&db, &last, inspect(&db, &last).unwrap()).unwrap());
+        let mut middle = first.clone();
+        middle.ordinal = 6;
+        middle.projection = Some(("middle".into(), 7));
+        middle.text = "中途调整要求".into();
+        assert!(apply(&db, &middle, inspect(&db, &middle).unwrap()).unwrap());
+        for c in [&first, &middle, &last] {
+            assert!(matches!(inspect(&db, c).unwrap(), Action::Present));
+        }
+        assert_eq!(counts(&db), (3, Some("recovered-user-turn-3".into())));
+        middle.text = "相同位置的冲突原文".into();
+        assert!(matches!(inspect(&db, &middle).unwrap(), Action::Skip(_)));
+        middle.ordinal = 2;
+        middle.projection = Some(("earlier".into(), 2));
+        assert!(matches!(inspect(&db, &middle).unwrap(), Action::Skip(_)));
+        assert_eq!(counts(&db).0, 3);
+    }
+
+    #[test]
+    fn session_index_uses_native_completion_identity_and_preserves_repeated_text() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        let mut first = candidate();
+        first.projection = Some(("native-first".into(), 4));
+        apply(&db, &first, inspect(&db, &first).unwrap()).unwrap();
+        assert!(matches!(inspect(&db, &first).unwrap(), Action::Present));
+        let mut repeated = first.clone();
+        repeated.ordinal = 8;
+        repeated.projection = Some(("native-second".into(), 9));
+        assert!(apply(&db, &repeated, inspect(&db, &repeated).unwrap()).unwrap());
+        assert_eq!(counts(&db), (2, Some("native-first".into())));
+        let ordinal: i64 = db
+            .query_row(
+                "SELECT rollout_ordinal FROM thread_items WHERE item_id='native-second'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap();
+        assert_eq!(ordinal, 9);
+        first.projection = None;
+        assert!(matches!(inspect(&db, &first).unwrap(), Action::Skip(_)));
+        repeated.projection = None;
+        repeated.text = "没有完成事件的待补消息".into();
+        assert!(matches!(inspect(&db, &repeated).unwrap(), Action::Skip(_)));
+    }
+
+    #[test]
+    fn session_index_scans_multiple_messages_without_turn_wide_conflict() {
+        let home = home_with_rollout();
+        let path = home.path().join("sessions/rollout.jsonl");
+        let mut raw = fs::read_to_string(&path).unwrap();
+        let first_completed = json!({"type":"event_msg","ordinal":4,"payload":{"type":"item_completed","thread_id":"thread","turn_id":"turn","item":{"id":"first","type":"UserMessage","content":[{"type":"Text","text":"完整原文"}]}}});
+        raw.push_str(&format!("{first_completed}\n"));
+        for (ordinal, text) in [(5, "补充要求"), (7, "再次补充")] {
+            for row in [
+                json!({"type":"event_msg","payload":{"type":"user_message","message":text}}),
+                json!({"type":"response_item","ordinal":ordinal,"timestamp":"2026-09-14T12:25:30Z","payload":{"role":"user","content":[{"type":"input_text","text":text}],"internal_chat_message_metadata_passthrough":{"turn_id":"turn"}}}),
+                json!({"type":"event_msg","ordinal":ordinal+1,"payload":{"type":"item_completed","thread_id":"thread","turn_id":"turn","item":{"id":format!("item-{ordinal}"),"type":"UserMessage","content":[{"type":"Text","text":text}]}}}),
+            ] {
+                raw.push_str(&format!("{row}\n"));
+            }
+        }
+        fs::write(&path, raw).unwrap();
+        let report = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!((report.repaired_items, report.skipped_items), (3, 0));
+        let again = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!(
+            (
+                again.repaired_items,
+                again.already_present,
+                again.skipped_items
+            ),
+            (0, 3, 0)
+        );
+    }
+
+    #[test]
+    fn session_index_rejects_conflicting_candidates_at_same_ordinal() {
+        let home = home_with_rollout();
+        let path = home.path().join("sessions/conflict.jsonl");
+        let raw = fs::read_to_string(home.path().join("sessions/rollout.jsonl")).unwrap();
+        fs::write(path, raw.replace("完整原文", "冲突原文")).unwrap();
+        let report = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!((report.repaired_items, report.skipped_items), (0, 2));
+        let db = Connection::open(home.path().join("thread_history_1.sqlite")).unwrap();
+        assert_eq!(counts(&db), (0, None));
+    }
+
+    fn counts(db: &Connection) -> (i64, Option<String>) {
+        (
+            db.query_row("SELECT count(*) FROM thread_items", [], |r| r.get(0))
+                .unwrap(),
+            db.query_row(
+                "SELECT first_user_item_id FROM thread_turns WHERE turn_id='turn'",
+                [],
+                |r| r.get(0),
+            )
+            .unwrap(),
+        )
+    }
+
+    #[test]
+    fn session_index_insert_is_atomic_and_idempotent() {
+        let mut db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        let c = candidate();
+        {
+            let tx = db.transaction().unwrap();
+            assert!(apply(&tx, &c, inspect(&tx, &c).unwrap()).unwrap());
+            assert_eq!(counts(&tx).0, 1);
+            // 未提交事务必须同时回滚消息和 first_user_item_id。
+        }
+        assert_eq!(counts(&db), (0, None));
+        let tx = db.transaction().unwrap();
+        assert!(apply(&tx, &c, inspect(&tx, &c).unwrap()).unwrap());
+        tx.commit().unwrap();
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Present));
+        assert!(!apply(&db, &c, inspect(&db, &c).unwrap()).unwrap());
+        assert_eq!(counts(&db), (1, Some("recovered-user-turn-3".into())));
+        let raw: String = db
+            .query_row("SELECT item_json FROM thread_items", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(
+            serde_json::from_str::<Value>(&raw).unwrap()["content"][0]["text"],
+            c.text
+        );
+    }
+
+    #[test]
+    fn session_index_preserves_conflicting_messages_and_skips_unfinished_turns() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        let c = candidate();
+        for status in ["inProgress", "running", "unknown"] {
+            db.execute("UPDATE thread_turns SET status=?1", [status])
+                .unwrap();
+            assert!(matches!(inspect(&db, &c).unwrap(), Action::Wait(_)));
+            assert_eq!(counts(&db), (0, None));
+        }
+        db.execute("UPDATE thread_turns SET status='completed'", [])
+            .unwrap();
+        let mut other = candidate();
+        other.text = "原有的不同内容".into();
+        apply(&db, &other, inspect(&db, &other).unwrap()).unwrap();
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Skip(_)));
+        db.execute("UPDATE thread_turns SET first_user_item_id=NULL", [])
+            .unwrap();
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Skip(_)));
+        assert_eq!(counts(&db), (1, None));
+        let mut missing = candidate();
+        missing.turn = "不存在的轮次".into();
+        assert!(matches!(inspect(&db, &missing).unwrap(), Action::Wait(_)));
+    }
+
+    #[test]
+    fn session_index_relinks_only_unambiguous_orphan_user_message() {
+        let db = Connection::open_in_memory().unwrap();
+        schema(&db);
+        let c = candidate();
+        apply(&db, &c, inspect(&db, &c).unwrap()).unwrap();
+        db.execute("UPDATE thread_turns SET first_user_item_id=NULL", [])
+            .unwrap();
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Link(_)));
+        assert!(apply(&db, &c, inspect(&db, &c).unwrap()).unwrap());
+        assert_eq!(counts(&db).0, 1);
+        db.execute(
+            "UPDATE thread_turns SET first_user_item_id='different-existing-item'",
+            [],
+        )
+        .unwrap();
+        assert!(matches!(inspect(&db, &c).unwrap(), Action::Skip(_)));
+        assert_eq!(counts(&db).1.as_deref(), Some("different-existing-item"));
+    }
+
+    fn home_with_rollout() -> tempfile::TempDir {
+        let home = tempfile::tempdir().unwrap();
+        let db = Connection::open(home.path().join("thread_history_1.sqlite")).unwrap();
+        schema(&db);
+        fs::create_dir(home.path().join("sessions")).unwrap();
+        let rows = [
+            json!({"type":"session_meta","payload":{"id":"thread"}}),
+            json!({"type":"event_msg","ordinal":1,"payload":{"type":"task_started","turn_id":"turn"}}),
+            json!({"type":"event_msg","payload":{"type":"user_message","message":"完整原文"}}),
+            json!({"type":"response_item","ordinal":3,"timestamp":"2026-09-14T12:24:30Z","payload":{"role":"user","content":[{"type":"input_text","text":"完整原文"}],"internal_chat_message_metadata_passthrough":{"turn_id":"turn"}}}),
+        ];
+        fs::write(
+            home.path().join("sessions/rollout.jsonl"),
+            rows.iter().map(|r| format!("{r}\n")).collect::<String>(),
+        )
+        .unwrap();
+        home
+    }
+
+    #[test]
+    fn session_index_cached_scan_rechecks_rebuilt_native_index_and_saves_report() {
+        let home = home_with_rollout();
+        let report = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!(
+            (
+                report.scanned_files,
+                report.cached_files,
+                report.repaired_items
+            ),
+            (1, 0, 1)
+        );
+        let backup = Connection::open(report.backup_path.unwrap()).unwrap();
+        assert_eq!(counts(&backup), (0, None));
+        let again = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!(
+            (
+                again.cached_files,
+                again.repaired_items,
+                again.already_present
+            ),
+            (1, 0, 1)
+        );
+        assert!(again.backup_path.is_none());
+        let db = Connection::open(home.path().join("thread_history_1.sqlite")).unwrap();
+        db.execute_batch(
+            "DELETE FROM thread_items; UPDATE thread_turns SET first_user_item_id=NULL;",
+        )
+        .unwrap();
+        let rebuilt = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!((rebuilt.cached_files, rebuilt.repaired_items), (1, 1));
+        assert_eq!(counts(&db).0, 1);
+        let saved = load_session_index_repair_report(Some(home.path()))
+            .unwrap()
+            .unwrap();
+        assert_eq!((saved.cached_files, saved.repaired_items), (1, 1));
+        assert!(!home.path().join("session-index-repair/report.tmp").exists());
+    }
+
+    #[test]
+    fn session_index_waiting_clears_after_native_turn_finishes_without_backups_while_waiting() {
+        let home = home_with_rollout();
+        let db = Connection::open(home.path().join("thread_history_1.sqlite")).unwrap();
+        db.execute("UPDATE thread_turns SET status='inProgress'", [])
+            .unwrap();
+        let waiting = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!(waiting.deferred_items, 1);
+        assert!(waiting.backup_path.is_none());
+        assert_eq!(waiting.pending_details[0].checks, 1);
+        let cache =
+            Connection::open(home.path().join("session-index-repair/scan-cache.sqlite")).unwrap();
+        cache
+            .execute("UPDATE pending SET first_seen=0", [])
+            .unwrap();
+        let blocked = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!((blocked.deferred_items, blocked.skipped_items), (0, 1));
+        assert!(blocked.backup_path.is_none());
+        db.execute("UPDATE thread_turns SET status='completed'", [])
+            .unwrap();
+        let fixed = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!(
+            (
+                fixed.repaired_items,
+                fixed.deferred_items,
+                fixed.skipped_items
+            ),
+            (1, 0, 0)
+        );
+        assert!(fixed.pending_details.is_empty());
+        let remaining: i64 = cache
+            .query_row("SELECT count(*) FROM pending", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(remaining, 0);
+        let repeat = repair_session_index(Some(home.path())).unwrap();
+        assert_eq!(
+            (
+                repeat.repaired_items,
+                repeat.already_present,
+                repeat.deferred_items
+            ),
+            (0, 1, 0)
+        );
+        assert!(repeat.backup_path.is_none());
+    }
+}
diff --git a/crates/codex-plus-data/src/session_index_scan.rs b/crates/codex-plus-data/src/session_index_scan.rs
new file mode 100644
index 000000000..3d101fe3a
--- /dev/null
+++ b/crates/codex-plus-data/src/session_index_scan.rs
@@ -0,0 +1,1233 @@
+//! 仅从相互印证的原始事件生成候选；不把自动续执行当作新指令。
+use anyhow::{Context, bail};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::{
+    collections::{HashMap, HashSet},
+    fs::File,
+    io::{BufRead, BufReader},
+    path::Path,
+};
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub(crate) struct TurnBoundary {
+    pub start_ordinal: i64,
+    pub started_at: i64,
+    pub end_ordinal: i64,
+    pub completed_at: i64,
+    pub status: String,
+    pub error_json: Option<String>,
+    pub duration_ms: Option<i64>,
+    pub first_user_projection: Option<(String, i64)>,
+    pub first_user_response_ordinal: Option<i64>,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub(crate) struct Candidate {
+    pub thread: String,
+    pub turn: String,
+    pub text: String,
+    pub ordinal: i64,
+    pub created: i64,
+    pub goal: bool,
+    #[serde(default)]
+    pub goal_key: Option<String>,
+    #[serde(default)]
+    pub projection: Option<(String, i64)>,
+    #[serde(default)]
+    pub following_turn: Option<(String, i64)>,
+    #[serde(default)]
+    pub turn_boundary: Option<TurnBoundary>,
+    #[serde(default)]
+    pub start_ordinal: Option<i64>,
+    #[serde(default)]
+    pub first_user_projection: Option<(String, i64)>,
+    #[serde(default)]
+    pub first_user_response_ordinal: Option<i64>,
+    #[serde(default)]
+    pub first_user_evidence_valid: bool,
+}
+
+fn event_time(row: &Value, field: &str) -> Option<i64> {
+    let payload = &row["payload"][field];
+    if !payload.is_null() {
+        return payload.as_i64().filter(|time| *time >= 0);
+    }
+    chrono::DateTime::parse_from_rfc3339(row["timestamp"].as_str()?)
+        .ok()
+        .map(|time| time.timestamp())
+        .filter(|time| *time >= 0)
+}
+
+fn text_content(value: &Value) -> Option<String> {
+    let parts = value.as_array()?;
+    if parts.is_empty() {
+        return None;
+    }
+    let mut text = String::new();
+    for part in parts {
+        if !matches!(part["type"].as_str(), Some("text" | "input_text" | "Text")) {
+            return None;
+        }
+        text.push_str(part["text"].as_str()?);
+    }
+    (!text.is_empty()).then_some(text)
+}
+
+fn ordinary(text: &str) -> bool {
+    let text = text.trim_start();
+    !text.is_empty()
+        && ![
+            "<codex_internal_context",
+            "<environment_context",
+            "<permissions instructions>",
+            "# AGENTS.md instructions",
+            "<INSTRUCTIONS>",
+        ]
+        .iter()
+        .any(|prefix| text.starts_with(prefix))
+}
+
+pub(crate) fn scan(path: &Path) -> anyhow::Result<Vec<Candidate>> {
+    let mut reader = BufReader::new(File::open(path)?);
+    let mut line = String::new();
+    let mut thread = None::<String>;
+    let mut inherited_from = HashSet::<String>::new();
+    let mut own_history_start = None::<i64>;
+    let mut current_turn = None::<String>;
+    let mut last_started_turn = None::<String>;
+    let mut following_turns = HashMap::<String, Option<(String, i64)>>::new();
+    let mut active_boundary = None::<(String, i64, i64)>;
+    let mut started_turns = HashSet::<String>::new();
+    let mut duplicate_starts = HashSet::<String>::new();
+    let mut start_ordinals = HashMap::<String, i64>::new();
+    let mut ended_turns = HashSet::<String>::new();
+    let mut invalid_boundaries = HashSet::<String>::new();
+    let mut invalid_first_user_evidence = HashSet::<String>::new();
+    let mut boundaries = HashMap::<String, TurnBoundary>::new();
+    let mut first_user_projections = HashMap::<String, (String, i64)>::new();
+    let mut first_user_responses = HashMap::<String, i64>::new();
+    let mut responses = HashMap::<String, Vec<(Candidate, usize)>>::new();
+    let mut evidence = HashMap::<(String, String), Vec<usize>>::new();
+    let mut projections = HashMap::<(String, String), Vec<(String, i64)>>::new();
+    let mut goal = None::<(String, i64, String, Option<String>)>;
+    let mut seen_goals = HashSet::<(String, String)>::new();
+    let mut result = Vec::new();
+    let mut stream_position = 0usize;
+    loop {
+        line.clear();
+        if reader.read_line(&mut line)? == 0 {
+            break;
+        }
+        stream_position = stream_position.saturating_add(1);
+        let row: Value = match serde_json::from_str(&line) {
+            Ok(row) => row,
+            Err(_) if !line.ends_with('\n') => bail!("会话记录尾行不完整"),
+            Err(error) => return Err(error).context("会话记录包含无效 JSON 行"),
+        };
+        let p = &row["payload"];
+        if row["type"] == "session_meta" {
+            let id = p["id"].as_str().context("会话标识缺失")?;
+            if thread.as_deref().is_some_and(|old| old != id) {
+                // 子任务日志包含父任务快照；只有元数据声明的父任务及边界内记录可忽略。
+                if inherited_from.contains(id)
+                    && own_history_start.is_some_and(|start| {
+                        row["ordinal"]
+                            .as_i64()
+                            .is_some_and(|ordinal| ordinal >= 0 && ordinal < start)
+                    })
+                {
+                    continue;
+                }
+                bail!("文件混入其他会话");
+            }
+            if thread.is_none() {
+                own_history_start = p["subagent_history_start_ordinal"]
+                    .as_i64()
+                    .filter(|v| *v >= 0);
+                for key in ["parent_thread_id", "forked_from_id"] {
+                    if let Some(parent) = p[key].as_str() {
+                        inherited_from.insert(parent.to_owned());
+                    }
+                }
+            }
+            thread = Some(id.to_owned());
+            continue;
+        }
+        let thread = thread.as_ref().context("首条记录缺少 session_meta")?;
+        if let Some(start) = own_history_start {
+            let ordinal = row["ordinal"]
+                .as_i64()
+                .context("分叉记录缺少序号，无法确认历史边界")?;
+            if ordinal < start {
+                continue;
+            }
+        }
+        if p["thread_id"].as_str().is_some_and(|id| id != thread)
+            || p["threadId"].as_str().is_some_and(|id| id != thread)
+        {
+            continue;
+        }
+        if row["type"] == "event_msg" {
+            if p["type"] == "item_completed"
+                && matches!(
+                    p["item"]["type"].as_str(),
+                    Some("UserMessage" | "userMessage")
+                )
+            {
+                if let Some(turn) = p["turn_id"].as_str().or(current_turn.as_deref()) {
+                    if let (Some(id), Some(ordinal)) = (
+                        p["item"]["id"].as_str().filter(|id| !id.is_empty()),
+                        row["ordinal"].as_i64().filter(|ordinal| *ordinal >= 0),
+                    ) {
+                        if p["thread_id"].as_str() == Some(thread)
+                            && p["turn_id"].as_str() == Some(turn)
+                        {
+                            let first = first_user_projections
+                                .entry(turn.to_owned())
+                                .or_insert((id.to_owned(), ordinal));
+                            if ordinal < first.1 {
+                                *first = (id.to_owned(), ordinal);
+                            } else if ordinal == first.1 && id != first.0 {
+                                invalid_boundaries.insert(turn.to_owned());
+                                invalid_first_user_evidence.insert(turn.to_owned());
+                            }
+                        } else {
+                            invalid_boundaries.insert(turn.to_owned());
+                            invalid_first_user_evidence.insert(turn.to_owned());
+                        }
+                    } else {
+                        invalid_boundaries.insert(turn.to_owned());
+                        invalid_first_user_evidence.insert(turn.to_owned());
+                    }
+                }
+            }
+            match p["type"].as_str() {
+                Some("task_started") => {
+                    if let Some((interrupted, _, _)) = active_boundary.take() {
+                        invalid_boundaries.insert(interrupted);
+                    }
+                    current_turn = p["turn_id"].as_str().map(str::to_owned);
+                    let started = current_turn.as_ref().filter(|id| !id.is_empty());
+                    if let Some(turn) = started {
+                        if !started_turns.insert(turn.clone()) {
+                            duplicate_starts.insert(turn.clone());
+                            invalid_boundaries.insert(turn.clone());
+                        }
+                        if ended_turns.contains(turn) {
+                            invalid_boundaries.insert(turn.clone());
+                        }
+                        if let Some(ordinal) =
+                            row["ordinal"].as_i64().filter(|ordinal| *ordinal >= 0)
+                        {
+                            start_ordinals.entry(turn.clone()).or_insert(ordinal);
+                        }
+                        if let (Some(ordinal), Some(time)) = (
+                            row["ordinal"].as_i64().filter(|ordinal| *ordinal >= 0),
+                            event_time(&row, "started_at"),
+                        ) {
+                            active_boundary = Some((turn.clone(), ordinal, time));
+                        }
+                    }
+                    if let Some(previous) = last_started_turn.take() {
+                        if started.is_none_or(|id| id != &previous) {
+                            // 只接受紧接的真实开始边界；缺失信息不得跳到更晚的轮次补猜。
+                            let following = started.and_then(|id| {
+                                row["ordinal"]
+                                    .as_i64()
+                                    .filter(|ordinal| *ordinal >= 0)
+                                    .map(|ordinal| (id.clone(), ordinal))
+                            });
+                            following_turns.entry(previous).or_insert(following);
+                        }
+                    }
+                    last_started_turn = started.cloned();
+                    if let Some((_, _, _, turn)) = goal.as_mut() {
+                        if turn.is_none() {
+                            *turn = current_turn.clone();
+                        }
+                    }
+                }
+                Some(kind @ ("task_complete" | "task_failed" | "turn_aborted")) => {
+                    let terminal_turn = p["turn_id"].as_str().or(current_turn.as_deref());
+                    if let Some(turn) = terminal_turn {
+                        if !ended_turns.insert(turn.to_owned()) {
+                            invalid_boundaries.insert(turn.to_owned());
+                        }
+                        if let Some((active, start_ordinal, started_at)) = active_boundary.take() {
+                            if active != turn
+                                || (!p["started_at"].is_null()
+                                    && p["started_at"].as_i64() != Some(started_at))
+                            {
+                                invalid_boundaries.insert(active);
+                                invalid_boundaries.insert(turn.to_owned());
+                            } else if let (Some(end_ordinal), Some(completed_at)) = (
+                                row["ordinal"]
+                                    .as_i64()
+                                    .filter(|ordinal| *ordinal > start_ordinal),
+                                event_time(&row, "completed_at").filter(|time| *time >= started_at),
+                            ) {
+                                let error_json =
+                                    (!p["error"].is_null()).then(|| p["error"].to_string());
+                                let status = match kind {
+                                    "turn_aborted" => "interrupted",
+                                    "task_failed" => "failed",
+                                    _ if error_json.is_some() => "failed",
+                                    _ => "completed",
+                                };
+                                boundaries.insert(
+                                    turn.to_owned(),
+                                    TurnBoundary {
+                                        start_ordinal,
+                                        started_at,
+                                        end_ordinal,
+                                        completed_at,
+                                        status: status.to_owned(),
+                                        error_json,
+                                        duration_ms: p["duration_ms"]
+                                            .as_i64()
+                                            .filter(|duration| *duration >= 0),
+                                        first_user_projection: None,
+                                        first_user_response_ordinal: None,
+                                    },
+                                );
+                            }
+                        }
+                    } else if let Some((active, _, _)) = active_boundary.take() {
+                        invalid_boundaries.insert(active);
+                    }
+                    current_turn = None;
+                }
+                Some("thread_goal_updated") => {
+                    if p["threadId"].as_str() != Some(thread) {
+                        continue;
+                    }
+                    if let (Some(text), Some(created)) = (
+                        p["goal"]["objective"].as_str(),
+                        p["goal"]["createdAt"].as_i64(),
+                    ) {
+                        let goal_key = p["goal"]["id"]
+                            .as_str()
+                            .filter(|id| !id.is_empty())
+                            .map(|id| format!("id:{id}:created:{created}"))
+                            .unwrap_or_else(|| format!("created:{created}"));
+                        if !text.is_empty()
+                            && !seen_goals.contains(&(goal_key.clone(), text.to_owned()))
+                        {
+                            goal = created
+                                .checked_mul(1000)
+                                .map(|created| (text.to_owned(), created, goal_key, None));
+                        }
+                    }
+                }
+                Some("user_message") => {
+                    let turn = p["turn_id"].as_str().or(current_turn.as_deref());
+                    if let (Some(turn), Some(text)) = (turn, p["message"].as_str()) {
+                        if ordinary(text) {
+                            evidence
+                                .entry((turn.to_owned(), text.to_owned()))
+                                .or_default()
+                                .push(stream_position);
+                        }
+                    }
+                }
+                Some("item_completed")
+                    if p["thread_id"].as_str() == Some(thread)
+                        && matches!(
+                            p["item"]["type"].as_str(),
+                            Some("UserMessage" | "userMessage")
+                        ) =>
+                {
+                    if let (Some(turn), Some(text)) =
+                        (p["turn_id"].as_str(), text_content(&p["item"]["content"]))
+                    {
+                        if ordinary(&text) {
+                            if let (Some(id), Some(ordinal)) =
+                                (p["item"]["id"].as_str(), row["ordinal"].as_i64())
+                            {
+                                if !id.is_empty() && ordinal >= 0 {
+                                    projections
+                                        .entry((turn.to_owned(), text.clone()))
+                                        .or_default()
+                                        .push((id.to_owned(), ordinal));
+                                }
+                            }
+                        }
+                    }
+                }
+                _ => {}
+            }
+        }
+        if row["type"] != "response_item" || p["role"] != "user" {
+            continue;
+        }
+        let meta = &p["internal_chat_message_metadata_passthrough"];
+        // 附件消息也占据首条用户输入位置，不能因无法恢复为纯文本而被后续 steering 越过。
+        let parts = p["content"].as_array();
+        let visible_text = parts
+            .map(|parts| {
+                parts
+                    .iter()
+                    .filter_map(|part| part["text"].as_str())
+                    .collect::<String>()
+            })
+            .unwrap_or_default();
+        let has_attachment = parts.is_some_and(|parts| {
+            parts
+                .iter()
+                .any(|part| !matches!(part["type"].as_str(), Some("text" | "input_text" | "Text")))
+        });
+        if ordinary(&visible_text) || has_attachment {
+            if let Some(turn) = meta["turn_id"].as_str().or(current_turn.as_deref()) {
+                if let Some(ordinal) = row["ordinal"].as_i64().filter(|ordinal| *ordinal >= 0) {
+                    if meta["turn_id"].as_str() == Some(turn) {
+                        let first = first_user_responses
+                            .entry(turn.to_owned())
+                            .or_insert(ordinal);
+                        *first = (*first).min(ordinal);
+                    } else {
+                        invalid_boundaries.insert(turn.to_owned());
+                        invalid_first_user_evidence.insert(turn.to_owned());
+                    }
+                } else {
+                    invalid_boundaries.insert(turn.to_owned());
+                    invalid_first_user_evidence.insert(turn.to_owned());
+                }
+            }
+        }
+        let (Some(turn), Some(ordinal), Some(text)) = (
+            meta["turn_id"].as_str(),
+            row["ordinal"].as_i64(),
+            text_content(&p["content"]),
+        ) else {
+            continue;
+        };
+        if ordinal < 0 {
+            continue;
+        }
+        if let Some((objective, created, goal_key, Some(goal_turn))) = &goal {
+            if turn == goal_turn
+                && text.starts_with("<codex_internal_context")
+                && (text.contains(&format!("<objective>\n{objective}\n</objective>"))
+                    || text.contains(&format!("<objective>\r\n{objective}\r\n</objective>")))
+            {
+                result.push(Candidate {
+                    thread: thread.clone(),
+                    turn: turn.to_owned(),
+                    text: objective.clone(),
+                    ordinal,
+                    created: *created,
+                    goal: true,
+                    goal_key: Some(goal_key.clone()),
+                    projection: None,
+                    following_turn: None,
+                    turn_boundary: None,
+                    start_ordinal: None,
+                    first_user_projection: None,
+                    first_user_response_ordinal: None,
+                    first_user_evidence_valid: false,
+                });
+                seen_goals.insert((goal_key.clone(), objective.clone()));
+                goal = None;
+                continue;
+            }
+        }
+        if !ordinary(&text) {
+            continue;
+        }
+        let created = meta["create_time"]
+            .as_f64()
+            .filter(|v| v.is_finite() && *v > 0.0)
+            .map(|v| (v * 1000.0) as i64)
+            .or_else(|| {
+                chrono::DateTime::parse_from_rfc3339(row["timestamp"].as_str()?)
+                    .ok()
+                    .map(|t| t.timestamp_millis())
+            });
+        if let Some(created) = created {
+            responses.entry(turn.to_owned()).or_default().push((
+                Candidate {
+                    thread: thread.clone(),
+                    turn: turn.to_owned(),
+                    text,
+                    ordinal,
+                    created,
+                    goal: false,
+                    goal_key: None,
+                    projection: None,
+                    following_turn: None,
+                    turn_boundary: None,
+                    start_ordinal: None,
+                    first_user_projection: None,
+                    first_user_response_ordinal: None,
+                    first_user_evidence_valid: false,
+                },
+                stream_position,
+            ));
+        }
+    }
+    for candidates in responses.into_values() {
+        let mut response_counts = HashMap::<String, usize>::new();
+        for (candidate, _) in &candidates {
+            *response_counts.entry(candidate.text.clone()).or_default() += 1;
+        }
+        if let Some((first, _)) = candidates.first() {
+            for (text, expected) in &response_counts {
+                if evidence
+                    .get(&(first.turn.clone(), text.clone()))
+                    .is_some_and(|items| items.len() != *expected)
+                {
+                    bail!("重复文本用户事件数量与响应不一致，无法可靠匹配");
+                }
+            }
+        }
+        for (index, (source, _)) in candidates.iter().enumerate() {
+            let mut candidate = source.clone();
+            let key = (candidate.turn.clone(), candidate.text.clone());
+            let next = candidates[index + 1..]
+                .iter()
+                .find(|(c, _)| c.text == candidate.text)
+                .map(|(c, _)| c.ordinal);
+            if let Some(items) = projections.get_mut(&key) {
+                // 每条 response 只匹配其后、下一次同文 response 之前的完成事件。
+                let matches = items
+                    .iter()
+                    .filter(|(_, ordinal)| {
+                        *ordinal > candidate.ordinal && next.is_none_or(|next| *ordinal < next)
+                    })
+                    .count();
+                if matches > 1 {
+                    bail!("同一用户响应对应多个完成事件，无法可靠确认原生消息身份");
+                }
+                if let Some(index) = items.iter().position(|(_, ordinal)| {
+                    *ordinal > candidate.ordinal && next.is_none_or(|next| *ordinal < next)
+                }) {
+                    candidate.projection = Some(items.remove(index));
+                }
+            }
+            let matched_event = if let Some(items) = evidence.get_mut(&key) {
+                // 独立 user_message 与同文 response 按各自出现次数一一对应；
+                // 事件在 response 前后的历史格式都可匹配，但任何事件只消费一次。
+                items.remove(0);
+                true
+            } else {
+                false
+            };
+            if candidate.projection.is_some() || matched_event {
+                result.push(candidate);
+            }
+        }
+    }
+    for (turn, boundary) in &mut boundaries {
+        boundary.first_user_projection = first_user_projections.get(turn).cloned();
+        boundary.first_user_response_ordinal = first_user_responses.get(turn).copied();
+        if boundary
+            .first_user_projection
+            .as_ref()
+            .is_some_and(|(_, ordinal)| {
+                *ordinal <= boundary.start_ordinal || *ordinal >= boundary.end_ordinal
+            })
+            || boundary.first_user_response_ordinal.is_some_and(|ordinal| {
+                ordinal <= boundary.start_ordinal || ordinal >= boundary.end_ordinal
+            })
+        {
+            invalid_boundaries.insert(turn.clone());
+        }
+    }
+    for candidate in &mut result {
+        candidate.start_ordinal = start_ordinals
+            .get(&candidate.turn)
+            .filter(|ordinal| {
+                !duplicate_starts.contains(&candidate.turn) && **ordinal < candidate.ordinal
+            })
+            .copied();
+        candidate.first_user_projection = first_user_projections.get(&candidate.turn).cloned();
+        candidate.first_user_response_ordinal = first_user_responses.get(&candidate.turn).copied();
+        candidate.first_user_evidence_valid = if candidate.goal {
+            candidate.start_ordinal.is_some()
+                && !duplicate_starts.contains(&candidate.turn)
+                && !invalid_first_user_evidence.contains(&candidate.turn)
+        } else {
+            candidate
+                .first_user_response_ordinal
+                .is_some_and(|response| {
+                    !invalid_first_user_evidence.contains(&candidate.turn)
+                        && candidate.start_ordinal.is_none_or(|start| start < response)
+                        && candidate
+                            .first_user_projection
+                            .as_ref()
+                            .is_none_or(|(_, projection)| response < *projection)
+                })
+        };
+        candidate.turn_boundary = boundaries
+            .get(&candidate.turn)
+            .filter(|boundary| {
+                !invalid_boundaries.contains(&candidate.turn)
+                    && boundary.start_ordinal < candidate.ordinal
+                    && candidate.ordinal < boundary.end_ordinal
+                    && candidate.projection.as_ref().is_none_or(|(_, ordinal)| {
+                        boundary.start_ordinal < *ordinal && *ordinal < boundary.end_ordinal
+                    })
+            })
+            .cloned();
+        candidate.following_turn = following_turns
+            .get(&candidate.turn)
+            .and_then(|following| following.as_ref())
+            .filter(|(id, ordinal)| {
+                candidate.start_ordinal.is_some()
+                    && !duplicate_starts.contains(id)
+                    && id != &candidate.turn
+                    && *ordinal > candidate.ordinal
+            })
+            .cloned();
+    }
+    result.sort_by_key(|candidate| candidate.ordinal);
+    Ok(result)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+    use std::io::Write;
+
+    /// 只读现场核验，路径与目标轮次均由调用方显式提供；不输出消息原文。
+    #[test]
+    #[ignore = "需要显式指定 CODEX_INDEX_VERIFY_ROLLOUT、CODEX_INDEX_VERIFY_TURN 和 CODEX_INDEX_VERIFY_TEXT"]
+    fn verify_real_rollout_read_only() {
+        let path = std::env::var_os("CODEX_INDEX_VERIFY_ROLLOUT")
+            .expect("需要 CODEX_INDEX_VERIFY_ROLLOUT");
+        let turn = std::env::var("CODEX_INDEX_VERIFY_TURN").expect("需要 CODEX_INDEX_VERIFY_TURN");
+        let text = std::env::var("CODEX_INDEX_VERIFY_TEXT").expect("需要 CODEX_INDEX_VERIFY_TEXT");
+        assert!(!text.is_empty(), "核验原文不能为空");
+        let candidates = scan(Path::new(&path)).unwrap();
+        let found = candidates
+            .iter()
+            .any(|candidate| candidate.turn == turn && candidate.text.starts_with(&text));
+        eprintln!("candidates={}, confirmed_message={found}", candidates.len());
+        assert!(found, "未找到指定轮次的待核验原文");
+    }
+
+    fn scan_rows(rows: Vec<Value>, tail: &str) -> anyhow::Result<Vec<Candidate>> {
+        let mut file = tempfile::NamedTempFile::new().unwrap();
+        for row in rows {
+            writeln!(file, "{row}").unwrap();
+        }
+        write!(file, "{tail}").unwrap();
+        scan(file.path())
+    }
+    fn message(turn: &str, text: &str, ordinal: i64) -> Value {
+        json!({"type":"response_item","ordinal":ordinal,"timestamp":"2026-09-14T12:24:30Z", "payload":{"role":"user","content":[{"type":"input_text","text":text}],"internal_chat_message_metadata_passthrough":{"turn_id":turn}}})
+    }
+    fn start(turn: &str) -> Value {
+        json!({"type":"event_msg","payload":{"type":"task_started","turn_id":turn}})
+    }
+    fn goal(thread: &str) -> Value {
+        goal_at(thread, 100)
+    }
+    fn goal_at(thread: &str, created: i64) -> Value {
+        json!({"type":"event_msg","payload":{"type":"thread_goal_updated","threadId":thread,"goal":{"objective":"原文","createdAt":created}}})
+    }
+    fn meta() -> Value {
+        json!({"type":"session_meta","payload":{"id":"t"}})
+    }
+
+    fn start_at(turn: &str, ordinal: i64) -> Value {
+        let mut row = start(turn);
+        row["ordinal"] = json!(ordinal);
+        row
+    }
+
+    fn confirmed_message_rows() -> Vec<Value> {
+        vec![
+            meta(),
+            start_at("a", 1),
+            message("a", "原文", 2),
+            json!({"type":"event_msg","ordinal":3,"payload":{"type":"user_message","turn_id":"a","message":"原文"}}),
+        ]
+    }
+
+    #[test]
+    fn following_turn_uses_immediate_task_started_even_without_completion() {
+        for completed in [false, true] {
+            let mut rows = confirmed_message_rows();
+            if completed {
+                rows.push(json!({"type":"event_msg","ordinal":4,"payload":{"type":"task_complete","turn_id":"a"}}));
+            }
+            rows.extend([start_at("b", 5), start_at("c", 8)]);
+            let candidates = scan_rows(rows, "").unwrap();
+            assert_eq!(candidates.len(), 1);
+            assert_eq!(candidates[0].following_turn, Some(("b".into(), 5)));
+        }
+    }
+
+    #[test]
+    fn following_turn_requires_distinct_valid_boundary_after_candidate() {
+        for next in [
+            None,
+            Some(start("b")),
+            Some(start_at("b", -1)),
+            Some(start_at("b", 2)),
+            Some(start_at("a", 5)),
+            Some(
+                json!({"type":"event_msg","ordinal":5,"payload":{"type":"task_complete","turn_id":"b"}}),
+            ),
+            Some(
+                json!({"type":"event_msg","ordinal":5,"payload":{"type":"task_started","turn_id":"b","thread_id":"other"}}),
+            ),
+            Some(
+                json!({"type":"event_msg","ordinal":5,"payload":{"type":"task_started","turn_id":"b","threadId":"other"}}),
+            ),
+        ] {
+            let mut rows = confirmed_message_rows();
+            rows.extend(next);
+            let candidates = scan_rows(rows, "").unwrap();
+            assert_eq!(candidates.len(), 1);
+            assert!(candidates[0].following_turn.is_none());
+        }
+    }
+
+    #[test]
+    fn following_turn_does_not_skip_an_unverifiable_intermediate_start() {
+        for next in [
+            start("b"),
+            start_at("b", -1),
+            start_at("", 5),
+            json!({"type":"event_msg","ordinal":5,"payload":{"type":"task_started"}}),
+        ] {
+            let mut rows = confirmed_message_rows();
+            rows.extend([next, start_at("c", 8)]);
+            let candidates = scan_rows(rows, "").unwrap();
+            assert_eq!(candidates.len(), 1);
+            assert!(candidates[0].following_turn.is_none());
+        }
+    }
+
+    #[test]
+    fn following_turn_ignores_inherited_history_and_foreign_thread_events() {
+        let rows = vec![
+            json!({"type":"session_meta","ordinal":0,"payload":{"id":"t","parent_thread_id":"parent","subagent_history_start_ordinal":5}}),
+            json!({"type":"session_meta","ordinal":1,"payload":{"id":"parent"}}),
+            start_at("parent-a", 2),
+            start_at("parent-b", 3),
+            start_at("a", 5),
+            message("a", "原文", 6),
+            json!({"type":"event_msg","ordinal":7,"payload":{"type":"user_message","turn_id":"a","message":"原文"}}),
+            json!({"type":"event_msg","ordinal":8,"payload":{"type":"task_started","turn_id":"parent-c","thread_id":"parent"}}),
+            start_at("b", 9),
+        ];
+        let candidates = scan_rows(rows, "").unwrap();
+        assert_eq!(candidates.len(), 1);
+        assert_eq!(candidates[0].following_turn, Some(("b".into(), 9)));
+    }
+
+    #[test]
+    fn following_turn_is_optional_in_older_cached_candidates() {
+        let candidate: Candidate = serde_json::from_value(json!({
+            "thread":"t", "turn":"a", "text":"原文", "ordinal":2,
+            "created":1000, "goal":false,
+        }))
+        .unwrap();
+        assert!(candidate.following_turn.is_none());
+        assert!(candidate.turn_boundary.is_none());
+        assert!(candidate.start_ordinal.is_none());
+        assert!(candidate.goal_key.is_none());
+        assert!(candidate.first_user_projection.is_none());
+        assert!(candidate.first_user_response_ordinal.is_none());
+        assert!(!candidate.first_user_evidence_valid);
+    }
+
+    #[test]
+    fn following_turn_rejects_repeated_source_or_destination_starts() {
+        for extra in [start_at("a", 8), start_at("b", 8)] {
+            let mut rows = confirmed_message_rows();
+            rows.extend([start_at("b", 5), extra]);
+            let items = scan_rows(rows, "").unwrap();
+            assert!(items[0].following_turn.is_none());
+        }
+        let mut rows = confirmed_message_rows();
+        rows.push(start_at("b", 5));
+        let items = scan_rows(rows, "").unwrap();
+        assert_eq!(items[0].start_ordinal, Some(1));
+        assert_eq!(items[0].following_turn, Some(("b".into(), 5)));
+    }
+
+    #[test]
+    fn following_turn_requires_unique_source_start_ordinal() {
+        let mut rows = confirmed_message_rows();
+        rows[1].as_object_mut().unwrap().remove("ordinal");
+        rows.push(start_at("b", 5));
+        let items = scan_rows(rows, "").unwrap();
+        assert!(items[0].start_ordinal.is_none());
+        assert!(items[0].following_turn.is_none());
+    }
+
+    fn complete_boundary_rows() -> Vec<Value> {
+        let mut rows = confirmed_message_rows();
+        rows[1]["payload"]["started_at"] = json!(100);
+        rows.push(json!({"type":"event_msg","ordinal":5,"payload":{
+            "type":"task_complete","turn_id":"a","started_at":100,
+            "completed_at":105,"duration_ms":5000,
+        }}));
+        rows
+    }
+
+    #[test]
+    fn complete_turn_boundary_preserves_terminal_status_times_and_errors() {
+        for (kind, error, status) in [
+            ("task_complete", Value::Null, "completed"),
+            (
+                "task_complete",
+                json!({"message":"failure","codex_error_info":"other"}),
+                "failed",
+            ),
+            ("task_failed", Value::Null, "failed"),
+            ("turn_aborted", Value::Null, "interrupted"),
+        ] {
+            let mut rows = complete_boundary_rows();
+            rows[4]["payload"]["type"] = json!(kind);
+            rows[4]["payload"]["error"] = error.clone();
+            let items = scan_rows(rows, "").unwrap();
+            let boundary = items[0].turn_boundary.as_ref().unwrap();
+            assert_eq!((boundary.start_ordinal, boundary.end_ordinal), (1, 5));
+            assert_eq!((boundary.started_at, boundary.completed_at), (100, 105));
+            assert_eq!(boundary.status, status);
+            assert_eq!(boundary.duration_ms, Some(5000));
+            assert_eq!(
+                boundary.error_json,
+                (!error.is_null()).then(|| error.to_string())
+            );
+        }
+    }
+
+    #[test]
+    fn complete_turn_boundary_requires_identity_ordinals_and_times() {
+        for invalid in 0..11 {
+            let mut rows = complete_boundary_rows();
+            match invalid {
+                0 => {
+                    rows[1].as_object_mut().unwrap().remove("ordinal");
+                }
+                1 => {
+                    rows[1]["payload"]
+                        .as_object_mut()
+                        .unwrap()
+                        .remove("started_at");
+                }
+                2 => {
+                    rows[4].as_object_mut().unwrap().remove("ordinal");
+                }
+                3 => {
+                    rows[4]["payload"]
+                        .as_object_mut()
+                        .unwrap()
+                        .remove("completed_at");
+                }
+                4 => rows[4]["payload"]["turn_id"] = json!("other"),
+                5 => rows[4]["payload"]["started_at"] = json!(101),
+                6 => rows[4]["payload"]["completed_at"] = json!(99),
+                7 => rows[4]["ordinal"] = json!(2),
+                8 => rows[1]["ordinal"] = json!(2),
+                9 => rows[1]["ordinal"] = json!(-1),
+                10 => rows[4]["payload"]["thread_id"] = json!("other"),
+                _ => unreachable!(),
+            }
+            let items = scan_rows(rows, "").unwrap();
+            assert_eq!(items.len(), 1);
+            assert!(items[0].turn_boundary.is_none(), "invalid case {invalid}");
+        }
+    }
+
+    #[test]
+    fn complete_turn_boundary_rejects_duplicates_interleaving_and_late_projection() {
+        for invalid in 0..6 {
+            let mut rows = complete_boundary_rows();
+            match invalid {
+                0 => rows.insert(2, rows[1].clone()),
+                1 => rows.push(rows[4].clone()),
+                2 => rows.insert(4, start_at("other", 4)),
+                3 => rows.push(start_at("a", 8)),
+                4 => rows.insert(1, json!({"type":"event_msg","ordinal":0,"payload":{"type":"task_complete","turn_id":"a","completed_at":99}})),
+                5 => rows.push(json!({"type":"event_msg","ordinal":6,"payload":{"type":"item_completed","thread_id":"t","turn_id":"a","item":{"id":"late","type":"UserMessage","content":[{"type":"text","text":"原文"}]}}})),
+                _ => unreachable!(),
+            }
+            let items = scan_rows(rows, "").unwrap();
+            assert_eq!(items.len(), 1);
+            assert!(items[0].turn_boundary.is_none(), "invalid case {invalid}");
+        }
+    }
+
+    #[test]
+    fn complete_turn_boundary_can_use_timestamp_and_unambiguous_current_turn() {
+        let mut rows = complete_boundary_rows();
+        rows[1]["payload"]
+            .as_object_mut()
+            .unwrap()
+            .remove("started_at");
+        rows[1]["timestamp"] = json!("1970-01-01T00:01:40Z");
+        rows[4]["payload"]
+            .as_object_mut()
+            .unwrap()
+            .remove("completed_at");
+        rows[4]["payload"]
+            .as_object_mut()
+            .unwrap()
+            .remove("turn_id");
+        rows[4]["timestamp"] = json!("1970-01-01T00:01:45Z");
+        let items = scan_rows(rows, "").unwrap();
+        let boundary = items[0].turn_boundary.as_ref().unwrap();
+        assert_eq!((boundary.started_at, boundary.completed_at), (100, 105));
+    }
+
+    #[test]
+    fn complete_turn_boundary_applies_to_goals_and_verified_projection() {
+        let mut rows = complete_boundary_rows();
+        rows[3] = json!({"type":"event_msg","ordinal":4,"payload":{"type":"item_completed","thread_id":"t","turn_id":"a","item":{"id":"native","type":"UserMessage","content":[{"type":"text","text":"原文"}]}}});
+        let projected = scan_rows(rows, "").unwrap();
+        assert_eq!(projected[0].projection, Some(("native".into(), 4)));
+        assert!(projected[0].turn_boundary.is_some());
+
+        let mut rows = complete_boundary_rows();
+        rows.insert(1, goal("t"));
+        rows[3] = message(
+            "a",
+            "<codex_internal_context><objective>\n原文\n</objective></codex_internal_context>",
+            2,
+        );
+        rows.remove(4);
+        rows.push(start_at("b", 6));
+        let goals = scan_rows(rows, "").unwrap();
+        assert_eq!(goals.len(), 1);
+        assert!(goals[0].goal);
+        assert!(goals[0].turn_boundary.is_some());
+        assert_eq!(goals[0].following_turn, Some(("b".into(), 6)));
+    }
+
+    #[test]
+    fn boundary_first_user_evidence_includes_attachments_before_text_steering() {
+        let mut rows = complete_boundary_rows();
+        rows[2] = message("a", "原文", 6);
+        rows[3]["ordinal"] = json!(7);
+        rows[4]["ordinal"] = json!(10);
+        rows.insert(2, json!({"type":"response_item","ordinal":2,"payload":{"role":"user","content":[{"type":"input_image","image_url":"local"}],"internal_chat_message_metadata_passthrough":{"turn_id":"a"}}}));
+        rows.insert(3, json!({"type":"event_msg","ordinal":3,"payload":{"type":"item_completed","thread_id":"t","turn_id":"a","item":{"id":"attachment","type":"UserMessage","content":[{"type":"image","image_url":"local"}]}}}));
+        let items = scan_rows(rows, "").unwrap();
+        assert_eq!(items.len(), 1);
+        let boundary = items[0].turn_boundary.as_ref().unwrap();
+        assert_eq!(
+            boundary.first_user_projection,
+            Some(("attachment".into(), 3))
+        );
+        assert_eq!(boundary.first_user_response_ordinal, Some(2));
+        assert_eq!(items[0].ordinal, 6);
+    }
+
+    #[test]
+    fn boundary_rejects_unidentified_user_projection_and_response() {
+        for invalid in 0..6 {
+            let mut rows = complete_boundary_rows();
+            let mut item = json!({"type":"event_msg","ordinal":4,"payload":{"type":"item_completed","thread_id":"t","turn_id":"a","item":{"id":"native","type":"UserMessage","content":[{"type":"image","image_url":"local"}]}}});
+            match invalid {
+                0 => {
+                    item["payload"]["item"]
+                        .as_object_mut()
+                        .unwrap()
+                        .remove("id");
+                }
+                1 => {
+                    item.as_object_mut().unwrap().remove("ordinal");
+                }
+                2 => {
+                    item["payload"].as_object_mut().unwrap().remove("turn_id");
+                }
+                3 => {
+                    item["payload"].as_object_mut().unwrap().remove("thread_id");
+                }
+                4 | 5 => {
+                    item = message("a", "附件", 4);
+                    item["payload"]["content"] =
+                        json!([{"type":"input_image","image_url":"local"}]);
+                    if invalid == 4 {
+                        item.as_object_mut().unwrap().remove("ordinal");
+                    } else {
+                        item["payload"]["internal_chat_message_metadata_passthrough"]
+                            .as_object_mut()
+                            .unwrap()
+                            .remove("turn_id");
+                    }
+                }
+                _ => unreachable!(),
+            }
+            rows.insert(4, item);
+            let items = scan_rows(rows, "").unwrap();
+            assert_eq!(items.len(), 1);
+            assert!(items[0].turn_boundary.is_none(), "invalid case {invalid}");
+        }
+    }
+
+    #[test]
+    fn boundary_first_user_response_excludes_internal_and_environment_messages() {
+        let mut rows = complete_boundary_rows();
+        rows[2]["ordinal"] = json!(6);
+        rows[3]["ordinal"] = json!(7);
+        rows[4]["ordinal"] = json!(8);
+        rows.insert(
+            2,
+            message("a", "<environment_context>环境</environment_context>", 2),
+        );
+        rows.insert(
+            3,
+            message(
+                "a",
+                "<codex_internal_context>续执行</codex_internal_context>",
+                3,
+            ),
+        );
+        let items = scan_rows(rows, "").unwrap();
+        let boundary = items[0].turn_boundary.as_ref().unwrap();
+        assert_eq!(boundary.first_user_response_ordinal, Some(6));
+    }
+
+    #[test]
+    fn goals_require_same_thread_and_first_matching_turn_and_deduplicate_continuations() {
+        let body = "<codex_internal_context source=\"goal\"><objective>\n原文\n</objective></codex_internal_context>";
+        let rows = vec![
+            meta(),
+            goal("wrong"),
+            start("a"),
+            message("a", body, 3),
+            goal("t"),
+            start("b"),
+            message("wrong", body, 6),
+            message("b", body, 7),
+            start("c"),
+            message("c", body, 9),
+            goal("t"),
+            start("d"),
+            message("d", body, 12),
+        ];
+        let items = scan_rows(rows, "").unwrap();
+        assert_eq!(items.len(), 1);
+        assert_eq!(
+            (&items[0].turn, &items[0].text, items[0].created),
+            (&"b".into(), &"原文".into(), 100000)
+        );
+    }
+
+    #[test]
+    fn repeated_goal_text_uses_goal_identity_instead_of_body() {
+        let body = "<codex_internal_context source=\"goal\"><objective>\n原文\n</objective></codex_internal_context>";
+        let rows = vec![
+            meta(),
+            goal_at("t", 100),
+            start_at("a", 2),
+            message("a", body, 3),
+            start_at("continuation", 4),
+            message("continuation", body, 5),
+            goal_at("t", 200),
+            start_at("b", 7),
+            message("b", body, 8),
+        ];
+        let items = scan_rows(rows, "").unwrap();
+        assert_eq!(items.len(), 2);
+        assert_eq!(items[0].goal_key.as_deref(), Some("created:100"));
+        assert_eq!(items[1].goal_key.as_deref(), Some("created:200"));
+        assert_eq!(items[0].text, items[1].text);
+    }
+
+    #[test]
+    fn completed_item_and_user_message_evidence_are_consumed_once() {
+        let completed = vec![
+            meta(),
+            start_at("a", 1),
+            message("a", "重复", 2),
+            message("a", "重复", 3),
+            json!({"type":"event_msg","ordinal":4,"payload":{"type":"item_completed","thread_id":"t","turn_id":"a","item":{"id":"one","type":"UserMessage","content":[{"type":"text","text":"重复"}]}}}),
+        ];
+        let items = scan_rows(completed, "").unwrap();
+        assert_eq!(items.len(), 1);
+        assert_eq!(items[0].ordinal, 3);
+        assert_eq!(items[0].projection, Some(("one".into(), 4)));
+
+        let event = vec![
+            meta(),
+            start_at("a", 1),
+            message("a", "重复", 2),
+            json!({"type":"event_msg","payload":{"type":"user_message","turn_id":"a","message":"重复"}}),
+            message("a", "重复", 4),
+        ];
+        let error = scan_rows(event, "").unwrap_err();
+        assert!(
+            error
+                .to_string()
+                .contains("重复文本用户事件数量与响应不一致")
+        );
+
+        let balanced = vec![
+            meta(),
+            start_at("a", 1),
+            json!({"type":"event_msg","payload":{"type":"user_message","turn_id":"a","message":"重复"}}),
+            message("a", "重复", 3),
+            message("a", "重复", 4),
+            json!({"type":"event_msg","payload":{"type":"user_message","turn_id":"a","message":"重复"}}),
+        ];
+        let items = scan_rows(balanced, "").unwrap();
+        assert_eq!(items.len(), 2);
+    }
+
+    #[test]
+    fn multiple_completions_for_one_response_require_review() {
+        let mut rows = confirmed_message_rows();
+        for (id, ordinal) in [("native-a", 4), ("native-b", 5)] {
+            rows.push(json!({"type":"event_msg","ordinal":ordinal,"payload":{"type":"item_completed","thread_id":"t","turn_id":"a","item":{"id":id,"type":"UserMessage","content":[{"type":"text","text":"原文"}]}}}));
+        }
+        assert!(
+            scan_rows(rows, "")
+                .unwrap_err()
+                .to_string()
+                .contains("多个完成事件")
+        );
+    }
+
+    #[test]
+    fn first_user_evidence_survives_missing_terminal_boundary() {
+        let rows = vec![
+            meta(),
+            start_at("a", 1),
+            json!({"type":"response_item","ordinal":2,"payload":{"role":"user","content":[{"type":"input_image","image_url":"local"}],"internal_chat_message_metadata_passthrough":{"turn_id":"a"}}}),
+            json!({"type":"event_msg","ordinal":3,"payload":{"type":"item_completed","thread_id":"t","turn_id":"a","item":{"id":"attachment","type":"UserMessage","content":[{"type":"image","image_url":"local"}]}}}),
+            message("a", "后续文本", 4),
+            json!({"type":"event_msg","ordinal":5,"payload":{"type":"user_message","turn_id":"a","message":"后续文本"}}),
+        ];
+        let items = scan_rows(rows, "").unwrap();
+        assert_eq!(items.len(), 1);
+        assert!(items[0].turn_boundary.is_none());
+        assert_eq!(
+            items[0].first_user_projection,
+            Some(("attachment".into(), 3))
+        );
+        assert_eq!(items[0].first_user_response_ordinal, Some(2));
+        assert!(items[0].first_user_evidence_valid);
+    }
+
+    #[test]
+    fn missing_first_user_identity_or_ordinal_marks_evidence_invalid() {
+        for invalid in 0..2 {
+            let mut first = json!({"type":"response_item","ordinal":2,"payload":{"role":"user","content":[{"type":"input_image","image_url":"local"}],"internal_chat_message_metadata_passthrough":{"turn_id":"a"}}});
+            if invalid == 0 {
+                first.as_object_mut().unwrap().remove("ordinal");
+            } else {
+                first["payload"]["internal_chat_message_metadata_passthrough"] = json!({});
+            }
+            let rows = vec![
+                meta(),
+                start_at("a", 1),
+                first,
+                message("a", "后续文本", 4),
+                json!({"type":"event_msg","ordinal":5,"payload":{"type":"user_message","turn_id":"a","message":"后续文本"}}),
+            ];
+            let items = scan_rows(rows, "").unwrap();
+            assert_eq!(items.len(), 1);
+            assert!(
+                !items[0].first_user_evidence_valid,
+                "invalid case {invalid}"
+            );
+        }
+    }
+
+    #[test]
+    fn ordinary_messages_need_corroboration_in_same_turn_and_exclude_environment() {
+        let rows = vec![
+            meta(),
+            start("a"),
+            message("a", "第一条", 2),
+            json!({"type":"event_msg","ordinal":3,"payload":{"type":"item_completed","thread_id":"t","turn_id":"a","item":{"id":"first","type":"UserMessage","content":[{"type":"text","text":"第一条"}]}}}),
+            start("b"),
+            message("b", "第二条", 5),
+            json!({"type":"event_msg","payload":{"type":"user_message","message":"第二条"}}),
+            message("b", "无证据", 7),
+            message("a", "第二条", 8),
+            message("b", "<environment_context>环境</environment_context>", 9),
+            json!({"type":"event_msg","payload":{"type":"user_message","message":"<environment_context>环境</environment_context>"}}),
+        ];
+        let items = scan_rows(rows, "").unwrap();
+        assert_eq!(
+            items.iter().map(|c| c.text.as_str()).collect::<Vec<_>>(),
+            ["第一条", "第二条"]
+        );
+    }
+
+    #[test]
+    fn fork_scans_own_messages_without_inheriting_parent_turns() {
+        let rows = vec![
+            json!({"type":"session_meta","ordinal":0,"payload":{"id":"t","parent_thread_id":"parent","forked_from_id":"parent","subagent_history_start_ordinal":5}}),
+            json!({"type":"session_meta","ordinal":1,"payload":{"id":"parent"}}),
+            message("parent-turn", "父任务指令", 2),
+            json!({"type":"event_msg","ordinal":3,"payload":{"type":"user_message","turn_id":"parent-turn","message":"父任务指令"}}),
+            json!({"type":"event_msg","ordinal":4,"payload":{"type":"task_started","turn_id":"parent-turn"}}),
+            message("child-turn", "子任务指令", 5),
+            json!({"type":"event_msg","ordinal":6,"payload":{"type":"user_message","turn_id":"child-turn","message":"子任务指令"}}),
+        ];
+        let items = scan_rows(rows.clone(), "").unwrap();
+        assert_eq!(items.len(), 1);
+        assert_eq!(
+            (&items[0].thread[..], &items[0].turn[..], &items[0].text[..]),
+            ("t", "child-turn", "子任务指令")
+        );
+
+        let mut foreign = rows.clone();
+        foreign[1]["payload"]["id"] = json!("unrelated");
+        assert!(scan_rows(foreign, "").is_err());
+        let mut outside_boundary = rows.clone();
+        outside_boundary[1]["ordinal"] = json!(5);
+        assert!(scan_rows(outside_boundary, "").is_err());
+        let mut no_boundary = rows.clone();
+        no_boundary[0]["payload"]
+            .as_object_mut()
+            .unwrap()
+            .remove("subagent_history_start_ordinal");
+        assert!(scan_rows(no_boundary, "").is_err());
+        let mut no_ordinal = rows;
+        no_ordinal[2].as_object_mut().unwrap().remove("ordinal");
+        assert!(scan_rows(no_ordinal, "").is_err());
+    }
+
+    #[test]
+    fn rejects_mixed_sessions_and_nonterminal_corruption() {
+        assert!(
+            scan_rows(
+                vec![
+                    meta(),
+                    json!({"type":"session_meta","payload":{"id":"other"}})
+                ],
+                ""
+            )
+            .is_err()
+        );
+        assert!(scan_rows(vec![meta()], "bad\n").is_err());
+        let error = scan_rows(vec![meta()], "{\"type\":").unwrap_err();
+        assert!(error.to_string().contains("会话记录尾行不完整"));
+    }
+
+    #[test]
+    fn missing_identity_or_ordinal_and_partial_attachments_are_not_recovered() {
+        let mut missing_ordinal = message("a", "原文", 2);
+        missing_ordinal.as_object_mut().unwrap().remove("ordinal");
+        let mut missing_turn = message("a", "原文", 3);
+        missing_turn["payload"]["internal_chat_message_metadata_passthrough"] = json!({});
+        let mut attachment = message("a", "原文", 4);
+        attachment["payload"]["content"]
+            .as_array_mut()
+            .unwrap()
+            .push(json!({"type":"input_image","image_url":"file:///image.png"}));
+        let rows = vec![
+            meta(),
+            start("a"),
+            missing_ordinal,
+            missing_turn,
+            attachment,
+            json!({"type":"event_msg","payload":{"type":"user_message","message":"原文"}}),
+        ];
+        assert!(scan_rows(rows, "").unwrap().is_empty());
+    }
+}
diff --git a/docs/session-index-repair.md b/docs/session-index-repair.md
new file mode 100644
index 000000000..0611cca2c
--- /dev/null
+++ b/docs/session-index-repair.md
@@ -0,0 +1,23 @@
+# 会话索引修复
+
+管理工具 → 会话管理 → “修复会话索引”（位于“修复历史会话”右侧）。无需指定会话或记住缺失内容。操作完成后页面显示读取文件、缓存命中、恢复数量、跳过原因、耗时和备份位置。重新打开会话可读取修复后的原生消息流。
+
+报告先核验消息是否已存在，轮次残留 inProgress 不再把已显示消息计入等待。缺少原生轮次、暂时不允许写入、文件正在变化才列为“短暂等待”。持续等待至少 30 分钟，或消息及文件均已超过 24 小时未更新，则列入“需核查”，仍在后续检查中重新核验，不永久忽略。详情包含任务和轮次 ID、原因、首次发现和复查次数；解决后清除等待计时。计数按候选消息或失败文件计算，不是任务数量。
+
+页面显示最后检查时间，并每 15 秒读取后台报告。读取报告不会启动修复，自动检查仍需要启动器运行及已保存的开关开启。活跃任务的新缺失目标可短暂出现，不能承诺使用期间永久为零。
+
+“启动前自动修复历史会话”开启并保存后，启动器在启动前执行检查，并在 Codex 运行期间每次检查完成后间隔 60 秒再次检查。每次读取已保存开关；关闭后停止后续自动检查。正在进行的事务会正常结束。检查失败记录诊断，不因消息修复失败阻止启动。
+
+扫描范围包括当前 CODEX_HOME 的 sessions 和 archived_sessions。普通纯文本消息需要原始用户事件与 response_item 原文、轮次相互印证；目标消息需要目标事件与首轮内部上下文原文一致。恢复使用 thread_history_1.sqlite 的原生 userMessage 投影，不创建新任务、不发送新指令、不修改 rollout。
+
+旧 inProgress 只有原始日志与原生数据库都证明后继轮次已经开始才允许补消息，不修改原有执行状态。顺序明确的目标可以补到已有问答回复前并修正首条指针，完整保留卡片中的模型提问和用户选项回复。分片归属需核对原生轮次、完成事件 ID、位置与全文，不能仅凭文件名或相同正文映射。原生轮次缺失时，仅在任务归属可信、完整开始/结束事件及首条消息身份均确定时补建历史轮次投影；不推进原生投影游标。
+
+已有不同消息、混合附件、无法确认位置、未知数据库格式和正在变化的文件不会被猜测恢复。该工具不能恢复从未保存或已彻底删除的原文，也不保证修复所有形式的历史损坏。供应商和任务目录问题继续使用“修复历史会话”。
+
+即使原生轮次已经存在，补入或重新连接首条用户消息也必须确认原始首条身份。首条包含不支持的附件时，后续纯文字不能取代它；悬空指针只能连接同一原生消息 ID。位置冲突按整个任务核验，与原生数据库唯一索引一致，冲突项进入核查，不阻止其他安全项修复。
+
+普通消息的独立用户事件与完成事件按记录顺序逐个核验、一次使用；相同正文不代表同一次发送。目标按目标身份及创建时间区分，自动续执行沿用同目标，同文新目标保留独立记录。尾行 JSON 不完整时不缓存成功结果：先等待后续写入，持续未完成或超过 24 小时未更新则明确需核查。
+
+同一响应区间出现多个完成事件时，无法唯一确认原生消息身份，明确进入核查。同一目标跨分片重复出现时，优先保留已在原生索引中的分片，包括旧版本的恢复记录，避免升级后重复补入。
+
+性能：按文件大小和修改时间缓存解析结果，未变化文件不重复读取正文；每次仍核对原生索引，因此原生索引重建后也能重新恢复。变化文件采用流式解析，首次检查时间取决于全部历史记录大小。缓存、最近报告和每次发生修复前的数据库备份存放在 CODEX_HOME/session-index-repair。备份可能较大，仅发生实际待修复项时创建，不会在无变化检查时生成。