From c604dd22366e1ba611f0845476e976381590a14e Mon Sep 17 00:00:00 2001 From: Lyon <88232613+pikasTech@users.noreply.github.com> Date: Wed, 24 Jun 2026 22:45:17 +0800 Subject: [PATCH] fix(workbench): replay realtime outbox by after seq (#2068) --- internal/cloud/server-workbench-http.test.ts | 52 ++++++++++++++++++- internal/cloud/server-workbench-http.ts | 23 +++++++- .../src/api/workbench-events.ts | 11 +++- web/hwlab-cloud-web/src/stores/workbench.ts | 20 +++++++ 4 files changed, 102 insertions(+), 4 deletions(-) diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index 6b76b394..d9407c98 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -1570,6 +1570,55 @@ test("workbench realtime stream accepts session authority without project or wor } }); +test("workbench realtime stream replays durable outbox after requested seq", async () => { + const traceId = "trc_workbench_realtime_after_seq"; + const session = { + id: "ses_workbench_realtime_after_seq", + projectId: "prj_hwpod_workbench", + agentId: "hwlab-code-agent", + status: "running", + ownerUserId: ACTOR.id, + conversationId: "cnv_workbench_realtime_after_seq", + threadId: "thread-workbench-realtime-after-seq", + lastTraceId: traceId, + updatedAt: "2026-06-24T14:00:00.000Z", + session: { sessionStatus: "running", lastTraceId: traceId } + }; + const outboxQueries = []; + const runtimeStore = { + async readWorkbenchProjectionOutbox(params = {}) { + outboxQueries.push({ ...params }); + return [ + { outboxSeq: 11, projectedSeq: 7, traceId, sessionId: session.id, turnId: traceId, commitType: "event", terminal: false, sealed: false, createdAt: "2026-06-24T14:00:01.000Z" } + ]; + } + }; + const accessController = { + store: { + async getAgentSession(sessionId) { return sessionId === session.id ? session : null; }, + async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; } + }, + async ensureBootstrap() {}, + async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; } + }; + const server = createCloudApiServer({ accessController, runtimeStore, env: { HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000" } }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + + try { + const { port } = server.address(); + const events = await getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}&afterSeq=10`, 2); + assert.deepEqual(events.map((event) => event.event), ["workbench.connected", "workbench.trace.event"]); + assert.equal(events[0].id, "10"); + assert.equal(events[1].id, "11"); + assert.equal(events[1].data.cursor.outboxSeq, 11); + assert.equal(events[1].data.cursor.traceSeq, 7); + assert.equal(outboxQueries[0].afterSeq, 10); + assert.equal(outboxQueries[0].sessionId, session.id); + } finally { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + test("workbench read model exposes runtime trace projection query failures as projection blockers", async () => { const traceStore = createCodeAgentTraceStore(); const results = createCodeAgentChatResultStore(); @@ -2021,6 +2070,7 @@ async function getSseEvents(port, path, count) { function parseSseBlock(block) { const lines = block.split("\n"); const event = lines.find((line) => line.startsWith("event:"))?.slice(6).trim() ?? "message"; + const id = lines.find((line) => line.startsWith("id:"))?.slice(3).trim() ?? null; const data = lines.filter((line) => line.startsWith("data:")).map((line) => line.slice(5).trim()).join("\n"); - return { event, data: JSON.parse(data) }; + return { event, id, data: JSON.parse(data) }; } diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 938986f5..dc0a2453 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -129,6 +129,7 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option const traceStore = options.traceStore ?? defaultCodeAgentTraceStore; const readModel = createWorkbenchReadModel(options, auth.actor); + const requestedAfterSeq = workbenchRealtimeAfterSeq(request, url); let closed = false; let realtimeCloseReason = "client_close"; let realtimeCloseSignal = null; @@ -152,7 +153,9 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option const startedAt = nowMs(); const eventCreatedAt = realtimeEventCreatedAt(payload); const traceSeq = realtimeTraceSeq(payload); + const eventId = realtimeEventId(payload); response.write(`event: ${name}\n`); + if (eventId) response.write(`id: ${eventId}\n`); response.write(`data: ${JSON.stringify({ contractVersion: "workbench-events-v1", serverSentAt: new Date().toISOString(), eventCreatedAt, traceSeq, ...payload })}\n\n`); perf?.recordPhase({ phase: "sse_write", durationMs: nowMs() - startedAt, outcome: "ok" }); }; @@ -192,6 +195,7 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option type: "connected", status: "connected", heartbeatMs, + cursor: { outboxSeq: requestedAfterSeq }, filters: { sessionId: requestedSessionId, traceId: requestedTraceId @@ -216,12 +220,12 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option outboxMode: typeof (options.runtimeStore ?? null)?.readWorkbenchProjectionOutbox === "function" }); emitWorkbenchRealtimeAcceptedOtelSpan(request, options.env); - if (activeTraceId) { + if (activeTraceId && requestedAfterSeq <= 0) { await (perf ? perf.measure("workbench_initial_trace", () => writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" })) : writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" })); } const runtimeStore = options.runtimeStore ?? null; const outboxPollMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_OUTBOX_POLL_MS, 500); - let outboxCursor = 0; + let outboxCursor = requestedAfterSeq; const outboxTraceSnapshots = new Set(activeTraceId ? [activeTraceId] : []); if (typeof runtimeStore?.readWorkbenchProjectionOutbox === "function" && (streamSessionId || activeTraceId)) { const pollOutbox = async () => { @@ -530,6 +534,21 @@ function realtimeTraceSeq(payload) { return Number.isFinite(seq) && seq >= 0 ? seq : null; } +function realtimeEventId(payload) { + const raw = payload?.cursor?.outboxSeq ?? payload?.outboxSeq ?? payload?.id; + const text = textValue(raw); + if (!text || /[\r\n]/u.test(text)) return null; + return text.slice(0, 128); +} + +function workbenchRealtimeAfterSeq(request, url) { + for (const value of [url.searchParams.get("afterSeq"), url.searchParams.get("afterOutboxSeq"), request?.headers?.["last-event-id"]]) { + const parsed = nonNegativeInteger(value, 0); + if (parsed > 0) return parsed; + } + return 0; +} + function nowMs() { if (typeof performance !== "undefined" && typeof performance.now === "function") return performance.now(); return Date.now(); diff --git a/web/hwlab-cloud-web/src/api/workbench-events.ts b/web/hwlab-cloud-web/src/api/workbench-events.ts index 00f47a36..d99c34a3 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.ts @@ -34,13 +34,15 @@ export interface WorkbenchRealtimeEvent { reason?: string | null; error?: { code?: string; message?: string; [key: string]: unknown } | null; traceSeq?: number | null; - cursor?: { traceSeq?: number | null; [key: string]: unknown }; + outboxSeq?: number | null; + cursor?: { traceSeq?: number | null; outboxSeq?: number | null; [key: string]: unknown }; [key: string]: unknown; } export interface WorkbenchEventStreamOptions { sessionId?: string | null; traceId?: string | null; + afterSeq?: number | null; onEvent: (event: WorkbenchRealtimeEvent, eventName: string) => void; onOpen?: () => void; onError?: (event: Event) => void; @@ -66,6 +68,7 @@ export function connectWorkbenchEvents(options: WorkbenchEventStreamOptions): Wo const params = new URLSearchParams(); appendParam(params, "sessionId", options.sessionId); appendParam(params, "traceId", options.traceId); + appendNumberParam(params, "afterSeq", options.afterSeq); const eventRoute = `/v1/workbench/events?${params.toString()}`; const connectStartedAt = typeof performance === "undefined" ? Date.now() : performance.now(); recordWorkbenchSseLifecycle({ state: "connect", route: eventRoute, sessionId: options.sessionId, traceId: options.traceId }); @@ -105,6 +108,12 @@ function appendParam(params: URLSearchParams, key: string, value: string | null if (text) params.set(key, text); } +function appendNumberParam(params: URLSearchParams, key: string, value: number | null | undefined): void { + const parsed = Number(value); + if (!Number.isFinite(parsed) || parsed <= 0) return; + params.set(key, String(Math.trunc(parsed))); +} + function parseRealtimeEvent(raw: string): WorkbenchRealtimeEvent | null { try { const value = JSON.parse(raw); diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 6207f12e..a8a4118e 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -79,6 +79,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const realtimeSessionRefreshInFlight = new Set(); let realtimeStream: WorkbenchEventStream | null = null; let realtimeKey = ""; + const realtimeOutboxSeqByKey = new Map(); const sessionListRefreshInFlight = new Map>(); const sessionListRefreshTimers = new Map(); const sessionListLastRefreshAtByKey = new Map(); @@ -1082,9 +1083,11 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project stopRealtime(); realtimeKey = key; if (!sessionId && !traceId) return; + const afterSeq = realtimeOutboxSeqByKey.get(key) ?? null; realtimeStream = connectWorkbenchEvents({ sessionId, traceId, + afterSeq, onOpen: () => undefined, onError: () => undefined, onEvent: (event, eventName) => applyRealtimeEvent(event, eventName) @@ -1108,6 +1111,22 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project return normalizeWorkbenchSessionId(firstNonEmptyString(event.sessionId, turn?.sessionId, snapshot?.sessionId, traceEvent?.sessionId)); } + function rememberRealtimeOutboxSeq(event: WorkbenchRealtimeEvent): void { + const outboxSeq = realtimeOutboxSeq(event); + if (outboxSeq === null) return; + const sessionId = realtimeEventSessionId(event) ?? selectedSessionId.value ?? null; + const traceId = firstNonEmptyString(event.traceId, event.event?.traceId, event.snapshot?.traceId, currentRequest.value?.traceId, activeTraceIdFromMessages(messages.value, turnStatusAuthority.value)); + const key = [sessionId ?? "", traceId ?? ""].join("|"); + if (!key.trim()) return; + const previous = realtimeOutboxSeqByKey.get(key) ?? 0; + if (outboxSeq > previous) realtimeOutboxSeqByKey.set(key, outboxSeq); + } + + function realtimeOutboxSeq(event: WorkbenchRealtimeEvent): number | null { + const parsed = Number(event.cursor?.outboxSeq ?? event.outboxSeq); + return Number.isFinite(parsed) && parsed >= 0 ? Math.trunc(parsed) : null; + } + function traceResultSessionId(result: AgentChatResultResponse | TraceSnapshot | Record | null | undefined): string | null { const value = recordValue(result); const runnerTrace = recordValue(value?.runnerTrace); @@ -1171,6 +1190,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project function applyRealtimeEvent(event: WorkbenchRealtimeEvent, eventName: string): void { recordActivity(event.type ? `realtime:${event.type}` : `realtime:${eventName}`); + rememberRealtimeOutboxSeq(event); if (event.type === "trace.snapshot") { applyRealtimeTraceSnapshot(event.traceId, event.snapshot); return;