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, kafkaDnsLookup, kafkaEventBridgeConfig, kafkaMessageKeyCanMatchPartitionKey, kafkaPartitionForKey, projectAgentRunKafkaEventToHwlabEvent, publishAgentRunKafkaMessageLive, queryKafkaEventStream, relayHwlabKafkaOutboxOnce, startHwlabKafkaEventBridge } from "./kafka-event-bridge.ts";

test("Workbench refresh replay resolves the same Kafka partition as the session-keyed producer", () => {
  const offsets = [
    { partition: 0, startOffset: "0", endOffset: "100" },
    { partition: 1, startOffset: "0", endOffset: "200" },
    { partition: 2, startOffset: "0", endOffset: "300" }
  ];
  const sessionId = "ses_refresh_partition_scope";
  const first = kafkaPartitionForKey(sessionId, "hwlab.event.v1", offsets);
  const repeated = kafkaPartitionForKey(sessionId, "hwlab.event.v1", [...offsets].reverse());

  assert.ok([0, 1, 2].includes(first));
  assert.equal(repeated, first);
  assert.equal(kafkaPartitionForKey(null, "hwlab.event.v1", offsets), null);
});

test("Workbench refresh replay rejects foreign keyed values before JSON parsing", () => {
  assert.equal(kafkaMessageKeyCanMatchPartitionKey("ses_target", "ses_target"), true);
  assert.equal(kafkaMessageKeyCanMatchPartitionKey("ses_foreign", "ses_target"), false);
  assert.equal(kafkaMessageKeyCanMatchPartitionKey(null, "ses_target"), true);
  assert.equal(kafkaMessageKeyCanMatchPartitionKey("ses_target", null), true);
});

test("Agent observer retains a bounded latest window while scanning through the Kafka barrier", async () => {
  const messages = Array.from({ length: 5 }, (_, offset) => ({
    offset: String(offset),
    key: Buffer.from(`run-${offset}`),
    value: Buffer.from(JSON.stringify({ schema: "hwlab.event.v1", eventId: `event-${offset}` }))
  }));
  const queried = await queryKafkaEventStream({
    env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
    topic: "hwlab.event.v1",
    limit: 2,
    scanLimit: 10,
    retainLatestMatches: true,
    groupIdPrefix: "hwlab-agent-observer-test",
    kafkaFactory: kafkaQueryTestFactory(messages)
  });

  assert.equal(queried.completionReason, "end-offset");
  assert.equal(queried.completion.complete, true);
  assert.equal(queried.scannedCount, 5);
  assert.equal(queried.matchedCount, 2);
  assert.equal(queried.totalMatchedCount, 5);
  assert.equal(queried.droppedMatchedCount, 3);
  assert.equal(queried.retainLatestMatches, true);
  assert.deepEqual(queried.events.map((event) => event.offset), ["3", "4"]);
});

test("Workbench Kafka query keeps the existing stop-at-match-limit behavior", async () => {
  const messages = Array.from({ length: 5 }, (_, offset) => ({
    offset: String(offset),
    value: Buffer.from(JSON.stringify({ schema: "hwlab.event.v1", eventId: `event-${offset}` }))
  }));
  const queried = await queryKafkaEventStream({
    env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
    topic: "hwlab.event.v1",
    limit: 2,
    scanLimit: 10,
    groupIdPrefix: "hwlab-workbench-query-test",
    kafkaFactory: kafkaQueryTestFactory(messages)
  });

  assert.equal(queried.completionReason, "limit");
  assert.equal(queried.completion.complete, false);
  assert.equal(queried.scannedCount, 2);
  assert.equal(queried.totalMatchedCount, 2);
  assert.equal(queried.droppedMatchedCount, 0);
  assert.equal(queried.retainLatestMatches, false);
  assert.deepEqual(queried.events.map((event) => event.offset), ["0", "1"]);
});

function kafkaQueryTestFactory(messages) {
  return () => ({
    admin() {
      return {
        async connect() {},
        async fetchTopicOffsets() { return [{ partition: 0, low: "0", high: String(messages.length) }]; },
        async disconnect() {}
      };
    },
    consumer() {
      return {
        async connect() {},
        async subscribe() {},
        async run({ eachBatch }) {
          await eachBatch({ batch: { topic: "hwlab.event.v1", partition: 0, messages } });
        },
        async stop() {},
        async disconnect() {}
      };
    }
  });
}

test("Workbench session index uses explicit budgets and defaults missing non-core values with warnings", () => {
  const explicit = kafkaEventBridgeConfig({
    ...LIVE_REFRESH_ENV,
    HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_ENABLED: "true",
    HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_MAX_EVENTS: "100000",
    HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_MAX_EVENTS_PER_SESSION: "10000",
    HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_BOOTSTRAP_TIMEOUT_MS: "30000",
    HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_REBUILD_INTERVAL_MS: "300000",
    HWLAB_WORKBENCH_KAFKA_REFRESH_INDEX_REBUILD_COOLDOWN_MS: "30000"
  });
  assert.deepEqual(explicit.refreshReplay.sessionIndex, {
    enabled: true,
    maxEvents: 100000,
    maxEventsPerSession: 10000,
    bootstrapTimeoutMs: 30000,
    rebuildIntervalMs: 300000,
    rebuildCooldownMs: 30000,
    warnings: [],
    valuesRedacted: true
  });

  const missing = kafkaEventBridgeConfig(LIVE_REFRESH_ENV);
  assert.equal(missing.refreshReplay.sessionIndex.enabled, false);
  assert.equal(missing.refreshReplay.sessionIndex.warnings[0].blocking, false);
});

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

const LIVE_REFRESH_ENV = Object.freeze({
  ...LIVE_ENV,
  HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED: "true",
  HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX: "hwlab-live-refresh-test",
  HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS: "30000",
  HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT: "1000000",
  HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT: "2000",
  HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT: "2000"
});

test("uses explicitly configured DNS servers for Kafka broker metadata hostnames", async () => {
  class FakeResolver {
    static instance;
    constructor() { FakeResolver.instance = this; }
    setServers(servers) { this.servers = servers; }
    resolve4(hostname, callback) {
      this.hostnames ??= [];
      this.hostnames.push(hostname);
      if (hostname.endsWith(".cluster.local")) callback(null, ["10.42.0.64"]);
      else callback(Object.assign(new Error("not found"), { code: "ENOTFOUND" }));
    }
  }
  const lookup = kafkaDnsLookup(["10.43.0.10"], ["cluster.local"], FakeResolver);
  const resolved = await new Promise((resolve, reject) => {
    lookup("platform-infra-kafka-dual-role-0.platform-infra-kafka-kafka-brokers.platform-infra.svc", {}, (error, address, family) => {
      if (error) reject(error);
      else resolve({ address, family });
    });
  });
  assert.deepEqual(FakeResolver.instance.servers, ["10.43.0.10"]);
  assert.deepEqual(FakeResolver.instance.hostnames, [
    "platform-infra-kafka-dual-role-0.platform-infra-kafka-kafka-brokers.platform-infra.svc",
    "platform-infra-kafka-dual-role-0.platform-infra-kafka-kafka-brokers.platform-infra.svc.cluster.local"
  ]);
  assert.deepEqual(resolved, { address: "10.42.0.64", family: 4 });
  assert.equal(kafkaDnsLookup([]), undefined);
  const config = kafkaEventBridgeConfig({
    ...LIVE_ENV,
    HWLAB_KAFKA_DNS_SERVERS: "10.43.0.10,10.43.0.11",
    HWLAB_KAFKA_DNS_SEARCH_DOMAINS: "cluster.local"
  });
  assert.deepEqual(config.dnsServers, ["10.43.0.10", "10.43.0.11"]);
  assert.deepEqual(config.dnsSearchDomains, ["cluster.local"]);
});

test("projects only explicit AgentRun lifecycle timestamps", () => {
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.run.event",
    producedAt: "2026-07-09T10:00:09.000Z",
    run: {
      runId: "run_explicit_lifecycle",
      sessionId: "ses_explicit_lifecycle",
      startedAt: "2026-07-09T10:00:00.000Z",
      finishedAt: "2026-07-09T10:00:08.000Z"
    },
    command: { commandId: "cmd_explicit_lifecycle" },
    event: { seq: 1, type: "assistant_message", createdAt: "2026-07-09T10:00:05.000Z", payload: { text: "progress" } }
  }, { source: "hwlab-test" });
  assert.equal(projected.event.startedAt, "2026-07-09T10:00:00.000Z");
  assert.equal(projected.event.finishedAt, "2026-07-09T10:00:08.000Z");

  const withoutFacts = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.run.event",
    producedAt: "2026-07-09T10:00:09.000Z",
    run: { runId: "run_unknown_lifecycle", sessionId: "ses_unknown_lifecycle" },
    command: { commandId: "cmd_unknown_lifecycle" },
    event: { seq: 1, type: "user_message", createdAt: "2026-07-09T10:00:05.000Z", payload: { text: "hello" } }
  }, { source: "hwlab-test" });
  assert.equal(withoutFacts.event.startedAt, null);
  assert.equal(withoutFacts.event.finishedAt, null);
  assert.equal(withoutFacts.event.createdAt, "2026-07-09T10:00:05.000Z");
  assert.equal(withoutFacts.event.submittedAt, null);
  assert.equal(withoutFacts.event.timingClock, "event-created-at-fallback");
  assert.equal(withoutFacts.event.blocking, false);
});

test("projects AgentRun semantic infrastructure retry without reclassification", () => {
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.run.event",
    producedAt: "2026-07-20T10:00:01.000Z",
    run: { runId: "run_retry", sessionId: "ses_retry", status: "running" },
    command: { commandId: "cmd_retry", state: "pending" },
    event: {
      id: "evt_retry",
      seq: 5,
      type: "backend_status",
      createdAt: "2026-07-20T10:00:01.000Z",
      payload: {
        phase: "runner-startup-retry-scheduled",
        retryPhase: "retryScheduled",
        failureDomain: "infrastructure",
        component: "runner-image-pull",
        code: "runner-image-registry-unreachable",
        summary: "Runner 镜像仓库暂时不可达",
        retryable: true,
        attempt: 2,
        maxAttempts: 5,
        backoffMs: 8_000,
        nextRetryAt: "2026-07-20T10:00:09.000Z",
        firstObservedAt: "2026-07-20T09:59:55.000Z",
        observedAt: "2026-07-20T10:00:01.000Z",
        runnerJobId: "rjob_retry"
      }
    }
  }, { source: "hwlab-test" });
  assert.equal(projected.event.status, "retrying");
  assert.equal(projected.event.failureDomain, "infrastructure");
  assert.equal(projected.event.failureComponent, "runner-image-pull");
  assert.equal(projected.event.failureCode, "runner-image-registry-unreachable");
  assert.equal(projected.event.retryPhase, "retryScheduled");
  assert.equal(projected.event.retryAttempt, 2);
  assert.equal(projected.event.retryMaxAttempts, 5);
  assert.equal(projected.event.retryBackoffMs, 8_000);
  assert.equal(projected.event.nextRetryAt, "2026-07-20T10:00:09.000Z");
  assert.equal(projected.event.runnerJobId, "rjob_retry");
});

test("normalizes sparse AgentRun provider retry metadata for Workbench visibility", () => {
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.run.event",
    producedAt: "2026-07-20T19:32:07.000Z",
    run: { runId: "run_provider_retry", sessionId: "ses_provider_retry", status: "running" },
    command: { commandId: "cmd_provider_retry", state: "running" },
    event: {
      id: "evt_provider_retry",
      seq: 52,
      type: "error",
      createdAt: "2026-07-20T19:32:07.000Z",
      payload: {
        failureKind: "provider-stream-disconnected",
        message: "Provider stream disconnected",
        willRetry: true,
        retryAttempt: 1,
        retryMax: 5,
        retryDelayMs: 240_000
      }
    }
  }, { source: "hwlab-test" });

  assert.equal(projected.event.status, "retrying");
  assert.equal(projected.event.failureDomain, "upstream");
  assert.equal(projected.event.retryPhase, "retryScheduled");
  assert.equal(projected.event.retryAttempt, 1);
  assert.equal(projected.event.retryMaxAttempts, 5);
  assert.equal(projected.event.retryBackoffMs, 240_000);
  assert.equal(projected.event.nextRetryAt, "2026-07-20T19:36:07.000Z");
  assert.equal(projected.event.terminal, false);
});

test("keeps sparse non-retryable provider errors failed", () => {
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.run.event",
    producedAt: "2026-07-20T19:32:07.000Z",
    run: { runId: "run_provider_failed", sessionId: "ses_provider_failed", status: "failed" },
    command: { commandId: "cmd_provider_failed", state: "failed" },
    event: {
      id: "evt_provider_failed",
      seq: 53,
      type: "error",
      createdAt: "2026-07-20T19:32:07.000Z",
      payload: {
        failureKind: "provider-unavailable",
        message: "Provider unavailable",
        willRetry: false
      }
    }
  }, { source: "hwlab-test" });

  assert.equal(projected.event.status, "failed");
  assert.equal(projected.event.failureDomain, "upstream");
  assert.equal(projected.event.retryPhase, null);
});

test("projects AgentRun Git mirror fetch as running preparation", () => {
  const projected = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.run.event",
    producedAt: "2026-07-20T10:00:01.000Z",
    run: { runId: "run_fetch", sessionId: "ses_fetch", status: "running" },
    command: { commandId: "cmd_fetch", state: "pending" },
    event: {
      id: "evt_fetch",
      seq: 5,
      type: "backend_status",
      createdAt: "2026-07-20T10:00:01.000Z",
      payload: {
        phase: "runner-source-fetch-started",
        retryPhase: "initialAttempt",
        failureDomain: "infrastructure",
        component: "git-mirror",
        code: "git-mirror-fetch-in-progress",
        summary: "正在从 Git mirror 获取 Runner 源码",
        attempt: 1,
        maxAttempts: 3
      }
    }
  }, { source: "hwlab-test" });
  assert.equal(projected.event.status, "running");
  assert.equal(projected.event.code, "git-mirror-fetch-in-progress");
  assert.equal(projected.event.failureComponent, "git-mirror");
});

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 truncatedFileChange = projectAgentRunKafkaEventToHwlabEvent({
    schema: "agentrun.event.v1",
    eventType: "agentrun.event.committed",
    eventId: "evt_file_change_truncated",
    traceId: "trc_file_change",
    hwlabSessionId: "ses_file_change",
    runId: "run_file_change",
    event: { id: "evt_file_change_truncated", seq: 22, type: "tool_call", payload: { method: "item/completed", itemId: "item_large", type: "fileChange", toolName: "fileChange", status: "completed", changes: [{ path: "src/large.ts", kind: "add", additions: 400, deletions: 0, diffTruncated: true }], fileCount: 1, summary: { text: "bounded file change metadata", outputTruncated: true }, outputTruncated: true } }
  }, { source: "hwlab-test", sourceTopic: "agentrun.event.debug.v1", sourceOffset: "22" });
  assert.equal(truncatedFileChange.event.message, "AgentRun fileChange (1 file)");
  assert.deepEqual(truncatedFileChange.event.changes, [{ path: "src/large.ts", kind: "add", additions: 400, deletions: 0, diffTruncated: true }]);
  assert.doesNotMatch(truncatedFileChange.event.message, /\[object Object\]/u);

  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",
        submittedAt: "2026-07-10T09:59:55.000Z",
        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.createdAt, "2026-07-10T09:59:55.000Z");
  assert.equal(projected.event.submittedAt, "2026-07-10T09:59:55.000Z");
  assert.equal(projected.event.timingClock, "user-submitted-at");
  assert.equal(projected.event.blocking, false);
  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("Workbench bridge fixes direct publish and live Kafka authority", () => {
  const config = kafkaEventBridgeConfig(LIVE_ENV);
  assert.deepEqual(config?.capabilities, {
    directPublish: true,
    liveKafkaSse: true,
    kafkaRefreshReplay: false,
    transactionalProjector: false,
    projectionOutboxRelay: false,
    projectionRealtime: false
  });
  assert.equal(config?.directPublishGroupId, "hwlab-live-bridge-fixed");
  assert.equal(config?.hwlabEventGroupId, "hwlab-live-fanout-fixed");
  assert.equal(config?.projectorHeartbeatIntervalMs, null);
  assert.equal(config?.relay, null);
});

test("live Kafka startup does not touch transactional v8 readiness", async () => {
  let runtimeReads = 0;
  const runtimeStore = new Proxy({}, { get() { runtimeReads += 1; throw new Error("live Kafka must not require transactional v8 schema"); } });
  const consumers = [];
  const kafkaFactory = () => ({
    consumer() {
      const consumer = { async connect() {}, async subscribe() {}, async run() {}, async stop() {}, async disconnect() {} };
      consumers.push(consumer);
      return consumer;
    },
    producer() { return { async connect() {}, async send() {}, async disconnect() {} }; }
  });
  const bridge = startHwlabKafkaEventBridge({ env: LIVE_ENV, runtimeStore, kafkaFactory, logger: null });
  await bridge.ready;
  assert.equal(runtimeReads, 0);
  assert.equal(consumers.length, 2);
  await bridge.stop();
});

test("live Kafka publishes Artificer runs without HWLAB trace lineage for the unscoped observer", async () => {
  const published = [];
  const message = canonicalMessage({ offset: "52", traceId: null, hwlabSessionId: null, sessionId: "sess_artificer_test" });
  const result = await publishAgentRunKafkaMessageLive({
    topic: "agentrun.event.v1",
    partition: 0,
    message
  }, {
    producer: { async send(input) { published.push(input); return [{ partition: 0, baseOffset: "7" }]; } },
    config: kafkaEventBridgeConfig(LIVE_ENV),
    env: LIVE_ENV,
    logger: null,
    otelSpanEmitter() {}
  });

  assert.equal(result.published, true);
  assert.equal(result.envelope.runId, "run_projector_test");
  assert.equal(result.envelope.sessionId, "sess_artificer_test");
  assert.equal(result.envelope.traceId, null);
  assert.equal(published.length, 1);
  assert.equal(published[0].topic, "hwlab.event.v1");
});

test("Cloud API starts with all six Kafka capabilities and without transactional v8 schema", async () => {
  let transactionalV8Reads = 0;
  const runtimeStore = new Proxy({}, {
    get(_target, property) {
      if (["assertWorkbenchTransactionalRealtimeReady", "workbenchTransactionalRealtimeReadiness"].includes(String(property))) {
        transactionalV8Reads += 1;
        throw new Error("transactional v8 schema must not gate live Kafka Cloud API startup");
      }
      return undefined;
    }
  });
  const kafkaFactory = () => ({
    consumer() { return { async connect() {}, async subscribe() {}, async run() {}, async stop() {}, async disconnect() {} }; },
    producer() { return { async connect() {}, async send() {}, async disconnect() {} }; }
  });
  let bunServeCalls = 0;
  const cloud = await createCloudApiBunServer({
    env: { ...LIVE_REFRESH_ENV },
    runtimeStore,
    kafkaFactory,
    accessController: { async ensureBootstrap() {}, async authenticate() { return { ok: false }; } },
    bunServe() { bunServeCalls += 1; return { port: 31888, stop() {} }; },
    installSignalHandlers: false,
    logger: { log() {}, info() {}, warn() {}, error() {} }
  });
  assert.equal(bunServeCalls, 1);
  assert.equal(transactionalV8Reads, 0);
  await cloud.stop("test");
});

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", sessionId = hwlabSessionId }) {
  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,
      sessionId,
      hwlabSessionId,
      runId,
      commandId: "cmd_projector_test",
      run: { runId, status: "running", sessionId, hwlabSessionId, 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
    }))
  };
}