pikasTech-HWLAB/internal/cloud/workbench-kafka-session-index.test.ts

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

import { createWorkbenchKafkaSessionIndex } from "./workbench-kafka-session-index.ts";

test("process session index bootstraps once and serves repeated scoped queries from memory", async () => {
  let queryCount = 0;
  let now = 1000;
  const index = createWorkbenchKafkaSessionIndex({
    topic: "hwlab.event.v1",
    maxEvents: 20,
    maxEventsPerSession: 10,
    bootstrapTimeoutMs: 5000,
    rebuildIntervalMs: 60000,
    rebuildCooldownMs: 1000,
    scanLimit: 100,
    nowMs: () => now++,
    logger: quietLogger(),
    queryAll: async () => {
      queryCount += 1;
      return completeResult([
        record(0, "ses_a", "trc_a"),
        record(1, "ses_b", "trc_b"),
        record(2, "ses_a", "trc_a")
      ], "3");
    }
  });

  assert.equal(await index.start(), true);
  const first = index.query({ sessionId: "ses_a", partitionKey: "ses_a", limit: 10, scanLimit: 100, timeoutMs: 5000 });
  const second = index.query({ sessionId: "ses_a", partitionKey: "ses_a", limit: 10, scanLimit: 100, timeoutMs: 5000 });
  assert.equal(first.hit, true);
  assert.equal(second.hit, true);
  assert.equal(first.result.events.length, 2);
  assert.equal(first.result.scannedCount, 2);
  assert.equal(first.result.index.hit, true);
  assert.equal(first.result.index.indexedEventCount, 3);
  assert.equal(queryCount, 1);
  assert.deepEqual(index.sessionSummaries().sessions, [
    { sessionId: "ses_a", firstUserMessagePreview: "prompt ses_a", lastUserMessageAt: "2026-07-20T00:00:01.000Z", valuesRedacted: true }
  ]);

  index.observeLive(envelope("ses_a", "trc_a", 4), { topic: "hwlab.event.v1", partition: 0, offset: "3" });
  const live = index.query({ sessionId: "ses_a", limit: 10 });
  assert.equal(live.hit, true);
  assert.equal(live.result.events.length, 3);
  assert.equal(live.result.endOffsets[0].endOffset, "4");
  index.stop();
});

test("process session index falls back instead of returning an incomplete bounded bucket", async () => {
  const index = createWorkbenchKafkaSessionIndex({
    topic: "hwlab.event.v1",
    maxEvents: 10,
    maxEventsPerSession: 1,
    bootstrapTimeoutMs: 5000,
    rebuildIntervalMs: 60000,
    rebuildCooldownMs: 1000,
    scanLimit: 100,
    logger: quietLogger(),
    queryAll: async () => completeResult([
      record(0, "ses_overflow", "trc_overflow"),
      record(1, "ses_overflow", "trc_overflow")
    ], "2")
  });

  assert.equal(await index.start(), true);
  const result = index.query({ sessionId: "ses_overflow", limit: 10 });
  assert.equal(result.hit, false);
  assert.equal(result.fallbackReason, "session-capacity-exceeded");
  assert.equal(result.warning.blocking, false);
  index.stop();
});

function completeResult(events: ReturnType<typeof record>[], endOffset: string) {
  return {
    completionReason: "end-offset",
    completion: {
      complete: true,
      barrierReached: true,
      retentionStartVerified: true
    },
    reachedEndOffsets: true,
    endOffsetsAvailable: true,
    endOffsets: [{ partition: 0, startOffset: "0", endOffset }],
    events
  };
}

function record(offset: number, sessionId: string, traceId: string) {
  const value = envelope(sessionId, traceId, offset + 1);
  return {
    topic: "hwlab.event.v1",
    partition: 0,
    offset: String(offset),
    key: sessionId,
    timestamp: value.event.createdAt,
    valueSha256: `sha-${offset}`,
    value
  };
}

function envelope(sessionId: string, traceId: string, seq: number) {
  return {
    schema: "hwlab.event.v1",
    eventId: `evt_${sessionId}_${seq}`,
    sourceEventId: `src_${sessionId}_${seq}`,
    sessionId,
    traceId,
    event: {
      type: seq === 1 ? "user" : "assistant",
      text: seq === 1 ? `prompt ${sessionId}` : "reply",
      sessionId,
      traceId,
      createdAt: `2026-07-20T00:00:0${seq}.000Z`
    }
  };
}

function quietLogger() {
  return { log() {}, warn() {} };
}