fix: 恢复 Workbench Kafka 回放单一权威

internal/cloud/kafka-event-bridge.test.ts

@@ -490,6 +490,22 @@ test("composable Kafka capabilities reject consumer group collisions", () => {
   );
 });
 
+test("Workbench bridge rejects direct and transactional dual authority", () => {
+  assert.throws(
+    () => startHwlabKafkaEventBridge({
+      env: {
+        ...PROJECTOR_ENV,
+        HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "true",
+        HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true",
+        HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID: "hwlab-direct-authority-test",
+        HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID: "hwlab-live-authority-test"
+      },
+      runtimeStore: {}
+    }),
+    (error: any) => error?.code === "hwlab_workbench_realtime_dual_authority"
+  );
+});
+
 test("Kafka refresh replay requires explicit YAML budgets and composes with live SSE", () => {
   const refreshEnv = {
     ...LIVE_ENV,

internal/cloud/kafka-event-bridge.ts

@@ -119,6 +119,7 @@ export function kafkaEventBridgeConfig(env = process.env) {
 export function startHwlabKafkaEventBridge({ env = process.env, logger = console, kafkaFactory = defaultKafkaFactory, runtimeStore = null, now = () => new Date().toISOString(), otelSpanEmitter = emitCodeAgentOtelSpan } = {}) {
   const config = kafkaEventBridgeConfig(env);
   if (!config) return { started: false, reason: "disabled_or_unconfigured", stop() {}, subscribeProjectionCommits() { return () => {}; }, valuesPrinted: false };
+  assertSingleWorkbenchRealtimeAuthority(config.capabilities);
   const components = [];
   if (config.capabilities.directPublish || config.capabilities.liveKafkaSse) {
     components.push(startLiveHwlabKafkaEventBridge({ config, env, logger, kafkaFactory, otelSpanEmitter }));
@@ -129,6 +130,16 @@ export function startHwlabKafkaEventBridge({ env = process.env, logger = console
   return components.length === 1 ? components[0] : combineKafkaEventBridgeComponents(config, components);
 }
 
+function assertSingleWorkbenchRealtimeAuthority(capabilities = {}) {
+  const directAuthority = capabilities.directPublish || capabilities.liveKafkaSse || capabilities.kafkaRefreshReplay;
+  const projectionAuthority = capabilities.transactionalProjector || capabilities.projectionOutboxRelay || capabilities.projectionRealtime;
+  if (!directAuthority || !projectionAuthority) return;
+  throw contractError(
+    "hwlab_workbench_realtime_dual_authority",
+    "Workbench direct/live Kafka authority cannot run beside the transactional projector authority. Disable direct publish, live-only SSE, and Kafka retention replay before enabling the projector chain."
+  );
+}
+
 function startTransactionalHwlabKafkaEventBridge({ config, logger = console, kafkaFactory = defaultKafkaFactory, runtimeStore = null, now = () => new Date().toISOString() } = {}) {
   requireKafkaProjectorStore(runtimeStore, config.capabilities);
 

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

@@ -90,7 +90,7 @@ test("projection outbox emits immutable assistant versions in row order", () =>
   assert.deepEqual(events.map((item) => item.payload.cursor.outboxSeq), [10, 11]);
 });
 
-test("live Kafka SSE transparently fans out one envelope without DB, snapshot, cursor, replay, or SSE id", async () => {
+test.skip("removed live-only product authority transparently fans out one envelope", async () => {
   const sessionId = "ses_live_kafka_sse";
   const traceId = "trc_live_kafka_sse";
   const subscribers = new Set();
@@ -189,7 +189,7 @@ test("live Kafka SSE transparently fans out one envelope without DB, snapshot, c
   }
 });
 
-test("Kafka refresh SSE replays retained user and lifecycle envelopes before one connected handoff without waiting for projector readiness", async () => {
+test.skip("removed Kafka retention replay product authority replays retained envelopes", async () => {
   const sessionId = "ses_kafka_refresh_replay";
   const traceId = "trc_kafka_refresh_replay";
   const records = [
@@ -249,7 +249,7 @@ test("Kafka refresh SSE replays retained user and lifecycle envelopes before one
   }
 });
 
-test("Kafka refresh client abort while live readiness is pending never starts retention or live subscription", async () => {
+test.skip("removed Kafka retention replay product authority handles client abort", async () => {
   const sessionId = "ses_kafka_refresh_ready_abort";
   let releaseLiveReady: (() => void) | null = null;
   const liveReady = new Promise<void>((resolve) => { releaseLiveReady = resolve; });
@@ -297,7 +297,7 @@ test("Kafka refresh client abort while live readiness is pending never starts re
   }
 });
 
-test("trace-only Kafka refresh binds the authorized session and rejects same-trace foreign-session retention", async () => {
+test.skip("removed Kafka retention replay product authority binds trace ownership", async () => {
   const sessionId = "ses_kafka_refresh_trace_owner";
   const foreignSessionId = "ses_kafka_refresh_trace_foreign";
   const traceId = "trc_kafka_refresh_trace_owner";
@@ -329,7 +329,7 @@ test("trace-only Kafka refresh binds the authorized session and rejects same-tra
   }
 });
 
-test("session-scoped Kafka refresh gap emits typed workbench.error and closes without connected or fallback", async () => {
+test.skip("removed Kafka retention replay product authority reports retention gaps", async () => {
   const sessionId = "ses_kafka_refresh_gap";
   const traceId = "trc_kafka_refresh_gap";
   const retained = refreshRecord(0, sessionId, traceId, "user", { userMessageId: "msg_kafka_refresh_gap", messageId: "msg_kafka_refresh_gap", text: "retained input" });
@@ -366,7 +366,7 @@ test("session-scoped Kafka refresh gap emits typed workbench.error and closes wi
   }
 });
 
-test("live Kafka SSE heartbeat keeps the transport open without DB, cursor, snapshot, or replay", async () => {
+test.skip("removed live-only product authority keeps its transport open", async () => {
   const sessionId = "ses_live_kafka_heartbeat";
   let dbCalls = 0;
   const server = createCloudApiServer({
@@ -409,7 +409,7 @@ test("live Kafka SSE heartbeat keeps the transport open without DB, cursor, snap
   }
 });
 
-test("live Kafka SSE rejects foreign or inconsistent ownership scopes before fanout subscription", async () => {
+test.skip("removed live-only product authority rejects foreign ownership", async () => {
   const ownedSession = { id: "ses_live_kafka_owned", ownerUserId: ACTOR.id, lastTraceId: "trc_live_kafka_owned" };
   const foreignSession = { id: "ses_live_kafka_foreign", ownerUserId: "usr_live_kafka_foreign", lastTraceId: "trc_live_kafka_foreign" };
   let subscriptions = 0;
@@ -460,7 +460,7 @@ test("live Kafka SSE rejects foreign or inconsistent ownership scopes before fan
   }
 });
 
-test("live Kafka SSE fails closed when ownership lookup is not configured", async () => {
+test.skip("removed live-only product authority fails closed without ownership lookup", async () => {
   let subscriptions = 0;
   const server = createCloudApiServer({
     accessController: {
@@ -497,7 +497,7 @@ test("live Kafka SSE fails closed when ownership lookup is not configured", asyn
   }
 });
 
-test("live Kafka SSE lets admin subscribe to another owner's consistent session and trace", async () => {
+test.skip("removed live-only product authority permits admin ownership scope", async () => {
   const session = { id: "ses_live_kafka_admin", ownerUserId: "usr_live_kafka_owner", lastTraceId: "trc_live_kafka_admin" };
   const admin = { ...ACTOR, id: "usr_live_kafka_admin", role: "admin" };
   let subscriptions = 0;
@@ -573,7 +573,7 @@ test("workbench realtime initial connection emits current snapshot without repla
   }
 });
 
-test("live and projection realtime remain independently reachable when both capabilities are enabled", async () => {
+test("product realtime uses projection outbox even when obsolete live capability values remain present", async () => {
   const sessionId = "ses_composable_realtime";
   const traceId = "trc_composable_realtime";
   let projectionReads = 0;
@@ -610,7 +610,7 @@ test("live and projection realtime remain independently reachable when both capa
   });
   await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
   try {
-    const events = await getSseEvents(server.address().port, `/v1/workbench/projection-events?sessionId=${sessionId}`, 2);
+    const events = await getSseEvents(server.address().port, `/v1/workbench/events?sessionId=${sessionId}`, 2);
     assert.deepEqual(events.map((event) => event.event), ["workbench.connected", "workbench.message.snapshot"]);
     assert.equal(events[0].data.realtimeSource, "projection-outbox");
     assert.equal(projectionReads, 1);

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

@@ -185,19 +185,19 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
     if (request.method !== "GET") return methodNotAllowed(response, "GET");
     const realtimeCapabilities = workbenchRealtimeCapabilities(options.env ?? process.env);
     const projectionOnly = url.pathname === "/v1/workbench/projection-events";
-    if (!projectionOnly && realtimeCapabilities.liveKafkaSse) {
-      await handleLiveKafkaWorkbenchRealtimeHttp(request, response, url, options);
+    if (projectionOnly) {
+      sendJson(response, 410, workbenchError("workbench_projection_events_route_removed", "Use /v1/workbench/events for the single transactional projection authority."));
       return;
     }
     if (!realtimeCapabilities.projectionRealtime) {
-      sendJson(response, 503, workbenchError("workbench_realtime_disabled", "No Workbench realtime SSE capability is enabled."));
+      sendJson(response, 503, workbenchError("workbench_projection_realtime_disabled", "Workbench product realtime requires the transactional projection outbox authority."));
       return;
     }
     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);
     attachWorkbenchRealtimeOtelContext(request, {
-      route: projectionOnly ? "/v1/workbench/projection-events" : "/v1/workbench/events",
+      route: "/v1/workbench/events",
       sessionId: requestedSessionId,
       traceId: requestedTraceId,
       heartbeatMs,

web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts

@@ -83,22 +83,23 @@ test("Workbench API uses metadata-only session detail and bounded messages paths
   assert.equal(workbenchSessionMessagesPathForTest("ses_metadata", { limit: 9 }), "/v1/workbench/sessions/ses_metadata/messages?limit=9");
 });
 
-test("live SSE transport never sends an afterSeq cursor while projection transport can", () => {
+test("only the projection authority sends the durable afterSeq cursor", () => {
   assert.equal(workbenchEventStreamPath({ realtimeCapabilities: { liveKafkaSse: true, kafkaRefreshReplay: false, projectionRealtime: false }, sessionId: "ses_live", traceId: "trc_live", afterSeq: 42 }), "/v1/workbench/events?sessionId=ses_live&traceId=trc_live");
   assert.equal(workbenchEventStreamPath({ realtimeCapabilities: { liveKafkaSse: false, kafkaRefreshReplay: false, projectionRealtime: true }, sessionId: "ses_projection", traceId: "trc_projection", afterSeq: 42 }), "/v1/workbench/events?sessionId=ses_projection&traceId=trc_projection&afterSeq=42");
-  assert.equal(workbenchEventStreamPath({ realtimeCapabilities: { liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: true }, sessionId: "ses_both", traceId: "trc_both", afterSeq: 42 }), "/v1/workbench/events?sessionId=ses_both&traceId=trc_both");
+  assert.equal(workbenchEventStreamPath({ realtimeCapabilities: { liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: true }, sessionId: "ses_both", traceId: "trc_both", afterSeq: 42 }), "/v1/workbench/events?sessionId=ses_both&traceId=trc_both&afterSeq=42");
   assert.equal(workbenchProjectionEventStreamPath({ sessionId: "ses_both", traceId: "trc_both", afterSeq: 42 }), "/v1/workbench/projection-events?sessionId=ses_both&traceId=trc_both&afterSeq=42");
   assert.equal(workbenchRealtimeTransportEnabled({ liveKafkaSse: false, kafkaRefreshReplay: false, projectionRealtime: false }), false);
-  assert.equal(workbenchRealtimeTransportEnabled({ liveKafkaSse: true, kafkaRefreshReplay: false, projectionRealtime: false }), true);
+  assert.equal(workbenchRealtimeTransportEnabled({ liveKafkaSse: true, kafkaRefreshReplay: false, projectionRealtime: false }), false);
   assert.equal(workbenchRealtimeTransportEnabled({ liveKafkaSse: false, kafkaRefreshReplay: false, projectionRealtime: true }), true);
   assert.equal(workbenchRealtimeTransportEnabled({ liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: true }), true);
 });
 
-test("live Kafka keeps the pre-submit session SSE key when a turn trace becomes active", () => {
+test("obsolete live flags cannot erase the active projection trace scope", () => {
   const capabilities = { liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: false };
   const beforeSubmit = workbenchRealtimeScopeKey("ses_live", workbenchRealtimeTraceIdForCapabilities(capabilities, null, null));
   const afterSubmit = workbenchRealtimeScopeKey("ses_live", workbenchRealtimeTraceIdForCapabilities(capabilities, "trc_current_request", "trc_message"));
-  assert.equal(afterSubmit, beforeSubmit);
+  assert.notEqual(afterSubmit, beforeSubmit);
+  assert.equal(afterSubmit, workbenchRealtimeScopeKey("ses_live", "trc_current_request"));
   assert.equal(workbenchRealtimeTraceIdForCapabilities({ liveKafkaSse: false, kafkaRefreshReplay: false, projectionRealtime: true }, "trc_current_request", "trc_message"), "trc_current_request");
 });
 

web/hwlab-cloud-web/src/api/workbench-events.test.ts

@@ -1,7 +1,7 @@
 import assert from "node:assert/strict";
 import { test } from "bun:test";
 
-import { connectWorkbenchEvents, realtimeCoalesceKey, type WorkbenchRealtimeEvent, type WorkbenchSseIngressFrame } from "./workbench-events";
+import { connectWorkbenchEvents, realtimeCoalesceKey, workbenchEventStreamPath, type WorkbenchRealtimeEvent, type WorkbenchSseIngressFrame } from "./workbench-events";
 import { createCoalescedEventQueue } from "../utils/scheduler/coalesced-event-queue";
 
 test("turn snapshot coalescing keys by trace instead of per sequence", () => {
@@ -18,6 +18,18 @@ test("trace events keep sequence-specific coalescing keys", () => {
   assert.notEqual(realtimeCoalesceKey(first, "workbench.trace.event"), realtimeCoalesceKey(next, "workbench.trace.event"));
 });
 
+test("product EventSource resumes the transactional projection authority with one durable outbox cursor", () => {
+  assert.equal(
+    workbenchEventStreamPath({
+      realtimeCapabilities: { liveKafkaSse: true, kafkaRefreshReplay: false, projectionRealtime: true },
+      sessionId: "ses_projection_authority",
+      traceId: "trc_projection_authority",
+      afterSeq: 42
+    }),
+    "/v1/workbench/events?sessionId=ses_projection_authority&traceId=trc_projection_authority&afterSeq=42"
+  );
+});
+
 test("live Kafka envelopes from one trace retain every assistant, tool, output, and terminal event", () => {
   const delivered: WorkbenchRealtimeEvent[] = [];
   const queue = createCoalescedEventQueue<WorkbenchRealtimeEvent>({

web/hwlab-cloud-web/src/api/workbench-events.ts

@@ -213,7 +213,7 @@ export function workbenchEventStreamPath(options: Pick<WorkbenchEventStreamOptio
   const params = new URLSearchParams();
   appendParam(params, "sessionId", options.sessionId);
   appendParam(params, "traceId", options.traceId);
-  if (!options.realtimeCapabilities.liveKafkaSse && options.realtimeCapabilities.projectionRealtime) appendNumberParam(params, "afterSeq", options.afterSeq);
+  if (options.realtimeCapabilities.projectionRealtime) appendNumberParam(params, "afterSeq", options.afterSeq);
   return `/v1/workbench/events?${params.toString()}`;
 }
 

web/hwlab-cloud-web/src/stores/workbench-kafka-refresh-policy.test.ts

@@ -3,18 +3,18 @@ import { test } from "bun:test";
 
 import { workbenchHistoryAuthorityPolicy } from "./workbench-kafka-refresh-policy";
 
-test("Kafka refresh keeps metadata authorization but disables every HTTP history hydrate", () => {
+test("obsolete Kafka retention flags cannot replace projection replay or durable history", () => {
   assert.deepEqual(workbenchHistoryAuthorityPolicy({ liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: true }), {
-    kafkaRetention: true,
+    kafkaRetention: false,
     sessionMetadataRead: true,
-    sessionMessagesHydrate: false,
-    turnStatusHydrate: false,
-    traceEventsHydrate: false,
-    syncReplay: false
+    sessionMessagesHydrate: true,
+    turnStatusHydrate: true,
+    traceEventsHydrate: true,
+    syncReplay: true
   });
 });
 
-test("independent live-only and projection capabilities preserve their existing history policy", () => {
+test("projection capability alone controls product replay", () => {
   assert.deepEqual(workbenchHistoryAuthorityPolicy({ liveKafkaSse: true, kafkaRefreshReplay: false, projectionRealtime: false }), {
     kafkaRetention: false,
     sessionMetadataRead: true,

web/hwlab-cloud-web/src/stores/workbench-kafka-refresh-policy.ts

@@ -10,13 +10,12 @@ export interface WorkbenchHistoryAuthorityPolicy {
 }
 
 export function workbenchHistoryAuthorityPolicy(capabilities: WorkbenchRealtimeCapabilities): WorkbenchHistoryAuthorityPolicy {
-  const kafkaRetention = capabilities.liveKafkaSse && capabilities.kafkaRefreshReplay;
   return {
-    kafkaRetention,
+    kafkaRetention: false,
     sessionMetadataRead: true,
-    sessionMessagesHydrate: !kafkaRetention,
-    turnStatusHydrate: !kafkaRetention,
-    traceEventsHydrate: !kafkaRetention,
-    syncReplay: !kafkaRetention && capabilities.projectionRealtime
+    sessionMessagesHydrate: true,
+    turnStatusHydrate: true,
+    traceEventsHydrate: true,
+    syncReplay: capabilities.projectionRealtime
   };
 }

web/hwlab-cloud-web/src/stores/workbench.ts

@@ -1193,7 +1193,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   }
 
   async function refreshWorkbenchSyncReplay(sessionId: string | null, traceId: string | null, sinceOutboxSeq: number | null, reason: string): Promise<void> {
-    if (realtimeCapabilities.liveKafkaSse || !realtimeCapabilities.projectionRealtime) return;
+    if (!realtimeCapabilities.projectionRealtime) return;
     let cursor = firstFiniteNumber(sinceOutboxSeq) ?? 0;
     for (;;) {
       const result = await workbenchColadaQueries.fetchSyncReplay({ sessionId, traceId, since: cursor }, { timeoutMs: 8000, activityRef: () => activityRef.value });
@@ -1536,7 +1536,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   }
 
   function publishWorkbenchProjectionSignal(sessionId: string | null | undefined, traceId: string | null | undefined, reason: string): void {
-    if (realtimeCapabilities.liveKafkaSse || !realtimeCapabilities.projectionRealtime) return;
+    if (!realtimeCapabilities.projectionRealtime) return;
     if (typeof window === "undefined") return;
     const id = normalizeWorkbenchSessionId(sessionId);
     if (!id) return;
@@ -1546,7 +1546,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   }
 
   function handleWorkbenchProjectionSignal(value: unknown): void {
-    if (realtimeCapabilities.liveKafkaSse || !realtimeCapabilities.projectionRealtime) return;
+    if (!realtimeCapabilities.projectionRealtime) return;
     const record = recordValue(value);
     if (!record || record.sourceId === workbenchProjectionSignalSourceId) return;
     if (firstNonEmptyString(record.type) !== "session-projection") return;

web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts

@@ -78,10 +78,9 @@ interface WorkbenchTransportCursor {
 }
 
 export function workbenchRealtimeTraceIdForCapabilities(
-  capabilities: WorkbenchRealtimeCapabilities,
+  _capabilities: WorkbenchRealtimeCapabilities,
   ...candidates: Array<string | null | undefined>
 ): string | null {
-  if (capabilities.liveKafkaSse) return null;
   for (const candidate of candidates) {
     const traceId = typeof candidate === "string" ? candidate.trim() : "";
     if (traceId) return traceId;
@@ -90,7 +89,7 @@ export function workbenchRealtimeTraceIdForCapabilities(
 }
 
 export function workbenchRealtimeTransportEnabled(capabilities: WorkbenchRealtimeCapabilities): boolean {
-  return capabilities.liveKafkaSse || capabilities.projectionRealtime;
+  return capabilities.projectionRealtime;
 }
 
 export class WorkbenchStreamTransportRuntime {