fix: 消除会话列表解析阻塞并按需加载工具详情
This commit is contained in:
230
server.js
230
server.js
@@ -18,6 +18,7 @@ const {
|
||||
} = require('./lib/gitea-workflow-codex');
|
||||
const { createSessionSearchIndex } = require('./lib/session-search-index');
|
||||
const { createSessionHistoryStore } = require('./lib/session-history-store');
|
||||
const { summarizeTools, toolSignature } = require('./lib/session-query-transport');
|
||||
const { createUsageStatisticsIndex, UsageStatisticsError } = require('./lib/usage-statistics');
|
||||
const {
|
||||
createGiteaWebhookReceiver,
|
||||
@@ -5963,6 +5964,7 @@ const SESSION_META_PREVIEW_TOP_LEVEL_KEYS = new Set([
|
||||
'cwd',
|
||||
'giteaSource',
|
||||
]);
|
||||
const sessionMetadataCache = new Map();
|
||||
|
||||
function isJsonWhitespace(ch) {
|
||||
return ch === ' ' || ch === '\n' || ch === '\r' || ch === '\t';
|
||||
@@ -6079,35 +6081,75 @@ function parseJsonTopLevelPrefixFields(text) {
|
||||
return fields;
|
||||
}
|
||||
|
||||
function previousNonWhitespaceChar(text, index) {
|
||||
for (let i = index - 1; i >= 0; i -= 1) {
|
||||
if (!isJsonWhitespace(text[i])) return text[i];
|
||||
function skipJsonWhitespaceBackward(text, index) {
|
||||
while (index >= 0 && isJsonWhitespace(text[index])) index -= 1;
|
||||
return index;
|
||||
}
|
||||
|
||||
function readJsonStringStart(text, end) {
|
||||
if (text[end] !== '"') return -1;
|
||||
for (let i = end - 1; i >= 0; i -= 1) {
|
||||
if (text[i] !== '"') continue;
|
||||
let before = i - 1;
|
||||
while (before >= 0 && text[before] === '\\') before -= 1;
|
||||
if ((i - 1 - before) % 2 === 0) return i;
|
||||
i = before + 1;
|
||||
}
|
||||
return '';
|
||||
return -1;
|
||||
}
|
||||
|
||||
function readJsonValueStart(text, end) {
|
||||
if (text[end] === '"') return readJsonStringStart(text, end);
|
||||
if (text[end] === '}' || text[end] === ']') {
|
||||
const stack = [];
|
||||
for (let i = end; i >= 0; i -= 1) {
|
||||
const ch = text[i];
|
||||
if (ch === '"') {
|
||||
i = readJsonStringStart(text, i);
|
||||
if (i < 0) return -1;
|
||||
} else if (ch === '}' || ch === ']') {
|
||||
stack.push(ch);
|
||||
} else if (ch === '{' || ch === '[') {
|
||||
const close = stack.pop();
|
||||
if ((ch === '{' && close !== '}') || (ch === '[' && close !== ']')) return -1;
|
||||
if (!stack.length) return i;
|
||||
}
|
||||
}
|
||||
return -1;
|
||||
}
|
||||
let start = end;
|
||||
while (start >= 0 && !isJsonWhitespace(text[start]) && !'[:,{}'.includes(text[start])) start -= 1;
|
||||
const value = text.slice(start + 1, end + 1);
|
||||
return /^(?:true|false|null|-?(?:0|[1-9]\d*)(?:\.\d+)?(?:[eE][+-]?\d+)?)$/.test(value) ? start + 1 : -1;
|
||||
}
|
||||
|
||||
function parseJsonTopLevelSuffixFields(text) {
|
||||
let end = text.length;
|
||||
while (end > 0 && isJsonWhitespace(text[end - 1])) end -= 1;
|
||||
if (text[end - 1] !== '}') return {};
|
||||
const content = text.slice(0, end - 1);
|
||||
const keyPattern = /"(id|title|updated|updatedAt|created|pinnedAt|titleSource|createdFrom|hasUnread|agent|cwd)"\s*:/g;
|
||||
let match;
|
||||
while ((match = keyPattern.exec(content))) {
|
||||
const start = match.index;
|
||||
const prev = previousNonWhitespaceChar(content, start);
|
||||
if (prev && prev !== ',' && prev !== '{') continue;
|
||||
const fields = {};
|
||||
let i = skipJsonWhitespaceBackward(text, text.length - 1);
|
||||
if (text[i] !== '}') return fields;
|
||||
i = skipJsonWhitespaceBackward(text, i - 1);
|
||||
// 从根对象的结束符反向扫描一次;嵌套工具内容只跳过,绝不逐个候选重试解析整个后缀。
|
||||
while (i >= 0 && text[i] !== '{') {
|
||||
const valueEnd = i + 1;
|
||||
const valueStart = readJsonValueStart(text, i);
|
||||
if (valueStart < 0) break;
|
||||
i = skipJsonWhitespaceBackward(text, valueStart - 1);
|
||||
if (text[i] !== ':') break;
|
||||
const keyEnd = skipJsonWhitespaceBackward(text, i - 1);
|
||||
const keyStart = readJsonStringStart(text, keyEnd);
|
||||
if (keyStart < 0) break;
|
||||
i = skipJsonWhitespaceBackward(text, keyStart - 1);
|
||||
if (i >= 0 && text[i] !== ',' && text[i] !== '{') break;
|
||||
try {
|
||||
const parsed = JSON.parse(`{${content.slice(start)}}`);
|
||||
if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) continue;
|
||||
const fields = {};
|
||||
for (const key of SESSION_META_PREVIEW_TOP_LEVEL_KEYS) {
|
||||
if (Object.prototype.hasOwnProperty.call(parsed, key)) fields[key] = parsed[key];
|
||||
const key = JSON.parse(text.slice(keyStart, keyEnd + 1));
|
||||
if (SESSION_META_PREVIEW_TOP_LEVEL_KEYS.has(key) && !Object.prototype.hasOwnProperty.call(fields, key)) {
|
||||
fields[key] = JSON.parse(text.slice(valueStart, valueEnd));
|
||||
}
|
||||
return fields;
|
||||
} catch {}
|
||||
} catch { break; }
|
||||
if (text[i] !== ',') break;
|
||||
i = skipJsonWhitespaceBackward(text, i - 1);
|
||||
}
|
||||
return {};
|
||||
return fields;
|
||||
}
|
||||
|
||||
function readSessionPreviewTopLevelFields(preview) {
|
||||
@@ -6160,44 +6202,31 @@ function listSessionFiles() {
|
||||
return [...files];
|
||||
}
|
||||
|
||||
function refreshSessionSnapshotCache(id) {
|
||||
const revision = sessionHistoryStore.revision(id);
|
||||
if (!revision) return null;
|
||||
const filePath = sessionPath(id);
|
||||
let fd;
|
||||
let cachedRevision;
|
||||
try {
|
||||
fd = fs.openSync(filePath, 'r');
|
||||
const head = Buffer.alloc(512);
|
||||
const count = fs.readSync(fd, head, 0, head.length, 0);
|
||||
cachedRevision = /^\{\s*"historyArchiveRevision"\s*:\s*"([0-9a-f-]{36})"\s*,/.exec(head.toString('utf8', 0, count))?.[1];
|
||||
} catch (error) {
|
||||
if (error?.code !== 'ENOENT') throw error;
|
||||
} finally {
|
||||
if (fd !== undefined) fs.closeSync(fd);
|
||||
}
|
||||
if (cachedRevision === revision) return null;
|
||||
const session = loadSession(id, { count: SESSION_PERSIST_MAX_MESSAGES });
|
||||
if (!session) return null;
|
||||
const result = buildSessionJsonForPersist({ ...session, historyArchiveRevision: revision });
|
||||
try {
|
||||
writeFileAtomicSync(filePath, result.json);
|
||||
} catch (error) {
|
||||
plog('WARN', 'session_snapshot_cache_refresh_failed', { sessionId: id.slice(0, 8), error: error?.message || String(error) });
|
||||
}
|
||||
// 首次缓存失败的孤立归档仍须展示;列表只使用刚生成的有界预览。
|
||||
return { session: result.persisted, fileBytes: textByteLength(result.json) };
|
||||
}
|
||||
|
||||
function loadSessionMetaFromFile(filePath) {
|
||||
const fallbackId = path.basename(filePath, '.json');
|
||||
try {
|
||||
const refreshed = refreshSessionSnapshotCache(fallbackId);
|
||||
const stat = refreshed ? { size: refreshed.fileBytes, mtime: new Date() } : fs.statSync(filePath);
|
||||
if (refreshed || stat.size <= SESSION_META_FULL_PARSE_MAX_BYTES) {
|
||||
const session = normalizeSession(refreshed?.session || JSON.parse(fs.readFileSync(filePath, 'utf8')));
|
||||
// 列表只读归档元数据,不读取消息块,也不为陈旧快照重建、裁剪或序列化历史。
|
||||
const revision = sessionHistoryStore.revision(fallbackId);
|
||||
let stat;
|
||||
try { stat = fs.statSync(filePath); } catch (error) {
|
||||
if (!revision || error?.code !== 'ENOENT') throw error;
|
||||
stat = { size: 0, mtime: new Date(0) };
|
||||
}
|
||||
const cacheKey = `${revision || 'snapshot'}:${stat.size}:${stat.mtimeMs || 0}:${stat.ctimeMs || 0}`;
|
||||
const cached = sessionMetadataCache.get(filePath);
|
||||
if (cached?.key === cacheKey) return cached.meta;
|
||||
const remember = (meta) => {
|
||||
// 仅保留列表字段;每次检查磁盘修订,改标题、未读和运行目标不会命中过期缓存。
|
||||
if (sessionMetadataCache.size >= 4096) sessionMetadataCache.delete(sessionMetadataCache.keys().next().value);
|
||||
sessionMetadataCache.set(filePath, { key: cacheKey, meta });
|
||||
return meta;
|
||||
};
|
||||
const archived = revision ? sessionHistoryStore.loadMetadata(fallbackId) : null;
|
||||
if (archived || stat.size <= SESSION_META_FULL_PARSE_MAX_BYTES) {
|
||||
const source = archived || JSON.parse(fs.readFileSync(filePath, 'utf8'));
|
||||
const session = normalizeSession({ ...source, messages: [] });
|
||||
const cwd = session.cwd || '';
|
||||
return {
|
||||
return remember({
|
||||
id: session.id || fallbackId,
|
||||
title: session.title || 'Untitled',
|
||||
updated: session.updated || session.created || stat.mtime.toISOString(),
|
||||
@@ -6211,16 +6240,16 @@ function loadSessionMetaFromFile(filePath) {
|
||||
cwd,
|
||||
giteaSource: normalizeGiteaSource(session.giteaSource),
|
||||
projectName: cwd ? path.basename(cwd.replace(/[\\/]+$/, '')) : '',
|
||||
codexAppGoalActive: isCodexThreadGoalActive(session.codexAppGoal),
|
||||
fileBytes: stat.size,
|
||||
codexAppGoalActive: isCodexThreadGoalActive(session.codexAppGoal),
|
||||
fileBytes: stat.size,
|
||||
oversized: stat.size > SESSION_LOAD_MAX_BYTES,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
const preview = readSessionPreview(filePath, stat);
|
||||
const previewFields = readSessionPreviewTopLevelFields(preview);
|
||||
const cwd = previewStringField(previewFields, 'cwd') || '';
|
||||
return {
|
||||
return remember({
|
||||
id: previewStringField(previewFields, 'id') || fallbackId,
|
||||
title: previewStringField(previewFields, 'title') || 'Untitled',
|
||||
updated: previewStringField(previewFields, 'updated') || previewStringField(previewFields, 'updatedAt') || stat.mtime.toISOString(),
|
||||
@@ -6234,11 +6263,12 @@ function loadSessionMetaFromFile(filePath) {
|
||||
cwd,
|
||||
giteaSource: normalizeGiteaSource(previewFields.giteaSource),
|
||||
projectName: cwd ? path.basename(cwd.replace(/[\\/]+$/, '')) : '',
|
||||
codexAppGoalActive: false,
|
||||
fileBytes: stat.size,
|
||||
codexAppGoalActive: false,
|
||||
fileBytes: stat.size,
|
||||
oversized: stat.size > SESSION_LOAD_MAX_BYTES,
|
||||
};
|
||||
});
|
||||
} catch (err) {
|
||||
sessionMetadataCache.delete(filePath);
|
||||
plog('WARN', 'session_meta_load_failed', {
|
||||
sessionId: fallbackId.slice(0, 8),
|
||||
error: err?.message || String(err || ''),
|
||||
@@ -11829,20 +11859,28 @@ function attachClientRequestId(payload, source = {}) {
|
||||
return requestId ? { ...payload, requestId } : payload;
|
||||
}
|
||||
|
||||
const runtimeDetailScopes = new WeakMap();
|
||||
|
||||
function summarizeRuntimePayload(sessionId, entry, payload) {
|
||||
if (!runtimeDetailScopes.has(entry)) runtimeDetailScopes.set(entry, crypto.randomUUID());
|
||||
return { ...payload, toolCalls: summarizeTools(entry.toolCalls, {
|
||||
sessionId, scope: runtimeDetailScopes.get(entry), messageId: payload.assistantMessageId || null,
|
||||
}) };
|
||||
}
|
||||
|
||||
function buildSessionRuntimeSnapshot(sessionId) {
|
||||
const classic = activeProcesses.get(sessionId);
|
||||
if (classic) {
|
||||
return { type: 'resume_generating', sessionId, text: classic.fullText || '',
|
||||
toolCalls: sanitizeToolCallsForPersist(classic.toolCalls || []) };
|
||||
return summarizeRuntimePayload(sessionId, classic, { type: 'resume_generating', sessionId, text: classic.fullText || '' });
|
||||
}
|
||||
const active = activeCodexAppTurns.get(sessionId);
|
||||
if (active) return buildCodexAppResumePayload(sessionId, active);
|
||||
if (active) return summarizeRuntimePayload(sessionId, active, buildCodexAppResumePayload(sessionId, active));
|
||||
const retained = retainedCodexAppTurns.get(sessionId);
|
||||
if (retained) return buildCodexAppResumePayload(sessionId, retained, {
|
||||
if (retained) return summarizeRuntimePayload(sessionId, retained, buildCodexAppResumePayload(sessionId, retained, {
|
||||
isRunning: false, historySaved: retained.historySaved === true,
|
||||
});
|
||||
}));
|
||||
const retry = pendingCodexCapacityRetries.get(sessionId);
|
||||
if (retry?.entry) return buildCodexAppResumePayload(sessionId, retry.entry, { retryPending: true });
|
||||
if (retry?.entry) return summarizeRuntimePayload(sessionId, retry.entry, buildCodexAppResumePayload(sessionId, retry.entry, { retryPending: true }));
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -11864,6 +11902,39 @@ function handleSessionQueryApi(req, res, url, encodedId, historyPage = false) {
|
||||
try { sessionId = decodeURIComponent(encodedId); }
|
||||
catch { return respondError(400, 'history_invalid_id', '会话 ID 无效'); }
|
||||
if (!/^[A-Za-z0-9-]+$/.test(sessionId)) return respondError(400, 'history_invalid_id', '会话 ID 无效');
|
||||
const detail = url.searchParams.get('toolDetail');
|
||||
if (detail !== null) {
|
||||
phase = 'tool_detail';
|
||||
let ref;
|
||||
try { ref = JSON.parse(detail); } catch { return respondError(400, 'tool_detail_invalid', '工具详情引用无效'); }
|
||||
if (!ref || ref.sessionId !== sessionId || !Number.isSafeInteger(ref.index) || ref.index < 0) {
|
||||
return respondError(400, 'tool_detail_invalid', '工具详情引用无效');
|
||||
}
|
||||
const entries = [activeProcesses.get(sessionId), activeCodexAppTurns.get(sessionId),
|
||||
retainedCodexAppTurns.get(sessionId), pendingCodexCapacityRetries.get(sessionId)?.entry];
|
||||
const runtime = ref.scope && entries.find(entry => entry && runtimeDetailScopes.get(entry) === ref.scope);
|
||||
let tools = runtime?.toolCalls;
|
||||
if (!tools) {
|
||||
let message;
|
||||
if (Number.isSafeInteger(ref.messageIndex) && ref.messageIndex >= 0) {
|
||||
const saved = loadSession(sessionId, { window: { start: ref.messageIndex, end: ref.messageIndex + 1 }, throwOnError: true });
|
||||
const state = saved && sessionHistoryStore.state(saved);
|
||||
message = state?.source === 'archive' ? saved.messages[0] : saved?.messages?.[ref.messageIndex];
|
||||
if (ref.messageId && normalizeHistoryMessage(message)?.id !== ref.messageId) message = null;
|
||||
} else if (ref.messageId) {
|
||||
const saved = loadSession(sessionId, { messageId: ref.messageId, throwOnError: true });
|
||||
message = saved?.messages?.find(item => item.id === ref.messageId);
|
||||
} else if (ref.scope && ref.toolId) {
|
||||
message = sessionHistoryStore.findMessage(sessionId, item => item.toolCalls?.some(tool => tool.id === ref.toolId))?.message;
|
||||
}
|
||||
tools = message?.toolCalls;
|
||||
}
|
||||
const tool = tools?.[ref.index];
|
||||
if (!tool || (ref.toolId && tool.id !== ref.toolId) || (!ref.scope && toolSignature(tool) !== ref.signature)) {
|
||||
return respondError(409, 'tool_detail_changed', '工具记录已更新,请重新打开会话后查看');
|
||||
}
|
||||
return jsonResponse(res, 200, { type: 'session_tool_detail', sessionId, requestId, tool });
|
||||
}
|
||||
const beforeValue = url.searchParams.get('before');
|
||||
const before = beforeValue === null ? undefined : Number(beforeValue);
|
||||
if (beforeValue !== null && (!/^\d+$/.test(beforeValue) || !Number.isSafeInteger(before))) {
|
||||
@@ -11878,7 +11949,10 @@ function handleSessionQueryApi(req, res, url, encodedId, historyPage = false) {
|
||||
before: historyPage ? before : undefined,
|
||||
persistedOnly: true,
|
||||
});
|
||||
const messages = sanitizeMessagesForTransport(history.messages || []);
|
||||
const messages = (history.messages || []).map((message, index) => sanitizeMessageForTransport({ ...message,
|
||||
toolCalls: summarizeTools(message.toolCalls, { sessionId, messageId: message.id || null,
|
||||
messageIndex: (history.start || 0) + index }),
|
||||
}));
|
||||
const total = history.total ?? messages.length;
|
||||
const start = history.start || 0;
|
||||
const common = {
|
||||
@@ -11945,7 +12019,7 @@ function handleSubscribeSession(ws, msg = {}) {
|
||||
type: 'session_subscribed', sessionId, isRunning: isSessionRunning(sessionId),
|
||||
historyRevision, historyChanged,
|
||||
}, msg));
|
||||
attachActiveRuntimeToWs(ws, sessionId, msg);
|
||||
attachActiveRuntimeToWs(ws, sessionId, { ...msg, toolDetails: 'deferred' });
|
||||
} catch (error) {
|
||||
plog('WARN', 'session_subscribe_failed', { sessionId: sessionId.slice(0, 8), code: error?.code || 'session_subscribe_failed' });
|
||||
wsSend(ws, attachClientRequestId({
|
||||
@@ -12106,6 +12180,8 @@ function handleMigrateSessionHistory(ws, msg = {}) {
|
||||
}
|
||||
|
||||
function attachActiveRuntimeToWs(ws, sessionId, source = {}) {
|
||||
const sendResume = (entry, payload) => wsSend(ws, attachClientRequestId(
|
||||
source.toolDetails === 'deferred' ? summarizeRuntimePayload(sessionId, entry, payload) : payload, source));
|
||||
if (activeProcesses.has(sessionId)) {
|
||||
const entry = activeProcesses.get(sessionId);
|
||||
entry.ws = ws;
|
||||
@@ -12115,12 +12191,12 @@ function attachActiveRuntimeToWs(ws, sessionId, source = {}) {
|
||||
pid: entry.pid,
|
||||
responseLen: (entry.fullText || '').length,
|
||||
});
|
||||
wsSend(ws, attachClientRequestId({
|
||||
sendResume(entry, {
|
||||
type: 'resume_generating',
|
||||
sessionId,
|
||||
text: entry.fullText || '',
|
||||
toolCalls: sanitizeToolCallsForPersist(entry.toolCalls || []),
|
||||
}, source));
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -12134,7 +12210,7 @@ function attachActiveRuntimeToWs(ws, sessionId, source = {}) {
|
||||
turnId: entry.turnId || null,
|
||||
responseLen: (entry.fullText || '').length,
|
||||
});
|
||||
wsSend(ws, attachClientRequestId(buildCodexAppResumePayload(sessionId, entry), source));
|
||||
sendResume(entry, buildCodexAppResumePayload(sessionId, entry));
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -12149,10 +12225,10 @@ function attachActiveRuntimeToWs(ws, sessionId, source = {}) {
|
||||
responseLen: (retainedEntry.fullText || '').length,
|
||||
historySaved: !!retainedEntry.historySaved,
|
||||
});
|
||||
wsSend(ws, attachClientRequestId(buildCodexAppResumePayload(sessionId, retainedEntry, {
|
||||
sendResume(retainedEntry, buildCodexAppResumePayload(sessionId, retainedEntry, {
|
||||
isRunning: false,
|
||||
historySaved: retainedEntry.historySaved === true,
|
||||
}), source));
|
||||
}));
|
||||
wsSend(ws, attachClientRequestId({ type: 'done', sessionId, historySaved: retainedEntry.historySaved === true }, source));
|
||||
return true;
|
||||
}
|
||||
@@ -12167,7 +12243,7 @@ function attachActiveRuntimeToWs(ws, sessionId, source = {}) {
|
||||
attempt: pendingRetry.attempts || 0,
|
||||
responseLen: (pendingRetry.entry.fullText || '').length,
|
||||
});
|
||||
wsSend(ws, attachClientRequestId(buildCodexAppResumePayload(sessionId, pendingRetry.entry, { retryPending: true }), source));
|
||||
sendResume(pendingRetry.entry, buildCodexAppResumePayload(sessionId, pendingRetry.entry, { retryPending: true }));
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user