Merge pull request #2495 from pikasTech/fix/l1-workbench-kafka-sse

配置 L1 Workbench 纯 Kafka SSE

config/hwlab-node-lanes.yaml

@@ -131,6 +131,11 @@ lanes:
               runtimeWebUrl: http://hwlab-cloud-web.hwlab-v03.svc.cluster.local:8080
               terminalTimeoutMs: 600000
               pollIntervalMs: 1000
+            kafka:
+              configRef: codeAgentRuntime.kafkaEventBridge
+              bootstrapServers: 10.43.4.158:9092
+              clientId: hwlab-v03-workbench-native-l1
+              hwlabEventConsumerGroupId: hwlab-v03-workbench-native-l1-live-sse
             api:
               bindHost: 0.0.0.0
               probeHost: 127.0.0.1
@@ -150,9 +155,9 @@ lanes:
                   label: 北京时间
                 workbench:
                   realtimeFeatures:
-                    liveKafkaSse: false
+                    liveKafkaSse: true
                     kafkaRefreshReplay: false
-                    projectionRealtime: true
+                    projectionRealtime: false
                   debugCapabilities:
                     isolatedKafka: false
                     rawHwlabEventWindow:

scripts/src/hwlab-native-development.test.ts

@@ -16,10 +16,21 @@ test("L1 Workbench exposure uses YAML public IP and separate bind/probe addresse
   expect(env.AGENTRUN_API_KEY).toMatch(/^hwl_live_/u);
   expect(env.HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE).toBe("gpt.pika");
   expect(env.HWLAB_CODE_AGENT_AGENTRUN_SOURCE_COMMIT).toBe("0123456789abcdef0123456789abcdef01234567");
+  expect(env.HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED).toBe("false");
+  expect(env.HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED).toBe("true");
+  expect(env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED).toBe("false");
+  expect(env.HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED).toBe("false");
+  expect(env.HWLAB_KAFKA_BOOTSTRAP_SERVERS).toBe(native!.kafka.bootstrapServers);
+  expect(env.HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC).toBe(spec.codeAgentRuntime!.kafkaEventBridge!.agentRunEventTopic);
+  expect(env.HWLAB_KAFKA_EVENT_TOPIC).toBe(spec.codeAgentRuntime!.kafkaEventBridge!.hwlabEventTopic);
+  expect(env.HWLAB_KAFKA_CLIENT_ID).toBe(native!.kafka.clientId);
+  expect(env.HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID).toBe(native!.kafka.hwlabEventConsumerGroupId);
+  expect(env.HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID).not.toBe(spec.codeAgentRuntime!.kafkaEventBridge!.hwlabEventConsumerGroupId);
   expect(env.WORKBENCH_API_BIND_HOST).toBe(native!.api.bindHost);
   expect(env.WORKBENCH_API_PROBE_HOST).toBe(native!.api.probeHost);
   expect(env.WORKBENCH_API_PUBLIC_HOST).toBe(native!.publicHost);
   expect(env.WORKBENCH_WEB_PUBLIC_HOST).toBe(native!.publicHost);
   expect(JSON.parse(env.WORKBENCH_WEB_RUNTIME_CONFIG!)).toEqual(native!.web.runtimeConfig);
+  expect(native!.web.runtimeConfig.workbench.realtimeFeatures).toEqual({ liveKafkaSse: true, kafkaRefreshReplay: false, projectionRealtime: false });
   expect(hwlabNativeDevelopmentHelp()).toMatchObject({ ok: true, command: "hwlab nodes native-development workbench" });
 });

scripts/src/hwlab-native-development.ts

@@ -64,6 +64,9 @@ function workbenchEnvironment(spec: HwlabRuntimeLaneSpec, native: HwlabRuntimeNa
 export function hwlabNativeDevelopmentEnvironment(native: HwlabRuntimeNativeDevelopmentSpec["workbench"], spec: HwlabRuntimeLaneSpec, sourceCommit: string): NodeJS.ProcessEnv {
   const codeAgent = spec.codeAgentRuntime;
   if (codeAgent === undefined) throw new Error(`${hwlabRuntimeLaneConfigPath()}#lanes.${spec.lane}.targets.${spec.nodeId}.codeAgentRuntime is required`);
+  const kafkaBridge = codeAgent.kafkaEventBridge;
+  if (kafkaBridge === undefined) throw new Error(`${hwlabRuntimeLaneConfigPath()}#lanes.${spec.lane}.targets.${spec.nodeId}.${native.kafka.configRef} is required`);
+  if (native.kafka.hwlabEventConsumerGroupId === kafkaBridge.hwlabEventConsumerGroupId) throw new Error("native Workbench Kafka consumer group must be distinct from the Cloud API live SSE consumer group");
   if (!/^[0-9a-f]{40}$/iu.test(sourceCommit)) throw new Error("sourceCommit must be the current HWLAB workspace 40-hex git SHA");
   const authorizationSource = readEnvSourceFile({
     root: "/root/.unidesk/.state/secrets",
@@ -96,6 +99,17 @@ export function hwlabNativeDevelopmentEnvironment(native: HwlabRuntimeNativeDeve
     HWLAB_RUNTIME_LANE: spec.lane,
     HWLAB_RUNTIME_ENDPOINT_LOCKED: "1",
     HWLAB_CODE_AGENT_ASSEMBLED_RUNTIME: "1",
+    HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "false",
+    HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true",
+    HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED: "false",
+    HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED: "false",
+    HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED: "false",
+    HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false",
+    HWLAB_KAFKA_BOOTSTRAP_SERVERS: native.kafka.bootstrapServers,
+    HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC: kafkaBridge.agentRunEventTopic,
+    HWLAB_KAFKA_EVENT_TOPIC: kafkaBridge.hwlabEventTopic,
+    HWLAB_KAFKA_CLIENT_ID: native.kafka.clientId,
+    HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID: native.kafka.hwlabEventConsumerGroupId,
     WORKBENCH_NATIVE_SERVICE_STATE_DIR: native.stateDir,
     WORKBENCH_API_BIND_HOST: native.api.bindHost,
     WORKBENCH_API_PROBE_HOST: native.api.probeHost,

scripts/src/hwlab-node-lanes.ts

@@ -689,6 +689,12 @@ export interface HwlabRuntimeNativeDevelopmentSpec {
       readonly terminalTimeoutMs: number;
       readonly pollIntervalMs: number;
     };
+    readonly kafka: {
+      readonly configRef: "codeAgentRuntime.kafkaEventBridge";
+      readonly bootstrapServers: string;
+      readonly clientId: string;
+      readonly hwlabEventConsumerGroupId: string;
+    };
     readonly api: { readonly bindHost: string; readonly probeHost: string; readonly port: number };
     readonly worker: { readonly bindHost: string; readonly probeHost: string; readonly healthPort: number };
     readonly web: {
@@ -1466,6 +1472,7 @@ function nativeDevelopmentConfig(value: unknown, path: string): HwlabRuntimeNati
   const temporal = asRecord(workbench.temporal, `${path}.workbench.temporal`);
   const temporalActivity = asRecord(temporal.activity, `${path}.workbench.temporal.activity`);
   const agentRun = asRecord(workbench.agentRun, `${path}.workbench.agentRun`);
+  const kafka = asRecord(workbench.kafka, `${path}.workbench.kafka`);
   const api = asRecord(workbench.api, `${path}.workbench.api`);
   const worker = asRecord(workbench.worker, `${path}.workbench.worker`);
   const web = asRecord(workbench.web, `${path}.workbench.web`);
@@ -1500,6 +1507,12 @@ function nativeDevelopmentConfig(value: unknown, path: string): HwlabRuntimeNati
         terminalTimeoutMs: numberField(agentRun, "terminalTimeoutMs", `${path}.workbench.agentRun`),
         pollIntervalMs: numberField(agentRun, "pollIntervalMs", `${path}.workbench.agentRun`),
       },
+      kafka: {
+        configRef: enumStringField(kafka, "configRef", `${path}.workbench.kafka`, ["codeAgentRuntime.kafkaEventBridge"]),
+        bootstrapServers: stringField(kafka, "bootstrapServers", `${path}.workbench.kafka`),
+        clientId: stringField(kafka, "clientId", `${path}.workbench.kafka`),
+        hwlabEventConsumerGroupId: stringField(kafka, "hwlabEventConsumerGroupId", `${path}.workbench.kafka`),
+      },
       api: { bindHost: stringField(api, "bindHost", `${path}.workbench.api`), probeHost: stringField(api, "probeHost", `${path}.workbench.api`), port: numberField(api, "port", `${path}.workbench.api`) },
       worker: { bindHost: stringField(worker, "bindHost", `${path}.workbench.worker`), probeHost: stringField(worker, "probeHost", `${path}.workbench.worker`), healthPort: numberField(worker, "healthPort", `${path}.workbench.worker`) },
       web: {