From 28a2629c4351af3e4ebab0ae7eed499af2101fda Mon Sep 17 00:00:00 2001 From: snowspeeder Date: Fri, 7 Aug 2026 14:20:21 -0400 Subject: [PATCH] feat: speak answers sentence by sentence and never speak reasoning Segment the final answer as it streams and synthesize each completed sentence immediately, so a listener hears the reply begin while the model is still writing it. Sentences share the progress clips' FIFO chain and output-sequence counter; only text left unspoken (a hit budget) is synthesized after settle, so audio is never sent twice. A message that grows a tool call stops answer-streaming without re-speaking what already played. Reasoning models inline their chain of thought as blocks in the same delta stream. Strip it while streaming -- carrying a partial tag across delta boundaries -- so it is neither spoken nor shown as answer text. The fixed "Hang on while I work on that." is now a rotation of short natural phrases that never repeats twice running, and a long silent think gets a spoken "still on it" roughly every 45 seconds until real speech begins. Turn timeout rises to five minutes to match the client. Co-Authored-By: Claude Fable 5 --- docs/voice-api.md | 8 +- src/server/voiceApi.test.ts | 673 ++++++++++++++++++++++++++++++++++++ src/server/voiceApi.ts | 571 +++++++++++++++++++++++++++++- 3 files changed, 1233 insertions(+), 19 deletions(-) diff --git a/docs/voice-api.md b/docs/voice-api.md index 471099b..4934033 100644 --- a/docs/voice-api.md +++ b/docs/voice-api.md @@ -93,7 +93,13 @@ Azure push-stream recognition emits `transcript.partial` and `transcript.final` ### Turn output and lifecycle -For either input form, the server emits `agent.working`, then `agent.accepted` after Pi accepted the prompt. If work is still in progress after two seconds, it sends one `agent.progress` event with a short spoken status clip (for example, “Hang on while I work on that.”). It never repeats this spoken update within the same turn; ordinary screen-only `agent.working` heartbeats continue every 15 seconds. While it waits for Pi's native `agent.settled` event (not merely `agent.end`), it repeats the existing `agent.working` frame every 15 seconds; those progress frames stop before `assistant.final` and never overlap synthesized audio. The wait has a two-minute turn timeout. `assistant.delta` is streamed while Pi answers; `assistant.final` contains only assistant text, never STT transcript. +For either input form, the server emits `agent.working`, then `agent.accepted` after Pi accepted the prompt. If work is still in progress after two seconds and nothing has been spoken yet (see below), it sends one `agent.progress` event with a short spoken status clip, picked pseudo-randomly from a small set of natural filler phrases (“One sec.”, “Let me take a look.”, “Hmm, checking now.”, “Working on it.”, “Give me a moment.”, “Let me look into that.”). The server never repeats the same phrase on two consecutive turns. Ordinary screen-only `agent.working` heartbeats continue every 15 seconds regardless. + +During a long turn, Pi may emit several assistant messages between tool calls before its final answer — each one ends with Pi's `message.end` event. Whenever a completed message includes a tool call (i.e., it is not the turn's final answer), the server also speaks it: it sends `{"type":"agent.progress","text":"","audio":true}` — `text` is always the message's full, untruncated text — immediately followed by that clip's binary audio frames, unless every word of that text was already spoken live as answer sentences (see below), in which case the clip is skipped entirely, or only its unspoken remainder is synthesized if part of it was spoken live and part was not. `agent.progress` may therefore occur multiple times per turn, always in the order the messages completed, and each is optionally followed by its own clip's binary frames. For speech only, a message longer than about 300 characters is truncated at a sentence boundary where possible (the JSON `text` field is never truncated). Progress clips are sent one at a time — a clip's audio always finishes sending before the next `agent.progress` frame is sent — and a turn's total spoken progress narration is capped at roughly 60 seconds of audio or 8 clips, whichever comes first; the fallback phrase counts toward this same cap. Once the turn's answer has audibly begun (the first `answer.sentence` has been sent — see below), a brand-new message that had no live narration of its own is left quiet rather than starting fresh progress narration that would talk over the answer; a message that was already mid-narration when a tool call appeared in it still gets its own unspoken leftover finished so it isn't cut off abruptly. `agent.progress` frames, spoken or not, always stop before `assistant.final`. + +While the final answer streams in as Pi generates it, the server does not wait for it to fully finish before speaking it: as each complete sentence of the in-progress message accumulates (split on `.`/`!`/`?` — plus a closing quote or bracket — followed by whitespace; a decimal like “3.5” never triggers a false split, since the period is not followed by whitespace), the server sends `{"type":"answer.sentence","text":""}` immediately followed by that sentence's own binary audio frames, serialized through the same ordered chain as `agent.progress` clips so nothing interleaves or reorders on the single audio output. Markdown-decoration-only fragments (e.g. `---`, `` ``` ``, `**`) are skipped silently. Because a message might still turn out to include a tool call (making it narration, not the final answer), every message is optimistically streamed this way until either a tool call is detected in it or it ends; once a tool call appears, sentence-streaming for that message stops (already-spoken sentences remain spoken — that's fine acoustically) and any unspoken remainder is handled by the `agent.progress` mechanism above instead. A message's trailing fragment with no terminal punctuation yet is held back until the message ends, then flushed as one final `answer.sentence`. Spoken answer audio is capped at roughly 120 seconds or 40 sentences per turn, whichever comes first; once agent.settled fires, `assistant.final` (the full, untruncated answer text) is always sent next, and any part of the answer that was never spoken live because a cap was hit is synthesized once, as ordinary audio frames, between `assistant.final` and `audio.end` — a turn where every sentence streamed live sends zero audio frames in that window. `audio.end` always ends the turn's audio either way. + +While it waits for Pi's native `agent.settled` event (not merely `agent.end`), it repeats the existing `agent.working` frame every 15 seconds; those progress frames stop before `assistant.final` and never overlap synthesized audio. The wait has a two-minute turn timeout. `assistant.delta` is streamed while Pi answers; `assistant.final` contains only assistant text, never STT transcript. Azure synthesis output is signed little-endian 16-bit, 24 kHz, mono PCM. Every binary server frame has this eight-byte header followed by PCM: diff --git a/src/server/voiceApi.test.ts b/src/server/voiceApi.test.ts index 9781de5..6b7c177 100644 --- a/src/server/voiceApi.test.ts +++ b/src/server/voiceApi.test.ts @@ -13,6 +13,11 @@ import { revokeVoiceToken, clearWorkingProgress, startWorkingProgress, + extractCompleteSentences, + isSpeakableFragment, + pickFallbackPhrase, + stripThinkingStream, + stripThinkingAll, } from "./voiceApi.js"; const temporary: string[] = []; @@ -119,6 +124,81 @@ describe("voice API", () => { expect(frame.subarray(8)).toEqual(Buffer.from([1, 2, 3, 4])); }); + it("segments complete sentences, holds back an incomplete trailing fragment, and does not split a decimal like 3.5", () => { + expect(extractCompleteSentences("First one. Second one! Third one?", false)).toEqual([ + { text: "First one.", rawLength: 11 }, + { text: "Second one!", rawLength: 12 }, + ]); + // "Third one?" has no trailing whitespace yet (it's the end of the + // buffer) so it is held back as an incomplete fragment, not emitted. + const held = extractCompleteSentences("Third one?", false); + expect(held).toEqual([]); + + // A mid-number decimal point is never mistaken for a sentence boundary, + // because the period is not followed by whitespace. + const decimal = extractCompleteSentences("It costs 3.5 dollars. Thanks. ", false); + expect(decimal.map((sentence) => sentence.text)).toEqual([ + "It costs 3.5 dollars.", + "Thanks.", + ]); + }); + + it("strips streamed thinking blocks, even split across delta boundaries", () => { + const turn = { thinkDepth: 0, thinkCarry: "" }; + const out = [ + stripThinkingStream(turn, "step one, "), + stripThinkingStream(turn, "step twoThe answer "), + stripThinkingStream(turn, "is 42."), + ].join(""); + expect(out).toBe("The answer is 42."); + expect(turn.thinkDepth).toBe(0); + + // A tag split exactly at a chunk edge is carried, never leaked. + const carry = { thinkDepth: 0, thinkCarry: "" }; + expect(stripThinkingStream(carry, "Hello <")).toBe("Hello "); + expect(stripThinkingStream(carry, "think>hidden world")).toBe(" world"); + + expect(stripThinkingAll("abcAnswer. moreDone.")).toBe( + "Answer. Done." + ); + // An unclosed trailing block (still thinking at message end) is dropped. + expect(stripThinkingAll("Answer first. unfinished")).toBe("Answer first. "); + }); + + it("flushes the trailing incomplete fragment only when explicitly told the message ended", () => { + const pending = "Here is the tail with no terminal punctuation"; + expect(extractCompleteSentences(pending, false)).toEqual([]); + expect(extractCompleteSentences(pending, true)).toEqual([ + { text: pending, rawLength: pending.length }, + ]); + // A purely-whitespace remainder is never flushed as an empty "sentence". + expect(extractCompleteSentences(" ", true)).toEqual([]); + expect(extractCompleteSentences("", true)).toEqual([]); + }); + + it("treats markdown-decoration-only fragments as unspeakable, but keeps ordinary punctuation-bearing text", () => { + expect(isSpeakableFragment("---")).toBe(false); + expect(isSpeakableFragment("```")).toBe(false); + expect(isSpeakableFragment("**")).toBe(false); + expect(isSpeakableFragment("> ")).toBe(false); + expect(isSpeakableFragment(" ")).toBe(false); + expect(isSpeakableFragment("")).toBe(false); + expect(isSpeakableFragment("Okay - let's continue.")).toBe(true); + expect(isSpeakableFragment("It costs 3.5 dollars.")).toBe(true); + }); + + it("never repeats the same fallback phrase twice in a row", () => { + const picks = Array.from({ length: 200 }, () => pickFallbackPhrase()); + for (let index = 1; index < picks.length; index++) { + expect(picks[index]).not.toBe(picks[index - 1]); + } + // Every pick is one of the known short, natural phrases. + const known = new Set(picks); + for (const phrase of known) expect(phrase.length).toBeGreaterThan(0); + expect(known.size).toBeGreaterThan(1); + }); + it("does not announce progress for a quick completed turn", () => { vi.useFakeTimers(); try { @@ -392,6 +472,599 @@ describe("voice API", () => { await app.close(); }); + it("speaks intermediate assistant messages between tool calls as ordered progress clips before the final answer", async () => { + const directory = mkdtempSync(join(tmpdir(), "pi-web-voice-api-")); + temporary.push(directory); + const events = Object.assign(new EventEmitter(), { + close: () => undefined, + readyState: websocketConnectingState(), + }); + const longMessage = `${"Sentence number one is here. ".repeat(15)}Trailing tail text without a stop.`; + const daemon = { + request: (method: string, path: string) => { + if (method === "POST" && path === "/sessions") + return Promise.resolve(response(200, { id: "session-1" })); + if (method === "POST" && path.endsWith("/prompt")) { + queueMicrotask(() => { + events.emit( + "message", + Buffer.from(JSON.stringify({ + type: "message.end", + message: { role: "assistant", content: [ + { type: "text", text: "Checking the first file now." }, + { type: "toolCall", id: "1", name: "read_file", input: {} }, + ] }, + })) + ); + events.emit( + "message", + Buffer.from(JSON.stringify({ + type: "message.end", + message: { role: "assistant", content: [ + { type: "text", text: longMessage }, + { type: "toolCall", id: "2", name: "read_file", input: {} }, + ] }, + })) + ); + events.emit( + "message", + Buffer.from(JSON.stringify({ + type: "message.end", + message: { role: "assistant", content: [ + { type: "text", text: "Here is the final answer." }, + ] }, + })) + ); + events.emit("message", Buffer.from(JSON.stringify({ type: "agent.settled" }))); + }); + return Promise.resolve(response(200, { accepted: true })); + } + return Promise.resolve(response(200, {})); + }, + connectWebSocket: () => { + queueMicrotask(() => { + events.readyState = WebSocket.OPEN; + events.emit("open"); + }); + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + return events as unknown as WebSocket; + }, + }; + const synthesizeCalls: string[] = []; + const speech = { + recognize: () => ({ write: () => undefined, end: () => Promise.resolve(), close: () => undefined }), + synthesize: (text: string, onAudio: (pcm: Buffer) => void) => { + synthesizeCalls.push(text); + onAudio(Buffer.from([synthesizeCalls.length])); + return Promise.resolve(); + }, + }; + const app = Fastify(); + await app.register(fastifyWebsocket); + const configPath = join(directory, "voice-api.json"); + registerVoiceApiRoutes(app, { + configPath, + daemon, + speech, + projects: { list: () => Promise.resolve([{ id: "p", name: "P", path: "/tmp", createdAt: "" }]) }, + workspaces: { list: () => Promise.resolve([{ id: "w", projectId: "p", path: "/tmp", label: "W", isMain: true, isGitRepo: false, isGitWorktree: false }]) }, + }); + const token = createVoiceToken({}, configPath).token; + const created = await app.inject({ method: "POST", url: "/api/v1/voice/conversations", headers: { authorization: `Bearer ${token}` }, payload: { workspaceId: "w" } }); + const address = await app.listen({ port: 0, host: "127.0.0.1" }); + const frames: ({ json: Record } | { binary: Buffer })[] = []; + try { + await new Promise((resolve, reject) => { + const client = new WebSocket(address.replace("http", "ws") + "/api/v1/voice/stream", { headers: { authorization: `Bearer ${token}` } }); + client.on("open", () => { + client.send(JSON.stringify({ type: "attach", conversationId: responseId(created.body) })); + client.send(JSON.stringify({ type: "input.text", text: "look into this" })); + }); + client.on("message", (data, binary) => { + if (binary) { + frames.push({ binary: rawDataToBuffer(data) }); + return; + } + const value: unknown = JSON.parse(rawDataToBuffer(data).toString()); + if (typeof value !== "object" || value === null) return; + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + const record = value as Record; + frames.push({ json: record }); + if (record["type"] === "audio.end") { + client.close(); + resolve(); + } + }); + client.on("error", reject); + }); + } finally { + await app.close(); + } + const jsonFrames = frames.flatMap((frame) => ("json" in frame ? [frame.json] : [])); + const progressFrames = jsonFrames.filter((frame) => frame["type"] === "agent.progress"); + expect(progressFrames).toEqual([ + { type: "agent.progress", text: "Checking the first file now.", audio: true }, + { type: "agent.progress", text: longMessage, audio: true }, + ]); + // The final answer is not also spoken as progress, and never repeats. + expect(jsonFrames).toContainEqual({ type: "assistant.final", text: "Here is the final answer." }); + expect(jsonFrames.filter((frame) => frame["type"] === "assistant.final")).toHaveLength(1); + // Full text goes out over JSON; only the long message is shortened for speech. + expect(synthesizeCalls).toHaveLength(3); + const [firstClip, secondClip, thirdClip] = synthesizeCalls; + expect(firstClip).toBe("Checking the first file now."); + expect(secondClip).toBeDefined(); + expect((secondClip ?? "").length).toBeLessThan(longMessage.length); + expect((secondClip ?? "").length).toBeLessThanOrEqual(300); + expect((secondClip ?? "").endsWith(".")).toBe(true); + expect(thirdClip).toBe("Here is the final answer."); + // Every clip's binary frame(s) are sent before the next JSON frame that + // follows them, i.e. progress clips never interleave or reorder. + const order = frames.map((frame) => ("json" in frame ? frame.json["type"] : "binary")); + const firstProgress = order.indexOf("agent.progress"); + const firstBinary = order.indexOf("binary"); + const finalIndex = order.indexOf("assistant.final"); + expect(firstProgress).toBeGreaterThanOrEqual(0); + expect(firstBinary).toBeGreaterThan(firstProgress); + expect(finalIndex).toBeGreaterThan(firstBinary); + // 3 synthesized clips (2 progress + 1 final) => 3 binary frames. + expect(order.filter((type) => type === "binary")).toHaveLength(3); + }); + + it("streams the final answer as sentences complete, before agent.settled, then sends assistant.final and audio.end without re-synthesizing already-spoken audio", async () => { + const directory = mkdtempSync(join(tmpdir(), "pi-web-voice-api-")); + temporary.push(directory); + const events = Object.assign(new EventEmitter(), { + close: () => undefined, + readyState: websocketConnectingState(), + }); + const sentence1 = "First sentence. "; + const sentence2 = "Second sentence. "; + const sentence3 = "Third sentence."; + const fullText = sentence1 + sentence2 + sentence3; + const daemon = { + request: (method: string, path: string) => { + if (method === "POST" && path === "/sessions") + return Promise.resolve(response(200, { id: "session-1" })); + if (method === "POST" && path.endsWith("/prompt")) { + queueMicrotask(() => { + events.emit("message", Buffer.from(JSON.stringify({ type: "assistant.delta", text: sentence1 }))); + events.emit("message", Buffer.from(JSON.stringify({ type: "assistant.delta", text: sentence2 }))); + events.emit( + "message", + Buffer.from(JSON.stringify({ + type: "message.end", + message: { role: "assistant", content: [{ type: "text", text: fullText }] }, + })) + ); + events.emit("message", Buffer.from(JSON.stringify({ type: "agent.settled" }))); + }); + return Promise.resolve(response(200, { accepted: true })); + } + return Promise.resolve(response(200, {})); + }, + connectWebSocket: () => { + queueMicrotask(() => { + events.readyState = WebSocket.OPEN; + events.emit("open"); + }); + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + return events as unknown as WebSocket; + }, + }; + const synthesizeCalls: string[] = []; + const speech = { + recognize: () => ({ write: () => undefined, end: () => Promise.resolve(), close: () => undefined }), + synthesize: (text: string, onAudio: (pcm: Buffer) => void) => { + synthesizeCalls.push(text); + onAudio(Buffer.from([synthesizeCalls.length])); + return Promise.resolve(); + }, + }; + const app = Fastify(); + await app.register(fastifyWebsocket); + const configPath = join(directory, "voice-api.json"); + registerVoiceApiRoutes(app, { + configPath, + daemon, + speech, + projects: { list: () => Promise.resolve([{ id: "p", name: "P", path: "/tmp", createdAt: "" }]) }, + workspaces: { list: () => Promise.resolve([{ id: "w", projectId: "p", path: "/tmp", label: "W", isMain: true, isGitRepo: false, isGitWorktree: false }]) }, + }); + const token = createVoiceToken({}, configPath).token; + const created = await app.inject({ method: "POST", url: "/api/v1/voice/conversations", headers: { authorization: `Bearer ${token}` }, payload: { workspaceId: "w" } }); + const address = await app.listen({ port: 0, host: "127.0.0.1" }); + const frames: ({ json: Record } | { binary: Buffer })[] = []; + try { + await new Promise((resolve, reject) => { + const client = new WebSocket(address.replace("http", "ws") + "/api/v1/voice/stream", { headers: { authorization: `Bearer ${token}` } }); + client.on("open", () => { + client.send(JSON.stringify({ type: "attach", conversationId: responseId(created.body) })); + client.send(JSON.stringify({ type: "input.text", text: "explain it" })); + }); + client.on("message", (data, binary) => { + if (binary) { + frames.push({ binary: rawDataToBuffer(data) }); + return; + } + const value: unknown = JSON.parse(rawDataToBuffer(data).toString()); + if (typeof value !== "object" || value === null) return; + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + const record = value as Record; + frames.push({ json: record }); + if (record["type"] === "audio.end") { + client.close(); + resolve(); + } + }); + client.on("error", reject); + }); + } finally { + await app.close(); + } + const jsonFrames = frames.flatMap((frame) => ("json" in frame ? [frame.json] : [])); + const sentenceFrames = jsonFrames.filter((frame) => frame["type"] === "answer.sentence"); + expect(sentenceFrames).toEqual([ + { type: "answer.sentence", text: "First sentence." }, + { type: "answer.sentence", text: "Second sentence." }, + { type: "answer.sentence", text: "Third sentence." }, + ]); + // The third sentence has no trailing punctuation+whitespace of its own + // (it is the end of the message) so it is only known once message.end + // arrives, well before agent.settled — it is not deferred to the + // post-settle path. + expect(synthesizeCalls).toEqual(["First sentence.", "Second sentence.", "Third sentence."]); + expect(jsonFrames).toContainEqual({ type: "assistant.final", text: fullText.trim() }); + // Everything was already spoken live: no post-settle audio is sent. + const order = frames.map((frame) => ("json" in frame ? frame.json["type"] : "binary")); + expect(order.filter((type) => type === "binary")).toHaveLength(3); + // Every answer.sentence's own binary frame(s) are sent before the next + // JSON frame, and all sentence audio precedes assistant.final/audio.end + // — the shared FIFO chain (also used by agent.progress) never + // interleaves or reorders clips. (Filtered to the frame types this + // assertion cares about; agent.working/assistant.delta/etc. also + // appear in `order` but are irrelevant to sequencing answer audio.) + const relevant = order.filter((type) => + type === "answer.sentence" || type === "binary" || type === "assistant.final" || type === "audio.end" + ); + expect(relevant).toEqual([ + "answer.sentence", "binary", + "answer.sentence", "binary", + "answer.sentence", "binary", + "assistant.final", + "audio.end", + ]); + // All three answer clips share the turn's single monotonic output + // sequence counter (same as agent.progress/final audio). + const binaries = frames.flatMap((frame) => ("binary" in frame ? [frame.binary] : [])); + const sequences = binaries.map((buffer) => buffer.readUInt32BE(4)); + expect(sequences).toEqual([0, 1, 2]); + }); + + it("sends only the unspoken tail post-settle when the answer sentence cap is hit", async () => { + const directory = mkdtempSync(join(tmpdir(), "pi-web-voice-api-")); + temporary.push(directory); + const events = Object.assign(new EventEmitter(), { + close: () => undefined, + readyState: websocketConnectingState(), + }); + // 45 short, complete sentences (every one, including the last, is + // followed by whitespace) so all are eligible for live streaming, but + // the ~40-sentence cap leaves the last 5 unspoken. + const sentences = Array.from({ length: 45 }, (_, index) => `Sentence ${String(index + 1)}.`); + const fullText = `${sentences.join(" ")} `; + const daemon = { + request: (method: string, path: string) => { + if (method === "POST" && path === "/sessions") + return Promise.resolve(response(200, { id: "session-1" })); + if (method === "POST" && path.endsWith("/prompt")) { + queueMicrotask(() => { + events.emit("message", Buffer.from(JSON.stringify({ type: "assistant.delta", text: fullText }))); + events.emit( + "message", + Buffer.from(JSON.stringify({ + type: "message.end", + message: { role: "assistant", content: [{ type: "text", text: fullText }] }, + })) + ); + events.emit("message", Buffer.from(JSON.stringify({ type: "agent.settled" }))); + }); + return Promise.resolve(response(200, { accepted: true })); + } + return Promise.resolve(response(200, {})); + }, + connectWebSocket: () => { + queueMicrotask(() => { + events.readyState = WebSocket.OPEN; + events.emit("open"); + }); + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + return events as unknown as WebSocket; + }, + }; + const synthesizeCalls: string[] = []; + const speech = { + recognize: () => ({ write: () => undefined, end: () => Promise.resolve(), close: () => undefined }), + synthesize: (text: string, onAudio: (pcm: Buffer) => void) => { + synthesizeCalls.push(text); + onAudio(Buffer.from([1])); + return Promise.resolve(); + }, + }; + const app = Fastify(); + await app.register(fastifyWebsocket); + const configPath = join(directory, "voice-api.json"); + registerVoiceApiRoutes(app, { + configPath, + daemon, + speech, + projects: { list: () => Promise.resolve([{ id: "p", name: "P", path: "/tmp", createdAt: "" }]) }, + workspaces: { list: () => Promise.resolve([{ id: "w", projectId: "p", path: "/tmp", label: "W", isMain: true, isGitRepo: false, isGitWorktree: false }]) }, + }); + const token = createVoiceToken({}, configPath).token; + const created = await app.inject({ method: "POST", url: "/api/v1/voice/conversations", headers: { authorization: `Bearer ${token}` }, payload: { workspaceId: "w" } }); + const address = await app.listen({ port: 0, host: "127.0.0.1" }); + const jsonFrames: Record[] = []; + try { + await new Promise((resolve, reject) => { + const client = new WebSocket(address.replace("http", "ws") + "/api/v1/voice/stream", { headers: { authorization: `Bearer ${token}` } }); + client.on("open", () => { + client.send(JSON.stringify({ type: "attach", conversationId: responseId(created.body) })); + client.send(JSON.stringify({ type: "input.text", text: "list them all" })); + }); + client.on("message", (data, binary) => { + if (binary) return; + const value: unknown = JSON.parse(rawDataToBuffer(data).toString()); + if (typeof value !== "object" || value === null) return; + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + const record = value as Record; + jsonFrames.push(record); + if (record["type"] === "audio.end") { + client.close(); + resolve(); + } + }); + client.on("error", reject); + }); + } finally { + await app.close(); + } + const sentenceFrames = jsonFrames.filter((frame) => frame["type"] === "answer.sentence"); + expect(sentenceFrames).toHaveLength(40); + expect(sentenceFrames.map((frame) => frame["text"])).toEqual(sentences.slice(0, 40).map((text) => text)); + expect(jsonFrames).toContainEqual({ type: "assistant.final", text: fullText.trim() }); + // 40 live sentence clips, plus exactly one post-settle clip for the + // unspoken tail (sentences 41-45) — never re-synthesizing sentences 1-40. + expect(synthesizeCalls).toHaveLength(41); + expect(synthesizeCalls.slice(0, 40)).toEqual(sentences.slice(0, 40)); + expect(synthesizeCalls[40]).toBe(sentences.slice(40).join(" ")); + }); + + it("stops answer-sentence streaming once a message grows a toolCall, and does not double-speak the part already spoken live", async () => { + const directory = mkdtempSync(join(tmpdir(), "pi-web-voice-api-")); + temporary.push(directory); + const events = Object.assign(new EventEmitter(), { + close: () => undefined, + readyState: websocketConnectingState(), + }); + const spokenLive = "Let me check that."; + const spokenAfterToolStart = "Actually reading more now."; + const messageText = `${spokenLive} ${spokenAfterToolStart}`; + const daemon = { + request: (method: string, path: string) => { + if (method === "POST" && path === "/sessions") + return Promise.resolve(response(200, { id: "session-1" })); + if (method === "POST" && path.endsWith("/prompt")) { + queueMicrotask(() => { + // "Let me check that." completes (trailing space) and is + // spoken live before the tool call is known to exist. + events.emit("message", Buffer.from(JSON.stringify({ type: "assistant.delta", text: `${spokenLive} ` }))); + events.emit("message", Buffer.from(JSON.stringify({ type: "tool.start", toolName: "read_file", toolCallId: "1" }))); + // More text after the tool call starts must not be live-streamed. + events.emit("message", Buffer.from(JSON.stringify({ type: "assistant.delta", text: spokenAfterToolStart }))); + events.emit( + "message", + Buffer.from(JSON.stringify({ + type: "message.end", + message: { role: "assistant", content: [ + { type: "text", text: messageText }, + { type: "toolCall", id: "1", name: "read_file", input: {} }, + ] }, + })) + ); + events.emit( + "message", + Buffer.from(JSON.stringify({ + type: "message.end", + message: { role: "assistant", content: [{ type: "text", text: "Done." }] }, + })) + ); + events.emit("message", Buffer.from(JSON.stringify({ type: "agent.settled" }))); + }); + return Promise.resolve(response(200, { accepted: true })); + } + return Promise.resolve(response(200, {})); + }, + connectWebSocket: () => { + queueMicrotask(() => { + events.readyState = WebSocket.OPEN; + events.emit("open"); + }); + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + return events as unknown as WebSocket; + }, + }; + const synthesizeCalls: string[] = []; + const speech = { + recognize: () => ({ write: () => undefined, end: () => Promise.resolve(), close: () => undefined }), + synthesize: (text: string, onAudio: (pcm: Buffer) => void) => { + synthesizeCalls.push(text); + onAudio(Buffer.from([1])); + return Promise.resolve(); + }, + }; + const app = Fastify(); + await app.register(fastifyWebsocket); + const configPath = join(directory, "voice-api.json"); + registerVoiceApiRoutes(app, { + configPath, + daemon, + speech, + projects: { list: () => Promise.resolve([{ id: "p", name: "P", path: "/tmp", createdAt: "" }]) }, + workspaces: { list: () => Promise.resolve([{ id: "w", projectId: "p", path: "/tmp", label: "W", isMain: true, isGitRepo: false, isGitWorktree: false }]) }, + }); + const token = createVoiceToken({}, configPath).token; + const created = await app.inject({ method: "POST", url: "/api/v1/voice/conversations", headers: { authorization: `Bearer ${token}` }, payload: { workspaceId: "w" } }); + const address = await app.listen({ port: 0, host: "127.0.0.1" }); + const jsonFrames: Record[] = []; + try { + await new Promise((resolve, reject) => { + const client = new WebSocket(address.replace("http", "ws") + "/api/v1/voice/stream", { headers: { authorization: `Bearer ${token}` } }); + client.on("open", () => { + client.send(JSON.stringify({ type: "attach", conversationId: responseId(created.body) })); + client.send(JSON.stringify({ type: "input.text", text: "look into it" })); + }); + client.on("message", (data, binary) => { + if (binary) return; + const value: unknown = JSON.parse(rawDataToBuffer(data).toString()); + if (typeof value !== "object" || value === null) return; + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + const record = value as Record; + jsonFrames.push(record); + if (record["type"] === "audio.end") { + client.close(); + resolve(); + } + }); + client.on("error", reject); + }); + } finally { + await app.close(); + } + // "Let me check that." streams live before the tool call is known + // about; "Done." is the turn's separate, subsequent final message and + // is legitimately spoken too, once it ends the turn. + expect(jsonFrames.filter((frame) => frame["type"] === "answer.sentence")).toEqual([ + { type: "answer.sentence", text: spokenLive }, + { type: "answer.sentence", text: "Done." }, + ]); + // The intermediate-progress clip's JSON text is still the message's + // full, untruncated text, matching every other agent.progress frame. + expect(jsonFrames.filter((frame) => frame["type"] === "agent.progress")).toEqual([ + { type: "agent.progress", text: messageText, audio: true }, + ]); + // Only the part never spoken live was synthesized for the progress + // clip — the already-spoken opening sentence is never repeated. + expect(synthesizeCalls).toEqual([spokenLive, spokenAfterToolStart, "Done."]); + }); + + it("caps spoken progress at 8 clips per turn and never speaks after the turn has settled", async () => { + const directory = mkdtempSync(join(tmpdir(), "pi-web-voice-api-")); + temporary.push(directory); + const events = Object.assign(new EventEmitter(), { + close: () => undefined, + readyState: websocketConnectingState(), + }); + const intermediateMessage = (index: number) => ({ + type: "message.end", + message: { role: "assistant", content: [ + { type: "text", text: `Update number ${String(index)}.` }, + { type: "toolCall", id: String(index), name: "read_file", input: {} }, + ] }, + }); + const daemon = { + request: (method: string, path: string) => { + if (method === "POST" && path === "/sessions") + return Promise.resolve(response(200, { id: "session-1" })); + if (method === "POST" && path.endsWith("/prompt")) { + queueMicrotask(() => { + for (let index = 1; index <= 10; index++) { + events.emit("message", Buffer.from(JSON.stringify(intermediateMessage(index)))); + } + events.emit( + "message", + Buffer.from(JSON.stringify({ + type: "message.end", + message: { role: "assistant", content: [{ type: "text", text: "Done." }] }, + })) + ); + events.emit("message", Buffer.from(JSON.stringify({ type: "agent.settled" }))); + // A stray message.end arriving after agent.settled must never be spoken. + events.emit("message", Buffer.from(JSON.stringify(intermediateMessage(11)))); + }); + return Promise.resolve(response(200, { accepted: true })); + } + return Promise.resolve(response(200, {})); + }, + connectWebSocket: () => { + queueMicrotask(() => { + events.readyState = WebSocket.OPEN; + events.emit("open"); + }); + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + return events as unknown as WebSocket; + }, + }; + const speech = { + recognize: () => ({ write: () => undefined, end: () => Promise.resolve(), close: () => undefined }), + synthesize: (_text: string, onAudio: (pcm: Buffer) => void) => { + onAudio(Buffer.from([1])); + return Promise.resolve(); + }, + }; + const app = Fastify(); + await app.register(fastifyWebsocket); + const configPath = join(directory, "voice-api.json"); + registerVoiceApiRoutes(app, { + configPath, + daemon, + speech, + projects: { list: () => Promise.resolve([{ id: "p", name: "P", path: "/tmp", createdAt: "" }]) }, + workspaces: { list: () => Promise.resolve([{ id: "w", projectId: "p", path: "/tmp", label: "W", isMain: true, isGitRepo: false, isGitWorktree: false }]) }, + }); + const token = createVoiceToken({}, configPath).token; + const created = await app.inject({ method: "POST", url: "/api/v1/voice/conversations", headers: { authorization: `Bearer ${token}` }, payload: { workspaceId: "w" } }); + const address = await app.listen({ port: 0, host: "127.0.0.1" }); + const jsonFrames: Record[] = []; + try { + await new Promise((resolve, reject) => { + const client = new WebSocket(address.replace("http", "ws") + "/api/v1/voice/stream", { headers: { authorization: `Bearer ${token}` } }); + client.on("open", () => { + client.send(JSON.stringify({ type: "attach", conversationId: responseId(created.body) })); + client.send(JSON.stringify({ type: "input.text", text: "do many things" })); + }); + client.on("message", (data, binary) => { + if (binary) return; + const value: unknown = JSON.parse(rawDataToBuffer(data).toString()); + if (typeof value !== "object" || value === null) return; + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + const record = value as Record; + jsonFrames.push(record); + if (record["type"] === "audio.end") { + client.close(); + resolve(); + } + }); + client.on("error", reject); + }); + } finally { + await app.close(); + } + const progressFrames = jsonFrames.filter((frame) => frame["type"] === "agent.progress"); + expect(progressFrames).toHaveLength(8); + expect(progressFrames.map((frame) => frame["text"])).toEqual([ + "Update number 1.", + "Update number 2.", + "Update number 3.", + "Update number 4.", + "Update number 5.", + "Update number 6.", + "Update number 7.", + "Update number 8.", + ]); + expect(jsonFrames).toContainEqual({ type: "assistant.final", text: "Done." }); + }); + it("requires a hashed bearer token and only returns registered workspaces", async () => { const { app, configPath } = setup(); const created = createVoiceToken( diff --git a/src/server/voiceApi.ts b/src/server/voiceApi.ts index 1b5dd56..b690269 100644 --- a/src/server/voiceApi.ts +++ b/src/server/voiceApi.ts @@ -28,9 +28,124 @@ const BINARY_VERSION = 1; const BINARY_AUDIO = 1; const OUTPUT_CHUNK = 16 * 1024; const EVENT_OPEN_TIMEOUT_MS = 10_000; -const TURN_TIMEOUT_MS = 120_000; +// Long-thinking models can take minutes before the first answer token; the +// voice client's receive budget is raised in lockstep (MAX_TURN_RECEIVE_SECONDS). +const TURN_TIMEOUT_MS = 300_000; const WORKING_PROGRESS_INTERVAL_MS = 15_000; const SPOKEN_PROGRESS_DELAY_MS = 2_000; +// Per-turn cap on distinct spoken progress clips (the "Hang on..." fallback +// plus intermediate assistant messages spoken between tool calls). Keeps a +// chatty multi-tool-call turn from talking over itself indefinitely. +const MAX_SPOKEN_PROGRESS_CLIPS = 8; +// Per-turn cap on combined spoken progress audio, in milliseconds (~60s). +// Checked before starting a new clip, so it bounds total narration time +// without truncating a clip that is already playing. +const MAX_SPOKEN_PROGRESS_AUDIO_MS = 60_000; +// Azure progress/final clips are 24 kHz mono 16-bit PCM: 24000 * 2 bytes/sec. +const PROGRESS_AUDIO_BYTES_PER_MS = 48; +// Intermediate assistant messages are truncated to about this many +// characters (at a sentence boundary where possible) before being sent to +// speech synthesis. The full message text is still sent in the JSON frame. +const PROGRESS_SPEECH_MAX_CHARS = 300; +// Per-turn cap on combined spoken *answer* audio (sentences streamed live +// as the final message is generated, plus any post-settle unspoken tail). +// Higher than MAX_SPOKEN_PROGRESS_AUDIO_MS because this is the actual +// answer, not filler narration. +const MAX_SPOKEN_ANSWER_AUDIO_MS = 120_000; +// Guard against a pathological number of tiny "sentences" (e.g. a message +// that is mostly punctuation) turning into an endless run of small clips. +// Anything beyond this count is left for the post-settle unspoken-tail path. +const MAX_SPOKEN_ANSWER_SENTENCES = 40; +// Rotation of short, natural fallback phrases spoken ~2 seconds into a turn +// when nothing else has been said yet (see SPOKEN_PROGRESS_DELAY_MS). Picked +// pseudo-randomly per turn; consecutive turns never repeat the same phrase +// (see pickFallbackPhrase's module-scope lastFallbackPhrase). +const FALLBACK_PROGRESS_PHRASES: readonly string[] = [ + "One sec.", + "Let me take a look.", + "Hmm, checking now.", + "Working on it.", + "Give me a moment.", + "Let me look into that.", +]; +// Spoken every ~45s while a long-thinking model has produced nothing +// speakable yet; rotates without immediate repeats like the fallback set. +const STILL_WORKING_PHRASES: readonly string[] = [ + "Still on it.", + "Still thinking this through.", + "Almost there, still working.", + "This one needs a bit more thought.", + "Bear with me, still working on it.", +]; +// Reasoning models (e.g. qwen "thinking" variants) inline their chain of +// thought as / blocks in the same delta stream as the +// answer. That content must never be spoken or shown as answer text. +const THINK_OPEN_RE = //; +const THINK_CLOSE_RE = /<\/think(?:ing)?>/; +const THINK_TAG_MAX = "".length; +export function stripThinkingStream( + turn: Pick, + text: string +): string { + let pending = turn.thinkCarry + text; + turn.thinkCarry = ""; + let visible = ""; + while (pending !== "") { + const match = (turn.thinkDepth > 0 ? THINK_CLOSE_RE : THINK_OPEN_RE).exec(pending); + const opposite = (turn.thinkDepth > 0 ? THINK_OPEN_RE : THINK_CLOSE_RE).exec(pending); + // Nested opens inside reasoning are treated as plain text by Pi's own + // renderers; track only the matching tag for the current depth, but a + // stray close while at depth 0 is dropped as noise rather than shown. + const hit = match ?? (turn.thinkDepth === 0 ? opposite : null); + if (hit === null) { + // No complete tag: emit/discard all but a possible partial tag suffix. + const keep = partialTagSuffix(pending); + const emit = pending.slice(0, pending.length - keep.length); + if (turn.thinkDepth === 0) visible += emit; + turn.thinkCarry = keep; + break; + } + const before = pending.slice(0, hit.index); + if (turn.thinkDepth === 0 && match !== null) visible += before; + if (match !== null) turn.thinkDepth = turn.thinkDepth > 0 ? 0 : 1; + pending = pending.slice(hit.index + hit[0].length); + } + return visible; +} +function partialTagSuffix(text: string): string { + const window = text.slice(-THINK_TAG_MAX); + const start = window.lastIndexOf("<"); + if (start === -1) return ""; + const candidate = window.slice(start); + return "".startsWith(candidate) || "".startsWith(candidate) || + "".startsWith(candidate) || "".startsWith(candidate) + ? candidate + : ""; +} +export function stripThinkingAll(text: string): string { + return text + .replace(/[\s\S]*?(?:<\/think(?:ing)?>|$)/g, "") + .replace(/^\s+/, ""); +} + +let lastStillWorkingPhrase: string | undefined; +export function pickStillWorkingPhrase(): string { + const pool = STILL_WORKING_PHRASES.filter( + (phrase) => phrase !== lastStillWorkingPhrase + ); + const source = pool.length > 0 ? pool : STILL_WORKING_PHRASES; + const phrase = + source[Math.floor(Math.random() * source.length)] ?? source[0] ?? "Still on it."; + lastStillWorkingPhrase = phrase; + return phrase; +} +// Splits streamed answer text into complete sentences: one or more +// sentence-final marks (. ! ?), optionally followed by closing quotes or +// brackets, followed by required whitespace. Requiring the trailing +// whitespace is what keeps a mid-number decimal point like "3.5" (period +// immediately followed by a digit, no space) from being treated as a +// sentence boundary, with no special-cased abbreviation list needed. +const SENTENCE_BOUNDARY_RE = /[.!?]+[)\]"'”’]*\s+/; const DEVICE_WORKSPACE_ROOT = "/home/hope/workspaces"; const VOICE_CONVERSATION_INSTRUCTIONS = "This is a voice conversation. Respond in natural, concise spoken language suitable for Azure DragonHD text-to-speech. Do not use Markdown, headings, tables, bullet lists, code fences, URLs unless explicitly requested, or visual-only references. Speak punctuation naturally and expand ambiguous symbols when useful."; interface TokenRecord { @@ -100,7 +215,68 @@ interface SocketTurn { eventSocket: WebSocket | undefined; workingTimer: ReturnType | undefined; progressTimer: ReturnType | undefined; + /** True once any spoken progress clip (fallback or intermediate message) has started for the active turn. */ progressAnnounced: boolean | undefined; + /** + * True once Pi's agent.settled event has been observed for the active + * turn. The final answer supersedes any further spoken progress, so + * queued/in-flight progress clips check this and stop announcing. + */ + settled: boolean; + /** + * Serializes spoken progress clip synthesis and sending so clips are + * never interleaved or reordered on the shared audio output. The final + * answer's synthesis also waits on this chain before it starts, so it + * never overlaps a still-playing progress clip. + */ + progressChain: Promise; + /** Count of spoken progress clips sent so far this turn; see MAX_SPOKEN_PROGRESS_CLIPS. */ + progressClipsSent: number; + /** Combined spoken progress audio sent so far this turn, in ms; see MAX_SPOKEN_PROGRESS_AUDIO_MS. */ + progressAudioMs: number; + /** + * Delta-accumulated text of the assistant message currently streaming + * (i.e. since the last message.end, or since the turn started). Reset at + * every message boundary; see streamSpokenLength/streamToolCall. + */ + streamText: string; + /** + * Streaming / filter state for the current message: + * nesting depth (text inside any depth > 0 is reasoning, never shown or + * spoken) and a carried fragment that might be the start of a tag split + * across delta boundaries. Reset per turn and at every message.end. + */ + thinkDepth: number; + thinkCarry: string; + /** + * How many leading characters of the current message's text (streamText, + * or the message.end/settle-time authoritative text once the message has + * ended) have already been queued as spoken answer.sentence clips. The + * remainder is what still needs to be spoken, whether live, at + * message.end, or as the post-settle unspoken tail. + */ + streamSpokenLength: number; + /** + * True once a toolCall part has been observed for the message currently + * streaming (via a tool.start event, or defensively at its message.end). + * While true, further delta text for that message is not treated as + * answer sentences — it was narration ahead of a tool call, not the + * final answer. + */ + streamToolCall: boolean; + /** + * True once the turn's final message (the one with no toolCall) has had + * its trailing remainder flushed — either at its own message.end, or, if + * agent.settled arrived without an explicit message.end, defensively in + * runTurn. Prevents flushing the same trailing text twice. + */ + finalFlushDone: boolean; + /** True once any real (non-noise) answer.sentence clip has been queued this turn. */ + answerStreamingStarted: boolean; + /** Count of answer.sentence clips queued so far this turn; see MAX_SPOKEN_ANSWER_SENTENCES. */ + answerSentencesSent: number; + /** Combined spoken answer audio sent so far this turn, in ms; see MAX_SPOKEN_ANSWER_AUDIO_MS. */ + answerAudioMs: number; /** VAD/STT result only; never use it as an assistant response. */ transcript: string; /** Assistant output collected from Pi's session event stream only. */ @@ -420,6 +596,19 @@ function wireVoiceSocket( workingTimer: undefined, progressTimer: undefined, progressAnnounced: false, + settled: false, + progressChain: Promise.resolve(), + progressClipsSent: 0, + progressAudioMs: 0, + streamText: "", + thinkDepth: 0, + thinkCarry: "", + streamSpokenLength: 0, + streamToolCall: false, + finalFlushDone: false, + answerStreamingStarted: false, + answerSentencesSent: 0, + answerAudioMs: 0, }; sendJson(socket, { type: "hello", @@ -587,13 +776,36 @@ async function submitText( if (turn.closed) return; conversation.status = "working"; turn.assistant = ""; + turn.settled = false; + turn.progressAnnounced = false; + turn.progressChain = Promise.resolve(); + turn.progressClipsSent = 0; + turn.progressAudioMs = 0; + turn.streamText = ""; + turn.streamSpokenLength = 0; + turn.streamToolCall = false; + turn.thinkDepth = 0; + turn.thinkCarry = ""; + turn.finalFlushDone = false; + turn.answerStreamingStarted = false; + turn.answerSentencesSent = 0; + turn.answerAudioMs = 0; sendJson(socket, { type: "agent.working" }); - startWorkingProgress(socket, turn, () => { - sendJson(socket, { type: "agent.progress", text: "Hang on while I work on that.", audio: true }); - void speech.synthesize("Hang on while I work on that.", (pcm) => { - if (!turn.closed) sendAudio(socket, turn, pcm); - }); - }); + startWorkingProgress( + socket, + turn, + () => { + const phrase = pickFallbackPhrase(); + void speakProgress(socket, turn, speech, phrase, phrase); + }, + () => { + // Quiet down as soon as the real answer (or any narration mid-flow) + // has audibly begun; only fill genuinely dead air. + if (turn.settled || turn.answerStreamingStarted) return; + const phrase = pickStillWorkingPhrase(); + void speakProgress(socket, turn, speech, phrase, phrase); + } + ); try { await withTimeout( runTurn(socket, turn, conversation, deps, speech, text), @@ -630,7 +842,8 @@ async function runTurn( deps.daemon, conversation.sessionId, turn, - socket + socket, + speech ); try { await daemonJson( @@ -650,10 +863,32 @@ async function runTurn( } sendJson(socket, { type: "agent.accepted" }); await settled; + // Most runs get a message.end for the final (no-toolCall) message before + // agent.settled, and its handler in startSettledWaiter already flushed + // the trailing remainder live. But agent.settled can arrive without one + // (e.g. a very fast turn); this is the backstop for that case, using + // turn.assistant (the settled final text) as the authoritative source. + if (!turn.finalFlushDone) { + commitAnswerSentences( + turn, + socket, + speech, + extractCompleteSentences(turn.assistant.slice(turn.streamSpokenLength), true) + ); + } + // The final answer must never overlap or reorder against a spoken + // progress/answer clip that was still synthesizing/sending when the turn + // settled; draining the shared chain first guarantees strict ordering + // on the single audio output stream. + await turn.progressChain; const finalText = turn.assistant.trim(); sendJson(socket, { type: "assistant.final", text: finalText }); - if (finalText !== "") { - await speech.synthesize(finalText, (pcm) => { + // Anything not already streamed live (e.g. the sentence/audio budget was + // hit) is synthesized once here, as ordinary post-settle audio. When + // everything streamed live this is empty and no audio frame is sent. + const unspokenTail = turn.assistant.slice(turn.streamSpokenLength).trim(); + if (unspokenTail !== "") { + await speech.synthesize(prepareForSpeech(unspokenTail), (pcm) => { if (!turn.closed) sendAudio(socket, turn, pcm); }); } @@ -664,7 +899,8 @@ async function startSettledWaiter( daemon: SessionProxyDaemon, sessionId: string, turn: SocketTurn, - socket: WebSocket + socket: WebSocket, + speech: VoiceSpeechGateway ): Promise<{ settled: Promise }> { const events = daemon.connectWebSocket( `/sessions/${encodeURIComponent(sessionId)}/events` @@ -691,16 +927,82 @@ async function startSettledWaiter( event["type"] === "assistant.delta" && typeof event["text"] === "string" ) { - turn.assistant += event["text"]; - sendJson(socket, { type: "assistant.delta", text: event["text"] }); + const visible = stripThinkingStream(turn, event["text"]); + if (visible === "") return; + turn.assistant += visible; + sendJson(socket, { type: "assistant.delta", text: visible }); + // Which message will turn out to be "the final answer" isn't + // knowable mid-stream, so every message is optimistically + // streamed as answer sentences until either a toolCall part + // appears in it (see tool.start below) or it ends. + if (!turn.settled && !turn.streamToolCall) { + turn.streamText += visible; + processAnswerDelta(turn, socket, speech); + } } - if (event["type"] === "message.end") { - const text = assistantText(event["message"]); + // Pi emits tool.start once a message's tool call begins, which (for + // a message with narration before it) arrives after that + // narration's text deltas but before the message's message.end. + // It is the earliest signal that this message will not be the + // turn's final answer, so live sentence-streaming for it stops + // here rather than waiting for message.end to find out. + if (event["type"] === "tool.start" && !turn.settled) { + turn.streamToolCall = true; + } + // A stray message.end after agent.settled (the daemon should not + // send one, but nothing prevents it) must not overwrite the final + // answer already captured, nor be queued for spoken progress. + if (event["type"] === "message.end" && !turn.settled) { + const text = stripThinkingAll(assistantText(event["message"])); + turn.thinkDepth = 0; + turn.thinkCarry = ""; if (text !== "") turn.assistant = text; + // A message that includes a tool call is not the turn's final + // answer: Pi's agent loop only settles once an assistant message + // has no pending tool call, so a toolCall part here means more + // work (and likely another assistant message) is still coming. + if (messageHasToolCall(event["message"])) { + // Only the part of this message that was never spoken live (see + // assistant.delta above) still needs narrating — track the + // offset so a message spoken in full live is never re-spoken, + // and skip the clip entirely when nothing is left unspoken. + const spokenThisMessage = turn.streamSpokenLength > 0; + const unspoken = text.slice(turn.streamSpokenLength).trim(); + // Once the turn's answer has audibly begun, a brand-new message + // that had no live narration of its own (spokenThisMessage + // false) is left quiet rather than talking over it; a message + // that *was* mid-narration when the tool call appeared still + // gets its own leftover finished, so it isn't cut off abruptly. + const suppressed = turn.answerStreamingStarted && !spokenThisMessage; + if (unspoken !== "" && !suppressed) { + void speakProgress(socket, turn, speech, text, truncateForSpeech(unspoken, PROGRESS_SPEECH_MAX_CHARS)); + } + turn.streamText = ""; + turn.streamSpokenLength = 0; + turn.streamToolCall = false; + } else { + // No toolCall part: Pi's agent loop only settles once a + // message like this is reached, so this is definitively the + // turn's final answer. Flush its trailing remainder (held back + // live because it lacked trailing punctuation+whitespace) as a + // last answer.sentence; leave streamSpokenLength as-is (no + // "next message" is coming) so runTurn's post-settle tail + // computation and the finalFlushDone guard below both see it. + if (text !== "") { + commitAnswerSentences(turn, socket, speech, extractCompleteSentences(text.slice(turn.streamSpokenLength), true)); + } + turn.finalFlushDone = true; + } } // This is Pi's native session-level event, forwarded by // PiSessionService/sessiond without deriving it from agent_end. - if (event["type"] === "agent.settled") finish(); + if (event["type"] === "agent.settled") { + // Mark settled before finish() so any progress clip already + // queued on turn.progressChain (but not yet started) skips + // itself instead of speaking after the turn has concluded. + turn.settled = true; + finish(); + } } catch (error) { finish(error instanceof Error ? error : new Error(String(error))); } @@ -732,12 +1034,20 @@ function waitForWebSocketOpen(socket: WebSocket): Promise { export function startWorkingProgress( socket: Pick, turn: Pick & Partial>, - onSpokenProgress?: () => void + onSpokenProgress?: () => void, + onPeriodicSpokenUpdate?: () => void ): void { clearWorkingProgress(turn); + let ticks = 0; const timer = setInterval(() => { - if (!turn.closed && turn.workingTimer === timer && socket.readyState === 1) + if (!turn.closed && turn.workingTimer === timer && socket.readyState === 1) { socket.send(JSON.stringify({ type: "agent.working" })); + // A long-thinking model can produce nothing speakable for minutes; a + // spoken "still on it" every third heartbeat (~45s) keeps the room from + // going dead between the 2s opener and the first real answer sentence. + ticks += 1; + if (onPeriodicSpokenUpdate !== undefined && ticks % 3 === 0) onPeriodicSpokenUpdate(); + } }, WORKING_PROGRESS_INTERVAL_MS); timer.unref(); turn.workingTimer = timer; @@ -940,6 +1250,231 @@ function rawDataToBuffer(data: RawData): Buffer { if (data instanceof ArrayBuffer) return Buffer.from(data); return Buffer.concat(data); } +/** + * Queue one spoken progress clip (the "Hang on..." fallback or an + * intermediate assistant message) behind any clip already in flight. Callers + * do not need to await the result; runTurn awaits turn.progressChain once + * before the final answer so every queued clip finishes, in order, first. + */ +function speakProgress( + socket: WebSocket, + turn: SocketTurn, + speech: VoiceSpeechGateway, + fullText: string, + speechText: string +): Promise { + turn.progressChain = turn.progressChain.then(() => + sendProgressClip(socket, turn, speech, fullText, speechText) + ); + return turn.progressChain; +} +async function sendProgressClip( + socket: WebSocket, + turn: SocketTurn, + speech: VoiceSpeechGateway, + fullText: string, + speechText: string +): Promise { + // Whether this particular message should be spoken at all (including the + // "already settled" rule) is decided once, synchronously, by the caller at + // the moment its triggering event arrived — see the message.end handler in + // startSettledWaiter. Re-checking turn.settled here would be wrong: this + // function runs on the progressChain microtask queue, so a same-tick burst + // of message.end + agent.settled events (a single WebSocket read can + // deliver several frames back to back) could otherwise flip turn.settled + // to true before an already-approved clip gets to run, silently dropping + // it. A closed turn still has nowhere to send audio, so that check stays. + if (turn.closed) return; + if (turn.progressClipsSent >= MAX_SPOKEN_PROGRESS_CLIPS) return; + if (turn.progressAudioMs >= MAX_SPOKEN_PROGRESS_AUDIO_MS) return; + turn.progressClipsSent += 1; + turn.progressAnnounced = true; + sendJson(socket, { type: "agent.progress", text: fullText, audio: true }); + try { + await speech.synthesize(speechText, (pcm) => { + if (turn.closed) return; + turn.progressAudioMs += pcm.length / PROGRESS_AUDIO_BYTES_PER_MS; + sendAudio(socket, turn, pcm); + }); + } catch { + // Best-effort narration: a failed progress clip must not abort the + // turn, nor block progress clips already queued behind it. + } +} +let lastFallbackPhrase: string | undefined; +/** + * Pick the 2-second "still working" fallback phrase for a turn, pseudo- + * randomly, but never the same phrase spoken by the immediately preceding + * turn (tracked at module scope, since a fresh SocketTurn is created per + * turn and would otherwise have no memory of it). + */ +export function pickFallbackPhrase(): string { + const candidates = FALLBACK_PROGRESS_PHRASES.filter( + (phrase) => phrase !== lastFallbackPhrase + ); + const pool = candidates.length > 0 ? candidates : FALLBACK_PROGRESS_PHRASES; + const phrase = + pool[Math.floor(Math.random() * pool.length)] ?? pool[0] ?? "One sec."; + lastFallbackPhrase = phrase; + return phrase; +} +export interface ExtractedSentence { + text: string; + /** Length, in characters of the original (untrimmed) pending text, consumed by this sentence. */ + rawLength: number; +} +/** + * Split the not-yet-spoken tail of a streaming answer into complete + * sentences, holding back any incomplete trailing fragment (text with no + * sentence-ending punctuation+whitespace yet) for the next call. Pass + * `flushRemainder: true` at the point the source message is known to have + * ended (message.end, or turn settle with no message.end) to also emit + * that trailing fragment as one final "sentence". + */ +export function extractCompleteSentences( + pending: string, + flushRemainder: boolean +): ExtractedSentence[] { + const sentences: ExtractedSentence[] = []; + let rest = pending; + for (;;) { + const match = SENTENCE_BOUNDARY_RE.exec(rest); + if (match === null) break; + const end = match.index + match[0].length; + sentences.push({ text: rest.slice(0, end).trim(), rawLength: end }); + rest = rest.slice(end); + } + if (flushRemainder && rest.trim() !== "") { + sentences.push({ text: rest.trim(), rawLength: rest.length }); + } + return sentences; +} +/** + * False for a fragment that, once simple markdown decoration is stripped + * (heading/bullet/emphasis/code/table/link punctuation), has no actual + * word content left — e.g. "---", "```", "**", "> ". These are a normal + * byproduct of segmenting a markdown-ish stream into "sentences" and + * should be silently skipped rather than spoken. + */ +export function isSpeakableFragment(text: string): boolean { + const trimmed = text.trim(); + if (trimmed === "") return false; + const withoutMarkdownSyntax = trimmed + .replace(/[#*_`~=\-•>|[\]()]+/g, "") + .trim(); + return withoutMarkdownSyntax !== ""; +} +/** + * Text preparation immediately before handing a spoken answer fragment to + * speech synthesis. Matches how the final answer's text is prepared for + * TTS today (a trim; the server already instructs Pi not to use Markdown + * in voice turns, see VOICE_CONVERSATION_INSTRUCTIONS) so streamed + * sentences and the final/tail answer audio are prepared identically. + */ +function prepareForSpeech(text: string): string { + return text.trim(); +} +/** Segment newly streamed answer text and speak any complete sentences found. */ +function processAnswerDelta( + turn: SocketTurn, + socket: WebSocket, + speech: VoiceSpeechGateway +): void { + if (turn.streamToolCall) return; + commitAnswerSentences( + turn, + socket, + speech, + extractCompleteSentences(turn.streamText.slice(turn.streamSpokenLength), false) + ); +} +/** + * Advance turn.streamSpokenLength over each sentence and queue it to be + * spoken, in order, unless the per-turn answer sentence/audio budget is + * already exhausted — in which case this (and every later) sentence is + * left unconsumed so it becomes part of the post-settle unspoken tail. + * Noise-only fragments (see isSpeakableFragment) are consumed (skipped + * silently) without being queued, since they carry nothing to say. + */ +function commitAnswerSentences( + turn: SocketTurn, + socket: WebSocket, + speech: VoiceSpeechGateway, + sentences: ExtractedSentence[] +): void { + for (const sentence of sentences) { + if (turn.answerSentencesSent >= MAX_SPOKEN_ANSWER_SENTENCES) break; + if (turn.answerAudioMs >= MAX_SPOKEN_ANSWER_AUDIO_MS) break; + turn.streamSpokenLength += sentence.rawLength; + if (!isSpeakableFragment(sentence.text)) continue; + turn.answerSentencesSent += 1; + turn.answerStreamingStarted = true; + turn.progressAnnounced = true; + void speakAnswerSentence(socket, turn, speech, sentence.text); + } +} +/** + * Queue one spoken answer sentence behind any progress/answer clip already + * in flight, on the same FIFO chain as speakProgress, so answer sentences + * and progress clips are never interleaved or reordered on the shared + * audio output. + */ +function speakAnswerSentence( + socket: WebSocket, + turn: SocketTurn, + speech: VoiceSpeechGateway, + text: string +): Promise { + turn.progressChain = turn.progressChain.then(() => + sendAnswerSentenceClip(socket, turn, speech, text) + ); + return turn.progressChain; +} +async function sendAnswerSentenceClip( + socket: WebSocket, + turn: SocketTurn, + speech: VoiceSpeechGateway, + text: string +): Promise { + if (turn.closed) return; + sendJson(socket, { type: "answer.sentence", text }); + try { + await speech.synthesize(prepareForSpeech(text), (pcm) => { + if (turn.closed) return; + turn.answerAudioMs += pcm.length / PROGRESS_AUDIO_BYTES_PER_MS; + sendAudio(socket, turn, pcm); + }); + } catch { + // Best-effort narration: a failed sentence clip must not abort the + // turn, nor block sentence clips already queued behind it. + } +} +/** True if an assistant message includes a tool call content part. */ +function messageHasToolCall(value: unknown): boolean { + if (!isRecord(value) || !Array.isArray(value["content"])) return false; + return value["content"].some( + (part) => isRecord(part) && part["type"] === "toolCall" + ); +} +/** + * Truncate long intermediate text for speech, preferring a sentence + * boundary within the limit and otherwise a word boundary. The full text is + * always sent separately in the JSON agent.progress frame; only the audio is + * shortened. + */ +function truncateForSpeech(text: string, maxChars: number): string { + const trimmed = text.trim(); + if (trimmed.length <= maxChars) return trimmed; + const window = trimmed.slice(0, maxChars); + let sentenceEnd = -1; + for (const punctuation of [". ", "! ", "? "]) { + sentenceEnd = Math.max(sentenceEnd, window.lastIndexOf(punctuation)); + } + if (/[.!?]$/.test(window)) sentenceEnd = Math.max(sentenceEnd, window.length - 1); + if (sentenceEnd > maxChars * 0.4) return window.slice(0, sentenceEnd + 1).trim(); + const spaceIndex = window.lastIndexOf(" "); + return (spaceIndex > 0 ? window.slice(0, spaceIndex) : window).trim(); +} function assistantText(value: unknown): string { if ( !isRecord(value) ||