pikasTech-HWLAB/internal/cloud/workbench-kafka-sse-debug.test.ts

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

import { handleWorkbenchKafkaSseDebugHttp } from "./workbench-kafka-sse-debug.ts";

test("workbench Kafka SSE debug endpoint streams filtered raw HWLAB Kafka events", async () => {
  const fakeKafka = createFakeKafkaFactory();
  const env = {
    HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092",
    HWLAB_KAFKA_CLIENT_ID: "test-hwlab-cloud-api",
    HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1"
  };
  const server = createServer((request, response) => {
    const url = new URL(request.url ?? "/", "http://127.0.0.1");
    void handleWorkbenchKafkaSseDebugHttp(request, response, url, { env, kafkaFactory: fakeKafka.factory, accessController: fakeAccessController(), logger: null });
  });
  await listen(server);
  const abort = new AbortController();
  try {
    const response = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab&sessionId=ses_kafka_sse_debug`, { signal: abort.signal });
    assert.equal(response.status, 200);
    assert.match(response.headers.get("content-type") ?? "", /text\/event-stream/u);
    const reader = response.body?.getReader();
    assert.ok(reader, "SSE response must expose a readable stream");
    const connected = await readUntil(reader, "hwlab.kafka.connected");
    assert.match(connected, /hwlab\.event\.v1/u);
    assert.match(connected, /"consumerReady":true/u);
    assert.match(connected, new RegExp(String(fakeKafka.groupId).replace(/[.*+?^${}()|[\]\\]/gu, "\\$&"), "u"));

    await fakeKafka.emit({ eventType: "hwlab.trace.event.projected", sessionId: "ses_other", event: { label: "ignored" } });
    await fakeKafka.emit({ eventType: "hwlab.trace.event.projected", sessionId: "ses_kafka_sse_debug", traceId: "trc_kafka_sse_debug", context: { runId: "run_kafka_sse_debug", commandId: "cmd_kafka_sse_debug" }, event: { label: "agentrun:event:debug", status: "running" } });
    const streamed = await readUntil(reader, "trc_kafka_sse_debug");
    assert.match(streamed, /hwlab\.kafka\.event/u);
    assert.match(streamed, /ses_kafka_sse_debug/u);
    assert.doesNotMatch(streamed, /ses_other/u);
  } finally {
    abort.abort();
    await close(server);
  }
});

test("workbench Kafka SSE debug endpoint streams filtered codex stdio Kafka events", async () => {
  const fakeKafka = createFakeKafkaFactory();
  const env = {
    HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092",
    HWLAB_KAFKA_CLIENT_ID: "test-hwlab-cloud-api",
    HWLAB_KAFKA_STDIO_TOPIC: "codex-stdio.raw.v1"
  };
  const server = createServer((request, response) => {
    const url = new URL(request.url ?? "/", "http://127.0.0.1");
    void handleWorkbenchKafkaSseDebugHttp(request, response, url, { env, kafkaFactory: fakeKafka.factory, accessController: fakeAccessController(), logger: null });
  });
  await listen(server);
  const abort = new AbortController();
  try {
    const response = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=stdio&traceId=trc_stdio_sse_debug&fromBeginning=true`, { signal: abort.signal });
    assert.equal(response.status, 200);
    const reader = response.body?.getReader();
    assert.ok(reader, "SSE response must expose a readable stream");
    const connected = await readUntil(reader, "hwlab.kafka.connected");
    assert.match(connected, /codex-stdio\.raw\.v1/u);

    await fakeKafka.emit({ traceId: "trc_other", stream: "stdout", line: "ignored" });
    await fakeKafka.emit({ trace_id: "trc_stdio_sse_debug", metadata: { session_id: "ses_stdio_sse_debug", run_id: "run_stdio_sse_debug" }, stream: "stdout", line: "visible stdio frame" });
    const streamed = await readUntil(reader, "visible stdio frame");
    assert.match(streamed, /hwlab\.kafka\.event/u);
    assert.match(streamed, /codex-stdio\.raw\.v1/u);
    assert.match(streamed, /ses_stdio_sse_debug/u);
    assert.doesNotMatch(streamed, /trc_other/u);
  } finally {
    abort.abort();
    await close(server);
  }
});

test("workbench Kafka SSE debug endpoint resolves traceId to AgentRun Kafka ids", async () => {
  const fakeKafka = createFakeKafkaFactory();
  const traceStore = {
    snapshot: () => ({
      traceId: "trc_resolved_kafka_debug",
      events: [
        { type: "backend", runId: "run_resolved_kafka_debug", commandId: "cmd_resolved_kafka_debug", sessionId: "ses_agentrun_resolved_kafka_debug" }
      ],
      lastEvent: null
    })
  };
  const env = {
    HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092",
    HWLAB_KAFKA_CLIENT_ID: "test-hwlab-cloud-api",
    HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1"
  };
  const server = createServer((request, response) => {
    const url = new URL(request.url ?? "/", "http://127.0.0.1");
    void handleWorkbenchKafkaSseDebugHttp(request, response, url, { env, kafkaFactory: fakeKafka.factory, traceStore, accessController: fakeAccessController(), logger: null });
  });
  await listen(server);
  const abort = new AbortController();
  try {
    const response = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab&traceId=trc_resolved_kafka_debug`, { signal: abort.signal });
    assert.equal(response.status, 200);
    const reader = response.body?.getReader();
    assert.ok(reader, "SSE response must expose a readable stream");
    const connected = await readUntil(reader, "resolvedFilters");
    assert.match(connected, /trc_resolved_kafka_debug/u);
    assert.match(connected, /run_resolved_kafka_debug/u);

    await fakeKafka.emit({ eventType: "hwlab.trace.event.projected", traceId: null, sessionId: null, context: { runId: "run_other", commandId: "cmd_other" }, event: { label: "ignored" } });
    await fakeKafka.emit({ eventType: "hwlab.trace.event.projected", traceId: null, sessionId: null, context: { runId: "run_resolved_kafka_debug", commandId: "cmd_resolved_kafka_debug" }, event: { label: "resolved", status: "running" } });
    const streamed = await readUntil(reader, "run_resolved_kafka_debug");
    assert.match(streamed, /hwlab\.kafka\.event/u);
    assert.match(streamed, /resolved/u);
    assert.doesNotMatch(streamed, /run_other/u);
  } finally {
    abort.abort();
    await close(server);
  }
});

test("workbench Kafka SSE debug endpoint is admin-only", async () => {
  const fakeKafka = createFakeKafkaFactory();
  const server = createServer((request, response) => {
    const url = new URL(request.url ?? "/", "http://127.0.0.1");
    void handleWorkbenchKafkaSseDebugHttp(request, response, url, {
      env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092" },
      kafkaFactory: fakeKafka.factory,
      accessController: fakeAccessController("user"),
      logger: null
    });
  });
  await listen(server);
  try {
    const response = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab`);
    assert.equal(response.status, 403);
    assert.match(await response.text(), /admin_required/u);
    assert.equal(fakeKafka.consumerCreated, false);
  } finally {
    await close(server);
  }
});

test("workbench Kafka SSE debug endpoint reports connected only after consumer run is ready", async () => {
  const fakeKafka = createFakeKafkaFactory({ deferRun: true });
  const server = createServer((request, response) => {
    const url = new URL(request.url ?? "/", "http://127.0.0.1");
    void handleWorkbenchKafkaSseDebugHttp(request, response, url, {
      env: {
        HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092",
        HWLAB_KAFKA_CLIENT_ID: "test-hwlab-cloud-api",
        HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1"
      },
      kafkaFactory: fakeKafka.factory,
      accessController: fakeAccessController(),
      logger: null
    });
  });
  await listen(server);
  const abort = new AbortController();
  try {
    const response = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab`, { signal: abort.signal });
    const reader = response.body?.getReader();
    assert.ok(reader);
    const firstRead = reader.read();
    const early = await Promise.race([firstRead.then(() => "event"), delay(25).then(() => "pending")]);
    assert.equal(early, "pending");
    fakeKafka.releaseRun();
    const first = await firstRead;
    assert.equal(first.done, false);
    const connected = new TextDecoder().decode(first.value);
    assert.match(connected, /hwlab\.kafka\.connected/u);
    assert.match(connected, /"consumerReady":true/u);
    assert.match(connected, /"groupId":"test-hwlab-cloud-api-debug-sse-/u);
  } finally {
    abort.abort();
    fakeKafka.releaseRun();
    await close(server);
  }
});

test("workbench Kafka SSE debug endpoint stops a consumer that becomes ready after client close", async () => {
  const fakeKafka = createFakeKafkaFactory({ deferRun: true });
  const server = createServer((request, response) => {
    const url = new URL(request.url ?? "/", "http://127.0.0.1");
    void handleWorkbenchKafkaSseDebugHttp(request, response, url, {
      env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092", HWLAB_KAFKA_CLIENT_ID: "test-hwlab-cloud-api", HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1" },
      kafkaFactory: fakeKafka.factory,
      accessController: fakeAccessController(),
      logger: null
    });
  });
  await listen(server);
  const abort = new AbortController();
  try {
    const response = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab`, { signal: abort.signal });
    const reader = response.body?.getReader();
    assert.ok(reader);
    const firstRead = reader.read().catch(() => null);
    await delay(20);
    abort.abort();
    fakeKafka.releaseRun();
    await waitUntil(() => fakeKafka.stopCalls > 0);
    const first = await firstRead;
    const text = first && !first.done && first.value ? new TextDecoder().decode(first.value) : "";
    assert.doesNotMatch(text, /hwlab\.kafka\.connected/u);
    assert.equal(fakeKafka.disconnectCalls > 0, true);
  } finally {
    abort.abort();
    fakeKafka.releaseRun();
    await close(server);
  }
});

function createFakeKafkaFactory(options: { deferRun?: boolean } = {}) {
  let eachMessage: ((input: any) => Promise<void>) | null = null;
  let subscribedTopic = "hwlab.event.v1";
  let releaseRun: (() => void) | null = null;
  const runGate = options.deferRun ? new Promise<void>((resolve) => { releaseRun = resolve; }) : null;
  let groupId: string | null = null;
  let stopCalls = 0;
  let disconnectCalls = 0;
  let consumerCreated = false;
  const consumer = {
    connect: async () => undefined,
    subscribe: async (input: { topic?: string }) => { subscribedTopic = String(input.topic ?? subscribedTopic); },
    run: async (input: any) => {
      if (runGate) await runGate;
      eachMessage = input.eachMessage;
    },
    stop: async () => { stopCalls += 1; },
    disconnect: async () => { disconnectCalls += 1; }
  };
  return {
    factory: () => ({
      consumer: (input: { groupId?: string }) => {
        consumerCreated = true;
        groupId = String(input.groupId ?? "");
        return consumer;
      }
    }),
    get groupId() { return groupId; },
    get stopCalls() { return stopCalls; },
    get disconnectCalls() { return disconnectCalls; },
    get consumerCreated() { return consumerCreated; },
    releaseRun: () => releaseRun?.(),
    emit: async (value: Record<string, unknown>) => {
      assert.ok(eachMessage, "consumer.run must be called before emitting fake Kafka events");
      await eachMessage({
        topic: subscribedTopic,
        partition: 0,
        message: {
          offset: String(value.sessionId === "ses_other" ? 1 : 2),
          key: Buffer.from(String(value.sessionId ?? "unknown")),
          timestamp: "2026-07-09T18:30:00.000Z",
          value: Buffer.from(JSON.stringify(value))
        }
      });
    }
  };
}

function fakeAccessController(role: "admin" | "user" = "admin") {
  return {
    async ensureBootstrap() {},
    async authenticate() {
      return { ok: true, status: 200, actor: { id: role === "admin" ? "usr_debug_admin" : "usr_debug_user", role } };
    }
  };
}

async function delay(ms: number) {
  await new Promise<void>((resolve) => setTimeout(resolve, ms));
}

async function waitUntil(predicate: () => boolean, timeoutMs = 1000) {
  const deadline = Date.now() + timeoutMs;
  while (!predicate()) {
    if (Date.now() >= deadline) throw new Error("condition did not become true before timeout");
    await delay(10);
  }
}

async function readUntil(reader: ReadableStreamDefaultReader<Uint8Array>, pattern: string): Promise<string> {
  const decoder = new TextDecoder();
  let text = "";
  const deadline = Date.now() + 3000;
  while (!text.includes(pattern)) {
    if (Date.now() > deadline) throw new Error(`SSE stream did not include ${pattern}: ${text}`);
    const next = await reader.read();
    if (next.done) break;
    text += decoder.decode(next.value, { stream: true });
  }
  return text;
}

async function listen(server: ReturnType<typeof createServer>) {
  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
}

async function close(server: ReturnType<typeof createServer>) {
  await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
}

function serverUrl(server: ReturnType<typeof createServer>): string {
  const address = server.address();
  assert.equal(typeof address, "object");
  assert.ok(address && typeof address.port === "number");
  return `http://127.0.0.1:${address.port}`;
}