fix: sync all codex sessions
This commit is contained in:
mac
@@ -4,10 +4,12 @@
|
||||
const fs = require("fs");
|
||||
const path = require("path");
|
||||
const os = require("os");
|
||||
const { execFileSync } = require("child_process");
|
||||
const { execFileSync, spawnSync } = require("child_process");
|
||||
const initSqlJs = require("sql.js");
|
||||
|
||||
const CODEX_HOME = path.join(os.homedir(), ".codex");
|
||||
const SESSION_INDEX = path.join(CODEX_HOME, "session_index.jsonl");
|
||||
const GLOBAL_STATE = path.join(CODEX_HOME, ".codex-global-state.json");
|
||||
|
||||
// 检测 Codex 是否在运行（尽力而为，跨平台）
|
||||
function codexRunning() {
|
||||
@@ -55,6 +57,231 @@ function findRollouts() {
|
||||
return out;
|
||||
}
|
||||
|
||||
function isArchivedRollout(file) {
|
||||
return file.startsWith(path.join(CODEX_HOME, "archived_sessions") + path.sep);
|
||||
}
|
||||
|
||||
function loadIndexTitles() {
|
||||
const map = {};
|
||||
try {
|
||||
for (const line of fs.readFileSync(SESSION_INDEX, "utf-8").split("\n")) {
|
||||
if (!line.trim()) continue;
|
||||
try {
|
||||
const d = JSON.parse(line);
|
||||
if (d.id && d.thread_name) map[d.id] = d.thread_name;
|
||||
} catch {}
|
||||
}
|
||||
} catch {}
|
||||
return map;
|
||||
}
|
||||
|
||||
function parseTimeMs(tsText, fallbackMs = Date.now()) {
|
||||
const n = Date.parse(tsText || "");
|
||||
return Number.isFinite(n) ? n : fallbackMs;
|
||||
}
|
||||
|
||||
function firstText(value) {
|
||||
if (typeof value === "string") return value;
|
||||
if (Array.isArray(value)) {
|
||||
return value
|
||||
.map((x) => (typeof x === "string" ? x : x && typeof x.text === "string" ? x.text : ""))
|
||||
.filter(Boolean)
|
||||
.join("\n");
|
||||
}
|
||||
return "";
|
||||
}
|
||||
|
||||
function sql(v) {
|
||||
if (v === null || v === undefined) return "NULL";
|
||||
if (typeof v === "number") return Number.isFinite(v) ? String(v) : "NULL";
|
||||
return "'" + String(v).replace(/'/g, "''") + "'";
|
||||
}
|
||||
|
||||
function sqlite3Available() {
|
||||
const r = spawnSync("sqlite3", ["--version"], { encoding: "utf-8" });
|
||||
return !r.error && r.status === 0;
|
||||
}
|
||||
|
||||
function runSqlite3(dbPath, input) {
|
||||
const r = spawnSync("sqlite3", [dbPath], {
|
||||
input,
|
||||
encoding: "utf-8",
|
||||
maxBuffer: 20 * 1024 * 1024,
|
||||
});
|
||||
if (r.error || r.status !== 0) {
|
||||
throw new Error((r.stderr || r.error && r.error.message || "sqlite3 执行失败").trim());
|
||||
}
|
||||
return r.stdout || "";
|
||||
}
|
||||
|
||||
function sqliteCols(dbPath) {
|
||||
const out = runSqlite3(dbPath, "PRAGMA table_info(threads);\n");
|
||||
return new Set(out.split("\n").map((line) => line.split("|")[1]).filter(Boolean));
|
||||
}
|
||||
|
||||
function readRolloutRecord(file, titles) {
|
||||
let content;
|
||||
try {
|
||||
content = fs.readFileSync(file, "utf-8");
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
const stat = fs.statSync(file);
|
||||
const rec = {
|
||||
id: "",
|
||||
rollout_path: file,
|
||||
cwd: "",
|
||||
source: "vscode",
|
||||
model_provider: "",
|
||||
thread_source: "user",
|
||||
title: "",
|
||||
first_user_message: "",
|
||||
preview: "",
|
||||
has_user_event: content.includes('"user_message"') || content.includes('"user_input"') ? 1 : 0,
|
||||
archived: isArchivedRollout(file) ? 1 : 0,
|
||||
archived_at: null,
|
||||
created_at_ms: stat.birthtimeMs || stat.mtimeMs || Date.now(),
|
||||
updated_at_ms: stat.mtimeMs || Date.now(),
|
||||
sandbox_policy: JSON.stringify({ type: "disabled" }),
|
||||
approval_mode: "never",
|
||||
tokens_used: 0,
|
||||
cli_version: "",
|
||||
git_sha: null,
|
||||
git_branch: null,
|
||||
git_origin_url: null,
|
||||
model: null,
|
||||
reasoning_effort: null,
|
||||
history_mode: "legacy",
|
||||
};
|
||||
for (const line of content.split("\n")) {
|
||||
if (!line.trim()) continue;
|
||||
let d;
|
||||
try {
|
||||
d = JSON.parse(line);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
const p = d.payload || {};
|
||||
if (d.type === "session_meta" && !rec.id) {
|
||||
rec.id = p.id || p.session_id || rec.id;
|
||||
rec.cwd = p.cwd || rec.cwd;
|
||||
rec.source = p.source || rec.source;
|
||||
rec.model_provider = p.model_provider || rec.model_provider;
|
||||
rec.thread_source = p.thread_source || rec.thread_source;
|
||||
rec.cli_version = p.cli_version || rec.cli_version;
|
||||
rec.history_mode = p.history_mode || rec.history_mode;
|
||||
rec.created_at_ms = parseTimeMs(p.timestamp || d.timestamp, rec.created_at_ms);
|
||||
if (p.git && typeof p.git === "object") {
|
||||
rec.git_sha = p.git.commit_hash || null;
|
||||
rec.git_branch = p.git.branch || null;
|
||||
rec.git_origin_url = p.git.repository_url || null;
|
||||
}
|
||||
} else if (d.type === "event_msg" && (p.type === "user_message" || p.type === "user_input")) {
|
||||
rec.has_user_event = 1;
|
||||
if (!rec.first_user_message) rec.first_user_message = firstText(p.message || p.input).trim();
|
||||
}
|
||||
}
|
||||
rec.updated_at_ms = Math.max(rec.updated_at_ms, rec.created_at_ms);
|
||||
rec.title = titles[rec.id] || rec.first_user_message.replace(/\s+/g, " ").slice(0, 80) || "(无标题对话)";
|
||||
rec.preview = rec.first_user_message.replace(/\s+/g, " ").slice(0, 200);
|
||||
return rec.id ? rec : null;
|
||||
}
|
||||
|
||||
function buildThreadUpsertSql(records, target, cols) {
|
||||
const desired = [
|
||||
"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", "memory_mode",
|
||||
"model", "reasoning_effort", "created_at_ms", "updated_at_ms", "thread_source", "preview",
|
||||
"recency_at", "recency_at_ms", "history_mode",
|
||||
].filter((c) => cols.has(c));
|
||||
const valueFor = (r, c) => {
|
||||
const created = Math.floor(r.created_at_ms / 1000);
|
||||
const updated = Math.floor(r.updated_at_ms / 1000);
|
||||
const values = {
|
||||
id: r.id,
|
||||
rollout_path: r.rollout_path,
|
||||
created_at: created,
|
||||
updated_at: updated,
|
||||
source: r.source,
|
||||
model_provider: target,
|
||||
cwd: r.cwd,
|
||||
title: r.title,
|
||||
sandbox_policy: r.sandbox_policy,
|
||||
approval_mode: r.approval_mode,
|
||||
tokens_used: r.tokens_used,
|
||||
has_user_event: r.has_user_event,
|
||||
archived: r.archived,
|
||||
archived_at: r.archived ? updated : null,
|
||||
git_sha: r.git_sha,
|
||||
git_branch: r.git_branch,
|
||||
git_origin_url: r.git_origin_url,
|
||||
cli_version: r.cli_version,
|
||||
first_user_message: r.first_user_message,
|
||||
memory_mode: "enabled",
|
||||
model: r.model,
|
||||
reasoning_effort: r.reasoning_effort,
|
||||
created_at_ms: Math.round(r.created_at_ms),
|
||||
updated_at_ms: Math.round(r.updated_at_ms),
|
||||
thread_source: r.thread_source,
|
||||
preview: r.preview,
|
||||
recency_at: updated,
|
||||
recency_at_ms: Math.round(r.updated_at_ms),
|
||||
history_mode: r.history_mode,
|
||||
};
|
||||
return values[c];
|
||||
};
|
||||
const updateSet = [];
|
||||
for (const c of desired) {
|
||||
if (c === "id") continue;
|
||||
if (c === "has_user_event") updateSet.push(`${c}=CASE WHEN excluded.${c}=1 THEN 1 ELSE threads.${c} END`);
|
||||
else if (["cwd", "first_user_message", "preview", "thread_source"].includes(c)) {
|
||||
updateSet.push(`${c}=CASE WHEN COALESCE(excluded.${c},'')<>'' THEN excluded.${c} ELSE threads.${c} END`);
|
||||
} else if (c === "title") {
|
||||
updateSet.push(`${c}=CASE WHEN COALESCE(threads.${c},'')='' OR threads.${c}='(无标题对话)' THEN excluded.${c} ELSE threads.${c} END`);
|
||||
} else if (["updated_at", "updated_at_ms", "recency_at", "recency_at_ms"].includes(c)) {
|
||||
updateSet.push(`${c}=MAX(threads.${c}, excluded.${c})`);
|
||||
} else {
|
||||
updateSet.push(`${c}=excluded.${c}`);
|
||||
}
|
||||
}
|
||||
const lines = ["BEGIN;"];
|
||||
for (const r of records) {
|
||||
lines.push(
|
||||
`INSERT INTO threads (${desired.join(",")}) VALUES (${desired.map((c) => sql(valueFor(r, c))).join(",")}) ` +
|
||||
`ON CONFLICT(id) DO UPDATE SET ${updateSet.join(",")};`
|
||||
);
|
||||
}
|
||||
lines.push(`UPDATE threads SET model_provider=${sql(target)} WHERE COALESCE(model_provider,'')<>${sql(target)};`);
|
||||
lines.push("COMMIT;");
|
||||
return lines.join("\n") + "\n";
|
||||
}
|
||||
|
||||
function updateProjectState(records, log = () => {}) {
|
||||
if (!fs.existsSync(GLOBAL_STATE)) return { addedProjects: 0, removedProjectless: 0 };
|
||||
let state;
|
||||
try {
|
||||
state = JSON.parse(fs.readFileSync(GLOBAL_STATE, "utf-8"));
|
||||
} catch (e) {
|
||||
log(`全局项目索引跳过：${e.message || e}`);
|
||||
return { addedProjects: 0, removedProjectless: 0 };
|
||||
}
|
||||
const projects = [...new Set(records.map((r) => r.cwd).filter(Boolean))];
|
||||
const idsWithProject = new Set(records.filter((r) => r.cwd).map((r) => r.id));
|
||||
const before = Array.isArray(state["project-order"]) ? state["project-order"] : [];
|
||||
const seen = new Set(before);
|
||||
state["project-order"] = before.concat(projects.filter((p) => !seen.has(p)));
|
||||
const oldProjectless = Array.isArray(state["projectless-thread-ids"]) ? state["projectless-thread-ids"] : [];
|
||||
state["projectless-thread-ids"] = oldProjectless.filter((id) => !idsWithProject.has(id));
|
||||
const addedProjects = state["project-order"].length - before.length;
|
||||
const removedProjectless = oldProjectless.length - state["projectless-thread-ids"].length;
|
||||
if (addedProjects || removedProjectless) {
|
||||
fs.writeFileSync(GLOBAL_STATE, JSON.stringify(state, null, 2) + "\n", "utf-8");
|
||||
log(`项目索引更新：新增项目 ${addedProjects} 个，移出无项目会话 ${removedProjectless} 条`);
|
||||
}
|
||||
return { addedProjects, removedProjectless };
|
||||
}
|
||||
|
||||
function ts() {
|
||||
const d = new Date();
|
||||
const p = (n) => String(n).padStart(2, "0");
|
||||
@@ -74,14 +301,16 @@ async function syncProviderToCurrent(log = () => {}, onProgress = () => {}) {
|
||||
const src = path.join(CODEX_HOME, name);
|
||||
if (fs.existsSync(src)) fs.copyFileSync(src, path.join(backup, name));
|
||||
}
|
||||
for (const db of fs.readdirSync(CODEX_HOME).filter((f) => /^state_.*\.sqlite/.test(f))) {
|
||||
for (const db of fs.readdirSync(CODEX_HOME).filter((f) => /^state_.*\.sqlite(?:-(?:wal|shm))?$/.test(f))) {
|
||||
fs.copyFileSync(path.join(CODEX_HOME, db), path.join(backup, db));
|
||||
}
|
||||
log(`已备份到 ${backup}`);
|
||||
|
||||
// 2) 遍历 rollout：改首行 provider，收集 thread_id / cwd / has_user_event
|
||||
// 2) 遍历 rollout：改首行 provider，收集完整线程元信息
|
||||
const userEventIds = new Set();
|
||||
const cwdById = {};
|
||||
const titles = loadIndexTitles();
|
||||
const records = [];
|
||||
let changedFiles = 0;
|
||||
const rolloutFiles = findRollouts();
|
||||
const rtotal = rolloutFiles.length;
|
||||
@@ -126,11 +355,62 @@ async function syncProviderToCurrent(log = () => {}, onProgress = () => {}) {
|
||||
changedFiles++;
|
||||
}
|
||||
if (tid && hasUE) userEventIds.add(tid);
|
||||
const rec = readRolloutRecord(file, titles);
|
||||
if (rec) {
|
||||
rec.model_provider = target;
|
||||
records.push(rec);
|
||||
}
|
||||
}
|
||||
log(`改写 rollout 文件：${changedFiles} 个；含用户消息的会话：${userEventIds.size} 条`);
|
||||
log(`改写 rollout 文件：${changedFiles} 个；含用户消息的会话：${userEventIds.size} 条；可回填线程：${records.length} 条`);
|
||||
|
||||
// 2.5) Codex 项目分组来自全局 state，历史 cwd 需要合并到 project-order。
|
||||
const projectState = updateProjectState(records, log);
|
||||
|
||||
// 3) 更新所有 state_*.sqlite
|
||||
onProgress({ phase: "更新数据库", done: 0, total: 0 });
|
||||
if (sqlite3Available()) {
|
||||
let provRows = 0;
|
||||
let ueRows = 0;
|
||||
let cwdRows = 0;
|
||||
let insertedRows = 0;
|
||||
for (const name of fs.readdirSync(CODEX_HOME).filter((f) => /^state_.*\.sqlite$/.test(f))) {
|
||||
const dbPath = path.join(CODEX_HOME, name);
|
||||
try {
|
||||
const cols = sqliteCols(dbPath);
|
||||
if (!cols.has("model_provider") || !cols.has("has_user_event")) continue;
|
||||
const beforeProvider = Number(runSqlite3(dbPath, `SELECT COUNT(*) FROM threads WHERE COALESCE(model_provider,'')<>${sql(target)};\n`).trim() || 0);
|
||||
const beforeUserEvent = Number(runSqlite3(dbPath, "SELECT COUNT(*) FROM threads WHERE COALESCE(has_user_event,0)<>1;\n").trim() || 0);
|
||||
const beforeRows = Number(runSqlite3(dbPath, "SELECT COUNT(*) FROM threads;\n").trim() || 0);
|
||||
runSqlite3(dbPath, buildThreadUpsertSql(records, target, cols));
|
||||
runSqlite3(dbPath, "PRAGMA wal_checkpoint(TRUNCATE);\n");
|
||||
const afterUserEvent = Number(runSqlite3(dbPath, "SELECT COUNT(*) FROM threads WHERE COALESCE(has_user_event,0)<>1;\n").trim() || 0);
|
||||
const afterRows = Number(runSqlite3(dbPath, "SELECT COUNT(*) FROM threads;\n").trim() || 0);
|
||||
provRows += beforeProvider;
|
||||
ueRows += Math.max(0, beforeUserEvent - afterUserEvent);
|
||||
insertedRows += Math.max(0, afterRows - beforeRows);
|
||||
cwdRows += records.filter((r) => r.cwd).length;
|
||||
} catch (e) {
|
||||
log(` sqlite3 数据库跳过 ${name}：${e.message || e}`);
|
||||
}
|
||||
}
|
||||
log(`数据库更新：provider ${provRows} 行、has_user_event ${ueRows} 行、新增线程 ${insertedRows} 行、项目索引 +${projectState.addedProjects}`);
|
||||
onProgress({ phase: "完成", done: 0, total: 0 });
|
||||
|
||||
return {
|
||||
target,
|
||||
totalConversations: rtotal,
|
||||
changedFiles,
|
||||
userEvents: userEventIds.size,
|
||||
provRows,
|
||||
ueRows,
|
||||
cwdRows,
|
||||
insertedRows,
|
||||
projectRows: projectState.addedProjects,
|
||||
projectlessRemoved: projectState.removedProjectless,
|
||||
backup,
|
||||
};
|
||||
}
|
||||
|
||||
const SQL = await initSqlJs({
|
||||
// wasm 恒在【包内】node_modules（热更包不含 node_modules），故用 BUNDLE_ROOT 而非 __dirname
|
||||
locateFile: (f) =>
|
||||
@@ -187,6 +467,9 @@ async function syncProviderToCurrent(log = () => {}, onProgress = () => {}) {
|
||||
provRows,
|
||||
ueRows,
|
||||
cwdRows,
|
||||
insertedRows: 0,
|
||||
projectRows: projectState.addedProjects,
|
||||
projectlessRemoved: projectState.removedProjectless,
|
||||
backup,
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user