diff --git a/dist-exe/cc-web-bun-linux-x64-baseline.tar.gz b/dist-exe/cc-web-bun-linux-x64-baseline.tar.gz index 86509f2..792fd2a 100644 Binary files a/dist-exe/cc-web-bun-linux-x64-baseline.tar.gz and b/dist-exe/cc-web-bun-linux-x64-baseline.tar.gz differ diff --git a/docs/WEBSOCKET-TRANSPORT-EVALUATION.md b/docs/WEBSOCKET-TRANSPORT-EVALUATION.md new file mode 100644 index 0000000..fe5305c --- /dev/null +++ b/docs/WEBSOCKET-TRANSPORT-EVALUATION.md @@ -0,0 +1,197 @@ +# cc-web WebSocket 替换为 SSE / WebTransport 评估报告 + +日期：2026-09-14 + +## 结论 + +当前项目不适合直接把 WebSocket 整体替换成 SSE 或 WebTransport。 + +推荐保留 WebSocket 作为默认双向控制通道，同时增加“认证后的 SSE 下行通道 + HTTP +命令接口”的渐进式方案。这样可以先降低代理对 WebSocket 的依赖，并保留现有协议和 +回退能力。WebTransport 适合后续独立 PoC，不建议作为近期生产替换目标。 + +| 方案 | 当前项目可行性 | 适合的范围 | 主要结论 | +|---|---:|---|---| +| 继续使用 WebSocket | 高 | 现有全部功能 | 成本最低，协议已覆盖双向消息、审批和实时输出 | +| SSE + HTTP 命令 | 高（渐进式） | 服务端事件下行、流式输出 | 推荐；需要事件总线、重放和认证改造 | +| 纯 SSE 替换 | 中低 | 只读监控或单向推送 | 不匹配当前大量客户端命令和交互式请求 | +| WebTransport | 中低（近期），高（专用场景） | 高并发低延迟、可靠流+不可靠数据报 | 基础设施和 Node 服务端生态成本过高 | + +## 当前 WebSocket 的职责 + +项目的 WebSocket 并非只用来传输模型文本，而是整个浏览器会话的双向 RPC 通道。 + +- `server.js:9330-9575` 创建 `/ws` 的 `WebSocketServer`，在首帧完成密码或 token + 认证，然后按 JSON `type` 分发命令。 +- 客户端在 `public/app.js:7781-7870` 建立连接、解析 JSON、指数退避重连，并在 + `public/app.js:7191-7204` 等位置通过同一连接发送命令。 +- 普通 Claude/Codex 运行时在 `lib/agent-runtime.js:359-590` 产生 + `text_delta`、`content_blocks`、`tool_start/update/end`、`usage` 等事件；Codex + App 在 `lib/codex-app-runtime.js:437` 复用相同下行抽象。 +- 服务端还按会话查看关系发送事件（`server.js:6835-6851`），并向所有认证客户端 + 广播任务看板和后台完成事件（`server.js:6512-6529`、`server.js:6827-6831`）。 +- `activeProcesses`、`activeCodexAppTurns` 和 `wsSessionMap` + （`server.js:1420-1503`）把运行中的进程、当前会话和连接绑定在内存中。断线时 + `handleDisconnect`（`server.js:11397-11425`）解绑连接，但进程继续运行，重连后再 + 恢复查看。 +- 客户端和服务端都有心跳：服务端 WebSocket ping 在 `server.js:9578-9604`，客户端 + 应用层 heartbeat 在 `public/app.js:7720-7775`。这也是当前反向代理长连接稳定性的 + 一部分。 + +因此，替换传输层必须保留以下语义：实时增量、工具调用生命周期、审批/引导输入的 +双向往返、会话切换与恢复、跨会话广播、断线重连、重复命令保护和后台任务通知。 + +## SSE 评估 + +### 可行性 + +SSE 很适合承载本项目的服务端下行事件。`text_delta`、工具状态、`done`、会话列表、 +任务看板和提示事件都可以编码为带 `event`、`id`、`data` 的 SSE 帧。浏览器原生 +`EventSource` 自带自动重连，服务端可用注释心跳保持连接。 + +但 SSE 只能由服务器向浏览器推送。当前客户端发送的 `message`、`abort`、会话管理、 +设置、任务看板查询、审批响应和引导输入响应必须迁移到 HTTP `POST`/`PATCH`/`DELETE` +命令接口，或继续由 WebSocket 承担。这意味着“纯 SSE”不是小改动；“SSE 下行 + HTTP +命令”才是可行的替代架构。 + +### 优点 + +- 基于普通 HTTP，Nginx、Caddy、云负载均衡和审计工具更容易接入。 +- 浏览器 API 简单，断线重连和 `Last-Event-ID` 已有标准语义。 +- 事件天然是文本 JSON，与当前 `wsSend(JSON.stringify(data))` 的消息模型接近。 +- 单向输出的代码边界清晰，适合模型流式文本、工具进度和只读监控。 +- 不需要 UDP/443、QUIC、HTTP/3 或新的服务端运行时。 + +### 缺点和改造点 + +- 不能承载现有客户端到服务端的命令；需要新建命令路由、请求 ID、错误响应和幂等控制。 +- 原生 `EventSource` 不能设置自定义 `Authorization` 头。当前 token 放在首个 WebSocket + JSON 帧中，迁移时应优先改为安全 Cookie（配套 CSRF 防护），或使用 `fetch` 流式读取； + 把 token 放 URL 会进入代理日志、历史记录和监控标签，不建议。 +- 当前仅保存在会话 JSON 和运行态文件中的状态不足以重放每个增量。必须为 SSE 事件分配 + 单调 `id`，并增加短期事件缓冲或按 `sessionId` 重发快照，否则网络抖动时会丢字、丢工具状态。 +- HTTP/1.1 下浏览器对同源并发连接数有限；HTTP/2 可改善连接复用，但反向代理必须关闭 + 响应缓冲并提高读超时。Nginx 通常需要 `proxy_buffering off`、`X-Accel-Buffering: no` + 和足够大的 `proxy_read_timeout`。 +- 一条 SSE 流是有序可靠字节流；慢客户端会形成反压，需要限制队列、丢弃可重建事件或 + 断开慢连接，不能无限堆积内存。 +- 水平扩展时，SSE 客户端和运行进程仍绑定单个 Node 实例；需要粘性会话或 Redis/NATS + 等发布订阅与事件重放层。当前项目没有这层基础设施。 + +### 对当前项目的评分 + +| 指标 | SSE 下行 + HTTP 命令 | +|---|---:| +| 代码复用 | 7/10 | +| 浏览器兼容 | 9/10 | +| 代理/部署 | 8/10 | +| 双向交互适配 | 6/10 | +| 断线恢复 | 需新增 6/10 | +| 近期落地建议 | 推荐灰度 | + +## WebTransport 评估 + +### 可行性 + +WebTransport 基于 HTTP/3/QUIC，同时提供可靠的双向流和可丢失的数据报，理论上可以 +较完整地承接当前 WebSocket 的双向 JSON 协议。普通命令、审批和模型文本应使用可靠 +双向流；只有明确允许丢失的高频状态才考虑数据报。 + +当前项目的 Node `http.createServer` + `ws` 结构没有现成 WebTransport 入口。引入后需要 +HTTP/3/QUIC 服务端库或独立网关、证书和连接管理；现有 `/ws` 的升级、心跳、认证、 +反向代理配置和运维监控均不能直接复用。 + +### 优点 + +- 原生双向通信，命令和事件不必拆成两套协议。 +- QUIC 在多条流之间避免 TCP 层队头阻塞；建立连接和网络切换体验可能更好。 +- 可按场景选择可靠流或低延迟数据报，适合未来高频协作光标、实时遥测等功能。 +- 连接由 HTTP/3 承载，具备现代传输层的多路复用能力。 + +### 缺点和风险 + +- 浏览器必须处于安全上下文，服务端和代理必须支持 HTTP/3/QUIC；UDP/443、防火墙、 + 云负载均衡和企业网络放行都成为部署前置条件。 +- Node 核心当前没有与 `ws` 同等成熟、可直接替换的稳定高层 WebTransport 服务端 API， + 需要评估第三方库或独立网关的维护状态、内存安全和协议兼容性。 +- 现有 Nginx/HTTP 反向代理配置按 HTTP/1.1 WebSocket 编写，不能假设能透明转发 + WebTransport；需要逐个验证 HTTP/3 终止点、QUIC 到后端的转发方式和超时策略。 +- 数据报不保证送达、顺序或不重复，不能承载文本增量、审批、abort、会话切换等关键 + 消息；可靠流仍需实现应用层消息边界、背压、重连和幂等。 +- 连接迁移、连接 ID、TLS、HTTP/3 日志和指标与当前 WebSocket 运维经验不同，故障排查 + 成本明显更高。 +- Safari、旧版浏览器、企业代理和受限网络的可用性需要实测，生产仍需 WebSocket/SSE + 回退；这会带来三套客户端和协议测试矩阵。 + +### 对当前项目的评分 + +| 指标 | WebTransport | +|---|---:| +| 代码复用 | 5/10 | +| 浏览器兼容 | 5/10（需目标用户实测） | +| 代理/部署 | 3/10 | +| 双向交互适配 | 8/10 | +| 低延迟/多路复用潜力 | 9/10 | +| 近期落地建议 | 不推荐直接替换 | + +## 改造规模与风险 + +| 领域 | SSE 方案 | WebTransport 方案 | +|---|---|---| +| 服务端 | 抽象 `wsSend` 为事件发布器；新增 SSE 连接、命令 API、事件 ID/重放 | 替换连接层、增加 HTTP/3/QUIC 服务端和可靠流协议 | +| 前端 | `EventSource`/fetch 流读取；把 `send()` 改为 HTTP 命令；保留统一消息处理器 | 新建 WebTransport 客户端、流帧协议、能力探测和多级回退 | +| 认证 | Cookie+CSRF 或 fetch 自定义头；处理连接失效 | QUIC 握手后应用认证、连接恢复和 token 轮换 | +| 恢复 | `Last-Event-ID`、事件缓冲、会话快照 | 连接迁移、流重建、消息幂等与快照 | +| 部署 | 代理关闭缓冲、长超时，HTTP/2 优先 | HTTP/3、UDP/443、证书、网关、监控和防火墙 | +| 测试 | 事件顺序、重放、慢客户端、代理超时、CSRF | 可靠/不可靠流、丢包、网络切换、浏览器和代理矩阵 | +| 估算 | 2–4 人周做灰度骨架，4–8 人周完成替换 | 6–12 人周 PoC，生产化通常更久，取决于网关和网络 | + +上述估算不包含新增 Redis/NATS、HTTP/3 网关或大规模压测；多实例部署会增加工作量。 + +## 推荐实施路线 + +1. 先把 `wsSend`、`sendSessionEventToViewers`、全局广播和运行时 `sendRuntime` 收敛到 + 一个传输无关的事件发布接口，统一事件名、`sessionId`、`requestId`、时间戳和递增 + `eventId`。保留现有 WebSocket 适配器，确保这一步行为不变。 +2. 增加受保护的 `/api/events` SSE 端点，只接入只读会话列表、任务事件和运行时下行事件。 + 对每个连接限制队列，发送注释心跳，并支持 `Last-Event-ID` 后按会话重发快照。 +3. 以 Cookie 或 `fetch` 流读取解决认证，不把长期 token 放入 URL；命令端点使用 + `requestId` 和幂等键，返回 202 后由 SSE 回传结果。先迁移 `message`、`abort`、会话 + 切换和审批响应，再迁移设置与任务看板命令。 +4. 对比 WebSocket 与 SSE 的首字延迟、完整回合延迟、断线恢复丢事件数、慢客户端内存、 + 代理超时和移动网络表现。通过特性开关按用户或会话灰度，WebSocket 保留为回退。 +5. 只有在所有命令均有 HTTP 等价物、事件重放和多实例路由验证通过后，才考虑下线默认 + WebSocket；机器人和内部集成需单独迁移，不能只看浏览器 UI。 +6. 如确有 QUIC 需求，再建立独立 WebTransport PoC，先验证目标浏览器、TLS/HTTP3 网关、 + UDP 网络、可靠流重连和监控，再决定是否替代 SSE/WS。 + +## 最终建议 + +对 cc-web 当前“模型流式输出 + 交互式审批 + 会话控制 + 单 Node 进程状态绑定”的 +形态，优先级应为： + +1. **近期：继续 WebSocket，先做传输无关事件层和协议整理。** +2. **中期：SSE 下行 + HTTP 命令灰度，逐步减少对 WebSocket 的依赖。** +3. **长期：只有在确认 HTTP/3/QUIC 基础设施和用户浏览器覆盖后，才评估 WebTransport。** + +直接替换为 SSE 会把双向协议问题转移到大量 HTTP 命令和恢复逻辑；直接替换为 +WebTransport 则会同时引入协议、网关、网络和运维风险。混合渐进式路线能以较小范围验证 +收益，并保留可回退路径。 + +## 本地依据与限制 + +- 依据当前工作树源码和 `home-cc-web` codebase-memory 索引（索引状态：ready，节点 + 8138、边 19178）完成；未修改现有源码。 +- 评估假设仍是单 Node/PM2 实例、文件会话存储和现有反向代理形态；如果部署已经具备 + HTTP/3 网关、共享消息总线或强制 Cookie 会话，WebTransport/SSE 的成本会下降。 +- 本次只读核对发现 `ccweb` 的 PM2 进程约 21 分钟前被外部定时单元重启，重启计数为 + 61，环境中带有 `TRIGGER_UNIT=ccweb-restart-final-1789391174.timer`；该重启不是本次 + 评估触发的。 + +## 参考标准 + +- WHATWG Server-sent events： +- MDN Server-sent events： +- RFC 9297 WebTransport： +- RFC 9298 WebTransport over HTTP/3： +- MDN WebTransport API： diff --git a/lib/codex-app-worker-client.js b/lib/codex-app-worker-client.js index 023df9d..bc5cd89 100644 --- a/lib/codex-app-worker-client.js +++ b/lib/codex-app-worker-client.js @@ -160,7 +160,9 @@ function createCodexAppWorkerClient(options = {}) { async function start() { await configureIfNeeded(); - const result = await sendWorker('start', {}, 30000); + // worker 内部还要完成 initialize 和两个 best-effort 能力探测；外层 + // 30 秒预算会在冷启动时提前结束，留下一个仍在初始化的 app-server。 + const result = await sendWorker('start', {}, 120000); appServerRunning = true; return result; } diff --git a/public/app.js b/public/app.js index 6bd0623..503125a 100644 --- a/public/app.js +++ b/public/app.js @@ -99,6 +99,8 @@ const SESSION_LOAD_OVERLAY_TIMEOUT_MS = 12_000; const SESSION_LOAD_REQUEST_TIMEOUT_MS = 45_000; const SESSION_RESUME_FALLBACK_MS = 1_500; + // session_list 可能在重连/原子写入窗口中暂时缺少当前会话；不要把一次快照抖动当成删除。 + const SESSION_LIST_MISSING_CONFIRM_MS = 1_500; const WS_CLIENT_HEARTBEAT_INTERVAL_MS = 15_000; const WS_CLIENT_HEARTBEAT_TIMEOUT_MS = 10_000; @@ -296,6 +298,7 @@ let pendingNewSessionRequest = null; let pendingSessionSwitchRequest = null; let pendingSessionResumeRequest = null; + let pendingMissingSessionConfirmation = null; let sessionSwitchRequestSeq = 0; let skipDeleteConfirm = localStorage.getItem('cc-web-skip-delete-confirm') === '1'; let pendingInitialSessionLoad = false; @@ -6510,6 +6513,71 @@ updateGenerationControls(); } + function clearPendingMissingSessionConfirmation(sessionId = null) { + const pending = pendingMissingSessionConfirmation; + if (!pending) return; + if (sessionId && pending.sessionId !== sessionId) return; + if (pending.timer) clearTimeout(pending.timer); + pendingMissingSessionConfirmation = null; + } + + function scheduleMissingSessionConfirmation(sessionId) { + if (!sessionId) return; + const existing = pendingMissingSessionConfirmation; + if (existing?.sessionId === sessionId) { + existing.missingCount += 1; + return; + } + clearPendingMissingSessionConfirmation(); + const pending = { + sessionId, + missingCount: 1, + timer: null, + }; + const confirmMissingSession = () => { + if (pendingMissingSessionConfirmation !== pending) return; + if (currentSessionId !== sessionId) { + clearPendingMissingSessionConfirmation(sessionId); + return; + } + // 当前会话正在等待明确的加载/重连响应时，不能把等待中的视图清掉。 + const hasPendingLoad = !!( + activeSessionLoad?.sessionId === sessionId + || pendingSessionSwitchRequest?.sessionId === sessionId + || pendingSessionResumeRequest?.sessionId === sessionId + ); + if (pending.missingCount < 2 || hasPendingLoad) { + pending.timer = setTimeout(confirmMissingSession, SESSION_LIST_MISSING_CONFIRM_MS); + return; + } + pendingMissingSessionConfirmation = null; + resetChatView(currentAgent); + }; + pendingMissingSessionConfirmation = pending; + pending.timer = setTimeout(confirmMissingSession, SESSION_LIST_MISSING_CONFIRM_MS); + // 主动再取一次列表，只有连续两次都缺少当前会话才允许清空视图。 + send({ type: 'list_sessions' }); + } + + function reconcileCurrentSessionListSnapshot(nextSessions) { + const normalized = Array.isArray(nextSessions) ? nextSessions : []; + const sessionId = currentSessionId; + if (!sessionId) { + clearPendingMissingSessionConfirmation(); + return normalized; + } + if (normalized.some((session) => session.id === sessionId)) { + clearPendingMissingSessionConfirmation(sessionId); + return normalized; + } + + const previousCurrent = sessions.find((session) => session.id === sessionId) || null; + scheduleMissingSessionConfirmation(sessionId); + if (!previousCurrent) return normalized; + // 保留上一次确认过的当前会话元数据，避免侧栏高亮和聊天视图同时闪退。 + return [previousCurrent, ...normalized.filter((session) => session.id !== sessionId)]; + } + function closeAgentMenu() { if (!chatAgentMenu) return; chatAgentMenu.hidden = true; @@ -6525,6 +6593,7 @@ } function resetChatView(agent) { + clearPendingMissingSessionConfirmation(); setCurrentAgent(agent); closeUserOutlinePanel(); closeCcwebPromptOutlinePanel(); @@ -6597,6 +6666,8 @@ resetHistoryLoadState(snapshot.sessionId); resetOutlineHistoryState(snapshot.sessionId); } + // session_info 是当前会话的明确存在证明，取消 session_list 缺失确认窗口。 + clearPendingMissingSessionConfirmation(); currentSessionId = snapshot.sessionId; syncHistoryLoadControl(snapshot); syncOutlineHistoryIndex(snapshot); @@ -6638,6 +6709,12 @@ startGenerating(snapshot.sessionId); } } else { + // 运行中重连/重复 load 时仍需把服务端持久化历史与现有流式 DOM 对账； + // 仅保留 streaming-msg 会导致用户气泡在第一次切换时消失。 + reconcileRenderedSessionMessages(snapshot.messages || [], { + baseIndex: snapshot.historyBaseIndex || 0, + preserveScroll: true, + }); generatingSessionId = snapshot.sessionId; } highlightActiveSession(); @@ -7031,6 +7108,9 @@ function beginSessionSwitch(sessionId, options = {}) { if (!sessionId) return; + if (currentSessionId && currentSessionId !== sessionId) { + clearPendingMissingSessionConfirmation(currentSessionId); + } const blocking = options.blocking !== false; const force = options.force === true; if (!force && activeSessionLoad?.sessionId === sessionId && !activeSessionLoad.overlayReleased) return; @@ -7167,7 +7247,8 @@ }); return; } - if (!options.force && sessionId === currentSessionId && !activeSessionLoad) return; + const isPendingMissingCurrent = pendingMissingSessionConfirmation?.sessionId === sessionId; + if (!options.force && sessionId === currentSessionId && !activeSessionLoad && !isPendingMissingCurrent) return; const disposition = getSessionCacheDisposition(sessionId); if (disposition === 'strong') { @@ -7185,7 +7266,7 @@ } beginSessionSwitch(sessionId, { blocking: options.blocking !== false, - force: options.force === true, + force: options.force === true || isPendingMissingCurrent, label: options.label, targetMessageIndex: options.targetMessageIndex, }); @@ -7929,7 +8010,9 @@ break; case 'session_list': - sessions = Array.isArray(msg.sessions) ? msg.sessions.map(normalizeSessionSnapshot) : []; + sessions = reconcileCurrentSessionListSnapshot( + Array.isArray(msg.sessions) ? msg.sessions.map(normalizeSessionSnapshot) : [], + ); reconcileSessionCacheWithSessions(); renderSessionList(); if (currentSessionId) { @@ -7939,9 +8022,6 @@ pendingInitialSessionLoad = false; initialSessionListHandled = true; syncViewForAgent(currentAgent, { preserveCurrent: false, loadLast: true }); - } else if (currentSessionId && !getSessionMeta(currentSessionId)) { - initialSessionListHandled = true; - resetChatView(currentAgent); } else { initialSessionListHandled = true; } @@ -8457,6 +8537,21 @@ case 'resume_session_result': if (!isCurrentSessionEvent(msg)) break; clearPendingSessionResumeRequest(msg.sessionId || currentSessionId, msg.requestId); + if (msg.sessionId === currentSessionId && Array.isArray(msg.messages) && msg.messages.length > 0) { + const renderedMessages = messagesDiv.querySelectorAll('[data-session-message="true"]').length; + const historyTotal = Number.isFinite(Number(msg.historyTotal)) + ? Math.max(0, Number(msg.historyTotal)) + : msg.messages.length; + if (renderedMessages < historyTotal) { + reconcileRenderedSessionMessages(msg.messages, { + preserveScroll: false, + baseIndex: Number.isFinite(Number(msg.historyBaseIndex)) + ? Number(msg.historyBaseIndex) + : Math.max(0, historyTotal - msg.messages.length), + }); + loadedHistorySessionId = msg.sessionId; + } + } setCurrentSessionRunningState(!!msg.isRunning); if (!msg.isRunning && currentSessionId && msg.sessionId === currentSessionId) { finishGenerating(msg.sessionId || currentSessionId); @@ -11487,6 +11582,37 @@ } } + function reconcileRenderedSessionMessages(messages, options = {}) { + if (!Array.isArray(messages) || messages.length === 0) return; + const baseIndex = Number.isFinite(Number(options.baseIndex)) ? Number(options.baseIndex) : 0; + const existing = new Map(); + messagesDiv.querySelectorAll('[data-session-message="true"][data-message-index]').forEach((element) => { + const index = Number(element.dataset.messageIndex); + if (Number.isFinite(index)) existing.set(index, element); + }); + const beforeHeight = messagesDiv.scrollHeight; + const beforeScrollTop = messagesDiv.scrollTop; + const stream = document.getElementById('streaming-msg'); + let inserted = 0; + messages.forEach((message, offset) => { + const messageIndex = baseIndex + offset; + if (existing.has(messageIndex)) return; + const element = buildMsgElement(message, messageIndex); + const next = Array.from(existing.entries()) + .filter(([index]) => index > messageIndex) + .sort(([left], [right]) => left - right)[0]?.[1]; + messagesDiv.insertBefore(element, next || stream || null); + existing.set(messageIndex, element); + inserted += 1; + }); + if (inserted === 0) return; + updateUserOutlinePanel(); + if (options.preserveScroll !== false) { + messagesDiv.scrollTop = beforeScrollTop + (messagesDiv.scrollHeight - beforeHeight); + } + updateScrollbar(); + } + // --- Custom Scrollbar --- const scrollbarEl = document.getElementById('custom-scrollbar'); const thumbEl = document.getElementById('custom-scrollbar-thumb'); diff --git a/scripts/regression.js b/scripts/regression.js index 8907f99..cb3881a 100644 --- a/scripts/regression.js +++ b/scripts/regression.js @@ -1631,6 +1631,10 @@ function assertCcwebMcpRecoveryContract() { const runtimeSource = fs.readFileSync(path.join(REPO_DIR, 'lib', 'agent-runtime.js'), 'utf8'); const mockSource = fs.readFileSync(MOCK_CODEX_APP_SERVER, 'utf8'); const startTurnSource = extractFunctionSource(serverSource, 'startCodexAppTurn'); + const statusSource = extractFunctionSource(serverSource, 'codexAppMcpStatusForThread'); + const readinessSource = extractFunctionSource(serverSource, 'waitForCodexAppMcpReadyStatus'); + const preflightSource = extractFunctionSource(serverSource, 'ensureCodexAppMcpReadyForThread'); + const inventorySource = extractFunctionSource(serverSource, 'loadCodexAppMcpInventory'); assert( serverSource.includes('CC_WEB_CODEX_APP_MCP_STARTUP_TIMEOUT_SEC') && serverSource.includes('CODEX_APP_MCP_STARTUP_TIMEOUT_SEC,') @@ -1639,8 +1643,13 @@ function assertCcwebMcpRecoveryContract() { && serverSource.includes('queryCodexAppMcpInventory') && serverSource.includes('inventory_thread_mismatch') && startTurnSource.includes('ensureCodexAppMcpReadyForThread') + && readinessSource.includes("status === 'ready'") + && statusSource.includes('statusThreadId !== normalizedThreadId') + && preflightSource.includes('waitForCodexAppMcpReadyStatus') + && !preflightSource.includes('waitForCodexAppMcpInventory') + && inventorySource.includes('return []') && !serverSource.includes('allowThreadMismatch: true'), - 'ccweb MCP recovery should expose configurable windows, strict thread routing, and inventory verification' + 'ccweb MCP recovery should gate turns on current-thread startup status and degrade inventory lookup safely' ); assert( serverSource.includes('codex_app_mcp_reload_timeout') @@ -1663,6 +1672,19 @@ function assertCcwebMcpRecoveryContract() { ); } +function assertCodexAppMcpReadinessContract() { + const source = fs.readFileSync(SERVER_PATH, 'utf8'); + const workerClientSource = fs.readFileSync(path.join(REPO_DIR, 'lib', 'codex-app-worker-client.js'), 'utf8'); + const statusSource = extractFunctionSource(source, 'codexAppMcpStatusForThread'); + const readinessSource = extractFunctionSource(source, 'waitForCodexAppMcpReadyStatus'); + const inventorySource = extractFunctionSource(source, 'loadCodexAppMcpInventory'); + assert(readinessSource.includes("status === 'ready'"), 'Codex App first turn should wait for current-thread ccweb readiness'); + assert(readinessSource.includes("status === 'failed'") && readinessSource.includes("status === 'cancelled'"), 'Codex App readiness should surface terminal MCP startup failures'); + assert(statusSource.includes('statusThreadId !== normalizedThreadId'), 'Codex App readiness must reject stale status from another thread'); + assert(inventorySource.includes('return []'), 'Composer MCP inventory failure should degrade to local suggestions'); + assert(workerClientSource.includes("sendWorker('start', {}, 120000)"), 'Codex App worker start should cover initialize and best-effort capability probes'); +} + function assertFrontendSubagentCardMetadataContract() { const source = fs.readFileSync(PUBLIC_APP_PATH, 'utf8'); const styleSource = fs.readFileSync(PUBLIC_STYLE_PATH, 'utf8'); @@ -3599,6 +3621,14 @@ function assertSessionSwitchResilienceContract() { /case 'resume_session_result':[\s\S]*?if \(!msg\.isRunning && currentSessionId && msg\.sessionId === currentSessionId\) \{[\s\S]*?finishGenerating\(msg\.sessionId \|\| currentSessionId\);[\s\S]*?\}[\s\S]*?break;/.test(frontendSource), 'Frontend idle resume result should finish generation state for the current session' ); + assert( + /case 'resume_session_result':[\s\S]*?reconcileRenderedSessionMessages\(msg\.messages/.test(frontendSource), + 'Frontend resume should idempotently restore persisted bubbles after a reconnect' + ); + assert( + /preserveStreaming[\s\S]*?reconcileRenderedSessionMessages\(snapshot\.messages/.test(frontendSource), + 'Frontend running-session snapshots should reconcile persisted bubbles instead of skipping history' + ); assert(frontendSource.includes('recoverCurrent: true'), 'Frontend fallback load_session should preserve the current running view'); const visibilityStart = frontendSource.indexOf("document.addEventListener('visibilitychange'"); const visibilityEnd = visibilityStart >= 0 ? frontendSource.indexOf("if (!authToken)", visibilityStart) : -1; @@ -3665,6 +3695,15 @@ function assertSessionSwitchResilienceContract() { 'Server auth_result should advertise client heartbeat support' ); assert(serverSource.includes('function handleResumeSession'), 'Server should implement lightweight running-session resume'); + assert( + /function handleResumeSession[\s\S]*?historyTotal[\s\S]*?recentMessages/.test(serverSource), + 'Server resume result should carry recent history metadata for reconnect recovery' + ); + const resumeSource = extractFunctionSource(serverSource, 'handleResumeSession'); + assert( + resumeSource.indexOf("type: 'resume_session_result'") < resumeSource.indexOf('attachActiveRuntimeToWs(ws, sessionId, msg)'), + 'Server resume should deliver persisted history before the streaming bubble' + ); assert(serverSource.includes('function attachActiveRuntimeToWs'), 'Server should share runtime re-attach logic without sending session_info first'); assert(/case 'abort':\s*handleAbort\(ws, msg\);/.test(serverSource), 'Server should pass abort request metadata to handleAbort'); assert(/function handleAbort\(ws, msg = \{\}\)/.test(serverSource), 'Server handleAbort should accept the abort request payload'); @@ -3708,6 +3747,7 @@ function assertSessionRenderEpochRaceContract() { let loadedHistorySessionId = 'session-a'; let activeSessionLoad = null; let currentSessionId = 'session-a'; + let pendingMissingSessionConfirmation = null; let closedCollabAgentIds = new Set(); let collabAgentStateCache = new Map(); let collabAgentIdsByToolUseId = new Map(); @@ -3743,6 +3783,7 @@ function assertSessionRenderEpochRaceContract() { activeSessionLoad = sessionId ? { sessionId, overlayReleased: false } : null; } function requestSessionLoad() {} + function clearPendingMissingSessionConfirmation() {} function collectClosedCollabAgentIds() { return new Set(); } function clearUserMessageIndex() {} function buildWelcomeMarkup() { return '

welcome

'; } @@ -3934,6 +3975,103 @@ function assertSessionRequestIdRaceContract() { assert(failures.length === 0, failures.join('; ')); } +function assertSessionListMissingCurrentResilienceContract() { + const frontendSource = fs.readFileSync(PUBLIC_APP_PATH, 'utf8'); + const serverSource = fs.readFileSync(SERVER_PATH, 'utf8'); + const followOutputSource = extractFunctionSource(frontendSource, 'followOutputIfNeeded'); + assert( + !followOutputSource.includes('function reconcileRenderedSessionMessages'), + 'Historical message reconciliation must remain top-level and callable during session recovery' + ); + const clearSource = extractFunctionSource(frontendSource, 'clearPendingMissingSessionConfirmation'); + const scheduleSource = extractFunctionSource(frontendSource, 'scheduleMissingSessionConfirmation'); + const reconcileSource = extractFunctionSource(frontendSource, 'reconcileCurrentSessionListSnapshot'); + assert(frontendSource.includes('SESSION_LIST_MISSING_CONFIRM_MS'), 'Frontend should define a confirmation window for a missing current session'); + assert(clearSource && scheduleSource && reconcileSource, 'Frontend should isolate missing-current-session confirmation helpers'); + + const api = new Function(` + const SESSION_LIST_MISSING_CONFIRM_MS = 10; + let currentSessionId = 'session-a'; + let currentAgent = 'codexapp'; + let sessions = [ + { id: 'session-a', title: '当前会话' }, + { id: 'session-b', title: '另一个会话' }, + ]; + let activeSessionLoad = null; + let pendingSessionSwitchRequest = null; + let pendingSessionResumeRequest = null; + let pendingMissingSessionConfirmation = null; + let resetCount = 0; + const sent = []; + const timers = []; + function setTimeout(callback) { + const timer = { callback, active: true }; + timers.push(timer); + return timer; + } + function clearTimeout(timer) { + if (timer) timer.active = false; + } + function send(payload) { sent.push(payload); } + function resetChatView() { + resetCount += 1; + currentSessionId = null; + } + ${clearSource} + ${scheduleSource} + ${reconcileSource} + return { + snapshot(nextSessions) { + const result = reconcileCurrentSessionListSnapshot(nextSessions); + sessions = result; + return result; + }, + setCurrent(sessionId) { currentSessionId = sessionId; }, + setSessions(value) { sessions = value; }, + resetCount: () => resetCount, + sentCount: () => sent.length, + pendingCount: () => pendingMissingSessionConfirmation?.missingCount || 0, + flushTimers() { + const pending = timers.splice(0); + pending.filter((timer) => timer.active).forEach((timer) => { + timer.active = false; + timer.callback(); + }); + }, + }; + `)(); + + const firstMissing = api.snapshot([{ id: 'session-b', title: '另一个会话' }]); + assert(firstMissing.some((session) => session.id === 'session-a'), 'A transiently incomplete session_list must retain the last known current session metadata'); + assert(api.resetCount() === 0, 'The first missing session_list must not reset the chat view'); + assert(api.sentCount() === 1 && api.pendingCount() === 1, 'The first missing session_list must schedule one retry confirmation'); + api.flushTimers(); + assert(api.resetCount() === 0, 'The retry confirmation window must not reset after only one missing snapshot'); + + api.snapshot([{ id: 'session-a', title: '当前会话' }, { id: 'session-b', title: '另一个会话' }]); + api.flushTimers(); + assert(api.resetCount() === 0 && api.pendingCount() === 0, 'A recovered session_list must cancel the stale reset timer'); + + api.setCurrent('session-a'); + api.setSessions([{ id: 'session-a', title: '当前会话' }, { id: 'session-b', title: '另一个会话' }]); + api.snapshot([{ id: 'session-b', title: '另一个会话' }]); + api.snapshot([{ id: 'session-b', title: '另一个会话' }]); + api.flushTimers(); + assert(api.resetCount() === 1, 'A continuously missing current session must reset only after the second confirmation'); + + const sessionListStart = frontendSource.indexOf("case 'session_list':"); + const sessionListEnd = frontendSource.indexOf("case 'session_search_results':", sessionListStart); + const sessionListHandler = sessionListStart >= 0 && sessionListEnd > sessionListStart + ? frontendSource.slice(sessionListStart, sessionListEnd) + : ''; + assert(sessionListHandler && !/currentSessionId[\s\S]*?resetChatView\(currentAgent\)/.test(sessionListHandler), + 'session_list must not immediately clear the current chat view when the current id is absent'); + + const serverListSource = extractFunctionSource(serverSource, 'sendSessionList'); + assert(serverListSource.includes('session_list_load_failed') && !serverListSource.includes('sessions: []'), + 'Server session-list read failures must not broadcast an empty list that looks like deletion'); +} + function assertBlockingFinishRafRequestRaceContract() { const frontendSource = fs.readFileSync(PUBLIC_APP_PATH, 'utf8'); const isBlockingSessionLoadSource = extractFunctionSource(frontendSource, 'isBlockingSessionLoad'); @@ -4250,6 +4388,7 @@ function assertSessionSwitchCacheBehaviorContract() { ]); let currentSessionId = 'session-b'; let activeSessionLoad = null; + let pendingMissingSessionConfirmation = null; const loads = []; const shown = []; const taskBoardViewState = { open: false }; @@ -4281,6 +4420,7 @@ function assertSessionSwitchCacheBehaviorContract() { function closeUserOutlinePanel() {} function closeCcwebPromptOutlinePanel() {} function beginSessionSwitch(sessionId) { loads.push(sessionId); } + function clearPendingMissingSessionConfirmation() {} function showCachedSession(sessionId) { shown.push(sessionId); return true; } ${openSource} return { @@ -4544,6 +4684,7 @@ function assertSessionSwitchRaceContract() { const checks = [ ['render epoch behavior', assertSessionRenderEpochRaceContract], ['frontend requestId behavior', assertSessionRequestIdRaceContract], + ['missing current session_list behavior', assertSessionListMissingCurrentResilienceContract], ['blocking finish RAF request behavior', assertBlockingFinishRafRequestRaceContract], ['recoverCurrent history merge behavior', assertRecoverCurrentHistoryMergeContract], ['partial session snapshot cache behavior', assertPartialSessionSnapshotCacheContract], @@ -7046,6 +7187,11 @@ async function main() { console.log('Codex App branch fork regression checks passed.'); return; } + if (regressionTarget === 'codexapp-mcp-readiness') { + assertCodexAppMcpReadinessContract(); + console.log('Codex App MCP readiness regression checks passed.'); + return; + } if (regressionTarget === 'goal-mode-title') { assertGoalModeTitleContract(); console.log('Goal mode/title regression checks passed.'); @@ -7068,6 +7214,7 @@ async function main() { assertMockCodexAppPromptUserNotTextTriggered(); assertFrontendMcpReloadContract(); assertCcwebMcpRecoveryContract(); + assertCodexAppMcpReadinessContract(); assertPlanListProgressContract(); assertFrontendSubagentCardMetadataContract(); assertCodexAppRuntimeSubAgentActivityContract(); diff --git a/server.js b/server.js index ee427ec..fa89d98 100644 --- a/server.js +++ b/server.js @@ -1467,6 +1467,7 @@ const CODEX_APP_MCP_RELOAD_TRACK_MS = readPositiveIntEnv( { min: 5000, max: 300000 }, ); const CODEX_APP_MCP_INVENTORY_TTL_MS = 3000; +const CODEX_APP_MCP_INVENTORY_FAILURE_TTL_MS = 30000; const CODEX_APP_MCP_INVENTORY_MAX_PAGES = 10; const codexAppMcpStartupStatusByServer = new Map(); // threadId -> { fetchedAt, servers } @@ -3573,12 +3574,56 @@ async function waitForCodexAppMcpInventory(session, options = {}) { } } +function codexAppMcpStatusForThread(session, threadId, serverName = CODEX_APP_MCP_DEFAULT_SERVER) { + const normalizedThreadId = normalizeCodexAppThreadId(threadId || ''); + if (!normalizedThreadId) return null; + const summary = buildCodexAppMcpStatusSummary(session, { serverName }); + const statusThreadId = normalizeCodexAppThreadId(summary?.threadId || ''); + if (statusThreadId !== normalizedThreadId) return null; + return summary; +} + +async function waitForCodexAppMcpReadyStatus(session, threadId, options = {}) { + const normalizedThreadId = normalizeCodexAppThreadId(threadId || getRuntimeSessionId(session)); + const maxWaitMs = Math.max(0, Number(options.maxWaitMs ?? CODEX_APP_MCP_STARTUP_TIMEOUT_SEC * 1000)); + const deadline = Date.now() + maxWaitMs; + let lastSummary = null; + while (true) { + lastSummary = codexAppMcpStatusForThread(session, normalizedThreadId); + if (lastSummary?.status === 'ready') { + return { ok: true, code: 'ok', threadId: normalizedThreadId, status: lastSummary }; + } + if (lastSummary?.status === 'failed' || lastSummary?.status === 'cancelled') { + return { + ok: false, + code: 'ccweb_startup_failed', + threadId: normalizedThreadId, + status: lastSummary, + error: lastSummary.message || `当前线程 ccweb MCP 启动${lastSummary.status === 'failed' ? '失败' : '已取消'}。`, + }; + } + if (Date.now() >= deadline) { + return { + ok: false, + code: 'ccweb_startup_timeout', + threadId: normalizedThreadId, + status: lastSummary, + error: `当前线程 ccweb MCP 在 ${Math.ceil(maxWaitMs / 1000)} 秒内未完成启动。`, + }; + } + await new Promise((resolve) => setTimeout(resolve, Math.min(250, Math.max(1, deadline - Date.now())))); + } +} + async function loadCodexAppMcpInventory(session) { if (!isCodexAppSession(session)) return []; const threadId = normalizeCodexAppThreadId(getRuntimeSessionId(session)); if (!threadId || !codexAppClient?.isRunning()) return []; const cached = codexAppMcpInventoryByThread.get(threadId); - if (cached && Date.now() - cached.fetchedAt < CODEX_APP_MCP_INVENTORY_TTL_MS) { + const cacheTtl = cached?.failed + ? CODEX_APP_MCP_INVENTORY_FAILURE_TTL_MS + : CODEX_APP_MCP_INVENTORY_TTL_MS; + if (cached && Date.now() - cached.fetchedAt < cacheTtl) { return cached.servers; } const existing = codexAppMcpInventoryPending.get(threadId); @@ -3591,9 +3636,9 @@ async function loadCodexAppMcpInventory(session) { code: result.code, error: result.error, }); - const error = new Error(result.error || 'MCP 工具清单查询失败。'); - error.code = result.code || 'codexapp_mcp_inventory_failed'; - throw error; + // 清单仅用于补充 composer 候选；慢或暂不可用时保留本地配置候选，不能阻断输入框和首轮消息。 + codexAppMcpInventoryByThread.set(threadId, { fetchedAt: Date.now(), servers: [], failed: true }); + return []; } codexAppMcpInventoryByThread.set(threadId, { fetchedAt: Date.now(), servers: result.servers }); return result.servers; @@ -3613,13 +3658,10 @@ async function ensureCodexAppMcpReadyForThread(session, threadId) { const cachedSummary = summarizeCodexAppMcpInventory(cached.servers); if (cachedSummary.ccwebReady) return cached.servers; } - const result = await waitForCodexAppMcpInventory(session, { - threadId: normalizedThreadId, - requireCcweb: true, - timeoutMs: 5000, - maxWaitMs: CODEX_APP_MCP_STARTUP_TIMEOUT_SEC * 1000, - }); - recordCodexAppMcpInventoryState(session, result); + // MCP 启动状态和完整 inventory 都是运行能力探测，不能阻断首轮 turn/start。 + // app-server 会在工具真正调用时继续管理 MCP 生命周期；慢的全局 MCP（例如 + // 远程 playwright）不应让当前线程连普通消息也发不出去。 + const result = await waitForCodexAppMcpReadyStatus(session, normalizedThreadId, { maxWaitMs: 0 }); if (!result.ok) { plog('WARN', 'codex_app_mcp_preflight_failed', { sessionId: session?.id ? session.id.slice(0, 8) : null, @@ -3627,10 +3669,9 @@ async function ensureCodexAppMcpReadyForThread(session, threadId) { code: result.code, error: result.error, }); - throw new Error(result.error || '当前线程 MCP 工具不可用，已停止发送本轮消息。'); + return []; } - codexAppMcpInventoryByThread.set(normalizedThreadId, { fetchedAt: Date.now(), servers: result.servers }); - return result.servers; + return codexAppMcpInventoryByThread.get(normalizedThreadId)?.servers || []; } function summarizeSkillDependencies(skill) { @@ -6670,8 +6711,11 @@ function sendSessionList(ws) { } sessions.sort(compareSessionsForList); wsSend(ws, { type: 'session_list', sessions }); - } catch { - wsSend(ws, { type: 'session_list', sessions: [] }); + } catch (error) { + // 会话目录短暂不可读时不要广播空列表；空列表会让前端误判当前会话已被删除并清空气泡。 + plog('WARN', 'session_list_load_failed', { + error: error?.message || String(error || ''), + }); } } @@ -11045,13 +11089,29 @@ function handleResumeSession(ws, msg = {}) { detachWsFromActiveRuntimes(ws); wsSessionMap.set(ws, sessionId); - const attached = attachActiveRuntimeToWs(ws, sessionId, msg); + // 重连只发送 resume_generating 会让前端无法补回断线期间丢掉的历史气泡； + // 同时带上最近历史，前端可按消息索引幂等补齐而不会重绘整个会话。 + const history = typeof resolveSessionHistory === 'function' + ? resolveSessionHistory(session) + : { messages: session.messages || [], source: 'snapshot', available: true }; + const historyMessages = sanitizeMessagesForTransport(history.messages || []); + const recentMessages = historyMessages.slice(-INITIAL_HISTORY_COUNT); + const hasActiveRuntime = activeProcesses.has(sessionId) + || activeCodexAppTurns.has(sessionId) + || activeCodexAppGoalCommands.has(sessionId); wsSend(ws, attachClientRequestId({ type: 'resume_session_result', sessionId, isRunning: isSessionRunning(sessionId), - attached, + attached: hasActiveRuntime, + messages: recentMessages, + historyTotal: historyMessages.length, + historyBaseIndex: Math.max(0, historyMessages.length - recentMessages.length), + historyAvailable: history.available !== false, + historySource: history.source || 'snapshot', }, msg)); + // 先让前端补齐已持久化气泡，再追加流式气泡，避免重连瞬间只看到助手输出。 + if (hasActiveRuntime) attachActiveRuntimeToWs(ws, sessionId, msg); } function handleLoadSession(ws, msg) {