pikasTech-HWLAB/internal/cloud/kafka-event-bridge.test.ts

// SPEC: PJ2026-0104010803 Workbench事件流可见性 draft-2026-07-09-p0-kafka-authority.
// Responsibility: lock AgentRun Kafka event -> HWLAB Kafka event projection contract without requiring a live Kafka broker.

import assert from "node:assert/strict";
import { test } from "bun:test";

import { createCloudApiBunServer } from "./bun-server.ts";
import { buildCloudApiReadiness } from "./health-contract.ts";
import { decodeCanonicalAgentRunKafkaMessage, kafkaEventBridgeConfig, liveKafkaOtelSpanAttributes, projectAgentRunKafkaEventToHwlabEvent, publishAgentRunKafkaMessageLive, relayHwlabKafkaOutboxOnce, shouldEmitLiveKafkaOtelSpan, startHwlabKafkaEventBridge } from "./kafka-event-bridge.ts";

const PROJECTOR_ENV = Object.freeze({
  HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "false",
  HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "false",
  HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED: "false",
  HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED: "true",
  HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED: "true",
  HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "true",
  HWLAB_KAFKA_ENABLED: "true",
  HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED: "true",
  HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092",
  HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC: "agentrun.event.v1",
  HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1",
  HWLAB_KAFKA_CLIENT_ID: "hwlab-projector-test",
  HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID: "hwlab-projector-test-v1",
  HWLAB_KAFKA_PROJECTOR_GROUP_ID: "hwlab-projector-test-v1",
  HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS: "10",
  HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS: "60000",
  HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE: "1",
  HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS: "30000",
  HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS: "10000",
  HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS: "1000"
});

const LIVE_ENV = Object.freeze({
  HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "true",
  HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true",
  HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED: "false",
  HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED: "false",
  HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED: "false",
  HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false",
  HWLAB_KAFKA_ENABLED: "true",
  HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED: "true",
  HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092",
  HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC: "agentrun.event.v1",
  HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1",
  HWLAB_KAFKA_CLIENT_ID: "hwlab-live-test",
  HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID: "hwlab-live-bridge-fixed",
  HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID: "hwlab-live-fanout-fixed"
});

test("projects AgentRun assistant_message Kafka event into HWLAB trace event", () => {
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.run.event",
    source: "agentrun-manager",
    producedAt: "2026-07-09T10:00:00.000Z",
    run: {
      runId: "run_kafka_projection",
      sessionId: "ses_kafka_projection",
      conversationId: "cnv_kafka_projection",
      threadId: "thread-kafka-projection",
      projectId: "prj_kafka_projection",
      backendProfile: "deepseek"
    },
    command: { commandId: "cmd_kafka_projection" },
    event: {
      seq: 7,
      type: "assistant_message",
      createdAt: "2026-07-09T10:00:01.000Z",
      payload: {
        traceId: "trc_kafka_projection",
        commandId: "cmd_kafka_projection",
        text: "partial answer",
        itemId: "item_assistant_1",
        final: false
      }
    }
  }, { source: "hwlab-test", producedAt: "2026-07-09T10:00:02.000Z", sourceTopic: "agentrun.event.v1", sourceOffset: "42" });

  assert.equal(projected.schema, "hwlab.event.v1");
  assert.equal(projected.eventType, "hwlab.trace.event.projected");
  assert.equal(projected.source, "hwlab-test");
  assert.equal(projected.traceId, "trc_kafka_projection");
  assert.equal(projected.hwlabSessionId, "ses_kafka_projection");
  assert.equal(projected.sessionId, "ses_kafka_projection");
  assert.equal(projected.runId, "run_kafka_projection");
  assert.equal(projected.commandId, "cmd_kafka_projection");
  assert.equal(projected.context.runId, "run_kafka_projection");
  assert.equal(projected.context.commandId, "cmd_kafka_projection");
  assert.equal(projected.context.sourceSeq, 7);
  assert.equal(projected.context.agentRunEventType, "assistant_message");
  assert.equal(projected.event.type, "assistant");
  assert.equal(projected.event.sourceSeq, 7);
  assert.equal(Object.prototype.hasOwnProperty.call(projected.event, "projectedSeq"), false);
  assert.equal(projected.event.label, "agentrun:assistant:message");
  assert.equal(projected.event.status, "running");
  assert.equal(projected.event.text, "partial answer");
  assert.equal(projected.event.assistantText, "partial answer");
  assert.equal(projected.event.finalResponse, null);
  assert.equal(projected.event.terminal, false);
  assert.equal(projected.sourceEvent.topic, "agentrun.event.v1");
  assert.equal(projected.sourceEvent.offset, "42");
  assert.equal(projected.valuesPrinted, false);
});

test("projects AgentRun fileChange and diff events without dropping paths or unified diff", () => {
  const changes = [
    { path: "src/example.ts", kind: "update", diff: "@@ -1 +1 @@\n-export const value = 1;\n+export const value = 2;" },
    { path: "docs/说明.md", kind: "add", diff: "@@ -0,0 +1 @@\n+# 文件编辑可见" }
  ];
  const fileChange = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.event.committed",
    eventId: "evt_file_change_completed",
    traceId: "trc_file_change",
    hwlabSessionId: "ses_file_change",
    runId: "run_file_change",
    commandId: "cmd_file_change",
    event: { id: "evt_file_change_completed", seq: 21, type: "tool_call", payload: { method: "item/completed", itemId: "item_file_change", type: "fileChange", toolName: "fileChange", status: "completed", changes, fileCount: 2 } }
  }, { source: "hwlab-test", sourceTopic: "agentrun.event.debug.v1", sourceOffset: "21" });
  assert.equal(fileChange.event.type, "tool");
  assert.equal(fileChange.event.toolName, "fileChange");
  assert.equal(fileChange.event.itemId, "item_file_change");
  assert.deepEqual(fileChange.event.changes, changes);
  assert.equal(fileChange.event.fileCount, 2);

  const diff = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.event.committed",
    eventId: "evt_turn_diff_updated",
    traceId: "trc_file_change",
    hwlabSessionId: "ses_file_change",
    runId: "run_file_change",
    commandId: "cmd_file_change",
    event: { id: "evt_turn_diff_updated", seq: 22, type: "diff", payload: { phase: "turn/diff/updated", threadId: "thread_file_change", turnId: "turn_file_change", status: "updated", diff: changes.map((change) => change.diff).join("\n") } }
  }, { source: "hwlab-test", sourceTopic: "agentrun.event.debug.v1", sourceOffset: "22" });
  assert.equal(diff.event.type, "diff");
  assert.equal(diff.event.eventType, "diff");
  assert.equal(diff.event.threadId, "thread_file_change");
  assert.match(diff.event.diff, /docs\/说明\.md|文件编辑可见/u);
});

test("projects AgentRun assistant_progress as an identity-preserving HWLAB assistant lifecycle event", () => {
  const text = "p".repeat(500);
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.event.committed",
    eventId: "evt_assistant_progress",
    sourceSeq: 136,
    traceId: "trc_assistant_progress",
    hwlabSessionId: "ses_assistant_progress",
    runId: "run_assistant_progress",
    commandId: "cmd_assistant_progress",
    valuesPrinted: false,
    run: { runId: "run_assistant_progress", sessionId: "ses_agentrun_progress", hwlabSessionId: "ses_assistant_progress", valuesPrinted: false },
    command: { commandId: "cmd_assistant_progress", runId: "run_assistant_progress", seq: 1, type: "turn", state: "running", payloadHash: "hash", valuesPrinted: false },
    event: {
      id: "evt_assistant_progress",
      runId: "run_assistant_progress",
      seq: 136,
      type: "assistant_progress",
      createdAt: "2026-07-12T10:00:01.000Z",
      payload: {
        traceId: "trc_assistant_progress",
        commandId: "cmd_assistant_progress",
        text,
        itemId: "msg_assistant_progress",
        source: "agent-message-delta-progress",
        progress: true,
        progressFlush: false,
        replyAuthority: false,
        final: false,
        textBytes: 500,
        outputBytes: 500,
        valuesPrinted: false
      }
    }
  }, { source: "hwlab-test", sourceTopic: "agentrun.event.v1", sourceOffset: "136" });

  assert.equal(projected.context.agentRunEventType, "assistant_progress");
  assert.equal(projected.event.agentRunEventType, "assistant_progress");
  assert.equal(projected.event.type, "assistant");
  assert.equal(projected.event.eventType, "assistant_progress");
  assert.equal(projected.event.label, "agentrun:assistant:progress");
  assert.equal(projected.event.itemId, "msg_assistant_progress");
  assert.equal(projected.event.sourceSeq, 136);
  assert.equal(projected.event.assistantSource, "agent-message-delta-progress");
  assert.equal(projected.event.progress, true);
  assert.equal(projected.event.progressFlush, false);
  assert.equal(projected.event.replyAuthority, false);
  assert.equal(projected.event.final, false);
  assert.equal(projected.event.textBytes, 500);
  assert.equal(projected.event.assistantText, text);
  assert.equal(projected.event.finalResponse, null);
  assert.equal(projected.event.terminal, false);
});

test("projects AgentRun user_message into a stable HWLAB user event", () => {
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.event.committed",
    eventId: "evt_user_message",
    traceId: "trc_user_message",
    hwlabSessionId: "ses_user_message",
    run: { runId: "run_user_message", sessionId: "ses_agentrun_user_message" },
    command: { commandId: "cmd_user_message" },
    event: {
      id: "evt_user_message",
      seq: 2,
      type: "user_message",
      createdAt: "2026-07-10T10:00:00.000Z",
      payload: {
        commandId: "cmd_user_message",
        traceId: "trc_user_message",
        targetTraceId: "trc_user_message_target",
        userMessageId: "msg_user_message_user",
        text: "replay this input",
        source: "agentrun-turn-command-payload",
        valuesPrinted: false
      }
    }
  }, { source: "hwlab-test", sourceTopic: "agentrun.event.v1", sourcePartition: 1, sourceOffset: "7" });

  assert.equal(projected.eventId, "hwlab:evt_user_message");
  assert.equal(projected.traceId, "trc_user_message");
  assert.equal(projected.sessionId, "ses_user_message");
  assert.equal(projected.event.type, "user");
  assert.equal(projected.event.eventType, "user");
  assert.equal(projected.event.role, "user");
  assert.equal(projected.event.userMessageId, "msg_user_message_user");
  assert.equal(projected.event.messageId, "msg_user_message_user");
  assert.equal(projected.event.targetTraceId, "trc_user_message_target");
  assert.equal(projected.event.text, "replay this input");
  assert.equal(projected.event.sourceEventId, "evt_user_message");
  assert.equal(projected.sourceEvent.partition, 1);
  assert.equal(projected.sourceEvent.offset, "7");
});

test("projects final AgentRun assistant text without sealing before terminal_status", () => {
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.run.event",
    run: { runId: "run_final", sessionId: "ses_final" },
    command: { commandId: "cmd_final" },
    event: { seq: 9, type: "assistant_message", payload: { traceId: "trc_final", text: "durable final answer", final: true, replyAuthority: true } }
  }, { source: "hwlab-test" });

  assert.equal(projected.event.assistantText, "durable final answer");
  assert.equal(projected.event.status, "running");
  assert.equal(projected.event.terminal, false);
  assert.equal(projected.event.finalResponse.text, "durable final answer");
  assert.equal(projected.event.finalResponse.traceId, "trc_final");
});

test("projects structured AgentRun commandExecution visibility into HWLAB tool events", () => {
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.run.event",
    run: { runId: "run_tool", sessionId: "ses_tool" },
    command: { commandId: "cmd_tool" },
    event: {
      seq: 10,
      type: "tool_call",
      payload: {
        traceId: "trc_tool",
        type: "commandExecution",
        toolName: "commandExecution",
        method: "item/completed",
        itemId: "tool_rg",
        status: "completed",
        command: "/bin/bash -lc 'cd /workspace && rg -n target src'",
        cwd: "/workspace",
        exitCode: 0,
        durationMs: 1234,
        stdoutSummary: "src/main.ts:42:target",
        stderrSummary: "warning summary",
        outputSummary: "1 match",
        summary: {
          outputBytes: 24,
          outputTruncated: true
        }
      }
    }
  }, { source: "hwlab-test" });

  assert.equal(projected.event.toolName, "commandExecution");
  assert.equal(projected.event.toolType, "commandExecution");
  assert.equal(projected.event.command, "/bin/bash -lc 'cd /workspace && rg -n target src'");
  assert.equal(projected.event.cwd, "/workspace");
  assert.equal(projected.event.exitCode, 0);
  assert.equal(projected.event.durationMs, 1234);
  assert.equal(projected.event.stdoutSummary, "src/main.ts:42:target");
  assert.equal(projected.event.stderrSummary, "warning summary");
  assert.equal(projected.event.outputSummary, "1 match");
  assert.equal(projected.event.outputBytes, 24);
  assert.equal(projected.event.outputTruncated, true);
});

test("projects AgentRun terminal_status Kafka event into terminal HWLAB event", () => {
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.run.event",
    run: { runId: "run_terminal", sessionId: "ses_terminal" },
    command: { commandId: "cmd_terminal" },
    event: {
      seq: 12,
      type: "terminal_status",
      payload: {
        traceId: "trc_terminal",
        terminalStatus: "completed",
        message: "AgentRun completed"
      }
    }
  }, { source: "hwlab-test" });

  assert.equal(projected.traceId, "trc_terminal");
  assert.equal(projected.event.type, "result");
  assert.equal(projected.event.eventType, "terminal");
  assert.equal(projected.event.status, "completed");
  assert.equal(projected.event.label, "agentrun:terminal:completed");
  assert.equal(projected.event.terminal, true);
  assert.equal(projected.event.message, "AgentRun completed");
});

test("live Kafka publisher has no PG inbox, facts, checkpoint, or outbox dependency", async () => {
  const sent = [];
  const runtimeStore = new Proxy({}, { get() { throw new Error("unexpected DB access"); } });
  void runtimeStore;
  const result = await publishAgentRunKafkaMessageLive({ topic: "agentrun.event.v1", partition: 3, message: canonicalMessage({ offset: "71" }) }, {
    producer: { async send(input) { sent.push(input); } },
    config: { clientId: "hwlab-live-test", hwlabTopic: "hwlab.event.v1" },
    logger: null
  });
  assert.equal(result.published, true);
  assert.equal(sent.length, 1);
  assert.equal(JSON.parse(sent[0].messages[0].value).sourceEvent.offset, "71");
});

test("live Kafka OTel sampling is deterministic and bounded without payload attributes", () => {
  const envelope = (agentRunEventType, sourceSeq, event = {}) => ({
    schema: "hwlab.event.v1",
    eventId: `hwlab:evt_otel_${sourceSeq}`,
    sourceEventId: `evt_otel_${sourceSeq}`,
    traceId: "trc_live_otel",
    hwlabSessionId: "ses_live_otel",
    runId: "run_live_otel",
    commandId: "cmd_live_otel",
    context: { agentRunEventType, sourceSeq, valuesRedacted: true },
    event: { type: "backend", eventType: "backend", terminal: false, message: "must-not-leak", payload: { password: "must-not-leak" }, ...event }
  });

  assert.equal(shouldEmitLiveKafkaOtelSpan(envelope("assistant_message", 1, { type: "assistant" })), true);
  assert.equal(shouldEmitLiveKafkaOtelSpan(envelope("tool_call", 2, { type: "tool" })), true);
  assert.equal(shouldEmitLiveKafkaOtelSpan(envelope("terminal_status", 3, { type: "result", terminal: true })), true);
  const decisions = Array.from({ length: 64 }, (_, index) => shouldEmitLiveKafkaOtelSpan(envelope("command_output", index + 1, { type: "output" })));
  assert.deepEqual(decisions, Array.from({ length: 64 }, (_, index) => (index + 1) % 32 === 0));
  assert.deepEqual(decisions, Array.from({ length: 64 }, (_, index) => shouldEmitLiveKafkaOtelSpan(envelope("command_output", index + 1, { type: "output" }))));

  const attributes = liveKafkaOtelSpanAttributes(envelope("assistant_message", 1, { type: "assistant" }), {
    topic: "hwlab.event.v1",
    partition: 2,
    offset: "14"
  });
  assert.deepEqual(attributes, {
    businessTraceId: "trc_live_otel",
    hwlabSessionId: "ses_live_otel",
    runId: "run_live_otel",
    commandId: "cmd_live_otel",
    topic: "hwlab.event.v1",
    partition: 2,
    offset: "14",
    sourceTopic: null,
    sourcePartition: null,
    sourceOffset: null,
    eventType: "assistant_message",
    terminal: false,
    valuesRedacted: true
  });
  assert.equal(JSON.stringify(attributes).includes("must-not-leak"), false);
});

test("Workbench bridge ignores obsolete architecture env and fixes transactional authority", () => {
  const config = kafkaEventBridgeConfig({
    ...PROJECTOR_ENV,
    HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "true",
    HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true",
    HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED: "true",
    HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED: "false",
    HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED: "false",
    HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false"
  });

  assert.deepEqual(config?.capabilities, {
    directPublish: false,
    liveKafkaSse: false,
    kafkaRefreshReplay: false,
    transactionalProjector: true,
    projectionOutboxRelay: true,
    projectionRealtime: true
  });
  assert.equal(config?.directPublishGroupId, null);
  assert.equal(config?.hwlabEventGroupId, null);
  assert.equal(config?.refreshReplay, null);
});

test("Workbench bridge does not require the six obsolete architecture env", () => {
  const env = { ...PROJECTOR_ENV };
  for (const name of [
    "HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED",
    "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED",
    "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED",
    "HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED",
    "HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED",
    "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED"
  ]) delete env[name];

  assert.doesNotThrow(() => kafkaEventBridgeConfig(env));
});

test("projects AgentRun run-created envelope even when no source event is present", () => {
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.run.created",
    run: {
      runId: "run_created",
      sessionId: "ses_created",
      traceId: "trc_created",
      backendProfile: "codex"
    }
  }, { source: "hwlab-test" });

  assert.equal(projected.traceId, "trc_created");
  assert.equal(projected.sessionId, "ses_created");
  assert.equal(projected.context.runId, "run_created");
  assert.equal(projected.context.sourceEventType, "agentrun.run.created");
  assert.equal(projected.event.label, "agentrun:event:agentrun.run.created");
  assert.equal(projected.event.status, "running");
});

test("accepts canonical runner-job-created payload with redacted tool credential references", () => {
  const message = canonicalMessage({ offset: "60" });
  const envelope = JSON.parse(message.value.toString("utf8"));
  envelope.event.type = "backend_status";
  envelope.event.payload = {
    traceId: envelope.traceId,
    hwlabSessionId: envelope.hwlabSessionId,
    commandId: envelope.commandId,
    phase: "runner-job-created",
    toolCredentials: [{
      tool: "unidesk-ssh",
      purpose: "ssh-passthrough",
      name: "agentrun-v02-tool-unidesk-ssh",
      namespace: "agentrun-v02",
      keys: ["UNIDESK_SSH_CLIENT_TOKEN"],
      projection: { kind: "env", envName: "UNIDESK_SSH_CLIENT_TOKEN" },
      valuesPrinted: false
    }],
    valuesPrinted: false
  };
  message.value = Buffer.from(JSON.stringify(envelope));

  const decoded = decodeCanonicalAgentRunKafkaMessage({ topic: "agentrun.event.v1", partition: 0, message });
  assert.equal(decoded.ok, true);
  const projected = projectAgentRunKafkaEventToHwlabEvent(decoded.event, { source: "hwlab-test" });
  assert.equal(projected.event.label, "agentrun:backend:runner-job-created");
});

test("Kafka projector commits next offset only after durable projection", async () => {
  const calls = [];
  let commitInput = null;
  const harness = kafkaProjectorHarness({
    async commitAgentRunKafkaProjection(input) {
      commitInput = input;
      calls.push("db");
      return { projectedSeq: 1 };
    }
  }, calls);
  const bridge = startHwlabKafkaEventBridge({ env: PROJECTOR_ENV, runtimeStore: harness.runtimeStore, kafkaFactory: harness.kafkaFactory, logger: null });
  await bridge.ready;
  await harness.runBatch(canonicalMessage({ offset: "41" }));

  assert.ok(calls.indexOf("heartbeat") < calls.indexOf("db"));
  assert.ok(calls.indexOf("db") < calls.indexOf("resolve:41"));
  assert.equal(commitInput.partitionKey, "ses_projector_test");
  assert.deepEqual(calls.slice(-3), ["resolve:41", "commit:42", "heartbeat"]);
  await bridge.stop();
});

test("Kafka projector leaves offset uncommitted when durable projection throws", async () => {
  const calls = [];
  const expected = Object.assign(new Error("database unavailable"), { code: "db_unavailable" });
  const harness = kafkaProjectorHarness({
    async commitAgentRunKafkaProjection() {
      calls.push("db");
      throw expected;
    }
  }, calls);
  const bridge = startHwlabKafkaEventBridge({ env: PROJECTOR_ENV, runtimeStore: harness.runtimeStore, kafkaFactory: harness.kafkaFactory, logger: null });
  await bridge.ready;

  await assert.rejects(harness.runBatch(canonicalMessage({ offset: "42" })), expected);
  assert.equal(calls.some((entry) => entry.startsWith("commit:")), false);
  assert.equal(calls.some((entry) => entry.startsWith("resolve:")), false);
  await bridge.stop();
});

test("Kafka projector signals committed projection but does not commit a stale generation offset", async () => {
  const calls = [];
  let signal = null;
  let harness;
  harness = kafkaProjectorHarness({
    async commitAgentRunKafkaProjection() {
      calls.push("db");
      harness.setStale(true);
      return { projectedSeq: 1 };
    }
  }, calls);
  const bridge = startHwlabKafkaEventBridge({ env: PROJECTOR_ENV, runtimeStore: harness.runtimeStore, kafkaFactory: harness.kafkaFactory, logger: null });
  bridge.subscribeProjectionCommits((value) => { signal = value; });
  await bridge.ready;

  await harness.runBatch(canonicalMessage({ offset: "43" }));

  assert.equal(calls.includes("resolve:43"), false);
  assert.equal(calls.includes("commit:44"), false);
  assert.equal(signal.sessionId, "ses_projector_test");
  assert.equal(signal.traceId, "trc_projector_test");
  await bridge.stop();
});

test("Kafka projector commits a suppressed post-seal offset without publishing a projection signal", async () => {
  const calls = [];
  let signal = null;
  const harness = kafkaProjectorHarness({
    async commitAgentRunKafkaProjection() {
      calls.push("db");
      return { projectedSeq: 9, projectionWritten: false, suppressedAfterSeal: true };
    }
  }, calls);
  const bridge = startHwlabKafkaEventBridge({ env: PROJECTOR_ENV, runtimeStore: harness.runtimeStore, kafkaFactory: harness.kafkaFactory, logger: null });
  bridge.subscribeProjectionCommits((value) => { signal = value; });
  await bridge.ready;

  await harness.runBatch(canonicalMessage({ offset: "44" }));

  assert.equal(signal, null);
  assert.deepEqual(calls.slice(-3), ["resolve:44", "commit:45", "heartbeat"]);
  await bridge.stop();
});

test("Kafka projector persists malformed and non-HWLAB events before committing offsets", async () => {
  const calls = [];
  const harness = kafkaProjectorHarness({}, calls);
  const bridge = startHwlabKafkaEventBridge({ env: PROJECTOR_ENV, runtimeStore: harness.runtimeStore, kafkaFactory: harness.kafkaFactory, logger: null });
  await bridge.ready;
  await harness.runBatch({ offset: "50", value: Buffer.from("{}") });
  await harness.runBatch(canonicalMessage({ offset: "51", traceId: null, hwlabSessionId: null }));

  assert.deepEqual(calls.filter((entry) => ["dlq", "ignored", "commit:51", "commit:52"].includes(entry)), ["dlq", "commit:51", "ignored", "commit:52"]);
  await bridge.stop();
});

test("Kafka projector exposes startup failure as rejected readiness and blocked status", async () => {
  const expected = Object.assign(new Error("broker unavailable"), { code: "kafka_unavailable" });
  const harness = kafkaProjectorHarness({}, []);
  harness.kafkaFactory = () => ({
    consumer: () => harness.consumer,
    producer: () => ({ ...harness.producer, connect: async () => { throw expected; } })
  });
  const bridge = startHwlabKafkaEventBridge({ env: PROJECTOR_ENV, runtimeStore: harness.runtimeStore, kafkaFactory: harness.kafkaFactory, logger: null });

  await assert.rejects(bridge.ready, expected);
  assert.equal((await bridge.status()).status, "blocked");
  assert.equal((await bridge.status()).errorCode, "kafka_unavailable");
  await bridge.stop();
});

test("cloud API Bun entry does not listen when Kafka projector startup fails", async () => {
  const expected = Object.assign(new Error("broker unavailable before listen"), { code: "kafka_startup_blocked" });
  const harness = kafkaProjectorHarness({}, []);
  let bunServeCalls = 0;
  const kafkaFactory = () => ({
    consumer: () => harness.consumer,
    producer: () => ({ ...harness.producer, connect: async () => { throw expected; } })
  });

  await assert.rejects(createCloudApiBunServer({
    env: { ...PROJECTOR_ENV },
    runtimeStore: harness.runtimeStore,
    kafkaFactory,
    accessController: { async ensureBootstrap() {}, async authenticate() { return { ok: false }; } },
    bunServe() { bunServeCalls += 1; throw new Error("Bun.serve must not run"); },
    installSignalHandlers: false,
    logger: { log() {}, info() {}, warn() {}, error() {} }
  }), expected);

  assert.equal(bunServeCalls, 0);
});

test("cloud readiness is blocked while the Kafka projector is blocked", () => {
  const readiness = buildCloudApiReadiness({
    db: {},
    codeAgent: {},
    runtime: {},
    kafkaProjector: { started: true, status: "blocked", errorCode: "kafka_startup_blocked", message: "broker unavailable" }
  });

  assert.equal(readiness.components.kafkaProjector, "blocked");
  assert.equal(readiness.blockerCodes.includes("kafka_startup_blocked"), true);
  assert.equal(readiness.ready, false);
});

test("HWLAB Kafka relay claims just in time and forwards fencing token", async () => {
  const calls = [];
  const pending = [
    { outboxSeq: 1, eventId: "evt-relay-1", topic: "hwlab.event.v1", partitionKey: "trc-relay", payload: { eventId: "evt-relay-1" }, headers: {}, attemptCount: 2, leaseOwner: "relay-owner", leaseExpiresAt: "2026-07-10T10:02:30.000Z" },
    { outboxSeq: 2, eventId: "evt-relay-2", topic: "hwlab.event.v1", partitionKey: "trc-relay", payload: { eventId: "evt-relay-2" }, headers: {}, attemptCount: 1, leaseOwner: "relay-owner", leaseExpiresAt: "2026-07-10T10:02:30.000Z" }
  ];
  const runtimeStore = {
    async claimHwlabKafkaOutbox(params) { calls.push(`claim:${params.limit}`); return pending.length ? [pending.shift()] : []; },
    async completeHwlabKafkaOutbox(item) { calls.push(`complete:${item.outboxSeq}:${item.attemptCount}:${item.leaseOwner}`); }
  };
  const producer = { async send(record) { calls.push(`send:${record.messages[0].value.includes("evt-relay-1") ? 1 : 2}`); } };
  const config = { relay: { batchSize: 2, leaseMs: 30000, retryBackoffMs: 1000 } };

  const result = await relayHwlabKafkaOutboxOnce({ runtimeStore, producer, config, owner: "relay-owner", logger: null });

  assert.equal(result.deliveredCount, 2);
  assert.deepEqual(calls, ["claim:1", "send:1", "complete:1:2:relay-owner", "claim:1", "send:2", "complete:2:1:relay-owner"]);
});

test("HWLAB Kafka relay retries a failed send and clears backlog on the next cycle", async () => {
  const calls = [];
  let pending = true;
  let attemptCount = 0;
  const runtimeStore = {
    async claimHwlabKafkaOutbox() {
      if (!pending) return [];
      attemptCount += 1;
      return [{ outboxSeq: 7, eventId: "evt-relay-retry", topic: "hwlab.event.v1", partitionKey: "trc-relay-retry", payload: { eventId: "evt-relay-retry" }, headers: {}, attemptCount, leaseOwner: "relay-retry-owner", leaseExpiresAt: `2026-07-10T10:0${attemptCount}:30.000Z` }];
    },
    async completeHwlabKafkaOutbox(item) {
      calls.push(`complete:${item.attemptCount}`);
      pending = false;
    },
    async retryHwlabKafkaOutbox(item, error, retryAt) {
      calls.push(`retry:${item.attemptCount}:${error.message}:${retryAt}`);
    },
    async hwlabKafkaProjectorStatus() {
      return { backlogCount: pending ? 1 : 0, retryCount: pending && attemptCount > 0 ? 1 : 0, failedInboxCount: 0, dlqCount: 0 };
    }
  };
  let sendCount = 0;
  const producer = {
    async send() {
      sendCount += 1;
      calls.push(`send:${sendCount}`);
      if (sendCount === 1) throw new Error("broker unavailable");
    }
  };
  const config = { relay: { batchSize: 1, leaseMs: 30000, sendTimeoutMs: 10000, retryBackoffMs: 1000 } };

  const failedCycle = await relayHwlabKafkaOutboxOnce({ runtimeStore, producer, config, owner: "relay-retry-owner", now: () => "2026-07-10T10:00:00.000Z", logger: null });
  assert.deepEqual(failedCycle, { claimedCount: 1, deliveredCount: 0, retryCount: 1, valuesPrinted: false });
  assert.deepEqual(await runtimeStore.hwlabKafkaProjectorStatus(), { backlogCount: 1, retryCount: 1, failedInboxCount: 0, dlqCount: 0 });

  const recoveredCycle = await relayHwlabKafkaOutboxOnce({ runtimeStore, producer, config, owner: "relay-retry-owner", now: () => "2026-07-10T10:00:02.000Z", logger: null });
  assert.deepEqual(recoveredCycle, { claimedCount: 1, deliveredCount: 1, retryCount: 0, valuesPrinted: false });
  assert.deepEqual(await runtimeStore.hwlabKafkaProjectorStatus(), { backlogCount: 0, retryCount: 0, failedInboxCount: 0, dlqCount: 0 });
  assert.deepEqual(calls, [
    "send:1",
    "retry:1:broker unavailable:2026-07-10T10:00:01.000Z",
    "send:2",
    "complete:2"
  ]);
});

function kafkaProjectorHarness(overrides = {}, calls = []) {
  let runOptions = null;
  let stale = false;
  const consumer = {
    async connect() {},
    async subscribe() {},
    async run(options) { runOptions = options; },
    async commitOffsets(offsets) { calls.push(`commit:${offsets[0].offset}`); },
    async stop() {},
    async disconnect() {}
  };
  const producer = { async connect() {}, async send() {}, async disconnect() {} };
  const runtimeStore = {
    async assertWorkbenchTransactionalRealtimeReady() { return { ready: true }; },
    async subscribeWorkbenchProjectionCommits() { return () => {}; },
    async commitAgentRunKafkaProjection() { return { projectedSeq: 1 }; },
    async recordFailedAgentRunKafkaMessage() { calls.push("dlq"); },
    async recordIgnoredAgentRunKafkaMessage() { calls.push("ignored"); return { ignored: true }; },
    async claimHwlabKafkaOutbox() { return []; },
    async completeHwlabKafkaOutbox() {},
    async retryHwlabKafkaOutbox() {},
    async hwlabKafkaProjectorStatus() { return { backlogCount: 0 }; },
    ...overrides
  };
  return {
    consumer,
    producer,
    runtimeStore,
    kafkaFactory: () => ({ consumer: () => consumer, producer: () => producer }),
    setStale(value) { stale = value === true; },
    async runBatch(message) {
      assert.ok(runOptions?.eachBatch);
      await runOptions.eachBatch({
        batch: { topic: "agentrun.event.v1", partition: 2, messages: [message] },
        resolveOffset(offset) { calls.push(`resolve:${offset}`); },
        async heartbeat() { calls.push("heartbeat"); },
        isRunning: () => true,
        isStale: () => stale
      });
    }
  };
}

function canonicalMessage({ offset, traceId = "trc_projector_test", hwlabSessionId = "ses_projector_test" }) {
  const eventId = `evt_projector_${offset}`;
  const runId = "run_projector_test";
  return {
    offset,
    key: Buffer.from(runId),
    value: Buffer.from(JSON.stringify({
      schema: "agentrun.event.v1",
      eventType: "agentrun.event.committed",
      source: "agentrun-v02",
      eventId,
      sourceSeq: Number(offset) + 1,
      traceId,
      hwlabSessionId,
      runId,
      commandId: "cmd_projector_test",
      run: { runId, status: "running", valuesPrinted: false },
      command: { commandId: "cmd_projector_test", runId, seq: 1, type: "prompt", state: "running", valuesPrinted: false },
      event: { id: eventId, runId, seq: Number(offset) + 1, type: "assistant_message", payload: { traceId, hwlabSessionId, commandId: "cmd_projector_test", text: "projected answer" }, createdAt: "2026-07-10T10:00:00.000Z" },
      valuesPrinted: false
    }))
  };
}