feat(alerts): calls speak in an agent's own Kokoro voice, not a vendor's

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)}`,
+    },
+  });
+}

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;

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<string, string> = {
@@ -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": {

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<string> {
+  return mkdtemp(join(tmpdir(), "kokoro-test-"));
+}
+
+function env(extra: Record<string, string>): 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("<!doctype html><title>oops</title>", { 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);
+  });
+});

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";