From 499e0d8aa78f62bb83ba3afad78f3f4e11757a61 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 04:30:03 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A3=B0=E6=98=8E=20Kafka=20=E5=88=B7?= =?UTF-8?q?=E6=96=B0=E9=87=8D=E6=94=BE=E8=BF=90=E8=A1=8C=E6=97=B6=E9=85=8D?= =?UTF-8?q?=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/hwlab-node-lanes.yaml | 7 ++++ scripts/src/hwlab-node-lanes.ts | 34 +++++++++++++++++- scripts/src/hwlab-node/render.ts | 35 +++++++++++++++++++ .../web-probe-observe-actions.test.ts | 1 + scripts/src/hwlab-node/web-probe.ts | 29 +++++++++++++-- 5 files changed, 103 insertions(+), 3 deletions(-) diff --git a/config/hwlab-node-lanes.yaml b/config/hwlab-node-lanes.yaml index 629757f8..59499310 100644 --- a/config/hwlab-node-lanes.yaml +++ b/config/hwlab-node-lanes.yaml @@ -203,9 +203,16 @@ lanes: features: directPublish: true liveKafkaSse: true + kafkaRefreshReplay: true transactionalProjector: false projectionOutboxRelay: false projectionRealtime: false + refreshReplay: + groupIdPrefix: hwlab-v03-cloud-api-sse + timeoutMs: 30000 + scanLimit: 1000000 + matchedEventLimit: 2000 + liveBufferLimit: 2000 configRef: config/platform-infra/kafka.yaml#clients.hwlab-v03-cloud-api bootstrapServers: platform-infra-kafka-kafka-bootstrap.platform-infra.svc.cluster.local:9092 stdioTopic: codex-stdio.raw.v1 diff --git a/scripts/src/hwlab-node-lanes.ts b/scripts/src/hwlab-node-lanes.ts index 17d0090e..c28c807b 100644 --- a/scripts/src/hwlab-node-lanes.ts +++ b/scripts/src/hwlab-node-lanes.ts @@ -474,10 +474,18 @@ export interface HwlabRuntimeKafkaEventBridgeSpec { readonly features: { readonly directPublish: boolean; readonly liveKafkaSse: boolean; + readonly kafkaRefreshReplay: boolean; readonly transactionalProjector: boolean; readonly projectionOutboxRelay: boolean; readonly projectionRealtime: boolean; }; + readonly refreshReplay?: { + readonly groupIdPrefix: string; + readonly timeoutMs: number; + readonly scanLimit: number; + readonly matchedEventLimit: number; + readonly liveBufferLimit: number; + }; readonly configRef: string; readonly bootstrapServers: string; readonly stdioTopic: string; @@ -1087,6 +1095,17 @@ function codeAgentKafkaShadowProducerConfig(value: unknown, path: string): Hwlab function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRuntimeKafkaEventBridgeSpec { const raw = asRecord(value, path); const features = asRecord(raw.features, `${path}.features`); + const liveKafkaSse = booleanField(features, "liveKafkaSse", `${path}.features`); + const kafkaRefreshReplay = booleanField(features, "kafkaRefreshReplay", `${path}.features`); + const refreshReplay = raw.refreshReplay === undefined + ? undefined + : codeAgentKafkaRefreshReplayConfig(raw.refreshReplay, `${path}.refreshReplay`); + if (kafkaRefreshReplay && refreshReplay === undefined) { + throw new Error(`${path}.refreshReplay is required when ${path}.features.kafkaRefreshReplay is true`); + } + if (kafkaRefreshReplay && !liveKafkaSse) { + throw new Error(`${path}.features.liveKafkaSse must be true when ${path}.features.kafkaRefreshReplay is true`); + } const directPublishConsumerGroupId = stringField(raw, "directPublishConsumerGroupId", path); const transactionalProjectorConsumerGroupId = stringField(raw, "transactionalProjectorConsumerGroupId", path); const hwlabEventConsumerGroupId = stringField(raw, "hwlabEventConsumerGroupId", path); @@ -1097,11 +1116,13 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun enabled: booleanField(raw, "enabled", path), features: { directPublish: booleanField(features, "directPublish", `${path}.features`), - liveKafkaSse: booleanField(features, "liveKafkaSse", `${path}.features`), + liveKafkaSse, + kafkaRefreshReplay, transactionalProjector: booleanField(features, "transactionalProjector", `${path}.features`), projectionOutboxRelay: booleanField(features, "projectionOutboxRelay", `${path}.features`), projectionRealtime: booleanField(features, "projectionRealtime", `${path}.features`), }, + ...(refreshReplay === undefined ? {} : { refreshReplay }), configRef: stringField(raw, "configRef", path), bootstrapServers: stringField(raw, "bootstrapServers", path), stdioTopic: stringField(raw, "stdioTopic", path), @@ -1114,6 +1135,17 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun }; } +function codeAgentKafkaRefreshReplayConfig(value: unknown, path: string): NonNullable { + const raw = asRecord(value, path); + return { + groupIdPrefix: stringField(raw, "groupIdPrefix", path), + timeoutMs: numberField(raw, "timeoutMs", path), + scanLimit: numberField(raw, "scanLimit", path), + matchedEventLimit: numberField(raw, "matchedEventLimit", path), + liveBufferLimit: numberField(raw, "liveBufferLimit", path), + }; +} + function commandNameField(value: string, path: string): string { if (!/^[A-Za-z0-9._+-]+$/u.test(value)) throw new Error(`${path} must be a simple command name`); return value; diff --git a/scripts/src/hwlab-node/render.ts b/scripts/src/hwlab-node/render.ts index 3d0ac356..2a49b5f8 100644 --- a/scripts/src/hwlab-node/render.ts +++ b/scripts/src/hwlab-node/render.ts @@ -1810,6 +1810,13 @@ export function nodeRuntimePipelinePostprocessScript(): string[] { " delete item.valueFrom;", " return changed;", "}", + "function removeEnvValues(container, names) {", + " if (!isObject(container) || !Array.isArray(container.env)) return false;", + " const next = container.env.filter((env) => !(env && names.includes(env.name)));", + " if (next.length === container.env.length) return false;", + " container.env = next;", + " return true;", + "}", "function isEnvReuseContainer(container) { return envValue(container, 'HWLAB_RUNTIME_MODE') === 'env-reuse-git-mirror-checkout' || envValue(container, 'HWLAB_BOOT_SH') !== undefined || envValue(container, 'HWLAB_BOOT_COMMIT') !== undefined; }", "function workloadName(item) { return item && item.metadata && item.metadata.labels && item.metadata.labels['app.kubernetes.io/name'] ? String(item.metadata.labels['app.kubernetes.io/name']) : String(item && item.metadata && item.metadata.name || ''); }", "function expectedPublicEndpoint(item) { return workloadName(item) === 'hwlab-cloud-web' ? overlay.publicWebUrl : overlay.publicApiUrl; }", @@ -1819,6 +1826,7 @@ export function nodeRuntimePipelinePostprocessScript(): string[] { " const kafka = overlay.codeAgentRuntime && overlay.codeAgentRuntime.kafkaEventBridge;", " if (kafka) {", " entries.push({ name: 'HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED', value: String(kafka.enabled && kafka.features.liveKafkaSse) });", + " entries.push({ name: 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED', value: String(kafka.enabled && kafka.features.kafkaRefreshReplay) });", " entries.push({ name: 'HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED', value: String(kafka.enabled && kafka.features.projectionRealtime) });", " }", " return entries;", @@ -1929,6 +1937,17 @@ export function nodeRuntimePipelinePostprocessScript(): string[] { " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID', String(kafka.hwlabEventConsumerGroupId)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED', String(kafka.enabled && kafka.features.directPublish)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED', String(kafka.enabled && kafka.features.liveKafkaSse)) || codeAgentRuntimeChanged;", + " const refreshReplayEnabled = kafka.enabled && kafka.features.kafkaRefreshReplay;", + " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED', String(refreshReplayEnabled)) || codeAgentRuntimeChanged;", + " if (refreshReplayEnabled) {", + " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX', String(kafka.refreshReplay.groupIdPrefix)) || codeAgentRuntimeChanged;", + " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS', String(kafka.refreshReplay.timeoutMs)) || codeAgentRuntimeChanged;", + " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT', String(kafka.refreshReplay.scanLimit)) || codeAgentRuntimeChanged;", + " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT', String(kafka.refreshReplay.matchedEventLimit)) || codeAgentRuntimeChanged;", + " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT', String(kafka.refreshReplay.liveBufferLimit)) || codeAgentRuntimeChanged;", + " } else {", + " codeAgentRuntimeChanged = removeEnvValues(container, ['HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT']) || codeAgentRuntimeChanged;", + " }", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED', String(kafka.enabled && kafka.features.transactionalProjector)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED', String(kafka.enabled && kafka.features.projectionOutboxRelay)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED', String(kafka.enabled && kafka.features.projectionRealtime)) || codeAgentRuntimeChanged;", @@ -2259,6 +2278,7 @@ export function nodeRuntimePipelinePostprocessScript(): string[] { " const kafka = overlay.codeAgentRuntime && overlay.codeAgentRuntime.kafkaEventBridge;", " if (kafka) {", " entries.push({ name: 'HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED', value: String(kafka.enabled && kafka.features.liveKafkaSse) });", + " entries.push({ name: 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED', value: String(kafka.enabled && kafka.features.kafkaRefreshReplay) });", " entries.push({ name: 'HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED', value: String(kafka.enabled && kafka.features.projectionRealtime) });", " }", " return entries;", @@ -2285,6 +2305,10 @@ export function nodeRuntimePipelinePostprocessScript(): string[] { " const value = envValue(container, envName);", " if (value !== expected) wrongCodeAgentRuntimeEnvs.push({ ...workloadRef(item, file, container), envName, kind: 'value', expected, value: value ?? null });", " }", + " function checkCodeAgentRuntimeAbsent(item, file, container, envName) {", + " const value = envValue(container, envName);", + " if (value !== undefined) wrongCodeAgentRuntimeEnvs.push({ ...workloadRef(item, file, container), envName, kind: 'unexpected-value', expected: null, value });", + " }", " function checkCodeAgentRuntimeSecret(item, file, container, envName, expectedSecret, expectedKey) {", " const secretRef = envSecretRef(container, envName);", " if (secretRef.name !== expectedSecret || secretRef.key !== expectedKey) wrongCodeAgentRuntimeEnvs.push({ ...workloadRef(item, file, container), envName, kind: 'secretKeyRef', expectedSecret, expectedKey, secret: secretRef.name ?? null, key: secretRef.key ?? null });", @@ -2341,6 +2365,17 @@ export function nodeRuntimePipelinePostprocessScript(): string[] { " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID', String(kafka.hwlabEventConsumerGroupId));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED', String(kafka.enabled && kafka.features.directPublish));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED', String(kafka.enabled && kafka.features.liveKafkaSse));", + " const refreshReplayEnabled = kafka.enabled && kafka.features.kafkaRefreshReplay;", + " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED', String(refreshReplayEnabled));", + " if (refreshReplayEnabled) {", + " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX', String(kafka.refreshReplay.groupIdPrefix));", + " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS', String(kafka.refreshReplay.timeoutMs));", + " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT', String(kafka.refreshReplay.scanLimit));", + " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT', String(kafka.refreshReplay.matchedEventLimit));", + " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT', String(kafka.refreshReplay.liveBufferLimit));", + " } else {", + " for (const envName of ['HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT']) checkCodeAgentRuntimeAbsent(item, file, container, envName);", + " }", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED', String(kafka.enabled && kafka.features.transactionalProjector));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED', String(kafka.enabled && kafka.features.projectionOutboxRelay));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED', String(kafka.enabled && kafka.features.projectionRealtime));", diff --git a/scripts/src/hwlab-node/web-probe-observe-actions.test.ts b/scripts/src/hwlab-node/web-probe-observe-actions.test.ts index 2ce8ad31..466b9cf1 100644 --- a/scripts/src/hwlab-node/web-probe-observe-actions.test.ts +++ b/scripts/src/hwlab-node/web-probe-observe-actions.test.ts @@ -23,6 +23,7 @@ test("realtime fanout runner contract derives topics, groups, and independent ca assert.deepEqual(expectedKafka.capabilities, { directPublish: true, liveKafkaSse: true, + kafkaRefreshReplay: true, transactionalProjector: false, projectionOutboxRelay: false, projectionRealtime: false, diff --git a/scripts/src/hwlab-node/web-probe.ts b/scripts/src/hwlab-node/web-probe.ts index d44b5197..bd12830d 100644 --- a/scripts/src/hwlab-node/web-probe.ts +++ b/scripts/src/hwlab-node/web-probe.ts @@ -1596,6 +1596,10 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti } if (item.value !== expected) mismatches.push({ name, expected, value: item.value, present: true, kind: "value" }); }; + const expectAbsent = (name: string): void => { + const item = envByName.get(name); + if (item !== undefined) mismatches.push({ name, expected: null, value: item.value, present: true, kind: "unexpected-value" }); + }; const expectSecret = (name: string, expectedSecret: string, expectedKey: string): void => { const item = envByName.get(name); if (item === undefined) { @@ -1637,6 +1641,26 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti expectValue("HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID", kafkaEventBridge.hwlabEventConsumerGroupId); expectValue("HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.directPublish)); expectValue("HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.liveKafkaSse)); + const refreshReplayEnabled = kafkaEventBridge.enabled && kafkaEventBridge.features.kafkaRefreshReplay; + expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", String(refreshReplayEnabled)); + const refreshReplayEnvNames = [ + "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX", + "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS", + "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT", + "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT", + "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT", + ]; + if (refreshReplayEnabled) { + const refreshReplay = kafkaEventBridge.refreshReplay; + if (refreshReplay === undefined) throw new Error("codeAgentRuntime.kafkaEventBridge.refreshReplay is required when Kafka refresh replay is enabled"); + expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX", refreshReplay.groupIdPrefix); + expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS", String(refreshReplay.timeoutMs)); + expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT", String(refreshReplay.scanLimit)); + expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT", String(refreshReplay.matchedEventLimit)); + expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT", String(refreshReplay.liveBufferLimit)); + } else { + for (const name of refreshReplayEnvNames) expectAbsent(name); + } expectValue("HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.transactionalProjector)); expectValue("HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.projectionOutboxRelay)); expectValue("HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.projectionRealtime)); @@ -1646,7 +1670,7 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti deploymentExists: true, deploymentName: "hwlab-cloud-api", containerName: "hwlab-cloud-api", - checkedEnvCount: 9 + (kafkaShadowProducer === undefined ? 0 : 5) + (kafkaEventBridge === undefined ? 0 : 14), + checkedEnvCount: 9 + (kafkaShadowProducer === undefined ? 0 : 5) + (kafkaEventBridge === undefined ? 0 : 21), envCount: envByName.size, mismatches, result: compactCodeAgentDeploymentProbeResult(envProbe), @@ -1688,6 +1712,7 @@ function nodeRuntimeCodeAgentCloudWebEnvStatus(spec: HwlabRuntimeLaneSpec, runti if (item.value !== expected) mismatches.push({ name, expected, value: item.value, present: true, kind: "value" }); }; expectValue("HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.liveKafkaSse)); + expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.kafkaRefreshReplay)); expectValue("HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.projectionRealtime)); return { required: true, @@ -1695,7 +1720,7 @@ function nodeRuntimeCodeAgentCloudWebEnvStatus(spec: HwlabRuntimeLaneSpec, runti deploymentExists: true, deploymentName: "hwlab-cloud-web", containerName: "hwlab-cloud-web", - checkedEnvCount: 2, + checkedEnvCount: 3, envCount: envByName.size, mismatches, result: compactCodeAgentDeploymentProbeResult(envProbe),