fix(workbench): replay realtime outbox by after seq (#2068)

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) };
 }

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();

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);

web/hwlab-cloud-web/src/stores/workbench.ts

@@ -79,6 +79,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   const realtimeSessionRefreshInFlight = new Set<string>();
   let realtimeStream: WorkbenchEventStream | null = null;
   let realtimeKey = "";
+  const realtimeOutboxSeqByKey = new Map<string, number>();
   const sessionListRefreshInFlight = new Map<string, Promise<void>>();
   const sessionListRefreshTimers = new Map<string, number>();
   const sessionListLastRefreshAtByKey = new Map<string, number>();
@@ -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<string, unknown> | 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;