Merge pull request #2098 from pikasTech/fix/2089-kafka-refresh-replay-yaml

fix: 恢复 Workbench Kafka refresh replay YAML

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:

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;

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<HwlabRuntimeKafkaEventBridgeSpec["transactionalProjector"]> {
-  const raw = asRecord(value, path);
-  return { heartbeatIntervalMs: numberField(raw, "heartbeatIntervalMs", path) };
-}
-
-function codeAgentKafkaProjectionOutboxRelayConfig(value: unknown, path: string): NonNullable<HwlabRuntimeKafkaEventBridgeSpec["projectionOutboxRelay"]> {
+function codeAgentKafkaRefreshReplayConfig(value: unknown, path: string): NonNullable<HwlabRuntimeKafkaEventBridgeSpec["refreshReplay"]> {
   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<HwlabRuntimeKafkaEventBridgeSpec["projectionRealtime"]> {
-  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),
   };
 }
 

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));",
     "            }",
     "          }",
     "        }",

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<HwlabRuntimeLaneSpec["codeAgentRuntime"]>): Record<string, unknown> {
-  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<string, unknown> = {};
-  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<HwlabRuntimeLaneSpec["codeAgentRuntime"]>): Record<string, unknown> {
-  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<Record<string, unknown>> = [];
+  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 = {