From 2cb8be409c22a4dea667cf85d45b71d5cedb0ba7 Mon Sep 17 00:00:00 2001 From: root Date: Thu, 9 Jul 2026 21:31:12 +0200 Subject: [PATCH] Make workbench realtime SSE consume Kafka --- internal/cloud/server-workbench-http.test.ts | 77 ++++- internal/cloud/server-workbench-http.ts | 277 ++++++++++-------- internal/cloud/workbench-facts-store.ts | 30 -- internal/cloud/workbench-read-model.ts | 1 - ...kbench-realtime-authority-contract.test.ts | 73 ++++- 5 files changed, 292 insertions(+), 166 deletions(-) diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index 30d91f53..bf1ab963 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/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" } }); - await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const serverWithKafka = createCloudApiServer({ + 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 events = await getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}&afterSeq=10`, 2); + const { port } = serverWithKafka.address(); + 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].data.cursor.outboxSeq, 11); + assert.equal(events[1].id, "hwlab.event.v1:0:0"); + 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.equal(outboxQueries[0].sessionId, session.id); + assert.deepEqual(outboxQueries, []); } 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"); +} diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index b985b3c2..cf53c44e 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/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); - let outboxCursor = requestedAfterSeq; - const outboxTraceSnapshots = new Set(activeTraceId ? [activeTraceId] : []); - if (typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function" && (streamSessionId || activeTraceId)) { - const pollOutbox = async () => { + const kafkaFilters = resolveWorkbenchRealtimeKafkaFilters({ traceId: activeTraceId, sessionId: streamSessionId, traceStore }); + try { + const kafkaStream = await openKafkaEventStream({ + env: options.env ?? process.env, + stream: "hwlab", + fromBeginning: false, + ...kafkaFilters, + kafkaFactory: options.kafkaFactory, + onEvent: async (record) => { if (closed || response.destroyed) return; - try { - const query = { afterSeq: outboxCursor, limit: 50 }; - if (streamSessionId) query.sessionId = streamSessionId; - else query.traceId = activeTraceId; - const rows = await realtimeOutboxReader.readWorkbenchProjectionOutbox(query); - for (const row of rows) { - if (closed || response.destroyed) break; - outboxCursor = row.outboxSeq; - const rowTraceId = safeTraceId(row.traceId) ?? activeTraceId; - const rowSessionId = safeSessionId(row.sessionId ?? streamSessionId) ?? streamSessionId; - if (requestedAfterSeq <= 0 && rowTraceId && !outboxTraceSnapshots.has(rowTraceId)) { - outboxTraceSnapshots.add(rowTraceId); - await writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: rowTraceId, reason: "outbox-session" }); - } - if (row.commitType === "message") { - await writeMessageRealtimeSnapshotSafe({ - writeEvent, - readModel, - actor: auth.actor, - sessionId: rowSessionId, - threadId: streamThreadId, - traceId: rowTraceId, - reason: "outbox-message", - cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq }, - entity: workbenchRealtimeEntityDelta(row) - }); - 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 realtime = workbenchRealtimeEventFromKafka(record, { sessionId: streamSessionId, threadId: streamThreadId, traceId: activeTraceId }); + if (!realtime) return; + writeEvent("workbench.trace.event", realtime.traceEvent); + if (realtime.turnSnapshot) writeEvent("workbench.turn.snapshot", realtime.turnSnapshot); + }, + onError: (error) => { + writeEvent("workbench.error", { + type: "error", + realtimeSource: "kafka", + sessionId: streamSessionId, + threadId: streamThreadId, + traceId: activeTraceId, + error: { code: "workbench_kafka_stream_failed", message: error?.message ?? "Workbench Kafka realtime stream failed." } + }); + } + }); + cleanup.push(() => { void kafkaStream.stop?.(); }); + } catch (kafkaError) { + writeEvent("workbench.error", { + type: "error", + realtimeSource: "kafka", + sessionId: streamSessionId, + threadId: streamThreadId, + traceId: activeTraceId, + error: { code: "workbench_kafka_stream_unavailable", message: kafkaError?.message ?? "Workbench Kafka realtime stream is unavailable." } + }); + } 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(); diff --git a/internal/cloud/workbench-facts-store.ts b/internal/cloud/workbench-facts-store.ts index 2c63483f..e3cd815b 100644 --- a/internal/cloud/workbench-facts-store.ts +++ b/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) }; } diff --git a/internal/cloud/workbench-read-model.ts b/internal/cloud/workbench-read-model.ts index 9fc1db8f..fcde5182 100644 --- a/internal/cloud/workbench-read-model.ts +++ b/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 }) }; diff --git a/internal/cloud/workbench-realtime-authority-contract.test.ts b/internal/cloud/workbench-realtime-authority-contract.test.ts index 5a864463..ef8c3a37 100644 --- a/internal/cloud/workbench-realtime-authority-contract.test.ts +++ b/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); - assert.deepEqual(events.map((event) => event.event), ["workbench.connected", "workbench.trace.snapshot"]); - assert.equal(events[1].data.entity.family, sync.body.events[0].family); - assert.equal(events[1].data.entity.id, sync.body.events[0].id); - assert.equal(events[1].data.entity.entity.version, sync.body.events[0].entity.version); - assert.equal(events[1].data.entity.cursor.outboxSeq, sync.body.events[0].cursor.outboxSeq); - assert.equal(events[1].data.entity.projectionRevision, sync.body.events[0].projectionRevision); - assert.deepEqual(outboxQueries.map((query) => query.afterSeq), [10, 10]); + setTimeout(() => { void fakeKafka.emit({ + eventType: "hwlab.trace.event.projected", + sessionId, + traceId, + context: { sourceSeq: 8, runId: "run_realtime_authority_p1", commandId: "cmd_realtime_authority_p1" }, + event: { type: "result", eventType: "terminal", status: "completed", label: "agentrun:terminal:completed", message: "redacted final", terminal: true, sourceSeq: 8 } + }); }, 0); + 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)); }