From 2651a6ee5517aa4fd95d7d0e7de0bbdf4e3447a0 Mon Sep 17 00:00:00 2001
From: Lyon <88232613+pikasTech@users.noreply.github.com>
Date: Wed, 24 Jun 2026 23:21:02 +0800
Subject: [PATCH] fix(workbench): replay SSE outbox through runtime service
 (#2070)

---
 internal/cloud/server-workbench-http.test.ts |  4 ++--
 internal/cloud/server-workbench-http.ts      | 14 +++++++++-----
 2 files changed, 11 insertions(+), 7 deletions(-)

diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts
index d9407c98..e8e84ff6 100644
--- a/internal/cloud/server-workbench-http.test.ts
+++ b/internal/cloud/server-workbench-http.test.ts
@@ -1585,7 +1585,7 @@ test("workbench realtime stream replays durable outbox after requested seq", asy
     session: { sessionStatus: "running", lastTraceId: traceId }
   };
   const outboxQueries = [];
-  const runtimeStore = {
+  const workbenchRuntime = {
     async readWorkbenchProjectionOutbox(params = {}) {
       outboxQueries.push({ ...params });
       return [
@@ -1601,7 +1601,7 @@ test("workbench realtime stream replays durable outbox after requested seq", asy
     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" } });
+  const server = createCloudApiServer({ accessController, workbenchRuntime, env: { HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000" } });
   await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
 
   try {
diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts
index dc0a2453..48240477 100644
--- a/internal/cloud/server-workbench-http.ts
+++ b/internal/cloud/server-workbench-http.ts
@@ -112,11 +112,12 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
     const requestedSessionId = safeSessionId(url.searchParams.get("sessionId") ?? url.searchParams.get("includeSessionId"));
     const requestedTraceId = safeTraceId(url.searchParams.get("traceId"));
     const heartbeatMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_HEARTBEAT_MS, DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS);
+    const realtimeOutboxReader = workbenchRuntimeOutboxReader(request, options);
     attachWorkbenchRealtimeOtelContext(request, {
       sessionId: requestedSessionId,
       traceId: requestedTraceId,
       heartbeatMs,
-      outboxMode: typeof (options.runtimeStore ?? null)?.readWorkbenchProjectionOutbox === "function"
+      outboxMode: typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function"
     });
     const perf = options.backendPerformance;
     const auth = perf ? await perf.measure("workbench_auth", () => authenticateWorkbenchRead(request, response, options)) : await authenticateWorkbenchRead(request, response, options);
@@ -217,24 +218,23 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
     traceId: activeTraceId,
     threadId: streamThreadId,
     heartbeatMs,
-    outboxMode: typeof (options.runtimeStore ?? null)?.readWorkbenchProjectionOutbox === "function"
+    outboxMode: typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function"
   });
   emitWorkbenchRealtimeAcceptedOtelSpan(request, options.env);
   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 = requestedAfterSeq;
   const outboxTraceSnapshots = new Set(activeTraceId ? [activeTraceId] : []);
-  if (typeof runtimeStore?.readWorkbenchProjectionOutbox === "function" && (streamSessionId || activeTraceId)) {
+  if (typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function" && (streamSessionId || activeTraceId)) {
       const pollOutbox = async () => {
         if (closed || response.destroyed) return;
         try {
           const query = { afterSeq: outboxCursor, limit: 50 };
           if (streamSessionId) query.sessionId = streamSessionId;
           else query.traceId = activeTraceId;
-          const rows = await runtimeStore.readWorkbenchProjectionOutbox(query);
+          const rows = await realtimeOutboxReader.readWorkbenchProjectionOutbox(query);
           for (const row of rows) {
             if (closed || response.destroyed) break;
             outboxCursor = row.outboxSeq;
@@ -737,6 +737,10 @@ function workbenchRuntimeFactsReadModel(request, options, actor) {
   };
 }
 
+function workbenchRuntimeOutboxReader(request, options) {
+  return options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger, traceparent: request?.hwlabHttpRequestContext?.traceparent });
+}
+
 async function handleWorkbenchSessionListBunLegacy(response, url, options, actor) {
   if (url.searchParams.has("projectId") || url.searchParams.has("workspaceId")) return sendJson(response, 400, workbenchError("workbench_authority_removed", "Workbench session list is keyed by sessionId only."));
   const perf = options.backendPerformance;