Merge pull request #2457 from pikasTech/fix/workbench-realtime-kafka-only

让 Workbench 实时 SSE 纯 Kafka 消费

internal/cloud/server-workbench-http.test.ts

@@ -2344,7 +2344,8 @@ test("workbench realtime stream surfaces facts blocker instead of legacy trace f
   }
 });
 
-test("workbench realtime stream replays durable outbox after requested seq", async () => {
+test("workbench realtime stream forwards HWLAB Kafka events after initial connection", async () => {
+  const fakeKafka = createFakeKafkaFactory();
   const traceId = "trc_workbench_realtime_after_seq";
   const session = {
     id: "ses_workbench_realtime_after_seq",
@@ -2375,21 +2376,39 @@ 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, workbenchRuntime, env: { HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000" } });
+  const serverWithKafka = createCloudApiServer({
-  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
+    accessController,
+    workbenchRuntime,
+    kafkaFactory: fakeKafka.factory,
+    env: {
+      HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000",
+      HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092",
+      HWLAB_KAFKA_CLIENT_ID: "test-hwlab-cloud-api",
+      HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1"
+    }
+  });
+  await new Promise((resolve) => serverWithKafka.listen(0, "127.0.0.1", resolve));
 
   try {
-    const { port } = server.address();
+    const { port } = serverWithKafka.address();
-    const events = await getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}&afterSeq=10`, 2);
+    setTimeout(() => { void fakeKafka.emit({
+      eventType: "hwlab.trace.event.projected",
+      sessionId: session.id,
+      traceId,
+      context: { sourceSeq: 7, runId: "run_workbench_realtime_after_seq", commandId: "cmd_workbench_realtime_after_seq" },
+      event: { type: "backend", eventType: "backend", status: "running", label: "agentrun:event:test", message: "Kafka realtime event", sourceSeq: 7 }
+    }); }, 0);
+    const events = await getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}&traceId=${encodeURIComponent(traceId)}&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].id, "hwlab.event.v1:0:0");
-    assert.equal(events[1].data.cursor.outboxSeq, 11);
+    assert.equal(events[1].data.realtimeSource, "kafka");
+    assert.equal(events[1].data.event.message, "Kafka realtime event");
+    assert.equal(events[1].data.kafka.topic, "hwlab.event.v1");
     assert.equal(events[1].data.cursor.traceSeq, 7);
-    assert.equal(outboxQueries[0].afterSeq, 10);
+    assert.deepEqual(outboxQueries, []);
-    assert.equal(outboxQueries[0].sessionId, session.id);
   } finally {
-    await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
+    await new Promise((resolve, reject) => serverWithKafka.close((error) => error ? reject(error) : resolve()));
   }
 });
 
@@ -2931,3 +2950,41 @@ function parseSseBlock(block) {
   const data = lines.filter((line) => line.startsWith("data:")).map((line) => line.slice(5).trim()).join("\n");
   return { event, id, data: JSON.parse(data) };
 }
+
+function createFakeKafkaFactory() {
+  let eachMessage = null;
+  let offset = 0;
+  let subscribedTopic = "hwlab.event.v1";
+  const consumer = {
+    connect: async () => undefined,
+    subscribe: async (input = {}) => { subscribedTopic = String(input.topic ?? subscribedTopic); },
+    run: async (input = {}) => { eachMessage = input.eachMessage; },
+    stop: async () => undefined,
+    disconnect: async () => undefined
+  };
+  return {
+    factory: () => ({ consumer: () => consumer }),
+    emit: async (value = {}) => {
+      await waitFor(() => typeof eachMessage === "function");
+      await eachMessage({
+        topic: subscribedTopic,
+        partition: 0,
+        message: {
+          offset: String(offset++),
+          key: Buffer.from(value.traceId ?? value.sessionId ?? "fake"),
+          timestamp: String(Date.now()),
+          value: Buffer.from(JSON.stringify(value))
+        }
+      });
+    }
+  };
+}
+
+async function waitFor(predicate, timeoutMs = 1000) {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (predicate()) return;
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+  throw new Error("waitFor timed out");
+}

internal/cloud/server-workbench-http.ts

@@ -16,7 +16,8 @@ import {
 import { createWorkbenchReadModel } from "./workbench-read-model.ts";
 import { createWorkbenchRuntimeClient } from "./workbench-runtime-client.ts";
 import { buildWorkbenchSessionDetail, compactLaunchContext, includeMessagesForSessionDetail } from "./workbench-session-detail-response.ts";
-import { handleWorkbenchSyncHttp, workbenchRealtimeEntityDelta } from "./workbench-realtime-authority.ts";
+import { handleWorkbenchSyncHttp } from "./workbench-realtime-authority.ts";
+import { openKafkaEventStream } from "./kafka-event-bridge.ts";
 import { durableTraceStatus, RUNNING_STATUSES, terminalFinalResponse, TERMINAL_STATUSES } from "./workbench-turn-projection.ts";
 import { emitCodeAgentOtelSpan, emitHttpServerRequestSpan } from "./otel-trace.ts";
 
@@ -119,12 +120,11 @@ 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 realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function"
+      realtimeSource: "kafka"
     });
     const perf = options.backendPerformance;
     const auth = perf ? await perf.measure("workbench_auth", () => authenticateWorkbenchRead(request, response, options)) : await authenticateWorkbenchRead(request, response, options);
@@ -202,6 +202,8 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
   writeEvent("workbench.connected", {
     type: "connected",
     status: "connected",
+    realtimeSource: "kafka",
+    snapshotSource: "read-model",
     heartbeatMs,
     cursor: { outboxSeq: requestedAfterSeq },
     filters: {
@@ -238,119 +240,49 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
     traceId: activeTraceId,
     threadId: streamThreadId,
     heartbeatMs,
-    outboxMode: typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function"
+    realtimeSource: "kafka"
   });
   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 outboxPollMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_OUTBOX_POLL_MS, 500);
+  const kafkaFilters = resolveWorkbenchRealtimeKafkaFilters({ traceId: activeTraceId, sessionId: streamSessionId, traceStore });
-  let outboxCursor = requestedAfterSeq;
+  try {
-  const outboxTraceSnapshots = new Set(activeTraceId ? [activeTraceId] : []);
+    const kafkaStream = await openKafkaEventStream({
-  if (typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function" && (streamSessionId || activeTraceId)) {
+      env: options.env ?? process.env,
-      const pollOutbox = async () => {
+      stream: "hwlab",
+      fromBeginning: false,
+      ...kafkaFilters,
+      kafkaFactory: options.kafkaFactory,
+      onEvent: async (record) => {
         if (closed || response.destroyed) return;
-        try {
+        const realtime = workbenchRealtimeEventFromKafka(record, { sessionId: streamSessionId, threadId: streamThreadId, traceId: activeTraceId });
-          const query = { afterSeq: outboxCursor, limit: 50 };
+        if (!realtime) return;
-          if (streamSessionId) query.sessionId = streamSessionId;
+        writeEvent("workbench.trace.event", realtime.traceEvent);
-          else query.traceId = activeTraceId;
+        if (realtime.turnSnapshot) writeEvent("workbench.turn.snapshot", realtime.turnSnapshot);
-          const rows = await realtimeOutboxReader.readWorkbenchProjectionOutbox(query);
+      },
-          for (const row of rows) {
+      onError: (error) => {
-            if (closed || response.destroyed) break;
+        writeEvent("workbench.error", {
-            outboxCursor = row.outboxSeq;
+          type: "error",
-            const rowTraceId = safeTraceId(row.traceId) ?? activeTraceId;
+          realtimeSource: "kafka",
-            const rowSessionId = safeSessionId(row.sessionId ?? streamSessionId) ?? streamSessionId;
+          sessionId: streamSessionId,
-            if (requestedAfterSeq <= 0 && rowTraceId && !outboxTraceSnapshots.has(rowTraceId)) {
+          threadId: streamThreadId,
-              outboxTraceSnapshots.add(rowTraceId);
+          traceId: activeTraceId,
-              await writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: rowTraceId, reason: "outbox-session" });
+          error: { code: "workbench_kafka_stream_failed", message: error?.message ?? "Workbench Kafka realtime stream failed." }
-            }
+        });
-            if (row.commitType === "message") {
+      }
-              await writeMessageRealtimeSnapshotSafe({
+    });
-                writeEvent,
+    cleanup.push(() => { void kafkaStream.stop?.(); });
-                readModel,
+  } catch (kafkaError) {
-                actor: auth.actor,
+    writeEvent("workbench.error", {
-                sessionId: rowSessionId,
+      type: "error",
-                threadId: streamThreadId,
+      realtimeSource: "kafka",
-                traceId: rowTraceId,
+      sessionId: streamSessionId,
-                reason: "outbox-message",
+      threadId: streamThreadId,
-                cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq },
+      traceId: activeTraceId,
-                entity: workbenchRealtimeEntityDelta(row)
+      error: { code: "workbench_kafka_stream_unavailable", message: kafkaError?.message ?? "Workbench Kafka realtime stream is unavailable." }
-              });
+    });
-              continue;
+  }
-            }
-            if (row.commitType === "terminal") {
-              writeEvent("workbench.trace.snapshot", {
-                type: "trace.snapshot",
-                sessionId: rowSessionId,
-                threadId: streamThreadId,
-                traceId: rowTraceId,
-                reason: "outbox-terminal",
-                cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq },
-                entity: workbenchRealtimeEntityDelta(row),
-                projectionStatus: { terminal: row.terminal, sealed: row.sealed }
-              });
-              await writeMessageRealtimeSnapshotSafe({
-                writeEvent,
-                readModel,
-                actor: auth.actor,
-                sessionId: rowSessionId,
-                threadId: streamThreadId,
-                traceId: rowTraceId,
-                reason: "outbox-terminal",
-                cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq },
-                entity: workbenchRealtimeEntityDelta(row)
-              });
-              if (rowTraceId) void writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: rowTraceId, reason: "terminal" });
-            } else {
-              writeEvent("workbench.trace.event", {
-                type: "trace.event",
-                sessionId: rowSessionId,
-                threadId: streamThreadId,
-                traceId: rowTraceId,
-                cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq },
-                entity: workbenchRealtimeEntityDelta(row)
-              });
-            }
-          }
-        } 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 if (activeTraceId) {
-      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 writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "terminal" });
-        }
-      });
-      cleanup.push(unsubscribe);
-    }
 
   const heartbeat = setInterval(async () => {
     try {
@@ -400,10 +332,129 @@ function attachWorkbenchRealtimeOtelContext(request, fields = {}) {
     "workbench.trace_id": fields.traceId ?? null,
     "workbench.thread_id": fields.threadId ?? null,
     "workbench.sse.heartbeat_ms": fields.heartbeatMs ?? null,
-    "workbench.sse.outbox_mode": fields.outboxMode === true
+    "workbench.sse.realtime_source": fields.realtimeSource ?? null,
+    "workbench.sse.outbox_mode": false
   };
 }
 
+function resolveWorkbenchRealtimeKafkaFilters({ traceId = null, sessionId = null, traceStore = null } = {}) {
+  const resolved = traceId && typeof traceStore?.snapshot === "function" ? collectTraceLinkedIds(traceStore.snapshot(traceId)) : {};
+  const hasAgentRunKey = Boolean(resolved.runId || resolved.commandId);
+  return compactObject({
+    traceId: hasAgentRunKey ? null : traceId,
+    sessionId: hasAgentRunKey ? null : (resolved.sessionId || sessionId),
+    runId: resolved.runId,
+    commandId: resolved.commandId
+  });
+}
+
+function collectTraceLinkedIds(snapshot) {
+  const out = { sessionId: null, runId: null, commandId: null };
+  const events = Array.isArray(snapshot?.events) ? snapshot.events : [];
+  for (const event of events) collectTraceLinkedIdsFromRecord(event, out);
+  collectTraceLinkedIdsFromRecord(snapshot?.lastEvent, out);
+  return out;
+}
+
+function collectTraceLinkedIdsFromRecord(record, out) {
+  const value = record && typeof record === "object" && !Array.isArray(record) ? record : {};
+  const agentRun = value.agentRun && typeof value.agentRun === "object" && !Array.isArray(value.agentRun) ? value.agentRun : {};
+  const payload = value.payload && typeof value.payload === "object" && !Array.isArray(value.payload) ? value.payload : {};
+  out.sessionId ||= safeSessionId(value.sessionId ?? value.sourceSessionId ?? agentRun.sessionId ?? payload.sessionId);
+  out.runId ||= textValue(value.runId ?? value.sourceRunId ?? agentRun.runId ?? payload.runId);
+  out.commandId ||= textValue(value.commandId ?? value.sourceCommandId ?? agentRun.commandId ?? payload.commandId);
+}
+
+function workbenchRealtimeEventFromKafka(record, context = {}) {
+  const value = record?.value && typeof record.value === "object" && !Array.isArray(record.value) ? record.value : null;
+  if (!value) return null;
+  const hwlabEvent = value.event && typeof value.event === "object" && !Array.isArray(value.event) ? value.event : {};
+  const sourceContext = value.context && typeof value.context === "object" && !Array.isArray(value.context) ? value.context : {};
+  const traceId = safeTraceId(context.traceId ?? value.traceId ?? hwlabEvent.traceId) ?? textValue(context.traceId ?? value.traceId ?? hwlabEvent.traceId);
+  const sessionId = safeSessionId(context.sessionId ?? value.sessionId ?? hwlabEvent.sessionId) ?? textValue(context.sessionId ?? value.sessionId ?? hwlabEvent.sessionId);
+  const threadId = safeOpaqueId(context.threadId ?? sourceContext.threadId) ?? textValue(context.threadId ?? sourceContext.threadId);
+  if (!traceId && !sessionId) return null;
+  const sourceSeq = integerValue(hwlabEvent.sourceSeq ?? sourceContext.sourceSeq);
+  const kafka = {
+    topic: textValue(record.topic),
+    partition: integerValue(record.partition),
+    offset: textValue(record.offset),
+    key: textValue(record.key),
+    timestamp: textValue(record.timestamp),
+    valuesRedacted: true
+  };
+  const event = {
+    ...hwlabEvent,
+    traceId: traceId ?? hwlabEvent.traceId ?? null,
+    sessionId: sessionId ?? hwlabEvent.sessionId ?? null,
+    threadId: threadId ?? sourceContext.threadId ?? null,
+    runId: textValue(hwlabEvent.runId ?? sourceContext.runId),
+    commandId: textValue(hwlabEvent.commandId ?? sourceContext.commandId),
+    source: textValue(hwlabEvent.source) || "hwlab.kafka",
+    sourceSeq,
+    kafka,
+    valuesPrinted: false
+  };
+  const status = normalizeStatus(event.status);
+  const terminal = event.terminal === true || TERMINAL_STATUSES.has(status);
+  const cursor = {
+    traceSeq: sourceSeq ?? integerValue(record.offset),
+    kafkaOffset: textValue(record.offset),
+    kafkaPartition: integerValue(record.partition)
+  };
+  const traceEvent = {
+    id: kafka.topic && kafka.partition !== null && kafka.offset ? `${kafka.topic}:${kafka.partition}:${kafka.offset}` : null,
+    type: "trace.event",
+    realtimeSource: "kafka",
+    sessionId,
+    threadId,
+    traceId,
+    event,
+    snapshot: { traceId, sessionId, threadId, status: status || event.status || null, events: [event], eventCount: 1 },
+    cursor,
+    kafka,
+    valuesPrinted: false
+  };
+  const turnSnapshot = terminal ? {
+    type: "turn.snapshot",
+    realtimeSource: "kafka",
+    sessionId,
+    threadId,
+    traceId,
+    reason: "kafka-terminal",
+    cursor,
+    turn: {
+      traceId,
+      sessionId,
+      threadId,
+      status: status || event.status || "completed",
+      running: false,
+      terminal: true,
+      finalResponse: terminalFinalResponse(status || event.status || "completed", {
+        traceId,
+        status: status || event.status || "completed",
+        finalResponse: event.text ? { text: event.text, status: status || event.status || "completed", traceId, source: "kafka-terminal-event", valuesPrinted: false } : null,
+        finalText: event.text ?? event.message ?? null,
+        error: event.errorCode ? { message: event.message ?? event.errorCode } : null
+      }),
+      agentRun: compactObject({ runId: event.runId, commandId: event.commandId }),
+      valuesPrinted: false
+    },
+    kafka,
+    valuesPrinted: false
+  } : null;
+  return { traceEvent, turnSnapshot };
+}
+
+function compactObject(value) {
+  return Object.fromEntries(Object.entries(value || {}).filter(([, entry]) => textValue(entry)));
+}
+
+function integerValue(value) {
+  const number = Number(value);
+  return Number.isFinite(number) ? Math.trunc(number) : null;
+}
+
 function attachWorkbenchReadModelDiagnosticOtel(request, fields = {}) {
   const context = request?.hwlabHttpRequestContext;
   if (!context) return;
@@ -848,10 +899,6 @@ function workbenchRuntimeFactsQueryError(error) {
   return normalized;
 }
 
-function workbenchRuntimeOutboxReader(request, options) {
-  return workbenchRuntimeReader(request, options);
-}
-
 function queryWorkbenchSessionInclude(readModel, includeRouteId, perf = null) {
   const run = () => queryFactsByRouteId(readModel, includeRouteId, { families: WORKBENCH_SESSION_LIST_PAGE_FAMILIES });
   const promise = perf ? perf.measure("workbench_session_include_query", run) : run();

internal/cloud/workbench-facts-store.ts

@@ -149,34 +149,6 @@ export function createWorkbenchFactsStore(options = {}, actor = null) {
     return durable ?? memory;
   }
 
-  function subscribeTrace(traceId, listener) {
-    return traceStore.subscribe(traceId, listener);
-  }
-
-  function subscribeDurableProjection(traceId, listener, pollOptions = {}) {
-    if (typeof runtimeReader?.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 runtimeReader.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,
@@ -186,8 +158,6 @@ export function createWorkbenchFactsStore(options = {}, actor = null) {
     resultForTrace,
     traceSnapshot,
     traceSnapshotSync,
-    subscribeTrace,
-    subscribeDurableProjection,
     canReadOwner: (ownerUserId) => canActorReadOwner(ownerUserId, actor)
   };
 }

internal/cloud/workbench-read-model.ts

@@ -17,7 +17,6 @@ export function createWorkbenchReadModel(options = {}, actor = null) {
     resultForTrace: (traceId) => facts.resultForTrace(traceId),
     traceSnapshot: (traceId) => facts.traceSnapshot(traceId),
     traceSnapshotSync: (traceId) => facts.traceSnapshotSync(traceId),
-    subscribeTrace: (traceId, listener) => facts.subscribeTrace(traceId, listener),
     canReadOwner: (ownerUserId) => facts.canReadOwner(ownerUserId),
     projectionDiagnostics: ({ traceId, result = null, trace = null, projection = null, refreshError = null } = {}) => workbenchProjectionDiagnostics({ traceId, result, trace, projection, refreshError })
   };

internal/cloud/workbench-realtime-authority-contract.test.ts

@@ -8,7 +8,7 @@ import { createCloudApiServer } from "./server.ts";
 
 const ACTOR = { id: "usr_workbench_realtime_authority", username: "reader", displayName: "Reader", role: "user", status: "active" };
 
-test("workbench sync delta and SSE replay expose the same typed entity cursor", async () => {
+test("workbench sync delta stays durable while SSE realtime forwards Kafka events", async () => {
   const sessionId = "ses_realtime_authority_p1";
   const traceId = "trc_realtime_authority_p1";
   const outboxRows = [{
@@ -32,7 +32,18 @@ test("workbench sync delta and SSE replay expose the same typed entity cursor",
   }];
   const outboxQueries = [];
   const runtime = createRuntime({ facts: durableFacts({ sessionId, traceId }), outboxRows, outboxQueries });
-  const server = createCloudApiServer({ accessController: createAccessController(), workbenchRuntime: runtime, env: { HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000" } });
+  const fakeKafka = createFakeKafkaFactory();
+  const server = createCloudApiServer({
+    accessController: createAccessController(),
+    workbenchRuntime: runtime,
+    kafkaFactory: fakeKafka.factory,
+    env: {
+      HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000",
+      HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092",
+      HWLAB_KAFKA_CLIENT_ID: "test-hwlab-cloud-api",
+      HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1"
+    }
+  });
   await listen(server);
 
   try {
@@ -48,14 +59,18 @@ test("workbench sync delta and SSE replay expose the same typed entity cursor",
     assert.equal(sync.body.events[0].terminal, true);
     assert.equal(sync.body.events[0].sealed, true);
 
-    const events = await getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(sessionId)}&afterSeq=10`, 2);
+    setTimeout(() => { void fakeKafka.emit({
-    assert.deepEqual(events.map((event) => event.event), ["workbench.connected", "workbench.trace.snapshot"]);
+      eventType: "hwlab.trace.event.projected",
-    assert.equal(events[1].data.entity.family, sync.body.events[0].family);
+      sessionId,
-    assert.equal(events[1].data.entity.id, sync.body.events[0].id);
+      traceId,
-    assert.equal(events[1].data.entity.entity.version, sync.body.events[0].entity.version);
+      context: { sourceSeq: 8, runId: "run_realtime_authority_p1", commandId: "cmd_realtime_authority_p1" },
-    assert.equal(events[1].data.entity.cursor.outboxSeq, sync.body.events[0].cursor.outboxSeq);
+      event: { type: "result", eventType: "terminal", status: "completed", label: "agentrun:terminal:completed", message: "redacted final", terminal: true, sourceSeq: 8 }
-    assert.equal(events[1].data.entity.projectionRevision, sync.body.events[0].projectionRevision);
+    }); }, 0);
-    assert.deepEqual(outboxQueries.map((query) => query.afterSeq), [10, 10]);
+    const events = await getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(sessionId)}&traceId=${encodeURIComponent(traceId)}&afterSeq=10`, 2);
+    assert.deepEqual(events.map((event) => event.event), ["workbench.connected", "workbench.trace.event"]);
+    assert.equal(events[1].data.realtimeSource, "kafka");
+    assert.equal(events[1].data.event.message, "redacted final");
+    assert.deepEqual(outboxQueries.map((query) => query.afterSeq), [10]);
     assert.equal(outboxQueries.every((query) => query.sessionId === sessionId), true);
   } finally {
     await close(server);
@@ -291,6 +306,44 @@ function parseSseBlock(block) {
   return { event, id, data: JSON.parse(dataLine) };
 }
 
+function createFakeKafkaFactory() {
+  let eachMessage = null;
+  let offset = 0;
+  let subscribedTopic = "hwlab.event.v1";
+  const consumer = {
+    connect: async () => undefined,
+    subscribe: async (input = {}) => { subscribedTopic = String(input.topic ?? subscribedTopic); },
+    run: async (input = {}) => { eachMessage = input.eachMessage; },
+    stop: async () => undefined,
+    disconnect: async () => undefined
+  };
+  return {
+    factory: () => ({ consumer: () => consumer }),
+    emit: async (value = {}) => {
+      await waitFor(() => typeof eachMessage === "function");
+      await eachMessage({
+        topic: subscribedTopic,
+        partition: 0,
+        message: {
+          offset: String(offset++),
+          key: Buffer.from(value.traceId ?? value.sessionId ?? "fake"),
+          timestamp: String(Date.now()),
+          value: Buffer.from(JSON.stringify(value))
+        }
+      });
+    }
+  };
+}
+
+async function waitFor(predicate, timeoutMs = 1000) {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (predicate()) return;
+    await new Promise((resolve) => setTimeout(resolve, 10));
+  }
+  throw new Error("waitFor timed out");
+}
+
 function listen(server) {
   return new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
 }