perf(bridge): stream TTS per sentence to cut voice reply latency

The /converse turn synthesised the entire reply before any audio played, so
time-to-first-audio grew with reply length. Add a streaming /converse_stream
endpoint that emits the transcript/reply first, then one audio clip per
sentence as each finishes synthesising. The Discord voice layer enqueues each
clip on arrival via the existing FIFO playQueue, so the first sentence starts
speaking while the rest are still being synthesised.

STT and the reply engine still run to completion before the first clip; only
TTS is pipelined. The non-streaming /converse and /text endpoints are
unchanged.

- bridge: language-agnostic sentence splitter (bridge/text_utils.py) + NDJSON
  streaming route
- bot: ndjson() reader + converseStream() client; voice.ts plays clips
  progressively
- tests: splitter unit tests + bot ndjson/converseStream tests

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

bot/src/bridge.test.ts

@@ -0,0 +1,87 @@
+import { test, expect } from "bun:test";
+
+// bridge.ts imports the runtime `config`, which requires DISCORD_GUILD_ID.
+// Set it before the dynamic import so the module loads without a real .env.
+process.env.DISCORD_GUILD_ID ||= "test-guild";
+const { ndjson, converseStream } = await import("./bridge.ts");
+
+const enc = (s: string) => new TextEncoder().encode(s);
+
+async function* chunked(...cs: string[]): AsyncGenerator<Uint8Array> {
+  for (const c of cs) yield enc(c);
+}
+
+test("ndjson yields one object per line even when chunks split mid-line", async () => {
+  const out: any[] = [];
+  for await (const o of ndjson(chunked('{"a":1}\n{"b":', '2}\n{"c":3}'))) out.push(o);
+  expect(out).toEqual([{ a: 1 }, { b: 2 }, { c: 3 }]);
+});
+
+test("ndjson skips blank lines and a trailing newline", async () => {
+  const out: any[] = [];
+  for await (const o of ndjson(chunked('{"a":1}\n\n{"b":2}\n'))) out.push(o);
+  expect(out).toEqual([{ a: 1 }, { b: 2 }]);
+});
+
+test("converseStream surfaces meta first, then plays each sentence clip in order", async () => {
+  const clipA = Buffer.from("clipA");
+  const clipB = Buffer.from("clipB");
+  const body =
+    [
+      JSON.stringify({
+        type: "meta",
+        transcript: "안녕",
+        reply: "안녕하세요. 반갑습니다!",
+        broadcast_action: "start",
+      }),
+      JSON.stringify({ type: "audio", seq: 0, audio_b64: clipA.toString("base64") }),
+      JSON.stringify({ type: "audio", seq: 1, audio_b64: clipB.toString("base64") }),
+      JSON.stringify({ type: "end" }),
+    ].join("\n") + "\n";
+
+  const orig = globalThis.fetch;
+  globalThis.fetch = (async () => ({
+    ok: true,
+    body: chunked(body),
+    text: async () => "",
+  })) as any;
+
+  try {
+    const events: string[] = [];
+    const clips: Buffer[] = [];
+    let meta: any;
+    await converseStream(Buffer.from("wav"), true, {
+      onMeta: (m) => {
+        meta = m;
+        events.push("meta");
+      },
+      onAudio: (c) => {
+        clips.push(c);
+        events.push("audio");
+      },
+    });
+
+    expect(meta.transcript).toBe("안녕");
+    expect(meta.broadcast_action).toBe("start");
+    // Meta must arrive before any audio, and clips must stay in order.
+    expect(events).toEqual(["meta", "audio", "audio"]);
+    expect(clips.map((c) => c.toString())).toEqual(["clipA", "clipB"]);
+  } finally {
+    globalThis.fetch = orig;
+  }
+});
+
+test("converseStream throws on a non-ok bridge response", async () => {
+  const orig = globalThis.fetch;
+  globalThis.fetch = (async () => ({
+    ok: false,
+    status: 500,
+    body: null,
+    text: async () => "boom",
+  })) as any;
+  try {
+    await expect(converseStream(Buffer.from("wav"), undefined, {})).rejects.toThrow("500");
+  } finally {
+    globalThis.fetch = orig;
+  }
+});

bot/src/bridge.ts

@@ -38,6 +38,69 @@ export async function converse(wav: Buffer, broadcasting?: boolean): Promise<Con
   return (await res.json()) as ConverseResult;
 }
 
+/** Metadata for a streamed turn: everything except the audio clips. */
+export interface ConverseMeta {
+  transcript: string;
+  language?: string | null;
+  reply: string;
+  error?: string | null;
+  broadcast_action?: BroadcastAction | null;
+}
+
+export interface ConverseStreamHandlers {
+  /** Fired once, before any audio, with the transcript/reply/broadcast directive. */
+  onMeta?: (meta: ConverseMeta) => void | Promise<void>;
+  /** Fired per sentence as its audio finishes synthesising (in order). */
+  onAudio?: (wav: Buffer) => void | Promise<void>;
+}
+
+/** Parse a byte stream of newline-delimited JSON into objects, one per line. */
+export async function* ndjson(
+  stream: AsyncIterable<Uint8Array>,
+): AsyncGenerator<any> {
+  const decoder = new TextDecoder();
+  let buf = "";
+  for await (const chunk of stream) {
+    buf += decoder.decode(chunk, { stream: true });
+    let nl: number;
+    while ((nl = buf.indexOf("\n")) >= 0) {
+      const line = buf.slice(0, nl).trim();
+      buf = buf.slice(nl + 1);
+      if (line) yield JSON.parse(line);
+    }
+  }
+  const last = buf.trim();
+  if (last) yield JSON.parse(last);
+}
+
+/** Streaming voice turn: the bridge emits the transcript/reply first, then one
+ *  audio clip per sentence as it is synthesised. Handlers run in arrival order,
+ *  so playing each clip on arrival starts the first sentence while the rest are
+ *  still being spoken. Mirrors {@link converse} but pipelines TTS. */
+export async function converseStream(
+  wav: Buffer,
+  broadcasting: boolean | undefined,
+  handlers: ConverseStreamHandlers,
+): Promise<void> {
+  const qs = broadcasting === undefined ? "" : `?broadcasting=${broadcasting ? "1" : "0"}`;
+  const res = await fetch(`${config.bridgeUrl}/converse_stream${qs}`, {
+    method: "POST",
+    headers: { "content-type": "audio/wav" },
+    body: wav,
+  });
+  if (!res.ok || !res.body) {
+    throw new Error(`bridge /converse_stream ${res.status}: ${await res.text().catch(() => "")}`);
+  }
+  for await (const ev of ndjson(res.body as AsyncIterable<Uint8Array>)) {
+    if (ev.type === "meta") {
+      await handlers.onMeta?.(ev as ConverseMeta);
+    } else if (ev.type === "audio" && ev.audio_b64) {
+      const clip = decodeWav(ev.audio_b64);
+      if (clip) await handlers.onAudio?.(clip);
+    }
+  }
+}
+
 /** Text-only turn (used by /자비스 ask). */
 export async function ask(text: string): Promise<TextResult> {
   const res = await fetch(`${config.bridgeUrl}/text`, {

bot/src/voice.ts

@@ -23,7 +23,7 @@ import {
 } from "@discordjs/voice";
 import prism from "prism-media";
 import type { VoiceBasedChannel } from "discord.js";
-import { converse, decodeWav } from "./bridge.ts";
+import { converseStream } from "./bridge.ts";
 import { config } from "./config.ts";
 
 const DISCORD_RATE = 48000;
@@ -128,21 +128,27 @@ export class VoiceSession {
     const wav = pcm16MonoToWav(mono, DISCORD_RATE);
 
     try {
-      const result = await converse(wav, this.getBroadcasting?.());
-      if (result.transcript) {
-        this.onTurn?.({ user: userId, transcript: result.transcript, reply: result.reply });
-      }
-      // Apply any broadcast directive the brain requested (e.g. user said
-      // "방송 켜줘 / 꺼줘") before playing the reply.
-      if (result.broadcast_action && this.onBroadcastAction) {
-        try {
-          await this.onBroadcastAction(result.broadcast_action);
-        } catch (e) {
-          console.error("[voice] broadcast action failed:", e);
-        }
-      }
-      const audio = decodeWav(result.audio_b64);
-      if (audio) this.play(audio);
+      // Streaming turn: the brain sends transcript/reply first, then one audio
+      // clip per sentence as it is synthesised. We enqueue each clip on arrival
+      // so the first sentence starts playing while the rest are still spoken.
+      await converseStream(wav, this.getBroadcasting?.(), {
+        onMeta: async (meta) => {
+          if (meta.transcript) {
+            this.onTurn?.({ user: userId, transcript: meta.transcript, reply: meta.reply });
+          }
+          // Apply any broadcast directive the brain requested (e.g. user said
+          // "방송 켜줘 / 꺼줘") before the reply audio plays. The meta line
+          // always precedes the audio clips, so awaiting here preserves order.
+          if (meta.broadcast_action && this.onBroadcastAction) {
+            try {
+              await this.onBroadcastAction(meta.broadcast_action);
+            } catch (e) {
+              console.error("[voice] broadcast action failed:", e);
+            }
+          }
+        },
+        onAudio: (clip) => this.play(clip),
+      });
     } catch (err) {
       console.error("[voice] converse failed:", err);
     }