674 lines
30 KiB
JavaScript
674 lines
30 KiB
JavaScript
'use strict';
|
||
|
||
const nativeFs = require('fs');
|
||
const path = require('path');
|
||
const crypto = require('crypto');
|
||
const { createJsonReader, readJsonSync, writeJsonSync, cloneJsonValue, jsonValueEqual } = require('./json-persistence');
|
||
|
||
// 跨存储实例保护仍在使用的旧修订；弱引用不延长会话或消息的寿命。
|
||
const liveBaselines = new Set();
|
||
|
||
const SESSION_RUNTIME_KEYS = new Set([
|
||
'ws', 'tailer', 'codexAppStateTimer', 'codexAppStateDirty', 'codexAppStateCleaned',
|
||
'toolOutputDeltas', 'agentMessageItems',
|
||
]);
|
||
|
||
function createSessionHistoryStore(options = {}) {
|
||
const fs = options.fs || nativeFs;
|
||
const historyDir = path.join(options.sessionsDir, '_history');
|
||
const turnsDir = path.join(historyDir, '_turns');
|
||
// 身份只留在当前存储实例，不能通过 JSON 快照伪造可信完整加载或修订。
|
||
const loadedStates = new WeakMap();
|
||
const removedIds = new Set();
|
||
const chunkMaxBytes = options.chunkMaxBytes || 2 * 1024 * 1024;
|
||
const chunkMaxMessages = options.chunkMaxMessages || 64;
|
||
let commits = 0;
|
||
|
||
function validId(value) {
|
||
const id = String(value || '');
|
||
if (!/^[a-zA-Z0-9_-]+$/.test(id) || id === '_turns') throw Object.assign(new Error('会话归档 ID 无效'), { code: 'history_invalid_id' });
|
||
return id;
|
||
}
|
||
|
||
function filePath(id) {
|
||
return path.join(historyDir, `${validId(id)}.json`);
|
||
}
|
||
|
||
function archiveDir(id) {
|
||
return path.join(historyDir, validId(id));
|
||
}
|
||
|
||
function manifestPath(id) {
|
||
return path.join(archiveDir(id), 'manifest.json');
|
||
}
|
||
|
||
function chunkPath(id, file) {
|
||
if (!/^[a-f0-9]{64}\.json$/.test(file)) throw new Error('完整会话归档块引用无效');
|
||
return path.join(archiveDir(id), 'chunks', file);
|
||
}
|
||
|
||
function stringify(value) {
|
||
return JSON.stringify(value, (key, item) => {
|
||
if (typeof item === 'bigint') return String(item);
|
||
if (item instanceof Map) return Object.fromEntries(item);
|
||
if (item instanceof Set) return [...item];
|
||
return item;
|
||
});
|
||
}
|
||
|
||
function sessionMetadata(session) {
|
||
const metadata = Object.create(null);
|
||
for (const key of Object.keys(session)) {
|
||
if (key === 'messages' || key === 'historySnapshotBaseIndex' || key === 'historySnapshotCount'
|
||
|| key === 'historyArchiveRevision' || SESSION_RUNTIME_KEYS.has(key)) continue;
|
||
Object.defineProperty(metadata, key, { value: session[key], enumerable: true });
|
||
}
|
||
return cloneJsonValue(metadata);
|
||
}
|
||
|
||
function chunkWriter(id) {
|
||
const entries = [];
|
||
let pending = [];
|
||
let bytes = 2;
|
||
function flush() {
|
||
if (!pending.length) return;
|
||
const json = `[${pending.map(item => item.json).join(',')}]`;
|
||
const file = `${crypto.createHash('sha256').update(json).digest('hex')}.json`;
|
||
const target = chunkPath(id, file);
|
||
if (!fs.existsSync(target)) writeAtomicJson(target, null, json);
|
||
pending.forEach((item, index) => {
|
||
Object.assign(item.entry, { id: item.id || null, role: item.role, digest: item.digest, file, index, bytes: item.bytes });
|
||
entries.push(item.entry);
|
||
});
|
||
pending = [];
|
||
bytes = 2;
|
||
}
|
||
function add(message) {
|
||
const json = stringify(message);
|
||
if (json === undefined) throw new Error('完整会话消息不可序列化');
|
||
const size = Buffer.byteLength(json);
|
||
if (pending.length && (bytes + size + 1 > chunkMaxBytes || pending.length >= chunkMaxMessages)) flush();
|
||
const entry = {};
|
||
pending.push({ json, id: message?.id, role: message?.role || null,
|
||
digest: crypto.createHash('sha256').update(json).digest('hex'), bytes: size, entry });
|
||
bytes += size + 1;
|
||
// 单条超大消息仍完整保存，不能按预览限额截断。
|
||
if (bytes >= chunkMaxBytes || pending.length >= chunkMaxMessages) flush();
|
||
return entry;
|
||
}
|
||
return { add, finish() { flush(); return entries; } };
|
||
}
|
||
|
||
function migrateLegacy(id) {
|
||
if (!fs.existsSync(filePath(id))) return null;
|
||
const reader = createJsonReader(fs, filePath(id));
|
||
const record = Object.create(null);
|
||
const metadata = Object.create(null);
|
||
const writer = chunkWriter(id);
|
||
let hasMessages = false;
|
||
try {
|
||
reader.object(key => {
|
||
if (key !== 'session') { record[key] = reader.value(); return; }
|
||
reader.object(field => {
|
||
if (field === 'messages') {
|
||
hasMessages = true;
|
||
reader.array(message => { if (message?.ccwebPersistenceNotice !== true) writer.add(message); });
|
||
} else Object.defineProperty(metadata, field, { value: reader.value(), enumerable: true, configurable: true });
|
||
});
|
||
});
|
||
reader.end();
|
||
} finally {
|
||
reader.close();
|
||
}
|
||
if (record.version !== 1 || typeof record.revision !== 'string' || !record.revision
|
||
|| metadata.id !== id || !hasMessages || typeof record.complete !== 'boolean') {
|
||
throw Object.assign(new Error('完整会话归档格式无效'), { code: 'history_invalid_archive' });
|
||
}
|
||
const entries = writer.finish();
|
||
const manifest = { version: 2, revision: record.revision, complete: record.complete,
|
||
resetRevision: record.resetRevision || null, session: sessionMetadata(metadata), entries };
|
||
// 旧单文件保留为迁移来源；所有块成功落盘后，清单才成为唯一提交点。
|
||
writeAtomicJson(manifestPath(id), manifest);
|
||
return manifest;
|
||
}
|
||
|
||
function readRecord(id) {
|
||
let record;
|
||
try {
|
||
record = JSON.parse(fs.readFileSync(manifestPath(id), 'utf8'));
|
||
} catch (error) {
|
||
if (error?.code === 'ENOENT') return migrateLegacy(id);
|
||
throw error;
|
||
}
|
||
if (record?.version !== 2 || typeof record.revision !== 'string' || !record.revision
|
||
|| record.session?.id !== id || !Array.isArray(record.entries) || typeof record.complete !== 'boolean'
|
||
|| record.entries.some(entry => !/^[a-f0-9]{64}\.json$/.test(entry.file)
|
||
|| !/^[a-f0-9]{64}$/.test(entry.digest) || !Number.isSafeInteger(entry.index) || entry.index < 0)) {
|
||
throw Object.assign(new Error('完整会话归档格式无效'), { code: 'history_invalid_archive' });
|
||
}
|
||
return record;
|
||
}
|
||
|
||
function state(session) {
|
||
const value = session && loadedStates.get(session);
|
||
if (!value) return null;
|
||
return { id: value.id, revision: value.revision, complete: value.complete, source: value.source,
|
||
resetRevision: value.resetRevision || null, start: value.start || 0,
|
||
end: value.end, total: value.manifest?.entries.length, partial: value.partial === true };
|
||
}
|
||
|
||
function register(session, manifest, start, end, partial) {
|
||
const baseline = { id: session.id, revision: manifest.revision, complete: manifest.complete,
|
||
source: 'archive', resetRevision: manifest.resetRevision || null, manifest, start, end, partial, directory: archiveDir(session.id) };
|
||
loadedStates.set(session, baseline);
|
||
liveBaselines.add(new WeakRef(baseline));
|
||
}
|
||
|
||
function entryReader(id) {
|
||
const cache = new Map();
|
||
return entry => {
|
||
let messages = cache.get(entry.file);
|
||
if (!messages) {
|
||
const json = fs.readFileSync(chunkPath(id, entry.file), 'utf8');
|
||
if (crypto.createHash('sha256').update(json).digest('hex') !== entry.file.slice(0, -5)) {
|
||
throw new Error('完整会话归档块校验失败');
|
||
}
|
||
messages = JSON.parse(json);
|
||
if (!Array.isArray(messages)) throw new Error('完整会话归档块格式无效');
|
||
cache.set(entry.file, messages);
|
||
if (cache.size > 2) cache.delete(cache.keys().next().value);
|
||
}
|
||
const message = messages[entry.index];
|
||
if (message === undefined) throw new Error('完整会话归档消息引用无效');
|
||
return message;
|
||
};
|
||
}
|
||
|
||
function loadRange(id, record, start, end, partial) {
|
||
const session = cloneJsonValue(record.session);
|
||
const read = entryReader(id);
|
||
session.messages = record.entries.slice(start, end).map(read);
|
||
session.historySnapshotBaseIndex = start;
|
||
session.historySnapshotCount = session.messages.length;
|
||
register(session, record, start, end, partial);
|
||
return session;
|
||
}
|
||
|
||
function load(id) {
|
||
id = validId(id);
|
||
const record = readRecord(id);
|
||
return record ? loadRange(id, record, 0, record.entries.length, false) : null;
|
||
}
|
||
|
||
function loadTail(id, count = 180, limits = {}) {
|
||
id = validId(id);
|
||
const record = readRecord(id);
|
||
if (!record) return null;
|
||
const end = record.entries.length;
|
||
let start = Math.max(0, end - Math.max(0, Math.trunc(count)));
|
||
if (Number.isFinite(limits.maxBytes) && limits.maxBytes > 0) {
|
||
let bytes = 0;
|
||
for (let index = end - 1; index >= start; index -= 1) {
|
||
const size = record.entries[index].bytes || 0;
|
||
if (bytes && bytes + size > limits.maxBytes) { start = index + 1; break; }
|
||
bytes += size;
|
||
}
|
||
}
|
||
return loadRange(id, record, start, end, start > 0);
|
||
}
|
||
|
||
function loadMetadata(id) {
|
||
return loadTail(id, 0);
|
||
}
|
||
|
||
function loadWindow(id, start, end = start + 1) {
|
||
id = validId(id);
|
||
const record = readRecord(id);
|
||
if (!record) return null;
|
||
const total = record.entries.length;
|
||
start = Math.max(0, Math.min(total, Math.trunc(start)));
|
||
end = Math.max(start, Math.min(total, Math.trunc(end)));
|
||
return loadRange(id, record, start, end, start > 0 || end < total);
|
||
}
|
||
|
||
function forEachMessage(id, callback) {
|
||
id = validId(id);
|
||
const record = readRecord(id);
|
||
if (!record) return;
|
||
const read = entryReader(id);
|
||
for (let index = 0; index < record.entries.length; index += 1) callback(read(record.entries[index]), index);
|
||
}
|
||
|
||
function findMessage(id, predicate, before = Infinity) {
|
||
id = validId(id);
|
||
const record = readRecord(id);
|
||
if (!record) return null;
|
||
const read = entryReader(id);
|
||
for (let index = Math.min(record.entries.length, before) - 1; index >= 0; index -= 1) {
|
||
const message = read(record.entries[index]);
|
||
if (predicate(message, index)) return { message, index };
|
||
}
|
||
return null;
|
||
}
|
||
|
||
function loadHistoryPage(id, before, count = 24) {
|
||
id = validId(id);
|
||
const record = readRecord(id);
|
||
if (!record) return null;
|
||
const total = record.entries.length;
|
||
const end = Number.isFinite(before) ? Math.max(0, Math.min(total, Math.trunc(before))) : total;
|
||
const start = Math.max(0, end - Math.max(0, Math.trunc(count)));
|
||
const read = entryReader(id);
|
||
return { messages: record.entries.slice(start, end).map(read), total, start, end,
|
||
complete: record.complete, revision: record.revision };
|
||
}
|
||
|
||
function messageIndex(id, messageId) {
|
||
const record = readRecord(validId(id));
|
||
return record ? record.entries.findIndex(entry => entry.id === messageId) : -1;
|
||
}
|
||
|
||
function countMessages(id, role) {
|
||
const record = readRecord(validId(id));
|
||
return record ? record.entries.filter(entry => !role || entry.role === role).length : null;
|
||
}
|
||
|
||
function markLegacy(session) {
|
||
validId(session?.id);
|
||
loadedStates.set(session, {
|
||
id: session.id, revision: null, source: 'legacy',
|
||
complete: !session.messages?.some(message => message?.ccwebPersistenceNotice === true)
|
||
&& !(Number(session.historySnapshotBaseIndex) > 0),
|
||
});
|
||
return session;
|
||
}
|
||
|
||
function markComplete(session, options = {}) {
|
||
const previous = loadedStates.get(session);
|
||
if (!previous || previous.id !== session.id) {
|
||
throw Object.assign(new Error('只能确认可信加载会话的历史状态'), { code: 'history_untrusted_source' });
|
||
}
|
||
const confirmed = { ...previous, complete: true, reset: options.reset === true, partial: false };
|
||
loadedStates.set(session, confirmed);
|
||
if (confirmed.manifest) liveBaselines.add(new WeakRef(confirmed));
|
||
session.historySnapshotBaseIndex = 0;
|
||
session.historySnapshotCount = session.messages.length;
|
||
return () => loadedStates.set(session, previous);
|
||
}
|
||
|
||
function equal(left, right) {
|
||
return jsonValueEqual(left, right);
|
||
}
|
||
|
||
function plainObject(value) {
|
||
return value && typeof value === 'object' && !Array.isArray(value);
|
||
}
|
||
|
||
function identifiedArray(value) {
|
||
return Array.isArray(value) && value.every(item => item && typeof item.id === 'string' && item.id);
|
||
}
|
||
|
||
function mergeIdentifiedArray(baseline, incoming, current) {
|
||
const baseById = new Map(baseline.map(item => [item.id, item]));
|
||
const incomingById = new Map(incoming.map(item => [item.id, item]));
|
||
const currentById = new Map(current.map(item => [item.id, item]));
|
||
const output = [];
|
||
const seen = new Set();
|
||
for (const item of current) {
|
||
if (seen.has(item.id)) continue;
|
||
seen.add(item.id);
|
||
const base = baseById.get(item.id);
|
||
const next = incomingById.get(item.id);
|
||
if (base && !next && equal(item, base)) continue;
|
||
output.push(next ? mergeValue(base, next, item) : item);
|
||
}
|
||
// 只补本次相对加载基线新增的气泡；后来已经删除的旧气泡不能复活。
|
||
for (let index = 0; index < incoming.length; index += 1) {
|
||
const item = incoming[index];
|
||
if (baseById.has(item.id) || currentById.has(item.id) || output.some(existing => existing.id === item.id)) continue;
|
||
let position = -1;
|
||
for (let next = index + 1; next < incoming.length; next += 1) {
|
||
position = output.findIndex(existing => existing.id === incoming[next].id);
|
||
if (position >= 0) break;
|
||
}
|
||
if (position < 0) {
|
||
for (let previous = index - 1; previous >= 0; previous -= 1) {
|
||
const anchor = output.findIndex(existing => existing.id === incoming[previous].id);
|
||
if (anchor >= 0) { position = anchor + 1; break; }
|
||
}
|
||
}
|
||
output.splice(position < 0 ? output.length : position, 0, item);
|
||
}
|
||
return output;
|
||
}
|
||
|
||
function mergeValue(baseline, incoming, current) {
|
||
if (equal(incoming, baseline)) return current;
|
||
if (equal(current, baseline) || equal(incoming, current)) return incoming;
|
||
if ((baseline === undefined || identifiedArray(baseline)) && identifiedArray(incoming) && identifiedArray(current)) {
|
||
return mergeIdentifiedArray(baseline || [], incoming, current);
|
||
}
|
||
if (plainObject(incoming) && plainObject(current) && (baseline === undefined || plainObject(baseline))) {
|
||
const output = Object.create(null);
|
||
const base = baseline || {};
|
||
for (const key of new Set([...Object.keys(base), ...Object.keys(incoming), ...Object.keys(current)])) {
|
||
const own = (value) => Object.prototype.hasOwnProperty.call(value, key) ? value[key] : undefined;
|
||
const value = mergeValue(own(base), own(incoming), own(current));
|
||
if (value !== undefined) output[key] = value;
|
||
}
|
||
return output;
|
||
}
|
||
// 同字段冲突保留已经提交的值；互不重叠的正文、工具与元数据修改会逐字段合并。
|
||
return current;
|
||
}
|
||
|
||
function synchronizeValue(target, persisted, root = false) {
|
||
if (Array.isArray(target) && Array.isArray(persisted)) {
|
||
const identified = identifiedArray(target) && identifiedArray(persisted);
|
||
const byId = identified ? new Map(target.map(item => [item.id, item])) : null;
|
||
const updated = persisted.map((item, index) => synchronizeValue(byId ? byId.get(item.id) : target[index], item));
|
||
target.length = updated.length;
|
||
for (let index = 0; index < updated.length; index += 1) target[index] = updated[index];
|
||
return target;
|
||
}
|
||
const prototype = target && typeof target === 'object' ? Object.getPrototypeOf(target) : undefined;
|
||
if ((prototype === Object.prototype || prototype === null) && plainObject(persisted)) {
|
||
for (const key of Object.keys(target)) {
|
||
if ((!root || !SESSION_RUNTIME_KEYS.has(key)) && !Object.prototype.hasOwnProperty.call(persisted, key)) delete target[key];
|
||
}
|
||
for (const [key, value] of Object.entries(persisted)) {
|
||
const existing = Object.prototype.hasOwnProperty.call(target, key) ? target[key] : undefined;
|
||
Object.defineProperty(target, key, {
|
||
value: synchronizeValue(existing, value), enumerable: true, writable: true, configurable: true,
|
||
});
|
||
}
|
||
return target;
|
||
}
|
||
return persisted;
|
||
}
|
||
|
||
function writeAtomicJson(target, record, json) {
|
||
fs.mkdirSync(path.dirname(target), { recursive: true });
|
||
const temporary = `${target}.${process.pid}.${crypto.randomUUID()}.tmp`;
|
||
let fd;
|
||
try {
|
||
fd = fs.openSync(temporary, 'wx', 0o600);
|
||
if (json === undefined) writeJsonSync(fs, fd, record);
|
||
else fs.writeFileSync(fd, json, 'utf8');
|
||
fs.fsyncSync(fd);
|
||
fs.closeSync(fd);
|
||
fd = undefined;
|
||
fs.renameSync(temporary, target);
|
||
} finally {
|
||
if (fd !== undefined) {
|
||
try { fs.closeSync(fd); } catch {}
|
||
}
|
||
try { fs.unlinkSync(temporary); } catch {}
|
||
}
|
||
}
|
||
|
||
function turnPath(id) {
|
||
return path.join(turnsDir, `${validId(id)}.json`);
|
||
}
|
||
|
||
function saveTurn(id, state) {
|
||
writeAtomicJson(turnPath(id), state);
|
||
}
|
||
|
||
function loadTurn(id) {
|
||
try {
|
||
const state = readJsonSync(fs, turnPath(id));
|
||
if (state?.sessionId !== id || state.agent !== 'codexapp') throw new Error('完整运行态归档无效');
|
||
return state;
|
||
} catch (error) {
|
||
if (error?.code === 'ENOENT') return null;
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
function removeTurn(id) {
|
||
try { fs.unlinkSync(turnPath(id)); } catch (error) {
|
||
if (error?.code !== 'ENOENT') throw error;
|
||
}
|
||
}
|
||
|
||
function listTurnIds() {
|
||
try {
|
||
return fs.readdirSync(turnsDir).filter(name => /^[a-zA-Z0-9_-]+\.json$/.test(name)).map(name => name.slice(0, -5));
|
||
} catch (error) {
|
||
if (error?.code === 'ENOENT') return [];
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
function save(session, options = {}) {
|
||
const id = validId(session?.id);
|
||
const previous = loadedStates.get(session);
|
||
if (removedIds.has(id)) throw Object.assign(new Error('已删除会话不能由旧对象重新保存'), { code: 'history_removed' });
|
||
if (previous?.source === 'archive' && !fs.existsSync(manifestPath(id))) {
|
||
throw Object.assign(new Error('完整会话归档已移除，拒绝旧对象重新创建'), { code: 'history_removed' });
|
||
}
|
||
// 同修订只核对文件头，可信清单已经是不可变基线；旧修订才重新读取当前清单。
|
||
const current = previous?.source === 'archive' && revision(id) === previous.revision
|
||
? previous.manifest : readRecord(id);
|
||
if (current) {
|
||
if (previous?.source !== 'archive' || previous.id !== id) {
|
||
throw Object.assign(new Error('受限预览不能覆盖完整会话归档'), { code: 'history_untrusted_source' });
|
||
}
|
||
} else if (previous?.source === 'archive') {
|
||
throw Object.assign(new Error('完整会话归档已移除，拒绝旧对象重新创建'), { code: 'history_removed' });
|
||
}
|
||
const complete = current?.complete || (previous ? previous.complete
|
||
: !session.messages?.some(message => message?.ccwebPersistenceNotice === true)
|
||
&& !(Number(session.historySnapshotBaseIndex) > 0));
|
||
if (!Array.isArray(session.messages)) throw new Error('完整会话消息必须为数组');
|
||
let metadata = sessionMetadata(session);
|
||
let taskTrackingPreserved = null;
|
||
const currentTracking = current?.session.taskTracking;
|
||
if (!options.taskBoardMetadata && currentTracking && typeof currentTracking === 'object') {
|
||
const persistedVersion = Number.isSafeInteger(currentTracking.version) ? currentTracking.version : 0;
|
||
const incomingVersion = Number.isSafeInteger(metadata.taskTracking?.version) ? metadata.taskTracking.version : -1;
|
||
if (incomingVersion < persistedVersion) {
|
||
metadata.taskTracking = cloneJsonValue(currentTracking);
|
||
taskTrackingPreserved = { incomingVersion, persistedVersion };
|
||
}
|
||
}
|
||
const rebased = !!current && previous.revision !== current.revision;
|
||
const baseEntries = previous?.manifest?.entries || [];
|
||
const viewedEntries = previous?.partial ? baseEntries.slice(previous.start, previous.end) : baseEntries;
|
||
const descriptors = [];
|
||
const positions = new Map();
|
||
for (const message of session.messages) {
|
||
if (message?.ccwebPersistenceNotice === true) continue;
|
||
const json = stringify(message);
|
||
if (json === undefined) throw new Error('完整会话消息不可序列化');
|
||
const descriptor = { id: message?.id || null, digest: crypto.createHash('sha256').update(json).digest('hex'), message };
|
||
// 同 ID 的最新片段保留最初位置，不同气泡即使同 turn 也分别保存。
|
||
if (descriptor.id && positions.has(descriptor.id)) descriptors[positions.get(descriptor.id)] = descriptor;
|
||
else {
|
||
if (descriptor.id) positions.set(descriptor.id, descriptors.length);
|
||
descriptors.push(descriptor);
|
||
}
|
||
}
|
||
let ordered = descriptors;
|
||
let start = previous?.partial ? previous.start : 0;
|
||
let end;
|
||
const read = entryReader(id);
|
||
const currentEntries = current?.entries || [];
|
||
const resetChanged = rebased && (current.resetRevision || null) !== (previous.resetRevision || null);
|
||
if (rebased && !resetChanged) {
|
||
if ([viewedEntries, descriptors, currentEntries].some(entries => !entries.every(entry => typeof entry.id === 'string' && entry.id))) {
|
||
throw Object.assign(new Error('旧修订缺少可靠消息 ID，拒绝覆盖新历史'), { code: 'history_stale_unidentified' });
|
||
}
|
||
if (!previous.partial && baseEntries.length > 0 && descriptors.length === 0
|
||
&& !equal(currentEntries.map(entry => entry.digest), baseEntries.map(entry => entry.digest))) {
|
||
throw Object.assign(new Error('清空前历史已经更新，请重新加载'), { code: 'history_stale_clear' });
|
||
}
|
||
const baseById = new Map(viewedEntries.map(entry => [entry.id, entry]));
|
||
const incomingById = new Map(descriptors.map(entry => [entry.id, entry]));
|
||
const currentIds = new Set(currentEntries.map(entry => entry.id));
|
||
ordered = [];
|
||
for (const entry of currentEntries) {
|
||
const base = baseById.get(entry.id);
|
||
const next = incomingById.get(entry.id);
|
||
if (!base) { ordered.push(entry); continue; }
|
||
if (!next) {
|
||
if (entry.digest !== base.digest) ordered.push(entry);
|
||
continue;
|
||
}
|
||
if (next.digest === base.digest || next.digest === entry.digest) ordered.push(entry);
|
||
else if (entry.digest === base.digest) ordered.push(next);
|
||
else ordered.push({ id: next.id, incoming: next.message, base, current: entry });
|
||
}
|
||
const orderedIds = new Set(ordered.map(entry => entry.id));
|
||
for (let index = 0; index < descriptors.length; index += 1) {
|
||
const item = descriptors[index];
|
||
if (baseById.has(item.id) || currentIds.has(item.id) || orderedIds.has(item.id)) continue;
|
||
let position = -1;
|
||
for (let next = index + 1; next < descriptors.length; next += 1) {
|
||
position = ordered.findIndex(entry => entry.id === descriptors[next].id);
|
||
if (position >= 0) break;
|
||
}
|
||
if (position < 0) {
|
||
for (let anchor = index - 1; anchor >= 0; anchor -= 1) {
|
||
const found = ordered.findIndex(entry => entry.id === descriptors[anchor].id);
|
||
if (found >= 0) { position = found + 1; break; }
|
||
}
|
||
}
|
||
ordered.splice(position < 0 ? ordered.length : position, 0, item);
|
||
orderedIds.add(item.id);
|
||
}
|
||
if (previous.partial) {
|
||
// 只协调原加载窗口及窗口内的并发新增，不能把未加载的后缀变成内存基线。
|
||
const prefixIds = new Set(baseEntries.slice(0, previous.start).map(entry => entry.id));
|
||
const suffixIds = new Set(baseEntries.slice(previous.end).map(entry => entry.id));
|
||
const lastPrefix = ordered.reduce((found, entry, index) => prefixIds.has(entry.id) ? index : found, -1);
|
||
start = lastPrefix >= 0 ? lastPrefix + 1 : Math.min(previous.start, ordered.length);
|
||
const firstSuffix = ordered.findIndex(entry => suffixIds.has(entry.id));
|
||
end = firstSuffix >= 0 ? firstSuffix : ordered.length;
|
||
if (previous.start === previous.end && descriptors.length === 0) start = end;
|
||
}
|
||
} else if (resetChanged) {
|
||
// 清空后的旧对象只同步元数据，旧线程及输出不能复活。
|
||
ordered = currentEntries;
|
||
start = previous.partial ? currentEntries.length : 0;
|
||
end = currentEntries.length;
|
||
for (const key of ['claudeSessionId', 'codexThreadId', 'codexAppThreadId']) {
|
||
if (Object.prototype.hasOwnProperty.call(previous.manifest.session, key)) metadata[key] = previous.manifest.session[key];
|
||
else delete metadata[key];
|
||
}
|
||
} else if (previous?.partial) {
|
||
const prefix = currentEntries.slice(0, previous.start);
|
||
const suffix = currentEntries.slice(previous.end);
|
||
const unseenIds = new Set([...prefix, ...suffix].filter(entry => entry.id).map(entry => entry.id));
|
||
// 重试的稳定 ID 若已存在于未加载区间，保留已提交原文和位置。
|
||
ordered = [...prefix, ...descriptors.filter(entry => !entry.id || !unseenIds.has(entry.id)), ...suffix];
|
||
end = ordered.length - suffix.length;
|
||
}
|
||
if (rebased) metadata = mergeValue(previous.manifest.session, metadata, current.session);
|
||
const writer = chunkWriter(id);
|
||
const currentById = new Map(currentEntries.filter(entry => entry.id).map(entry => [entry.id, entry]));
|
||
const entries = ordered.map((item, index) => {
|
||
if (item.file) return item;
|
||
const same = item.id ? currentById.get(item.id) : currentEntries[index];
|
||
if (same && same.digest === item.digest) return same;
|
||
const message = item.current
|
||
? mergeValue(read(item.base), cloneJsonValue(item.incoming), read(item.current))
|
||
: item.message;
|
||
return writer.add(message);
|
||
});
|
||
writer.finish();
|
||
const nextRevision = crypto.randomUUID();
|
||
const resetRevision = previous?.reset ? nextRevision : current?.resetRevision || null;
|
||
const record = { version: 2, revision: nextRevision, complete, resetRevision, session: metadata, entries };
|
||
end ??= entries.length;
|
||
let updated;
|
||
let synchronizedMetadata;
|
||
if (rebased || end - start !== session.messages.length) {
|
||
const byId = new Map(session.messages.filter(message => message?.id).map(message => [message.id, message]));
|
||
// 提交前完成所有可能失败的块读取；提交失败不修改活跃消息和可信基线。
|
||
updated = entries.slice(start, end).map(entry => ({ target: byId.get(entry.id), persisted: read(entry) }));
|
||
// 活跃元数据可以原位更新，不能与不可变清单基线共用新增的对象引用。
|
||
synchronizedMetadata = cloneJsonValue(metadata);
|
||
}
|
||
const synchronizedTracking = taskTrackingPreserved && !updated ? cloneJsonValue(metadata.taskTracking) : null;
|
||
writeAtomicJson(manifestPath(id), record);
|
||
if (updated) {
|
||
session.messages.length = updated.length;
|
||
updated.forEach((item, index) => { session.messages[index] = synchronizeValue(item.target, item.persisted); });
|
||
synchronizeValue(session, { ...synchronizedMetadata, messages: session.messages }, true);
|
||
}
|
||
if (synchronizedTracking) session.taskTracking = synchronizeValue(session.taskTracking, synchronizedTracking);
|
||
session.historySnapshotBaseIndex = start;
|
||
session.historySnapshotCount = session.messages.length;
|
||
register(session, record, start, end, start > 0 || end < entries.length);
|
||
if (++commits % 32 === 0) {
|
||
// 只回收当前清单和活跃旧修订都不再引用的块，清理失败不改变已提交结果。
|
||
try { collectChunks(id, record); } catch {}
|
||
}
|
||
return { revision: nextRevision, complete, session, rebased, taskTrackingPreserved };
|
||
}
|
||
|
||
function collectChunks(id, current) {
|
||
const files = new Set(current.entries.map(entry => entry.file));
|
||
for (const reference of liveBaselines) {
|
||
const baseline = reference.deref();
|
||
if (!baseline) { liveBaselines.delete(reference); continue; }
|
||
if (baseline.id === id && baseline.directory === archiveDir(id)) baseline.manifest.entries.forEach(entry => files.add(entry.file));
|
||
}
|
||
const directory = path.join(archiveDir(id), 'chunks');
|
||
for (const file of fs.readdirSync(directory)) {
|
||
if (/^[a-f0-9]{64}\.json$/.test(file) && !files.has(file)) fs.unlinkSync(path.join(directory, file));
|
||
}
|
||
}
|
||
|
||
function listIds() {
|
||
try {
|
||
const ids = new Set();
|
||
for (const name of fs.readdirSync(historyDir)) {
|
||
// 运行态归档属于保留目录，不参与会话归档枚举。
|
||
if (name === '_turns' || name === '_turns.json') continue;
|
||
if (/^[a-zA-Z0-9_-]+\.json$/.test(name)) ids.add(name.slice(0, -5));
|
||
else if (/^[a-zA-Z0-9_-]+$/.test(name) && fs.existsSync(manifestPath(name))) ids.add(name);
|
||
}
|
||
return [...ids];
|
||
} catch (error) {
|
||
if (error?.code === 'ENOENT') return [];
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
function revision(id) {
|
||
let fd;
|
||
try {
|
||
const recordPath = fs.existsSync(manifestPath(id)) ? manifestPath(id) : filePath(id);
|
||
fd = fs.openSync(recordPath, 'r');
|
||
const head = Buffer.alloc(256);
|
||
const count = fs.readSync(fd, head, 0, head.length, 0);
|
||
const matched = /^\{"version":[12],"revision":"([0-9a-f-]{36})",/.exec(head.toString('utf8', 0, count));
|
||
if (!matched) throw new Error('完整会话归档修订无效');
|
||
return matched[1];
|
||
} catch (error) {
|
||
if (error?.code === 'ENOENT') return null;
|
||
throw error;
|
||
} finally {
|
||
if (fd !== undefined) fs.closeSync(fd);
|
||
}
|
||
}
|
||
|
||
function remove(id) {
|
||
id = validId(id);
|
||
try { fs.unlinkSync(filePath(id)); } catch (error) {
|
||
if (error?.code !== 'ENOENT') throw error;
|
||
}
|
||
fs.rmSync(archiveDir(id), { recursive: true, force: true });
|
||
removedIds.add(id);
|
||
removeTurn(id);
|
||
}
|
||
|
||
return { load, loadTail, loadMetadata, loadWindow, forEachMessage, findMessage, loadHistoryPage, messageIndex, countMessages, save, state, markLegacy, markComplete,
|
||
listIds, revision, remove, saveTurn, loadTurn, removeTurn, listTurnIds, historyDir };
|
||
}
|
||
|
||
module.exports = { createSessionHistoryStore };
|