mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: PJ2026-0104010803 P2 SSE/realtime 无状态化 - durable outbox cursor replay (#1756)

Browse Source 
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
This commit is contained in:
Lyon
2026-06-20 19:31:49 +08:00
committed by GitHub
parent b254c1875e
commit 7c5a1f8c5b
2 changed files with 99 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/server-workbench-http.ts
+72 -28
View File
@@ -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 () => {
internal/cloud/workbench-facts-store.ts
+27 -1
View File
@@ -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)
};
}