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?.());
+      // Streaming turn: the brain sends transcript/reply first, then one audio
-      if (result.transcript) {
+      // clip per sentence as it is synthesised. We enqueue each clip on arrival
-        this.onTurn?.({ user: userId, transcript: result.transcript, reply: result.reply });
+      // 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 playing the reply.
+          // "방송 켜줘 / 꺼줘") before the reply audio plays. The meta line
-      if (result.broadcast_action && this.onBroadcastAction) {
+          // always precedes the audio clips, so awaiting here preserves order.
+          if (meta.broadcast_action && this.onBroadcastAction) {
             try {
-          await this.onBroadcastAction(result.broadcast_action);
+              await this.onBroadcastAction(meta.broadcast_action);
             } catch (e) {
               console.error("[voice] broadcast action failed:", e);
             }
           }
-      const audio = decodeWav(result.audio_b64);
+        },
-      if (audio) this.play(audio);
+        onAudio: (clip) => this.play(clip),
+      });
     } catch (err) {
       console.error("[voice] converse failed:", err);
     }

bridge/server.py

@@ -37,13 +37,22 @@ import wave
 from pathlib import Path
 from typing import Optional
 
-# Ensure repo-root/src is importable (jarvis package lives in src/jarvis)
+# Ensure repo-root/src is importable (jarvis package lives in src/jarvis) and
+# the repo root itself (so ``bridge.text_utils`` resolves whether this module is
+# launched as ``python -m bridge.server`` or ``python bridge/server.py``).
 _REPO_ROOT = Path(__file__).resolve().parent.parent
 _SRC = _REPO_ROOT / "src"
 if str(_SRC) not in sys.path:
     sys.path.insert(0, str(_SRC))
+if str(_REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(_REPO_ROOT))
 
-from flask import Flask, request, jsonify
+from flask import Flask, request, jsonify, Response, stream_with_context
+
+try:  # package-relative when imported as ``bridge.server``
+    from bridge.text_utils import split_sentences
+except ImportError:  # script-relative when run as ``bridge/server.py``
+    from text_utils import split_sentences
 
 app = Flask(__name__)
 
@@ -372,6 +381,59 @@ def http_converse():
     )
 
 
+@app.post("/converse_stream")
+def http_converse_stream():
+    """Streaming full turn: speech in -> transcript -> reply -> speech out.
+
+    Reduces perceived latency by synthesising the reply one sentence at a time
+    and emitting each clip as soon as it is ready, so the Discord layer can play
+    the first sentence while the rest are still being spoken. The response is
+    newline-delimited JSON (NDJSON):
+
+        {"type":"meta","transcript":..,"language":..,"reply":..,"error":..,"broadcast_action":..}
+        {"type":"audio","seq":0,"audio_b64":..}
+        {"type":"audio","seq":1,"audio_b64":..}
+        {"type":"end"}
+
+    STT and the reply engine still run to completion before the meta line; only
+    TTS is pipelined. The non-streaming /converse endpoint is unchanged.
+    """
+    raw = request.get_data()
+    if not raw:
+        return jsonify({"error": "empty body; send a WAV blob"}), 400
+    broadcasting = _coerce_bool(request.args.get("broadcasting"))
+
+    def gen():
+        stt = transcribe(raw)
+        transcript = stt.get("text", "")
+        if not transcript:
+            yield json.dumps({"type": "meta", "transcript": "", "language": stt.get("language"),
+                              "reply": "", "error": stt.get("error"), "broadcast_action": None}) + "\n"
+            yield json.dumps({"type": "end"}) + "\n"
+            return
+        result = think(transcript, stt.get("language"), broadcasting)
+        reply = result.get("reply", "")
+        yield json.dumps({
+            "type": "meta",
+            "transcript": transcript,
+            "language": stt.get("language"),
+            "reply": reply,
+            "error": result.get("error"),
+            "broadcast_action": result.get("broadcast_action"),
+        }) + "\n"
+        for seq, sentence in enumerate(split_sentences(reply)):
+            audio = synthesize(sentence)
+            if audio:
+                yield json.dumps({
+                    "type": "audio",
+                    "seq": seq,
+                    "audio_b64": base64.b64encode(audio).decode("ascii"),
+                }) + "\n"
+        yield json.dumps({"type": "end"}) + "\n"
+
+    return Response(stream_with_context(gen()), mimetype="application/x-ndjson")
+
+
 def main():
     print(f"[bridge] listening on http://{BRIDGE_HOST}:{BRIDGE_PORT}", flush=True)
     # threaded=True so STT (slow) on one request doesn't block /health, etc.

bridge/text_utils.py

@@ -0,0 +1,50 @@
+"""Small, dependency-free text helpers for the brain bridge.
+
+Kept separate from ``bridge.server`` (which imports Flask and the heavy brain)
+so the pure logic here can be unit-tested in isolation.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import List, Optional
+
+# Sentence terminators across scripts: ASCII . ! ? plus the CJK fullwidth forms
+# and the ellipsis. Runs of terminators ("?!", "...") collapse into one boundary.
+# A run of newlines is also a boundary. This is punctuation-only and therefore
+# language-agnostic (no hardcoded words), per the project's multilingual rule.
+_BOUNDARY = re.compile(r"([.!?。！？…]+|\n+)")
+
+
+def split_sentences(text: Optional[str], min_len: int = 5) -> List[str]:
+    """Split ``text`` into sentence-sized chunks for streaming TTS.
+
+    Each chunk ends at a sentence boundary so it can be synthesised and played
+    while later chunks are still being spoken. Fragments shorter than
+    ``min_len`` characters (interjections like "네.", "응") are merged into an
+    adjacent chunk so we don't emit choppy micro-clips. Returns an empty list
+    for blank input and never loses visible content.
+    """
+    text = (text or "").strip()
+    if not text:
+        return []
+
+    parts = _BOUNDARY.split(text)
+    chunks: List[str] = []
+    buf = ""
+    for i in range(0, len(parts), 2):
+        seg = parts[i]
+        delim = parts[i + 1] if i + 1 < len(parts) else ""
+        buf += seg + delim
+        # Flush at a real boundary once the buffer is a worthwhile clip.
+        if delim and len(buf.strip()) >= min_len:
+            chunks.append(buf.strip())
+            buf = ""
+
+    tail = buf.strip()
+    if tail:
+        if chunks and len(tail) < min_len:
+            chunks[-1] = chunks[-1] + " " + tail
+        else:
+            chunks.append(tail)
+    return chunks

src/jarvis/memory/embeddings.py

@@ -6,11 +6,14 @@ def get_embedding(text: str, base_url: str, model: str, timeout_sec: float = 15.
     try:
         resp = requests.post(
             f"{base_url.rstrip('/')}/api/embeddings",
-            # keep_alive=0 unloads the embedding model right after the call so
+            # Short positive keep_alive keeps the embed model warm across the
-            # it does not sit resident in VRAM alongside the chat model. The
+            # consecutive turns of an active conversation. With keep_alive=0
-            # chat model is pinned separately (llm.py keep_alive=30m); only the
+            # Ollama unloads it ~2s after every call, so each turn after a brief
-            # actively-used chat model should stay loaded.
+            # idle gap pays a cold reload of the embed model. The embed model is
-            json={"model": model, "prompt": text, "keep_alive": 0},
+            # tiny (~0.3 GB) and coexists in VRAM with the chat model (pinned at
+            # keep_alive=30m in llm.py) with ample headroom, so holding it for a
+            # few minutes is effectively free and removes the per-turn reload.
+            json={"model": model, "prompt": text, "keep_alive": "5m"},
             timeout=timeout_sec,
         )
         resp.raise_for_status()

tests/test_bridge_sentence_split.py

@@ -0,0 +1,77 @@
+"""Unit tests for the bridge sentence splitter that drives streaming TTS.
+
+The splitter is the only new logic on the bridge's streaming path: it chops a
+reply into sentence-sized chunks so the first sentence can be synthesised and
+played while the rest are still being spoken. It must be language-agnostic
+(punctuation only, no hardcoded words) per the project rule.
+"""
+
+import pytest
+
+from bridge.text_utils import split_sentences
+
+
+@pytest.mark.unit
+def test_empty_text_yields_no_chunks():
+    assert split_sentences("") == []
+    assert split_sentences("   ") == []
+    assert split_sentences(None) == []  # type: ignore[arg-type]
+
+
+@pytest.mark.unit
+def test_text_without_terminal_punctuation_is_one_chunk():
+    assert split_sentences("오늘 날씨 맑음") == ["오늘 날씨 맑음"]
+
+
+@pytest.mark.unit
+def test_splits_on_sentence_ending_punctuation():
+    chunks = split_sentences("안녕하세요. 반갑습니다!")
+    assert chunks == ["안녕하세요.", "반갑습니다!"]
+
+
+@pytest.mark.unit
+def test_splits_on_fullwidth_cjk_punctuation():
+    chunks = split_sentences("これはペンです。あれは何ですか？")
+    assert chunks == ["これはペンです。", "あれは何ですか？"]
+
+
+@pytest.mark.unit
+def test_splits_english_sentences():
+    chunks = split_sentences("Hello there. How are you? I am fine!")
+    assert chunks == ["Hello there.", "How are you?", "I am fine!"]
+
+
+@pytest.mark.unit
+def test_short_leading_fragment_merges_forward():
+    # "네." is below the min length, so it should ride along with the next
+    # sentence rather than become its own micro-clip.
+    chunks = split_sentences("네. 지금 바로 처리하겠습니다.")
+    assert chunks == ["네. 지금 바로 처리하겠습니다."]
+
+
+@pytest.mark.unit
+def test_short_trailing_fragment_merges_backward():
+    chunks = split_sentences("지금 바로 처리하겠습니다. 응")
+    assert chunks == ["지금 바로 처리하겠습니다. 응"]
+
+
+@pytest.mark.unit
+def test_newline_is_a_boundary():
+    chunks = split_sentences("첫 번째 줄입니다\n두 번째 줄입니다")
+    assert chunks == ["첫 번째 줄입니다", "두 번째 줄입니다"]
+
+
+@pytest.mark.unit
+def test_chunks_preserve_all_visible_content_in_order():
+    text = "안녕하세요. 오늘 일정 알려드릴게요. 회의가 세 개 있습니다!"
+    chunks = split_sentences(text)
+    assert len(chunks) >= 2
+    # No content lost: stripping spaces, the concatenation matches the source.
+    joined = "".join(chunks)
+    assert joined.replace(" ", "") == text.replace(" ", "")
+
+
+@pytest.mark.unit
+def test_collapses_repeated_terminators_into_one_chunk():
+    chunks = split_sentences("정말요?! 네 맞습니다.")
+    assert chunks == ["정말요?!", "네 맞습니다."]

tests/test_embeddings.py

@@ -0,0 +1,52 @@
+"""Tests for the Ollama embedding client.
+
+Behaviour under test: the embedding request keeps the embed model warm across
+consecutive conversation turns. With ``keep_alive=0`` Ollama unloads the embed
+model ~2s after every call, so each turn after a short idle gap pays a cold
+reload. A short positive ``keep_alive`` keeps it resident between turns at a
+negligible VRAM cost (nomic-embed-text is ~0.3 GB).
+"""
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from jarvis.memory.embeddings import get_embedding
+
+
+def _mock_response(vec):
+    resp = MagicMock()
+    resp.raise_for_status.return_value = None
+    resp.json.return_value = {"embedding": vec}
+    return resp
+
+
+def test_get_embedding_posts_to_embeddings_endpoint():
+    with patch("jarvis.memory.embeddings.requests") as mock_requests:
+        mock_requests.post.return_value = _mock_response([0.1, 0.2, 0.3])
+        vec = get_embedding("hello", "http://localhost:11434", "nomic-embed-text")
+
+    assert vec == [0.1, 0.2, 0.3]
+    args, kwargs = mock_requests.post.call_args
+    assert args[0].endswith("/api/embeddings")
+    assert kwargs["json"]["model"] == "nomic-embed-text"
+    assert kwargs["json"]["prompt"] == "hello"
+
+
+def test_get_embedding_keeps_model_warm_between_turns():
+    """The request must not unload the model after each call (keep_alive > 0)."""
+    with patch("jarvis.memory.embeddings.requests") as mock_requests:
+        mock_requests.post.return_value = _mock_response([0.0])
+        get_embedding("warm me", "http://localhost:11434", "nomic-embed-text")
+
+    _, kwargs = mock_requests.post.call_args
+    keep_alive = kwargs["json"].get("keep_alive")
+    # A falsy/zero keep_alive evicts the model immediately, forcing a cold
+    # reload on the next turn. Anything truthy positive keeps it resident.
+    assert keep_alive, f"embedding keep_alive should be a positive duration, got {keep_alive!r}"
+    assert keep_alive != 0
+
+
+def test_get_embedding_returns_none_on_error():
+    with patch("jarvis.memory.embeddings.requests") as mock_requests:
+        mock_requests.post.side_effect = RuntimeError("boom")
+        assert get_embedding("x", "http://localhost:11434", "nomic-embed-text") is None