Lumi/plugins/lumi_transcription/tests/verify.js

const assert = require("assert");
const crypto = require("crypto");
const fs = require("fs");
const os = require("os");
const path = require("path");
const http = require("http");
const { EventEmitter } = require("events");
const { PassThrough } = require("stream");
const Database = require("better-sqlite3");
const express = require("express");
const { WebSocket } = require("ws");
const { DeviceStore } = require("../backend/companion/device_store");
const { normalizeHost } = require("../backend/companion/device_store");
const { CompanionPackageService, safeArchivePath } = require("../backend/companion/package_service");
const { CompanionGateway } = require("../backend/companion/gateway");
const { resolveWorkerExecutable } = require("../backend/paths");
const protocol = require("../backend/companion/protocol");
const { RevisionStore } = require("../backend/config/revision_store");
const { JsonlDiagnosticLog, sanitize } = require("../backend/logs/jsonl_log");
const { ArtifactManager, sha256File } = require("../backend/models/artifact_manager");
const { BoundedQueue, SequenceTracker, RollingPcmBuffer } = require("../backend/sessions/bounded_queue");
const { SessionCoordinator } = require("../backend/sessions/session_coordinator");
const { CaptionStabilizer, LatestCaptionGate } = require("../backend/transcription/stabilizer");
const { WhisperWorkerSupervisor, WhisperCppServerProvider } = require("../backend/transcription/provider");
const plugin = require("../index");
const { createWebUpgradeRegistry } = require("../../../src/services/web-upgrades");

async function run() {
  const temp = fs.mkdtempSync(path.join(os.tmpdir(), "lumi-transcription-"));
  try {
    verifyProtocol();
    verifyPairingAndRevocation();
    verifyLocalhostTransportPolicy();
    verifyCompanionVersionOrdering();
    verifyWorkerResolution(temp);
    verifyRevisions();
    verifyQueues();
    verifyStabilization();
    await verifySessionLifecycle();
    await verifyWorkerRestart();
    await verifyNativeWorkerBoundary();
    await verifyAuthenticatedGateway();
    verifyArtifactsAndLogs(temp);
    await verifyCompanionPackage(temp);
    await verifyPluginIsolation();
    console.log("Lumi transcription verification passed: protocol, pairing, localhost-only HTTP, revocation, revisions, queues, stabilization, lifecycle, native worker boundary, paired Companion package, worker recovery, artifacts, logs, dashboard summary, and plugin isolation.");
  } finally { fs.rmSync(temp, { recursive: true, force: true }); }
}

function verifyProtocol() {
  const sessionId = crypto.randomUUID();
  const sourceUuid = crypto.randomUUID();
  const encoded = protocol.encodeAudioFrame({ session_id: sessionId, source_uuid: sourceUuid, sequence: 42, capture_timestamp_us: 987654321, active: true, muted: false, pcm: Buffer.alloc(640, 3) });
  const decoded = protocol.parseAudioFrame(encoded);
  assert.equal(decoded.session_id, sessionId);
  assert.equal(decoded.source_uuid, sourceUuid);
  assert.equal(decoded.sequence, 42);
  assert.equal(decoded.pcm.length, 640);
  assert.throws(() => protocol.parseAudioFrame(Buffer.alloc(64)), /magic/i);
  const hello = protocol.envelope("hello", { companion_version: "0.1.0", plugin_version: "0.1.0", capabilities: ["transcription.capture.v1"], audio: { codec: "pcm_s16le", sample_rate: 16000, channels: 1, bits: 16 } });
  assert.equal(protocol.validateHello(protocol.parseEnvelope(JSON.stringify(hello))).audio.codec, "pcm_s16le");
  assert.throws(() => protocol.parseEnvelope(JSON.stringify({ ...hello, version: 2 })), /unsupported/i);
  assert.throws(() => protocol.parseEnvelope(Buffer.alloc(protocol.MAX_JSON_BYTES + 1)), /size/i);
}

function verifyPairingAndRevocation() {
  const db = new Database(":memory:");
  let now = 1000;
  let randomSeed = 6;
  const store = new DeviceStore(db, { now: () => now, randomBytes: (size) => Buffer.alloc(size, ++randomSeed) });
  const pairing = store.issuePairing({ userId: "admin", host: "https://lumi.example" });
  const issued = store.exchange({ token: pairing.token, device: { install_id: "install", name: "Stream PC", companion_version: "0.1.0" } });
  assert.equal(store.authenticate(`LumiDevice ${issued.device_id}.${issued.device_secret}`, "transcription.capture.v1").allowed, true);
  assert.throws(() => store.exchange({ token: pairing.token, device: {} }), (error) => error.code === "PAIRING_ALREADY_USED");
  assert.deepEqual(store.setCapabilities(issued.device_id, ["transcription.settings.v1", "made.up"]), ["transcription.settings.v1"]);
  assert.equal(store.authenticate(`LumiDevice ${issued.device_id}.${issued.device_secret}`, "transcription.capture.v1").reason, "capability_revoked");
  assert.equal(store.revoke(issued.device_id), true);
  assert.equal(store.authenticate(`LumiDevice ${issued.device_id}.${issued.device_secret}`).reason, "device_revoked");
  const expired = store.issuePairing({ userId: "admin", host: "https://lumi.example", ttlMs: 50 });
  now += 51;
  assert.throws(() => store.exchange({ token: expired.token, device: {} }), (error) => error.code === "PAIRING_INVALID");
  db.close();
}

function verifyLocalhostTransportPolicy() {
  assert.equal(normalizeHost("http://localhost:3000/path"), "http://localhost:3000");
  assert.equal(normalizeHost("http://127.0.0.1:3000"), "http://127.0.0.1:3000");
  assert.throws(() => normalizeHost("http://lumi.example"), /localhost|HTTPS/i);
  const db = new Database(":memory:");
  const store = new DeviceStore(db);
  const local = store.issuePairing({ userId: "admin", host: "http://localhost:3000" });
  assert.equal(store.pairingAllowsHttp(local.token, "http://localhost:3000"), true);
  assert.equal(store.pairingAllowsHttp(local.token, "http://127.0.0.1:3000"), false);
  assert.equal(store.pairingAllowsHttp(local.token, "http://localhost:3001"), false);
  store.exchange({ token: local.token, device: {} });
  assert.equal(store.pairingAllowsHttp(local.token, "http://localhost:3000"), false);
  db.close();
}

function verifyCompanionVersionOrdering() {
  assert.equal(plugin.compareVersions("0.1.0-experimental.3", "0.1.0-experimental.2"), 1);
  assert.equal(plugin.compareVersions("0.1.0-experimental.2", "0.1.0-experimental.2"), 0);
  assert.equal(plugin.compareVersions("0.1.0", "0.1.0-experimental.9"), 1);
  assert.equal(plugin.compareVersions("0.1.1", "0.1.0"), 1);
}

function verifyWorkerResolution(temp) {
  const executable = path.join(temp, process.platform === "win32" ? "worker.exe" : "worker");
  fs.writeFileSync(executable, "worker");
  assert.equal(resolveWorkerExecutable(executable), path.resolve(executable));
  assert.throws(() => new WhisperWorkerSupervisor({ executable: "" }).start(), (error) => error.code === "WORKER_NOT_CONFIGURED");
}

function verifyRevisions() {
  const db = new Database(":memory:");
  const store = new RevisionStore(db, { now: () => 1234 });
  const first = store.apply([{ key: "selected_model_id", value: "small.en", base_revision: 0 }], "admin");
  assert.equal(first.applied[0].revision, 1);
  const merged = store.apply([{ key: "caption_max_chars", value: 80, base_revision: 0 }], "companion");
  assert.equal(merged.conflicts.length, 0);
  const conflict = store.apply([{ key: "selected_model_id", value: "base.en", base_revision: 0 }], "companion");
  assert.equal(conflict.conflicts[0].server.value, "small.en");
  assert.equal(store.list().selected_model_id.value, "small.en");
  db.close();
}

function verifyQueues() {
  let now = 0;
  const queue = new BoundedQueue({ maxItems: 2, maxBytes: 8, maxAgeMs: 50, now: () => now });
  queue.push(Buffer.alloc(4), { capturedAt: now }); queue.push(Buffer.alloc(4), { capturedAt: now }); queue.push(Buffer.alloc(4), { capturedAt: now });
  assert.equal(queue.metrics().dropped.capacity, 1);
  now = 51; assert.equal(queue.size(), 0); assert.equal(queue.metrics().dropped.stale, 2);
  const sequence = new SequenceTracker();
  assert.equal(sequence.accept(5).accepted, true); assert.equal(sequence.accept(8).gap, 2); assert.equal(sequence.accept(7).accepted, false);
  const rolling = new RollingPcmBuffer({ seconds: 1, now: () => now });
  rolling.push(Buffer.alloc(20000), now); rolling.push(Buffer.alloc(20000), now);
  assert.equal(rolling.snapshot().length, 20000);
}

function verifyStabilization() {
  let now = 0;
  const stabilizer = new CaptionStabilizer({ now: () => now, fragmentAfterMs: 1000 });
  const source = crypto.randomUUID();
  stabilizer.update(source, "hello wor");
  now += 600;
  const second = stabilizer.update(source, "hello world");
  assert.equal(second.stable_text, "hello");
  now += 600;
  const third = stabilizer.update(source, "hello world again");
  assert.equal(third.stable_text, "hello world");
  const fragmentSource = crypto.randomUUID();
  stabilizer.update(fragmentSource, "extraord", { trailingIncomplete: true });
  now += 1001;
  const fragment = stabilizer.update(fragmentSource, "extraord", { trailingIncomplete: true });
  assert.equal(fragment.incomplete_word, true);
  assert.equal(fragment.stable_text, "");
  const gate = new LatestCaptionGate();
  assert.equal(gate.accept({ session_id: source, caption_id: fragment.caption_id, revision: 2 }), true);
  assert.equal(gate.accept({ session_id: source, caption_id: fragment.caption_id, revision: 1 }), false);
}

async function verifySessionLifecycle() {
  class Provider extends EventEmitter {
    constructor() { super(); this.audio = []; this.stops = 0; }
    async health() { return { healthy: true }; }
    async startSession() {} async addTrack() {} async pushAudio(_session, _track, frame) { this.audio.push(frame); return true; }
    async stopSession() { this.stops += 1; }
  }
  const provider = new Provider();
  const delivered = [];
  const deliveryFactory = () => ({ start: async () => {}, stop: async () => {}, pause: async () => {}, resume: async () => {}, deliver: async (event) => { delivered.push(event); return { disposition: "simulated" }; } });
  const coordinator = new SessionCoordinator({ provider, deliveryFactory, graceMs: 25 });
  const { session } = coordinator.create({ id: "device" }, () => {});
  const primary = crypto.randomUUID();
  coordinator.updateSource(session.id, { source_uuid: primary, display_name: "Mic", primary: true, program_active: true });
  await assert.rejects(coordinator.start(session.id, { mode: "live" }), (error) => error.code === "OBS_NOT_STREAMING");
  await coordinator.updateObsState(session.id, { streaming: true, auto_start: false });
  await coordinator.start(session.id, { mode: "live" });
  const frame = protocol.parseAudioFrame(protocol.encodeAudioFrame({ session_id: session.id, source_uuid: primary, sequence: 1, capture_timestamp_us: 1, pcm: Buffer.alloc(640) }));
  assert.equal((await coordinator.audio(session.id, frame)).accepted, true);
  provider.emit("hypothesis", { session_id: session.id, track_id: primary, text: "hello", final: true, model_id: "small.en", backend: "cuda" });
  await new Promise((resolve) => setImmediate(resolve));
  assert.equal(delivered[0].stable_text, "hello");
  await coordinator.updateObsState(session.id, { streaming: false });
  assert.equal(coordinator.status(session.id).state, "grace");
  await new Promise((resolve) => setTimeout(resolve, 40));
  assert.equal(coordinator.status(session.id).state, "idle");
  assert.equal(provider.stops, 1);
  await coordinator.close();
}

async function verifyWorkerRestart() {
  const children = [];
  const fakeSpawn = () => {
    const child = new EventEmitter();
    child.stdin = new PassThrough(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.kill = () => child.emit("exit", null, "SIGTERM");
    children.push(child); return child;
  };
  const supervisor = new WhisperWorkerSupervisor({ executable: "fake-worker", spawn: fakeSpawn, maxRestarts: 1 });
  supervisor.on("error", () => {});
  supervisor.start();
  assert.equal(supervisor.send({ type: "audio", captured_at: Date.now() }, Buffer.alloc(640)), true);
  children[0].emit("exit", 1, null);
  await new Promise((resolve) => setTimeout(resolve, 320));
  assert.equal(children.length, 2);
  supervisor.stopping = true; children[1].emit("exit", 0, null);
}

async function verifyNativeWorkerBoundary() {
  class Worker extends EventEmitter {
    start() { this.started = true; }
    send(message) { if (message.type === "load_model") setImmediate(() => this.emit("message", { type: "model_loaded", model_id: message.model.id, backend: "cpu" })); return true; }
    health() { return { healthy: true, state: "running" }; }
    async stop() {}
  }
  const worker = new Worker();
  const provider = new WhisperCppServerProvider(worker);
  const health = await provider.loadModel({ id: "base.en", path: "verified-model.bin" });
  assert.equal(worker.started, true);
  assert.equal(health.model, "base.en");
  const sourceRoot = path.join(__dirname, "../backend/transcription/worker-native");
  const cmake = fs.readFileSync(path.join(sourceRoot, "CMakeLists.txt"), "utf8");
  const source = fs.readFileSync(path.join(sourceRoot, "src/main.cpp"), "utf8");
  assert.match(cmake, /f049fff95a089aa9969deb009cdd4892b3e74916/);
  assert.match(source, /max_samples = sample_rate \* 6/);
  assert.doesNotMatch(source, /ofstream|fwrite|WriteAllBytes/);
}

async function verifyAuthenticatedGateway() {
  const db = new Database(":memory:");
  const devices = new DeviceStore(db);
  const sessionId = crypto.randomUUID();
  let disconnected = false;
  const sessions = {
    create: (_device, send) => ({ session: { id: sessionId, state: "idle", send }, resumed: false }),
    disconnect: () => { disconnected = true; },
    audio: async () => ({ accepted: true }), updateSource: () => ({}), updateObsState: async () => ({}), start: async () => ({}), stop: async () => ({})
  };
  const gateway = new CompanionGateway({ devices, sessions });
  const registry = createWebUpgradeRegistry();
  registry.add("/plugins/lumi_transcription/live", (request, socket, head) => gateway.upgrade(request, socket, head));
  const server = http.createServer((_req, res) => res.end("ok"));
  registry.attach(server);
  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
  const pairing = devices.issuePairing({ userId: "admin", host: `http://127.0.0.1:${server.address().port}` });
  const credential = devices.exchange({ token: pairing.token, device: { name: "Stream PC" } });
  const client = new WebSocket(`ws://127.0.0.1:${server.address().port}/plugins/lumi_transcription/live`, {
    headers: { Authorization: `LumiDevice ${credential.device_id}.${credential.device_secret}` }
  });
  await new Promise((resolve, reject) => { client.once("open", resolve); client.once("error", reject); });
  client.send(JSON.stringify(protocol.envelope("hello", { companion_version: "0.1.0", plugin_version: "0.1.0", capabilities: ["transcription.capture.v1"], audio: { codec: "pcm_s16le", sample_rate: 16000, channels: 1, bits: 16 } })));
  const response = await new Promise((resolve, reject) => { client.once("message", (data) => resolve(JSON.parse(String(data)))); client.once("error", reject); });
  assert.equal(response.type, "hello_ack");
  assert.equal(response.session_id, sessionId);
  await new Promise((resolve) => { client.once("close", resolve); client.close(); });
  for (let attempt = 0; attempt < 20 && !disconnected; attempt += 1) await new Promise((resolve) => setTimeout(resolve, 5));
  assert.equal(disconnected, true);
  const productionPairing = devices.issuePairing({ userId: "admin", host: "https://lumi.example" });
  const productionCredential = devices.exchange({ token: productionPairing.token, device: { name: "Production Stream PC" } });
  const insecureProductionClient = new WebSocket(`ws://127.0.0.1:${server.address().port}/plugins/lumi_transcription/live`, {
    headers: { Authorization: `LumiDevice ${productionCredential.device_id}.${productionCredential.device_secret}` }
  });
  const rejectionStatus = await new Promise((resolve, reject) => {
    insecureProductionClient.once("unexpected-response", (_request, response) => { response.resume(); resolve(response.statusCode); });
    insecureProductionClient.once("open", () => reject(new Error("A production credential connected over insecure WebSocket.")));
    insecureProductionClient.once("error", (error) => { if (!String(error.message).includes("Unexpected server response")) reject(error); });
  });
  assert.equal(rejectionStatus, 426);
  await gateway.close(); registry.close();
  await new Promise((resolve) => server.close(resolve));
  db.close();
}

function verifyArtifactsAndLogs(temp) {
  const artifactRoot = path.join(temp, "artifacts"); fs.mkdirSync(artifactRoot);
  const file = path.join(artifactRoot, "model.bin"); fs.writeFileSync(file, "verified");
  const entry = { id: "model", filename: "model.bin", url: "https://example.invalid/model.bin", sha256: sha256File(file) };
  assert.equal(new ArtifactManager(artifactRoot).status(entry).valid, true);
  assert.equal(sanitize({ pcm: Buffer.alloc(10), device_secret: "secret", stable_text: "hello" }, false).pcm, "[redacted]");
  const logsRoot = path.join(temp, "logs");
  const logs = new JsonlDiagnosticLog(logsRoot, { retentionDays: 1, maxBytes: 100 });
  logs.append({ kind: "caption", stable_text: "hello", pcm: Buffer.alloc(10) });
  const old = path.join(logsRoot, "transcription-2000-01-01.jsonl"); fs.writeFileSync(old, `${"x".repeat(150)}\n`); fs.utimesSync(old, new Date(0), new Date(0));
  assert.ok(logs.cleanup().removed >= 1);
  assert.equal(fs.readdirSync(logsRoot).some((name) => name.endsWith(".wav") || name.endsWith(".pcm")), false);
}

async function verifyCompanionPackage(temp) {
  const root = path.join(temp, "companion-package");
  fs.mkdirSync(root, { recursive: true });
  const sourcePath = path.join(root, "companion.zip");
  const zip = new (require("adm-zip"))();
  zip.addFile("Lumi.Companion.App.exe", Buffer.from("portable-app"));
  zip.writeZip(sourcePath);
  const manifest = { version: "test", artifacts: [{
    id: "windows-x64", platform: "win32", architecture: "x64", filename: "companion.zip",
    entrypoint: "Lumi.Companion.App.exe", url: "https://example.invalid/companion.zip", sha256: sha256File(sourcePath), bytes: fs.statSync(sourcePath).size
  }] };
  const service = new CompanionPackageService(root, manifest);
  const bundle = await service.build({ pairing_id: "pairing", bootstrap: { format: "lumi-companion-bootstrap-v1", token: "single-use" } });
  const output = new (require("adm-zip"))(bundle.buffer);
  assert.equal(output.readAsText("Lumi.Companion.App.exe"), "portable-app");
  assert.match(output.readAsText("lumi-companion-pairing.lumi-pairing.json"), /single-use/);
  assert.match(output.readAsText("START-HERE.txt"), /works once/i);
  const manifestPath = path.join(root, "manifest.json");
  fs.writeFileSync(manifestPath, JSON.stringify(manifest));
  const liveManifestService = new CompanionPackageService(root, manifestPath);
  assert.equal(liveManifestService.status().version, "test");
  fs.writeFileSync(manifestPath, JSON.stringify({ ...manifest, version: "refreshed" }));
  assert.equal(liveManifestService.status().version, "refreshed");
  assert.throws(() => safeArchivePath("../escape.exe"), /unsafe path/i);
}

async function verifyPluginIsolation() {
  const db = new Database(":memory:");
  const mounts = [];
  let upgradeRemoved = false;
  const cleanup = plugin.init({
    db, logger: { info() {} },
    web: {
      createRouter: () => express.Router(),
      addUpgradeHandler: () => () => { upgradeRemoved = true; },
      emitEvent() {},
      mount: (mountPath) => mounts.push(mountPath)
    }
  });
  assert.deepEqual(mounts, ["/plugins/lumi_transcription"]);
  assert.equal(typeof global.lumiFrameworks.transcription.health, "function");
  const dashboard = await global.lumiFrameworks.transcription.dashboardSummary();
  assert.equal(dashboard.title, "Lumi Companion");
  assert.equal(dashboard.metrics.find((entry) => entry.label === "Connection").value, "Offline");
  await cleanup();
  assert.equal(upgradeRemoved, true);
  assert.equal(global.lumiFrameworks.transcription, undefined);
  db.close();
}

run().catch((error) => { console.error(error); process.exit(1); });