diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index a04ac5e9..1351e790 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -2403,6 +2403,10 @@ test("workbench realtime stream forwards HWLAB Kafka events after initial connec assert.equal(events[0].id, "10"); assert.equal(events[1].id, "hwlab.event.v1:0:0"); assert.equal(events[1].data.realtimeSource, "kafka"); + assert.equal(events[1].data.realtimeAuthority, "workbench-realtime-authority-v2"); + assert.equal(events[1].data.entity.family, "traceEvents"); + assert.equal(events[1].data.entity.version, 7); + assert.equal(events[1].data.entity.projectionRevision, "kafka:hwlab.event.v1:0:0"); 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); diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 618047be..7ba4a0d4 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -26,6 +26,7 @@ const DEFAULT_SESSION_LIST_LIMIT = 20; const MAX_PAGE_LIMIT = 100; const DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS = 15000; const WORKBENCH_REALTIME_DRAIN_TIMEOUT_MS = 2500; +const WORKBENCH_REALTIME_AUTHORITY_VERSION = "workbench-realtime-authority-v2"; const WORKBENCH_SESSION_LIST_PAGE_FAMILIES = Object.freeze(["sessions"]); const WORKBENCH_SESSION_DETAIL_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]); const WORKBENCH_SESSION_MESSAGE_PAGE_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]); @@ -410,9 +411,21 @@ function workbenchRealtimeEventFromKafka(record, context = {}) { kafkaOffset: textValue(record.offset), kafkaPartition: integerValue(record.partition) }; + const entityVersion = sourceSeq ?? integerValue(record.offset) ?? 0; + const projectionRevision = ["kafka", kafka.topic, kafka.partition, kafka.offset].filter((part) => textValue(part)).join(":"); + const traceEntity = { + family: "traceEvents", + id: [traceId ?? "trace", event.runId, event.commandId, kafka.partition, kafka.offset].filter((part) => textValue(part)).join(":"), + version: entityVersion, + traceSeq: cursor.traceSeq, + projectionRevision, + authority: WORKBENCH_REALTIME_AUTHORITY_VERSION + }; const traceEvent = { id: kafka.topic && kafka.partition !== null && kafka.offset ? `${kafka.topic}:${kafka.partition}:${kafka.offset}` : null, type: "trace.event", + contractVersion: "workbench-sync-v1", + realtimeAuthority: WORKBENCH_REALTIME_AUTHORITY_VERSION, realtimeSource: "kafka", sessionId, threadId, @@ -420,11 +433,15 @@ function workbenchRealtimeEventFromKafka(record, context = {}) { event, snapshot: { traceId, sessionId, threadId, status: status || event.status || null, events: [event], eventCount: 1 }, cursor, + entity: traceEntity, + projectionRevision, kafka, valuesPrinted: false }; const turnSnapshot = terminal ? { type: "turn.snapshot", + contractVersion: "workbench-sync-v1", + realtimeAuthority: WORKBENCH_REALTIME_AUTHORITY_VERSION, realtimeSource: "kafka", sessionId, threadId, @@ -448,6 +465,15 @@ function workbenchRealtimeEventFromKafka(record, context = {}) { agentRun: compactObject({ runId: event.runId, commandId: event.commandId }), valuesPrinted: false }, + entity: { + family: "turns", + id: traceId ?? event.runId ?? event.commandId ?? traceEntity.id, + version: entityVersion, + traceSeq: cursor.traceSeq, + projectionRevision, + authority: WORKBENCH_REALTIME_AUTHORITY_VERSION + }, + projectionRevision, kafka, valuesPrinted: false } : null; diff --git a/internal/cloud/workbench-realtime-authority-contract.test.ts b/internal/cloud/workbench-realtime-authority-contract.test.ts index ce2eed94..2033c8b9 100644 --- a/internal/cloud/workbench-realtime-authority-contract.test.ts +++ b/internal/cloud/workbench-realtime-authority-contract.test.ts @@ -69,6 +69,10 @@ test("workbench sync delta stays durable while SSE realtime forwards Kafka event 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.realtimeAuthority, "workbench-realtime-authority-v2"); + assert.equal(events[1].data.entity.family, "traceEvents"); + assert.equal(events[1].data.entity.version, 8); + assert.equal(events[1].data.entity.projectionRevision, "kafka:hwlab.event.v1:0:0"); 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);