feat: 发布热更 1.5.94,修复中转历史会话并支持一键清空

This commit is contained in:
mac
2026-09-11 09:06:41 +08:00
parent 0446d09b3b
commit a74576c3e1
28 changed files with 3217 additions and 155 deletions
+177 -9
View File
@@ -15,6 +15,7 @@ let GLOBAL_STATE = path.join(CODEX_HOME, ".codex-global-state.json");
// Codex 桌面端侧栏现在优先读这个本地目录库,而不是直接扫 state_*.sqlite。
// 只改 threads / rollout / session_index 而不同步这里,重启后侧栏仍会空。
let LOCAL_CATALOG_DB = path.join(CODEX_HOME, "sqlite", "codex-dev.db");
const STATE_DB_UPSERT_BATCH_SIZE = 200;
function setCodexDataDir(dir) {
if (typeof dir !== "string" || !path.isAbsolute(dir)) throw new Error("Codex 数据目录必须是绝对路径");
@@ -351,7 +352,12 @@ function threadValue(record, target, column) {
function threadUpdateSet(columns, target, parameterized = false) {
return columns.filter((c) => c !== "id").map((c) => {
if (c === "model_provider") return `${c}=${parameterized ? "?" : sql(target)}`;
if (c === "has_user_event") return `${c}=CASE WHEN excluded.${c}=1 THEN 1 ELSE threads.${c} END`;
if (c === "has_user_event") {
const subagent = columns.includes("thread_source")
? "LOWER(COALESCE(excluded.thread_source,''))='subagent'"
: "0";
return `${c}=CASE WHEN ${subagent} THEN 0 WHEN excluded.${c}=1 THEN 1 ELSE threads.${c} END`;
}
if (["cwd", "first_user_message", "preview", "thread_source"].includes(c)) {
return `${c}=CASE WHEN COALESCE(excluded.${c},'')<>'' THEN excluded.${c} ELSE threads.${c} END`;
}
@@ -449,13 +455,21 @@ function readRolloutRecord(file, titles) {
if (!rec.first_user_message) rec.first_user_message = firstText(p.message || p.input).trim();
}
}
if (
String(rec.thread_source || "").toLowerCase() === "subagent"
|| (rec.source && typeof rec.source === "object" && rec.source.subagent)
|| (typeof rec.source === "string" && /"subagent"\s*:/.test(rec.source))
) {
rec.thread_source = "subagent";
rec.has_user_event = 0;
}
rec.updated_at_ms = Math.max(rec.updated_at_ms, rec.created_at_ms);
rec.title = cleanThreadTitle(titles[rec.id] || "", rec.first_user_message, 80) || "(无标题对话)";
rec.preview = cleanUserVisibleText(rec.first_user_message).replace(/\s+/g, " ").trim().slice(0, 200);
return rec.id ? rec : null;
}
function buildThreadUpsertSql(records, target, cols) {
function buildThreadUpsertSql(records, target, cols, opts = {}) {
const desired = threadColumns(cols);
const updateSet = threadUpdateSet(desired, target);
const lines = ["BEGIN;"];
@@ -465,12 +479,20 @@ function buildThreadUpsertSql(records, target, cols) {
`ON CONFLICT(id) DO UPDATE SET ${updateSet.join(",")};`
);
}
// 【增强】确保数据库中所有会话的 provider 都更新为目标(兜底逻辑)
lines.push(`UPDATE threads SET model_provider=${sql(target)} WHERE COALESCE(model_provider,'')<>${sql(target)};`);
if (opts.updateAllProviders !== false) {
// 【增强】确保数据库中所有会话的 provider 都更新为目标(兜底逻辑)
lines.push(`UPDATE threads SET model_provider=${sql(target)} WHERE COALESCE(model_provider,'')<>${sql(target)};`);
}
lines.push("COMMIT;");
return lines.join("\n") + "\n";
}
function chunksOf(items, size) {
const chunks = [];
for (let i = 0; i < items.length; i += size) chunks.push(items.slice(i, i + size));
return chunks;
}
function updateProjectState(records, log = () => {}) {
if (!fs.existsSync(GLOBAL_STATE)) return { addedProjects: 0, removedProjectless: 0 };
let state;
@@ -580,6 +602,7 @@ function rebuildLocalThreadCatalog(records, target, log = () => {}) {
const seen = new Set();
for (const r of records) {
if (!r || !r.id || seen.has(r.id) || r.archived) continue;
if (String(r.thread_source || "").toLowerCase() === "subagent") continue;
const hasText = !!(r.first_user_message || r.preview || (r.title && r.title !== "(无标题对话)"));
const hasUser = !!r.has_user_event || hasText;
if (!hasUser) continue;
@@ -715,6 +738,7 @@ async function rebuildLocalThreadCatalogAsync(records, target, log = () => {}) {
const seen = new Set();
for (const r of records) {
if (!r || !r.id || seen.has(r.id) || r.archived) continue;
if (String(r.thread_source || "").toLowerCase() === "subagent") continue;
const hasText = !!(r.first_user_message || r.preview || (r.title && r.title !== "(无标题对话)"));
if (!(r.has_user_event || hasText)) continue;
seen.add(r.id);
@@ -823,6 +847,134 @@ function forceHasUserEventSql(userEventIds) {
.join("\n");
}
function forceSubagentNoUserEventSql(columns) {
if (!columns || !columns.has("has_user_event")) return "";
const checks = [];
if (columns.has("thread_source")) checks.push("LOWER(COALESCE(thread_source,''))='subagent'");
if (columns.has("source")) checks.push("LOWER(COALESCE(source,'')) LIKE '%\"subagent\"%'");
if (!checks.length) return "";
return `UPDATE threads SET has_user_event=0 WHERE COALESCE(has_user_event,0)<>0 AND (${checks.join(" OR ")});`;
}
function backupSqlite3(dbPath, suffix) {
const backupDir = path.join(CODEX_HOME, "backups_state", "visibility-repair");
fs.mkdirSync(backupDir, { recursive: true });
const backupPath = path.join(backupDir, `${path.basename(dbPath)}.${suffix}.bak`);
const escaped = backupPath.replace(/'/g, "''");
runSqlite3(dbPath, `.timeout 5000\n.backup '${escaped}'\n`);
return backupPath;
}
// 旧版本曾把 subagent 当成顶层用户会话写入 state/catalog。这里做轻量幂等修复,
// 不扫描或改写大体积 rollout;sqlite3 可用时允许与 Codex 的 WAL 连接并发事务。
// Windows 无 sqlite3 CLI 且 Codex 正在运行时不冒险整库覆盖,等下次 Codex 退出后再修。
async function repairLegacySessionVisibility(log = () => {}) {
const stateDbNames = (() => {
try { return fs.readdirSync(CODEX_HOME).filter((name) => /^state_.*\.sqlite$/.test(name)); } catch { return []; }
})();
const suffix = ts();
const result = { stateRows: 0, catalogRows: 0, backups: [], skipped: false };
if (sqlite3Available()) {
for (const name of stateDbNames) {
const dbPath = path.join(CODEX_HOME, name);
const cols = sqliteCols(dbPath);
const clearSql = forceSubagentNoUserEventSql(cols);
if (!clearSql) continue;
const before = Number(runSqlite3(
dbPath,
`SELECT COUNT(*) FROM threads WHERE COALESCE(has_user_event,0)<>0 AND (` +
`${cols.has("thread_source") ? "LOWER(COALESCE(thread_source,''))='subagent'" : "0"}` +
`${cols.has("source") ? `${cols.has("thread_source") ? " OR " : ""}LOWER(COALESCE(source,'')) LIKE '%\"subagent\"%'` : ""}` +
");\n"
).trim() || 0);
if (!before) continue;
result.backups.push(backupSqlite3(dbPath, suffix));
runSqlite3(dbPath, `.timeout 5000\nBEGIN IMMEDIATE;\n${clearSql}\nCOMMIT;\n`);
result.stateRows += before;
}
if (fs.existsSync(LOCAL_CATALOG_DB)) {
const catalogCols = new Set(
runSqlite3(LOCAL_CATALOG_DB, "PRAGMA table_info(local_thread_catalog);\n")
.split("\n").map((line) => line.split("|")[1]).filter(Boolean)
);
if (catalogCols.has("thread_source")) {
const before = Number(runSqlite3(
LOCAL_CATALOG_DB,
"SELECT COUNT(*) FROM local_thread_catalog WHERE LOWER(COALESCE(thread_source,''))='subagent';\n"
).trim() || 0);
if (before) {
result.backups.push(backupSqlite3(LOCAL_CATALOG_DB, suffix));
runSqlite3(
LOCAL_CATALOG_DB,
".timeout 5000\nBEGIN IMMEDIATE;\n" +
"DELETE FROM local_thread_catalog WHERE LOWER(COALESCE(thread_source,''))='subagent';\n" +
"UPDATE local_thread_catalog_metadata SET catalog_revision=COALESCE(catalog_revision,0)+1 WHERE id=1;\n" +
"COMMIT;\n"
);
result.catalogRows = before;
}
}
}
if (result.stateRows || result.catalogRows) {
log(`启动修复会话目录:state ${result.stateRows} 条,侧栏移除 subagent ${result.catalogRows} 条`);
}
return result;
}
if (codexRunning()) {
result.skipped = true;
result.reason = "Codex 正在运行且系统没有 sqlite3,已延后到下次启动";
log(`启动修复会话目录跳过:${result.reason}`);
return result;
}
const SQL = await loadSqlJs();
for (const name of stateDbNames) {
const dbPath = path.join(CODEX_HOME, name);
const db = new SQL.Database(fs.readFileSync(dbPath));
try {
const cols = sqlJsTableColumns(db, "threads");
const clearSql = forceSubagentNoUserEventSql(cols);
if (!clearSql) continue;
db.run(clearSql);
const changed = db.getRowsModified();
if (!changed) continue;
const backupPath = `${dbPath}.${suffix}.bak`;
fs.copyFileSync(dbPath, backupPath);
result.backups.push(backupPath);
fs.writeFileSync(dbPath, Buffer.from(db.export()));
result.stateRows += changed;
} finally {
db.close();
}
}
if (fs.existsSync(LOCAL_CATALOG_DB)) {
const db = new SQL.Database(fs.readFileSync(LOCAL_CATALOG_DB));
try {
const cols = sqlJsTableColumns(db, "local_thread_catalog");
if (cols.has("thread_source")) {
db.run("DELETE FROM local_thread_catalog WHERE LOWER(COALESCE(thread_source,''))='subagent'");
const changed = db.getRowsModified();
if (changed) {
try { db.run("UPDATE local_thread_catalog_metadata SET catalog_revision=COALESCE(catalog_revision,0)+1 WHERE id=1"); } catch {}
const backupPath = `${LOCAL_CATALOG_DB}.${suffix}.bak`;
fs.copyFileSync(LOCAL_CATALOG_DB, backupPath);
result.backups.push(backupPath);
fs.writeFileSync(LOCAL_CATALOG_DB, Buffer.from(db.export()));
result.catalogRows = changed;
}
}
} finally {
db.close();
}
}
if (result.stateRows || result.catalogRows) {
log(`启动修复会话目录:state ${result.stateRows} 条,侧栏移除 subagent ${result.catalogRows} 条`);
}
return result;
}
function sqlJsTableColumns(db, table) {
const result = db.exec(`PRAGMA table_info("${String(table).replace(/"/g, '""')}")`);
if (!result.length) return new Set();
@@ -921,6 +1073,10 @@ function upsertStateDbSqlJs(db, records, target, userEventIds, cwdById, log = ()
db.run("UPDATE threads SET has_user_event=1 WHERE id=? AND COALESCE(has_user_event,0)<>1", [id]);
ueRows += db.getRowsModified();
}
const checks = [];
if (cols.has("thread_source")) checks.push("LOWER(COALESCE(thread_source,''))='subagent'");
if (cols.has("source")) checks.push("LOWER(COALESCE(source,'')) LIKE '%\"subagent\"%'");
if (checks.length) db.run(`UPDATE threads SET has_user_event=0 WHERE COALESCE(has_user_event,0)<>0 AND (${checks.join(" OR ")})`);
}
if (cols.has("cwd")) {
for (const [id, cwd] of Object.entries(cwdById || {})) {
@@ -986,8 +1142,6 @@ async function syncProviderToCurrent(log = () => {}, onProgress = () => {}) {
continue;
}
if (!lines.length) continue;
const blob = lines.join("");
const hasUE = blob.includes('"user_message"') || blob.includes('"user_input"');
let tid = null;
let rewrote = false;
for (let i = 0; i < lines.length; i++) {
@@ -1015,10 +1169,10 @@ async function syncProviderToCurrent(log = () => {}, onProgress = () => {}) {
fs.writeFileSync(file, lines.join(""), "utf-8");
changedFiles++;
}
if (tid && hasUE) userEventIds.add(tid);
const rec = readRolloutRecord(file, titles);
if (rec) {
rec.model_provider = target;
if (rec.has_user_event) userEventIds.add(rec.id);
records.push(rec);
}
}
@@ -1063,9 +1217,17 @@ async function syncProviderToCurrent(log = () => {}, onProgress = () => {}) {
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));
let batchDone = 0;
for (const batch of chunksOf(records, STATE_DB_UPSERT_BATCH_SIZE)) {
batchDone += batch.length;
onProgress({ phase: `更新数据库 ${name}`, done: batchDone, total: records.length });
runSqlite3(dbPath, buildThreadUpsertSql(batch, target, cols, { updateAllProviders: false }));
}
runSqlite3(dbPath, `BEGIN;\nUPDATE threads SET model_provider=${sql(target)} WHERE COALESCE(model_provider,'')<>${sql(target)};\nCOMMIT;\n`);
// 兜底:强制把 rollout 已确认有用户消息的线程 has_user_event 拉起来
if (forceUeSql) runSqlite3(dbPath, "BEGIN;\n" + forceUeSql + "\nCOMMIT;\n");
const clearSubagentSql = forceSubagentNoUserEventSql(cols);
if (clearSubagentSql) runSqlite3(dbPath, "BEGIN;\n" + clearSubagentSql + "\nCOMMIT;\n");
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);
@@ -1172,4 +1334,10 @@ async function syncProviderToCurrent(log = () => {}, onProgress = () => {}) {
};
}
module.exports = { syncProviderToCurrent, codexRunning, targetProvider, setCodexDataDir };
module.exports = {
syncProviderToCurrent,
repairLegacySessionVisibility,
codexRunning,
targetProvider,
setCodexDataDir,
};