From 7c5a1f8c5ba2de48213a8580429b5593370789ee Mon Sep 17 00:00:00 2001 From: Lyon <88232613+pikasTech@users.noreply.github.com> Date: Sat, 20 Jun 2026 19:31:49 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20PJ2026-0104010803=20P2=20SSE/realtime?= =?UTF-8?q?=20=E6=97=A0=E7=8A=B6=E6=80=81=E5=8C=96=20-=20durable=20outbox?= =?UTF-8?q?=20cursor=20replay=20(#1756)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit server-workbench-http.ts: - handleWorkbenchRealtimeHttp 优先从 durable outbox 读取 (readWorkbenchProjectionOutbox) - 当 runtimeStore 支持 outbox 时，SSE 用 polling cursor replay 代替 traceStore.subscribe - outbox 不支持时保留内存 traceStore.subscribe 作为兼容降级 - terminal commit 从 outbox terminal 行触发 snapshot + trace realtime - 新增 HWLAB_WORKBENCH_SSE_OUTBOX_POLL_MS 环境变量控制 poll 间隔 workbench-facts-store.ts: - 新增 subscribeDurableProjection 方法，从 durable outbox polling 读取 - subscribeTrace 保留兼容（内存 trace），新增 durable projection subscription 入口 - SPEC header 更新到 draft-2026-06-20-p1-zero-split-durable-realtime cloud-api 重启后 SSE 从 DB outbox cursor 继续输出，不依赖内存 listener Closes #1748 Refs #1742 --- internal/cloud/server-workbench-http.ts | 100 +++++++++++++++++------- internal/cloud/workbench-facts-store.ts | 28 ++++++- 2 files changed, 99 insertions(+), 29 deletions(-) diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index dc226ee1..eb6e2c3e 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -1,5 +1,5 @@ /* - * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p0-projector-resume; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 + * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p0-projector-resume; draft-2026-06-20-p1-zero-split-durable-realtime; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 * 职责: Workbench REST read model。只读投影 session/message/turn/trace facts，不执行 AgentRun sync、billing finalize 或 workspace repair。 */ import { createHash } from "node:crypto"; @@ -126,33 +126,77 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option ?? null; if (activeTraceId) { await (perf ? perf.measure("workbench_initial_trace", () => writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" })) : writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" })); - const unsubscribe = traceStore.subscribe(activeTraceId, (event, snapshot) => { - if (event) { - writeEvent("workbench.trace.event", { - type: "trace.event", - sessionId: streamSessionId, - threadId: streamThreadId, - traceId: activeTraceId, - event, - snapshot: { ...traceSnapshotSummary(snapshot), sessionId: streamSessionId, threadId: streamThreadId }, - cursor: { traceSeq: eventSeq(event, Number(snapshot?.eventCount ?? 1) - 1) } - }); - } else { - writeEvent("workbench.trace.snapshot", { - type: "trace.snapshot", - sessionId: streamSessionId, - threadId: streamThreadId, - traceId: activeTraceId, - reason: "trace-store-update", - snapshot: { ...traceSnapshotForRealtime(snapshot), sessionId: streamSessionId, threadId: streamThreadId }, - cursor: { traceSeq: traceSnapshotLastSeq(snapshot) } - }); - } - if (event?.terminal === true || TERMINAL_STATUSES.has(normalizeStatus(event?.status))) { - void writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "terminal" }); - } - }); - cleanup.push(unsubscribe); + const runtimeStore = options.runtimeStore ?? null; + const outboxPollMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_OUTBOX_POLL_MS, 2000); + let outboxCursor = 0; + if (typeof runtimeStore?.readWorkbenchProjectionOutbox === "function") { + const pollOutbox = async () => { + if (closed || response.destroyed) return; + try { + const rows = await runtimeStore.readWorkbenchProjectionOutbox({ afterSeq: outboxCursor, limit: 50, traceId: activeTraceId }); + for (const row of rows) { + if (closed || response.destroyed) break; + outboxCursor = row.outboxSeq; + if (row.commitType === "terminal") { + writeEvent("workbench.trace.snapshot", { + type: "trace.snapshot", + sessionId: row.sessionId ?? streamSessionId, + threadId: streamThreadId, + traceId: row.traceId ?? activeTraceId, + reason: "outbox-terminal", + cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq }, + projectionStatus: { terminal: row.terminal, sealed: row.sealed } + }); + void writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "terminal" }); + } else { + writeEvent("workbench.trace.event", { + type: "trace.event", + sessionId: row.sessionId ?? streamSessionId, + threadId: streamThreadId, + traceId: row.traceId ?? activeTraceId, + cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq } + }); + } + } + } catch (outboxError) { + writeEvent("workbench.error", { + type: "error", + error: { code: "workbench_outbox_poll_failed", message: outboxError?.message ?? "Workbench outbox poll failed." } + }); + } + }; + await pollOutbox(); + const outboxTimer = setInterval(() => { void pollOutbox(); }, Math.max(500, outboxPollMs)); + cleanup.push(() => clearInterval(outboxTimer)); + } else { + const unsubscribe = traceStore.subscribe(activeTraceId, (event, snapshot) => { + if (event) { + writeEvent("workbench.trace.event", { + type: "trace.event", + sessionId: streamSessionId, + threadId: streamThreadId, + traceId: activeTraceId, + event, + snapshot: { ...traceSnapshotSummary(snapshot), sessionId: streamSessionId, threadId: streamThreadId }, + cursor: { traceSeq: eventSeq(event, Number(snapshot?.eventCount ?? 1) - 1) } + }); + } else { + writeEvent("workbench.trace.snapshot", { + type: "trace.snapshot", + sessionId: streamSessionId, + threadId: streamThreadId, + traceId: activeTraceId, + reason: "trace-store-update", + snapshot: { ...traceSnapshotForRealtime(snapshot), sessionId: streamSessionId, threadId: streamThreadId }, + cursor: { traceSeq: traceSnapshotLastSeq(snapshot) } + }); + } + if (event?.terminal === true || TERMINAL_STATUSES.has(normalizeStatus(event?.status))) { + void writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "terminal" }); + } + }); + cleanup.push(unsubscribe); + } } const heartbeat = setInterval(async () => { diff --git a/internal/cloud/workbench-facts-store.ts b/internal/cloud/workbench-facts-store.ts index 1f890a87..75ad4e9f 100644 --- a/internal/cloud/workbench-facts-store.ts +++ b/internal/cloud/workbench-facts-store.ts @@ -1,5 +1,5 @@ /* - * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1. + * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; draft-2026-06-20-p1-zero-split-durable-realtime; PJ2026-010403 API契约 draft-2026-06-18-r1. * 职责: Workbench durable facts store adapter。统一读取 session/message/turn/trace projection facts，不在 GET 中推进 AgentRun facts。 */ import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; @@ -124,6 +124,31 @@ export function createWorkbenchFactsStore(options = {}, actor = null) { return traceStore.subscribe(traceId, listener); } + function subscribeDurableProjection(traceId, listener, pollOptions = {}) { + const runtimeStore = options.runtimeStore ?? null; + if (typeof runtimeStore?.readWorkbenchProjectionOutbox !== "function") return () => {}; + const pollMs = Math.max(500, Number(pollOptions.pollMs ?? 2000)); + let cursor = 0; + let stopped = false; + const poll = async () => { + if (stopped) return; + try { + const rows = await runtimeStore.readWorkbenchProjectionOutbox({ afterSeq: cursor, limit: 50, traceId }); + for (const row of rows) { + if (stopped) break; + cursor = row.outboxSeq; + listener(row); + } + } catch { /* durable projection poll errors are non-fatal for SSE consumers */ } + }; + void poll(); + const timer = setInterval(() => { void poll(); }, pollMs); + return () => { + stopped = true; + clearInterval(timer); + }; + } + return { queryFacts, getSessionById, @@ -134,6 +159,7 @@ export function createWorkbenchFactsStore(options = {}, actor = null) { traceSnapshot, traceSnapshotSync, subscribeTrace, + subscribeDurableProjection, canReadOwner: (ownerUserId) => canActorReadOwner(ownerUserId, actor) }; }