From c1af68475ed90a41e542719cdc28d291dcda4060 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 13 Jul 2026 20:28:47 +0200 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20=E5=88=87=E6=8D=A2=20HWLAB=20?= =?UTF-8?q?=E6=8A=95=E5=BD=B1=E5=AE=9E=E6=97=B6=E6=9D=83=E5=A8=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/hwlab-node-lanes.yaml | 37 ++++-- .../hwlab/runtime-gitops-postprocess.mjs | 105 ++++++++++++++++-- scripts/src/hwlab-node-lanes.ts | 98 +++++++++++++++- scripts/src/hwlab-node/render.ts | 45 +++++++- .../web-probe-observe-actions.test.ts | 36 ++++-- scripts/src/hwlab-node/web-probe.ts | 95 +++++++++++++++- 6 files changed, 382 insertions(+), 34 deletions(-) diff --git a/config/hwlab-node-lanes.yaml b/config/hwlab-node-lanes.yaml index 69f10dee..0d5603d9 100644 --- a/config/hwlab-node-lanes.yaml +++ b/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 diff --git a/scripts/native/hwlab/runtime-gitops-postprocess.mjs b/scripts/native/hwlab/runtime-gitops-postprocess.mjs index f38ba2ee..24145745 100644 --- a/scripts/native/hwlab/runtime-gitops-postprocess.mjs +++ b/scripts/native/hwlab/runtime-gitops-postprocess.mjs @@ -6,7 +6,7 @@ import path from "node:path"; const repoDir = process.cwd(); const overlay = readOverlay(); -const runtimePath = requiredOverlayString("runtimePath"); +const runtimePath = process.env.UNIDESK_RUNTIME_GITOPS_RUNTIME_PATH || requiredOverlayString("runtimePath"); const runtimeDir = path.resolve(repoDir, runtimePath); const prometheusOperatorKinds = new Set(["ServiceMonitor", "PrometheusRule", "PodMonitor", "Probe"]); @@ -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); diff --git a/scripts/src/hwlab-node-lanes.ts b/scripts/src/hwlab-node-lanes.ts index 4331e231..84fc0e0c 100644 --- a/scripts/src/hwlab-node-lanes.ts +++ b/scripts/src/hwlab-node-lanes.ts @@ -545,6 +545,7 @@ export interface HwlabRuntimeKafkaShadowProducerSpec { export interface HwlabRuntimeKafkaEventBridgeSpec { readonly enabled: boolean; + readonly authority: "transactional-projection" | "direct-live"; 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", "direct-live"] 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,24 @@ 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 (authority === "transactional-projection" && (!enabled || directPublish || liveKafkaSse || kafkaRefreshReplay || !transactionalProjector || !projectionOutboxRelay || !projectionRealtime)) { + throw new Error(`${path}.authority transactional-projection requires enabled transactionalProjector/projectionOutboxRelay/projectionRealtime and disables directPublish/liveKafkaSse/kafkaRefreshReplay`); + } + if (authority === "direct-live" && (!enabled || !directPublish || !liveKafkaSse || transactionalProjector || projectionOutboxRelay || projectionRealtime)) { + throw new Error(`${path}.authority direct-live requires enabled directPublish/liveKafkaSse and disables transactional projection capabilities`); + } + 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 +1286,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 +1313,42 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun }; } +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 { + 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), + }; +} + function codeAgentKafkaRefreshReplayConfig(value: unknown, path: string): NonNullable { const raw = asRecord(value, path); return { diff --git a/scripts/src/hwlab-node/render.ts b/scripts/src/hwlab-node/render.ts index 17c84056..b594e09a 100644 --- a/scripts/src/hwlab-node/render.ts +++ b/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, @@ -1520,6 +1539,7 @@ export function renderNodeRuntimeControlPlaneOnNode(spec: HwlabRuntimeLaneSpec, `--web-endpoint ${shellQuote(spec.publicWebUrl)}`, `--out ${shellQuote(renderDir)}`, ].join(" "), + `UNIDESK_RUNTIME_GITOPS_OVERLAY_B64="$overlay_b64" UNIDESK_RUNTIME_GITOPS_RUNTIME_PATH="$render_dir/${spec.runtimeRenderDir}" node ${shellQuote(rootPath("scripts/native/hwlab/runtime-gitops-postprocess.mjs"))}`, ...nodeRuntimePipelinePostprocessScript(), ].join("\n"); return { result: runNodeHostScriptAsync(spec, script, timeoutSeconds, `${spec.nodeId.toLowerCase()}-${spec.lane}-render`), renderDir, worktreeDir, location: "node-host" }; @@ -1946,6 +1966,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;", + " }", " }", " }", " }", diff --git a/scripts/src/hwlab-node/web-probe-observe-actions.test.ts b/scripts/src/hwlab-node/web-probe-observe-actions.test.ts index 3a699328..d36c4dc0 100644 --- a/scripts/src/hwlab-node/web-probe-observe-actions.test.ts +++ b/scripts/src/hwlab-node/web-probe-observe-actions.test.ts @@ -7,7 +7,7 @@ import { test } from "bun:test"; import { hwlabRuntimeLaneSpecForNode } from "../hwlab-node-lanes"; 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; assert.deepEqual(expectedKafka.topics, { @@ -21,12 +21,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"], diff --git a/scripts/src/hwlab-node/web-probe.ts b/scripts/src/hwlab-node/web-probe.ts index 605bb694..18aa5448 100644 --- a/scripts/src/hwlab-node/web-probe.ts +++ b/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): Record { + 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 = {}; + 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), From afc56ad067eb4602926de6e1a51941f93d8c56b6 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 13 Jul 2026 20:44:40 +0200 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20=E6=94=B6=E6=95=9B=E6=8A=95=E5=BD=B1?= =?UTF-8?q?=E6=9D=83=E5=A8=81=E6=B8=B2=E6=9F=93=E8=81=8C=E8=B4=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/native/hwlab/runtime-gitops-postprocess.mjs | 2 +- scripts/src/hwlab-node-lanes.ts | 9 +++------ scripts/src/hwlab-node/render.ts | 1 - .../src/hwlab-node/web-probe-observe-actions.test.ts | 12 ++++++++++++ 4 files changed, 16 insertions(+), 8 deletions(-) diff --git a/scripts/native/hwlab/runtime-gitops-postprocess.mjs b/scripts/native/hwlab/runtime-gitops-postprocess.mjs index 24145745..5e183572 100644 --- a/scripts/native/hwlab/runtime-gitops-postprocess.mjs +++ b/scripts/native/hwlab/runtime-gitops-postprocess.mjs @@ -6,7 +6,7 @@ import path from "node:path"; const repoDir = process.cwd(); const overlay = readOverlay(); -const runtimePath = process.env.UNIDESK_RUNTIME_GITOPS_RUNTIME_PATH || requiredOverlayString("runtimePath"); +const runtimePath = requiredOverlayString("runtimePath"); const runtimeDir = path.resolve(repoDir, runtimePath); const prometheusOperatorKinds = new Set(["ServiceMonitor", "PrometheusRule", "PodMonitor", "Probe"]); diff --git a/scripts/src/hwlab-node-lanes.ts b/scripts/src/hwlab-node-lanes.ts index 84fc0e0c..2aaa75c0 100644 --- a/scripts/src/hwlab-node-lanes.ts +++ b/scripts/src/hwlab-node-lanes.ts @@ -545,7 +545,7 @@ export interface HwlabRuntimeKafkaShadowProducerSpec { export interface HwlabRuntimeKafkaEventBridgeSpec { readonly enabled: boolean; - readonly authority: "transactional-projection" | "direct-live"; + readonly authority: "transactional-projection"; readonly features: { readonly directPublish: boolean; readonly liveKafkaSse: boolean; @@ -1245,7 +1245,7 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun 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", "direct-live"] as const); + 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`); @@ -1261,12 +1261,9 @@ 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 (authority === "transactional-projection" && (!enabled || directPublish || liveKafkaSse || kafkaRefreshReplay || !transactionalProjector || !projectionOutboxRelay || !projectionRealtime)) { + 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`); } - if (authority === "direct-live" && (!enabled || !directPublish || !liveKafkaSse || transactionalProjector || projectionOutboxRelay || projectionRealtime)) { - throw new Error(`${path}.authority direct-live requires enabled directPublish/liveKafkaSse and disables transactional projection capabilities`); - } const transactionalProjectorConfig = raw.transactionalProjector === undefined ? undefined : codeAgentKafkaTransactionalProjectorConfig(raw.transactionalProjector, `${path}.transactionalProjector`); diff --git a/scripts/src/hwlab-node/render.ts b/scripts/src/hwlab-node/render.ts index b594e09a..7acc7a39 100644 --- a/scripts/src/hwlab-node/render.ts +++ b/scripts/src/hwlab-node/render.ts @@ -1539,7 +1539,6 @@ export function renderNodeRuntimeControlPlaneOnNode(spec: HwlabRuntimeLaneSpec, `--web-endpoint ${shellQuote(spec.publicWebUrl)}`, `--out ${shellQuote(renderDir)}`, ].join(" "), - `UNIDESK_RUNTIME_GITOPS_OVERLAY_B64="$overlay_b64" UNIDESK_RUNTIME_GITOPS_RUNTIME_PATH="$render_dir/${spec.runtimeRenderDir}" node ${shellQuote(rootPath("scripts/native/hwlab/runtime-gitops-postprocess.mjs"))}`, ...nodeRuntimePipelinePostprocessScript(), ].join("\n"); return { result: runNodeHostScriptAsync(spec, script, timeoutSeconds, `${spec.nodeId.toLowerCase()}-${spec.lane}-render`), renderDir, worktreeDir, location: "node-host" }; diff --git a/scripts/src/hwlab-node/web-probe-observe-actions.test.ts b/scripts/src/hwlab-node/web-probe-observe-actions.test.ts index d36c4dc0..39844fb9 100644 --- a/scripts/src/hwlab-node/web-probe-observe-actions.test.ts +++ b/scripts/src/hwlab-node/web-probe-observe-actions.test.ts @@ -3,8 +3,10 @@ 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 authority runner contract derives topics, groups, and transactional capabilities from owning YAML", () => { @@ -60,6 +62,16 @@ test("realtime authority runner contract derives topics, groups, and transaction }); }); +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,