From fc7fc23f4e19c8fef317e0ed7321bb0afe86aae9 Mon Sep 17 00:00:00 2001 From: lyon1998 Date: Sun, 28 Jun 2026 19:51:36 +0800 Subject: [PATCH] feat: shadow produce code agent kafka events --- internal/cloud/kafka-shadow-producer.ts | 185 +++++++++++++++++++++++ internal/cloud/server-code-agent-http.ts | 8 + package-lock.json | 10 ++ package.json | 1 + 4 files changed, 204 insertions(+) create mode 100644 internal/cloud/kafka-shadow-producer.ts diff --git a/internal/cloud/kafka-shadow-producer.ts b/internal/cloud/kafka-shadow-producer.ts new file mode 100644 index 00000000..d5ef390a --- /dev/null +++ b/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 })); +} diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index 0e4f265b..97b03fc6 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/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 }); diff --git a/package-lock.json b/package-lock.json index e9fc18d9..74b3c094 100644 --- a/package-lock.json +++ b/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", diff --git a/package.json b/package.json index e3ec8314..1663e886 100644 --- a/package.json +++ b/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"