feat(cli): stream long-form speech with say --stream

biome.json

@@ -1,7 +1,7 @@
 {
   "$schema": "https://biomejs.dev/schemas/2.5.1/schema.json",
   "files": {
-    "includes": ["**", "!**/dist", "!**/node_modules"]
+    "includes": ["**", "!**/dist", "!**/node_modules", "!**/.claude"]
   },
   "formatter": {
     "enabled": true,

pnpm-workspace.yaml

@@ -0,0 +1,6 @@
+# pnpm settings live here (pnpm 10+ no longer reads the "pnpm" field in package.json).
+# Dependency build scripts are blocked by default; esbuild needs its postinstall to
+# fetch/verify the platform binary tsup builds with. Without this, pnpm's dependency
+# check fails and `pnpm build` / `pnpm test` exit 1 (ERR_PNPM_IGNORED_BUILDS).
+allowBuilds:
+  esbuild: true

src/audio/sink.test.ts

@@ -0,0 +1,175 @@
+import { mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { Writable } from "node:stream";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { CliError } from "../core/errors.js";
+import { assertBinaryStdout, assertPathAvailable, writeStreamToFile, writeStreamToStdout } from "./sink.js";
+
+const encoder = new TextEncoder();
+
+async function* chunksOf(...parts: string[]): AsyncGenerator<Uint8Array> {
+  for (const part of parts) yield encoder.encode(part);
+}
+
+/** A source that fails part way through, like a connection dropping mid-download. */
+async function* failsAfterFirstChunk(err: Error): AsyncGenerator<Uint8Array> {
+  yield encoder.encode("first half");
+  throw err;
+}
+
+/** Collects everything written, and can be told to fail like a closed pipe. */
+class RecordingWritable extends Writable {
+  written = "";
+  constructor(private readonly failWith?: NodeJS.ErrnoException) {
+    super({ highWaterMark: 4 }); // tiny, so backpressure actually happens
+  }
+  override _write(chunk: Buffer, _encoding: BufferEncoding, done: (err?: Error | null) => void): void {
+    if (this.failWith) {
+      done(this.failWith);
+      return;
+    }
+    this.written += chunk.toString();
+    done();
+  }
+}
+
+let directory: string;
+
+beforeEach(async () => {
+  directory = await mkdtemp(join(tmpdir(), "speechify-sink-"));
+});
+
+afterEach(async () => {
+  await rm(directory, { recursive: true, force: true });
+});
+
+describe("writeStreamToFile", () => {
+  it("writes every chunk and leaves no temporary file behind", async () => {
+    const path = join(directory, "speech.mp3");
+    const bytes = await writeStreamToFile(chunksOf("one ", "two ", "three"), path);
+
+    expect(bytes).toBe("one two three".length);
+    expect(await readFile(path, "utf8")).toBe("one two three");
+    expect(await readdir(directory)).toEqual(["speech.mp3"]);
+  });
+
+  it("removes the partial file and rethrows when the stream dies part way", async () => {
+    const path = join(directory, "speech.mp3");
+    const boom = new CliError("The audio stream stalled.", { code: "stream_stalled" });
+
+    await expect(writeStreamToFile(failsAfterFirstChunk(boom), path)).rejects.toBe(boom);
+    // Neither a truncated destination nor a stray .part file survives.
+    expect(await readdir(directory)).toEqual([]);
+  });
+
+  it("refuses a stream that ends without sending any audio, and writes nothing", async () => {
+    const path = join(directory, "speech.mp3");
+
+    await expect(writeStreamToFile(chunksOf(), path)).rejects.toMatchObject({
+      code: "empty_stream",
+      exitCode: 69,
+    });
+    expect(await readdir(directory)).toEqual([]);
+  });
+
+  it("replaces an existing file (the caller decides whether that is allowed)", async () => {
+    const path = join(directory, "speech.mp3");
+    await writeFile(path, "stale");
+
+    await writeStreamToFile(chunksOf("fresh"), path);
+
+    expect(await readFile(path, "utf8")).toBe("fresh");
+    expect(await readdir(directory)).toEqual(["speech.mp3"]);
+  });
+
+  it("does not leave interrupt handlers behind", async () => {
+    const before = process.listenerCount("SIGINT");
+    await writeStreamToFile(chunksOf("audio"), join(directory, "speech.mp3"));
+    expect(process.listenerCount("SIGINT")).toBe(before);
+
+    await expect(writeStreamToFile(chunksOf(), join(directory, "empty.mp3"))).rejects.toThrow();
+    expect(process.listenerCount("SIGINT")).toBe(before);
+  });
+});
+
+describe("assertPathAvailable", () => {
+  it("passes when nothing is there", async () => {
+    await expect(assertPathAvailable(join(directory, "speech.mp3"))).resolves.toBeUndefined();
+  });
+
+  it("fails with output_exists, naming both ways past it", async () => {
+    const path = join(directory, "speech.mp3");
+    await writeFile(path, "existing");
+
+    await expect(assertPathAvailable(path)).rejects.toMatchObject({ code: "output_exists", exitCode: 65 });
+    await expect(assertPathAvailable(path)).rejects.toThrow(/--out .*--force/);
+  });
+});
+
+describe("assertBinaryStdout", () => {
+  const stdout = process.stdout as { isTTY?: boolean };
+  const original = stdout.isTTY;
+
+  afterEach(() => {
+    stdout.isTTY = original;
+  });
+
+  it("refuses to write raw audio to a terminal", () => {
+    stdout.isTTY = true;
+    expect(() => assertBinaryStdout()).toThrow(CliError);
+    expect(() => assertBinaryStdout()).toThrow(/Refusing to write raw audio to the terminal/);
+  });
+
+  it("allows a redirect or a pipe", () => {
+    stdout.isTTY = false;
+    expect(() => assertBinaryStdout()).not.toThrow();
+  });
+});
+
+describe("writeStreamToStdout", () => {
+  it("writes every chunk, waiting for the buffer to drain", async () => {
+    const target = new RecordingWritable();
+    const bytes = await writeStreamToStdout(chunksOf("aaaa", "bbbb", "cccc"), target);
+
+    expect(bytes).toBe(12);
+    expect(target.written).toBe("aaaabbbbcccc");
+  });
+
+  it("stops quietly when the reader closes the pipe", async () => {
+    const epipe: NodeJS.ErrnoException = Object.assign(new Error("write EPIPE"), { code: "EPIPE" });
+    const target = new RecordingWritable(epipe);
+
+    await expect(writeStreamToStdout(chunksOf("aaaa", "bbbb", "cccc"), target)).resolves.toBeGreaterThan(0);
+  });
+
+  it("surfaces any other write failure", async () => {
+    const denied: NodeJS.ErrnoException = Object.assign(new Error("write EACCES"), { code: "EACCES" });
+    const target = new RecordingWritable(denied);
+
+    await expect(writeStreamToStdout(chunksOf("aaaa", "bbbb", "cccc"), target)).rejects.toBe(denied);
+  });
+
+  it("refuses a stream that ends without sending any audio", async () => {
+    await expect(writeStreamToStdout(chunksOf(), new RecordingWritable())).rejects.toMatchObject({
+      code: "empty_stream",
+    });
+  });
+});
+
+describe("a destination that vanishes mid-write", () => {
+  it("fails instead of waiting forever for a drain that will not come", async () => {
+    // Never calls the write callback, so the buffer fills and stays full; the
+    // stream is then destroyed without an error, emitting only 'close'.
+    const target = new Writable({
+      highWaterMark: 1,
+      write: () => undefined,
+    });
+    setTimeout(() => target.destroy(), 5);
+
+    await expect(writeStreamToStdout(chunksOf("aaaa", "bbbb"), target)).rejects.toMatchObject({
+      code: "sink_closed",
+      exitCode: 69,
+    });
+  });
+});

src/audio/sink.ts

@@ -0,0 +1,195 @@
+// Where streamed audio bytes end up: a file, or stdout.
+//
+// Two rules shape this module. Nothing is buffered whole (a 20,000-character
+// synthesis is megabytes), and a file never exists in a half-written state: we
+// write to a temporary file in the destination directory and rename it into
+// place only once the stream has finished, removing it on any failure — an
+// interrupt included.
+import { randomBytes } from "node:crypto";
+import { createWriteStream, rmSync } from "node:fs";
+import { rename, rm, stat } from "node:fs/promises";
+import { basename, dirname, join, resolve } from "node:path";
+import type { Writable } from "node:stream";
+import { finished } from "node:stream/promises";
+import { CliError, ExitCode } from "../core/errors.js";
+
+function isErrnoCode(err: unknown, code: string): boolean {
+  return err instanceof Error && (err as NodeJS.ErrnoException).code === code;
+}
+
+function emptyStreamError(): CliError {
+  return new CliError("The stream ended without sending any audio.", {
+    exitCode: ExitCode.UNAVAILABLE,
+    code: "empty_stream",
+  });
+}
+
+interface PumpOptions {
+  /**
+   * Treat a closed downstream pipe as a clean stop rather than a failure. True
+   * for stdout (`… --out - | head`), false for a file, where EPIPE cannot occur.
+   */
+  stopOnBrokenPipe: boolean;
+}
+
+/**
+ * Wait for a full write buffer to empty. Resolves on 'drain'; fails if the
+ * target errors or closes first, so a destination that goes away mid-write can
+ * never leave us waiting for a drain that will not come.
+ */
+function waitForDrain(target: Writable): Promise<void> {
+  return new Promise<void>((resolveDrain, rejectDrain) => {
+    const cleanup = (): void => {
+      target.off("drain", onDrain);
+      target.off("error", onError);
+      target.off("close", onClose);
+    };
+    const onDrain = (): void => {
+      cleanup();
+      resolveDrain();
+    };
+    const onError = (err: Error): void => {
+      cleanup();
+      rejectDrain(err);
+    };
+    const onClose = (): void => {
+      cleanup();
+      rejectDrain(
+        new CliError("The destination closed before the audio finished writing.", {
+          exitCode: ExitCode.UNAVAILABLE,
+          code: "sink_closed",
+        }),
+      );
+    };
+    target.once("drain", onDrain);
+    target.once("error", onError);
+    target.once("close", onClose);
+  });
+}
+
+/** Copy chunks into `target`, honouring backpressure. Returns the bytes written. */
+async function pump(target: Writable, chunks: AsyncIterable<Uint8Array>, options: PumpOptions): Promise<number> {
+  let bytes = 0;
+  // A write error can land between our awaits; with no listener attached Node
+  // turns it into an uncaught 'error' event, so capture it and raise it in turn.
+  let failure: Error | undefined;
+  const capture = (err: Error): void => {
+    failure = err;
+  };
+  target.on("error", capture);
+
+  try {
+    for await (const chunk of chunks) {
+      if (failure) throw failure;
+      bytes += chunk.byteLength;
+      // Backpressure: a false return means the buffer is full.
+      if (!target.write(chunk)) await waitForDrain(target);
+    }
+    if (failure) throw failure;
+    return bytes;
+  } catch (err) {
+    // The downstream reader closed the pipe (`… | head`). That is how shell
+    // pipelines end, not a failure: report what we wrote and stop.
+    if (options.stopOnBrokenPipe && isErrnoCode(err, "EPIPE")) return bytes;
+    throw err;
+  } finally {
+    target.off("error", capture);
+  }
+}
+
+/** Refuse to spray raw audio over a terminal. */
+export function assertBinaryStdout(): void {
+  if (!process.stdout.isTTY) return;
+  throw new CliError(
+    "Refusing to write raw audio to the terminal. Redirect it (`--out - > speech.mp3`), pipe it (`--out - | ffplay -i -`), or drop `--out -` to write a file.",
+    { exitCode: ExitCode.DATA_ERR, code: "binary_to_tty" },
+  );
+}
+
+/** Fail if `path` is already taken, naming the two ways past it. */
+export async function assertPathAvailable(path: string): Promise<void> {
+  try {
+    await stat(path);
+  } catch (err) {
+    // Nothing there: the path is free. Anything else (permissions, a bad
+    // directory) is a real problem and must not be mistaken for "available".
+    if (isErrnoCode(err, "ENOENT")) return;
+    throw err;
+  }
+  throw new CliError(`${path} already exists. Pass --out <path> to write elsewhere, or --force to overwrite it.`, {
+    exitCode: ExitCode.DATA_ERR,
+    code: "output_exists",
+  });
+}
+
+/**
+ * Remove the partial file and exit if the run is interrupted mid-stream, so a
+ * cancelled download never leaves debris behind. Returns a function that puts
+ * the previous (default) signal behaviour back.
+ */
+function removeOnInterrupt(path: string): () => void {
+  const handler = (signal: NodeJS.Signals): void => {
+    rmSync(path, { force: true });
+    // The shell convention for "killed by signal N" is 128 + N: 130 for SIGINT,
+    // 143 for SIGTERM.
+    process.exit(signal === "SIGTERM" ? 143 : 130);
+  };
+  process.on("SIGINT", handler);
+  process.on("SIGTERM", handler);
+  return () => {
+    process.off("SIGINT", handler);
+    process.off("SIGTERM", handler);
+  };
+}
+
+/**
+ * Write a stream to `path` atomically: a temporary file in the same directory,
+ * renamed into place once the last chunk lands. A stream that dies halfway
+ * leaves no file at all rather than a truncated one that looks complete.
+ *
+ * An existing `path` is replaced. Callers that must not clobber check first with
+ * `assertPathAvailable`.
+ */
+export async function writeStreamToFile(chunks: AsyncIterable<Uint8Array>, path: string): Promise<number> {
+  const destination = resolve(path);
+  // Same directory, so the rename is atomic (never a cross-device copy).
+  const temporary = join(dirname(destination), `.${basename(destination)}.${randomBytes(6).toString("hex")}.part`);
+  const file = createWriteStream(temporary, { flags: "wx" });
+  const restoreSignals = removeOnInterrupt(temporary);
+
+  try {
+    const bytes = await pump(file, chunks, { stopOnBrokenPipe: false });
+    file.end();
+    await finished(file);
+    if (bytes === 0) throw emptyStreamError();
+    await rename(temporary, destination);
+    return bytes;
+  } catch (err) {
+    // Wait for the file to actually close before removing it. destroy() can
+    // leave an open() in flight, and that open would recreate the path moments
+    // after the rm, leaving a stray .part file behind. The close itself may
+    // reject (premature close is expected here); we are already unwinding with
+    // the real failure, and all we need from it is a closed descriptor.
+    file.destroy();
+    await finished(file).catch(() => undefined);
+    await rm(temporary, { force: true });
+    throw err;
+  } finally {
+    restoreSignals();
+  }
+}
+
+/**
+ * Write a stream to stdout. A closed downstream pipe ends it quietly.
+ *
+ * `target` is injected so the broken-pipe path can be driven in a test without
+ * tampering with the real stdout.
+ */
+export async function writeStreamToStdout(
+  chunks: AsyncIterable<Uint8Array>,
+  target: Writable = process.stdout,
+): Promise<number> {
+  const bytes = await pump(target, chunks, { stopOnBrokenPipe: true });
+  if (bytes === 0) throw emptyStreamError();
+  return bytes;
+}