Merge pull request #2460 from pikasTech/fix/workbench-kafka-session-filter

补齐 Workbench Kafka SSE 实时权威合同

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);

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;

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);