Merge pull request #1954 from pikasTech/fix/hwlab-v03-kafka-replay-authority

fix: 切换 HWLAB v03 到 transactional projection authority

config/hwlab-node-lanes.yaml

@@ -209,19 +209,32 @@ lanes:
           codexStdioSupervisor: repo-owned
           kafkaEventBridge:
             enabled: true
+            authority: transactional-projection
             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
+              directPublish: false
+              liveKafkaSse: false
+              kafkaRefreshReplay: false
+              transactionalProjector: true
+              projectionOutboxRelay: true
+              projectionRealtime: 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
             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

scripts/native/hwlab/runtime-gitops-postprocess.mjs

@@ -20,26 +20,117 @@ emit({ ok: true, runtimePath, ...result });
 
 function postprocessRuntimeGitops() {
   const prometheusOperatorDisabled = overlay?.observability?.prometheusOperator === false;
-  if (!prometheusOperatorDisabled) {
-    return { observabilityPrometheusOperator: overlay?.observability?.prometheusOperator ?? null, observabilityWorkloadsChanged: false, kustomizationChanged: false };
-  }
   const changedFiles = [];
   const deletedFiles = [];
   for (const file of listYamlFiles(runtimeDir)) {
-    const changed = stripPrometheusOperatorResourcesFromFile(file);
+    const changed = prometheusOperatorDisabled ? stripPrometheusOperatorResourcesFromFile(file) : "unchanged";
     if (changed === "deleted") deletedFiles.push(path.relative(repoDir, file));
     if (changed === "changed") changedFiles.push(path.relative(repoDir, file));
   }
-  const kustomizationChanged = pruneMissingKustomizationResources();
+  const kustomizationChanged = prometheusOperatorDisabled ? pruneMissingKustomizationResources() : false;
+  const codeAgentRuntimeChanged = patchCodeAgentRuntimeWorkloads();
   return {
-    observabilityPrometheusOperator: false,
+    observabilityPrometheusOperator: overlay?.observability?.prometheusOperator ?? null,
     observabilityWorkloadsChanged: changedFiles.length > 0 || deletedFiles.length > 0,
     kustomizationChanged,
+    codeAgentRuntimeChanged,
     changedFiles,
     deletedFiles,
   };
 }
 
+function patchCodeAgentRuntimeWorkloads() {
+  const runtime = overlay?.codeAgentRuntime;
+  if (!runtime?.enabled || !runtime.kafkaEventBridge) return false;
+  const kafka = runtime.kafkaEventBridge;
+  let changed = false;
+  for (const file of listYamlFiles(runtimeDir)) {
+    const parsed = parseJsonDocument(readFileSync(file, "utf8"));
+    if (parsed === null) continue;
+    const items = isKubernetesList(parsed) ? parsed.items : [parsed];
+    let fileChanged = false;
+    for (const item of items) {
+      const workloadName = String(item?.metadata?.labels?.["app.kubernetes.io/name"] ?? item?.metadata?.name ?? "");
+      const containers = item?.spec?.template?.spec?.containers;
+      if (!Array.isArray(containers)) continue;
+      for (const container of containers) {
+        if (!Array.isArray(container?.env)) continue;
+        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;
+        }
+      }
+    }
+    if (fileChanged) {
+      writeFileSync(file, `${JSON.stringify(parsed, null, 2)}\n`, "utf8");
+      changed = true;
+    }
+  }
+  return changed;
+}
+
+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 && (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;
+  changed = setEnvValue(container, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", String(kafka.enabled && kafka.features.kafkaRefreshReplay)) || 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 && kafka.features.transactionalProjector, {
+    HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS: kafka.transactionalProjector?.heartbeatIntervalMs,
+  }) || changed;
+  changed = patchOptionalEnvGroup(container, kafka.enabled && kafka.features.projectionOutboxRelay, {
+    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 && kafka.features.projectionRealtime, {
+    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;
+  return changed;
+}
+
+function patchOptionalEnvGroup(container, enabled, values) {
+  let changed = false;
+  for (const [name, value] of Object.entries(values)) {
+    changed = enabled ? setEnvValue(container, name, String(value)) || changed : removeEnvValue(container, name) || changed;
+  }
+  return changed;
+}
+
+function setEnvValue(container, name, value) {
+  const existing = container.env.find((item) => item?.name === name);
+  if (existing?.value === value && existing.valueFrom === undefined) return false;
+  if (existing) {
+    existing.value = value;
+    delete existing.valueFrom;
+  } else {
+    container.env.push({ name, value });
+  }
+  return true;
+}
+
+function removeEnvValue(container, name) {
+  const next = container.env.filter((item) => item?.name !== name);
+  if (next.length === container.env.length) return false;
+  container.env = next;
+  return true;
+}
+
 function stripPrometheusOperatorResourcesFromFile(file) {
   const original = readFileSync(file, "utf8");
   const jsonResult = stripPrometheusOperatorResourcesFromJsonFile(file, original);

scripts/src/hwlab-node-lanes.ts

@@ -545,6 +545,7 @@ export interface HwlabRuntimeKafkaShadowProducerSpec {
 
 export interface HwlabRuntimeKafkaEventBridgeSpec {
   readonly enabled: boolean;
+  readonly authority: "transactional-projection";
   readonly features: {
     readonly directPublish: boolean;
     readonly liveKafkaSse: boolean;
@@ -560,6 +561,28 @@ export interface HwlabRuntimeKafkaEventBridgeSpec {
     readonly matchedEventLimit: number;
     readonly liveBufferLimit: number;
   };
+  readonly transactionalProjector?: {
+    readonly heartbeatIntervalMs: number;
+  };
+  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 configRef: string;
   readonly bootstrapServers: string;
   readonly stdioTopic: string;
@@ -1221,8 +1244,14 @@ 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 enabled = booleanField(raw, "enabled", path);
+  const authority = enumStringField(raw, "authority", path, ["transactional-projection"] as const);
+  const directPublish = booleanField(features, "directPublish", `${path}.features`);
   const liveKafkaSse = booleanField(features, "liveKafkaSse", `${path}.features`);
   const kafkaRefreshReplay = booleanField(features, "kafkaRefreshReplay", `${path}.features`);
+  const transactionalProjector = booleanField(features, "transactionalProjector", `${path}.features`);
+  const projectionOutboxRelay = booleanField(features, "projectionOutboxRelay", `${path}.features`);
+  const projectionRealtime = booleanField(features, "projectionRealtime", `${path}.features`);
   const refreshReplay = raw.refreshReplay === undefined
     ? undefined
     : codeAgentKafkaRefreshReplayConfig(raw.refreshReplay, `${path}.refreshReplay`);
@@ -1232,6 +1261,21 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun
   if (kafkaRefreshReplay && !liveKafkaSse) {
     throw new Error(`${path}.features.liveKafkaSse must be true when ${path}.features.kafkaRefreshReplay is true`);
   }
+  if (!enabled || directPublish || liveKafkaSse || kafkaRefreshReplay || !transactionalProjector || !projectionOutboxRelay || !projectionRealtime) {
+    throw new Error(`${path}.authority transactional-projection requires enabled transactionalProjector/projectionOutboxRelay/projectionRealtime and disables directPublish/liveKafkaSse/kafkaRefreshReplay`);
+  }
+  const transactionalProjectorConfig = raw.transactionalProjector === 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 (transactionalProjector && transactionalProjectorConfig === undefined) throw new Error(`${path}.transactionalProjector is required when ${path}.features.transactionalProjector is true`);
+  if (projectionOutboxRelay && projectionOutboxRelayConfig === undefined) throw new Error(`${path}.projectionOutboxRelay is required when ${path}.features.projectionOutboxRelay is true`);
+  if (projectionRealtime && projectionRealtimeConfig === undefined) throw new Error(`${path}.projectionRealtime is required when ${path}.features.projectionRealtime is true`);
   const directPublishConsumerGroupId = stringField(raw, "directPublishConsumerGroupId", path);
   const transactionalProjectorConsumerGroupId = stringField(raw, "transactionalProjectorConsumerGroupId", path);
   const hwlabEventConsumerGroupId = stringField(raw, "hwlabEventConsumerGroupId", path);
@@ -1239,16 +1283,21 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun
     throw new Error(`${path} consumer group ids must be distinct so enabled capabilities receive independent event streams`);
   }
   return {
-    enabled: booleanField(raw, "enabled", path),
+    enabled,
+    authority,
     features: {
-      directPublish: booleanField(features, "directPublish", `${path}.features`),
+      directPublish,
       liveKafkaSse,
       kafkaRefreshReplay,
-      transactionalProjector: booleanField(features, "transactionalProjector", `${path}.features`),
-      projectionOutboxRelay: booleanField(features, "projectionOutboxRelay", `${path}.features`),
-      projectionRealtime: booleanField(features, "projectionRealtime", `${path}.features`),
+      transactionalProjector,
+      projectionOutboxRelay,
+      projectionRealtime,
     },
     ...(refreshReplay === undefined ? {} : { refreshReplay }),
+    ...(transactionalProjectorConfig === undefined ? {} : { transactionalProjector: transactionalProjectorConfig }),
+    ...(projectionOutboxRelayConfig === undefined ? {} : { projectionOutboxRelay: projectionOutboxRelayConfig }),
+    ...(projectionRealtimeConfig === undefined ? {} : { projectionRealtime: projectionRealtimeConfig }),
+    healthThresholds: codeAgentKafkaHealthThresholdsConfig(raw.healthThresholds, `${path}.healthThresholds`),
     configRef: stringField(raw, "configRef", path),
     bootstrapServers: stringField(raw, "bootstrapServers", path),
     stdioTopic: stringField(raw, "stdioTopic", path),
@@ -1261,6 +1310,42 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun
   };
 }
 
+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"]> {
+  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),
+  };
+}
+
 function codeAgentKafkaRefreshReplayConfig(value: unknown, path: string): NonNullable<HwlabRuntimeKafkaEventBridgeSpec["refreshReplay"]> {
   const raw = asRecord(value, path);
   return {

scripts/src/hwlab-node/render.ts

@@ -753,6 +753,8 @@ export function summarizeNodeRuntimeControlPlaneStatus(
   const pipelineRunDiagnostics = record(status.pipelineRunDiagnostics);
   const argo = record(status.argo);
   const runtime = record(status.runtime);
+  const codeAgentRuntime = record(runtime.codeAgentRuntime);
+  const kafkaAuthority = record(codeAgentRuntime.kafkaAuthority);
   const publicProbes = record(status.publicProbes);
   const gitMirror = record(status.gitMirror);
   const gitMirrorCompact = record(gitMirror.compact);
@@ -822,8 +824,25 @@ export function summarizeNodeRuntimeControlPlaneStatus(
       externalPostgresReady: runtime.externalPostgresBridge === undefined && runtime.externalPostgresSecrets === undefined
         ? null
         : record(runtime.externalPostgresBridge).ready === true && record(runtime.externalPostgresSecrets).ready === true,
-      codeAgentRuntimeReady: record(runtime.codeAgentRuntime).required === true ? record(runtime.codeAgentRuntime).ready === true : null,
-      codeAgentRuntimeReason: typeof record(runtime.codeAgentRuntime).degradedReason === "string" ? record(runtime.codeAgentRuntime).degradedReason : null,
+      codeAgentRuntimeReady: codeAgentRuntime.required === true ? codeAgentRuntime.ready === true : null,
+      codeAgentRuntimeReason: typeof codeAgentRuntime.degradedReason === "string" ? codeAgentRuntime.degradedReason : null,
+      kafkaAuthority: Object.keys(kafkaAuthority).length === 0 ? null : {
+        authority: kafkaAuthority.authority ?? null,
+        fixedGroups: kafkaAuthority.fixedGroups ?? null,
+        capabilities: kafkaAuthority.capabilities ?? null,
+        liveReplay: kafkaAuthority.liveReplay ?? null,
+        healthObserved: kafkaAuthority.healthObserved === true,
+        healthStatus: kafkaAuthority.healthStatus ?? null,
+        consumerLag: kafkaAuthority.consumerLag ?? null,
+        backlogCount: kafkaAuthority.backlogCount ?? null,
+        retryCount: kafkaAuthority.retryCount ?? null,
+        failedInboxCount: kafkaAuthority.failedInboxCount ?? null,
+        dlqCount: kafkaAuthority.dlqCount ?? null,
+        thresholds: kafkaAuthority.thresholds ?? null,
+        warning: kafkaAuthority.warning === true,
+        warningReasons: kafkaAuthority.warningReasons ?? [],
+        blocking: false,
+      },
     },
     publicProbe: {
       ready: publicProbes.ready === true,
@@ -1946,6 +1965,27 @@ export function nodeRuntimePipelinePostprocessScript(): string[] {
     "          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;",
+    "          if (kafka.enabled && kafka.features.transactionalProjector) {",
+    "            codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS', String(kafka.transactionalProjector.heartbeatIntervalMs)) || codeAgentRuntimeChanged;",
+    "          } else {",
+    "            codeAgentRuntimeChanged = removeEnvValues(container, ['HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS']) || codeAgentRuntimeChanged;",
+    "          }",
+    "          if (kafka.enabled && kafka.features.projectionOutboxRelay) {",
+    "            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 && kafka.features.projectionRealtime) {",
+    "            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;",
+    "          }",
     "        }",
     "      }",
     "    }",

scripts/src/hwlab-node/web-probe-observe-actions.test.ts

@@ -3,11 +3,13 @@ import { mkdtemp, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { test } from "bun:test";
+import { readFileSync } from "node:fs";
 
 import { hwlabRuntimeLaneSpecForNode } from "../hwlab-node-lanes";
+import { rootPath } from "../config";
 import { buildWebObserveCommandVisibility, nodeWebProbeRealtimeFanoutProfiles, nodeWebProbeWorkbenchKafkaDebugReplay, nodeWebProbeWorkbenchTraceReadability, resolveWebObserveActionJson } from "./web-probe-observe-actions";
 
-test("realtime fanout runner contract derives topics, groups, and independent capabilities from owning YAML", () => {
+test("realtime authority runner contract derives topics, groups, and transactional capabilities from owning YAML", () => {
   const profiles = nodeWebProbeRealtimeFanoutProfiles(hwlabRuntimeLaneSpecForNode("v03", "NC01"));
   const expectedKafka = profiles["pure-kafka-live"]?.expectedKafka as Record<string, any>;
   assert.deepEqual(expectedKafka.topics, {
@@ -21,12 +23,34 @@ test("realtime fanout runner contract derives topics, groups, and independent ca
     liveSse: "hwlab-v03-workbench-live-sse",
   });
   assert.deepEqual(expectedKafka.capabilities, {
-    directPublish: true,
-    liveKafkaSse: true,
-    kafkaRefreshReplay: true,
-    transactionalProjector: false,
-    projectionOutboxRelay: false,
-    projectionRealtime: false,
+    directPublish: false,
+    liveKafkaSse: false,
+    kafkaRefreshReplay: false,
+    transactionalProjector: true,
+    projectionOutboxRelay: true,
+    projectionRealtime: true,
+  });
+  const kafka = hwlabRuntimeLaneSpecForNode("v03", "NC01").codeAgentRuntime?.kafkaEventBridge;
+  assert.equal(kafka?.authority, "transactional-projection");
+  assert.deepEqual(kafka?.transactionalProjector, { heartbeatIntervalMs: 2_000 });
+  assert.deepEqual(kafka?.projectionOutboxRelay, {
+    intervalMs: 250,
+    batchSize: 100,
+    leaseMs: 30_000,
+    sendTimeoutMs: 10_000,
+    retryBackoffMs: 1_000,
+  });
+  assert.deepEqual(kafka?.projectionRealtime, {
+    outboxTailBatchSize: 100,
+    sseHeartbeatMs: 15_000,
+    sseDrainTimeoutMs: 2_500,
+  });
+  assert.deepEqual(kafka?.healthThresholds, {
+    consumerLagWarning: 0,
+    outboxBacklogWarning: 0,
+    retryCountWarning: 0,
+    failedInboxWarning: 0,
+    dlqWarning: 0,
   });
   assert.deepEqual(profiles["pure-kafka-live"]?.requiredEventTypes, {
     agentrun: ["user_message", "assistant_message", "terminal_status"],
@@ -38,6 +62,16 @@ test("realtime fanout runner contract derives topics, groups, and independent ca
   });
 });
 
+test("realtime authority parser and inline render expose no direct authority or duplicate native postprocess execution", () => {
+  const laneSource = readFileSync(rootPath("scripts", "src", "hwlab-node-lanes.ts"), "utf8");
+  const renderSource = readFileSync(rootPath("scripts", "src", "hwlab-node", "render.ts"), "utf8");
+  assert.doesNotMatch(laneSource, /direct-live/u);
+  assert.equal(renderSource.includes("UNIDESK_RUNTIME_GITOPS_RUNTIME_PATH"), false);
+  assert.equal(renderSource.includes('UNIDESK_RUNTIME_GITOPS_OVERLAY_B64="$overlay_b64" UNIDESK_RUNTIME_GITOPS_RUNTIME_PATH'), false);
+  assert.match(renderSource, /runtimeGitopsPostprocessScript\(\)/u);
+  assert.match(renderSource, /writeRuntimeGitopsNativeScript\("runtime-gitops-postprocess\.mjs"/u);
+});
+
 test("Workbench Kafka debug replay runner contract is derived from owning YAML", () => {
   assert.deepEqual(nodeWebProbeWorkbenchKafkaDebugReplay(hwlabRuntimeLaneSpecForNode("v03", "NC01")), {
     enabled: true,

scripts/src/hwlab-node/web-probe.ts

@@ -1482,6 +1482,7 @@ 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
@@ -1538,12 +1539,70 @@ 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?.authority ?? null, 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: kafka.authority,
+    fixedGroups: {
+      directPublish: kafka.directPublishConsumerGroupId,
+      transactionalProjector: kafka.transactionalProjectorConsumerGroupId,
+      liveSse: kafka.hwlabEventConsumerGroupId,
+    },
+    capabilities: kafka.features,
+    liveReplay: {
+      live: kafka.features.projectionRealtime || kafka.features.liveKafkaSse,
+      replay: kafka.features.projectionRealtime || kafka.features.kafkaRefreshReplay,
+      cursor: kafka.features.projectionRealtime ? "projection-outbox-seq" : kafka.features.kafkaRefreshReplay ? "kafka-refresh-replay" : "none",
+    },
+    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 {
@@ -1666,13 +1725,47 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti
     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));
+    if (kafkaEventBridge.enabled && kafkaEventBridge.features.transactionalProjector) {
+      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",
+    ];
+    if (kafkaEventBridge.enabled && kafkaEventBridge.features.projectionOutboxRelay) {
+      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));
+    } 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 && kafkaEventBridge.features.projectionRealtime) {
+      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);
+    }
   }
   return {
     ready: mismatches.length === 0,
     deploymentExists: true,
     deploymentName: "hwlab-cloud-api",
     containerName: "hwlab-cloud-api",
-    checkedEnvCount: 9 + (kafkaShadowProducer === undefined ? 0 : 5) + (kafkaEventBridge === undefined ? 0 : 21),
+    checkedEnvCount: 9 + (kafkaShadowProducer === undefined ? 0 : 5) + (kafkaEventBridge === undefined ? 0 : 30),
     envCount: envByName.size,
     mismatches,
     result: compactCodeAgentDeploymentProbeResult(envProbe),