mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make workbench realtime SSE consume Kafka

Browse Source
This commit is contained in:
root
2026-07-09 21:31:12 +02:00
parent 307db9f270
commit 2cb8be409c
5 changed files with 292 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/server-workbench-http.test.ts
+67 -10
View File
@@ -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");
}
internal/cloud/server-workbench-http.ts
+162 -115
View File
@@ -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();
internal/cloud/workbench-facts-store.ts
-30
View File
@@ -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
-1
View File
@@ -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
+63 -10
View File
@@ -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));
}