修复 Codex rollout 历史合并并重新打包
This commit is contained in:
204
server.js
204
server.js
@@ -4423,6 +4423,108 @@ function isPersistenceNoticeMessage(message) {
|
||||
return !!(message && typeof message === 'object' && message.ccwebPersistenceNotice === true);
|
||||
}
|
||||
|
||||
function stableMessageHash(value) {
|
||||
return crypto.createHash('sha256').update(String(value || '')).digest('hex').slice(0, 24);
|
||||
}
|
||||
|
||||
function messageReplyToRequestId(message) {
|
||||
return String(
|
||||
message?.replyToRequestId
|
||||
|| message?.crossConversation?.replyToRequestId
|
||||
|| message?.crossConversation?.replyRequestId
|
||||
|| '',
|
||||
).trim();
|
||||
}
|
||||
|
||||
function ensureStableMessageId(message) {
|
||||
if (!message || typeof message !== 'object' || Array.isArray(message)) return message;
|
||||
if (String(message.id || '').trim()) return message;
|
||||
const explicitId = String(message.clientMessageId || message.messageId || '').trim();
|
||||
const replyToRequestId = messageReplyToRequestId(message);
|
||||
const stableId = explicitId
|
||||
|| (replyToRequestId ? `reply:${replyToRequestId}` : '')
|
||||
|| (message.codexAppTurnKey ? `codexapp:${message.codexAppTurnKey}` : '')
|
||||
|| (message.nativeTurnKey ? `native:${message.nativeTurnKey}` : '');
|
||||
if (stableId) return { ...message, id: stableId };
|
||||
const content = typeof message.content === 'string' ? message.content : JSON.stringify(message.content || '');
|
||||
const identity = [message.role || '', message.timestamp || message.createdAt || '', content].join('\u001f');
|
||||
return { ...message, id: `message:${stableMessageHash(identity)}` };
|
||||
}
|
||||
|
||||
function normalizeHistoryMessage(message) {
|
||||
return ensureStableMessageId(message);
|
||||
}
|
||||
|
||||
function messagesEquivalent(left, right) {
|
||||
if (!left || !right || left.role !== right.role) return false;
|
||||
const leftId = String(left.id || '').trim();
|
||||
const rightId = String(right.id || '').trim();
|
||||
if (leftId && rightId && leftId === rightId) return true;
|
||||
if (!leftId.startsWith('message:') || !rightId.startsWith('message:')) return false;
|
||||
const leftContent = typeof left.content === 'string' ? left.content.trim() : JSON.stringify(left.content || '');
|
||||
const rightContent = typeof right.content === 'string' ? right.content.trim() : JSON.stringify(right.content || '');
|
||||
return !!leftContent
|
||||
&& leftContent === rightContent
|
||||
&& String(left.timestamp || '') === String(right.timestamp || '');
|
||||
}
|
||||
|
||||
function mergeHistorySegments(nativePrefix, snapshot) {
|
||||
const prefix = Array.isArray(nativePrefix) ? nativePrefix : [];
|
||||
const tail = Array.isArray(snapshot) ? snapshot : [];
|
||||
const snapshotIds = new Set(tail.map((message) => String(message?.id || '').trim()).filter(Boolean));
|
||||
const merged = [];
|
||||
const prefixIds = new Set();
|
||||
for (const message of prefix) {
|
||||
const id = String(message?.id || '').trim();
|
||||
if (id && snapshotIds.has(id)) continue;
|
||||
if (id && prefixIds.has(id)) continue;
|
||||
if (id) prefixIds.add(id);
|
||||
merged.push(message);
|
||||
}
|
||||
const snapshotPositions = new Map();
|
||||
for (const message of tail) {
|
||||
const id = String(message?.id || '').trim();
|
||||
if (id && snapshotPositions.has(id)) {
|
||||
merged[snapshotPositions.get(id)] = message;
|
||||
continue;
|
||||
}
|
||||
if (id) snapshotPositions.set(id, merged.length);
|
||||
merged.push(message);
|
||||
}
|
||||
return merged;
|
||||
}
|
||||
|
||||
function mergeNativeHistoryWithSnapshot(session, persistedMessages, nativeMessages) {
|
||||
const snapshot = (Array.isArray(persistedMessages) ? persistedMessages : []).map(normalizeHistoryMessage);
|
||||
const native = (Array.isArray(nativeMessages) ? nativeMessages : []).map(normalizeHistoryMessage);
|
||||
if (snapshot.length === 0) return native.length > 0 ? { messages: mergeHistorySegments(native, []), confirmed: true } : { messages: snapshot, confirmed: true };
|
||||
if (native.length === 0) return { messages: snapshot, confirmed: false };
|
||||
|
||||
// 从尾部寻找快照与 native 的最长有序重叠,允许 native 中存在旧版本额外事件。
|
||||
let nativeCursor = native.length - 1;
|
||||
let snapshotCursor = snapshot.length - 1;
|
||||
while (nativeCursor >= 0 && snapshotCursor >= 0) {
|
||||
if (messagesEquivalent(native[nativeCursor], snapshot[snapshotCursor])) {
|
||||
nativeCursor -= 1;
|
||||
snapshotCursor -= 1;
|
||||
continue;
|
||||
}
|
||||
nativeCursor -= 1;
|
||||
}
|
||||
if (snapshotCursor < 0) {
|
||||
const overlapStart = nativeCursor + 1;
|
||||
return { messages: mergeHistorySegments(native.slice(0, overlapStart), snapshot), confirmed: true };
|
||||
}
|
||||
|
||||
const baseIndex = Number(session?.historySnapshotBaseIndex);
|
||||
const hasSnapshotBoundary = Number.isSafeInteger(baseIndex) && baseIndex > 0;
|
||||
if (hasSnapshotBoundary && native.length >= baseIndex) {
|
||||
const prefix = native.slice(0, Math.min(baseIndex, native.length));
|
||||
return { messages: mergeHistorySegments(prefix, snapshot), confirmed: true };
|
||||
}
|
||||
return { messages: snapshot, confirmed: false };
|
||||
}
|
||||
|
||||
function sessionHistoryCacheKey(session) {
|
||||
if (!session) return '';
|
||||
if (isCodexLikeSession(session)) return `${getSessionAgent(session)}:${session.codexAppThreadId || session.codexThreadId || ''}`;
|
||||
@@ -4491,14 +4593,32 @@ function resolveSessionHistory(session) {
|
||||
const persistedMessages = persisted.filter((message) => !isPersistenceNoticeMessage(message));
|
||||
const notice = persisted.some(isPersistenceNoticeMessage);
|
||||
if (!notice) {
|
||||
return { messages: persisted, source: 'snapshot', available: true, recoverable: false };
|
||||
return {
|
||||
messages: persistedMessages.map(normalizeHistoryMessage),
|
||||
source: 'snapshot',
|
||||
available: true,
|
||||
recoverable: false,
|
||||
};
|
||||
}
|
||||
|
||||
const native = loadNativeSessionHistory(session);
|
||||
if (native?.messages?.length > persistedMessages.length) {
|
||||
return { messages: native.messages, source: 'native', available: true, recoverable: true };
|
||||
const merged = mergeNativeHistoryWithSnapshot(session, persistedMessages, native?.messages);
|
||||
if (merged.confirmed && merged.messages.length >= persistedMessages.length) {
|
||||
return {
|
||||
messages: merged.messages,
|
||||
source: merged.messages.length > persistedMessages.length ? 'merged' : 'snapshot',
|
||||
available: true,
|
||||
recoverable: merged.messages.length > persistedMessages.length,
|
||||
};
|
||||
}
|
||||
return { messages: persisted, source: 'snapshot', available: false, recoverable: false };
|
||||
if (native && !merged.confirmed) {
|
||||
plog('WARN', 'session_history_merge_unconfirmed', {
|
||||
sessionId: String(session?.id || '').slice(0, 8),
|
||||
persistedCount: persistedMessages.length,
|
||||
nativeCount: Array.isArray(native.messages) ? native.messages.length : 0,
|
||||
});
|
||||
}
|
||||
return { messages: persistedMessages.map(normalizeHistoryMessage), source: 'snapshot', available: false, recoverable: false };
|
||||
}
|
||||
|
||||
function normalizeAgent(agent) {
|
||||
@@ -4593,10 +4713,11 @@ function normalizeSession(session) {
|
||||
if (Array.isArray(session.messages)) {
|
||||
session.messages = session.messages.map((message) => {
|
||||
if (!message || typeof message !== 'object') return message;
|
||||
const normalizedMessage = ensureStableMessageId(message);
|
||||
if (message.attachments) {
|
||||
return { ...message, attachments: normalizeMessageAttachments(message.attachments) };
|
||||
return { ...normalizedMessage, attachments: normalizeMessageAttachments(message.attachments) };
|
||||
}
|
||||
return message;
|
||||
return normalizedMessage;
|
||||
});
|
||||
}
|
||||
return session;
|
||||
@@ -5409,7 +5530,7 @@ function sanitizeMessagesForPersist(messages, limits = {}) {
|
||||
const list = Array.isArray(messages) ? messages : [];
|
||||
const maxMessages = limits.maxMessages || SESSION_PERSIST_MAX_MESSAGES;
|
||||
const selected = list.length > maxMessages ? list.slice(-maxMessages) : list;
|
||||
const output = selected.map((message) => sanitizeMessageForPersist(message, limits));
|
||||
const output = selected.map((message) => sanitizeMessageForPersist(ensureStableMessageId(message), limits));
|
||||
if (list.length > selected.length) {
|
||||
output.unshift({
|
||||
role: 'system',
|
||||
@@ -5451,6 +5572,26 @@ function sanitizeSessionForPersist(session, limits = {}) {
|
||||
if (skipKeys.has(key)) continue;
|
||||
if (key === 'messages') {
|
||||
output.messages = sanitizeMessagesForPersist(value, limits);
|
||||
const sourceMessages = Array.isArray(value) ? value : [];
|
||||
const persistedMessages = sourceMessages.filter((message) => !isPersistenceNoticeMessage(message));
|
||||
const existingBaseIndex = Number(session?.historySnapshotBaseIndex);
|
||||
const existingCount = Number(session?.historySnapshotCount);
|
||||
if (sourceMessages.length > (limits.maxMessages || SESSION_PERSIST_MAX_MESSAGES)) {
|
||||
const maxPersistedMessages = limits.maxMessages || SESSION_PERSIST_MAX_MESSAGES;
|
||||
const logicalDroppedCount = Math.max(0, persistedMessages.length - maxPersistedMessages);
|
||||
output.historySnapshotBaseIndex = Number.isSafeInteger(existingBaseIndex) && existingBaseIndex > 0
|
||||
? existingBaseIndex + logicalDroppedCount
|
||||
: logicalDroppedCount;
|
||||
output.historySnapshotCount = Math.max(0, output.messages.length - (output.messages.some(isPersistenceNoticeMessage) ? 1 : 0));
|
||||
} else if (Number.isSafeInteger(existingBaseIndex) && existingBaseIndex > 0) {
|
||||
output.historySnapshotBaseIndex = existingBaseIndex;
|
||||
output.historySnapshotCount = Number.isSafeInteger(existingCount) && existingCount > 0
|
||||
? Math.max(existingCount, persistedMessages.length)
|
||||
: persistedMessages.length;
|
||||
} else {
|
||||
output.historySnapshotBaseIndex = 0;
|
||||
output.historySnapshotCount = persistedMessages.length;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
output[key] = sanitizePersistValue(value, {
|
||||
@@ -7883,6 +8024,8 @@ function deliverCrossConversationReply(requestId) {
|
||||
draft.status = 'delivering';
|
||||
});
|
||||
const replyMessage = {
|
||||
id: replyMessageId,
|
||||
replyToRequestId: requestId,
|
||||
role: 'assistant',
|
||||
content: replyContent,
|
||||
timestamp: now,
|
||||
@@ -11035,7 +11178,10 @@ function createPersistentConversationSession(args = {}, options = {}) {
|
||||
|
||||
function buildSessionInfoPayload(session) {
|
||||
const waitState = crossConversationWaitState(session.id);
|
||||
const messages = sanitizeMessagesForTransport(session.messages || []);
|
||||
const history = typeof resolveSessionHistory === 'function'
|
||||
? resolveSessionHistory(session)
|
||||
: { messages: session.messages || [], source: 'snapshot', available: true };
|
||||
const messages = sanitizeMessagesForTransport(history.messages || []);
|
||||
const taskTracking = taskTrackingSnapshotForSession(session);
|
||||
return {
|
||||
type: 'session_info',
|
||||
@@ -11057,6 +11203,8 @@ function buildSessionInfoPayload(session) {
|
||||
hasUnread: false,
|
||||
historyPending: false,
|
||||
isRunning: false,
|
||||
historyAvailable: history.available !== false,
|
||||
historySource: history.source || 'snapshot',
|
||||
waitingOnChildren: waitState.waitingOnChildren,
|
||||
pendingReplyCount: waitState.pendingReplyCount,
|
||||
readyReplyCount: waitState.readyReplyCount,
|
||||
@@ -11868,21 +12016,33 @@ function handleMessage(ws, msg, options = {}) {
|
||||
|
||||
let persistedUserMessage = null;
|
||||
if (!hideInHistory) {
|
||||
persistedUserMessage = {
|
||||
role: 'user',
|
||||
content: displayTextValue,
|
||||
attachments: savedAttachments,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
const giteaSource = normalizeGiteaSource(options.giteaSource);
|
||||
if (giteaSource) persistedUserMessage.giteaSource = giteaSource;
|
||||
if (options.crossConversation) {
|
||||
persistedUserMessage.crossConversation = options.crossConversation;
|
||||
const clientMessageId = String(msg?.clientMessageId || '').trim();
|
||||
const replyToRequestId = String(options.crossConversation?.replyToRequestId || '').trim();
|
||||
const messageId = clientMessageId || (replyToRequestId ? `reply-request:${replyToRequestId}` : crypto.randomUUID());
|
||||
const existingUserMessage = Array.isArray(session.messages)
|
||||
? session.messages.find((message) => message?.role === 'user' && message?.id === messageId)
|
||||
: null;
|
||||
if (existingUserMessage) {
|
||||
persistedUserMessage = existingUserMessage;
|
||||
} else {
|
||||
persistedUserMessage = {
|
||||
id: messageId,
|
||||
role: 'user',
|
||||
content: displayTextValue,
|
||||
attachments: savedAttachments,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
if (clientMessageId) persistedUserMessage.clientMessageId = clientMessageId;
|
||||
const giteaSource = normalizeGiteaSource(options.giteaSource);
|
||||
if (giteaSource) persistedUserMessage.giteaSource = giteaSource;
|
||||
if (options.crossConversation) {
|
||||
persistedUserMessage.crossConversation = options.crossConversation;
|
||||
}
|
||||
if (decoratorResolution.mentions.length > 0) {
|
||||
persistedUserMessage.composerMentions = decoratorResolution.mentions;
|
||||
}
|
||||
session.messages.push(persistedUserMessage);
|
||||
}
|
||||
if (decoratorResolution.mentions.length > 0) {
|
||||
persistedUserMessage.composerMentions = decoratorResolution.mentions;
|
||||
}
|
||||
session.messages.push(persistedUserMessage);
|
||||
}
|
||||
session.updated = new Date().toISOString();
|
||||
saveSession(session);
|
||||
|
||||
Reference in New Issue
Block a user