diff --git a/src/app/api/v1/alerts/audio/[token]/route.ts b/src/app/api/v1/alerts/audio/[token]/route.ts new file mode 100644 index 00000000..db0578a8 --- /dev/null +++ b/src/app/api/v1/alerts/audio/[token]/route.ts @@ -0,0 +1,32 @@ +/** + * The WAV an alert call plays, fetched by Telnyx's media loader. + * + * Unauthenticated on purpose: Telnyx fetches it with no credential of ours. + * What stands in for auth is the path — an HMAC of the call's id under a + * server secret — plus a ten-minute lifetime, so a URL is unguessable and + * stops working shortly after the call it belonged to. The body is speech + * this deployment just synthesized on the c0mpute network; there is nothing + * in it that is not about to be read down a phone line anyway. + */ + +import { NextResponse, type NextRequest } from "next/server"; +import { readAudio, TTL_MS } from "@/lib/voice/kokoro"; + +export async function GET( + _request: NextRequest, + { params }: { params: Promise<{ token: string }> } +) { + const { token } = await params; + const wav = await readAudio(token); + if (!wav) { + return NextResponse.json({ error: "Not found" }, { status: 404 }); + } + return new NextResponse(new Uint8Array(wav), { + status: 200, + headers: { + "content-type": "audio/wav", + "content-length": String(wav.length), + "cache-control": `private, max-age=${Math.floor(TTL_MS / 1000)}`, + }, + }); +} diff --git a/src/lib/alerts/__tests__/alerts.test.ts b/src/lib/alerts/__tests__/alerts.test.ts index 3af643c4..43bb1a9f 100644 --- a/src/lib/alerts/__tests__/alerts.test.ts +++ b/src/lib/alerts/__tests__/alerts.test.ts @@ -233,8 +233,12 @@ describe("alert call flow", () => { handleCallEvent(deps, LINE, { event_type, payload: { call_control_id: "cc-1", ...payload } }, s); expect(await send("call.initiated", { direction: "outgoing" }, state)).toBe("ignored: call.initiated"); - expect(await send("call.answered", {}, state)).toBe("alert read"); + expect(await send("call.answered", {}, state)).toBe("alert read in the agent's voice, rendered by Telnyx"); expect(cmds[0].action).toBe("speak"); + // The voice is the agent's Kokoro voice, not a generic one. (With a + // c0mpute speech gateway configured this would be a playback_start of + // audio we rendered ourselves; see lib/voice/kokoro.ts.) + expect(cmds[0].body.voice).toBe("Telnyx.KokoroTTS.am_michael"); expect(cmds[0].body.payload).toBe(alertSpeech("fleet-db-backup failed")); expect(String(cmds[0].body.payload).match(/fleet-db-backup failed/g)).toHaveLength(2); const after = decodeState(cmds[0].body.client_state) as CallState; diff --git a/src/lib/alerts/call.ts b/src/lib/alerts/call.ts index f8400eca..444b783e 100644 --- a/src/lib/alerts/call.ts +++ b/src/lib/alerts/call.ts @@ -3,15 +3,29 @@ * webhook as the voice menu. client_state carries al = alert id, m = the * call's log message, h = 1 once the alert is being read (hang up after it). * - * call.answered read the alert twice - * call.speak.ended hang up (the shared h = 1 rule in ivr.ts) - * call.hangup note answered / not answered on the log entry + * call.answered read the alert twice + * call.speak.ended hang up (the shared h = 1 rule in ivr.ts) + * call.playback.ended hang up, same rule, when we played our own audio + * call.hangup note answered / not answered on the log entry + * + * How the line is spoken: first choice is one of our agents' own voices, + * synthesized on the c0mpute network (Kokoro) and played back as a file — + * see lib/voice/kokoro.ts. If this deployment has no speech gateway, or the + * network cannot render it in time, the call falls back to a Telnyx `speak`. + * An alert is worth more than the voice it arrives in. */ import type { SupabaseClient } from "@supabase/supabase-js"; import { encodeState, type CallCommand } from "@/lib/voice/telnyx-voice"; import { callLength, RECORDING_NOTICE, type CallEvent, type CallState } from "@/lib/voice/ivr"; +import { speakOnC0mpute, telnyxKokoroVoice } from "@/lib/voice/kokoro"; +import { getSiteUrl } from "@/lib/site-url"; +/** + * The last-resort voice. Tier one is our own network (lib/voice/kokoro.ts), + * tier two is Telnyx's copy of the same Kokoro model; this is only what a + * `speak` falls back to if a Kokoro voice id is ever rejected. + */ const VOICE = { voice: "female", language: "en-US" }; const UNITS: Record = { @@ -80,12 +94,25 @@ export async function handleAlertCall( client_state: encodeState({ ...state }), }); } - await command(callId, "speak", { - payload: `${recordCalls ? `${RECORDING_NOTICE} ` : ""}${alertSpeech(String((alert as { message: string }).message))}`, - ...VOICE, - client_state: encodeState({ ...state, h: 1 }), + const line = `${recordCalls ? `${RECORDING_NOTICE} ` : ""}${alertSpeech(String((alert as { message: string }).message))}`; + const spoken = await speakOnC0mpute(line, { id: `${state.al}:${callId}`, publicBase: getSiteUrl() }); + if (spoken) { + await command(callId, "playback_start", { + audio_url: spoken.url, + client_state: encodeState({ ...state, h: 1 }), + }); + return `alert read in our own voice (${spoken.seconds}s)`; + } + const spokenState = encodeState({ ...state, h: 1 }); + const kokoro = await command(callId, "speak", { + payload: line, + ...telnyxKokoroVoice(), + client_state: spokenState, }); - return "alert read"; + if (kokoro.ok) return "alert read in the agent's voice, rendered by Telnyx"; + // A voice id Telnyx would not take is no reason to drop an alert. + await command(callId, "speak", { payload: line, ...VOICE, client_state: spokenState }); + return "alert read in the fallback voice"; } case "call.hangup": { diff --git a/src/lib/voice/__tests__/kokoro.test.ts b/src/lib/voice/__tests__/kokoro.test.ts new file mode 100644 index 00000000..200da18b --- /dev/null +++ b/src/lib/voice/__tests__/kokoro.test.ts @@ -0,0 +1,180 @@ +import { mkdtemp, readFile, utimes, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; +import { + alertVoice, + audioToken, + DEFAULT_AGENT_VOICE, + kokoroGateway, + readAudio, + speakOnC0mpute, + TTL_MS, + wavSeconds, + type SpeechEnv, +} from "../kokoro"; + +/** A minimal but real 16-bit PCM WAV header plus `frames` of silence. */ +function wav(frames: number, rate = 24000): Buffer { + const data = Buffer.alloc(frames * 2); + const head = Buffer.alloc(44); + head.write("RIFF", 0); + head.writeUInt32LE(36 + data.length, 4); + head.write("WAVE", 8); + head.write("fmt ", 12); + head.writeUInt32LE(16, 16); + head.writeUInt16LE(1, 20); + head.writeUInt16LE(1, 22); + head.writeUInt32LE(rate, 24); + head.writeUInt32LE(rate * 2, 28); + head.writeUInt16LE(2, 32); + head.writeUInt16LE(16, 34); + head.write("data", 36); + head.writeUInt32LE(data.length, 40); + return Buffer.concat([head, data]); +} + +async function scratch(): Promise { + return mkdtemp(join(tmpdir(), "kokoro-test-")); +} + +function env(extra: Record): SpeechEnv { + return { + C0MPUTE_SPEECH_URL: "http://speech.internal:8787", + C0MPUTE_SPEECH_TOKEN: "token-abc", + ALERT_AUDIO_SECRET: "secret", + ...extra, + }; +} + +describe("kokoroGateway", () => { + it("is off until both the url and the token are set", () => { + expect(kokoroGateway({})).toBeNull(); + expect(kokoroGateway({ C0MPUTE_SPEECH_URL: "http://x" })).toBeNull(); + expect(kokoroGateway({ C0MPUTE_SPEECH_TOKEN: "t" })).toBeNull(); + expect( + kokoroGateway({ C0MPUTE_SPEECH_URL: "http://x:8787/", C0MPUTE_SPEECH_TOKEN: "t" }) + ).toEqual({ baseUrl: "http://x:8787", token: "t" }); + }); +}); + +describe("alertVoice", () => { + it("defaults to riotcoder's voice and ignores a voice no node serves", () => { + expect(alertVoice({})).toBe(DEFAULT_AGENT_VOICE); + expect(alertVoice({ ALERT_KOKORO_VOICE: "af_bella" })).toBe("af_bella"); + expect(alertVoice({ ALERT_KOKORO_VOICE: "nope" })).toBe(DEFAULT_AGENT_VOICE); + }); +}); + +describe("speakOnC0mpute", () => { + it("renders the line on the network and returns a fetchable url", async () => { + const dir = await scratch(); + let sent: { url: string; body: unknown } | null = null; + const line = await speakOnC0mpute("alert, twice over", { + id: "alert-1:call-1", + publicBase: "https://smshub.dev/", + env: env({ ALERT_AUDIO_DIR: dir }), + fetcher: (async (url: string, init: RequestInit) => { + sent = { url: String(url), body: JSON.parse(String(init.body)) }; + return new Response(new Uint8Array(wav(24000)), { status: 200 }); + }) as unknown as typeof fetch, + }); + + expect(sent!.url).toBe("http://speech.internal:8787/v1/tts"); + expect(sent!.body).toMatchObject({ engine: "kokoro", voice: DEFAULT_AGENT_VOICE, text: "alert, twice over" }); + const token = audioToken("alert-1:call-1", env({ ALERT_AUDIO_DIR: dir })); + expect(line).toEqual({ url: `https://smshub.dev/api/v1/alerts/audio/${token}`, seconds: 1 }); + // The bytes are on disk under the token, ready for Telnyx to fetch. + expect((await readFile(join(dir, `${token}.wav`))).subarray(0, 4).toString()).toBe("RIFF"); + }); + + it("gives up quietly so the call can still speak the old way", async () => { + const dir = await scratch(); + const options = { id: "a:b", publicBase: "https://smshub.dev", env: env({ ALERT_AUDIO_DIR: dir }) }; + + // No gateway configured. + expect( + await speakOnC0mpute("x", { ...options, env: { ALERT_AUDIO_DIR: dir } }) + ).toBeNull(); + // Empty line. + expect(await speakOnC0mpute(" ", options)).toBeNull(); + // A base url we cannot build an absolute link from. + expect(await speakOnC0mpute("x", { ...options, publicBase: "smshub.dev" })).toBeNull(); + // The gateway refused. + expect( + await speakOnC0mpute("x", { + ...options, + fetcher: (async () => new Response("no nodes", { status: 503 })) as unknown as typeof fetch, + }) + ).toBeNull(); + // The network threw. + expect( + await speakOnC0mpute("x", { + ...options, + fetcher: (async () => { + throw new Error("connection reset"); + }) as unknown as typeof fetch, + }) + ).toBeNull(); + // A 200 that is not audio: an error page would otherwise be "played". + expect( + await speakOnC0mpute("x", { + ...options, + fetcher: (async () => new Response("oops", { status: 200 })) as unknown as typeof fetch, + }) + ).toBeNull(); + }); +}); + +describe("readAudio", () => { + it("serves only a well-formed token, and never walks out of its directory", async () => { + const dir = await scratch(); + const e = env({ ALERT_AUDIO_DIR: dir }); + const token = audioToken("alert-9:call-9", e); + await writeFile(join(dir, `${token}.wav`), wav(2400)); + + expect((await readAudio(token, e))?.length).toBe(44 + 4800); + expect(await readAudio("../../etc/passwd", e)).toBeNull(); + expect(await readAudio(`${token}/../${token}`, e)).toBeNull(); + expect(await readAudio("0".repeat(32), e)).toBeNull(); + expect(await readAudio("", e)).toBeNull(); + }); + + it("a token is tied to the secret, so another deployment's url is useless", async () => { + const dir = await scratch(); + const mine = audioToken("same-id", env({ ALERT_AUDIO_DIR: dir, ALERT_AUDIO_SECRET: "mine" })); + const theirs = audioToken("same-id", env({ ALERT_AUDIO_DIR: dir, ALERT_AUDIO_SECRET: "theirs" })); + expect(mine).not.toBe(theirs); + expect(mine).toMatch(/^[0-9a-f]{32}$/); + }); + + it("sweeps a line older than its ttl on the next write", async () => { + const dir = await scratch(); + const e = env({ ALERT_AUDIO_DIR: dir }); + const stale = audioToken("stale", e); + const stalePath = join(dir, `${stale}.wav`); + await writeFile(stalePath, wav(2400)); + const old = (Date.now() - TTL_MS - 60_000) / 1000; + await utimes(stalePath, old, old); + + await speakOnC0mpute("fresh", { + id: "fresh", + publicBase: "https://smshub.dev", + env: e, + fetcher: (async () => new Response(new Uint8Array(wav(2400)), { status: 200 })) as unknown as typeof fetch, + }); + // The sweep runs in the background; give it a tick. + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(await readAudio(stale, e)).toBeNull(); + expect(await readAudio(audioToken("fresh", e), e)).not.toBeNull(); + }); +}); + +describe("wavSeconds", () => { + it("reads the length out of the header", () => { + expect(wavSeconds(wav(24000))).toBe(1); + expect(wavSeconds(wav(12000))).toBe(0.5); + expect(wavSeconds(wav(16000, 16000))).toBe(1); + expect(wavSeconds(Buffer.alloc(10))).toBe(0); + }); +}); diff --git a/src/lib/voice/ivr.ts b/src/lib/voice/ivr.ts index d7a5f076..77cae887 100644 --- a/src/lib/voice/ivr.ts +++ b/src/lib/voice/ivr.ts @@ -103,7 +103,10 @@ export async function handleCallEvent( const callId = String(p.call_control_id ?? ""); if (!callId) return "ignored: no call_control_id"; if (state?.t) return "ignored: transferred leg"; - if (state?.al && !["call.speak.ended", "call.recording.saved"].includes(event.event_type)) { + if ( + state?.al && + !["call.speak.ended", "call.playback.ended", "call.recording.saved"].includes(event.event_type) + ) { return handleAlertCall(deps, event, state as CallState & { al: string }, Boolean(line.record_calls)); } @@ -204,7 +207,10 @@ export async function handleCallEvent( return "recording saved"; } - case "call.speak.ended": { + // Telnyx finished reading our text, or finished playing a file we + // synthesized ourselves. Either way the goodbye has been said. + case "call.speak.ended": + case "call.playback.ended": { if (!state?.h) return "ignored: not a goodbye"; await command(callId, "hangup", {}); return "hung up after goodbye"; diff --git a/src/lib/voice/kokoro.ts b/src/lib/voice/kokoro.ts new file mode 100644 index 00000000..6b9f4f7c --- /dev/null +++ b/src/lib/voice/kokoro.ts @@ -0,0 +1,211 @@ +/** + * Calls spoken in one of our agents' own voices, synthesized on the c0mpute + * network instead of a hosted TTS vendor. + * + * A Telnyx `speak` command renders text on Telnyx's side. This renders it on + * ours: POST /v1/tts on the c0mpute speech gateway auctions a `tts.synthesize` + * job to a node running Kokoro (an int8 ONNX export of hexgrad/Kokoro-82M, + * which every c0mpute node provisions itself), and hands back a WAV. The call + * then plays that file, so the voice is the agent's, the model is ours, and + * there is no per-character bill. + * + * The WAV lands in a short-lived cache on disk with an HMAC-named path, and + * `/api/v1/alerts/audio/` serves it unauthenticated, because the only + * fetcher is Telnyx's media loader. The token is derived from the bytes and a + * server secret, so a path cannot be guessed or walked, and anything older + * than TTL_MS is swept on the next write. + * + * Every step is optional: no gateway configured, a node that cannot be + * reached, a synth that fails — all return null, and the caller falls back to + * a plain Telnyx `speak`. An alert must still go out when the network is + * having a bad day. + */ + +import { createHmac, randomBytes } from "node:crypto"; +import { mkdir, readdir, readFile, stat, unlink, writeFile } from "node:fs/promises"; +import { join } from "node:path"; + +/** Keep a rendered line only as long as a call needs to fetch it. */ +export const TTL_MS = 10 * 60 * 1000; + +/** The longest line we ask a node to read; the gateway caps this too. */ +const MAX_CHARS = 6000; + +/** riotcoder's voice in its OpenProfile, and the default for alert calls. */ +export const DEFAULT_AGENT_VOICE = "am_michael"; + +/** Voices the gateway's Kokoro engine serves. */ +export const KOKORO_VOICES = ["af_sarah", "af_bella", "am_onyx", "am_michael"] as const; + +/** Just the environment variables this module reads. */ +export type SpeechEnv = Record; + +export interface KokoroGateway { + baseUrl: string; + token: string; +} + +/** The c0mpute speech gateway, when this deployment has one. */ +export function kokoroGateway(env: SpeechEnv = process.env): KokoroGateway | null { + const baseUrl = env.C0MPUTE_SPEECH_URL?.trim().replace(/\/+$/, ""); + const token = env.C0MPUTE_SPEECH_TOKEN?.trim(); + if (!baseUrl || !token) return null; + return { baseUrl, token }; +} + +/** The voice alert calls speak in: an env override, else riotcoder's. */ +export function alertVoice(env: SpeechEnv = process.env): string { + const chosen = env.ALERT_KOKORO_VOICE?.trim(); + return chosen && (KOKORO_VOICES as readonly string[]).includes(chosen) ? chosen : DEFAULT_AGENT_VOICE; +} + +/** + * The same model, rendered by Telnyx instead of by us. + * + * Telnyx hosts Kokoro as `Telnyx.KokoroTTS.`, so when our own gateway + * is out of reach the call still speaks in the agent's voice rather than a + * generic one — it just costs a vendor round trip instead of a node's CPU. + * This is the middle of three tiers: our network, Telnyx's copy, then + * whatever the account's default voice is. + */ +export function telnyxKokoroVoice(env: SpeechEnv = process.env): { voice: string; language: string } { + return { voice: `Telnyx.KokoroTTS.${alertVoice(env)}`, language: "en-US" }; +} + +/** Where rendered lines wait to be fetched. Never /tmp: that is quota-bound. */ +export function audioDir(env: SpeechEnv = process.env): string { + return env.ALERT_AUDIO_DIR?.trim() || join(process.cwd(), ".cache", "alert-audio"); +} + +/** + * The secret the path token is signed with. A deployment without one gets a + * per-process random secret: tokens then stop working after a restart, which + * only matters for a call in flight during a deploy. + */ +function audioSecret(env: SpeechEnv = process.env): string { + return env.ALERT_AUDIO_SECRET?.trim() || env.SUPABASE_SERVICE_ROLE_KEY?.trim() || processSecret; +} +const processSecret = randomBytes(32).toString("hex"); + +/** A path token nobody can guess or walk: 32 hex chars, no separators. */ +export function audioToken(id: string, env: SpeechEnv = process.env): string { + return createHmac("sha256", audioSecret(env)).update(id).digest("hex").slice(0, 32); +} + +/** Drop rendered lines past their TTL, so the cache cannot grow without bound. */ +async function sweep(dir: string, now = Date.now()): Promise { + let names: string[]; + try { + names = await readdir(dir); + } catch { + return; + } + await Promise.all( + names + .filter((name) => name.endsWith(".wav")) + .map(async (name) => { + const path = join(dir, name); + try { + const info = await stat(path); + if (now - info.mtimeMs > TTL_MS) await unlink(path); + } catch { + // Someone else swept it, or it is being written. Either is fine. + } + }) + ); +} + +/** Read a cached line back out, by the token in its URL. */ +export async function readAudio( + token: string, + env: SpeechEnv = process.env +): Promise { + if (!/^[0-9a-f]{32}$/.test(token)) return null; + try { + const bytes = await readFile(join(audioDir(env), `${token}.wav`)); + return bytes.length > 44 ? bytes : null; + } catch { + return null; + } +} + +export interface SpokenLine { + /** The absolute URL to hand Telnyx. */ + url: string; + seconds: number; +} + +/** + * Render `text` in `voice` on the c0mpute network and cache it for Telnyx. + * + * `id` names the cache entry (the alert id and call id make a good one) and + * `publicBase` is the origin Telnyx will fetch from. Returns null whenever + * this deployment cannot do it, which is never an error: the caller speaks + * the line the old way instead. + */ +export async function speakOnC0mpute( + text: string, + options: { + id: string; + publicBase: string; + voice?: string; + env?: SpeechEnv; + fetcher?: typeof fetch; + timeoutMs?: number; + } +): Promise { + const env = options.env ?? process.env; + const gateway = kokoroGateway(env); + const line = text.trim(); + if (!gateway || !line || line.length > MAX_CHARS) return null; + const base = options.publicBase.trim().replace(/\/+$/, ""); + if (!/^https?:\/\//.test(base)) return null; + + const doFetch = options.fetcher ?? fetch; + let wav: Buffer; + try { + const response = await doFetch(`${gateway.baseUrl}/v1/tts`, { + method: "POST", + headers: { authorization: `Bearer ${gateway.token}`, "content-type": "application/json" }, + body: JSON.stringify({ + text: line, + engine: "kokoro", + voice: options.voice ?? alertVoice(env), + locale: "en-us", + speed: 1, + }), + signal: AbortSignal.timeout(options.timeoutMs ?? 90_000), + }); + if (!response.ok) return null; + wav = Buffer.from(await response.arrayBuffer()); + } catch { + return null; + } + // A WAV, not an error page: RIFF....WAVE, and long enough to be speech. + if (wav.length < 1024 || wav.subarray(0, 4).toString() !== "RIFF" || wav.subarray(8, 12).toString() !== "WAVE") { + return null; + } + + const dir = audioDir(env); + const token = audioToken(options.id, env); + try { + await mkdir(dir, { recursive: true }); + await writeFile(join(dir, `${token}.wav`), wav); + } catch { + return null; + } + void sweep(dir); + + return { url: `${base}/api/v1/alerts/audio/${token}`, seconds: wavSeconds(wav) }; +} + +/** How long a 16-bit PCM WAV runs, for the log line. */ +export function wavSeconds(wav: Buffer): number { + if (wav.length < 44) return 0; + const rate = wav.readUInt32LE(24); + const channels = wav.readUInt16LE(22) || 1; + const bits = wav.readUInt16LE(34) || 16; + const bytesPerFrame = (bits / 8) * channels; + if (!rate || !bytesPerFrame) return 0; + return Math.round(((wav.length - 44) / bytesPerFrame / rate) * 100) / 100; +} diff --git a/src/lib/voice/telnyx-voice.ts b/src/lib/voice/telnyx-voice.ts index 9edaef1c..37c98fe5 100644 --- a/src/lib/voice/telnyx-voice.ts +++ b/src/lib/voice/telnyx-voice.ts @@ -237,7 +237,16 @@ export async function configureVoice( export type CallCommand = ( callControlId: string, - action: "answer" | "gather_using_speak" | "speak" | "transfer" | "hangup" | "record_start", + action: + | "answer" + | "gather_using_speak" + | "speak" + // playback_start plays a file we host, which is how a call speaks in one + // of our agents' voices instead of a vendor's (see lib/voice/kokoro.ts). + | "playback_start" + | "transfer" + | "hangup" + | "record_start", body: Record ) => Promise>;