diff --git a/config/hwlab-node-lanes.yaml b/config/hwlab-node-lanes.yaml index bb3faaa0..e6b72af8 100644 --- a/config/hwlab-node-lanes.yaml +++ b/config/hwlab-node-lanes.yaml @@ -211,30 +211,26 @@ lanes: codexStdioSupervisor: repo-owned kafkaEventBridge: enabled: true - transactionalProjector: - heartbeatIntervalMs: 2000 - projectionOutboxRelay: - intervalMs: 250 - batchSize: 100 - leaseMs: 30000 - sendTimeoutMs: 10000 - retryBackoffMs: 1000 - projectionRealtime: - outboxTailBatchSize: 100 - sseHeartbeatMs: 15000 - sseDrainTimeoutMs: 2500 - healthThresholds: - consumerLagWarning: 0 - outboxBacklogWarning: 0 - retryCountWarning: 0 - failedInboxWarning: 0 - dlqWarning: 0 + 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 agentRunEventTopic: agentrun.event.v1 hwlabEventTopic: hwlab.event.v1 clientId: hwlab-v03-cloud-api + directPublishConsumerGroupId: hwlab-v03-agentrun-event-direct-publish transactionalProjectorConsumerGroupId: hwlab-v03-agentrun-event-projector hwlabEventConsumerGroupId: hwlab-v03-workbench-live-sse publicExposure: diff --git a/scripts/native/hwlab/runtime-gitops-postprocess.mjs b/scripts/native/hwlab/runtime-gitops-postprocess.mjs index 604035d7..6388989f 100644 --- a/scripts/native/hwlab/runtime-gitops-postprocess.mjs +++ b/scripts/native/hwlab/runtime-gitops-postprocess.mjs @@ -62,6 +62,11 @@ function patchCodeAgentRuntimeWorkloads() { if (workloadName === "hwlab-cloud-api" && container.name === "hwlab-cloud-api") { fileChanged = patchCloudApiKafkaEnv(container, kafka) || fileChanged; } + if (workloadName === "hwlab-cloud-web" && container.name === "hwlab-cloud-web") { + fileChanged = setEnvValue(container, "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", String(kafka.enabled && kafka.features.liveKafkaSse)) || fileChanged; + fileChanged = setEnvValue(container, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", String(kafka.enabled && kafka.features.kafkaRefreshReplay)) || fileChanged; + fileChanged = setEnvValue(container, "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED", String(kafka.enabled && kafka.features.projectionRealtime)) || fileChanged; + } } } } @@ -76,24 +81,35 @@ function patchCodeAgentRuntimeWorkloads() { function patchCloudApiKafkaEnv(container, kafka) { let changed = false; changed = setEnvValue(container, "HWLAB_KAFKA_ENABLED", String(kafka.enabled)) || changed; - changed = setEnvValue(container, "HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED", String(kafka.enabled)) || changed; + changed = setEnvValue(container, "HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED", String(kafka.enabled && (kafka.features.directPublish || kafka.features.transactionalProjector))) || changed; + changed = setEnvValue(container, "HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED", String(kafka.enabled && kafka.features.directPublish)) || changed; + changed = setEnvValue(container, "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", String(kafka.enabled && kafka.features.liveKafkaSse)) || changed; + const refreshReplayEnabled = kafka.enabled && kafka.features.kafkaRefreshReplay; + changed = setEnvValue(container, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", String(refreshReplayEnabled)) || changed; + changed = setEnvValue(container, "HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED", String(kafka.enabled && kafka.features.transactionalProjector)) || changed; + changed = setEnvValue(container, "HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED", String(kafka.enabled && kafka.features.projectionOutboxRelay)) || changed; + changed = setEnvValue(container, "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED", String(kafka.enabled && kafka.features.projectionRealtime)) || changed; + changed = setEnvValue(container, "HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID", String(kafka.directPublishConsumerGroupId)) || changed; changed = setEnvValue(container, "HWLAB_KAFKA_PROJECTOR_GROUP_ID", String(kafka.transactionalProjectorConsumerGroupId)) || changed; changed = setEnvValue(container, "HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID", String(kafka.hwlabEventConsumerGroupId)) || changed; - changed = patchOptionalEnvGroup(container, kafka.enabled, { - HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS: kafka.transactionalProjector?.heartbeatIntervalMs, - }) || changed; - changed = patchOptionalEnvGroup(container, kafka.enabled, { - HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS: kafka.projectionOutboxRelay?.intervalMs, - HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE: kafka.projectionOutboxRelay?.batchSize, - HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS: kafka.projectionOutboxRelay?.leaseMs, - HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS: kafka.projectionOutboxRelay?.sendTimeoutMs, - HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS: kafka.projectionOutboxRelay?.retryBackoffMs, - }) || changed; - changed = patchOptionalEnvGroup(container, kafka.enabled, { - HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE: kafka.projectionRealtime?.outboxTailBatchSize, - HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: kafka.projectionRealtime?.sseHeartbeatMs, - HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS: kafka.projectionRealtime?.sseDrainTimeoutMs, + changed = patchOptionalEnvGroup(container, refreshReplayEnabled, { + HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX: kafka.refreshReplay?.groupIdPrefix, + HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS: kafka.refreshReplay?.timeoutMs, + HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT: kafka.refreshReplay?.scanLimit, + HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT: kafka.refreshReplay?.matchedEventLimit, + HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT: kafka.refreshReplay?.liveBufferLimit, }) || changed; + changed = removeEnvValues(container, [ + "HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS", + "HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS", + "HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE", + "HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS", + "HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS", + "HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS", + "HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE", + "HWLAB_WORKBENCH_SSE_HEARTBEAT_MS", + "HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS", + ]) || changed; return changed; } @@ -105,6 +121,13 @@ function patchOptionalEnvGroup(container, enabled, values) { return changed; } +function removeEnvValues(container, names) { + const next = container.env.filter((item) => !names.includes(item?.name)); + if (next.length === container.env.length) return false; + container.env = next; + return true; +} + function setEnvValue(container, name, value) { const existing = container.env.find((item) => item?.name === name); if (existing?.value === value && existing.valueFrom === undefined) return false; diff --git a/scripts/src/hwlab-node-lanes.ts b/scripts/src/hwlab-node-lanes.ts index 3d3d9be3..4856c92a 100644 --- a/scripts/src/hwlab-node-lanes.ts +++ b/scripts/src/hwlab-node-lanes.ts @@ -559,27 +559,20 @@ export interface HwlabRuntimeKafkaShadowProducerSpec { export interface HwlabRuntimeKafkaEventBridgeSpec { readonly enabled: boolean; - readonly transactionalProjector?: { - readonly heartbeatIntervalMs: number; + readonly features: { + readonly directPublish: boolean; + readonly liveKafkaSse: boolean; + readonly kafkaRefreshReplay: boolean; + readonly transactionalProjector: boolean; + readonly projectionOutboxRelay: boolean; + readonly projectionRealtime: boolean; }; - readonly projectionOutboxRelay?: { - readonly intervalMs: number; - readonly batchSize: number; - readonly leaseMs: number; - readonly sendTimeoutMs: number; - readonly retryBackoffMs: number; - }; - readonly projectionRealtime?: { - readonly outboxTailBatchSize: number; - readonly sseHeartbeatMs: number; - readonly sseDrainTimeoutMs: number; - }; - readonly healthThresholds: { - readonly consumerLagWarning: number; - readonly outboxBacklogWarning: number; - readonly retryCountWarning: number; - readonly failedInboxWarning: number; - readonly dlqWarning: number; + readonly refreshReplay?: { + readonly groupIdPrefix: string; + readonly timeoutMs: number; + readonly scanLimit: number; + readonly matchedEventLimit: number; + readonly liveBufferLimit: number; }; readonly configRef: string; readonly bootstrapServers: string; @@ -587,6 +580,7 @@ export interface HwlabRuntimeKafkaEventBridgeSpec { readonly agentRunEventTopic: string; readonly hwlabEventTopic: string; readonly clientId: string; + readonly directPublishConsumerGroupId: string; readonly transactionalProjectorConsumerGroupId: string; readonly hwlabEventConsumerGroupId: string; } @@ -1245,75 +1239,55 @@ function codeAgentKafkaShadowProducerConfig(value: unknown, path: string): Hwlab function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRuntimeKafkaEventBridgeSpec { const raw = asRecord(value, path); - if (raw.refreshReplay !== undefined) throw new Error(`${path}.refreshReplay was removed with the fixed projection realtime architecture`); - const enabled = booleanField(raw, "enabled", path); - const transactionalProjectorConfig = raw.transactionalProjector === undefined + 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 - : codeAgentKafkaTransactionalProjectorConfig(raw.transactionalProjector, `${path}.transactionalProjector`); - const projectionOutboxRelayConfig = raw.projectionOutboxRelay === undefined - ? undefined - : codeAgentKafkaProjectionOutboxRelayConfig(raw.projectionOutboxRelay, `${path}.projectionOutboxRelay`); - const projectionRealtimeConfig = raw.projectionRealtime === undefined - ? undefined - : codeAgentKafkaProjectionRealtimeConfig(raw.projectionRealtime, `${path}.projectionRealtime`); - if (transactionalProjectorConfig === undefined) throw new Error(`${path}.transactionalProjector is required by the fixed transactional projection architecture`); - if (projectionOutboxRelayConfig === undefined) throw new Error(`${path}.projectionOutboxRelay is required by the fixed transactional projection architecture`); - if (projectionRealtimeConfig === undefined) throw new Error(`${path}.projectionRealtime is required by the fixed transactional projection architecture`); + : 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); - if (transactionalProjectorConsumerGroupId === hwlabEventConsumerGroupId) { - throw new Error(`${path} projector and realtime consumer group ids must be distinct`); + if (new Set([directPublishConsumerGroupId, transactionalProjectorConsumerGroupId, hwlabEventConsumerGroupId]).size !== 3) { + throw new Error(`${path} consumer group ids must be distinct so enabled capabilities receive independent event streams`); } return { - enabled, - transactionalProjector: transactionalProjectorConfig, - projectionOutboxRelay: projectionOutboxRelayConfig, - projectionRealtime: projectionRealtimeConfig, - healthThresholds: codeAgentKafkaHealthThresholdsConfig(raw.healthThresholds, `${path}.healthThresholds`), + enabled: booleanField(raw, "enabled", path), + features: { + directPublish: booleanField(features, "directPublish", `${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), agentRunEventTopic: stringField(raw, "agentRunEventTopic", path), hwlabEventTopic: stringField(raw, "hwlabEventTopic", path), clientId: stringField(raw, "clientId", path), + directPublishConsumerGroupId, transactionalProjectorConsumerGroupId, hwlabEventConsumerGroupId, }; } -function codeAgentKafkaTransactionalProjectorConfig(value: unknown, path: string): NonNullable { - const raw = asRecord(value, path); - return { heartbeatIntervalMs: numberField(raw, "heartbeatIntervalMs", path) }; -} - -function codeAgentKafkaProjectionOutboxRelayConfig(value: unknown, path: string): NonNullable { +function codeAgentKafkaRefreshReplayConfig(value: unknown, path: string): NonNullable { const raw = asRecord(value, path); return { - intervalMs: numberField(raw, "intervalMs", path), - batchSize: numberField(raw, "batchSize", path), - leaseMs: numberField(raw, "leaseMs", path), - sendTimeoutMs: numberField(raw, "sendTimeoutMs", path), - retryBackoffMs: numberField(raw, "retryBackoffMs", path), - }; -} - -function codeAgentKafkaProjectionRealtimeConfig(value: unknown, path: string): NonNullable { - const raw = asRecord(value, path); - return { - outboxTailBatchSize: numberField(raw, "outboxTailBatchSize", path), - sseHeartbeatMs: numberField(raw, "sseHeartbeatMs", path), - sseDrainTimeoutMs: numberField(raw, "sseDrainTimeoutMs", path), - }; -} - -function codeAgentKafkaHealthThresholdsConfig(value: unknown, path: string): HwlabRuntimeKafkaEventBridgeSpec["healthThresholds"] { - const raw = asRecord(value, path); - return { - consumerLagWarning: nonNegativeIntegerField(raw, "consumerLagWarning", path), - outboxBacklogWarning: nonNegativeIntegerField(raw, "outboxBacklogWarning", path), - retryCountWarning: nonNegativeIntegerField(raw, "retryCountWarning", path), - failedInboxWarning: nonNegativeIntegerField(raw, "failedInboxWarning", path), - dlqWarning: nonNegativeIntegerField(raw, "dlqWarning", path), + 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), }; } diff --git a/scripts/src/hwlab-node/render.ts b/scripts/src/hwlab-node/render.ts index bd3eb10a..588f53e0 100644 --- a/scripts/src/hwlab-node/render.ts +++ b/scripts/src/hwlab-node/render.ts @@ -1859,6 +1859,12 @@ export function nodeRuntimePipelinePostprocessScript(featureConfigSchemaStage: s "function cloudWebRuntimeEnvEntries() {", " const exposure = overlay.publicExposure;", " const entries = !exposure || !Array.isArray(exposure.extraProxies) ? [] : exposure.extraProxies.filter((proxy) => proxy && proxy.cloudWebEnvName && proxy.publicBaseUrl).map((proxy) => ({ name: String(proxy.cloudWebEnvName), value: String(proxy.publicBaseUrl) }));", + " 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;", "}", "function startupProbeFrom(probe) {", @@ -1956,35 +1962,31 @@ export function nodeRuntimePipelinePostprocessScript(featureConfigSchemaStage: s " if (codeAgentRuntime.kafkaEventBridge) {", " const kafka = codeAgentRuntime.kafkaEventBridge;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_ENABLED', String(kafka.enabled)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED', String(kafka.enabled)) || codeAgentRuntimeChanged;", + " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED', String(kafka.enabled && (kafka.features.directPublish || kafka.features.transactionalProjector))) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_BOOTSTRAP_SERVERS', String(kafka.bootstrapServers)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_STDIO_TOPIC', String(kafka.stdioTopic)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC', String(kafka.agentRunEventTopic)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_EVENT_TOPIC', String(kafka.hwlabEventTopic)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_CLIENT_ID', String(kafka.clientId)) || codeAgentRuntimeChanged;", + " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID', String(kafka.directPublishConsumerGroupId)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_PROJECTOR_GROUP_ID', String(kafka.transactionalProjectorConsumerGroupId)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID', String(kafka.hwlabEventConsumerGroupId)) || codeAgentRuntimeChanged;", - " if (kafka.enabled) {", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS', String(kafka.transactionalProjector.heartbeatIntervalMs)) || 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_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS']) || codeAgentRuntimeChanged;", - " }", - " if (kafka.enabled) {", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS', String(kafka.projectionOutboxRelay.intervalMs)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE', String(kafka.projectionOutboxRelay.batchSize)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS', String(kafka.projectionOutboxRelay.leaseMs)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS', String(kafka.projectionOutboxRelay.sendTimeoutMs)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS', String(kafka.projectionOutboxRelay.retryBackoffMs)) || codeAgentRuntimeChanged;", - " } else {", - " codeAgentRuntimeChanged = removeEnvValues(container, ['HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS', 'HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE', 'HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS', 'HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS', 'HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS']) || codeAgentRuntimeChanged;", - " }", - " if (kafka.enabled) {", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE', String(kafka.projectionRealtime.outboxTailBatchSize)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_SSE_HEARTBEAT_MS', String(kafka.projectionRealtime.sseHeartbeatMs)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS', String(kafka.projectionRealtime.sseDrainTimeoutMs)) || codeAgentRuntimeChanged;", - " } else {", - " codeAgentRuntimeChanged = removeEnvValues(container, ['HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE', 'HWLAB_WORKBENCH_SSE_HEARTBEAT_MS', 'HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS']) || codeAgentRuntimeChanged;", + " 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;", " }", " }", " }", @@ -2331,6 +2333,12 @@ export function nodeRuntimePipelinePostprocessScript(featureConfigSchemaStage: s "function cloudWebRuntimeEnvEntries() {", " const exposure = overlay.publicExposure;", " const entries = !exposure || !Array.isArray(exposure.extraProxies) ? [] : exposure.extraProxies.filter((proxy) => proxy && proxy.cloudWebEnvName && proxy.publicBaseUrl).map((proxy) => ({ name: String(proxy.cloudWebEnvName), value: String(proxy.publicBaseUrl) }));", + " 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;", "}", "function workloadRef(item, file, container) { return { file, kind: item && item.kind, name: item && item.metadata && item.metadata.name, container: container && container.name }; }", @@ -2404,14 +2412,31 @@ export function nodeRuntimePipelinePostprocessScript(featureConfigSchemaStage: s " if (codeAgentRuntime.kafkaEventBridge) {", " const kafka = codeAgentRuntime.kafkaEventBridge;", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_ENABLED', String(kafka.enabled));", - " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED', String(kafka.enabled));", + " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED', String(kafka.enabled && (kafka.features.directPublish || kafka.features.transactionalProjector)));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_BOOTSTRAP_SERVERS', String(kafka.bootstrapServers));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_STDIO_TOPIC', String(kafka.stdioTopic));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC', String(kafka.agentRunEventTopic));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_EVENT_TOPIC', String(kafka.hwlabEventTopic));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_CLIENT_ID', String(kafka.clientId));", + " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID', String(kafka.directPublishConsumerGroupId));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_PROJECTOR_GROUP_ID', String(kafka.transactionalProjectorConsumerGroupId));", " 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.ts b/scripts/src/hwlab-node/web-probe.ts index 19f2b921..fb9a7346 100644 --- a/scripts/src/hwlab-node/web-probe.ts +++ b/scripts/src/hwlab-node/web-probe.ts @@ -1493,7 +1493,6 @@ export function nodeRuntimeCodeAgentRuntimeStatus(spec: HwlabRuntimeLaneSpec, na const apiKey = secretKeyStatus(spec, runtime.apiKeySecretName, runtime.apiKeySecretKey); const cloudApiEnv = nodeRuntimeCodeAgentCloudApiEnvStatus(spec, runtime); const cloudWebEnv = nodeRuntimeCodeAgentCloudWebEnvStatus(spec, runtime); - const kafkaAuthority = nodeRuntimeKafkaAuthorityStatus(spec, runtime); const ready = runnerNamespace.exitCode === 0 && secretNamespace.exitCode === 0 && managerNamespace.exitCode === 0 @@ -1550,73 +1549,12 @@ export function nodeRuntimeCodeAgentRuntimeStatus(spec: HwlabRuntimeLaneSpec, na apiKey, cloudApiEnv, cloudWebEnv, - kafkaAuthority, repoUrlFrom: runtime.repoUrlFrom, providerIdFrom: runtime.providerIdFrom, valuesPrinted: false, }; } -function nodeRuntimeKafkaAuthorityStatus(spec: HwlabRuntimeLaneSpec, runtime: NonNullable): Record { - const kafka = runtime.kafkaEventBridge; - if (kafka === undefined || kafka.enabled === false) return { required: false, authority: kafka === undefined ? null : "transactional-projection", valuesPrinted: false }; - const healthUrl = `${spec.publicApiUrl.replace(/\/+$/u, "")}/health/ready`; - const result = runCommand(["curl", "-k", "-sS", "--connect-timeout", "5", "--max-time", "12", healthUrl], repoRoot, { timeoutMs: 15_000 }); - let health: Record = {}; - try { - health = result.exitCode === 0 ? record(JSON.parse(result.stdout)) : {}; - } catch { - health = {}; - } - const projector = record(health.kafkaProjector); - const components = Array.isArray(projector.components) ? projector.components.map(record) : []; - const observations = [projector, ...components]; - const consumerLag = observations.map((item) => record(item.consumerLag).total).find((value) => typeof value === "number") ?? null; - const backlogCount = observations.map((item) => item.backlogCount).find((value) => typeof value === "number") ?? null; - const retryCount = observations.map((item) => item.retryCount).find((value) => typeof value === "number") ?? null; - const failedInboxCount = observations.map((item) => item.failedInboxCount).find((value) => typeof value === "number") ?? null; - const dlqCount = observations.map((item) => item.dlqCount).find((value) => typeof value === "number") ?? null; - const thresholds = kafka.healthThresholds; - const warningReasons = [ - typeof consumerLag === "number" && consumerLag > thresholds.consumerLagWarning ? "consumer-lag" : null, - typeof backlogCount === "number" && backlogCount > thresholds.outboxBacklogWarning ? "outbox-backlog" : null, - typeof retryCount === "number" && retryCount > thresholds.retryCountWarning ? "outbox-retry" : null, - typeof failedInboxCount === "number" && failedInboxCount > thresholds.failedInboxWarning ? "failed-inbox" : null, - typeof dlqCount === "number" && dlqCount > thresholds.dlqWarning ? "dlq" : null, - ].filter((value): value is string => value !== null); - return { - required: true, - authority: "transactional-projection", - fixedGroups: { - transactionalProjector: kafka.transactionalProjectorConsumerGroupId, - liveSse: kafka.hwlabEventConsumerGroupId, - }, - capabilities: { - transactionalProjector: true, - projectionOutboxRelay: true, - projectionRealtime: true, - }, - liveReplay: { - live: true, - replay: true, - cursor: "projection-outbox-seq", - }, - healthObserved: Object.keys(projector).length > 0, - healthStatus: projector.status ?? null, - consumerLag, - backlogCount, - retryCount, - failedInboxCount, - dlqCount, - thresholds, - warning: warningReasons.length > 0, - warningReasons, - blocking: false, - result: compactRuntimeCommand(result), - valuesPrinted: false, - }; -} - function nodeRuntimeCodeAgentManagerTarget(managerUrl: string, fallbackNamespace: string): { namespace: string; serviceName: string } { let parsed: URL | null = null; try { @@ -1705,55 +1643,47 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti const kafkaEventBridge = runtime.kafkaEventBridge; if (kafkaEventBridge !== undefined) { expectValue("HWLAB_KAFKA_ENABLED", String(kafkaEventBridge.enabled)); - expectValue("HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED", String(kafkaEventBridge.enabled)); + expectValue("HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED", String(kafkaEventBridge.enabled && (kafkaEventBridge.features.directPublish || kafkaEventBridge.features.transactionalProjector))); expectValue("HWLAB_KAFKA_BOOTSTRAP_SERVERS", kafkaEventBridge.bootstrapServers); expectValue("HWLAB_KAFKA_STDIO_TOPIC", kafkaEventBridge.stdioTopic); expectValue("HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC", kafkaEventBridge.agentRunEventTopic); expectValue("HWLAB_KAFKA_EVENT_TOPIC", kafkaEventBridge.hwlabEventTopic); expectValue("HWLAB_KAFKA_CLIENT_ID", kafkaEventBridge.clientId); + expectValue("HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID", kafkaEventBridge.directPublishConsumerGroupId); expectValue("HWLAB_KAFKA_PROJECTOR_GROUP_ID", kafkaEventBridge.transactionalProjectorConsumerGroupId); expectValue("HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID", kafkaEventBridge.hwlabEventConsumerGroupId); - if (kafkaEventBridge.enabled) { - if (kafkaEventBridge.transactionalProjector === undefined) throw new Error("codeAgentRuntime.kafkaEventBridge.transactionalProjector is required when transactional projector is enabled"); - expectValue("HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS", String(kafkaEventBridge.transactionalProjector.heartbeatIntervalMs)); - } else { - expectAbsent("HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS"); - } - const relayEnvNames = [ - "HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS", - "HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE", - "HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS", - "HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS", - "HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS", + 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 (kafkaEventBridge.enabled) { - const relay = kafkaEventBridge.projectionOutboxRelay; - if (relay === undefined) throw new Error("codeAgentRuntime.kafkaEventBridge.projectionOutboxRelay is required when projection outbox relay is enabled"); - expectValue("HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS", String(relay.intervalMs)); - expectValue("HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE", String(relay.batchSize)); - expectValue("HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS", String(relay.leaseMs)); - expectValue("HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS", String(relay.sendTimeoutMs)); - expectValue("HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS", String(relay.retryBackoffMs)); + 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 relayEnvNames) expectAbsent(name); - } - const projectionRealtimeEnvNames = ["HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE", "HWLAB_WORKBENCH_SSE_HEARTBEAT_MS", "HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS"]; - if (kafkaEventBridge.enabled) { - const realtime = kafkaEventBridge.projectionRealtime; - if (realtime === undefined) throw new Error("codeAgentRuntime.kafkaEventBridge.projectionRealtime is required when projection realtime is enabled"); - expectValue("HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE", String(realtime.outboxTailBatchSize)); - expectValue("HWLAB_WORKBENCH_SSE_HEARTBEAT_MS", String(realtime.sseHeartbeatMs)); - expectValue("HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS", String(realtime.sseDrainTimeoutMs)); - } else { - for (const name of projectionRealtimeEnvNames) expectAbsent(name); + 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)); } return { ready: mismatches.length === 0, deploymentExists: true, deploymentName: "hwlab-cloud-api", containerName: "hwlab-cloud-api", - checkedEnvCount: 9 + (kafkaShadowProducer === undefined ? 0 : 5) + (kafkaEventBridge === undefined ? 0 : 22), + checkedEnvCount: 9 + (kafkaShadowProducer === undefined ? 0 : 5) + (kafkaEventBridge === undefined ? 0 : 21), envCount: envByName.size, mismatches, result: compactCodeAgentDeploymentProbeResult(envProbe), @@ -1762,9 +1692,53 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti } function nodeRuntimeCodeAgentCloudWebEnvStatus(spec: HwlabRuntimeLaneSpec, runtime: NonNullable): Record { - void spec; - void runtime; - return { required: false, ready: true, checkedEnvCount: 0, valuesPrinted: false }; + const kafkaEventBridge = runtime.kafkaEventBridge; + if (kafkaEventBridge === undefined) return { required: false, ready: true, checkedEnvCount: 0, valuesPrinted: false }; + const envProbe = runNodeK3sArgs(spec, [ + "kubectl", + "-n", + spec.runtimeNamespace, + "get", + "deployment", + "hwlab-cloud-web", + "-o", + 'jsonpath={range .spec.template.spec.containers[?(@.name=="hwlab-cloud-web")].env[*]}{.name}{"\\t"}{.value}{"\\t"}{.valueFrom.secretKeyRef.name}{"\\t"}{.valueFrom.secretKeyRef.key}{"\\n"}{end}', + ], 60); + if (envProbe.exitCode !== 0) { + return { + required: true, + ready: false, + deploymentExists: false, + mismatches: [{ name: "hwlab-cloud-web", reason: "deployment-missing" }], + result: compactCodeAgentDeploymentProbeResult(envProbe), + valuesPrinted: false, + }; + } + const envByName = parseCodeAgentEnvProbe(envProbe.stdout); + const mismatches: Array> = []; + const expectValue = (name: string, expected: string): void => { + const item = envByName.get(name); + if (item === undefined) { + mismatches.push({ name, expected, present: false, kind: "value" }); + return; + } + 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, + ready: mismatches.length === 0, + deploymentExists: true, + deploymentName: "hwlab-cloud-web", + containerName: "hwlab-cloud-web", + checkedEnvCount: 3, + envCount: envByName.size, + mismatches, + result: compactCodeAgentDeploymentProbeResult(envProbe), + valuesPrinted: false, + }; } type CodeAgentEnvProbeRow = {