diff --git a/config/hwlab-node-lanes.yaml b/config/hwlab-node-lanes.yaml index 942e0c22..eafbefbf 100644 --- a/config/hwlab-node-lanes.yaml +++ b/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: diff --git a/scripts/src/hwlab-native-development.test.ts b/scripts/src/hwlab-native-development.test.ts index 2f1cfe69..5018b994 100644 --- a/scripts/src/hwlab-native-development.test.ts +++ b/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" }); }); diff --git a/scripts/src/hwlab-native-development.ts b/scripts/src/hwlab-native-development.ts index 0e7d0820..5603a2fd 100644 --- a/scripts/src/hwlab-native-development.ts +++ b/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, diff --git a/scripts/src/hwlab-node-lanes.ts b/scripts/src/hwlab-node-lanes.ts index 10403eec..04a8d122 100644 --- a/scripts/src/hwlab-node-lanes.ts +++ b/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: {