pikasTech-HWLAB/internal/cloud/kafka-shadow-producer.ts

// 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 }));
}