chuan/cdxs
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 10 Wiki Activity
Files
292c7b017a1ab25fde150d79f7d82b77794787d3
cdxs/src/session.rs
T

1297 lines
43 KiB
Rust
Raw Blame History

//! Inspect and repair Codex session state.
//!
//! Codex stores session visibility across SQLite (`state_5.sqlite`),
//! `session_index.jsonl`, and rollout JSONL files. This module keeps those
//! pieces consistent, implements a reversible cdxs trash, and can copy missing
//! threads across managed homes.
use std::collections::{HashMap, HashSet};
use std::fs;
use std::path::{Path, PathBuf};
use anyhow::{anyhow, Context, Result};
use chrono::{DateTime, Utc};
use rusqlite::{params, Connection, Row};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::config_store::Store;
use crate::{atomic, paths};
#[derive(Debug, Clone, Serialize)]
pub struct SessionSummary {
/// Name of the managed home that owns this session.
pub home_name: String,
pub home_path: String,
pub id: String,
pub title: Option<String>,
pub cwd: Option<String>,
pub updated_at_ms: Option<i64>,
pub tokens_used: i64,
pub rollout_path: String,
pub archived: bool,
}
#[derive(Debug, Clone, Serialize)]
pub struct SessionStats {
pub session: SessionSummary,
pub rollout_bytes: Option<u64>,
pub rollout_lines: Option<usize>,
pub sqlite_tokens_used: i64,
pub rollout_total_tokens: Option<i64>,
pub rollout_input_tokens: Option<i64>,
pub rollout_output_tokens: Option<i64>,
}
#[derive(Debug, Clone, Serialize)]
pub struct VisibilityIssue {
pub home_name: String,
pub home_path: String,
pub session_id: String,
pub issue: String,
pub detail: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct RepairAction {
pub home_name: String,
pub session_id: String,
pub action: String,
pub detail: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct SyncThreadAction {
pub session_id: String,
pub source_home: String,
pub target_home: String,
pub action: String,
pub detail: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct TrashManifest {
/// Manifest schema version for future migrations.
version: u32,
deleted_at: String,
home_name: String,
home_path: String,
session_id: String,
original_rollout_path: String,
rollout_backup_file: Option<String>,
session_index_entries: Vec<String>,
thread: ThreadRowData,
}
#[derive(Debug, Clone, Serialize)]
struct TrashEntry {
home_name: String,
home_path: String,
session_id: String,
title: Option<String>,
cwd: Option<String>,
deleted_at: String,
trash_dir: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ThreadRowData {
// Mirrors the Codex `threads` table. Keeping this shape explicit makes
// repair/restore code independent from rusqlite row lifetimes.
id: String,
rollout_path: String,
created_at: Option<i64>,
updated_at: Option<i64>,
source: Option<String>,
model_provider: Option<String>,
cwd: Option<String>,
title: Option<String>,
sandbox_policy: Option<String>,
approval_mode: Option<String>,
tokens_used: Option<i64>,
has_user_event: Option<i64>,
archived: Option<i64>,
archived_at: Option<i64>,
git_sha: Option<String>,
git_branch: Option<String>,
git_origin_url: Option<String>,
cli_version: Option<String>,
first_user_message: Option<String>,
agent_nickname: Option<String>,
agent_role: Option<String>,
memory_mode: Option<String>,
model: Option<String>,
reasoning_effort: Option<String>,
agent_path: Option<String>,
created_at_ms: Option<i64>,
updated_at_ms: Option<i64>,
}
#[derive(Debug, Clone)]
struct HomeTarget {
name: String,
path: PathBuf,
}
#[derive(Debug, Clone)]
struct RolloutMeta {
session_id: String,
relative_path: String,
cwd: Option<String>,
model_provider: Option<String>,
source: Option<String>,
cli_version: Option<String>,
created_at_ms: Option<i64>,
updated_at_ms: Option<i64>,
}
pub fn list_sessions(all_homes: bool, json: bool) -> Result<()> {
let mut sessions = Vec::new();
for home in homes(all_homes)? {
sessions.extend(read_sessions_for_home(&home)?);
}
sessions.sort_by(|a, b| b.updated_at_ms.cmp(&a.updated_at_ms));
if json {
println!("{}", serde_json::to_string_pretty(&sessions)?);
return Ok(());
}
if sessions.is_empty() {
println!("没有找到 Codex 会话。");
return Ok(());
}
println!(
"{:<22} {:<12} {:<10} {:<20} {}",
"ID", "Home", "Tokens", "Updated", "Title"
);
for item in sessions {
println!(
"{:<22} {:<12} {:<10} {:<20} {}",
shorten(&item.id, 22),
shorten(&item.home_name, 12),
item.tokens_used,
format_time(item.updated_at_ms),
shorten(item.title.as_deref().unwrap_or("-"), 80)
);
}
Ok(())
}
pub fn session_stats(session_id: &str, all_homes: bool, json: bool) -> Result<()> {
let (home, row) =
find_thread(session_id, all_homes)?.ok_or_else(|| anyhow!("会话不存在: {session_id}"))?;
let summary = summary_from_thread(&home, &row);
let rollout = resolve_rollout_path(&home.path, &row.rollout_path);
let (rollout_bytes, rollout_lines, token_usage) = rollout_stats(&rollout)?;
let stats = SessionStats {
sqlite_tokens_used: row.tokens_used.unwrap_or_default(),
session: summary,
rollout_bytes,
rollout_lines,
rollout_total_tokens: token_usage.as_ref().and_then(|v| v.total_tokens),
rollout_input_tokens: token_usage.as_ref().and_then(|v| v.input_tokens),
rollout_output_tokens: token_usage.as_ref().and_then(|v| v.output_tokens),
};
if json {
println!("{}", serde_json::to_string_pretty(&stats)?);
return Ok(());
}
println!("id: {}", stats.session.id);
println!(
"home: {} ({})",
stats.session.home_name, stats.session.home_path
);
println!("title: {}", stats.session.title.as_deref().unwrap_or("-"));
println!("cwd: {}", stats.session.cwd.as_deref().unwrap_or("-"));
println!("updated: {}", format_time(stats.session.updated_at_ms));
println!("sqlite_tokens_used: {}", stats.sqlite_tokens_used);
println!(
"rollout_tokens: total={}, input={}, output={}",
opt_i64(stats.rollout_total_tokens),
opt_i64(stats.rollout_input_tokens),
opt_i64(stats.rollout_output_tokens)
);
println!(
"rollout_file: {} (bytes={}, lines={})",
stats.session.rollout_path,
opt_u64(stats.rollout_bytes),
stats
.rollout_lines
.map(|v| v.to_string())
.unwrap_or_else(|| "-".to_string())
);
Ok(())
}
pub fn trash_sessions(session_ids: Vec<String>, all_homes: bool) -> Result<()> {
if session_ids.is_empty() {
return Err(anyhow!("至少需要提供一个 session id"));
}
for session_id in session_ids {
let (home, row) = find_thread(&session_id, all_homes)?
.ok_or_else(|| anyhow!("会话不存在: {session_id}"))?;
trash_one(&home, row)?;
}
Ok(())
}
pub fn list_trash(all_homes: bool, json: bool) -> Result<()> {
let mut entries = Vec::new();
for home in homes(all_homes)? {
entries.extend(read_trash_entries(&home)?);
}
entries.sort_by(|a, b| b.deleted_at.cmp(&a.deleted_at));
if json {
println!("{}", serde_json::to_string_pretty(&entries)?);
return Ok(());
}
if entries.is_empty() {
println!("垃圾箱为空。");
return Ok(());
}
println!(
"{:<22} {:<12} {:<24} {}",
"ID", "Home", "Deleted At", "Title"
);
for item in entries {
println!(
"{:<22} {:<12} {:<24} {}",
shorten(&item.session_id, 22),
shorten(&item.home_name, 12),
item.deleted_at,
shorten(item.title.as_deref().unwrap_or("-"), 80)
);
}
Ok(())
}
pub fn restore_sessions(session_ids: Vec<String>, all_homes: bool) -> Result<()> {
if session_ids.is_empty() {
return Err(anyhow!("至少需要提供一个 session id"));
}
let manifests = trash_manifests(all_homes)?;
for session_id in session_ids {
let (manifest_path, manifest) = manifests
.iter()
.find(|(_, item)| item.session_id == session_id)
.cloned()
.ok_or_else(|| anyhow!("垃圾箱中没有找到会话: {session_id}"))?;
restore_one(&manifest_path, &manifest)?;
}
Ok(())
}
pub fn visibility_check(all_homes: bool, json: bool) -> Result<()> {
let issues = collect_visibility_issues(all_homes)?;
if json {
println!("{}", serde_json::to_string_pretty(&issues)?);
return Ok(());
}
if issues.is_empty() {
println!("没有发现会话可见性问题。");
return Ok(());
}
println!(
"{:<12} {:<22} {:<18} {}",
"Home", "Session", "Issue", "Detail"
);
for issue in issues {
println!(
"{:<12} {:<22} {:<18} {}",
shorten(&issue.home_name, 12),
shorten(&issue.session_id, 22),
issue.issue,
issue.detail
);
}
Ok(())
}
pub fn visibility_repair(all_homes: bool, json: bool) -> Result<()> {
let mut actions = Vec::new();
for home in homes(all_homes)? {
actions.extend(repair_home_visibility(&home)?);
}
if json {
println!("{}", serde_json::to_string_pretty(&actions)?);
return Ok(());
}
if actions.is_empty() {
println!("没有需要修复的会话可见性问题。");
return Ok(());
}
for action in actions {
println!(
"{} {}: {}",
action.home_name, action.session_id, action.detail
);
}
Ok(())
}
pub fn sync_threads(all_homes: bool, dry_run: bool, json: bool) -> Result<()> {
let selected_homes = homes(all_homes)?;
if selected_homes.len() < 2 {
return Err(anyhow!(
"sync-threads 至少需要两个 home，请先用 `cdxs home create` 添加实例"
));
}
// Choose the first seen copy of each thread as the source of truth, then
// fill in homes that do not have that thread yet.
let mut rows_by_id: HashMap<String, (HomeTarget, ThreadRowData)> = HashMap::new();
let mut indexes: HashMap<String, HashSet<String>> = HashMap::new();
for home in &selected_homes {
let rows = read_thread_rows_for_home(home)?;
indexes.insert(home.name.clone(), read_session_index_ids(&home.path)?);
for row in rows {
rows_by_id
.entry(row.id.clone())
.or_insert((home.clone(), row));
}
}
let mut actions = Vec::new();
for target in &selected_homes {
let target_rows = read_thread_rows_for_home(target)?;
let target_ids: HashSet<String> = target_rows.into_iter().map(|row| row.id).collect();
for (session_id, (source, source_row)) in &rows_by_id {
if target_ids.contains(session_id) {
continue;
}
let source_rollout = resolve_rollout_path(&source.path, &source_row.rollout_path);
if !source_rollout.exists() {
actions.push(SyncThreadAction {
session_id: session_id.clone(),
source_home: source.name.clone(),
target_home: target.name.clone(),
action: "skip".to_string(),
detail: "source rollout missing".to_string(),
});
continue;
}
let mut target_row = source_row.clone();
let target_relative = portable_rollout_path(source_row, source, target);
let target_rollout = resolve_rollout_path(&target.path, &target_relative);
target_row.rollout_path = target_relative.clone();
actions.push(SyncThreadAction {
session_id: session_id.clone(),
source_home: source.name.clone(),
target_home: target.name.clone(),
action: if dry_run { "would_sync" } else { "synced" }.to_string(),
detail: target_relative.clone(),
});
if dry_run {
continue;
}
backup_state_files(&target.path)?;
if let Some(parent) = target_rollout.parent() {
fs::create_dir_all(parent)
.with_context(|| format!("创建 rollout 目录失败: {}", parent.display()))?;
}
fs::copy(&source_rollout, &target_rollout).with_context(|| {
format!(
"复制 rollout 失败: source={}, target={}",
source_rollout.display(),
target_rollout.display()
)
})?;
let conn = open_state_db(&target.path)?;
insert_thread(&conn, &target_row)?;
if !indexes
.get(&target.name)
.map(|ids| ids.contains(session_id))
.unwrap_or(false)
{
append_session_index_entry(&target.path, &index_entry_from_thread(&target_row)?)?;
}
}
}
if json {
println!("{}", serde_json::to_string_pretty(&actions)?);
return Ok(());
}
if actions.is_empty() {
println!("所有 home 的会话线程已经一致。");
return Ok(());
}
for action in actions {
println!(
"{} -> {} {}: {}",
action.source_home, action.target_home, action.session_id, action.action
);
}
Ok(())
}
fn read_sessions_for_home(home: &HomeTarget) -> Result<Vec<SessionSummary>> {
let db = state_db_path(&home.path);
if !db.exists() {
return Ok(Vec::new());
}
let conn = Connection::open(&db)
.with_context(|| format!("打开 Codex 状态库失败: {}", db.display()))?;
let rows = read_thread_rows(&conn)?;
Ok(rows
.iter()
.map(|row| summary_from_thread(home, row))
.collect())
}
fn read_thread_rows_for_home(home: &HomeTarget) -> Result<Vec<ThreadRowData>> {
let db = state_db_path(&home.path);
if !db.exists() {
return Ok(Vec::new());
}
let conn = open_state_db(&home.path)?;
read_thread_rows(&conn)
}
fn read_thread_rows(conn: &Connection) -> Result<Vec<ThreadRowData>> {
let mut stmt = conn.prepare(
"SELECT id, rollout_path, created_at, updated_at, source, model_provider, cwd, title,
sandbox_policy, approval_mode, tokens_used, has_user_event, archived, archived_at,
git_sha, git_branch, git_origin_url, cli_version, first_user_message,
agent_nickname, agent_role, memory_mode, model, reasoning_effort, agent_path,
created_at_ms, updated_at_ms
FROM threads",
)?;
let rows = stmt
.query_map([], thread_from_row)?
.collect::<std::result::Result<Vec<_>, _>>()?;
Ok(rows)
}
fn find_thread(session_id: &str, all_homes: bool) -> Result<Option<(HomeTarget, ThreadRowData)>> {
for home in homes(all_homes)? {
let db = state_db_path(&home.path);
if !db.exists() {
continue;
}
let conn = Connection::open(&db)
.with_context(|| format!("打开 Codex 状态库失败: {}", db.display()))?;
let mut stmt = conn.prepare(
"SELECT id, rollout_path, created_at, updated_at, source, model_provider, cwd, title,
sandbox_policy, approval_mode, tokens_used, has_user_event, archived, archived_at,
git_sha, git_branch, git_origin_url, cli_version, first_user_message,
agent_nickname, agent_role, memory_mode, model, reasoning_effort, agent_path,
created_at_ms, updated_at_ms
FROM threads
WHERE id = ?1",
)?;
let mut rows = stmt.query(params![session_id])?;
if let Some(row) = rows.next()? {
return Ok(Some((home, thread_from_row(row)?)));
}
}
Ok(None)
}
fn collect_visibility_issues(all_homes: bool) -> Result<Vec<VisibilityIssue>> {
let mut issues = Vec::new();
for home in homes(all_homes)? {
let threads = read_thread_rows_for_home(&home)?;
let thread_ids: HashSet<String> = threads.iter().map(|row| row.id.clone()).collect();
let index_ids = read_session_index_ids(&home.path)?;
let rollouts = scan_rollouts(&home.path)?;
let rollout_ids: HashSet<String> = rollouts.keys().cloned().collect();
// A visible Codex thread needs a SQLite row, a rollout file, and a
// session_index.jsonl entry. Report whichever side is missing.
for row in &threads {
if !resolve_rollout_path(&home.path, &row.rollout_path).exists() {
issues.push(issue(
&home,
&row.id,
"missing_rollout",
format!("rollout not found: {}", row.rollout_path),
));
}
if !index_ids.contains(&row.id) {
issues.push(issue(
&home,
&row.id,
"missing_index",
"session_index.jsonl missing entry".to_string(),
));
}
}
for id in index_ids.difference(&thread_ids) {
issues.push(issue(
&home,
id,
"orphan_index",
"session_index.jsonl entry has no SQLite thread".to_string(),
));
}
for id in rollout_ids.difference(&thread_ids) {
issues.push(issue(
&home,
id,
"orphan_rollout",
"rollout file has no SQLite thread".to_string(),
));
}
}
Ok(issues)
}
fn repair_home_visibility(home: &HomeTarget) -> Result<Vec<RepairAction>> {
let mut actions = Vec::new();
let db = state_db_path(&home.path);
if !db.exists() {
return Ok(actions);
}
let conn = open_state_db(&home.path)?;
let threads = read_thread_rows(&conn)?;
let mut index_ids = read_session_index_ids(&home.path)?;
let rollouts = scan_rollouts(&home.path)?;
let thread_ids: HashSet<String> = threads.iter().map(|row| row.id.clone()).collect();
// Existing SQLite rows are preferred. If their rollout path or index entry
// is missing, repair those pieces in place.
for row in &threads {
let current_rollout = resolve_rollout_path(&home.path, &row.rollout_path);
if !current_rollout.exists() {
if let Some(found) = rollouts.get(&row.id) {
backup_state_files(&home.path)?;
conn.execute(
"UPDATE threads SET rollout_path = ?1 WHERE id = ?2",
params![found.relative_path, row.id],
)?;
actions.push(repair_action(
home,
&row.id,
"repair_rollout_path",
format!("updated rollout_path to {}", found.relative_path),
));
}
}
if !index_ids.contains(&row.id) {
backup_state_files(&home.path)?;
append_session_index_entry(&home.path, &index_entry_from_thread(row)?)?;
index_ids.insert(row.id.clone());
actions.push(repair_action(
home,
&row.id,
"append_index",
"added session_index.jsonl entry".to_string(),
));
}
}
// Rollout files without SQLite rows can still be made visible by creating a
// minimal thread row from the rollout metadata.
for (session_id, rollout) in &rollouts {
if thread_ids.contains(session_id) {
continue;
}
backup_state_files(&home.path)?;
let row = thread_from_rollout(rollout);
insert_thread(&conn, &row)?;
if !index_ids.contains(session_id) {
append_session_index_entry(&home.path, &index_entry_from_thread(&row)?)?;
index_ids.insert(session_id.clone());
}
actions.push(repair_action(
home,
session_id,
"insert_thread",
"created minimal SQLite thread from rollout metadata".to_string(),
));
}
Ok(actions)
}
fn trash_one(home: &HomeTarget, row: ThreadRowData) -> Result<()> {
let stamp = Utc::now().format("%Y%m%d-%H%M%S%.3f").to_string();
let trash_dir = home.path.join("cdxs-trash").join(format!(
"{}-{}-{}",
stamp,
safe_name(&row.id),
safe_name(&home.name)
));
fs::create_dir_all(&trash_dir)
.with_context(|| format!("创建垃圾箱目录失败: {}", trash_dir.display()))?;
// Trash is reversible: save the rollout copy, removed index lines and the
// original SQLite row in a manifest before deleting visibility state.
let rollout = resolve_rollout_path(&home.path, &row.rollout_path);
let rollout_backup_file = if rollout.exists() {
let file_name = rollout
.file_name()
.and_then(|value| value.to_str())
.unwrap_or("rollout.jsonl")
.to_string();
fs::copy(&rollout, trash_dir.join(&file_name)).with_context(|| {
format!(
"备份 rollout 失败: source={}, trash={}",
rollout.display(),
trash_dir.display()
)
})?;
Some(file_name)
} else {
None
};
let index_entries = remove_session_index_entries(&home.path, &row.id)?;
let manifest = TrashManifest {
version: 1,
deleted_at: Utc::now().to_rfc3339(),
home_name: home.name.clone(),
home_path: home.path.to_string_lossy().to_string(),
session_id: row.id.clone(),
original_rollout_path: row.rollout_path.clone(),
rollout_backup_file,
session_index_entries: index_entries,
thread: row.clone(),
};
let manifest_content = serde_json::to_string_pretty(&manifest)?;
atomic::write_atomic(&trash_dir.join("manifest.json"), &manifest_content)?;
let db = state_db_path(&home.path);
let conn = Connection::open(&db)
.with_context(|| format!("打开 Codex 状态库失败: {}", db.display()))?;
conn.execute("DELETE FROM threads WHERE id = ?1", params![row.id])?;
if rollout.exists() {
fs::remove_file(&rollout)
.with_context(|| format!("删除 rollout 文件失败: {}", rollout.display()))?;
}
println!(
"已移入垃圾箱: {} ({})",
manifest.session_id,
trash_dir.display()
);
Ok(())
}
fn restore_one(manifest_path: &Path, manifest: &TrashManifest) -> Result<()> {
let home_path = PathBuf::from(&manifest.home_path);
let db = state_db_path(&home_path);
let conn = Connection::open(&db)
.with_context(|| format!("打开 Codex 状态库失败: {}", db.display()))?;
// Restore SQLite first, then rollout and index entries. This makes a failed
// restore easy to retry from the manifest.
insert_thread(&conn, &manifest.thread)?;
if let Some(file_name) = manifest.rollout_backup_file.as_deref() {
let backup = manifest_path
.parent()
.ok_or_else(|| anyhow!("manifest 路径无父目录: {}", manifest_path.display()))?
.join(file_name);
let target = resolve_rollout_path(&home_path, &manifest.original_rollout_path);
if let Some(parent) = target.parent() {
fs::create_dir_all(parent)
.with_context(|| format!("创建 rollout 目录失败: {}", parent.display()))?;
}
fs::copy(&backup, &target).with_context(|| {
format!(
"恢复 rollout 失败: source={}, target={}",
backup.display(),
target.display()
)
})?;
}
restore_session_index_entries(&home_path, &manifest.session_index_entries)?;
let trash_dir = manifest_path
.parent()
.ok_or_else(|| anyhow!("manifest 路径无父目录: {}", manifest_path.display()))?;
fs::remove_dir_all(trash_dir)
.with_context(|| format!("清理垃圾箱条目失败: {}", trash_dir.display()))?;
println!("已恢复会话: {}", manifest.session_id);
Ok(())
}
fn remove_session_index_entries(home: &Path, session_id: &str) -> Result<Vec<String>> {
let path = home.join("session_index.jsonl");
if !path.exists() {
return Ok(Vec::new());
}
atomic::backup_if_exists(&path, home, "session_index.jsonl")?;
let content = fs::read_to_string(&path)
.with_context(|| format!("读取 session_index 失败: {}", path.display()))?;
let mut kept = Vec::new();
let mut removed = Vec::new();
for line in content.lines() {
if line_session_id(line).as_deref() == Some(session_id) {
removed.push(line.to_string());
} else {
kept.push(line.to_string());
}
}
let mut output = kept.join("\n");
if !output.is_empty() {
output.push('\n');
}
atomic::write_atomic(&path, &output)?;
Ok(removed)
}
fn restore_session_index_entries(home: &Path, entries: &[String]) -> Result<()> {
if entries.is_empty() {
return Ok(());
}
let path = home.join("session_index.jsonl");
let mut existing = if path.exists() {
fs::read_to_string(&path)
.with_context(|| format!("读取 session_index 失败: {}", path.display()))?
} else {
String::new()
};
let existing_ids: HashSet<String> = existing.lines().filter_map(line_session_id).collect();
if !existing.is_empty() && !existing.ends_with('\n') {
existing.push('\n');
}
for entry in entries {
let should_append = line_session_id(entry)
.map(|id| !existing_ids.contains(&id))
.unwrap_or(true);
if should_append {
existing.push_str(entry);
existing.push('\n');
}
}
atomic::backup_if_exists(&path, home, "session_index.jsonl")?;
atomic::write_atomic(&path, &existing)?;
Ok(())
}
fn read_session_index_ids(home: &Path) -> Result<HashSet<String>> {
let path = home.join("session_index.jsonl");
if !path.exists() {
return Ok(HashSet::new());
}
let content = fs::read_to_string(&path)
.with_context(|| format!("读取 session_index 失败: {}", path.display()))?;
Ok(content.lines().filter_map(line_session_id).collect())
}
fn append_session_index_entry(home: &Path, entry: &str) -> Result<()> {
let path = home.join("session_index.jsonl");
let mut content = if path.exists() {
fs::read_to_string(&path)
.with_context(|| format!("读取 session_index 失败: {}", path.display()))?
} else {
String::new()
};
if !content.is_empty() && !content.ends_with('\n') {
content.push('\n');
}
content.push_str(entry);
content.push('\n');
atomic::write_atomic(&path, &content)?;
Ok(())
}
fn index_entry_from_thread(row: &ThreadRowData) -> Result<String> {
// Codex only needs a compact JSONL index entry for the session picker.
let title = row
.title
.as_deref()
.or(row.first_user_message.as_deref())
.unwrap_or(&row.id);
let updated_at = row
.updated_at_ms
.or(row.updated_at.map(|value| value * 1000))
.and_then(DateTime::<Utc>::from_timestamp_millis)
.unwrap_or_else(Utc::now);
let value = serde_json::json!({
"id": row.id,
"thread_name": title,
"updated_at": updated_at.to_rfc3339(),
});
Ok(serde_json::to_string(&value)?)
}
fn scan_rollouts(home: &Path) -> Result<HashMap<String, RolloutMeta>> {
let mut out = HashMap::new();
for dirname in ["sessions", "archived_sessions"] {
let root = home.join(dirname);
if root.exists() {
scan_rollout_dir(home, &root, &mut out)?;
}
}
Ok(out)
}
fn scan_rollout_dir(home: &Path, dir: &Path, out: &mut HashMap<String, RolloutMeta>) -> Result<()> {
for entry in fs::read_dir(dir).with_context(|| format!("读取目录失败: {}", dir.display()))?
{
let entry = entry?;
let path = entry.path();
if path.is_dir() {
scan_rollout_dir(home, &path, out)?;
continue;
}
if path.extension().and_then(|value| value.to_str()) != Some("jsonl") {
continue;
}
if let Some(meta) = read_rollout_meta(home, &path)? {
out.entry(meta.session_id.clone()).or_insert(meta);
}
}
Ok(())
}
fn read_rollout_meta(home: &Path, path: &Path) -> Result<Option<RolloutMeta>> {
let content = fs::read_to_string(path)
.with_context(|| format!("读取 rollout 失败: {}", path.display()))?;
// session_meta is normally at the top of the rollout. Limit scanning so a
// malformed large transcript cannot make repair unexpectedly expensive.
for line in content.lines().take(25) {
let Ok(value) = serde_json::from_str::<Value>(line) else {
continue;
};
if value.get("type").and_then(Value::as_str) != Some("session_meta") {
continue;
}
let payload = value.get("payload").unwrap_or(&Value::Null);
let Some(session_id) = payload.get("id").and_then(Value::as_str) else {
continue;
};
let metadata = fs::metadata(path)
.with_context(|| format!("读取 rollout 元数据失败: {}", path.display()))?;
let updated_at_ms = metadata
.modified()
.ok()
.and_then(|value| value.duration_since(std::time::UNIX_EPOCH).ok())
.map(|value| value.as_millis() as i64);
return Ok(Some(RolloutMeta {
session_id: session_id.to_string(),
relative_path: relative_path(home, path),
cwd: payload
.get("cwd")
.and_then(Value::as_str)
.map(str::to_string),
model_provider: payload
.get("model_provider")
.and_then(Value::as_str)
.map(str::to_string),
source: payload
.get("source")
.and_then(Value::as_str)
.map(str::to_string),
cli_version: payload
.get("cli_version")
.and_then(Value::as_str)
.map(str::to_string),
created_at_ms: payload
.get("timestamp")
.and_then(Value::as_str)
.and_then(parse_rfc3339_ms),
updated_at_ms,
}));
}
Ok(None)
}
fn thread_from_rollout(meta: &RolloutMeta) -> ThreadRowData {
let now_ms = Utc::now().timestamp_millis();
let created_at_ms = meta.created_at_ms.or(meta.updated_at_ms).unwrap_or(now_ms);
let updated_at_ms = meta.updated_at_ms.unwrap_or(created_at_ms);
ThreadRowData {
id: meta.session_id.clone(),
rollout_path: meta.relative_path.clone(),
created_at: Some(created_at_ms / 1000),
updated_at: Some(updated_at_ms / 1000),
source: meta.source.clone(),
model_provider: meta
.model_provider
.clone()
.or_else(|| Some("openai".to_string())),
cwd: meta.cwd.clone(),
title: Some(meta.session_id.clone()),
sandbox_policy: None,
approval_mode: None,
tokens_used: Some(0),
has_user_event: Some(1),
archived: Some(0),
archived_at: None,
git_sha: None,
git_branch: None,
git_origin_url: None,
cli_version: meta.cli_version.clone().or_else(|| Some(String::new())),
first_user_message: Some(String::new()),
agent_nickname: None,
agent_role: None,
memory_mode: Some("enabled".to_string()),
model: None,
reasoning_effort: None,
agent_path: None,
created_at_ms: Some(created_at_ms),
updated_at_ms: Some(updated_at_ms),
}
}
fn backup_state_files(home: &Path) -> Result<()> {
// These are the Codex files session operations may modify.
let db = state_db_path(home);
atomic::backup_if_exists(&db, home, "state_5.sqlite")?;
atomic::backup_if_exists(
&home.join("session_index.jsonl"),
home,
"session_index.jsonl",
)?;
Ok(())
}
fn open_state_db(home: &Path) -> Result<Connection> {
let db = state_db_path(home);
Connection::open(&db).with_context(|| format!("打开 Codex 状态库失败: {}", db.display()))
}
fn portable_rollout_path(row: &ThreadRowData, source: &HomeTarget, target: &HomeTarget) -> String {
let source_rollout = resolve_rollout_path(&source.path, &row.rollout_path);
let relative = relative_path(&source.path, &source_rollout);
let target_rollout = resolve_rollout_path(&target.path, &relative);
if target_rollout.exists() {
// Avoid overwriting an existing rollout in the target home.
let file_name = source_rollout
.file_name()
.and_then(|value| value.to_str())
.unwrap_or("rollout.jsonl");
format!(
"sessions/cdxs-sync/{}/{}",
safe_name(&source.name),
file_name
)
} else {
relative
}
}
fn relative_path(base: &Path, path: &Path) -> String {
path.strip_prefix(base)
.unwrap_or(path)
.to_string_lossy()
.replace('\\', "/")
}
fn parse_rfc3339_ms(value: &str) -> Option<i64> {
DateTime::parse_from_rfc3339(value)
.ok()
.map(|value| value.timestamp_millis())
}
fn issue(home: &HomeTarget, session_id: &str, issue: &str, detail: String) -> VisibilityIssue {
VisibilityIssue {
home_name: home.name.clone(),
home_path: home.path.to_string_lossy().to_string(),
session_id: session_id.to_string(),
issue: issue.to_string(),
detail,
}
}
fn repair_action(
home: &HomeTarget,
session_id: &str,
action: &str,
detail: String,
) -> RepairAction {
RepairAction {
home_name: home.name.clone(),
session_id: session_id.to_string(),
action: action.to_string(),
detail,
}
}
fn trash_manifests(all_homes: bool) -> Result<Vec<(PathBuf, TrashManifest)>> {
let mut manifests = Vec::new();
for home in homes(all_homes)? {
let root = home.path.join("cdxs-trash");
if !root.exists() {
continue;
}
for entry in
fs::read_dir(&root).with_context(|| format!("读取垃圾箱失败: {}", root.display()))?
{
let entry = entry?;
let manifest_path = entry.path().join("manifest.json");
if !manifest_path.exists() {
continue;
}
let content = fs::read_to_string(&manifest_path)
.with_context(|| format!("读取 manifest 失败: {}", manifest_path.display()))?;
let manifest: TrashManifest =
serde_json::from_str(&content).context("解析 manifest 失败")?;
manifests.push((manifest_path, manifest));
}
}
Ok(manifests)
}
fn read_trash_entries(home: &HomeTarget) -> Result<Vec<TrashEntry>> {
let mut out = Vec::new();
let root = home.path.join("cdxs-trash");
if !root.exists() {
return Ok(out);
}
for entry in
fs::read_dir(&root).with_context(|| format!("读取垃圾箱失败: {}", root.display()))?
{
let entry = entry?;
let manifest_path = entry.path().join("manifest.json");
if !manifest_path.exists() {
continue;
}
let content = fs::read_to_string(&manifest_path)
.with_context(|| format!("读取 manifest 失败: {}", manifest_path.display()))?;
let manifest: TrashManifest =
serde_json::from_str(&content).context("解析 manifest 失败")?;
out.push(TrashEntry {
home_name: manifest.home_name,
home_path: manifest.home_path,
session_id: manifest.session_id,
title: manifest.thread.title,
cwd: manifest.thread.cwd,
deleted_at: manifest.deleted_at,
trash_dir: entry.path().to_string_lossy().to_string(),
});
}
Ok(out)
}
fn homes(all_homes: bool) -> Result<Vec<HomeTarget>> {
let default_home = paths::codex_home(None)?;
if !all_homes {
return Ok(vec![HomeTarget {
name: "default".to_string(),
path: default_home,
}]);
}
let store = Store::load(&default_home)?;
let mut seen = HashSet::new();
let mut result = Vec::new();
// De-duplicate paths because multiple names can point to the same home.
for home in &store.homes {
let path = paths::expand_home(PathBuf::from(&home.path));
let key = path.to_string_lossy().to_string();
if seen.insert(key) {
result.push(HomeTarget {
name: home.name.clone(),
path,
});
}
}
Ok(result)
}
fn thread_from_row(row: &Row<'_>) -> rusqlite::Result<ThreadRowData> {
Ok(ThreadRowData {
id: row.get(0)?,
rollout_path: row.get(1)?,
created_at: row.get(2)?,
updated_at: row.get(3)?,
source: row.get(4)?,
model_provider: row.get(5)?,
cwd: row.get(6)?,
title: row.get(7)?,
sandbox_policy: row.get(8)?,
approval_mode: row.get(9)?,
tokens_used: row.get(10)?,
has_user_event: row.get(11)?,
archived: row.get(12)?,
archived_at: row.get(13)?,
git_sha: row.get(14)?,
git_branch: row.get(15)?,
git_origin_url: row.get(16)?,
cli_version: row.get(17)?,
first_user_message: row.get(18)?,
agent_nickname: row.get(19)?,
agent_role: row.get(20)?,
memory_mode: row.get(21)?,
model: row.get(22)?,
reasoning_effort: row.get(23)?,
agent_path: row.get(24)?,
created_at_ms: row.get(25)?,
updated_at_ms: row.get(26)?,
})
}
fn insert_thread(conn: &Connection, row: &ThreadRowData) -> Result<()> {
conn.execute(
"INSERT OR REPLACE INTO threads (
id, rollout_path, created_at, updated_at, source, model_provider, cwd, title,
sandbox_policy, approval_mode, tokens_used, has_user_event, archived, archived_at,
git_sha, git_branch, git_origin_url, cli_version, first_user_message,
agent_nickname, agent_role, memory_mode, model, reasoning_effort, agent_path,
created_at_ms, updated_at_ms
) VALUES (
?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17,
?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26, ?27
)",
params![
row.id,
row.rollout_path,
row.created_at,
row.updated_at,
row.source,
row.model_provider,
row.cwd,
row.title,
row.sandbox_policy,
row.approval_mode,
row.tokens_used,
row.has_user_event,
row.archived,
row.archived_at,
row.git_sha,
row.git_branch,
row.git_origin_url,
row.cli_version,
row.first_user_message,
row.agent_nickname,
row.agent_role,
row.memory_mode,
row.model,
row.reasoning_effort,
row.agent_path,
row.created_at_ms,
row.updated_at_ms,
],
)?;
Ok(())
}
fn summary_from_thread(home: &HomeTarget, row: &ThreadRowData) -> SessionSummary {
SessionSummary {
home_name: home.name.clone(),
home_path: home.path.to_string_lossy().to_string(),
id: row.id.clone(),
title: row.title.clone(),
cwd: row.cwd.clone(),
updated_at_ms: row
.updated_at_ms
.or(row.updated_at.map(|value| value * 1000)),
tokens_used: row.tokens_used.unwrap_or_default(),
rollout_path: row.rollout_path.clone(),
archived: row.archived.unwrap_or_default() != 0,
}
}
fn rollout_stats(path: &Path) -> Result<(Option<u64>, Option<usize>, Option<TokenUsage>)> {
if !path.exists() {
return Ok((None, None, None));
}
let metadata = fs::metadata(path)
.with_context(|| format!("读取 rollout 元数据失败: {}", path.display()))?;
let content = fs::read_to_string(path)
.with_context(|| format!("读取 rollout 失败: {}", path.display()))?;
let mut usage = None;
for line in content.lines() {
// Keep the last token usage record, which represents the latest model
// accounting in the rollout.
if let Ok(value) = serde_json::from_str::<Value>(line) {
if let Some(next) = find_token_usage(&value) {
usage = Some(next);
}
}
}
Ok((Some(metadata.len()), Some(content.lines().count()), usage))
}
#[derive(Debug, Clone)]
struct TokenUsage {
total_tokens: Option<i64>,
input_tokens: Option<i64>,
output_tokens: Option<i64>,
}
fn find_token_usage(value: &Value) -> Option<TokenUsage> {
match value {
Value::Object(map) => {
if let Some(total) = map.get("total_token_usage") {
return Some(TokenUsage {
total_tokens: total.get("total_tokens").and_then(Value::as_i64),
input_tokens: total.get("input_tokens").and_then(Value::as_i64),
output_tokens: total.get("output_tokens").and_then(Value::as_i64),
});
}
for child in map.values() {
if let Some(usage) = find_token_usage(child) {
return Some(usage);
}
}
None
}
Value::Array(items) => items.iter().find_map(find_token_usage),
_ => None,
}
}
fn line_session_id(line: &str) -> Option<String> {
serde_json::from_str::<Value>(line)
.ok()
.and_then(|value| value.get("id").and_then(Value::as_str).map(str::to_string))
}
fn state_db_path(home: &Path) -> PathBuf {
home.join("state_5.sqlite")
}
fn resolve_rollout_path(home: &Path, rollout_path: &str) -> PathBuf {
let path = PathBuf::from(rollout_path);
if path.is_absolute() {
path
} else {
home.join(path)
}
}
fn format_time(ms: Option<i64>) -> String {
let Some(ms) = ms else {
return "-".to_string();
};
DateTime::<Utc>::from_timestamp_millis(ms)
.map(|value| value.format("%Y-%m-%d %H:%M:%S").to_string())
.unwrap_or_else(|| ms.to_string())
}
fn opt_i64(value: Option<i64>) -> String {
value
.map(|value| value.to_string())
.unwrap_or_else(|| "-".to_string())
}
fn opt_u64(value: Option<u64>) -> String {
value
.map(|value| value.to_string())
.unwrap_or_else(|| "-".to_string())
}
fn safe_name(value: &str) -> String {
value
.chars()
.map(|ch| {
if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
ch
} else {
'_'
}
})
.collect()
}
fn shorten(value: &str, width: usize) -> String {
if value.chars().count() <= width {
return value.to_string();
}
if width <= 1 {
return "...".to_string();
}
let mut out = value
.chars()
.take(width.saturating_sub(3))
.collect::<String>();
out.push_str("...");
out
}
Reference in New Issue View Git Blame Copy Permalink