From 48bdc96dd2301006748840935c5d88a13b86b00a Mon Sep 17 00:00:00 2001 From: Jason Brashear <115739170+webdevtodayjason@users.noreply.github.com> Date: Sat, 11 Apr 2026 21:07:41 -0500 Subject: [PATCH] =?UTF-8?q?voice-in-groq:=20arecord=20=E2=86=92=20Groq=20w?= =?UTF-8?q?hisper-large-v3=20voice-in=20channel?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task 021 engineer-router. Adds createVoiceInChannel with start/stop/abort and VoiceInError. Uploads buffered PCM as multipart/form-data to https://api.groq.com/openai/v1/audio/transcriptions and publishes the transcript as a bus prompt message to agentId (default "router"). Co-Authored-By: Claude Opus 4.6 (1M context) --- src/channels/voice-in.ts | 203 ++++++++++++++++++++++++++++++++ tests/channels/voice-in.test.ts | 178 ++++++++++++++++++++++++++++ 2 files changed, 381 insertions(+) create mode 100644 src/channels/voice-in.ts create mode 100644 tests/channels/voice-in.test.ts diff --git a/src/channels/voice-in.ts b/src/channels/voice-in.ts new file mode 100644 index 0000000..2e441ad --- /dev/null +++ b/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; + agentId?: string; + model?: string; + language?: string; + spawn?: VoiceInSpawn; + fetchImpl?: typeof fetch; + now?: () => number; +} + +export interface VoiceInChannel { + start(): Promise; + stop(): Promise; + abort(): Promise; +} + +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; +} + +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 { + if (state !== null) { + throw new VoiceInError("voice-in already capturing"); + } + const proc = spawn("arecord", ARECORD_ARGS); + const chunks: Buffer[] = []; + const ended = new Promise((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 { + 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 { + const active = state; + if (active === null) { + return; + } + state = null; + active.proc.kill("SIGTERM"); + await active.ended; + } + + return { + start: beginCapture, + stop: endCapture, + abort: abortCapture, + }; +} diff --git a/tests/channels/voice-in.test.ts b/tests/channels/voice-in.test.ts new file mode 100644 index 0000000..f409e25 --- /dev/null +++ b/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; +} + +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; + 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(); + }); +});