feat: shadow produce code agent kafka events

internal/cloud/kafka-shadow-producer.ts

@@ -0,0 +1,185 @@
+// SPEC: PJ2026-010205 HWLAB接入 draft-2026-06-25-p0-session-warm-runner-contract.
+// Responsibility: best-effort Kafka shadow production for HWLAB -> AgentRun command admission.
+
+import { createHash } from "node:crypto";
+
+import { Kafka, logLevel } from "kafkajs";
+
+const TRUE_VALUES = new Set(["1", "true", "yes", "on"]);
+const warnedCodes = new Set();
+let producerState = null;
+
+export function publishHwlabAgentRunCommandShadow({ params = {}, payload = {}, traceId, lifecycle = {}, env = process.env } = {}) {
+  const config = kafkaShadowConfig(env);
+  if (!config) return;
+  const message = buildHwlabCommandShadowMessage({ params, payload, traceId, lifecycle, config });
+  void producerForConfig(config)
+    .then((producer) => producer.send({ topic: config.topic, messages: [message] }))
+    .catch((error) => warnShadowProducer("hwlab-kafka-shadow-produce-failed", {
+      message: error instanceof Error ? error.message : String(error),
+      topic: config.topic,
+      clientId: config.clientId
+    }));
+}
+
+function kafkaShadowConfig(env) {
+  if (!truthy(env.HWLAB_KAFKA_SHADOW_PRODUCE_ENABLED)) return null;
+  if (truthy(env.HWLAB_KAFKA_SHADOW_CONSUME_ENABLED)) {
+    warnOnce("hwlab-kafka-shadow-consume-ignored", {
+      message: "HWLAB Kafka consumer cutover is disabled in this stage; producer continues in shadow mode.",
+      valuesPrinted: false
+    });
+  }
+  const brokers = csv(env.HWLAB_KAFKA_BOOTSTRAP_SERVERS);
+  const topic = stringValue(env.HWLAB_KAFKA_COMMAND_TOPIC);
+  const clientId = stringValue(env.HWLAB_KAFKA_CLIENT_ID);
+  const missing = [];
+  if (brokers.length === 0) missing.push("HWLAB_KAFKA_BOOTSTRAP_SERVERS");
+  if (!topic) missing.push("HWLAB_KAFKA_COMMAND_TOPIC");
+  if (!clientId) missing.push("HWLAB_KAFKA_CLIENT_ID");
+  if (missing.length > 0) {
+    warnOnce("hwlab-kafka-shadow-config-missing", { missing, valuesPrinted: false });
+    return null;
+  }
+  return { brokers, topic, clientId, key: `${clientId}|${topic}|${brokers.join(",")}` };
+}
+
+function buildHwlabCommandShadowMessage({ params, payload, traceId, lifecycle, config }) {
+  const agentRun = objectValue(payload.agentRun);
+  const prompt = promptEvidence(params);
+  const event = {
+    schema: "hwlab.agentrun.command.shadow.v1",
+    eventType: "hwlab.command.admitted",
+    source: config.clientId,
+    producedAt: new Date().toISOString(),
+    mode: "shadow-produce-only",
+    traceId: safeText(traceId ?? payload.traceId ?? params.traceId),
+    hwlab: {
+      sessionId: safeText(params.sessionId ?? payload.sessionId ?? agentRun.hwlabSessionId),
+      conversationId: safeText(params.conversationId ?? payload.conversationId ?? agentRun.conversationId),
+      turnId: safeText(lifecycle.turnId ?? payload.turnId),
+      userMessageId: safeText(lifecycle.userMessageId ?? payload.userMessageId),
+      assistantMessageId: safeText(lifecycle.assistantMessageId ?? payload.assistantMessageId),
+      projectId: safeText(params.projectId ?? payload.projectId),
+      providerProfile: safeText(params.providerProfile ?? agentRun.backendProfile)
+    },
+    agentRun: {
+      adapter: safeText(agentRun.adapter),
+      runId: safeText(agentRun.runId),
+      commandId: safeText(agentRun.commandId),
+      attemptId: safeText(agentRun.attemptId),
+      runnerId: safeText(agentRun.runnerId),
+      jobName: safeText(agentRun.jobName),
+      namespace: safeText(agentRun.namespace),
+      backendProfile: safeText(agentRun.backendProfile),
+      status: safeText(agentRun.status),
+      commandState: safeText(agentRun.commandState),
+      terminalStatus: safeText(agentRun.terminalStatus)
+    },
+    prompt,
+    diagnostics: {
+      payloadSha256: jsonSha256({ traceId: traceId ?? null, agentRun: eventKeyParts(agentRun), prompt }),
+      payloadBytes: jsonBytes({ traceId: traceId ?? null, agentRun: eventKeyParts(agentRun), prompt }),
+      valuesPrinted: false,
+      shadowConsumeEnabled: false
+    },
+    valuesPrinted: false
+  };
+  return {
+    key: safeText(agentRun.commandId ?? traceId ?? params.sessionId) || config.clientId,
+    value: JSON.stringify(event),
+    headers: {
+      "x-shadow-mode": "produce-only",
+      "x-values-printed": "false",
+      "x-trace-id": event.traceId ?? ""
+    }
+  };
+}
+
+async function producerForConfig(config) {
+  if (producerState?.key === config.key && producerState.producer) return producerState.producer;
+  if (producerState?.key === config.key && producerState.connecting) return producerState.connecting;
+  const kafka = new Kafka({ clientId: config.clientId, brokers: config.brokers, logLevel: logLevel.NOTHING });
+  const producer = kafka.producer({ allowAutoTopicCreation: false });
+  const state = {
+    key: config.key,
+    producer,
+    connecting: producer.connect().then(() => {
+      state.connecting = null;
+      return producer;
+    }).catch((error) => {
+      if (producerState === state) producerState = null;
+      throw error;
+    })
+  };
+  producerState = state;
+  return state.connecting;
+}
+
+function promptEvidence(params) {
+  const text = firstString(params.prompt, params.message, params.input, objectValue(params.userMessage).text, objectValue(params.userMessage).content);
+  if (!text) return { present: false, bytes: 0, sha256: null, valuesPrinted: false };
+  return {
+    present: true,
+    bytes: Buffer.byteLength(text, "utf8"),
+    sha256: createHash("sha256").update(text).digest("hex"),
+    valuesPrinted: false
+  };
+}
+
+function eventKeyParts(agentRun) {
+  return {
+    runId: safeText(agentRun.runId),
+    commandId: safeText(agentRun.commandId),
+    attemptId: safeText(agentRun.attemptId),
+    backendProfile: safeText(agentRun.backendProfile)
+  };
+}
+
+function truthy(value) {
+  return TRUE_VALUES.has(String(value ?? "").trim().toLowerCase());
+}
+
+function csv(value) {
+  return String(value ?? "").split(",").map((entry) => entry.trim()).filter(Boolean);
+}
+
+function stringValue(value) {
+  const text = String(value ?? "").trim();
+  return text.length > 0 ? text : null;
+}
+
+function safeText(value, max = 240) {
+  const text = stringValue(value);
+  if (!text) return null;
+  return text.length > max ? `${text.slice(0, max)}...` : text;
+}
+
+function firstString(...values) {
+  for (const value of values) {
+    if (typeof value === "string" && value.trim().length > 0) return value;
+  }
+  return null;
+}
+
+function objectValue(value) {
+  return value && typeof value === "object" && !Array.isArray(value) ? value : {};
+}
+
+function jsonSha256(value) {
+  return createHash("sha256").update(JSON.stringify(value)).digest("hex");
+}
+
+function jsonBytes(value) {
+  return Buffer.byteLength(JSON.stringify(value), "utf8");
+}
+
+function warnOnce(code, details) {
+  if (warnedCodes.has(code)) return;
+  warnedCodes.add(code);
+  warnShadowProducer(code, details);
+}
+
+function warnShadowProducer(code, details = {}) {
+  console.warn(JSON.stringify({ code, component: "hwlab-kafka-shadow-producer", ...details, valuesPrinted: false }));
+}

internal/cloud/server-code-agent-http.ts

@@ -21,6 +21,7 @@ import { scheduleWorkbenchProjectionFinalizer } from "./workbench-projection-fin
 import { writeWorkbenchProjectionSession, writeWorkbenchSessionAdmissionFact } from "./workbench-projection-writer.ts";
 import { buildAgentRunProjectionEventsFetchPlan } from "./workbench-projection-cursor.ts";
 import { createWorkbenchReadModel } from "./workbench-read-model.ts";
+import { publishHwlabAgentRunCommandShadow } from "./kafka-shadow-producer.ts";
 import { codeAgentOtelTraceFields, emitCodeAgentOtelSpan } from "./otel-trace.ts";
 import {
   firstHeaderValue,
@@ -946,6 +947,13 @@ async function submitCodeAgentChatTurn({ params, options, traceId }) {
         });
         if (isCodeAgentResultCanceled(results.get(traceId))) return;
         const owned = annotateOwner(withCodeAgentBillingReservation(payload, dispatchParams), dispatchParams);
+        publishHwlabAgentRunCommandShadow({
+          params: dispatchParams,
+          payload: owned,
+          traceId,
+          lifecycle: codeAgentTurnLifecycleFields(traceId, dispatchParams),
+          env: executionOptions.env ?? process.env
+        });
         await recordCodeAgentSessionOwner({ payload: owned, params: dispatchParams, options: executionOptions, status: "running" });
         results.set(traceId, owned);
         scheduleAgentRunProjectionSync({ traceId, params: dispatchParams, options: executionOptions, traceStore, currentResult: owned });

package-lock.json

@@ -10,6 +10,7 @@
       "dependencies": {
         "@openai/codex": "^0.128.0",
         "fzstd": "0.1.1",
+        "kafkajs": "^2.2.4",
         "pg": "^8.21.0",
         "playwright": "1.59.1",
         "yaml": "^2.8.3"
@@ -159,6 +160,15 @@
       "resolved": "https://registry.npmjs.org/fzstd/-/fzstd-0.1.1.tgz",
       "integrity": "sha512-dkuVSOKKwh3eas5VkJy1AW1vFpet8TA/fGmVA5krThl8YcOVE/8ZIoEA1+U1vEn5ckxxhLirSdY837azmbaNHA=="
     },
+    "node_modules/kafkajs": {
+      "version": "2.2.4",
+      "resolved": "https://registry.npmjs.org/kafkajs/-/kafkajs-2.2.4.tgz",
+      "integrity": "sha512-j/YeapB1vfPT2iOIUn/vxdyKEuhuY2PxMBvf5JWux6iSaukAccrMtXEY/Lb7OvavDhOWME589bpLrEdnVHjfjA==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=14.0.0"
+      }
+    },
     "node_modules/pg": {
       "version": "8.21.0",
       "resolved": "https://registry.npmjs.org/pg/-/pg-8.21.0.tgz",

package.json

@@ -48,6 +48,7 @@
   "dependencies": {
     "@openai/codex": "^0.128.0",
     "fzstd": "0.1.1",
+    "kafkajs": "^2.2.4",
     "pg": "^8.21.0",
     "playwright": "1.59.1",
     "yaml": "^2.8.3"