Files
cc-web/scripts/codexapp-retry-history-unit.js
shiyue 69b6db0f5b 修复完整会话历史与长气泡交互并更新 CentOS 7 发布包
提交当前工作区全部源码、回归脚本、任务记录及重新构建的 Bun baseline 发布包。发布单文件 MCP 初始化及前端资源一致性验证通过。
2026-10-10 23:40:16 +08:00

763 lines
43 KiB
JavaScript
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env node
'use strict';
// 抽取服务端真实重试、开轮、完成与重连函数,只把模型边界和时钟替换为隔离夹具。
// 会话及完整运行态均使用真实存储实现,所有读写限制在临时目录。
const assert = require('node:assert/strict');
const crypto = require('node:crypto');
const fs = require('node:fs');
const os = require('node:os');
const path = require('node:path');
const { inspect } = require('node:util');
const vm = require('node:vm');
const { createServerHarness } = require('./session-history-store-unit');
const { createCodexAppServerClient } = require('../lib/codex-app-server-client');
const { LIFECYCLE_EVENT_TYPES: TASK_BOARD_LIFECYCLE_EVENTS } = require('../lib/task-board-lifecycle');
const { CLASSIFICATION_EVENT_TYPES: TASK_STATUS_CLASSIFICATION_EVENTS } = require('../lib/task-board-classifier');
const REPO_DIR = path.join(__dirname, '..');
const baselineIndex = process.argv.indexOf('--baseline');
const sourcePath = baselineIndex >= 0 ? process.argv[baselineIndex + 1] : path.join(REPO_DIR, 'server.js');
assert(sourcePath, '--baseline 需要提供修复前工作区 server.js 快照路径');
const serverSource = fs.readFileSync(sourcePath, 'utf8');
const runtimePath = baselineIndex >= 0 && fs.existsSync(path.join(path.dirname(sourcePath), 'ccweb-streaming-retry-baseline-runtime.js'))
? path.join(path.dirname(sourcePath), 'ccweb-streaming-retry-baseline-runtime.js')
: path.join(REPO_DIR, 'lib/codex-app-runtime.js');
const { createCodexAppRuntime } = require(runtimePath);
function extractFunction(name, optional = false) {
const match = new RegExp(`(?:^|\\n)(?:async )?function ${name}\\(`).exec(serverSource);
if (!match && optional) return '';
assert(match, `缺少生产函数 ${name}`);
const start = match.index + (serverSource[match.index] === '\n' ? 1 : 0);
const closingLine = /\n\}/g;
closingLine.lastIndex = start;
let end;
while ((end = closingLine.exec(serverSource))) {
const candidate = serverSource.slice(start, end.index + 2);
try {
new vm.Script(`(${candidate})`, { filename: sourcePath });
return candidate;
} catch (error) {
if (!(error instanceof SyntaxError)) throw error;
}
}
throw new Error(`${name} 函数体不完整`);
}
function extractConst(name) {
const match = serverSource.match(new RegExp(`^const ${name} = .*;$`, 'm'));
assert(match, `缺少生产常量 ${name}`);
return match[0];
}
function createApi(dir, options = {}) {
const storeApi = createServerHarness(dir);
const sent = [];
const logs = [];
const starts = [];
const runtimeRef = { current: null };
const timers = new Map();
const maps = Object.fromEntries([
'activeProcesses', 'activeCodexAppTurns', 'activeCodexAppGoalCommands', 'activeCodexAppCompactions',
'codexAppGoalStates', 'pendingCompactRetries', 'pendingCodexCapacityRetries', 'retainedCodexAppTurns',
'retiredCodexAppRuntimeTurns', 'wsSessionMap', 'ccwebMcpChildThreads',
].map(name => [name, new Map()]));
const realSaveTurn = storeApi.sessionHistoryStore.saveTurn.bind(storeApi.sessionHistoryStore);
storeApi.sessionHistoryStore.saveTurn = (...args) => {
if (options.failTurnSave) { const error = new Error('夹具:完整运行态保存失败'); error.code = 'ENOSPC'; throw error; }
return realSaveTurn(...args);
};
const requiredNames = [
'readPositiveIntEnv', 'sanitizeId', 'runDir', 'writeFileAtomicSync', 'cleanRunDir',
'textByteLength', 'truncateTextValue', 'sanitizePersistValue', 'sanitizeToolCallsForPersist',
'mapToPairs', 'codexAppStatePath', 'codexAppTurnKey', 'serializeCodexAppEntry',
'buildCodexAppStateJson', 'writeCodexAppTurnState', 'persistCodexAppTurnState',
'cleanupCodexAppTurnState', 'loadCodexAppTurnState', 'hasCodexAppTurnMessage',
'appendCodexAppRecoveryNotice', 'recoverCodexAppTurnState',
'hasRuntimeOutput', 'retryTailText', 'retryPreviewValue', 'buildCodexRetryToolSummary',
'shouldUseCodexAppContinuationRetry', 'buildCodexAppContinuationRetryText',
'isCodexTransientCapacityError', 'isCodexTransientConnectionError', 'isCodexTransientRetryableError',
'codexTransientRetryDelayMs', 'cancelCodexCapacityRetry', 'shouldRetryCodexTransientFailure',
'scheduleCodexCapacityRetry', 'stableMessageHash', 'messageReplyToRequestId', 'ensureStableMessageId',
'handleMessage', 'handleCodexAppMessage', 'handleCodexAppTurnComplete', 'handleCodexAppTurnFailure',
'attachActiveRuntimeToWs', 'handleResumeSession', 'isSessionRunning',
'bindAbortSessionToWs', 'handleCodexAppAbortSession', 'handleAbort',
'codexAppRuntimeThreadId', 'codexAppRuntimeTurnId', 'findCodexAppEntryByRuntime',
'isCodexAppAdoptableRuntimeMethod', 'adoptCodexAppUnroutedTurn', 'findCodexAppRouteByRuntime',
'handleCodexAppNotification', 'handleCodexAppServerExit',
];
const constNames = [
'CODEX_APP_STATE_FULL_TEXT_MAX_CHARS', 'CODEX_APP_STATE_MAP_VALUE_MAX_CHARS', 'CODEX_APP_STATE_MAX_MAP_ENTRIES',
'CODEX_APP_STATE_MAX_BYTES', 'CODEX_APP_STATE_LOAD_MAX_BYTES', 'CODEX_APP_STATE_FILE',
'CODEX_APP_STATE_FLUSH_DELAY_MS', 'SESSION_MESSAGE_CONTENT_MAX_CHARS',
'SESSION_TOOL_INPUT_MAX_CHARS', 'SESSION_TOOL_RESULT_MAX_CHARS', 'SESSION_MAX_TOOL_CALLS_PER_MESSAGE',
'PERSIST_TRUNCATED_TEXT',
];
const optionalNames = [
'ensureCodexAppAssistantMessageId', 'buildCodexAppResumePayload', 'retirePendingCodexCapacityRetry',
'retireCodexAppRuntimeTurn', 'codexAppAssistantMessageIdForEntry',
];
const api = new Function('deps', `
const { fs, path, crypto, dir, storeApi, sent, logs, starts, timers, maps, options, runtimeRef,
TASK_BOARD_LIFECYCLE_EVENTS, TASK_STATUS_CLASSIFICATION_EVENTS } = deps;
const SESSIONS_DIR = dir;
const INITIAL_HISTORY_COUNT = 12;
const DEFAULT_CODEX_CONFIG = { retry: { intervalSeconds: 1 } };
const { activeProcesses, activeCodexAppTurns, activeCodexAppGoalCommands, activeCodexAppCompactions,
codexAppGoalStates, pendingCompactRetries, pendingCodexCapacityRetries, retainedCodexAppTurns,
retiredCodexAppRuntimeTurns, wsSessionMap, ccwebMcpChildThreads } = maps;
const sessionHistoryStore = storeApi.sessionHistoryStore;
const loadSession = id => options.failSessionLoad ? null : storeApi.loadSession(id);
const saveSession = session => options.failSessionSave ? false : storeApi.saveSession(session);
const wsSend = (ws, payload) => sent.push({ ...payload, _ws: ws });
const plog = (level, event, data) => logs.push({ level, event, ...data });
const setTimeout = (callback, delay) => {
const timer = { callback, delay, unref() {} }; timers.set(timer, timer); return timer;
};
const clearTimeout = timer => timers.delete(timer);
const getCodexRetryConfig = () => ({ mode: 'limited', maxAttempts: 3, intervalSeconds: 1, ...options.retryConfig });
const crossConversationReplyRequestIds = () => [];
const resolveMessageAttachments = () => [];
const resolveComposerDecorators = text => ({ runtimeText: text, mentions: [] });
const attachClientRequestId = (payload, source) => ({ ...payload, requestId: source?.requestId });
const isCodexAppGoalActive = () => false;
const isCodexAppSession = session => session?.agent === 'codexapp';
const getSessionAgent = session => session?.agent || 'claude';
const getRuntimeSessionId = session => session?.codexAppThreadId || null;
const getDefaultSessionCwd = () => dir;
const normalizeAgent = agent => agent || 'codexapp';
const getDefaultCodexModel = () => 'fixture-model';
const normalizeSession = session => session;
const normalizeGiteaSource = value => value || null;
const reconcilePendingCrossConversationReplies = () => false;
const flushPendingCrossConversationReplies = () => false;
const detachWsFromActiveRuntimes = () => {};
const sendSessionList = () => {};
const broadcastSessionList = () => {};
const dispatchTaskBoardLifecycle = () => Promise.resolve(null);
const enqueueTaskStatusClassification = () => Promise.resolve(null);
const resolvePendingCodexAppUserInputsForSession = () => {};
const resolvePendingCodexAppApprovalsForSession = () => {};
const resolvePendingCodexAppElicitationsForSession = () => {};
const formatRuntimeError = (agent, raw) => String(raw || '');
const completeCrossConversationReply = () => false;
const recoverCcwebMcpChildThreadsFromPersistedToolCalls = () => {};
const setRuntimeSessionId = (session, threadId) => { session.codexAppThreadId = threadId; };
const resolveSessionHistory = storeApi.resolveSessionHistory;
const sanitizeMessageForTransport = value => value;
const sanitizeMessagesForTransport = value => value;
const wss = { clients: [] };
const buildNotifyContent = async () => ({ title: '', content: '' });
const sendNotification = () => {};
const giteaWorkflowWaiters = new Map();
const settleGiteaWorkflowTurn = () => Promise.resolve(null);
let codexAppClient = options.runtimeClient || null;
let codexAppClientSignature = 'fixture-signature';
const pauseCodexAppGoalForAbort = () => false;
const cancelCodexAppGoalCommand = () => false;
const cancelCodexAppCompaction = () => false;
const codexAppInputFromMessage = text => text ? [{ type: 'text', text }] : [];
const normalizeCodexAppThreadId = value => String(value || '').trim();
const getCodexAppClient = () => ({ client: options.runtimeClient });
const codexAppThreadParams = () => ({});
const ensureCodexAppMcpReadyForThread = async () => {};
const codexAppTurnParams = (session, input, threadId, clientUserMessageId) => ({ input, threadId, clientUserMessageId });
const codexAppRuntime = { processCodexAppNotification: (...args) => runtimeRef.current.processCodexAppNotification(...args) };
const handleCodexAppGoalNotification = () => false;
const handleCodexAppMcpStartupStatusNotification = () => false;
const shouldLogCodexAppUnroutedNotification = () => false;
const syncCcwebMcpChildAgentsFromCollabItem = () => {};
const processCcwebMcpChildNotification = () => ({ changed: false });
const sendCcwebMcpChildAgentUpdate = () => {};
const findViewingSessionWs = id => [...wsSessionMap].find(([, sessionId]) => sessionId === id)?.[0] || null;
const findCodexAppSessionByThreadId = threadId => {
for (const id of new Set(wsSessionMap.values())) {
const session = loadSession(id);
if (session?.codexAppThreadId === threadId) return { sessionId: id, session };
}
return null;
};
// 仅替换真正的模型边界;调度器、hideInHistory 与新 entry 的交接全部执行生产函数。
let startCodexAppTurn = async (sessionId, input) => {
const entry = activeCodexAppTurns.get(sessionId);
starts.push({ sessionId, input });
if (options.failStart) throw new Error('夹具:线程启动失败');
if (options.startTurn) { await options.startTurn(sessionId, input, entry); return; }
entry.turnId = 'fixture-turn-' + starts.length;
persistCodexAppTurnState(sessionId, entry, { immediate: true });
};
${requiredNames.map(name => extractFunction(name)).join('\n')}
${optionalNames.map(name => extractFunction(name, true)).join('\n')}
${constNames.map(name => extractConst(name)).join('\n')}
if (options.realStart) startCodexAppTurn = (${extractFunction('startCodexAppTurn')});
return {
handleMessage, handleCodexAppMessage, handleCodexAppTurnComplete, handleCodexAppTurnFailure,
scheduleCodexCapacityRetry, cancelCodexCapacityRetry, attachActiveRuntimeToWs, handleResumeSession,
handleAbort, writeCodexAppTurnState, loadCodexAppTurnState, persistCodexAppTurnState,
recoverCodexAppTurnState, codexAppTurnKey, isSessionRunning, findCodexAppEntryByRuntime,
handleCodexAppNotification, handleCodexAppServerExit,
};
`)({ fs, path, crypto, dir, storeApi, sent, logs, starts, timers, maps, options, runtimeRef,
TASK_BOARD_LIFECYCLE_EVENTS, TASK_STATUS_CLASSIFICATION_EVENTS });
const runtime = createCodexAppRuntime({ wsSend: (ws, payload) => sent.push({ ...payload, _ws: ws }),
loadSession: storeApi.loadSession, saveSession: storeApi.saveSession });
runtimeRef.current = runtime;
async function fireRetry(sessionId) {
const retry = maps.pendingCodexCapacityRetries.get(sessionId);
assert(retry?.timer, '应已安排自动重试计时器');
timers.delete(retry.timer);
retry.timer.callback();
await Promise.resolve();
await Promise.resolve();
}
return { ...api, ...maps, runtime, storeApi, sent, logs, starts, timers, fireRetry, options };
}
const ERROR = 'Our servers are currently overloaded. Please try again later.';
const A = 'A正文首段唯一\n' + '甲'.repeat(220000) + '\nA正文尾段唯一';
const B = 'B正文唯一\n' + '乙'.repeat(15000) + '\nB正文结束唯一';
const C = 'C正文唯一\n' + '丙'.repeat(15000) + '\nC正文结束唯一';
function seed(api, id = 'retry-history', text = '继续长任务') {
const session = { id, title: '重试历史夹具', agent: 'codexapp', codexAppThreadId: 'fixture-thread',
model: 'fixture-model', permissionMode: 'yolo', cwd: '/fixture', messages: [] };
assert.equal(api.storeApi.saveSession(session), true);
const ws = { readyState: 1, label: 'initial' };
assert.equal(api.handleMessage(ws, { type: 'message', sessionId: id, agent: 'codexapp',
clientMessageId: 'original-user', text, mode: 'yolo' }).ok, true);
return { id, ws };
}
function emit(api, id, text, label, reusedId = false) {
const entry = api.activeCodexAppTurns.get(id);
assert(entry, '应有活动 entry');
const base = { threadId: entry.threadId, turnId: entry.turnId };
const agentId = reusedId ? 'agent-fixed' : 'agent-' + label;
const toolId = reusedId ? 'tool-fixed' : 'tool-' + label;
for (let offset = 0; offset < text.length; offset += 8192) {
api.runtime.processCodexAppNotification(entry, { method: 'item/agentMessage/delta',
params: { ...base, itemId: agentId, delta: text.slice(offset, offset + 8192) } }, id);
}
api.runtime.processCodexAppNotification(entry, { method: 'item/completed',
params: { ...base, item: { id: agentId, type: 'agentMessage', text } } }, id);
api.runtime.processCodexAppNotification(entry, { method: 'item/started',
params: { ...base, item: { id: toolId, type: 'commandExecution',
command: 'fixture-' + label + '-' + '输入'.repeat(10000), status: 'inProgress' } } }, id);
api.runtime.processCodexAppNotification(entry, { method: 'item/commandExecution/outputDelta',
params: { ...base, itemId: toolId, delta: '工具' + label + '结果' + '输出'.repeat(20000) } }, id);
api.writeCodexAppTurnState(id, entry);
return entry;
}
function assertToolOrder(tools, labels) {
assert.equal(tools.length, labels.length, '工具调用数量不能缩水或重复');
labels.forEach((label, index) => assert(tools[index].result.includes('工具' + label + '结果'), '工具结果必须按开轮顺序保留'));
assert.equal(new Set(tools.map(tool => tool.id)).size, labels.length, '跨轮同原生 ID 必须独立保存');
}
function assertOnceInOrder(text, parts) {
let end = -1;
for (const part of parts) {
const at = text.indexOf(part);
assert(at > end, '正文应完整保留并按 A → B → C 排序');
assert.equal(text.indexOf(part, at + part.length), -1, '同一段正文不能重复');
end = at + part.length - 1;
}
}
let passed = 0;
const failures = [];
async function check(name, callback) {
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ccweb-retry-history-'));
try {
await callback(dir);
passed += 1;
console.log(`通过:${name}`);
} catch (error) {
failures.push(name);
console.error(`失败:${name}:${inspect({ name: error.name, message: error.message.slice(0, 600),
stack: error.stack?.split('\n').filter(line => /^\s+at /.test(line)).slice(0, 3),
actual: error.actual, expected: error.expected,
}, { depth: 1, maxArrayLength: 3, maxStringLength: 160 })}`);
} finally {
fs.rmSync(dir, { recursive: true, force: true });
}
}
async function run() {
if (baselineIndex < 0 && !process.argv.includes('--only-handoff')) {
for (const startOutcome of ['response', 'timeout', 'exit']) {
const label = startOutcome === 'response'
? '取消启动时原生通知先于 RPC 返回,旧 A 不得进入或结束新 B'
: startOutcome === 'timeout'
? '取消启动 RPC 超时后保留门禁,晚到 ID 仅中断 A 且 B 正常完成'
: '取消启动未知 ID 时 app-server 明确退出,提交 A 前文并释放 B 门禁';
await check(label, async dir => {
const deferred = () => {
let resolve, reject;
const promise = new Promise((done, fail) => { resolve = done; reject = fail; });
return { promise, resolve, reject };
};
const aResponse = deferred();
const bResponse = deferred();
const nativeStarts = [];
const nativeInterrupts = [];
let clientRunning = true;
const runtimeClient = {
start: async () => { clientRunning = true; },
isRunning: () => clientRunning,
request: async (method, params) => {
if (method === 'thread/resume') return { thread: { id: params.threadId } };
if (method === 'turn/interrupt') { nativeInterrupts.push(params); return {}; }
assert.equal(method, 'turn/start', '夹具只替换线程 RPC 边界');
nativeStarts.push(params);
return nativeStarts.length === 1 ? aResponse.promise : bResponse.promise;
},
};
const api = createApi(dir, { realStart: true, runtimeClient });
const pump = async () => { for (let count = 0; count < 20; count += 1) await Promise.resolve(); };
// 路由、增量累积、完整运行态提交和完成函数均来自生产代码。
const receive = notification => {
const routed = api.findCodexAppEntryByRuntime(notification.params);
api.handleCodexAppNotification(notification);
return !!routed;
};
const { id, ws } = seed(api, 'cancel-start-race', '启动任务 A');
await pump();
assert.equal(nativeStarts.length, 1, 'A 的真实 turn/start RPC 应已发出但尚未返回');
assert.equal(api.activeCodexAppTurns.get(id).turnId, null);
if (startOutcome === 'exit') {
// 延续任务在新原生 ID 确认前已经拥有上轮正文,退出也不能丢弃它。
const carriedEntry = api.activeCodexAppTurns.get(id);
carriedEntry.fullText = '已经产生的 A 完整前文';
api.persistCodexAppTurnState(id, carriedEntry, { immediate: true });
}
api.handleAbort(ws, { sessionId: id });
const nextMessage = { sessionId: id, agent: 'codexapp', clientMessageId: 'race-b-user', text: '新任务 B' };
const beforeIdentity = api.handleMessage(ws, nextMessage);
await pump();
assert.equal(beforeIdentity.ok, false, '旧 A 身份未知时必须阻止同线程 B 开始');
assert.equal(nativeStarts.length, 1, '取消启动的 A 获取精确身份前不能发第二个 turn/start');
if (startOutcome !== 'response') {
aResponse.reject(new Error('夹具:turn/start RPC request timed out'));
await pump();
assert.equal(api.activeCodexAppTurns.get(id)?.turnId, null, 'RPC 超时不代表旧 A 没有启动,未知 ID 门禁应保留');
assert.equal(api.handleMessage(ws, nextMessage).ok, false, 'RPC 超时后 B 仍不能绕过未知身份门禁');
}
if (startOutcome === 'exit') {
clientRunning = false;
api.handleCodexAppServerExit('fixture-signature', { code: 1, stderr: '夹具:app-server 已明确退出' });
assert.equal(api.activeCodexAppTurns.has(id), false, 'app-server 明确退出后取消 A 不应永久占用门禁');
assert.equal(createServerHarness(dir).loadSession(id).messages.find(message => message.role === 'assistant')?.content,
'已经产生的 A 完整前文', '释放门禁前须提交 A 既有正文');
assert.equal(nativeInterrupts.length, 0, '已退出的进程无需请求未知 turn interrupt');
} else {
receive({ method: 'turn/started', params: { threadId: 'fixture-thread', turn: { id: 'native-A', status: 'inProgress' } } });
receive({ method: 'item/agentMessage/delta', params: { threadId: 'fixture-thread', turnId: 'native-A', itemId: 'agent-A', delta: '旧 A 的迟到正文' } });
if (startOutcome === 'response') aResponse.resolve({ turn: { id: 'native-A' } });
await pump();
assert(nativeInterrupts.some(params => params.turnId === 'native-A'), '旧 A 身份确认后必须按精确回合身份中断');
}
assert.equal(api.handleMessage(ws, nextMessage).ok, true, '旧 A 身份处理后 B 可以正常开始');
await pump();
assert.equal(nativeStarts.length, 2);
const bEntry = api.activeCodexAppTurns.get(id);
receive({ method: 'turn/started', params: { threadId: 'fixture-thread', turn: { id: 'native-B', status: 'inProgress' } } });
const bText = 'B 第一段完整正文\n' + '乙'.repeat(9000) + '\nB 最后一段完整正文';
receive({ method: 'item/agentMessage/delta', params: { threadId: 'fixture-thread', turnId: 'native-B', itemId: 'agent-B', delta: bText } });
if (startOutcome !== 'exit') {
assert.equal(receive({ method: 'turn/completed', params: { threadId: 'fixture-thread', turn: { id: 'native-A', status: 'interrupted' } } }), false);
}
assert.strictEqual(api.activeCodexAppTurns.get(id), bEntry, '旧 A 的迟到完成不能结束 B');
assert.equal(bEntry.fullText, bText, 'B 正文不能混入旧 A 通知');
bResponse.resolve({ turn: { id: 'native-B' } });
await pump();
receive({ method: 'item/completed', params: { threadId: 'fixture-thread', turnId: 'native-B', item: { id: 'agent-B', type: 'agentMessage', text: bText } } });
receive({ method: 'turn/completed', params: { threadId: 'fixture-thread', turn: { id: 'native-B', status: 'completed' } } });
const session = createServerHarness(dir).loadSession(id);
const replies = session.messages.filter(message => message.role === 'assistant' && message.content.includes('B 第一段完整正文'));
assert.equal(replies.length, 1, 'B 完成后只能归档一次');
assert.equal(replies[0].content, bText, 'B 完成与存储重开后前后文完整');
assert.equal(session.messages.filter(message => message.role === 'user' && message.id === 'race-b-user').length, 1);
assert.equal(api.activeCodexAppTurns.has(id), false);
if (startOutcome !== 'exit') {
assert.equal(receive({ method: 'turn/completed', params: { threadId: 'fixture-thread', turn: { id: 'native-A', status: 'interrupted' } } }), false);
}
});
}
if (process.argv.includes('--only-start-race')) return;
}
await check('真实自动重试开轮必须继承超过提示和快照上限的 A 正文及工具映射', async dir => {
const api = createApi(dir);
const { id } = seed(api);
const original = emit(api, id, A, 'A');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
await api.fireRetry(id);
const resumed = api.activeCodexAppTurns.get(id);
assert(resumed?.fullText.includes(A), '真实 schedule → handleMessage → handleCodexAppMessage 丢失了重试前 A 正文');
assert.deepEqual(resumed.toolCalls, original.toolCalls, '工具调用不得丢失或只剩预览');
assert.deepEqual(Array.from(resumed.toolOutputDeltas), Array.from(original.toolOutputDeltas));
assert.deepEqual(Array.from(resumed.agentMessageItems), Array.from(original.agentMessageItems));
assert.equal(resumed.codexRetry.attempt, 1);
assert.equal(api.storeApi.loadSession(id).messages.filter(message => message.role === 'user').length, 1);
});
if (process.argv.includes('--only-handoff')) return;
await check('A → B → C 两次重试后最终助手归档重开仍各保留一次且用户不重复', async dir => {
const api = createApi(dir);
const { id } = seed(api);
emit(api, id, A, 'A');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
await api.fireRetry(id);
emit(api, id, B, 'B');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
await api.fireRetry(id);
const entry = emit(api, id, C, 'C');
assert.equal(entry.codexRetry.attempt, 2, '重试次数必须单调增长');
api.handleCodexAppTurnComplete(id);
const reloaded = createServerHarness(dir).loadSession(id);
const assistants = reloaded.messages.filter(message => message.role === 'assistant');
assert.equal(assistants.length, 1, '一个逻辑任务只保存一个累积助手气泡');
assertOnceInOrder(assistants[0].content, [A, B, C]);
assertToolOrder(assistants[0].toolCalls, ['A', 'B', 'C']);
for (const tool of assistants[0].toolCalls) assert(tool.result.length > 32768, '最终工具结果必须完整');
assert.equal(reloaded.messages.filter(message => message.role === 'user').length, 1);
assert.equal(api.storeApi.sessionHistoryStore.loadTurn(id), null, '完成提交后才清理完整运行态');
});
await check('等待重试期间及续跑启动后重连均提供全部已流出的正文与工具', async dir => {
const api = createApi(dir);
const { id } = seed(api);
const first = emit(api, id, A, 'A');
api.attachActiveRuntimeToWs({ readyState: 1, label: 'before-error' }, id);
const originalId = api.sent.findLast(message => message.type === 'resume_generating')?.assistantMessageId;
assert(originalId, '累积气泡必须有稳定身份');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
assert.equal(api.isSessionRunning(id), true, '等待重试仍属于运行中的逻辑任务');
api.handleResumeSession({ readyState: 1, label: 'wait-reconnect' }, { sessionId: id, requestId: 'waiting' });
let resume = api.sent.findLast(message => message.type === 'resume_generating');
assert.equal(resume?.text, first.fullText, '等待期刷新/重连不能清空或截断正文');
assert.deepEqual(resume.toolCalls, first.toolCalls);
assert.equal(resume.assistantMessageId, originalId, '等待期必须复用同一累积气泡身份');
await api.fireRetry(id);
emit(api, id, B, 'B');
assert.equal(api.attachActiveRuntimeToWs({ readyState: 1, label: 'running-reconnect' }, id), true);
resume = api.sent.findLast(message => message.type === 'resume_generating');
assertOnceInOrder(resume.text, [A, B]);
assertToolOrder(resume.toolCalls, ['A', 'B']);
assert.equal(resume.assistantMessageId, originalId, '新重试 turn 必须复用累积气泡身份');
api.handleCodexAppTurnComplete(id);
const assistant = createServerHarness(dir).loadSession(id).messages.find(message => message.role === 'assistant');
assert.equal(assistant.id, originalId, '归档及重连必须指向同一个助手气泡');
});
await check('等待期完整运行态存储重开保留 A 正文及两个映射', async dir => {
const api = createApi(dir);
const { id } = seed(api);
const entry = emit(api, id, A, 'A');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
const state = createServerHarness(dir).sessionHistoryStore.loadTurn(id);
assert.equal(state?.fullText, A, '安排重试前不能删除唯一完整运行态');
assert.deepEqual(state.toolCalls, entry.toolCalls);
assert.deepEqual(state.toolOutputDeltas, Array.from(entry.toolOutputDeltas));
assert.deepEqual(state.agentMessageItems, Array.from(entry.agentMessageItems));
});
await check('重试耗尽保存所有已产生正文,且不会再安排新轮', async dir => {
const api = createApi(dir, { retryConfig: { maxAttempts: 1 } });
const { id } = seed(api);
emit(api, id, A, 'A');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
await api.fireRetry(id);
emit(api, id, B, 'B');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
assert.equal(api.pendingCodexCapacityRetries.has(id), false);
assert.equal(api.activeCodexAppTurns.has(id), false);
const assistant = createServerHarness(dir).loadSession(id).messages.find(message => message.role === 'assistant');
assertOnceInOrder(assistant?.content || '', [A, B]);
});
await check('用户取消等待中的重试仍提交已产生正文', async dir => {
const api = createApi(dir);
const { id, ws } = seed(api);
emit(api, id, A, 'A');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
api.handleAbort(ws, { sessionId: id });
assert.equal(api.pendingCodexCapacityRetries.has(id), false);
const assistant = createServerHarness(dir).loadSession(id).messages.find(message => message.role === 'assistant');
assert.equal(assistant?.content, A, '取消等待不得删除尚未归档正文');
});
await check('重试前完整运行态保存失败时保留内存与上次落盘正文', async dir => {
const options = {};
const api = createApi(dir, options);
const { id } = seed(api);
const entry = emit(api, id, A, 'A');
entry.fullText += '\n最新未落盘正文';
options.failTurnSave = true;
api.handleCodexAppTurnFailure(id, new Error(ERROR));
const state = createServerHarness(dir).sessionHistoryStore.loadTurn(id);
assert.equal(state?.fullText, A, '保存失败不能清理此前完整运行态');
const retained = api.activeCodexAppTurns.get(id) || api.pendingCodexCapacityRetries.get(id)?.entry || api.retainedCodexAppTurns.get(id);
assert(retained?.fullText.includes('最新未落盘正文'), '保存失败不能丢弃唯一内存正文');
});
await check('续跑启动失败仍归档此前正文', async dir => {
const options = {};
const api = createApi(dir, options);
const { id } = seed(api);
emit(api, id, A, 'A');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
options.failStart = true;
await api.fireRetry(id);
const assistant = createServerHarness(dir).loadSession(id).messages.find(message => message.role === 'assistant');
assert.equal(assistant?.content, A);
});
await check('最终会话保存失败仍留下可重开的 A/B/C 完整运行态', async dir => {
const options = {};
const api = createApi(dir, options);
const { id } = seed(api);
for (const [text, label] of [[A, 'A'], [B, 'B']]) {
emit(api, id, text, label);
api.handleCodexAppTurnFailure(id, new Error(ERROR));
await api.fireRetry(id);
}
emit(api, id, C, 'C');
options.failSessionSave = true;
api.handleCodexAppTurnComplete(id);
const state = createServerHarness(dir).sessionHistoryStore.loadTurn(id);
assertOnceInOrder(state?.fullText || '', [A, B, C]);
assertToolOrder(state.toolCalls, ['A', 'B', 'C']);
assert.equal(api.sent.findLast(message => message.type === 'done')?.historySaved, false);
});
await check('普通新用户轮次从空正文开始而不继承上一气泡', async dir => {
const api = createApi(dir);
const { id, ws } = seed(api);
emit(api, id, A, 'A');
api.handleCodexAppTurnComplete(id);
assert.equal(api.handleMessage(ws, { sessionId: id, clientMessageId: 'next-user', text: '新任务' }).ok, true);
const next = api.activeCodexAppTurns.get(id);
assert.equal(next.fullText, '');
assert.deepEqual(next.toolCalls, []);
assert.equal(next.toolOutputDeltas.size, 0);
assert.equal(next.agentMessageItems.size, 0);
});
await check('跨轮复用原生 item ID 不会覆盖工具或误拼已完成助手消息', async dir => {
const api = createApi(dir);
const { id } = seed(api);
for (const [text, label] of [[A, 'A'], [B, 'B']]) {
emit(api, id, text, label, true);
api.handleCodexAppTurnFailure(id, new Error(ERROR));
await api.fireRetry(id);
}
const entry = emit(api, id, C, 'C', true);
assertOnceInOrder(entry.fullText, [A, B, C]);
assertToolOrder(entry.toolCalls, ['A', 'B', 'C']);
assert.equal(entry.agentMessageItems.size, 3);
assert.equal(entry.toolOutputDeltas.size, 3);
api.handleCodexAppTurnComplete(id);
const assistant = createServerHarness(dir).loadSession(id).messages.find(message => message.role === 'assistant');
assertOnceInOrder(assistant.content, [A, B, C]);
assertToolOrder(assistant.toolCalls, ['A', 'B', 'C']);
});
await check('未完成消息的空白前缀及完整映射跨重试落盘且交接不丢', async dir => {
const api = createApi(dir);
const { id } = seed(api);
const oldEntry = emit(api, id, A, 'A');
api.runtime.processCodexAppNotification(oldEntry, { method: 'item/agentMessage/delta',
params: { threadId: oldEntry.threadId, turnId: oldEntry.turnId, itemId: 'pending-prefix', delta: '\n ' } }, id);
assert.equal(oldEntry.agentMessagePendingPrefixes.size, 1);
api.handleCodexAppTurnFailure(id, new Error(ERROR));
const state = createServerHarness(dir).sessionHistoryStore.loadTurn(id);
assert.deepEqual(state?.agentMessagePendingPrefixes, Array.from(oldEntry.agentMessagePendingPrefixes));
await api.fireRetry(id);
const resumed = api.activeCodexAppTurns.get(id);
assert.deepEqual(Array.from(resumed.agentMessagePendingPrefixes), Array.from(oldEntry.agentMessagePendingPrefixes));
assert.deepEqual(Array.from(resumed.agentMessageItems), Array.from(oldEntry.agentMessageItems));
});
await check('最终保存失败后重连能看完整正文,存储恢复前不得启动新用户轮', async dir => {
const options = {};
const api = createApi(dir, options);
const { id, ws } = seed(api);
const output = emit(api, id, A, 'A');
output.fullText += '\n最新完成段';
options.failSessionSave = true;
options.failTurnSave = true;
api.handleCodexAppTurnComplete(id);
assert(api.retainedCodexAppTurns.get(id)?.fullText.includes('最新完成段'), '双保存失败仍须保留最新内存正文');
api.attachActiveRuntimeToWs({ readyState: 1 }, id);
assert.equal(api.sent.findLast(message => message.type === 'resume_generating')?.text, output.fullText);
assert.equal(api.sent.findLast(message => message.type === 'done')?.historySaved, false);
assert.equal(api.handleMessage(ws, { sessionId: id, clientMessageId: 'next-user', text: '下一任务' }).ok, false);
assert.equal(api.starts.length, 1);
options.failSessionSave = false;
options.failTurnSave = false;
assert.equal(api.handleMessage(ws, { sessionId: id, clientMessageId: 'next-user', text: '下一任务' }).ok, true);
const reloaded = createServerHarness(dir).loadSession(id);
assert.equal(reloaded.messages.find(message => message.role === 'assistant')?.content, output.fullText);
assert.equal(reloaded.messages.filter(message => message.role === 'user').length, 2);
assert.equal(api.activeCodexAppTurns.get(id).fullText, '');
});
await check('等待期计时触发时会话暂不可读仍保留 A 的完整运行态及内存', async dir => {
const options = {};
const api = createApi(dir, options);
const { id } = seed(api);
emit(api, id, A, 'A');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
options.failSessionLoad = true;
await api.fireRetry(id);
assert.equal(createServerHarness(dir).sessionHistoryStore.loadTurn(id)?.fullText, A);
const held = api.retainedCodexAppTurns.get(id) || api.pendingCodexCapacityRetries.get(id)?.entry;
assert.equal(held?.fullText, A);
assert.equal(api.starts.length, 1, '会话暂时不可读时不能启动空白新线程');
});
await check('等待期收到新用户任务先提交 A,旧重试取消且新气泡从空开始', async dir => {
const api = createApi(dir);
const { id, ws } = seed(api);
emit(api, id, A, 'A');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
const oldTimer = api.pendingCodexCapacityRetries.get(id)?.timer;
assert(oldTimer);
assert.equal(api.handleMessage(ws, { sessionId: id, text: '新用户任务', clientMessageId: 'new-during-wait' }).ok, true);
assert.equal(api.pendingCodexCapacityRetries.has(id), false);
assert.equal(api.activeCodexAppTurns.get(id).fullText, '');
const reloaded = createServerHarness(dir).loadSession(id);
assert.deepEqual(reloaded.messages.map(message => message.role), ['user', 'assistant', 'user']);
assert.equal(reloaded.messages[1].content, A);
oldTimer.callback();
await Promise.resolve();
assert.equal(api.starts.length, 2, '已取消的计时器不能再启动旧任务');
});
await check('超过映射及工具预览数量的 105 项完整状态跨重试与归档保留', async dir => {
const api = createApi(dir);
const { id } = seed(api);
const first = api.activeCodexAppTurns.get(id);
const base = { threadId: first.threadId, turnId: first.turnId };
for (let index = 0; index < 105; index += 1) {
api.runtime.processCodexAppNotification(first, { method: 'item/agentMessage/delta',
params: { ...base, itemId: 'agent-many-' + index, delta: '正文段落-' + index } }, id);
api.runtime.processCodexAppNotification(first, { method: 'item/started',
params: { ...base, item: { id: 'tool-many-' + index, type: 'commandExecution', command: 'fixture-' + index } } }, id);
api.runtime.processCodexAppNotification(first, { method: 'item/commandExecution/outputDelta',
params: { ...base, itemId: 'tool-many-' + index, delta: '输出-' + index } }, id);
}
api.handleCodexAppTurnFailure(id, new Error(ERROR));
const state = createServerHarness(dir).sessionHistoryStore.loadTurn(id);
assert.equal(state?.toolCalls.length, 105);
assert.equal(state.toolOutputDeltas.length, 105);
assert.equal(state.agentMessageItems.length, 105);
await api.fireRetry(id);
const resumed = api.activeCodexAppTurns.get(id);
assert.equal(resumed.toolCalls.length, 105);
assert.equal(resumed.toolOutputDeltas.size, 105);
assert.equal(resumed.agentMessageItems.size, 105);
api.attachActiveRuntimeToWs({ readyState: 1 }, id);
assert.equal(api.sent.findLast(message => message.type === 'resume_generating').toolCalls.length, 105);
api.handleCodexAppTurnComplete(id);
const assistant = createServerHarness(dir).loadSession(id).messages.find(message => message.role === 'assistant');
assert.equal(assistant.toolCalls.length, 105);
assert.equal(assistant.content, first.fullText);
});
await check('等待期进程重开从完整运行态恢复 A/B,一次归档且重复恢复不再追加', async dir => {
const api = createApi(dir);
const { id } = seed(api);
emit(api, id, A, 'A');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
await api.fireRetry(id);
emit(api, id, B, 'B');
api.handleCodexAppTurnFailure(id, new Error(ERROR));
const savedState = createServerHarness(dir).sessionHistoryStore.loadTurn(id);
assertOnceInOrder(savedState?.fullText || '', [A, B]);
const reopened = createApi(dir);
assert.equal(reopened.recoverCodexAppTurnState(id), true);
let session = createServerHarness(dir).loadSession(id);
let assistants = session.messages.filter(message => message.role === 'assistant');
assert.equal(assistants.length, 1);
assertOnceInOrder(assistants[0].content, [A, B]);
assertToolOrder(assistants[0].toolCalls, ['A', 'B']);
assert.equal(assistants[0].id, savedState.assistantMessageId);
assert.equal(assistants[0].interrupted, true);
assert.equal(session.messages.filter(message => message.role === 'user').length, 1);
// 模拟运行态清理失败后下次重启再次读到同一状态,恢复入口必须幂等。
reopened.storeApi.sessionHistoryStore.saveTurn(id, savedState);
assert.equal(reopened.recoverCodexAppTurnState(id), true);
session = createServerHarness(dir).loadSession(id);
assistants = session.messages.filter(message => message.role === 'assistant');
assert.equal(assistants.length, 1);
assert.equal(reopened.storeApi.sessionHistoryStore.loadTurn(id), null);
});
if (baselineIndex >= 0 || process.argv.includes('--skip-mock')) return;
await check('真实 mock 协议三轮续接:A/B/C 正文、复用 ID 工具及用户归档完整', async dir => {
const options = {};
const api = createApi(dir, options);
const client = createCodexAppServerClient({ command: process.execPath,
args: [path.join(__dirname, 'mock-codex-app-server.js'), 'app-server'], cwd: dir,
env: { ...process.env, CODEX_HOME: path.join(dir, 'codex-home') },
onNotification: notification => {
const entry = api.activeCodexAppTurns.get('mock-history');
if (!entry) return;
const result = api.runtime.processCodexAppNotification(entry, notification, 'mock-history');
if (result.done) api.handleCodexAppTurnComplete('mock-history');
},
});
const waitFor = async predicate => {
const deadline = Date.now() + 5000;
while (!predicate()) {
assert(Date.now() < deadline, '隔离 mock 事件等待超过 5 秒');
await new Promise(resolve => setTimeout(resolve, 10));
}
};
try {
await client.start();
await client.request('thread/resume', { threadId: 'fixture-thread' });
options.startTurn = async (sessionId, input) => {
await client.request('turn/start', { threadId: 'fixture-thread', input });
};
const { id } = seed(api, 'mock-history', 'codexapp retry full history');
for (let attempt = 1; attempt <= 2; attempt += 1) {
await waitFor(() => api.pendingCodexCapacityRetries.has(id));
assert.equal(api.pendingCodexCapacityRetries.get(id).attempts, attempt);
api.handleResumeSession({ readyState: 1 }, { sessionId: id });
const resume = api.sent.findLast(message => message.type === 'resume_generating');
assert(resume.text.includes('retry-history-A-begin'));
await api.fireRetry(id);
}
await waitFor(() => !api.activeCodexAppTurns.has(id) && !api.pendingCodexCapacityRetries.has(id));
const reloaded = createServerHarness(dir).loadSession(id);
const assistant = reloaded.messages.find(message => message.role === 'assistant');
for (const label of ['A', 'B', 'C']) {
assert.equal(assistant.content.split(`retry-history-${label}-begin`).length, 2);
assert.equal(assistant.content.split(`retry-history-${label}-end`).length, 2);
}
assert.equal(assistant.toolCalls.length, 3);
assert.equal(new Set(assistant.toolCalls.map(tool => tool.id)).size, 3);
['A', 'B', 'C'].forEach((label, index) => {
assert(assistant.toolCalls[index].result.startsWith(`tool-${label}-full-result`));
assert(assistant.toolCalls[index].result.length > 32768);
});
assert.equal(reloaded.messages.filter(message => message.role === 'user').length, 1);
} finally {
await client.stop();
}
});
}
run().then(() => {
if (failures.length) {
console.error(`运行中气泡续接回归失败:${failures.length} 项;通过 ${passed} 项`);
process.exitCode = 1;
} else console.log(`运行中气泡续接回归通过:${passed} 项`);
}).catch(error => { console.error(error); process.exitCode = 1; });