import { afterAll, describe, expect, test } from "bun:test";
import { mkdtemp, readdir, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import {
  createSidecar,
  createTranscribeHandlers,
  parseLanguageAndTask,
  type Sidecar,
  type SidecarSegment,
  type SidecarTranscribeRequest,
} from "../src/server/transcribe";
import { isTranscribeStreamLine, type TranscribeStreamLine } from "../src/lib/engine";
import type { SessionUser } from "../src/server/auth";

// M5 server engine: HTTP handlers against a fake sidecar, and the real
// Bun.spawn supervisor against a protocol-faithful Python stand-in
// (tests/fixtures/fake_sidecar.py) — no faster-whisper needed. The end-to-end
// run with real inference is manual/opt-in: see tests/server-engine.md.

const FIXTURE_SIDECAR = join(import.meta.dir, "fixtures/fake_sidecar.py");
const REAL_SIDECAR = join(import.meta.dir, "../server-py/sidecar.py");

const tempDirs: string[] = [];
async function makeTempDir(): Promise<string> {
  const dir = await mkdtemp(join(tmpdir(), "transcribe-test-"));
  tempDirs.push(dir);
  return dir;
}
afterAll(async () => {
  await Promise.all(tempDirs.map((dir) => rm(dir, { recursive: true, force: true })));
});

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

const USER: SessionUser = { id: 1, username: "tester" };
const sessionFromCookie = (request: Request): SessionUser | null =>
  request.headers.get("cookie") === "sid=ok" ? USER : null;

interface FakeSidecar extends Sidecar {
  requests: SidecarTranscribeRequest[];
  /** Bytes of each request's audio file, captured before cleanup. */
  audioPayloads: Uint8Array[];
}

function fakeSidecar(
  behavior: {
    available?: boolean;
    reason?: string;
    segments?: SidecarSegment[];
    error?: string;
    waitForAbort?: boolean;
    /** Quiet time before the first segment (model load, VAD, first batch). */
    delayMs?: number;
  } = {},
): FakeSidecar {
  const segments = behavior.segments ?? [
    { start: 0, end: 1.5, text: " Hello" },
    { start: 1.5, end: 3, text: " world." },
  ];
  const sidecar: FakeSidecar = {
    requests: [],
    audioPayloads: [],
    async status() {
      return behavior.available === false
        ? { available: false, reason: behavior.reason ?? "nope" }
        : { available: true };
    },
    async transcribe(request, onSegment, signal) {
      sidecar.requests.push(request);
      sidecar.audioPayloads.push(
        new Uint8Array(await Bun.file(request.audioPath).arrayBuffer()),
      );
      if (behavior.error) throw new Error(behavior.error);
      if (behavior.delayMs) await Bun.sleep(behavior.delayMs);
      onSegment(segments[0]!);
      if (behavior.waitForAbort) {
        await new Promise<void>((resolve) => {
          if (signal?.aborted) return resolve();
          signal?.addEventListener("abort", () => resolve(), { once: true });
        });
        return { cancelled: true };
      }
      for (const segment of segments.slice(1)) onSegment(segment);
      return { cancelled: false };
    },
  };
  return sidecar;
}

function pcmBody(samples: number[]): Uint8Array<ArrayBuffer> {
  const audio = new Float32Array(samples);
  return new Uint8Array(audio.buffer as ArrayBuffer);
}

function postRequest(
  query: string,
  body: BodyInit | null,
  init: { cookie?: string; signal?: AbortSignal; contentLength?: boolean } = {},
): Request {
  const headers: Record<string, string> = {
    "Content-Type": "application/octet-stream",
  };
  if (init.cookie !== undefined) headers.cookie = init.cookie;
  return new Request(`http://localhost/api/transcribe${query}`, {
    method: "POST",
    headers,
    body,
    signal: init.signal,
  });
}

async function readStreamLines(response: Response): Promise<TranscribeStreamLine[]> {
  const text = await response.text();
  return text
    .split("\n")
    .filter((line) => line.trim().length > 0)
    .map((line) => {
      const parsed: unknown = JSON.parse(line);
      if (!isTranscribeStreamLine(parsed)) throw new Error(`bad line: ${line}`);
      return parsed;
    });
}

// ---------------------------------------------------------------------------
// HTTP handlers (fake sidecar)
// ---------------------------------------------------------------------------

// ---------------------------------------------------------------------------
// parseLanguageAndTask — shared with the WS live route (src/server/live.ts,
// src/index.ts) so it can't skip the validation the HTTP route enforces.
// ---------------------------------------------------------------------------

describe("parseLanguageAndTask", () => {
  test("defaults to transcribe with no language", () => {
    expect(parseLanguageAndTask(new URL("http://localhost/x"))).toEqual({
      language: undefined,
      task: "transcribe",
    });
  });

  test("accepts a valid language and task", () => {
    expect(parseLanguageAndTask(new URL("http://localhost/x?language=da&task=translate"))).toEqual(
      { language: "da", task: "translate" },
    );
  });

  test('"auto"/empty language means no language override', () => {
    expect(parseLanguageAndTask(new URL("http://localhost/x?language=auto"))).toEqual({
      language: undefined,
      task: "transcribe",
    });
    expect(parseLanguageAndTask(new URL("http://localhost/x?language="))).toEqual({
      language: undefined,
      task: "transcribe",
    });
  });

  test("rejects an invalid language or task", () => {
    expect(parseLanguageAndTask(new URL("http://localhost/x?language=Danish!"))).toBe(
      "invalid language: Danish!",
    );
    expect(parseLanguageAndTask(new URL("http://localhost/x?task=summarize"))).toBe(
      "invalid task: summarize",
    );
  });
});

describe("GET /api/transcribe/status", () => {
  test("401 without a session", async () => {
    const handlers = createTranscribeHandlers({
      sidecar: fakeSidecar(),
      getSessionUser: sessionFromCookie,
    });
    const response = await handlers.status(
      new Request("http://localhost/api/transcribe/status"),
    );
    expect(response.status).toBe(401);
  });

  test("passes the sidecar's availability through", async () => {
    const handlers = createTranscribeHandlers({
      sidecar: fakeSidecar({ available: false, reason: "python3 missing" }),
      getSessionUser: sessionFromCookie,
    });
    const response = await handlers.status(
      new Request("http://localhost/api/transcribe/status", {
        headers: { cookie: "sid=ok" },
      }),
    );
    expect(response.status).toBe(200);
    expect(await response.json()).toEqual({
      available: false,
      reason: "python3 missing",
    });
  });
});

describe("POST /api/transcribe", () => {
  test("401 without a session — the server engine is logged-in only", async () => {
    const sidecar = fakeSidecar();
    const handlers = createTranscribeHandlers({
      sidecar,
      getSessionUser: sessionFromCookie,
    });
    const response = await handlers.post(
      postRequest("?model=onnx-community/whisper-tiny.en", pcmBody([0.1])),
    );
    expect(response.status).toBe(401);
    expect(sidecar.requests).toHaveLength(0);
  });

  test("400 for missing/unknown model, bad language, bad task", async () => {
    const handlers = createTranscribeHandlers({
      sidecar: fakeSidecar(),
      getSessionUser: sessionFromCookie,
      tempDir: await makeTempDir(),
    });
    const cases = [
      "",
      "?model=someone/other-model",
      "?model=onnx-community/whisper-tiny&language=Danish!",
      "?model=onnx-community/whisper-tiny&task=summarize",
    ];
    for (const query of cases) {
      const response = await handlers.post(
        postRequest(query, pcmBody([0.1]), { cookie: "sid=ok" }),
      );
      expect(response.status).toBe(400);
    }
  });

  test("503 with the reason when the sidecar is unavailable", async () => {
    const sidecar = fakeSidecar({ available: false, reason: "faster-whisper is not installed" });
    const handlers = createTranscribeHandlers({
      sidecar,
      getSessionUser: sessionFromCookie,
    });
    const response = await handlers.post(
      postRequest("?model=onnx-community/whisper-tiny.en", pcmBody([0.1]), {
        cookie: "sid=ok",
      }),
    );
    expect(response.status).toBe(503);
    expect(await response.json()).toEqual({ error: "faster-whisper is not installed" });
    expect(sidecar.requests).toHaveLength(0);
  });

  test("413 over the upload limit; 400 for empty or non-float32 bodies", async () => {
    const tempDir = await makeTempDir();
    const handlers = createTranscribeHandlers({
      sidecar: fakeSidecar(),
      getSessionUser: sessionFromCookie,
      maxUploadBytes: 16,
      tempDir,
    });
    const over = await handlers.post(
      postRequest("?model=onnx-community/whisper-tiny.en", pcmBody([1, 2, 3, 4, 5]), {
        cookie: "sid=ok",
      }),
    );
    expect(over.status).toBe(413);

    const empty = await handlers.post(
      postRequest("?model=onnx-community/whisper-tiny.en", new Uint8Array(0), {
        cookie: "sid=ok",
      }),
    );
    expect(empty.status).toBe(400);

    const misaligned = await handlers.post(
      postRequest("?model=onnx-community/whisper-tiny.en", new Uint8Array(6), {
        cookie: "sid=ok",
      }),
    );
    expect(misaligned.status).toBe(400);

    // Rejected uploads must not leave temp files behind.
    expect(await readdir(tempDir)).toEqual([]);
  });

  test("streams segments then the result, maps the model, cleans up", async () => {
    const tempDir = await makeTempDir();
    const sidecar = fakeSidecar();
    const handlers = createTranscribeHandlers({
      sidecar,
      getSessionUser: sessionFromCookie,
      tempDir,
    });
    const audio = [0.25, -0.25, 0.5, -0.5];
    const response = await handlers.post(
      postRequest(
        "?model=onnx-community/whisper-base&language=da&task=translate",
        pcmBody(audio),
        { cookie: "sid=ok" },
      ),
    );
    expect(response.status).toBe(200);
    expect(response.headers.get("content-type")).toContain("application/x-ndjson");

    const lines = await readStreamLines(response);
    expect(lines).toEqual([
      { type: "segment", text: " Hello", start: 0, end: 1.5 },
      { type: "segment", text: " world.", start: 1.5, end: 3 },
      {
        type: "result",
        text: "Hello world.",
        chunks: [
          { text: " Hello", timestamp: [0, 1.5] },
          { text: " world.", timestamp: [1.5, 3] },
        ],
      },
    ]);

    // Catalog id → faster-whisper name (D-013), options passed through.
    expect(sidecar.requests).toEqual([
      expect.objectContaining({ model: "base", language: "da", task: "translate" }),
    ]);
    // The sidecar saw exactly the uploaded PCM bytes…
    expect([...new Float32Array(sidecar.audioPayloads[0]!.buffer.slice(0))]).toEqual(
      audio,
    );
    // …and the temp file is gone once the stream finished.
    expect(await readdir(tempDir)).toEqual([]);
  });

  test("heartbeat blank lines keep a quiet stream alive before the first segment (D-021)", async () => {
    // The batched pipeline (D-WS5a) runs VAD over the whole file and decodes a
    // full batch before its first segment, so the quiet stretch is longer.
    const handlers = createTranscribeHandlers({
      sidecar: fakeSidecar({ delayMs: 200 }),
      getSessionUser: sessionFromCookie,
      tempDir: await makeTempDir(),
      heartbeatIntervalMs: 20,
    });
    const response = await handlers.post(
      postRequest("?model=onnx-community/whisper-base", pcmBody([0.1]), { cookie: "sid=ok" }),
    );
    const text = await response.text();
    const beforeFirstSegment = text.slice(0, text.indexOf("{"));
    expect(beforeFirstSegment.length).toBeGreaterThanOrEqual(2);
    expect(beforeFirstSegment.trim()).toBe("");
    expect(text).toContain('"type":"result"');
  });

  test("sidecar errors travel in-stream after the 200 header", async () => {
    const tempDir = await makeTempDir();
    const handlers = createTranscribeHandlers({
      sidecar: fakeSidecar({ error: "model load failed" }),
      getSessionUser: sessionFromCookie,
      tempDir,
    });
    const response = await handlers.post(
      postRequest("?model=onnx-community/whisper-tiny.en", pcmBody([0.1]), {
        cookie: "sid=ok",
      }),
    );
    expect(response.status).toBe(200);
    const lines = await readStreamLines(response);
    expect(lines).toEqual([{ type: "error", message: "model load failed" }]);
    expect(await readdir(tempDir)).toEqual([]);
  });

  test("an aborted request cancels the sidecar run", async () => {
    const tempDir = await makeTempDir();
    const sidecar = fakeSidecar({ waitForAbort: true });
    const handlers = createTranscribeHandlers({
      sidecar,
      getSessionUser: sessionFromCookie,
      tempDir,
    });
    const controller = new AbortController();
    const response = await handlers.post(
      postRequest("?model=onnx-community/whisper-tiny.en", pcmBody([0.1]), {
        cookie: "sid=ok",
        signal: controller.signal,
      }),
    );
    expect(response.status).toBe(200);
    controller.abort();
    // Cancelled runs still terminate the stream with the partial result.
    const lines = await readStreamLines(response);
    expect(lines.at(-1)?.type).toBe("result");
    expect(await readdir(tempDir)).toEqual([]);
  });
});

// ---------------------------------------------------------------------------
// createSidecar — the real Bun.spawn supervisor against a Python child
// ---------------------------------------------------------------------------

describe("createSidecar (fake python sidecar)", () => {
  test("handshake, segment streaming, and request/option pass-through", async () => {
    const sidecar = createSidecar(["python3", FIXTURE_SIDECAR]);
    expect(await sidecar.status()).toEqual({ available: true });

    const audioPath = join(await makeTempDir(), "audio.f32");
    await Bun.write(audioPath, pcmBody([0.1, 0.2, 0.3]));

    const segments: SidecarSegment[] = [];
    const outcome = await sidecar.transcribe(
      { model: "ok", audioPath, language: "da", task: "translate" },
      (segment) => segments.push(segment),
    );
    expect(outcome).toEqual({ cancelled: false });
    expect(segments).toHaveLength(2);
    expect(segments[0]).toEqual({ start: 0, end: 1.5, text: " Hello" });
    // The fixture echoes what it received (audio bytes, language, task, mode).
    expect(segments[1]!.text).toContain("model=ok bytes=12 lang=da task=translate mode=file");

    const live: SidecarSegment[] = [];
    await sidecar.transcribe({ model: "ok", audioPath, mode: "live" }, (s) => live.push(s));
    expect(live[1]!.text).toContain("mode=live");
  });

  test("sidecar 'error' lines reject that request only", async () => {
    const sidecar = createSidecar(["python3", FIXTURE_SIDECAR]);
    const audioPath = join(await makeTempDir(), "audio.f32");
    await Bun.write(audioPath, pcmBody([0.1]));

    expect(
      sidecar.transcribe({ model: "boom", audioPath }, () => {}),
    ).rejects.toThrow("kaboom");
    // The process survived; the next request works.
    const outcome = await sidecar.transcribe({ model: "ok", audioPath }, () => {});
    expect(outcome).toEqual({ cancelled: false });
  });

  test("a crash rejects in-flight requests and the next request respawns", async () => {
    const sidecar = createSidecar(["python3", FIXTURE_SIDECAR]);
    const audioPath = join(await makeTempDir(), "audio.f32");
    await Bun.write(audioPath, pcmBody([0.1]));

    expect(
      sidecar.transcribe({ model: "die", audioPath }, () => {}),
    ).rejects.toThrow(/exited unexpectedly/);
    // Supervision: a fresh process serves the next request.
    const segments: SidecarSegment[] = [];
    const outcome = await sidecar.transcribe(
      { model: "ok", audioPath },
      (segment) => segments.push(segment),
    );
    expect(outcome).toEqual({ cancelled: false });
    expect(segments).toHaveLength(2);
  });

  test("abort sends a cancel and resolves {cancelled: true}", async () => {
    const sidecar = createSidecar(["python3", FIXTURE_SIDECAR]);
    const audioPath = join(await makeTempDir(), "audio.f32");
    await Bun.write(audioPath, pcmBody([0.1]));

    const controller = new AbortController();
    const outcome = await sidecar.transcribe(
      { model: "wait-for-cancel", audioPath },
      () => controller.abort(), // abort as soon as the first segment arrives
      controller.signal,
    );
    expect(outcome).toEqual({ cancelled: true });
  });
});

describe("createSidecar (degradation, D-013)", () => {
  test("the real sidecar without faster-whisper reports a permanent reason", async () => {
    // -S/-I keep site-packages off sys.path, so this behaves like a
    // Bun-only deployment even on machines that have faster-whisper.
    const sidecar = createSidecar(["python3", "-S", "-I", REAL_SIDECAR]);
    const first = await sidecar.status();
    expect(first.available).toBe(false);
    expect(first.reason).toContain("faster-whisper is not installed");
    // Remembered as permanent — no respawn storm, same answer immediately.
    const second = await sidecar.status();
    expect(second).toEqual(first);
  });

  test("a missing python binary reports unavailable", async () => {
    const sidecar = createSidecar(["definitely-not-a-python-binary"]);
    const status = await sidecar.status();
    expect(status.available).toBe(false);
  });
});

// ---------------------------------------------------------------------------
// The real server-py/sidecar.py against a stubbed faster_whisper/numpy
// (tests/fixtures/stub_fw): request routing, model cache, cancel (D-WS5a).
// `-S` keeps real site-packages out, so this runs the same everywhere.
// ---------------------------------------------------------------------------

describe("server-py/sidecar.py (stubbed faster-whisper, D-WS5a)", () => {
  const STUB_DIR = join(import.meta.dir, "fixtures/stub_fw");
  const stubbedSidecar = () =>
    createSidecar(["env", `PYTHONPATH=${STUB_DIR}`, "python3", "-S", REAL_SIDECAR]);

  async function run(
    sidecar: Sidecar,
    request: Omit<SidecarTranscribeRequest, "audioPath">,
    signal?: AbortSignal,
    onSegment: (segment: SidecarSegment) => void = () => {},
  ): Promise<{ texts: string[]; cancelled: boolean }> {
    const audioPath = join(await makeTempDir(), "audio.f32");
    await Bun.write(audioPath, pcmBody([0, 0.5, -0.5, 0]));
    const texts: string[] = [];
    const outcome = await sidecar.transcribe(
      { ...request, audioPath },
      (segment) => {
        texts.push(segment.text);
        onSegment(segment);
      },
      signal,
    );
    return { texts, cancelled: outcome.cancelled };
  }

  test("file mode (and no mode) runs the batched pipeline with VAD, batch_size=8", async () => {
    const sidecar = stubbedSidecar();
    const explicit = await run(sidecar, { model: "m1", mode: "file", language: "da" });
    expect(explicit.texts).toHaveLength(1);
    expect(explicit.texts[0]).toContain("batched model=m1");
    expect(explicit.texts[0]).toContain("vad_filter=True");
    expect(explicit.texts[0]).toContain("batch_size=8");
    // Tuned Silero VAD: the default (0.5 / 400 ms) dropped quiet speech (D-WS5a).
    expect(explicit.texts[0]).toContain("vad_parameters={'threshold': 0.2, 'speech_pad_ms': 800}");
    expect(explicit.texts[0]).toContain("language=da");
    expect(explicit.texts[0]).toContain("task=transcribe");

    const implicit = await run(sidecar, { model: "m1" });
    expect(implicit.texts[0]).toContain("batched model=m1");
  });

  test("live mode stays sequential without VAD (short passes, D-WS5a)", async () => {
    const sidecar = stubbedSidecar();
    const { texts } = await run(sidecar, { model: "m1", mode: "live", task: "translate" });
    expect(texts[0]).toContain("seq model=m1");
    expect(texts[0]).not.toContain("vad_filter");
    expect(texts[0]).not.toContain("batch_size");
    expect(texts[0]).toContain("task=translate");
  });

  test("one model stays loaded across modes; a switch reloads and rewraps", async () => {
    const sidecar = stubbedSidecar();
    expect((await run(sidecar, { model: "m1" })).texts[0]).toContain("loads=1");
    expect((await run(sidecar, { model: "m1", mode: "live" })).texts[0]).toContain("loads=1");
    const switched = (await run(sidecar, { model: "m2" })).texts[0];
    expect(switched).toContain("batched model=m2");
    expect(switched).toContain("loads=2");
  });

  test("cancel lands between batched segments", async () => {
    const sidecar = stubbedSidecar();
    const controller = new AbortController();
    const { texts, cancelled } = await run(sidecar, { model: "slow" }, controller.signal, () =>
      controller.abort(),
    );
    expect(cancelled).toBe(true);
    expect(texts.length).toBeGreaterThanOrEqual(1);
    expect(texts.length).toBeLessThan(20);
  });

  test("audio with no speech finishes with zero segments, not an error", async () => {
    const sidecar = stubbedSidecar();
    const { texts, cancelled } = await run(sidecar, { model: "silence" });
    expect(texts).toEqual([]);
    expect(cancelled).toBe(false);
  });
});

// ---------------------------------------------------------------------------
// Opt-in: real faster-whisper end to end (needs python3 + requirements.txt
// installed and network/cached weights). See tests/server-engine.md.
//   FW_INTEGRATION=1 bun test tests/transcribe.test.ts
// ---------------------------------------------------------------------------

describe.skipIf(!process.env.FW_INTEGRATION)("real faster-whisper sidecar", () => {
  test(
    "transcribes PCM through the real sidecar",
    async () => {
      const sidecar = createSidecar();
      const status = await sidecar.status();
      expect(status).toEqual({ available: true });

      // One second of a 440 Hz tone — not speech, but exercises model load,
      // numpy decode, and the full protocol round-trip.
      const samples = new Float32Array(16_000);
      for (let i = 0; i < samples.length; i++) {
        samples[i] = 0.1 * Math.sin((2 * Math.PI * 440 * i) / 16_000);
      }
      const audioPath = join(await makeTempDir(), "tone.f32");
      await Bun.write(audioPath, new Uint8Array(samples.buffer));

      const outcome = await sidecar.transcribe(
        { model: "tiny.en", audioPath },
        () => {},
      );
      expect(outcome.cancelled).toBe(false);
    },
    { timeout: 300_000 },
  );

  test(
    "file mode (batched + VAD) and live mode both transcribe real speech (D-WS5a)",
    async () => {
      const sidecar = createSidecar();
      // First 60 s of en_alien as raw float32 16 kHz mono — what the browser uploads.
      const clip = join(import.meta.dir, "audio_files/en_alien.opus");
      const pcm =
        await Bun.$`ffmpeg -v error -i ${clip} -t 60 -f f32le -ac 1 -ar 16000 -`.arrayBuffer();
      const audioPath = join(await makeTempDir(), "alien.f32");
      await Bun.write(audioPath, pcm);

      for (const mode of ["file", "live"] as const) {
        const segments: SidecarSegment[] = [];
        const outcome = await sidecar.transcribe(
          { model: "tiny.en", audioPath, language: "en", mode },
          (segment) => segments.push(segment),
        );
        expect(outcome.cancelled).toBe(false);
        expect(segments.length).toBeGreaterThan(0);
        const text = segments.map((segment) => segment.text).join("").toLowerCase();
        expect(text).toContain("voices of the void");
        // Timestamps stay in source-audio time after VAD drops silence.
        expect(segments.at(-1)!.end).toBeLessThanOrEqual(61);
      }
    },
    { timeout: 300_000 },
  );

  // The whole 15-min en_alien clip (≈ 17 VAD chunks ⇒ 3 batches) as uploaded PCM.
  async function wholeAlienPcm(): Promise<string> {
    const clip = join(import.meta.dir, "audio_files/en_alien.opus");
    const pcm = await Bun.$`ffmpeg -v error -i ${clip} -f f32le -ac 1 -ar 16000 -`.arrayBuffer();
    const audioPath = join(await makeTempDir(), "alien-full.f32");
    await Bun.write(audioPath, pcm);
    return audioPath;
  }

  test(
    "a long file streams in bursts behind heartbeats over HTTP (D-021, D-WS5a)",
    async () => {
      const handlers = createTranscribeHandlers({
        sidecar: createSidecar(),
        getSessionUser: sessionFromCookie,
        tempDir: await makeTempDir(),
        heartbeatIntervalMs: 100,
      });
      const body = await Bun.file(await wholeAlienPcm()).arrayBuffer();
      const response = await handlers.post(
        postRequest("?model=onnx-community/whisper-tiny.en&language=en", body, {
          cookie: "sid=ok",
        }),
      );
      expect(response.status).toBe(200);

      // Timestamp every line as it arrives.
      const started = performance.now();
      const lines: { at: number; line: string }[] = [];
      const decoder = new TextDecoder();
      let buffered = "";
      for await (const chunk of response.body!) {
        buffered += decoder.decode(chunk, { stream: true });
        let newline: number;
        while ((newline = buffered.indexOf("\n")) >= 0) {
          lines.push({ at: performance.now() - started, line: buffered.slice(0, newline) });
          buffered = buffered.slice(newline + 1);
        }
      }
      const segmentLines = lines.filter(({ line }) => line.includes('"type":"segment"'));
      const firstSegment = lines.findIndex(({ line }) => line.includes('"type":"segment"'));
      // VAD over 15 min of audio + the first batch: heartbeats fill the gap.
      expect(lines.slice(0, firstSegment).filter(({ line }) => line === "").length).toBeGreaterThan(0);
      expect(segmentLines.length).toBeGreaterThan(3);
      // Streamed, not one blob at the end: later bursts arrive measurably later.
      expect(segmentLines.at(-1)!.at - segmentLines[0]!.at).toBeGreaterThan(100);
      expect(lines.at(-1)!.line).toContain('"type":"result"');
      console.log(
        "[D-WS5a] segment arrival (ms):",
        segmentLines.map(({ at }) => Math.round(at)).join(" "),
      );
    },
    { timeout: 600_000 },
  );

  test(
    "cancelling a batched file run stops it early and the next request still works",
    async () => {
      const sidecar = createSidecar();
      const audioPath = await wholeAlienPcm();
      const controller = new AbortController();
      const cancelledRun: SidecarSegment[] = [];
      const outcome = await sidecar.transcribe(
        { model: "tiny.en", audioPath, language: "en" },
        (segment) => {
          cancelledRun.push(segment);
          controller.abort();
        },
        controller.signal,
      );
      expect(outcome.cancelled).toBe(true);
      // At most the first burst (8 chunks) got out; the full clip is ≈ 17 chunks.
      expect(cancelledRun.length).toBeGreaterThan(0);
      expect(cancelledRun.length).toBeLessThanOrEqual(8);
      expect(cancelledRun.at(-1)!.end).toBeLessThan(800);

      const again: SidecarSegment[] = [];
      const next = await sidecar.transcribe(
        { model: "tiny.en", audioPath, language: "en" },
        (segment) => again.push(segment),
      );
      expect(next.cancelled).toBe(false);
      expect(again.length).toBeGreaterThan(cancelledRun.length);
    },
    { timeout: 600_000 },
  );
});