use anyhow::{Context, Result, bail};
use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
use serde_json::{Value, json};
use std::{
collections::{BTreeMap, HashSet},
fs,
path::{Path, PathBuf},
time::{SystemTime, UNIX_EPOCH},
};
use walkdir::WalkDir;
mod model;
mod output;
mod storage;
use model::{
Assignment, IndexDiagnostics, RolloutDiagnostics, RolloutRecovery, SessionMetaRepair,
SessionRecord, SidebarThread,
};
pub use model::{DoctorReport, Paths, Report};
pub use output::{print_doctor_report, print_report};
use storage::{atomic_copy, atomic_write, copy_database, open_database};
const MAX_RECOVERY_FILE_SIZE: u64 = 1024 * 1024 * 1024;
impl Paths {
pub fn resolve(codex_home: Option<PathBuf>) -> Result<Self> {
let home = dirs::home_dir().context("无法确定用户主目录")?;
let codex_home = codex_home.unwrap_or_else(|| home.join(".codex"));
let paths = Self {
config: codex_home.join("config.toml"),
database: codex_home.join("state_5.sqlite"),
session_index: codex_home.join("session_index.jsonl"),
backup_root: codex_home.join("codex-sync-backups"),
codex_home,
};
paths.validate()?;
Ok(paths)
}
fn validate(&self) -> Result<()> {
if !self.codex_home.is_dir() {
bail!("Codex 数据目录不存在：{}", self.codex_home.display());
}
if !self.config.is_file() {
bail!("Codex 配置不存在：{}", self.config.display());
}
if !self.database.is_file() {
bail!("Codex 历史数据库不存在：{}", self.database.display());
}
Ok(())
}
}
pub fn inspect(paths: &Paths) -> Result<Report> {
let target = active_assignment(paths)?;
let records = session_records(paths)?;
let mut rollout_counts = BTreeMap::new();
let mut rollout_model_counts = BTreeMap::new();
let mut rollout_files_to_change = 0;
let mut duplicate_session_meta_lines = 0;
for record in &records {
*rollout_counts
.entry(display_value(&record.provider))
.or_insert(0) += 1;
*rollout_model_counts
.entry(display_value(record.model.as_deref().unwrap_or("")))
.or_insert(0) += 1;
if assignment_differs(&record.provider, record.model.as_deref(), &target) {
rollout_files_to_change += 1;
}
duplicate_session_meta_lines += record.session_meta_lines.saturating_sub(1);
}
let conn = open_database(&paths.database, true)?;
let columns = thread_columns(&conn)?;
if !columns.contains("model_provider") {
bail!("threads 表缺少 model_provider 字段，无法安全归并历史");
}
let database_counts = grouped_counts(&conn, "model_provider")?;
let model_counts = if columns.contains("model") {
grouped_counts(&conn, "model")?
} else {
BTreeMap::new()
};
let total_threads = conn.query_row("SELECT COUNT(*) FROM threads", [], |row| row.get(0))?;
let database_rows_to_change = count_database_changes(&conn, &columns, &target)?;
let sidebar_threads = sidebar_threads(&conn, &columns)?;
let missing_rollout_files = sidebar_threads
.iter()
.filter(|thread| !rollout_exists(&paths.codex_home, thread.rollout_path.as_deref()))
.count();
let database_ids = indexable_ids(&paths.codex_home, &sidebar_threads);
let index_ids = read_session_index(&paths.session_index)?
.into_keys()
.collect::<HashSet<_>>();
Ok(Report {
current_provider: target.provider,
current_model: target.model,
total_threads,
database_counts,
model_counts,
database_rows_to_change,
rollout_counts,
rollout_model_counts,
rollout_files: records.len(),
rollout_files_to_change,
duplicate_session_meta_lines,
missing_session_index_entries: database_ids.difference(&index_ids).count(),
stale_session_index_entries: index_ids.difference(&database_ids).count(),
missing_rollout_files,
backup: None,
})
}
pub fn doctor(
paths: &Paths,
apply: bool,
restore_missing: &[PathBuf],
dedupe_session_meta: Option<&Path>,
) -> Result<DoctorReport> {
let report = inspect_doctor(paths)?;
if !apply {
if !restore_missing.is_empty() || dedupe_session_meta.is_some() {
bail!("恢复或去重参数必须与 --apply 一起使用");
}
return Ok(report);
}
if !report.database_integrity_ok {
bail!("数据库完整性检查失败；doctor 未修改任何数据");
}
if report.invalid_rollout_files != 0
|| report.missing_session_meta_files != 0
|| report.invalid_session_index_lines != 0
{
bail!("存在无法安全自动修复的 JSONL；doctor 未修改任何数据");
}
let recoveries = prepare_rollout_recoveries(paths, restore_missing)?;
let session_repair = dedupe_session_meta
.map(|path| prepare_session_meta_repair(paths, path))
.transpose()?;
let needs_index_repair = report.missing_session_index_entries != 0
|| report.stale_session_index_entries != 0
|| report.duplicate_session_index_ids != 0;
if !needs_index_repair && recoveries.is_empty() && session_repair.is_none() {
return Ok(report);
}
let original = fs::read(&paths.session_index).ok();
let backup = backup_doctor_state(paths, &report, original.as_deref(), session_repair.as_ref())?;
let mut installed = Vec::new();
let mut session_replaced = false;
let index_touched = needs_index_repair || !recoveries.is_empty();
let repair = (|| -> Result<DoctorReport> {
let mut connection = open_database(&paths.database, false)?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
for recovery in &recoveries {
if recovery.target.exists() {
bail!(
"待恢复的 rollout 已被其他进程创建：{}",
recovery.target.display()
);
}
atomic_write(&recovery.target, &recovery.content)?;
installed.push(recovery.target.clone());
}
if let Some(repair) = &session_repair {
if fs::read(&repair.target)? != repair.original {
bail!(
"rollout 在诊断后发生变化，拒绝覆盖：{}",
repair.target.display()
);
}
atomic_write(&repair.target, &repair.repaired)?;
session_replaced = true;
validate_recovery_rollout(&repair.target, &repair.repaired)?;
}
if index_touched {
let content = render_session_index(paths, &transaction)?;
atomic_write(&paths.session_index, &content)?;
}
let verified = inspect_doctor(paths)?;
if (index_touched
&& (verified.invalid_session_index_lines != 0
|| verified.duplicate_session_index_ids != 0
|| verified.missing_session_index_entries != 0
|| verified.stale_session_index_entries != 0))
|| verified.missing_rollout_files + recoveries.len() != report.missing_rollout_files
{
bail!("rollout/侧栏索引原子替换后的复检失败");
}
transaction.commit()?;
Ok(verified)
})();
match repair {
Ok(mut verified) => {
verified.repaired = true;
verified.repaired_session_index = index_touched;
verified.restored_rollout_files = recoveries.len();
verified.deduplicated_session_meta_files = usize::from(session_repair.is_some());
verified.backup = Some(backup);
Ok(verified)
}
Err(error) => {
for target in installed.iter().rev() {
let _ = fs::remove_file(target);
}
if session_replaced && let Some(repair) = &session_repair {
atomic_write(&repair.target, &repair.original)
.context("修复失败，且恢复原 rollout 失败")?;
}
if index_touched {
restore_original_index(paths, original.as_deref())
.context("修复失败，且恢复原侧栏索引失败")?;
}
Err(error.context("修复失败；已原子恢复所有已替换文件"))
}
}
}
fn inspect_doctor(paths: &Paths) -> Result<DoctorReport> {
paths.validate()?;
let connection = open_database(&paths.database, true)?;
let database_integrity_messages = database_integrity(&connection)?;
let database_integrity_ok = database_integrity_messages.as_slice() == ["ok"];
let columns = thread_columns(&connection)?;
let total_threads =
connection.query_row("SELECT COUNT(*) FROM threads", [], |row| row.get(0))?;
let sidebar = sidebar_threads(&connection, &columns)?;
let database_ids = all_database_ids(&connection)?;
let indexable = indexable_ids(&paths.codex_home, &sidebar);
let missing_rollout_files = sidebar
.iter()
.filter(|thread| !rollout_exists(&paths.codex_home, thread.rollout_path.as_deref()))
.count();
let index = inspect_session_index(&paths.session_index)?;
let index_ids = index.entries.keys().cloned().collect::<HashSet<_>>();
let rollouts = inspect_rollouts(paths)?;
let orphan_rollout_files = rollouts
.thread_ids
.iter()
.filter(|id| !database_ids.contains(*id))
.count();
let mut report = DoctorReport {
database_integrity_ok,
database_integrity_messages,
total_threads,
sidebar_threads: sidebar.len(),
rollout_files: rollouts.files,
invalid_rollout_files: rollouts.invalid_files,
missing_session_meta_files: rollouts.missing_session_meta_files,
duplicate_session_meta_lines: rollouts.duplicate_session_meta_lines,
orphan_rollout_files,
missing_rollout_files,
invalid_session_index_lines: index.invalid_lines,
duplicate_session_index_ids: index.duplicate_ids,
missing_session_index_entries: indexable.difference(&index_ids).count(),
stale_session_index_entries: index_ids.difference(&indexable).count(),
clean: false,
repaired: false,
repaired_session_index: false,
restored_rollout_files: 0,
deduplicated_session_meta_files: 0,
backup: None,
};
report.clean = doctor_is_clean(&report);
Ok(report)
}
fn doctor_is_clean(report: &DoctorReport) -> bool {
report.database_integrity_ok
&& report.invalid_rollout_files == 0
&& report.missing_session_meta_files == 0
&& report.duplicate_session_meta_lines == 0
&& report.orphan_rollout_files == 0
&& report.missing_rollout_files == 0
&& report.invalid_session_index_lines == 0
&& report.duplicate_session_index_ids == 0
&& report.missing_session_index_entries == 0
&& report.stale_session_index_entries == 0
}
pub fn merge(paths: &Paths, apply: bool) -> Result<Report> {
let mut report = inspect(paths)?;
let needs_change = report.database_rows_to_change != 0
|| report.rollout_files_to_change != 0
|| report.duplicate_session_meta_lines != 0
|| report.missing_session_index_entries != 0
|| report.stale_session_index_entries != 0;
if !apply || !needs_change {
return Ok(report);
}
let target = active_assignment(paths)?;
let backup_path = backup(paths, "history-merge")?;
update_database(paths, &target)?;
rewrite_sessions(paths, &target)?;
rebuild_session_index(paths)?;
let verified = inspect(paths)?;
if verified.database_rows_to_change != 0
|| verified.rollout_files_to_change != 0
|| verified.duplicate_session_meta_lines != 0
|| verified.missing_session_index_entries != 0
|| verified.stale_session_index_entries != 0
{
bail!(
"历史归并后验证失败；修改前备份位于 {}，请先恢复备份",
backup_path.display()
);
}
report.backup = Some(backup_path);
Ok(report)
}
pub fn backup(paths: &Paths, label: &str) -> Result<PathBuf> {
paths.validate()?;
fs::create_dir_all(&paths.backup_root)?;
let stamp = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
let destination = paths.backup_root.join(format!("{label}-{stamp}"));
fs::create_dir_all(&destination)?;
copy_database(&paths.database, &destination.join("state_5.sqlite"))?;
if paths.session_index.is_file() {
fs::copy(
&paths.session_index,
destination.join("session_index.jsonl"),
)?;
}
for source in rollout_paths(paths) {
let relative = source.strip_prefix(&paths.codex_home)?;
let target = destination.join(relative);
if let Some(parent) = target.parent() {
fs::create_dir_all(parent)?;
}
fs::copy(source, target)?;
}
Ok(destination)
}
pub fn validate_backup(path: &Path) -> Result<()> {
if !path.is_dir() {
bail!("历史备份目录不存在：{}", path.display());
}
if !path.join("state_5.sqlite").is_file() {
bail!("备份缺少 state_5.sqlite：{}", path.display());
}
Ok(())
}
pub fn restore(paths: &Paths, source: &Path) -> Result<PathBuf> {
validate_backup(source)?;
let safety = backup(paths, "pre-history-restore")?;
copy_database(&source.join("state_5.sqlite"), &paths.database)?;
let index = source.join("session_index.jsonl");
if index.is_file() {
atomic_copy(&index, &paths.session_index)?;
}
for directory in ["sessions", "archived_sessions"] {
let root = source.join(directory);
if !root.is_dir() {
continue;
}
for entry in WalkDir::new(&root).follow_links(false) {
let entry = entry?;
if !entry.file_type().is_file() {
continue;
}
let relative = entry.path().strip_prefix(source)?;
atomic_copy(entry.path(), &paths.codex_home.join(relative))?;
}
}
Ok(safety)
}
pub fn reconcile_session_index(codex_home: &Path) -> Result<usize> {
let paths = Paths {
config: codex_home.join("config.toml"),
database: codex_home.join("state_5.sqlite"),
session_index: codex_home.join("session_index.jsonl"),
backup_root: codex_home.join("codex-sync-backups"),
codex_home: codex_home.to_path_buf(),
};
if !paths.database.is_file() {
return Ok(0);
}
let conn = open_database(&paths.database, true)?;
let columns = thread_columns(&conn)?;
let database_ids = indexable_ids(&paths.codex_home, &sidebar_threads(&conn, &columns)?);
let index_ids = read_session_index(&paths.session_index)?
.into_keys()
.collect::<HashSet<_>>();
let missing = database_ids.difference(&index_ids).count();
drop(conn);
rebuild_session_index(&paths)?;
Ok(missing)
}
fn active_assignment(paths: &Paths) -> Result<Assignment> {
let text = fs::read_to_string(&paths.config).context("无法读取 Codex config.toml")?;
let value: toml::Value = toml::from_str(&text).context("Codex config.toml 格式错误")?;
let provider = value
.get("model_provider")
.and_then(toml::Value::as_str)
.map(str::to_owned)
.unwrap_or_else(|| "openai".to_owned());
let model = value
.get("model")
.and_then(toml::Value::as_str)
.map(str::to_owned);
Ok(Assignment { provider, model })
}
fn assignment_differs(provider: &str, model: Option<&str>, target: &Assignment) -> bool {
provider != target.provider
|| target
.model
.as_deref()
.is_some_and(|expected| model != Some(expected))
}
fn display_value(value: &str) -> String {
if value.is_empty() {
"(empty)".into()
} else {
value.into()
}
}
fn database_integrity(connection: &Connection) -> Result<Vec<String>> {
let mut statement = connection.prepare("PRAGMA quick_check")?;
Ok(statement
.query_map([], |row| row.get(0))?
.collect::<rusqlite::Result<Vec<_>>>()?)
}
fn all_database_ids(connection: &Connection) -> Result<HashSet<String>> {
let mut statement = connection.prepare("SELECT id FROM threads")?;
Ok(statement
.query_map([], |row| row.get(0))?
.collect::<rusqlite::Result<HashSet<_>>>()?)
}
fn thread_columns(conn: &Connection) -> Result<HashSet<String>> {
let mut statement = conn.prepare("PRAGMA table_info(threads)")?;
let names = statement
.query_map([], |row| row.get::<_, String>(1))?
.collect::<rusqlite::Result<HashSet<_>>>()?;
Ok(names)
}
fn grouped_counts(conn: &Connection, column: &str) -> Result<BTreeMap<String, usize>> {
let sql = format!(
"SELECT COALESCE({column}, ''), COUNT(*) FROM threads GROUP BY {column} ORDER BY {column}"
);
let mut statement = conn.prepare(&sql)?;
let rows = statement.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, usize>(1)?))
})?;
let mut counts = BTreeMap::new();
for row in rows {
let (value, count) = row?;
counts.insert(display_value(&value), count);
}
Ok(counts)
}
fn count_database_changes(
conn: &Connection,
columns: &HashSet<String>,
target: &Assignment,
) -> Result<usize> {
if columns.contains("model")
&& let Some(model) = &target.model
{
return Ok(conn.query_row(
"SELECT COUNT(*) FROM threads WHERE model_provider IS NULL OR model_provider<>?1 OR model IS NULL OR model<>?2",
params![target.provider, model],
|row| row.get(0),
)?);
}
Ok(conn.query_row(
"SELECT COUNT(*) FROM threads WHERE model_provider IS NULL OR model_provider<>?1",
[&target.provider],
|row| row.get(0),
)?)
}
fn sidebar_threads(conn: &Connection, columns: &HashSet<String>) -> Result<Vec<SidebarThread>> {
let title = if columns.contains("title") {
"COALESCE(title, id)"
} else {
"id"
};
let updated = if columns.contains("updated_at") {
"COALESCE(strftime('%Y-%m-%dT%H:%M:%SZ', CASE WHEN updated_at>100000000000 THEN updated_at/1000 ELSE updated_at END, 'unixepoch'), '')"
} else {
"''"
};
let rollout_path = if columns.contains("rollout_path") {
"rollout_path"
} else {
"NULL"
};
let source = if columns.contains("source") {
"COALESCE(source, '')"
} else {
"''"
};
let thread_source = if columns.contains("thread_source") {
"COALESCE(thread_source, '')"
} else {
"''"
};
let model = if columns.contains("model") {
"COALESCE(model, '')"
} else {
"''"
};
let archived = if columns.contains("archived") {
" WHERE archived=0"
} else {
""
};
let sql = format!(
"SELECT id, {title}, {updated}, {rollout_path}, {source}, {thread_source}, {model} FROM threads{archived}"
);
let mut statement = conn.prepare(&sql)?;
let rows = statement.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, String>(4)?,
row.get::<_, String>(5)?,
row.get::<_, String>(6)?,
))
})?;
let mut threads = Vec::new();
for row in rows {
let (id, title, updated_at, rollout_path, source, thread_source, model) = row?;
if is_technical_thread(&source, &thread_source, &model, &title) {
continue;
}
threads.push(SidebarThread {
id,
title,
updated_at,
rollout_path: rollout_path.map(PathBuf::from),
});
}
Ok(threads)
}
fn is_subagent_source(source: &str) -> bool {
serde_json::from_str::<Value>(source)
.ok()
.is_some_and(|value| value.get("subagent").is_some())
}
fn is_technical_thread(source: &str, thread_source: &str, model: &str, title: &str) -> bool {
is_subagent_source(source)
|| thread_source == "subagent"
|| (model == "codex-auto-review" && title.trim().is_empty())
}
fn rollout_exists(codex_home: &Path, rollout_path: Option<&Path>) -> bool {
let Some(path) = rollout_path else {
return true;
};
if path.is_absolute() {
path.is_file()
} else {
codex_home.join(path).is_file()
}
}
fn indexable_ids(codex_home: &Path, threads: &[SidebarThread]) -> HashSet<String> {
threads
.iter()
.filter(|thread| rollout_exists(codex_home, thread.rollout_path.as_deref()))
.map(|thread| thread.id.clone())
.collect()
}
fn rollout_paths(paths: &Paths) -> Vec<PathBuf> {
["sessions", "archived_sessions"]
.into_iter()
.flat_map(|directory| {
WalkDir::new(paths.codex_home.join(directory))
.follow_links(false)
.into_iter()
.filter_map(Result::ok)
.filter(|entry| {
entry.file_type().is_file()
&& entry
.path()
.extension()
.is_some_and(|value| value == "jsonl")
})
.map(|entry| entry.into_path())
})
.collect()
}
fn inspect_rollouts(paths: &Paths) -> Result<RolloutDiagnostics> {
let mut diagnostics = RolloutDiagnostics::default();
for path in rollout_paths(paths) {
diagnostics.files += 1;
let text = fs::read_to_string(&path)
.with_context(|| format!("无法读取会话文件：{}", path.display()))?;
let mut invalid = false;
let mut session_meta_lines = 0usize;
let mut first_id = None;
for line in text.lines().filter(|line| !line.trim().is_empty()) {
let value: Value = match serde_json::from_str(line) {
Ok(value) => value,
Err(_) => {
invalid = true;
continue;
}
};
if value.get("type").and_then(Value::as_str) != Some("session_meta") {
continue;
}
session_meta_lines += 1;
if first_id.is_none() {
first_id = value
.get("payload")
.and_then(|payload| payload.get("id"))
.and_then(Value::as_str)
.map(str::to_owned);
}
}
diagnostics.invalid_files += usize::from(invalid);
diagnostics.missing_session_meta_files += usize::from(session_meta_lines == 0);
diagnostics.duplicate_session_meta_lines += session_meta_lines.saturating_sub(1);
if let Some(id) = first_id {
diagnostics.thread_ids.push(id);
}
}
Ok(diagnostics)
}
fn prepare_rollout_recoveries(paths: &Paths, sources: &[PathBuf]) -> Result<Vec<RolloutRecovery>> {
if sources.is_empty() {
return Ok(Vec::new());
}
let connection = open_database(&paths.database, true)?;
let columns = thread_columns(&connection)?;
let missing = sidebar_threads(&connection, &columns)?
.into_iter()
.filter(|thread| !rollout_exists(&paths.codex_home, thread.rollout_path.as_deref()))
.collect::<Vec<_>>();
let mut targets = HashSet::new();
let mut recoveries = Vec::new();
for source in sources {
let metadata = fs::symlink_metadata(source)
.with_context(|| format!("恢复来源不存在：{}", source.display()))?;
if !metadata.file_type().is_file() || metadata.file_type().is_symlink() {
bail!("恢复来源不是普通文件：{}", source.display());
}
if metadata.len() > MAX_RECOVERY_FILE_SIZE {
bail!("恢复来源超过大小上限：{}", source.display());
}
let content = fs::read(source)?;
let thread_id = validate_recovery_rollout(source, &content)?;
let thread = missing
.iter()
.find(|thread| thread.id == thread_id)
.with_context(|| format!("恢复来源不对应任何缺失线程：{}", source.display()))?;
let relative = thread
.rollout_path
.as_deref()
.context("缺失线程没有 rollout_path，无法自动恢复")?;
let target = if relative.is_absolute() {
relative.to_path_buf()
} else {
paths.codex_home.join(relative)
};
ensure_safe_recovery_target(&paths.codex_home, &target)?;
if source.file_name() != target.file_name() {
bail!("恢复来源文件名与数据库 rollout_path 不一致");
}
if !targets.insert(target.clone()) {
bail!("同一缺失 rollout 被重复指定：{}", target.display());
}
recoveries.push(RolloutRecovery { target, content });
}
Ok(recoveries)
}
fn prepare_session_meta_repair(paths: &Paths, target: &Path) -> Result<SessionMetaRepair> {
let target = if target.is_absolute() {
target.to_path_buf()
} else {
paths.codex_home.join(target)
};
ensure_safe_recovery_target(&paths.codex_home, &target)?;
let relative = target.strip_prefix(&paths.codex_home)?;
if !relative.starts_with("sessions") && !relative.starts_with("archived_sessions") {
bail!("只能去重 sessions 或 archived_sessions 中的 rollout");
}
let metadata = fs::symlink_metadata(&target)?;
if !metadata.file_type().is_file() || metadata.file_type().is_symlink() {
bail!("rollout 不是普通文件：{}", target.display());
}
if metadata.len() > MAX_RECOVERY_FILE_SIZE {
bail!("rollout 超过大小上限：{}", target.display());
}
let connection = open_database(&paths.database, true)?;
let database_id = connection
.query_row(
"SELECT id FROM threads WHERE rollout_path=?1",
[target.to_string_lossy().as_ref()],
|row| row.get::<_, String>(0),
)
.optional()?
.with_context(|| format!("数据库中找不到 rollout_path：{}", target.display()))?;
let original = fs::read(&target)?;
let repaired = deduplicate_session_meta(&target, &original, &database_id)?;
Ok(SessionMetaRepair {
target,
original,
repaired,
})
}
fn deduplicate_session_meta(path: &Path, content: &[u8], expected_id: &str) -> Result<Vec<u8>> {
let text = std::str::from_utf8(content)
.with_context(|| format!("rollout 不是 UTF-8 JSONL：{}", path.display()))?;
let mut output = Vec::with_capacity(content.len());
let mut session_meta_lines = 0usize;
let mut kept_expected = false;
for segment in text.split_inclusive('\n') {
let line = segment.strip_suffix('\n').unwrap_or(segment);
let line = line.strip_suffix('\r').unwrap_or(line);
let value: Value = serde_json::from_str(line)
.with_context(|| format!("rollout 包含非法 JSONL：{}", path.display()))?;
if value.get("type").and_then(Value::as_str) == Some("session_meta") {
session_meta_lines += 1;
let id = value
.get("payload")
.and_then(|payload| payload.get("id"))
.and_then(Value::as_str)
.context("session_meta 缺少线程 ID")?;
if id != expected_id || kept_expected {
continue;
}
kept_expected = true;
}
output.extend_from_slice(segment.as_bytes());
}
if session_meta_lines < 2 {
bail!("rollout 没有重复 session_meta，无需修改");
}
if !kept_expected {
bail!("rollout 中没有与数据库线程 ID 一致的 session_meta");
}
Ok(output)
}
fn validate_recovery_rollout(source: &Path, content: &[u8]) -> Result<String> {
let text = std::str::from_utf8(content)
.with_context(|| format!("恢复来源不是 UTF-8 JSONL：{}", source.display()))?;
let mut thread_id = None;
let mut session_meta_lines = 0usize;
for line in text.lines().filter(|line| !line.trim().is_empty()) {
let value: Value = serde_json::from_str(line)
.with_context(|| format!("恢复来源包含非法 JSONL：{}", source.display()))?;
if value.get("type").and_then(Value::as_str) != Some("session_meta") {
continue;
}
session_meta_lines += 1;
let id = value
.get("payload")
.and_then(|payload| payload.get("id"))
.and_then(Value::as_str)
.context("恢复来源的 session_meta 缺少线程 ID")?;
if let Some(existing) = &thread_id
&& existing != id
{
bail!("恢复来源包含不同线程的 session_meta");
}
thread_id = Some(id.to_owned());
}
if session_meta_lines != 1 {
bail!("恢复来源必须恰好包含一条 session_meta");
}
thread_id.context("恢复来源缺少 session_meta")
}
fn ensure_safe_recovery_target(codex_home: &Path, target: &Path) -> Result<()> {
if !target.starts_with(codex_home)
|| target
.strip_prefix(codex_home)?
.components()
.any(|component| !matches!(component, std::path::Component::Normal(_)))
{
bail!(
"数据库 rollout_path 超出 Codex 数据目录：{}",
target.display()
);
}
let mut current = codex_home.to_path_buf();
for component in target.strip_prefix(codex_home)?.components() {
current.push(component);
if current
.symlink_metadata()
.is_ok_and(|metadata| metadata.file_type().is_symlink())
{
bail!("拒绝通过符号链接恢复 rollout：{}", current.display());
}
}
Ok(())
}
fn session_records(paths: &Paths) -> Result<Vec<SessionRecord>> {
let mut records = Vec::new();
for path in rollout_paths(paths) {
let text = fs::read_to_string(&path)
.with_context(|| format!("无法读取会话文件：{}", path.display()))?;
let mut record = None;
let mut session_meta_lines = 0;
for line in text.lines() {
let value: Value = serde_json::from_str(line)
.with_context(|| format!("会话 JSONL 格式错误：{}", path.display()))?;
if value.get("type").and_then(Value::as_str) != Some("session_meta") {
continue;
}
session_meta_lines += 1;
if record.is_some() {
continue;
}
let payload = value
.get("payload")
.and_then(Value::as_object)
.with_context(|| format!("session_meta 缺少 payload：{}", path.display()))?;
let provider = payload
.get("model_provider")
.and_then(Value::as_str)
.unwrap_or_default()
.to_owned();
let model = payload
.get("model")
.and_then(Value::as_str)
.map(str::to_owned);
record = Some(SessionRecord {
path: path.clone(),
provider,
model,
session_meta_lines: 0,
});
}
if let Some(mut record) = record {
record.session_meta_lines = session_meta_lines;
records.push(record);
}
}
Ok(records)
}
fn read_session_index(path: &Path) -> Result<BTreeMap<String, Value>> {
let diagnostics = inspect_session_index(path)?;
if diagnostics.invalid_lines != 0 {
bail!("session_index.jsonl 包含非法行");
}
Ok(diagnostics.entries)
}
fn inspect_session_index(path: &Path) -> Result<IndexDiagnostics> {
let mut diagnostics = IndexDiagnostics::default();
if !path.is_file() {
return Ok(diagnostics);
}
for line in fs::read_to_string(path)?
.lines()
.filter(|line| !line.trim().is_empty())
{
let Ok(value) = serde_json::from_str::<Value>(line) else {
diagnostics.invalid_lines += 1;
continue;
};
let Some(id) = value.get("id").and_then(Value::as_str) else {
diagnostics.invalid_lines += 1;
continue;
};
if diagnostics.entries.insert(id.to_owned(), value).is_some() {
diagnostics.duplicate_ids += 1;
}
}
Ok(diagnostics)
}
fn update_database(paths: &Paths, target: &Assignment) -> Result<()> {
let mut conn = open_database(&paths.database, false)?;
let columns = thread_columns(&conn)?;
let transaction = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
if columns.contains("model")
&& let Some(model) = &target.model
{
transaction.execute(
"UPDATE threads SET model_provider=?1, model=?2 WHERE model_provider IS NULL OR model_provider<>?1 OR model IS NULL OR model<>?2",
params![target.provider, model],
)?;
} else {
transaction.execute(
"UPDATE threads SET model_provider=?1 WHERE model_provider IS NULL OR model_provider<>?1",
[&target.provider],
)?;
}
transaction.commit()?;
conn.execute_batch("PRAGMA wal_checkpoint(PASSIVE)")?;
Ok(())
}
fn rewrite_sessions(paths: &Paths, target: &Assignment) -> Result<usize> {
let records = session_records(paths)?;
let mut updated = 0;
for record in records {
if !assignment_differs(&record.provider, record.model.as_deref(), target)
&& record.session_meta_lines == 1
{
continue;
}
rewrite_session_meta(&record.path, target)?;
updated += 1;
}
Ok(updated)
}
fn rewrite_session_meta(path: &Path, target: &Assignment) -> Result<()> {
let text = fs::read_to_string(path)?;
let mut output = String::with_capacity(text.len());
let mut changed = false;
for segment in text.split_inclusive('\n') {
let (line, ending) = if let Some(line) = segment.strip_suffix("\r\n") {
(line, "\r\n")
} else if let Some(line) = segment.strip_suffix('\n') {
(line, "\n")
} else {
(segment, "")
};
let mut value: Value = serde_json::from_str(line)?;
if value.get("type").and_then(Value::as_str) == Some("session_meta") {
if changed {
continue;
}
if let Some(payload) = value.get_mut("payload").and_then(Value::as_object_mut) {
payload.insert(
"model_provider".into(),
Value::String(target.provider.clone()),
);
if let Some(model) = &target.model {
payload.insert("model".into(), Value::String(model.clone()));
}
output.push_str(&serde_json::to_string(&value)?);
output.push_str(ending);
changed = true;
continue;
}
}
output.push_str(segment);
}
if !changed {
bail!("会话文件缺少 session_meta：{}", path.display());
}
atomic_write(path, output.as_bytes())
}
fn rebuild_session_index(paths: &Paths) -> Result<()> {
let conn = open_database(&paths.database, true)?;
atomic_write(&paths.session_index, &render_session_index(paths, &conn)?)
}
fn render_session_index(paths: &Paths, connection: &Connection) -> Result<Vec<u8>> {
let columns = thread_columns(connection)?;
let rows = sidebar_threads(connection, &columns)?;
let mut existing = read_session_index(&paths.session_index)?;
let mut merged = Vec::new();
for thread in rows
.into_iter()
.filter(|thread| rollout_exists(&paths.codex_home, thread.rollout_path.as_deref()))
{
let entry = existing.remove(&thread.id).unwrap_or_else(|| {
json!({
"id": thread.id,
"thread_name": thread.title,
"updated_at": thread.updated_at
})
});
merged.push(entry);
}
merged.sort_by_key(index_sort_key);
let mut output = String::new();
for entry in merged {
output.push_str(&serde_json::to_string(&entry)?);
output.push('\n');
}
Ok(output.into_bytes())
}
fn backup_doctor_state(
paths: &Paths,
report: &DoctorReport,
original: Option<&[u8]>,
session_repair: Option<&SessionMetaRepair>,
) -> Result<PathBuf> {
fs::create_dir_all(&paths.backup_root)?;
let stamp = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
let destination = paths.backup_root.join(format!("history-doctor-{stamp}"));
fs::create_dir(&destination)?;
let result = (|| -> Result<()> {
atomic_write(
&destination.join("doctor-report.json"),
&serde_json::to_vec_pretty(report)?,
)?;
if let Some(content) = original {
atomic_write(&destination.join("session_index.jsonl"), content)?;
} else {
atomic_write(&destination.join("session-index-was-absent"), b"")?;
}
if let Some(repair) = session_repair {
let relative = repair.target.strip_prefix(&paths.codex_home)?;
atomic_write(
&destination.join("rollouts").join(relative),
&repair.original,
)?;
}
Ok(())
})();
if let Err(error) = result {
let _ = fs::remove_dir_all(&destination);
return Err(error);
}
Ok(destination)
}
fn restore_original_index(paths: &Paths, original: Option<&[u8]>) -> Result<()> {
if let Some(content) = original {
atomic_write(&paths.session_index, content)
} else if paths.session_index.exists() {
fs::remove_file(&paths.session_index)?;
Ok(())
} else {
Ok(())
}
}
fn index_sort_key(value: &Value) -> (String, String) {
(
value
.get("updated_at")
.and_then(Value::as_str)
.unwrap_or_default()
.to_owned(),
value
.get("id")
.and_then(Value::as_str)
.unwrap_or_default()
.to_owned(),
)
}
#[cfg(test)]
mod tests {
use super::*;
fn fixture() -> Result<(PathBuf, Paths)> {
let root = std::env::temp_dir().join(format!(
"codex-sync-history-{}-{}",
std::process::id(),
SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos()
));
let codex = root.join(".codex");
fs::create_dir_all(codex.join("sessions/2026/07/19"))?;
fs::write(
codex.join("config.toml"),
"model_provider = \"current\"\nmodel = \"gpt-current\"\n",
)?;
let conn = Connection::open(codex.join("state_5.sqlite"))?;
conn.execute_batch(
"CREATE TABLE threads (id TEXT PRIMARY KEY, title TEXT, model_provider TEXT, model TEXT, archived INTEGER, updated_at INTEGER, source TEXT);
INSERT INTO threads VALUES ('old', 'Old chat', 'old-provider', 'gpt-old', 0, 1700000000, 'vscode');
INSERT INTO threads VALUES ('current', 'Current chat', 'current', 'gpt-current', 0, 1700000010, 'vscode');
INSERT INTO threads VALUES ('archived', 'Archived chat', 'current', 'gpt-current', 1, 1700000020, 'vscode');
INSERT INTO threads VALUES ('guardian', 'Guardian', 'current', 'gpt-current', 0, 1700000030, '{\"subagent\":{\"other\":\"guardian\"}}');",
)?;
drop(conn);
fs::write(
codex.join("sessions/2026/07/19/rollout-old.jsonl"),
"{\"type\":\"session_meta\",\"payload\":{\"id\":\"old\",\"model_provider\":\"old-provider\",\"model\":\"gpt-old\"}}\n{\"type\":\"message\",\"payload\":{\"text\":\"keep me\"}}\n{\"type\":\"session_meta\",\"payload\":{\"id\":\"old\",\"model_provider\":\"older-provider\",\"model\":\"gpt-older\"}}\n",
)?;
fs::write(
codex.join("session_index.jsonl"),
"{\"id\":\"current\",\"thread_name\":\"Current chat\",\"updated_at\":\"2023-11-14T22:13:30Z\"}\n{\"id\":\"archived\",\"thread_name\":\"Archived chat\",\"updated_at\":\"2023-11-14T22:13:40Z\"}\n{\"id\":\"guardian\",\"thread_name\":\"Guardian\",\"updated_at\":\"2023-11-14T22:13:50Z\"}\n",
)?;
let paths = Paths::resolve(Some(codex))?;
Ok((root, paths))
}
#[test]
fn merge_updates_database_session_meta_and_index_with_backup() -> Result<()> {
let (root, paths) = fixture()?;
let preview = merge(&paths, false)?;
assert_eq!(preview.database_rows_to_change, 1);
assert_eq!(preview.rollout_files_to_change, 1);
assert_eq!(preview.duplicate_session_meta_lines, 1);
assert_eq!(preview.missing_session_index_entries, 1);
assert_eq!(preview.stale_session_index_entries, 2);
let applied = merge(&paths, true)?;
assert!(applied.backup.as_ref().is_some_and(|path| path.is_dir()));
let verified = inspect(&paths)?;
assert_eq!(verified.database_rows_to_change, 0);
assert_eq!(verified.rollout_files_to_change, 0);
assert_eq!(verified.duplicate_session_meta_lines, 0);
assert_eq!(verified.missing_session_index_entries, 0);
assert_eq!(verified.stale_session_index_entries, 0);
let conn = open_database(&paths.database, true)?;
let assignment: (String, String) = conn.query_row(
"SELECT model_provider, model FROM threads WHERE id='old'",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
assert_eq!(assignment, ("current".into(), "gpt-current".into()));
let rollout = fs::read_to_string(
paths
.codex_home
.join("sessions/2026/07/19/rollout-old.jsonl"),
)?;
assert!(rollout.contains("\"model_provider\":\"current\""));
assert!(rollout.contains("keep me"));
assert_eq!(rollout.matches("session_meta").count(), 1);
let index = fs::read_to_string(&paths.session_index)?;
assert!(index.contains("\"id\":\"old\""));
assert!(!index.contains("\"id\":\"archived\""));
assert!(!index.contains("\"id\":\"guardian\""));
fs::remove_dir_all(root)?;
Ok(())
}
#[test]
fn doctor_repairs_only_the_index_and_keeps_a_rollback_copy() -> Result<()> {
let (root, paths) = fixture()?;
let preview = doctor(&paths, false, &[], None)?;
assert!(preview.database_integrity_ok);
assert_eq!(preview.invalid_rollout_files, 0);
assert_eq!(preview.duplicate_session_meta_lines, 1);
assert_eq!(preview.missing_session_index_entries, 1);
assert_eq!(preview.stale_session_index_entries, 2);
let original = fs::read(&paths.session_index)?;
let repaired = doctor(&paths, true, &[], None)?;
assert!(repaired.repaired);
assert!(repaired.repaired_session_index);
assert_eq!(repaired.missing_session_index_entries, 0);
assert_eq!(repaired.stale_session_index_entries, 0);
assert!(!repaired.clean);
let backup = repaired.backup.context("doctor backup missing")?;
assert_eq!(fs::read(backup.join("session_index.jsonl"))?, original);
let index = fs::read_to_string(&paths.session_index)?;
assert!(index.contains("\"id\":\"old\""));
assert!(index.contains("\"id\":\"current\""));
assert!(!index.contains("\"id\":\"archived\""));
assert!(!index.contains("\"id\":\"guardian\""));
fs::remove_dir_all(root)?;
Ok(())
}
#[test]
fn doctor_refuses_invalid_input_without_modifying_it() -> Result<()> {
let (root, paths) = fixture()?;
let invalid = b"{not-json}\n";
fs::write(&paths.session_index, invalid)?;
assert!(doctor(&paths, true, &[], None).is_err());
assert_eq!(fs::read(&paths.session_index)?, invalid);
fs::remove_dir_all(root)?;
Ok(())
}
#[test]
fn doctor_atomically_restores_a_valid_missing_rollout() -> Result<()> {
let root = std::env::temp_dir().join(format!(
"codex-sync-doctor-recovery-{}-{}",
std::process::id(),
SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos()
));
let codex = root.join(".codex");
let target = codex.join("sessions/2026/07/19/rollout-missing.jsonl");
let source = root.join("backup/rollout-missing.jsonl");
fs::create_dir_all(source.parent().context("backup parent")?)?;
fs::create_dir_all(&codex)?;
fs::write(codex.join("config.toml"), "model_provider = \"current\"\n")?;
let content = b"{\"type\":\"session_meta\",\"payload\":{\"id\":\"missing\",\"model_provider\":\"current\"}}\n{\"type\":\"message\",\"payload\":{\"text\":\"preserved\"}}\n";
fs::write(&source, content)?;
let connection = Connection::open(codex.join("state_5.sqlite"))?;
connection.execute(
"CREATE TABLE threads (id TEXT PRIMARY KEY, title TEXT, archived INTEGER, updated_at INTEGER, source TEXT, rollout_path TEXT)",
[],
)?;
connection.execute(
"INSERT INTO threads VALUES ('missing', 'Missing', 0, 1700000000, 'vscode', ?1)",
[target.to_string_lossy().as_ref()],
)?;
drop(connection);
let paths = Paths::resolve(Some(codex))?;
let preview = doctor(&paths, false, &[], None)?;
assert_eq!(preview.missing_rollout_files, 1);
let repaired = doctor(&paths, true, std::slice::from_ref(&source), None)?;
assert_eq!(repaired.restored_rollout_files, 1);
assert_eq!(repaired.missing_rollout_files, 0);
assert_eq!(repaired.missing_session_index_entries, 0);
assert_eq!(fs::read(&target)?, content);
let mut duplicated = content.to_vec();
duplicated.extend_from_slice(
b"{\"type\":\"session_meta\",\"payload\":{\"id\":\"other\",\"model_provider\":\"current\"}}\n",
);
fs::write(&target, &duplicated)?;
let deduplicated = doctor(&paths, true, &[], Some(&target))?;
assert_eq!(deduplicated.deduplicated_session_meta_files, 1);
assert!(!deduplicated.repaired_session_index);
assert_eq!(deduplicated.duplicate_session_meta_lines, 0);
assert_eq!(fs::read(&target)?, content);
let backup = deduplicated.backup.context("dedupe backup missing")?;
assert_eq!(
fs::read(backup.join("rollouts/sessions/2026/07/19/rollout-missing.jsonl"))?,
duplicated
);
fs::remove_dir_all(root)?;
Ok(())
}
#[test]
fn restore_recovers_previous_assignments() -> Result<()> {
let (root, paths) = fixture()?;
let applied = merge(&paths, true)?;
let backup_path = applied.backup.context("missing backup")?;
let safety = restore(&paths, &backup_path)?;
assert!(safety.is_dir());
let restored = inspect(&paths)?;
assert_eq!(restored.database_rows_to_change, 1);
assert_eq!(restored.rollout_files_to_change, 1);
assert_eq!(restored.missing_session_index_entries, 1);
assert_eq!(restored.stale_session_index_entries, 2);
fs::remove_dir_all(root)?;
Ok(())
}
#[test]
fn missing_environment_is_rejected() {
assert!(Paths::resolve(Some(PathBuf::from("/missing-codex-history"))).is_err());
}
#[test]
fn technical_threads_are_excluded_from_the_user_sidebar() {
assert!(is_technical_thread(
r#"{"subagent":{"other":"guardian"}}"#,
"",
"gpt-current",
"Guardian"
));
assert!(is_technical_thread(
"vscode",
"subagent",
"gpt-current",
"Worker"
));
assert!(is_technical_thread("unknown", "", "codex-auto-review", ""));
assert!(!is_technical_thread(
"vscode",
"",
"codex-auto-review",
"用户主动创建的审核任务"
));
}
#[test]
fn missing_provider_uses_codex_builtin_openai_provider() -> Result<()> {
let (root, paths) = fixture()?;
fs::write(&paths.config, "model = \"gpt-current\"\n")?;
let report = inspect(&paths)?;
assert_eq!(report.current_provider, "openai");
assert_eq!(report.database_rows_to_change, 4);
fs::remove_dir_all(root)?;
Ok(())
}
}