voice-in-groq: arecord → Groq whisper-large-v3 voice-in channel

src/channels/voice-in.ts

@@ -0,0 +1,203 @@
+import { spawn as nodeSpawn } from "node:child_process";
+import { PassThrough } from "node:stream";
+
+export interface VoiceInBus {
+  send(msg: unknown): void;
+}
+
+export interface VoiceInSpawnedProcess {
+  stdout: NodeJS.ReadableStream;
+  stderr: NodeJS.ReadableStream;
+  kill: (sig?: NodeJS.Signals) => void;
+  exitCode: number | null;
+}
+
+export type VoiceInSpawn = (
+  cmd: string,
+  args: string[],
+) => VoiceInSpawnedProcess;
+
+export interface VoiceInOptions {
+  bus: VoiceInBus;
+  getKey: () => Promise<string> | string;
+  agentId?: string;
+  model?: string;
+  language?: string;
+  spawn?: VoiceInSpawn;
+  fetchImpl?: typeof fetch;
+  now?: () => number;
+}
+
+export interface VoiceInChannel {
+  start(): Promise<void>;
+  stop(): Promise<string>;
+  abort(): Promise<void>;
+}
+
+export class VoiceInError extends Error {
+  readonly status?: number;
+  constructor(message: string, status?: number) {
+    super(message);
+    this.name = "VoiceInError";
+    this.status = status;
+  }
+}
+
+const ARECORD_ARGS = [
+  "-q",
+  "-f",
+  "S16_LE",
+  "-r",
+  "16000",
+  "-c",
+  "1",
+  "-t",
+  "wav",
+  "-",
+];
+
+const TRANSCRIPTIONS_URL =
+  "https://api.groq.com/openai/v1/audio/transcriptions";
+
+interface CaptureState {
+  proc: VoiceInSpawnedProcess;
+  chunks: Buffer[];
+  ended: Promise<void>;
+}
+
+function isTranscriptionResponse(
+  value: unknown,
+): value is { text: string } {
+  return (
+    typeof value === "object" &&
+    value !== null &&
+    typeof (value as { text?: unknown }).text === "string"
+  );
+}
+
+const defaultSpawn: VoiceInSpawn = (cmd, args) => {
+  const child = nodeSpawn(cmd, args, { stdio: ["ignore", "pipe", "pipe"] });
+  return {
+    stdout: child.stdout,
+    stderr: child.stderr,
+    kill: (sig) => {
+      child.kill(sig);
+    },
+    get exitCode(): number | null {
+      return child.exitCode;
+    },
+  } as VoiceInSpawnedProcess;
+};
+
+export function createVoiceInChannel(opts: VoiceInOptions): VoiceInChannel {
+  const bus = opts.bus;
+  const getKey = opts.getKey;
+  const agentId = opts.agentId ?? "router";
+  const model = opts.model ?? "whisper-large-v3";
+  const language = opts.language ?? "en";
+  const spawn = opts.spawn ?? defaultSpawn;
+  const fetchImpl = opts.fetchImpl ?? fetch;
+  const now = opts.now ?? (() => Date.now());
+
+  let state: CaptureState | null = null;
+
+  async function beginCapture(): Promise<void> {
+    if (state !== null) {
+      throw new VoiceInError("voice-in already capturing");
+    }
+    const proc = spawn("arecord", ARECORD_ARGS);
+    const chunks: Buffer[] = [];
+    const ended = new Promise<void>((resolve) => {
+      proc.stdout.on("data", (chunk: Buffer | string) => {
+        const buf =
+          typeof chunk === "string" ? Buffer.from(chunk) : Buffer.from(chunk);
+        chunks.push(buf);
+      });
+      proc.stdout.on("end", () => {
+        resolve();
+      });
+      proc.stdout.on("close", () => {
+        resolve();
+      });
+      proc.stdout.on("error", () => {
+        resolve();
+      });
+    });
+    // Drain stderr so it does not stall the child process.
+    const stderrDrain = new PassThrough();
+    proc.stderr.on("data", (chunk: Buffer | string) => {
+      stderrDrain.write(chunk);
+    });
+    proc.stderr.on("error", () => {});
+    state = { proc, chunks, ended };
+  }
+
+  async function endCapture(): Promise<string> {
+    const active = state;
+    if (active === null) {
+      throw new VoiceInError("voice-in not capturing");
+    }
+    state = null;
+
+    active.proc.kill("SIGTERM");
+    await active.ended;
+
+    const pcm = Buffer.concat(active.chunks);
+    const key = await getKey();
+
+    const form = new FormData();
+    const blob = new Blob([pcm], { type: "audio/wav" });
+    form.append("file", blob, "capture.wav");
+    form.append("model", model);
+    form.append("language", language);
+    form.append("response_format", "json");
+
+    const res = await fetchImpl(TRANSCRIPTIONS_URL, {
+      method: "POST",
+      headers: {
+        Authorization: `Bearer ${key}`,
+      },
+      body: form,
+    });
+
+    if (!res.ok) {
+      throw new VoiceInError(
+        `groq transcription failed: ${res.status}`,
+        res.status,
+      );
+    }
+
+    const parsed: unknown = await res.json();
+    if (!isTranscriptionResponse(parsed)) {
+      throw new VoiceInError("groq transcription returned invalid payload");
+    }
+
+    const text = parsed.text;
+    const ts = now();
+    bus.send({
+      id: `voice-${ts}`,
+      kind: "prompt",
+      from: "voice-in",
+      to: agentId,
+      payload: { prompt: text },
+      ts,
+    });
+    return text;
+  }
+
+  async function abortCapture(): Promise<void> {
+    const active = state;
+    if (active === null) {
+      return;
+    }
+    state = null;
+    active.proc.kill("SIGTERM");
+    await active.ended;
+  }
+
+  return {
+    start: beginCapture,
+    stop: endCapture,
+    abort: abortCapture,
+  };
+}

tests/channels/voice-in.test.ts

@@ -0,0 +1,178 @@
+import { PassThrough } from "node:stream";
+import { describe, expect, it, vi } from "vitest";
+import {
+  VoiceInError,
+  createVoiceInChannel,
+  type VoiceInSpawn,
+  type VoiceInSpawnedProcess,
+} from "../../src/channels/voice-in.js";
+
+interface FakeProcess extends VoiceInSpawnedProcess {
+  stdout: PassThrough;
+  stderr: PassThrough;
+  kill: ReturnType<typeof vi.fn>;
+}
+
+function makeFakeSpawn(): {
+  spawn: VoiceInSpawn;
+  proc: FakeProcess;
+  calls: Array<{ cmd: string; args: string[] }>;
+} {
+  const calls: Array<{ cmd: string; args: string[] }> = [];
+  const stdout = new PassThrough();
+  const stderr = new PassThrough();
+  const kill = vi.fn((_sig?: NodeJS.Signals) => {
+    stdout.end();
+    stderr.end();
+  });
+  const proc: FakeProcess = {
+    stdout,
+    stderr,
+    kill,
+    exitCode: null,
+  };
+  const spawn: VoiceInSpawn = (cmd, args) => {
+    calls.push({ cmd, args });
+    return proc;
+  };
+  return { spawn, proc, calls };
+}
+
+describe("createVoiceInChannel", () => {
+  it("captures audio, uploads to Groq, and publishes a bus prompt message", async () => {
+    const { spawn, proc, calls } = makeFakeSpawn();
+    const bus = { send: vi.fn() };
+
+    const fetchImpl = vi.fn(async (input: string | URL, init?: RequestInit) => {
+      const url = typeof input === "string" ? input : input.toString();
+      expect(url).toBe("https://api.groq.com/openai/v1/audio/transcriptions");
+      expect(init?.method).toBe("POST");
+      const headers = init?.headers as Record<string, string>;
+      expect(headers.Authorization).toBe("Bearer test-key");
+      const body = init?.body;
+      expect(body).toBeInstanceOf(FormData);
+      const form = body as FormData;
+      expect(form.get("model")).toBe("whisper-large-v3");
+      expect(form.get("language")).toBe("en");
+      expect(form.get("response_format")).toBe("json");
+      const file = form.get("file");
+      expect(file).toBeInstanceOf(Blob);
+      return new Response(JSON.stringify({ text: "hello world" }), {
+        status: 200,
+        headers: { "content-type": "application/json" },
+      });
+    });
+
+    const channel = createVoiceInChannel({
+      bus,
+      getKey: () => "test-key",
+      spawn,
+      fetchImpl: fetchImpl as unknown as typeof fetch,
+      now: () => 1234,
+    });
+
+    await channel.start();
+    expect(calls).toHaveLength(1);
+    expect(calls[0].cmd).toBe("arecord");
+    expect(calls[0].args).toEqual([
+      "-q",
+      "-f",
+      "S16_LE",
+      "-r",
+      "16000",
+      "-c",
+      "1",
+      "-t",
+      "wav",
+      "-",
+    ]);
+
+    proc.stdout.write(Buffer.from([0x01, 0x02, 0x03, 0x04]));
+    proc.stdout.write(Buffer.from([0x05, 0x06]));
+
+    const transcript = await channel.stop();
+
+    expect(transcript).toBe("hello world");
+    expect(proc.kill).toHaveBeenCalledWith("SIGTERM");
+    expect(fetchImpl).toHaveBeenCalledTimes(1);
+    expect(bus.send).toHaveBeenCalledTimes(1);
+    expect(bus.send).toHaveBeenCalledWith({
+      id: "voice-1234",
+      kind: "prompt",
+      from: "voice-in",
+      to: "router",
+      payload: { prompt: "hello world" },
+      ts: 1234,
+    });
+  });
+
+  it("honors agentId override when publishing", async () => {
+    const { spawn, proc } = makeFakeSpawn();
+    const bus = { send: vi.fn() };
+    const fetchImpl = vi.fn(
+      async () =>
+        new Response(JSON.stringify({ text: "ok" }), { status: 200 }),
+    );
+
+    const channel = createVoiceInChannel({
+      bus,
+      agentId: "planner",
+      getKey: () => "k",
+      spawn,
+      fetchImpl: fetchImpl as unknown as typeof fetch,
+      now: () => 7,
+    });
+
+    await channel.start();
+    proc.stdout.write(Buffer.from([0xaa]));
+    await channel.stop();
+
+    const call = bus.send.mock.calls[0][0] as { to: string };
+    expect(call.to).toBe("planner");
+  });
+
+  it("abort() kills the recorder and does not transcribe or publish", async () => {
+    const { spawn, proc } = makeFakeSpawn();
+    const bus = { send: vi.fn() };
+    const fetchImpl = vi.fn();
+
+    const channel = createVoiceInChannel({
+      bus,
+      getKey: () => "test-key",
+      spawn,
+      fetchImpl: fetchImpl as unknown as typeof fetch,
+    });
+
+    await channel.start();
+    proc.stdout.write(Buffer.from([0xff]));
+    await channel.abort();
+
+    expect(proc.kill).toHaveBeenCalledWith("SIGTERM");
+    expect(fetchImpl).not.toHaveBeenCalled();
+    expect(bus.send).not.toHaveBeenCalled();
+  });
+
+  it("throws VoiceInError when Groq returns 401", async () => {
+    const { spawn, proc } = makeFakeSpawn();
+    const bus = { send: vi.fn() };
+    const fetchImpl = vi.fn(
+      async () =>
+        new Response(JSON.stringify({ error: "unauthorized" }), {
+          status: 401,
+        }),
+    );
+
+    const channel = createVoiceInChannel({
+      bus,
+      getKey: () => "bad-key",
+      spawn,
+      fetchImpl: fetchImpl as unknown as typeof fetch,
+    });
+
+    await channel.start();
+    proc.stdout.write(Buffer.from([0x00]));
+
+    await expect(channel.stop()).rejects.toBeInstanceOf(VoiceInError);
+    expect(bus.send).not.toHaveBeenCalled();
+  });
+});