fix: enable durable Kafka event stream

config/platform-infra/kafka.yaml

@@ -8,15 +8,16 @@ metadata:
   relatedIssues:
     - 2243
     - 2449
+    - 2462
 
 defaults:
   targetId: NC01
   switch:
     enabled: true
-    mode: shadow-produce-only
+    mode: durable-event-stream
     appIntegrationEnabled: true
-    shadowProduceEnabled: true
-    shadowConsumeEnabled: false
+    produceEnabled: true
+    consumeEnabled: true
 
 operator:
   implementation: strimzi
@@ -178,5 +179,5 @@ management:
           pattern: ^hwlab-v03-cloud-api-sse-([0-9]{13})-[0-9a-f]{8}$
           captureGroup: 1
           unit: unix-ms
-  defaultShadowTopic: hwlab.event.v1
-  shadowProducerId: unidesk-platform-infra-kafka-cli
+  defaultTopic: hwlab.event.v1
+  diagnosticProducerId: unidesk-platform-infra-kafka-cli

scripts/src/platform-infra-kafka.ts

@@ -74,10 +74,10 @@ interface PlatformKafkaConfig {
     targetId: string;
     switch: {
       enabled: boolean;
-      mode: string;
+      mode: "durable-event-stream";
       appIntegrationEnabled: boolean;
-      shadowProduceEnabled: boolean;
-      shadowConsumeEnabled: boolean;
+      produceEnabled: boolean;
+      consumeEnabled: boolean;
     };
   };
   operator: {
@@ -127,8 +127,8 @@ interface PlatformKafkaConfig {
       deleteBatchSize: number;
       policies: KafkaGroupCleanupPolicy[];
     };
-    defaultShadowTopic: string;
-    shadowProducerId: string;
+    defaultTopic: string;
+    diagnosticProducerId: string;
   };
 }
 
@@ -230,7 +230,7 @@ export function kafkaHelp(): Record<string, unknown> {
       "bun scripts/cli.ts platform-infra kafka tail --node NC01 --topic hwlab.event.v1 --limit 5",
       "bun scripts/cli.ts platform-infra kafka produce --node NC01 --topic hwlab.event.v1 --key <trace-or-session>",
     ],
-    boundary: "Kafka runtime belongs to platform-infra. P2 supports shadow produce/query only; Kafka consumer cutover stays disabled.",
+    boundary: "Kafka runtime belongs to platform-infra. Durable producers and fixed-group consumers are enabled; browser delivery remains a database-projected SSE concern.",
   };
 }
 
@@ -262,10 +262,10 @@ function readKafkaConfig(): PlatformKafkaConfig {
       targetId: y.stringField(defaults, "targetId", "defaults"),
       switch: {
         enabled: y.booleanField(switchRecord, "enabled", "defaults.switch"),
-        mode: y.stringField(switchRecord, "mode", "defaults.switch"),
+        mode: y.enumField(switchRecord, "mode", "defaults.switch", ["durable-event-stream"] as const),
         appIntegrationEnabled: y.booleanField(switchRecord, "appIntegrationEnabled", "defaults.switch"),
-        shadowProduceEnabled: y.booleanField(switchRecord, "shadowProduceEnabled", "defaults.switch"),
-        shadowConsumeEnabled: y.booleanField(switchRecord, "shadowConsumeEnabled", "defaults.switch"),
+        produceEnabled: y.booleanField(switchRecord, "produceEnabled", "defaults.switch"),
+        consumeEnabled: y.booleanField(switchRecord, "consumeEnabled", "defaults.switch"),
       },
     },
     operator: {
@@ -363,8 +363,8 @@ function parseManagement(root: Record<string, unknown>): PlatformKafkaConfig["ma
       deleteBatchSize: positiveInteger(cleanup, "deleteBatchSize", "management.consumerGroupCleanup"),
       policies: y.arrayOfRecords(cleanup.policies, "management.consumerGroupCleanup.policies").map(parseGroupCleanupPolicy),
     },
-    defaultShadowTopic: topicNameField(management, "defaultShadowTopic", "management"),
-    shadowProducerId: y.stringField(management, "shadowProducerId", "management"),
+    defaultTopic: topicNameField(management, "defaultTopic", "management"),
+    diagnosticProducerId: y.stringField(management, "diagnosticProducerId", "management"),
   };
 }
 
@@ -406,7 +406,7 @@ function validateConfig(kafka: PlatformKafkaConfig): void {
   if (kafka.cluster.replicas !== 1) throw new Error(`${configLabel}.cluster.replicas must be 1 for the target-scoped Kafka POC`);
   if (!/^[0-9]+Gi$/u.test(kafka.cluster.storage.size)) throw new Error(`${configLabel}.cluster.storage.size must be a Gi quantity`);
   if (!kafka.topics.some((topic) => topic.name === kafka.validation.smokeTopic)) throw new Error(`${configLabel}.validation.smokeTopic must reference one of topics[].name`);
-  if (!kafka.topics.some((topic) => topic.name === kafka.management.defaultShadowTopic)) throw new Error(`${configLabel}.management.defaultShadowTopic must reference one of topics[].name`);
+  if (!kafka.topics.some((topic) => topic.name === kafka.management.defaultTopic)) throw new Error(`${configLabel}.management.defaultTopic must reference one of topics[].name`);
   if (kafka.management.defaultTailLimit > kafka.management.maxTailLimit) throw new Error(`${configLabel}.management.defaultTailLimit must be <= management.maxTailLimit`);
   if (kafka.management.defaultGroupListLimit > kafka.management.maxGroupListLimit) throw new Error(`${configLabel}.management.defaultGroupListLimit must be <= management.maxGroupListLimit`);
   if (kafka.management.consumerGroupCleanup.defaultCandidateLimit > kafka.management.consumerGroupCleanup.maxCandidateLimit) throw new Error(`${configLabel}.management.consumerGroupCleanup.defaultCandidateLimit must be <= maxCandidateLimit`);
@@ -418,7 +418,9 @@ function validateConfig(kafka: PlatformKafkaConfig): void {
     cleanupPolicyIds.add(policy.id);
     cleanupPrefixes.add(policy.prefix);
   }
-  if (kafka.defaults.switch.shadowConsumeEnabled) throw new Error(`${configLabel}.defaults.switch.shadowConsumeEnabled must stay false for shadow-produce-only stage`);
+  if (!kafka.defaults.switch.enabled || !kafka.defaults.switch.appIntegrationEnabled || !kafka.defaults.switch.produceEnabled || !kafka.defaults.switch.consumeEnabled) {
+    throw new Error(`${configLabel}.defaults.switch must enable durable event production and consumption`);
+  }
   const topicNames = new Set<string>();
   for (const topic of kafka.topics) {
     if (topicNames.has(topic.name)) throw new Error(`${configLabel}.topics contains duplicate topic ${topic.name}`);
@@ -644,7 +646,7 @@ async function offsets(config: UniDeskConfig, options: KafkaInspectOptions): Pro
 async function tail(config: UniDeskConfig, options: KafkaInspectOptions): Promise<Record<string, unknown>> {
   const kafka = readKafkaConfig();
   const target = resolveTarget(kafka, options.targetId);
-  const topic = resolveTopic(kafka, options.topic, kafka.management.defaultShadowTopic);
+  const topic = resolveTopic(kafka, options.topic, kafka.management.defaultTopic);
   const limit = boundedLimit(kafka, options.limit);
   const result = await capture(config, target.route, ["sh"], tailScript(kafka, target, topic, limit));
   const parsed = parseJsonOutput(result.stdout);
@@ -662,15 +664,15 @@ async function tail(config: UniDeskConfig, options: KafkaInspectOptions): Promis
 async function produce(config: UniDeskConfig, options: KafkaProduceOptions): Promise<Record<string, unknown>> {
   const kafka = readKafkaConfig();
   const target = resolveTarget(kafka, options.targetId);
-  const topic = resolveTopic(kafka, options.topic, kafka.management.defaultShadowTopic);
-  const payload = buildShadowPayload(kafka, target, options, topic);
+  const topic = resolveTopic(kafka, options.topic, kafka.management.defaultTopic);
+  const payload = buildDiagnosticPayload(kafka, target, options, topic);
   const result = await capture(config, target.route, ["sh"], produceScript(kafka, target, topic, options.key, payload));
   const parsed = parseJsonOutput(result.stdout);
   return {
     ok: result.exitCode === 0 && parsed?.ok === true,
     action: "platform-infra-kafka-produce",
     mutation: true,
-    mode: "shadow-produce-only",
+    mode: "diagnostic-produce",
     target: targetSummary(target),
     config: compactConfigSummary(kafka, target),
     production: parsed ?? null,
@@ -763,6 +765,9 @@ metadata:
   annotations:
     unidesk.ai/spec: ${kafka.metadata.spec}
     unidesk.ai/app-integration-enabled: "${kafka.defaults.switch.appIntegrationEnabled}"
+    unidesk.ai/event-stream-mode: "${kafka.defaults.switch.mode}"
+    unidesk.ai/event-produce-enabled: "${kafka.defaults.switch.produceEnabled}"
+    unidesk.ai/event-consume-enabled: "${kafka.defaults.switch.consumeEnabled}"
 spec:
   kafka:
     version: ${kafka.cluster.kafkaVersion}
@@ -1453,7 +1458,7 @@ PY
 }
 
 function produceScript(kafka: PlatformKafkaConfig, target: KafkaTarget, topic: string, key: string | null, payload: string): string {
-  const keyValue = key ?? `${kafka.management.shadowProducerId}-${Date.now()}`;
+  const keyValue = key ?? `${kafka.management.diagnosticProducerId}-${Date.now()}`;
   const separator = "|";
   if (keyValue.includes(separator) || keyValue.includes("\n")) throw new Error("kafka produce --key must not contain | or newline");
   return `
@@ -1496,9 +1501,9 @@ payload_obj = {
         "bytes": len(payload.encode()),
         "valuesPrinted": False,
     },
-    "producer": "${kafka.management.shadowProducerId}",
-    "mode": "shadow-produce-only",
-    "consume": {"enabled": False, "performed": False},
+    "producer": "${kafka.management.diagnosticProducerId}",
+    "mode": "diagnostic-produce",
+    "consume": {"enabled": True, "performed": False},
     "steps": {"produce": {"exitCode": produce_rc, "stdoutTail": text("produce.out"), "stderrTail": text("produce.err")}},
     "valuesPrinted": False,
 }
@@ -1543,13 +1548,13 @@ function compactConfigSummary(kafka: PlatformKafkaConfig, target: KafkaTarget):
     topicCount: kafka.topics.length,
     clientCount: kafka.clients.length,
     management: {
-      defaultShadowTopic: kafka.management.defaultShadowTopic,
+      defaultTopic: kafka.management.defaultTopic,
       defaultTailLimit: kafka.management.defaultTailLimit,
       maxTailLimit: kafka.management.maxTailLimit,
       defaultGroupListLimit: kafka.management.defaultGroupListLimit,
       maxGroupListLimit: kafka.management.maxGroupListLimit,
       consumerGroupCleanup: kafka.management.consumerGroupCleanup,
-      shadowProducerId: kafka.management.shadowProducerId,
+      diagnosticProducerId: kafka.management.diagnosticProducerId,
     },
     valuesPrinted: false,
   };
@@ -1622,9 +1627,9 @@ function policyChecks(kafka: PlatformKafkaConfig, target: KafkaTarget, manifest:
     { name: "no-public-exposure", ok: !/^\s*type:\s*(NodePort|LoadBalancer)\s*$/mu.test(manifest) && !/^\s*kind:\s*Ingress\s*$/mu.test(manifest), detail: "Kafka POC is ClusterIP/internal only." },
     { name: "single-broker-poc", ok: kafka.cluster.replicas === 1, detail: "The target-scoped POC uses one KRaft broker/controller; production HA remains out of scope." },
     { name: "allow-all-network-policy", ok: manifest.includes("kind: NetworkPolicy") && manifest.includes("name: allow-all") && manifest.includes(`namespace: ${target.namespace}`), detail: `NetworkPolicy/allow-all is rendered in ${target.namespace}.` },
-    { name: "shadow-produce-enabled", ok: kafka.defaults.switch.shadowProduceEnabled === true, detail: "Apps may publish shadow events to Kafka." },
-    { name: "app-integration-shadow-only", ok: kafka.defaults.switch.appIntegrationEnabled === true && kafka.defaults.switch.mode === "shadow-produce-only", detail: "App integration is limited to shadow production in this stage." },
-    { name: "shadow-consume-disabled", ok: kafka.defaults.switch.shadowConsumeEnabled === false, detail: "Kafka consumer cutover remains disabled." },
+    { name: "durable-produce-enabled", ok: kafka.defaults.switch.produceEnabled === true, detail: "Transactional outbox relays may publish durable application events." },
+    { name: "durable-event-stream-mode", ok: kafka.defaults.switch.appIntegrationEnabled === true && kafka.defaults.switch.mode === "durable-event-stream", detail: "Application integration uses the durable event stream contract." },
+    { name: "fixed-group-consume-enabled", ok: kafka.defaults.switch.consumeEnabled === true, detail: "Durable projector consumers are enabled." },
   ];
 }
 
@@ -1686,7 +1691,7 @@ function renderPlan(result: Record<string, unknown>): RenderedCliResult {
     `  status: ${stringValue(next.status)}`,
     `  validate: ${stringValue(next.validate)}`,
     "",
-    "Boundary: Kafka runtime is platform-infra only; P2 allows shadow produce/query and keeps consumer cutover disabled.",
+    "Boundary: Kafka runtime is platform-infra only; durable event production and fixed-group projector consumption are enabled.",
     "Disclosure: Secret values are not printed; client entries show only object/key metadata.",
   ]);
 }
@@ -1960,7 +1965,7 @@ async function parseProduceOptions(args: string[]): Promise<KafkaProduceOptions>
   let topic: string | null = null;
   let key: string | null = null;
   let source = "manual-cli";
-  let eventType = "platform-infra.kafka.shadow-produce.v1";
+  let eventType = "platform-infra.kafka.diagnostic-produce.v1";
   let payload: string | null = null;
   for (let index = 0; index < args.length; index += 1) {
     const arg = args[index];
@@ -2026,7 +2031,7 @@ function boundedGroupCleanupLimit(kafka: PlatformKafkaConfig, limit: number | nu
   return value;
 }
 
-function buildShadowPayload(kafka: PlatformKafkaConfig, target: KafkaTarget, options: KafkaProduceOptions, topic: string): string {
+function buildDiagnosticPayload(kafka: PlatformKafkaConfig, target: KafkaTarget, options: KafkaProduceOptions, topic: string): string {
   let userPayload: unknown = null;
   if (options.payload !== null) {
     const trimmed = options.payload.trim();
@@ -2042,20 +2047,19 @@ function buildShadowPayload(kafka: PlatformKafkaConfig, target: KafkaTarget, opt
     eventId: `evt_${target.id}_${Date.now()}_${Math.random().toString(16).slice(2)}`,
     schemaVersion: 1,
     eventType: options.eventType,
-    producer: kafka.management.shadowProducerId,
+    producer: kafka.management.diagnosticProducerId,
     source: options.source,
     node: target.id,
     namespace: target.namespace,
     topic,
-    mode: "shadow-produce-only",
-    consumeEnabled: false,
+    mode: kafka.defaults.switch.mode,
     appIntegrationEnabled: kafka.defaults.switch.appIntegrationEnabled,
-    shadowProduceEnabled: kafka.defaults.switch.shadowProduceEnabled,
-    shadowConsumeEnabled: kafka.defaults.switch.shadowConsumeEnabled,
+    produceEnabled: kafka.defaults.switch.produceEnabled,
+    consumeEnabled: kafka.defaults.switch.consumeEnabled,
     partitionKey: options.key ?? null,
     occurredAt: new Date().toISOString(),
     payload: userPayload ?? {
-      kind: "manual-shadow-produce",
+      kind: "manual-diagnostic-produce",
       spec: kafka.metadata.spec,
       valuesPrinted: false,
     },