From fbcd99bcd603b2135238095e3394c1010c723a79 Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Thu, 27 Aug 2026 00:40:19 +0200 Subject: [PATCH 01/10] Speak finalized Brunch responses with OpenAI Select only finalized canonical assistant text and validated brunch_ask questions, then forward the exact fingerprinted text through the app-owned OpenAI Speech edge. Queue cancelable MP3 playback behind Brunch, keep the microphone closed until speech drains, reject stale generations, disclose the AI-generated voice, and preserve visible text on failure. Amp-Thread-ID: https://ampcode.com/threads/T-01a03fb3-fd3d-737f-b4c6-1fc9282950bf Co-authored-by: Amp --- apps/petrinaut-website/README.md | 30 +- apps/petrinaut-website/api/voice/speech.ts | 12 + .../src/canonical-speech-fingerprint.ts | 9 + .../voice-interview/canonical-speech.test.ts | 178 +++++++++++ .../app/voice-interview/canonical-speech.ts | 95 ++++++ .../speech-playback-controller.test.ts | 197 +++++++++++++ .../speech-playback-controller.ts | 172 +++++++++++ .../voice-interview-control.test.tsx | 34 +++ .../voice-interview-control.tsx | 29 +- .../voice-turn-controller.test.ts | 278 +++++++++++++++++- .../voice-interview/voice-turn-controller.ts | 246 ++++++++++++---- .../src/server/voice/openai-speech.test.ts | 269 +++++++++++++++++ .../src/server/voice/openai-speech.ts | 243 +++++++++++++++ apps/petrinaut-website/vercel.json | 3 + apps/petrinaut-website/vite.config.ts | 1 + .../adr/0009-openai-voice-ui-turn-shell.md | 20 ++ .../@hashintel/petrinaut/docs/ai-assistant.md | 7 + 17 files changed, 1745 insertions(+), 78 deletions(-) create mode 100644 apps/petrinaut-website/api/voice/speech.ts create mode 100644 apps/petrinaut-website/src/canonical-speech-fingerprint.ts create mode 100644 apps/petrinaut-website/src/main/app/voice-interview/canonical-speech.test.ts create mode 100644 apps/petrinaut-website/src/main/app/voice-interview/canonical-speech.ts create mode 100644 apps/petrinaut-website/src/main/app/voice-interview/speech-playback-controller.test.ts create mode 100644 apps/petrinaut-website/src/main/app/voice-interview/speech-playback-controller.ts create mode 100644 apps/petrinaut-website/src/server/voice/openai-speech.test.ts create mode 100644 apps/petrinaut-website/src/server/voice/openai-speech.ts diff --git a/apps/petrinaut-website/README.md b/apps/petrinaut-website/README.md index 583ccbe749c..b66405ed9cf 100644 --- a/apps/petrinaut-website/README.md +++ b/apps/petrinaut-website/README.md @@ -61,20 +61,20 @@ provides a fake optimizer for isolated UI development. ## Environment variables -| Name | Required | Used by | Notes | -| -------------------------------- | ---------------- | ---------------- | ---------------------------------------------------------- | -| `OPENAI_API_KEY` | for chat to work | `api/chat.ts` | OpenAI key the function uses to call `streamText`. | -| `OPENAI_VOICE_API_KEY` | for voice input | voice API | Dedicated OpenAI key used only by the Realtime call proxy. | -| `PETRINAUT_OPENAI_VOICE_ENABLED` | no | voice API | Set to `true` to enable voice outside production. | -| `PETRINAUT_AI_MODEL` | no | `api/chat.ts` | Overrides the default OpenAI model id. | -| `PETRINAUT_OPT_ORIGIN` | no | `vite.config.ts` | Overrides the local optimizer proxy target. | -| `VITE_BRUNCH_CHAT_ENDPOINT` | for voice input | website | Full Brunch Petrinaut chat endpoint used by the panel. | -| `VITE_PETRINAUT_OPT_PROVIDER` | no | website | Set to `service` to enable the optimization route. | -| `SENTRY_DSN` | no | `vite.config.ts` | Wired into the bundle via `__SENTRY_DSN__` at build time. | +| Name | Required | Used by | Notes | +| -------------------------------- | ---------------- | ---------------- | --------------------------------------------------------- | +| `OPENAI_API_KEY` | for chat to work | `api/chat.ts` | OpenAI key the function uses to call `streamText`. | +| `OPENAI_VOICE_API_KEY` | for voice | voice API | Dedicated OpenAI key used by Realtime and Speech proxies. | +| `PETRINAUT_OPENAI_VOICE_ENABLED` | no | voice API | Set to `true` to enable voice outside production. | +| `PETRINAUT_AI_MODEL` | no | `api/chat.ts` | Overrides the default OpenAI model id. | +| `PETRINAUT_OPT_ORIGIN` | no | `vite.config.ts` | Overrides the local optimizer proxy target. | +| `VITE_BRUNCH_CHAT_ENDPOINT` | for voice input | website | Full Brunch Petrinaut chat endpoint used by the panel. | +| `VITE_PETRINAUT_OPT_PROVIDER` | no | website | Set to `service` to enable the optimization route. | +| `SENTRY_DSN` | no | `vite.config.ts` | Wired into the bundle via `__SENTRY_DSN__` at build time. | Local values live in `.env.local`; Vite's `loadEnv` (see [`vite.config.ts`](vite.config.ts)) copies them into `process.env` for both the dev server and the API functions. In production, set these in the Vercel project settings. -### Brunch voice-input preview +### Brunch voice preview Voice input is disabled by default and always unavailable when `VERCEL_ENV` is `production`. To exercise the preview locally or in a Vercel preview, set a @@ -90,6 +90,14 @@ SDK transport. Partial transcripts remain display-only. The preview derives a stable conversation id from the locally saved net; it is diagnostic identity, not production authentication or conversation authority. +While voice is active, finalized assistant text and validated structured Brunch +questions are spoken with OpenAI's dedicated Speech API. The server fixes the +model and voice and forwards the selected canonical text without rewriting it; +Realtime remains transcription-only. The microphone stays closed while Brunch +is working and while AI-generated speech is being synthesized or played. The UI +discloses that the voice is AI-generated. Ending voice cancels playback, and a +speech failure leaves the exact response visible for reading. + The Brunch deployment must allow the website origin through its `BRUNCH_PETRINAUT_ORIGINS` setting. Starting voice input requests browser microphone permission. Denying permission leaves the existing text composer diff --git a/apps/petrinaut-website/api/voice/speech.ts b/apps/petrinaut-website/api/voice/speech.ts new file mode 100644 index 00000000000..a02b0f1b8b2 --- /dev/null +++ b/apps/petrinaut-website/api/voice/speech.ts @@ -0,0 +1,12 @@ +import { createOpenAISpeechHandler } from "../../src/server/voice/openai-speech"; + +declare const process: { + env: Record; +}; + +export default { + fetch: createOpenAISpeechHandler({ + environment: process.env, + fetch: globalThis.fetch.bind(globalThis), + }), +}; diff --git a/apps/petrinaut-website/src/canonical-speech-fingerprint.ts b/apps/petrinaut-website/src/canonical-speech-fingerprint.ts new file mode 100644 index 00000000000..323ca9d40d6 --- /dev/null +++ b/apps/petrinaut-website/src/canonical-speech-fingerprint.ts @@ -0,0 +1,9 @@ +export const hashCanonicalSpeechText = (text: string): string => { + let hash = 0x81_1c_9d_c5; + for (const byte of new TextEncoder().encode(text)) { + // eslint-disable-next-line no-bitwise -- FNV-1a requires byte-wise XOR. + hash = Math.imul(hash ^ byte, 0x01_00_01_93); + } + // eslint-disable-next-line no-bitwise -- Convert the signed result to uint32. + return `fnv1a32:${(hash >>> 0).toString(16).padStart(8, "0")}`; +}; diff --git a/apps/petrinaut-website/src/main/app/voice-interview/canonical-speech.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/canonical-speech.test.ts new file mode 100644 index 00000000000..9e6b1c3d80f --- /dev/null +++ b/apps/petrinaut-website/src/main/app/voice-interview/canonical-speech.test.ts @@ -0,0 +1,178 @@ +import { describe, expect, test } from "vitest"; + +import { ASK_TOOL_NAME } from "@hashintel/brunch-agent-transport-aisdk/client-tools"; + +import { + hashCanonicalSpeechText, + selectCanonicalSpeechSegments, +} from "./canonical-speech"; + +import type { PetrinautAiMessage } from "@hashintel/petrinaut/ui"; + +const select = (messages: PetrinautAiMessage[]) => + selectCanonicalSpeechSegments(messages); + +describe("canonical speech selection", () => { + test("selects only finalized assistant text without changing it", () => { + const messages = [ + { + id: "user-1", + role: "user", + parts: [{ type: "text", text: "Do not speak the user." }], + }, + { + id: "assistant-1", + role: "assistant", + parts: [ + { + type: "reasoning", + text: "Do not speak reasoning.", + state: "done", + }, + { + type: "text", + text: "Do not speak partial text.", + state: "streaming", + }, + { + type: "text", + text: " ", + state: "done", + }, + { + type: "text", + text: " Keep this exact finalized response. ", + state: "done", + }, + { + type: "text", + text: "Loaded finalized response.", + }, + { + type: "dynamic-tool", + toolCallId: "diagnostic-1", + toolName: "diagnostic", + state: "output-available", + input: {}, + output: { text: "Do not speak tool output." }, + }, + ], + }, + { + id: "system-1", + role: "system", + parts: [{ type: "text", text: "Do not speak system text." }], + }, + ] satisfies PetrinautAiMessage[]; + + const selected = select(messages); + const firstHash = hashCanonicalSpeechText( + " Keep this exact finalized response. ", + ); + const secondHash = hashCanonicalSpeechText("Loaded finalized response."); + expect(selected).toEqual([ + { + contentHash: firstHash, + id: `canonical-speech:assistant-1:text%3A3:${firstHash}`, + messageId: "assistant-1", + partId: "text:3", + source: "assistant-text", + text: " Keep this exact finalized response. ", + }, + { + contentHash: secondHash, + id: `canonical-speech:assistant-1:text%3A4:${secondHash}`, + messageId: "assistant-1", + partId: "text:4", + source: "assistant-text", + text: "Loaded finalized response.", + }, + ]); + }); + + test("selects one exact validated brunch_ask question", () => { + const messages = [ + { + id: "assistant-ask", + role: "assistant", + parts: [ + { + type: "dynamic-tool", + toolCallId: "ask-1", + toolName: ASK_TOOL_NAME, + state: "input-available", + input: { question: "Which operator confirms the batch?" }, + }, + { + type: "dynamic-tool", + toolCallId: "ask-malformed", + toolName: ASK_TOOL_NAME, + state: "input-available", + input: { question: 42 }, + }, + { + type: "dynamic-tool", + toolCallId: "ask-submitted", + toolName: ASK_TOOL_NAME, + state: "output-available", + input: { question: "Do not repeat an answered question." }, + output: { answer: "Already answered." }, + }, + { + type: "dynamic-tool", + toolCallId: "other-tool", + toolName: "other_tool", + state: "input-available", + input: { question: "Do not speak another tool." }, + }, + ], + }, + ] satisfies PetrinautAiMessage[]; + + const selected = select(messages); + const contentHash = hashCanonicalSpeechText( + "Which operator confirms the batch?", + ); + expect(selected).toEqual([ + { + contentHash, + id: `canonical-speech:assistant-ask:ask-1:${contentHash}`, + messageId: "assistant-ask", + partId: "ask-1", + source: "brunch-ask", + text: "Which operator confirms the batch?", + }, + ]); + }); + + test("uses stable source identity plus an exact-text fingerprint", () => { + expect(hashCanonicalSpeechText("hello")).toBe("fnv1a32:4f9f2cab"); + + const first = select([ + { + id: "assistant/id", + role: "assistant", + parts: [{ type: "text", text: "Exact text", state: "done" }], + }, + ]); + const repeated = select([ + { + id: "assistant/id", + role: "assistant", + parts: [{ type: "text", text: "Exact text", state: "done" }], + }, + ]); + const changed = select([ + { + id: "assistant/id", + role: "assistant", + parts: [{ type: "text", text: "Exact text ", state: "done" }], + }, + ]); + + expect(repeated).toEqual(first); + expect(changed[0]?.partId).toBe(first[0]?.partId); + expect(changed[0]?.contentHash).not.toBe(first[0]?.contentHash); + expect(changed[0]?.id).not.toBe(first[0]?.id); + }); +}); diff --git a/apps/petrinaut-website/src/main/app/voice-interview/canonical-speech.ts b/apps/petrinaut-website/src/main/app/voice-interview/canonical-speech.ts new file mode 100644 index 00000000000..151db772ab5 --- /dev/null +++ b/apps/petrinaut-website/src/main/app/voice-interview/canonical-speech.ts @@ -0,0 +1,95 @@ +import { + ASK_TOOL_NAME, + parseBrunchAskInput, +} from "@hashintel/brunch-agent-transport-aisdk/client-tools"; + +import { hashCanonicalSpeechText } from "../../../canonical-speech-fingerprint"; + +import type { PetrinautAiMessage } from "@hashintel/petrinaut/ui"; + +export { hashCanonicalSpeechText }; + +export interface CanonicalSpeechSegment { + readonly contentHash: string; + readonly id: string; + readonly messageId: string; + readonly partId: string; + readonly source: "assistant-text" | "brunch-ask"; + readonly text: string; +} + +const createSegment = ( + messageId: string, + partId: string, + source: CanonicalSpeechSegment["source"], + text: string, +): CanonicalSpeechSegment => { + const contentHash = hashCanonicalSpeechText(text); + return { + contentHash, + id: [ + "canonical-speech", + encodeURIComponent(messageId), + encodeURIComponent(partId), + contentHash, + ].join(":"), + messageId, + partId, + source, + text, + }; +}; + +export const selectCanonicalSpeechSegments = ( + messages: PetrinautAiMessage[], +): CanonicalSpeechSegment[] => { + const segments: CanonicalSpeechSegment[] = []; + + for (const message of messages) { + if (message.role !== "assistant") { + continue; + } + + for (const [partIndex, part] of message.parts.entries()) { + if ( + part.type === "text" && + part.state !== "streaming" && + part.text.trim() + ) { + segments.push( + createSegment( + message.id, + `text:${partIndex}`, + "assistant-text", + part.text, + ), + ); + continue; + } + + if ( + part.type !== "dynamic-tool" || + part.toolName !== ASK_TOOL_NAME || + part.state !== "input-available" + ) { + continue; + } + + try { + const input = parseBrunchAskInput(part.input); + segments.push( + createSegment( + message.id, + part.toolCallId, + "brunch-ask", + input.question, + ), + ); + } catch { + // Malformed tool inputs remain visible as tool errors; they are not spoken. + } + } + } + + return segments; +}; diff --git a/apps/petrinaut-website/src/main/app/voice-interview/speech-playback-controller.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/speech-playback-controller.test.ts new file mode 100644 index 00000000000..8349c64f699 --- /dev/null +++ b/apps/petrinaut-website/src/main/app/voice-interview/speech-playback-controller.test.ts @@ -0,0 +1,197 @@ +import { describe, expect, test, vi } from "vitest"; + +import { SpeechPlaybackController } from "./speech-playback-controller"; + +import type { CanonicalSpeechSegment } from "./canonical-speech"; + +const segment: CanonicalSpeechSegment = { + contentHash: "fnv1a32:69f1e741", + id: "canonical-speech:assistant-1:text%3A0:fnv1a32:69f1e741", + messageId: "assistant-1", + partId: "text:0", + source: "assistant-text", + text: " Preserve this exact canonical text. ", +}; + +const createAudioHarness = () => { + const listeners = new Map void>>(); + const audio = { + addEventListener: vi.fn((type: string, listener: () => void) => { + const typeListeners = listeners.get(type) ?? new Set(); + typeListeners.add(listener); + listeners.set(type, typeListeners); + }), + pause: vi.fn(), + play: vi.fn(async () => undefined), + removeEventListener: vi.fn((type: string, listener: () => void) => { + listeners.get(type)?.delete(listener); + }), + }; + return { + audio, + emit: (type: "ended" | "error") => { + for (const listener of listeners.get(type) ?? []) { + listener(); + } + }, + }; +}; + +const createHarness = ( + fetch: typeof globalThis.fetch = vi.fn(async () => + Promise.resolve( + new Response(new Uint8Array([1, 2, 3]), { + headers: { "content-type": "audio/mpeg" }, + }), + ), + ), +) => { + const audio = createAudioHarness(); + const createAudio = vi.fn(() => audio.audio); + const createObjectURL = vi.fn(() => "blob:canonical-speech"); + const revokeObjectURL = vi.fn(); + const controller = new SpeechPlaybackController({ + createAudio, + createObjectURL, + fetch, + revokeObjectURL, + }); + return { + audio, + controller, + createAudio, + createObjectURL, + fetch, + revokeObjectURL, + }; +}; + +describe("SpeechPlaybackController", () => { + test("posts the exact canonical text and resolves after audio playback ends", async () => { + const harness = createHarness(); + const onPlaying = vi.fn(); + + const playback = harness.controller.play(segment, { onPlaying }); + await vi.waitFor(() => expect(harness.createAudio).toHaveBeenCalledOnce()); + await vi.waitFor(() => expect(onPlaying).toHaveBeenCalledOnce()); + + expect(harness.fetch).toHaveBeenCalledOnce(); + const [url, request] = vi.mocked(harness.fetch).mock.calls[0]!; + expect(url).toBe("/api/voice/speech"); + expect(request).toMatchObject({ + body: JSON.stringify({ segmentId: segment.id, text: segment.text }), + cache: "no-store", + headers: { "content-type": "application/json" }, + method: "POST", + }); + expect(request?.signal).toBeInstanceOf(AbortSignal); + expect(harness.createObjectURL).toHaveBeenCalledWith(expect.any(Blob)); + expect(harness.createAudio).toHaveBeenCalledWith("blob:canonical-speech"); + expect(harness.audio.audio.play).toHaveBeenCalledOnce(); + + harness.audio.emit("ended"); + await expect(playback).resolves.toBeUndefined(); + expect(harness.revokeObjectURL).toHaveBeenCalledWith( + "blob:canonical-speech", + ); + }); + + test("rejects failed and non-audio speech responses without creating audio", async () => { + const fetch = vi + .fn() + .mockResolvedValueOnce(new Response("failed", { status: 502 })) + .mockResolvedValueOnce(Response.json({ not: "audio" })) + .mockResolvedValueOnce( + new Response(new Uint8Array([1]), { + headers: { "content-type": "audio/wav" }, + }), + ); + const harness = createHarness(fetch); + + await expect(harness.controller.play(segment)).rejects.toThrow( + "The response could not be spoken. Read the visible text instead.", + ); + await expect(harness.controller.play(segment)).rejects.toThrow( + "The response could not be spoken. Read the visible text instead.", + ); + await expect(harness.controller.play(segment)).rejects.toThrow( + "The response could not be spoken. Read the visible text instead.", + ); + expect(harness.createAudio).not.toHaveBeenCalled(); + }); + + test("rejects text that does not match its canonical fingerprint", async () => { + const harness = createHarness(); + + await expect( + harness.controller.play({ ...segment, text: "Tampered text" }), + ).rejects.toThrow( + "The response could not be spoken. Read the visible text instead.", + ); + + expect(harness.fetch).not.toHaveBeenCalled(); + }); + + test("aborts synthesis and ignores a response from a canceled generation", async () => { + let resolveFetch: ((response: Response) => void) | undefined; + const fetch = vi.fn( + () => + new Promise((resolve) => { + resolveFetch = resolve; + }), + ); + const harness = createHarness(fetch); + + const playback = harness.controller.play(segment); + await vi.waitFor(() => expect(fetch).toHaveBeenCalledOnce()); + harness.controller.cancel(); + + await expect(playback).rejects.toMatchObject({ name: "AbortError" }); + expect(fetch.mock.calls[0]?.[1]?.signal?.aborted).toBe(true); + resolveFetch?.( + new Response(new Uint8Array([1]), { + headers: { "content-type": "audio/mpeg" }, + }), + ); + await Promise.resolve(); + expect(harness.createAudio).not.toHaveBeenCalled(); + }); + + test("pauses active audio, revokes its URL, and rejects stale completion on cancel", async () => { + const harness = createHarness(); + const playback = harness.controller.play(segment); + await vi.waitFor(() => expect(harness.audio.audio.play).toHaveBeenCalled()); + + harness.controller.cancel(); + harness.audio.emit("ended"); + + await expect(playback).rejects.toMatchObject({ name: "AbortError" }); + expect(harness.audio.audio.pause).toHaveBeenCalledOnce(); + expect(harness.revokeObjectURL).toHaveBeenCalledWith( + "blob:canonical-speech", + ); + }); + + test("turns audio startup and playback errors into the visible-text fallback", async () => { + const harness = createHarness(); + harness.audio.audio.play.mockRejectedValueOnce( + new DOMException("blocked", "NotAllowedError"), + ); + + await expect(harness.controller.play(segment)).rejects.toThrow( + "The response could not be spoken. Read the visible text instead.", + ); + expect(harness.revokeObjectURL).toHaveBeenCalledOnce(); + + const secondHarness = createHarness(); + const playback = secondHarness.controller.play(segment); + await vi.waitFor(() => + expect(secondHarness.audio.audio.play).toHaveBeenCalledOnce(), + ); + secondHarness.audio.emit("error"); + await expect(playback).rejects.toThrow( + "The response could not be spoken. Read the visible text instead.", + ); + expect(secondHarness.revokeObjectURL).toHaveBeenCalledOnce(); + }); +}); diff --git a/apps/petrinaut-website/src/main/app/voice-interview/speech-playback-controller.ts b/apps/petrinaut-website/src/main/app/voice-interview/speech-playback-controller.ts new file mode 100644 index 00000000000..d6cbc3936f4 --- /dev/null +++ b/apps/petrinaut-website/src/main/app/voice-interview/speech-playback-controller.ts @@ -0,0 +1,172 @@ +import { + hashCanonicalSpeechText, + type CanonicalSpeechSegment, +} from "./canonical-speech"; + +const SPEECH_ERROR_MESSAGE = + "The response could not be spoken. Read the visible text instead."; + +interface SpeechAudio { + addEventListener(type: "ended" | "error", listener: () => void): void; + pause(): void; + play(): Promise; + removeEventListener(type: "ended" | "error", listener: () => void): void; +} + +interface SpeechPlaybackDependencies { + readonly createAudio: (source: string) => SpeechAudio; + readonly createObjectURL: (blob: Blob) => string; + readonly fetch: typeof globalThis.fetch; + readonly revokeObjectURL: (url: string) => void; +} + +interface SpeechPlaybackEvents { + readonly onPlaying?: () => void; +} + +interface ActiveAudio { + cancel(reason: DOMException): void; + readonly generation: number; +} + +const fallbackError = (): Error => new Error(SPEECH_ERROR_MESSAGE); +const abortError = (): DOMException => + new DOMException("Speech playback was canceled.", "AbortError"); + +const waitForAbort = ( + promise: Promise, + signal: AbortSignal, +): Promise => { + if (signal.aborted) { + return Promise.reject(signal.reason); + } + + return new Promise((resolve, reject) => { + const handleAbort = () => reject(signal.reason); + signal.addEventListener("abort", handleAbort, { once: true }); + void promise.then( + (value) => { + signal.removeEventListener("abort", handleAbort); + resolve(value); + }, + (error: unknown) => { + signal.removeEventListener("abort", handleAbort); + reject(error); + }, + ); + }); +}; + +const isAbortError = (error: unknown): error is DOMException => + error instanceof DOMException && error.name === "AbortError"; + +export class SpeechPlaybackController { + readonly #dependencies: SpeechPlaybackDependencies; + #abortController: AbortController | null = null; + #activeAudio: ActiveAudio | null = null; + #generation = 0; + + public constructor(dependencies: SpeechPlaybackDependencies) { + this.#dependencies = dependencies; + } + + public async play( + segment: CanonicalSpeechSegment, + events: SpeechPlaybackEvents = {}, + ): Promise { + this.cancel(); + if ( + segment.contentHash !== hashCanonicalSpeechText(segment.text) || + !segment.id.endsWith(`:${segment.contentHash}`) + ) { + throw fallbackError(); + } + const generation = this.#generation; + const abortController = new AbortController(); + this.#abortController = abortController; + + try { + const response = await waitForAbort( + this.#dependencies.fetch("/api/voice/speech", { + body: JSON.stringify({ segmentId: segment.id, text: segment.text }), + cache: "no-store", + headers: { "content-type": "application/json" }, + method: "POST", + signal: abortController.signal, + }), + abortController.signal, + ); + const contentType = response.headers + .get("content-type") + ?.split(";", 1)[0] + ?.trim() + .toLowerCase(); + if (!response.ok || contentType !== "audio/mpeg") { + await response.body?.cancel(); + throw fallbackError(); + } + + const blob = await waitForAbort(response.blob(), abortController.signal); + if (generation !== this.#generation || blob.size === 0) { + throw generation === this.#generation ? fallbackError() : abortError(); + } + + const objectUrl = this.#dependencies.createObjectURL(blob); + const audio = this.#dependencies.createAudio(objectUrl); + await new Promise((resolve, reject) => { + let settled = false; + let cleanup = () => undefined; + const settle = (finish: () => void) => { + if (settled) { + return; + } + settled = true; + cleanup(); + finish(); + }; + const handleEnded = () => settle(resolve); + const handleError = () => settle(() => reject(fallbackError())); + cleanup = () => { + audio.removeEventListener("ended", handleEnded); + audio.removeEventListener("error", handleError); + this.#dependencies.revokeObjectURL(objectUrl); + if (this.#activeAudio?.generation === generation) { + this.#activeAudio = null; + } + }; + this.#activeAudio = { + cancel: (reason) => { + audio.pause(); + settle(() => reject(reason)); + }, + generation, + }; + audio.addEventListener("ended", handleEnded); + audio.addEventListener("error", handleError); + void audio.play().then(() => { + if (!settled && generation === this.#generation) { + events.onPlaying?.(); + } + }, handleError); + }); + } catch (error) { + if (isAbortError(error)) { + throw error; + } + throw fallbackError(); + } finally { + if (this.#abortController === abortController) { + this.#abortController = null; + } + } + } + + public cancel(): void { + ++this.#generation; + const reason = abortError(); + this.#abortController?.abort(reason); + this.#abortController = null; + this.#activeAudio?.cancel(reason); + this.#activeAudio = null; + } +} diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-interview-control.test.tsx b/apps/petrinaut-website/src/main/app/voice-interview/voice-interview-control.test.tsx index 106d3ce7857..21c9411095f 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-interview-control.test.tsx +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-interview-control.test.tsx @@ -176,4 +176,38 @@ describe("voice interview control", () => { container.remove(); } }); + + test("announces synthesis, playback, and the AI-generated voice disclosure", () => { + const renderPhase = (phase: "synthesizing" | "playing") => + renderToStaticMarkup( + , + ); + + const synthesizing = renderPhase("synthesizing"); + expect(synthesizing).toContain( + "Microphone off. Creating AI-generated speech.", + ); + expect(synthesizing).toContain( + "Spoken responses use an AI-generated OpenAI voice.", + ); + + const playing = renderPhase("playing"); + expect(playing).toContain("Microphone off. Playing AI-generated speech."); + expect(playing).toContain( + "Spoken responses use an AI-generated OpenAI voice.", + ); + }); }); diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-interview-control.tsx b/apps/petrinaut-website/src/main/app/voice-interview/voice-interview-control.tsx index 32afc58f1d0..7eec4491ae5 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-interview-control.tsx +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-interview-control.tsx @@ -10,7 +10,9 @@ import { FaMicrophone, FaMicrophoneSlash } from "react-icons/fa6"; import { Button } from "@hashintel/ds-components"; import { css } from "@hashintel/ds-helpers/css"; +import { selectCanonicalSpeechSegments } from "./canonical-speech"; import { OpenAIRealtimeSession } from "./openai-realtime-session"; +import { SpeechPlaybackController } from "./speech-playback-controller"; import { VoiceTurnController, type VoiceTurnSnapshot, @@ -89,6 +91,12 @@ const statusStyle = css({ lineHeight: "relaxed", }); +const disclosureStyle = css({ + color: "neutral.s70", + fontSize: "xs", + lineHeight: "relaxed", +}); + const liveRegionStyle = css({ position: "absolute", width: "[1px]", @@ -162,6 +170,10 @@ const statusText = (snapshot: VoiceTurnSnapshot): string => { return "Microphone off. Sending the finalized transcript to Brunch."; case "waiting": return "Microphone off. Waiting for Brunch."; + case "synthesizing": + return "Microphone off. Creating AI-generated speech."; + case "playing": + return "Microphone off. Playing AI-generated speech."; case "recoverable-error": return `Microphone off. ${snapshot.errorMessage}`; } @@ -238,6 +250,9 @@ export const VoiceInterviewControlView = ({ {!isIdle && (

{statusText(snapshot)}

+

+ Spoken responses use an AI-generated OpenAI voice. +

{snapshot.partialText && (

@@ -311,8 +326,15 @@ const AvailableVoiceInterviewControl = ({ getUserMedia: (constraints) => navigator.mediaDevices.getUserMedia(constraints), }); + const playback = new SpeechPlaybackController({ + createAudio: (source) => new Audio(source), + createObjectURL: (blob) => URL.createObjectURL(blob), + fetch: globalThis.fetch.bind(globalThis), + revokeObjectURL: (url) => URL.revokeObjectURL(url), + }); const controller = new VoiceTurnController({ conversationId: context.conversationId, + playback, session, submitText: context.submitVoiceInput, }); @@ -331,8 +353,11 @@ const AvailableVoiceInterviewControl = ({ const [correction, setCorrection] = useState(""); useLayoutEffect(() => { - store.controller.updateChatStatus(context.status); - }, [context.status, store]); + store.controller.updateChat({ + canonicalSegments: selectCanonicalSpeechSegments(context.messages), + status: context.status, + }); + }, [context.messages, context.status, store]); useEffect( () => () => { diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts index 4f1384d530b..325a553cbb1 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts @@ -5,6 +5,7 @@ import { VoiceTurnController, } from "./voice-turn-controller"; +import type { CanonicalSpeechSegment } from "./canonical-speech"; import type { OpenAIRealtimeSessionEvent } from "./openai-realtime-session"; const createHarness = () => { @@ -22,8 +23,20 @@ const createHarness = () => { }), }; const submitText = vi.fn(async () => ({ kind: "message" as const })); + const playback = { + cancel: vi.fn(), + play: vi.fn( + async ( + _segment: CanonicalSpeechSegment, + events: { onPlaying?: () => void } = {}, + ) => { + events.onPlaying?.(); + }, + ), + }; const controller = new VoiceTurnController({ conversationId: "preview/net 1", + playback, session, submitText, }); @@ -31,6 +44,7 @@ const createHarness = () => { return { controller, emit: (event: OpenAIRealtimeSessionEvent) => listener?.(event), + playback, session, submitText, }; @@ -42,10 +56,45 @@ const key = (connectionEpoch: number, itemId: string, contentIndex = 0) => ({ itemId, }); +const canonicalSegment = ( + id: string, + text = `Canonical text for ${id}`, +): CanonicalSpeechSegment => ({ + contentHash: "fnv1a32:12345678", + id, + messageId: "assistant-1", + partId: "text:0", + source: "assistant-text", + text, +}); + +const updateChatStatus = ( + controller: VoiceTurnController, + status: "ready" | "submitted" | "streaming" | "error", +) => controller.updateChat({ canonicalSegments: [], status }); + describe("VoiceTurnController", () => { + test("does not surface an unrelated Brunch error before voice starts", async () => { + const harness = createHarness(); + + harness.controller.updateChat({ + canonicalSegments: [], + status: "error", + }); + + expect(harness.controller.getSnapshot().phase).toBe("idle"); + expect(harness.playback.cancel).not.toHaveBeenCalled(); + + await harness.controller.start(); + expect(harness.controller.getSnapshot()).toMatchObject({ + errorMessage: "Wait for Brunch to finish before starting voice input.", + phase: "recoverable-error", + }); + }); + test("refuses to open the microphone while Brunch is busy", async () => { const harness = createHarness(); - harness.controller.updateChatStatus("streaming"); + updateChatStatus(harness.controller, "streaming"); await harness.controller.start(); @@ -63,7 +112,7 @@ describe("VoiceTurnController", () => { const harness = createHarness(); const starting = harness.controller.start(); - harness.controller.updateChatStatus("streaming"); + updateChatStatus(harness.controller, "streaming"); await starting; expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( @@ -71,7 +120,7 @@ describe("VoiceTurnController", () => { ); expect(harness.controller.getSnapshot().phase).toBe("waiting"); - harness.controller.updateChatStatus("ready"); + updateChatStatus(harness.controller, "ready"); expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith(true); expect(harness.controller.getSnapshot().phase).toBe("listening"); @@ -81,14 +130,14 @@ describe("VoiceTurnController", () => { const harness = createHarness(); await harness.controller.start(); - harness.controller.updateChatStatus("submitted"); + updateChatStatus(harness.controller, "submitted"); expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( false, ); expect(harness.controller.getSnapshot().phase).toBe("waiting"); - harness.controller.updateChatStatus("ready"); + updateChatStatus(harness.controller, "ready"); expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith(true); expect(harness.controller.getSnapshot().phase).toBe("listening"); @@ -144,10 +193,10 @@ describe("VoiceTurnController", () => { phase: "waiting", }); - harness.controller.updateChatStatus("ready"); + updateChatStatus(harness.controller, "ready"); expect(harness.controller.getSnapshot().phase).toBe("waiting"); - harness.controller.updateChatStatus("streaming"); - harness.controller.updateChatStatus("ready"); + updateChatStatus(harness.controller, "streaming"); + updateChatStatus(harness.controller, "ready"); expect(harness.controller.getSnapshot().phase).toBe("listening"); expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith(true); }); @@ -305,8 +354,8 @@ describe("VoiceTurnController", () => { type: "completed", }); await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledOnce()); - harness.controller.updateChatStatus("streaming"); - harness.controller.updateChatStatus("ready"); + updateChatStatus(harness.controller, "streaming"); + updateChatStatus(harness.controller, "ready"); harness.emit({ key: key(1, "item-a"), @@ -372,7 +421,7 @@ describe("VoiceTurnController", () => { itemId: "empty-item", type: "input-committed", }); - harness.controller.updateChatStatus("streaming"); + updateChatStatus(harness.controller, "streaming"); harness.emit({ key: key(1, "empty-item"), @@ -386,7 +435,7 @@ describe("VoiceTurnController", () => { ); expect(harness.controller.getSnapshot().phase).toBe("waiting"); - harness.controller.updateChatStatus("ready"); + updateChatStatus(harness.controller, "ready"); expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith(true); expect(harness.controller.getSnapshot().phase).toBe("listening"); }); @@ -524,8 +573,8 @@ describe("VoiceTurnController", () => { type: "completed", }); await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledOnce()); - harness.controller.updateChatStatus("streaming"); - harness.controller.updateChatStatus("ready"); + updateChatStatus(harness.controller, "streaming"); + updateChatStatus(harness.controller, "ready"); await harness.controller.submitCorrection( "The incident manager closes it.", @@ -540,4 +589,205 @@ describe("VoiceTurnController", () => { false, ); }); + + test("seeds finalized history without replaying it when voice starts or reconnects", async () => { + const harness = createHarness(); + const history = canonicalSegment( + "canonical-speech:history:text%3A0:fnv1a32:12345678", + ); + harness.controller.updateChat({ + canonicalSegments: [history], + status: "ready", + }); + + await harness.controller.start(); + await harness.controller.reconnect(); + + expect(harness.playback.play).not.toHaveBeenCalled(); + expect(harness.controller.getSnapshot().phase).toBe("listening"); + }); + + test("speaks a finalized segment that arrives while voice is connecting", async () => { + const harness = createHarness(); + const response = canonicalSegment( + "canonical-speech:connecting:text%3A0:fnv1a32:12345678", + ); + + const starting = harness.controller.start(); + expect(harness.controller.getSnapshot().phase).toBe("connecting"); + harness.controller.updateChat({ + canonicalSegments: [response], + status: "ready", + }); + + expect(harness.playback.play).not.toHaveBeenCalled(); + await starting; + await vi.waitFor(() => + expect(harness.playback.play).toHaveBeenCalledWith( + response, + expect.any(Object), + ), + ); + }); + + test("queues canonical speech before an atomic ready update can reopen the microphone", async () => { + const harness = createHarness(); + let finishPlayback: (() => void) | undefined; + harness.playback.play.mockImplementationOnce(async (_segment, events) => { + events?.onPlaying?.(); + await new Promise((resolve) => { + finishPlayback = resolve; + }); + }); + await harness.controller.start(); + harness.emit({ + key: key(1, "answer"), + text: "A finalized answer", + type: "completed", + }); + await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledOnce()); + harness.controller.updateChat({ + canonicalSegments: [], + status: "streaming", + }); + const response = canonicalSegment( + "canonical-speech:response:text%3A0:fnv1a32:12345678", + ); + + harness.controller.updateChat({ + canonicalSegments: [response], + status: "ready", + }); + + expect(harness.playback.play).toHaveBeenCalledOnce(); + expect(harness.playback.play.mock.calls[0]?.[0]).toBe(response); + expect(typeof harness.playback.play.mock.calls[0]?.[1]?.onPlaying).toBe( + "function", + ); + expect(harness.controller.getSnapshot().phase).toBe("playing"); + expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( + false, + ); + + finishPlayback?.(); + await vi.waitFor(() => + expect(harness.controller.getSnapshot().phase).toBe("listening"), + ); + expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith(true); + }); + + test("speaks finalized segments in order and reopens only after the queue drains", async () => { + const harness = createHarness(); + const playbackResolvers: Array<() => void> = []; + harness.playback.play.mockImplementation(async (_segment, events) => { + events?.onPlaying?.(); + await new Promise((resolve) => playbackResolvers.push(resolve)); + }); + await harness.controller.start(); + const first = canonicalSegment( + "canonical-speech:first:text%3A0:fnv1a32:12345678", + ); + const second = canonicalSegment( + "canonical-speech:second:text%3A0:fnv1a32:12345678", + ); + + harness.controller.updateChat({ + canonicalSegments: [first], + status: "ready", + }); + await vi.waitFor(() => + expect(harness.playback.play).toHaveBeenCalledOnce(), + ); + harness.controller.updateChat({ + canonicalSegments: [first, second], + status: "ready", + }); + expect(harness.playback.play).toHaveBeenCalledOnce(); + expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( + false, + ); + + playbackResolvers.shift()?.(); + await vi.waitFor(() => + expect(harness.playback.play).toHaveBeenCalledTimes(2), + ); + expect(harness.playback.play).toHaveBeenNthCalledWith( + 2, + second, + expect.any(Object), + ); + expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( + false, + ); + + playbackResolvers.shift()?.(); + await vi.waitFor(() => + expect(harness.controller.getSnapshot().phase).toBe("listening"), + ); + expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith(true); + }); + + test("keeps the microphone closed and visible text available when speech fails", async () => { + const harness = createHarness(); + harness.playback.play.mockRejectedValueOnce( + new Error( + "The response could not be spoken. Read the visible text instead.", + ), + ); + await harness.controller.start(); + const response = canonicalSegment( + "canonical-speech:failed:text%3A0:fnv1a32:12345678", + "The visible response remains available.", + ); + + harness.controller.updateChat({ + canonicalSegments: [response], + status: "ready", + }); + + await vi.waitFor(() => + expect(harness.controller.getSnapshot()).toMatchObject({ + errorMessage: + "The response could not be spoken. Read the visible text instead.", + phase: "recoverable-error", + }), + ); + expect(response.text).toBe("The visible response remains available."); + expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( + false, + ); + }); + + test("cancels speech synchronously and rejects stale playback events when voice ends", async () => { + const harness = createHarness(); + let playbackEvents: { onPlaying?: () => void } | undefined; + let finishPlayback: (() => void) | undefined; + harness.playback.play.mockImplementationOnce(async (_segment, events) => { + playbackEvents = events; + await new Promise((resolve) => { + finishPlayback = resolve; + }); + }); + await harness.controller.start(); + harness.controller.updateChat({ + canonicalSegments: [ + canonicalSegment("canonical-speech:stale:text%3A0:fnv1a32:12345678"), + ], + status: "ready", + }); + await vi.waitFor(() => + expect(harness.playback.play).toHaveBeenCalledOnce(), + ); + + const ending = harness.controller.end(); + expect(harness.playback.cancel).toHaveBeenCalledOnce(); + playbackEvents?.onPlaying?.(); + finishPlayback?.(); + await ending; + + expect(harness.controller.getSnapshot().phase).toBe("idle"); + expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( + false, + ); + }); }); diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts index adffa720b0c..4958ef8f257 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts @@ -1,3 +1,4 @@ +import type { CanonicalSpeechSegment } from "./canonical-speech"; import type { OpenAIRealtimeSessionEvent, OpenAIRealtimeTranscriptKey, @@ -10,6 +11,8 @@ export type VoiceTurnPhase = | "transcribing" | "delivering" | "waiting" + | "synthesizing" + | "playing" | "recoverable-error"; export interface VoiceTurnSnapshot { @@ -34,12 +37,26 @@ interface SubmitTextInput { readonly text: string; } +interface SpeechPlayback { + cancel(): void; + play( + segment: CanonicalSpeechSegment, + events?: { readonly onPlaying?: () => void }, + ): Promise; +} + interface VoiceTurnControllerDependencies { readonly conversationId: string; + readonly playback: SpeechPlayback; readonly session: RealtimeSession; readonly submitText: (input: SubmitTextInput) => Promise; } +interface ChatUpdate { + readonly canonicalSegments: CanonicalSpeechSegment[]; + readonly status: ChatStatus; +} + type SnapshotListener = (snapshot: VoiceTurnSnapshot) => void; const transcriptKey = (key: OpenAIRealtimeTranscriptKey): string => @@ -67,25 +84,32 @@ const initialSnapshot: VoiceTurnSnapshot = { export class VoiceTurnController { readonly #conversationId: string; readonly #listeners = new Set(); + readonly #playback: SpeechPlayback; readonly #session: RealtimeSession; readonly #submitText: (input: SubmitTextInput) => Promise; readonly #completedKeys = new Set(); + readonly #seenSpeechSegmentIds = new Set(); + readonly #speechQueue: CanonicalSpeechSegment[] = []; #activeEpoch: number | null = null; #activeItemId: string | null = null; #activeKey: string | null = null; + #activeSpeechSegmentId: string | null = null; #awaitingChatCycle = false; #chatStatus: ChatStatus = "ready"; #generation = 0; #pendingDelivery: SubmitTextInput | null = null; #sawBusyChatStatus = false; #snapshot = initialSnapshot; + #speechLoopGeneration: number | null = null; public constructor({ conversationId, + playback, session, submitText, }: VoiceTurnControllerDependencies) { this.#conversationId = conversationId; + this.#playback = playback; this.#session = session; this.#submitText = submitText; session.subscribe((event) => this.#handleSessionEvent(event)); @@ -124,12 +148,13 @@ export class VoiceTurnController { this.#activeItemId = null; this.#activeKey = null; this.#completedKeys.clear(); - const canListen = this.#isChatReady(); + const canListen = this.#isChatReady() && this.#speechQueue.length === 0; this.#session.setMicrophoneEnabled(canListen); this.#update({ partialText: "", phase: canListen ? "listening" : "waiting", }); + this.#startSpeechQueueIfNeeded(); } catch (error) { if (generation !== this.#generation) { return; @@ -150,9 +175,13 @@ export class VoiceTurnController { this.#activeEpoch = null; this.#activeItemId = null; this.#activeKey = null; + this.#activeSpeechSegmentId = null; this.#awaitingChatCycle = false; this.#pendingDelivery = null; this.#sawBusyChatStatus = false; + this.#speechLoopGeneration = null; + this.#speechQueue.length = 0; + this.#playback.cancel(); this.#session.setMicrophoneEnabled(false); await this.#session.disconnect(); this.#update({ @@ -186,58 +215,69 @@ export class VoiceTurnController { }); } - public updateChatStatus(status: ChatStatus): void { + public updateChat({ canonicalSegments, status }: ChatUpdate): void { this.#chatStatus = status; - if (!this.#awaitingChatCycle) { - if (status === "ready" && this.#pendingDelivery !== null) { - const pendingDelivery = this.#pendingDelivery; - this.#pendingDelivery = null; - this.#session.setMicrophoneEnabled(false); - this.#update({ errorMessage: "", phase: "delivering" }); - void this.#deliver(pendingDelivery); - return; + const canQueueSpeech = + status !== "error" && + this.#snapshot.phase !== "idle" && + this.#snapshot.phase !== "recoverable-error"; + for (const segment of canonicalSegments) { + if (this.#seenSpeechSegmentIds.has(segment.id)) { + continue; } - if ( - (status === "submitted" || status === "streaming") && - (this.#snapshot.phase === "listening" || - this.#snapshot.phase === "transcribing") - ) { - this.#session.setMicrophoneEnabled(false); - this.#update({ phase: "waiting" }); - } else if ( - status === "ready" && - this.#activeEpoch !== null && - this.#snapshot.phase === "waiting" - ) { - this.#session.setMicrophoneEnabled(true); - this.#update({ errorMessage: "", phase: "listening" }); - } else if (status === "error" && this.#snapshot.phase === "waiting") { - this.#session.setMicrophoneEnabled(false); - this.#update({ - errorMessage: - "Brunch could not complete the current turn. Use the composer to retry.", - phase: "recoverable-error", - }); + this.#seenSpeechSegmentIds.add(segment.id); + if (canQueueSpeech) { + this.#speechQueue.push(segment); } - return; } - if (status === "submitted" || status === "streaming") { - this.#sawBusyChatStatus = true; - this.#update({ phase: "waiting" }); + if ( + status === "ready" && + !this.#awaitingChatCycle && + this.#pendingDelivery !== null + ) { + const pendingDelivery = this.#pendingDelivery; + this.#pendingDelivery = null; + this.#session.setMicrophoneEnabled(false); + this.#update({ errorMessage: "", phase: "delivering" }); + void this.#deliver(pendingDelivery); return; } + if (status === "error") { + if (this.#snapshot.phase === "idle") { + return; + } + const errorMessage = this.#awaitingChatCycle + ? "Brunch could not accept the voice turn. Use the composer to retry." + : "Brunch could not complete the current turn. Use the composer to retry."; this.#awaitingChatCycle = false; + this.#sawBusyChatStatus = false; + this.#speechQueue.length = 0; + this.#activeSpeechSegmentId = null; + this.#speechLoopGeneration = null; + this.#playback.cancel(); this.#session.setMicrophoneEnabled(false); - this.#update({ - errorMessage: - "Brunch could not accept the voice turn. Use the composer to retry.", - phase: "recoverable-error", - }); + this.#update({ errorMessage, phase: "recoverable-error" }); return; } - this.#reopenListeningIfReady(); + + this.#startSpeechQueueIfNeeded(); + if (status === "submitted" || status === "streaming") { + if (this.#awaitingChatCycle) { + this.#sawBusyChatStatus = true; + } + this.#session.setMicrophoneEnabled(false); + if ( + this.#speechLoopGeneration === null && + this.#snapshot.phase !== "idle" && + this.#snapshot.phase !== "recoverable-error" + ) { + this.#update({ phase: "waiting" }); + } + return; + } + this.#settleListeningIfReady(); } async #deliver(input: SubmitTextInput): Promise { @@ -260,7 +300,7 @@ export class VoiceTurnController { return; } this.#update({ phase: "waiting" }); - this.#reopenListeningIfReady(); + this.#settleListeningIfReady(); } catch { if (generation !== this.#generation) { return; @@ -299,6 +339,10 @@ export class VoiceTurnController { this.#awaitingChatCycle = false; this.#pendingDelivery = null; this.#sawBusyChatStatus = false; + this.#activeSpeechSegmentId = null; + this.#speechLoopGeneration = null; + this.#speechQueue.length = 0; + this.#playback.cancel(); this.#session.setMicrophoneEnabled(false); this.#update({ errorMessage: event.message, phase: "recoverable-error" }); return; @@ -358,12 +402,11 @@ export class VoiceTurnController { this.#activeItemId = null; this.#activeKey = null; if (!finalText) { - const canListen = this.#isChatReady(); - this.#session.setMicrophoneEnabled(canListen); this.#update({ partialText: "", - phase: canListen ? "listening" : "waiting", + phase: "waiting", }); + this.#settleListeningIfReady(); return; } @@ -386,16 +429,117 @@ export class VoiceTurnController { } } - #reopenListeningIfReady(): void { + #startSpeechQueueIfNeeded(): void { if ( - !this.#awaitingChatCycle || - !this.#sawBusyChatStatus || - this.#chatStatus !== "ready" + this.#speechLoopGeneration !== null || + this.#speechQueue.length === 0 || + this.#activeEpoch === null ) { return; } - this.#awaitingChatCycle = false; - this.#sawBusyChatStatus = false; + + const generation = this.#generation; + this.#speechLoopGeneration = generation; + this.#session.setMicrophoneEnabled(false); + this.#update({ errorMessage: "", phase: "synthesizing" }); + void this.#drainSpeechQueue(generation); + } + + async #drainSpeechQueue(generation: number): Promise { + while ( + generation === this.#generation && + this.#speechLoopGeneration === generation && + this.#activeEpoch !== null + ) { + const segment = this.#speechQueue.shift(); + if (!segment) { + break; + } + + this.#activeSpeechSegmentId = segment.id; + this.#session.setMicrophoneEnabled(false); + this.#update({ errorMessage: "", phase: "synthesizing" }); + try { + await this.#playback.play(segment, { + onPlaying: () => { + if ( + generation === this.#generation && + this.#speechLoopGeneration === generation && + this.#activeSpeechSegmentId === segment.id + ) { + this.#update({ phase: "playing" }); + } + }, + }); + } catch { + if ( + generation !== this.#generation || + this.#speechLoopGeneration !== generation + ) { + return; + } + this.#activeSpeechSegmentId = null; + this.#speechLoopGeneration = null; + this.#speechQueue.length = 0; + this.#session.setMicrophoneEnabled(false); + this.#update({ + errorMessage: + "The response could not be spoken. Read the visible text instead.", + phase: "recoverable-error", + }); + return; + } + + if ( + generation !== this.#generation || + this.#speechLoopGeneration !== generation + ) { + return; + } + this.#activeSpeechSegmentId = null; + } + + if ( + generation !== this.#generation || + this.#speechLoopGeneration !== generation + ) { + return; + } + this.#activeSpeechSegmentId = null; + this.#speechLoopGeneration = null; + this.#settleListeningIfReady(); + } + + #settleListeningIfReady(): void { + if ( + this.#activeEpoch === null || + this.#snapshot.phase === "recoverable-error" || + this.#speechLoopGeneration !== null || + this.#speechQueue.length > 0 + ) { + return; + } + + if (this.#awaitingChatCycle) { + if (!this.#sawBusyChatStatus || this.#chatStatus !== "ready") { + this.#session.setMicrophoneEnabled(false); + return; + } + this.#awaitingChatCycle = false; + this.#sawBusyChatStatus = false; + } + + if (!this.#isChatReady()) { + this.#session.setMicrophoneEnabled(false); + if ( + this.#snapshot.phase !== "transcribing" && + this.#snapshot.phase !== "delivering" + ) { + this.#update({ phase: "waiting" }); + } + return; + } + this.#session.setMicrophoneEnabled(true); this.#update({ errorMessage: "", phase: "listening" }); } diff --git a/apps/petrinaut-website/src/server/voice/openai-speech.test.ts b/apps/petrinaut-website/src/server/voice/openai-speech.test.ts new file mode 100644 index 00000000000..27410833978 --- /dev/null +++ b/apps/petrinaut-website/src/server/voice/openai-speech.test.ts @@ -0,0 +1,269 @@ +import { afterEach, describe, expect, test, vi } from "vitest"; + +import { + createOpenAISpeechHandler, + OPENAI_SPEECH_TIMEOUT_MS, +} from "./openai-speech"; + +const enabledEnvironment = { + OPENAI_VOICE_API_KEY: "server-secret", + PETRINAUT_OPENAI_VOICE_ENABLED: "true", + VERCEL_ENV: "preview", +}; + +const validSpeechRequest = { + segmentId: "canonical-speech:assistant-1:text%3A0:fnv1a32:69f1e741", + text: " Preserve this exact canonical text. ", +}; + +const createRequest = ( + body: BodyInit | null = JSON.stringify(validSpeechRequest), + overrides: ConstructorParameters[1] = {}, +) => + new Request("https://petrinaut.test/api/voice/speech", { + body, + headers: { + "content-type": "application/json", + origin: "https://petrinaut.test", + }, + method: "POST", + ...overrides, + }); + +describe("OpenAI Speech handler", () => { + afterEach(() => { + vi.useRealTimers(); + }); + + test("rejects unavailable, cross-origin, malformed, and oversized requests before OpenAI", async () => { + const fetch = vi.fn(); + const disabledHandler = createOpenAISpeechHandler({ + environment: {}, + fetch, + }); + const disabled = await disabledHandler(createRequest()); + expect(disabled.status).toBe(404); + expect(disabled.headers.get("cache-control")).toBe("no-store"); + + const handler = createOpenAISpeechHandler({ + environment: enabledEnvironment, + fetch, + }); + const cases = [ + createRequest(null, { method: "GET" }), + createRequest(undefined, { + headers: { + "content-type": "application/json", + origin: "https://attacker.test", + }, + }), + createRequest(undefined, { + headers: { origin: "https://petrinaut.test" }, + }), + createRequest("not-json"), + createRequest(JSON.stringify(null)), + createRequest(JSON.stringify({ ...validSpeechRequest, extra: true })), + createRequest(JSON.stringify({ ...validSpeechRequest, text: " " })), + createRequest( + JSON.stringify({ ...validSpeechRequest, text: "🙂".repeat(4_097) }), + ), + createRequest( + JSON.stringify({ ...validSpeechRequest, segmentId: "untrusted" }), + ), + createRequest( + JSON.stringify({ + ...validSpeechRequest, + segmentId: "canonical-speech:assistant-1:text%3A0:fnv1a32:12345678", + }), + ), + createRequest("{}", { + headers: { + "content-length": "32769", + "content-type": "application/json", + origin: "https://petrinaut.test", + }, + }), + ]; + + const responses = await Promise.all( + cases.map((request) => handler(request)), + ); + + expect(responses.map(({ status }) => status)).toEqual([ + 405, 403, 415, 400, 400, 400, 400, 400, 400, 400, 413, + ]); + expect( + responses.every( + (response) => response.headers.get("cache-control") === "no-store", + ), + ).toBe(true); + expect(fetch).not.toHaveBeenCalled(); + }); + + test("streams audio for the exact canonical text with fixed server policy", async () => { + const firstChunk = new Uint8Array([1, 2, 3]); + const secondChunk = new Uint8Array([4, 5]); + const fetch = vi.fn( + async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(firstChunk); + controller.enqueue(secondChunk); + controller.close(); + }, + }), + { headers: { "content-type": "audio/mpeg" } }, + ), + ); + const handler = createOpenAISpeechHandler({ + environment: enabledEnvironment, + fetch, + }); + + const response = await handler(createRequest()); + + expect(response.status).toBe(200); + expect(response.headers.get("cache-control")).toBe("no-store"); + expect(response.headers.get("content-type")).toBe("audio/mpeg"); + expect(fetch).toHaveBeenCalledOnce(); + const [url, request] = fetch.mock.calls[0]!; + expect(url).toBe("https://api.openai.com/v1/audio/speech"); + expect(request?.method).toBe("POST"); + expect(new Headers(request?.headers).get("authorization")).toBe( + "Bearer server-secret", + ); + expect(new Headers(request?.headers).get("content-type")).toBe( + "application/json", + ); + expect(request?.signal).toBeInstanceOf(AbortSignal); + const upstreamBody = JSON.parse(request?.body as string) as Record< + string, + unknown + >; + expect(upstreamBody).toEqual({ + input: validSpeechRequest.text, + model: "gpt-4o-mini-tts", + response_format: "mp3", + stream_format: "audio", + voice: "marin", + }); + expect(upstreamBody).not.toHaveProperty("instructions"); + expect(JSON.stringify(upstreamBody)).not.toContain("say exactly"); + expect(JSON.stringify(upstreamBody)).not.toContain("response.create"); + expect(new Uint8Array(await response.arrayBuffer())).toEqual( + new Uint8Array([1, 2, 3, 4, 5]), + ); + }); + + test("sanitizes upstream and non-audio failures", async () => { + const fetch = vi + .fn() + .mockResolvedValueOnce( + new Response("upstream secret diagnostics", { status: 400 }), + ) + .mockResolvedValueOnce( + Response.json({ secret: "not audio" }, { status: 200 }), + ) + .mockResolvedValueOnce( + new Response(new Uint8Array([1, 2, 3]), { + headers: { "content-type": "audio/wav" }, + }), + ); + const handler = createOpenAISpeechHandler({ + environment: enabledEnvironment, + fetch, + }); + + const upstreamFailure = await handler(createRequest()); + const nonAudio = await handler(createRequest()); + const wrongAudioFormat = await handler(createRequest()); + + for (const response of [upstreamFailure, nonAudio, wrongAudioFormat]) { + expect(response.status).toBe(502); + expect(await response.text()).toBe( + "The response could not be spoken. Read the visible text instead.", + ); + expect(response.headers.get("cache-control")).toBe("no-store"); + } + }); + + test("aborts a slow upstream request after the speech timeout", async () => { + vi.useFakeTimers(); + const fetch = vi.fn( + (_input, init) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => + reject(new DOMException("aborted", "AbortError")), + ); + }), + ); + const handler = createOpenAISpeechHandler({ + environment: enabledEnvironment, + fetch, + }); + + const responsePromise = handler(createRequest()); + await vi.advanceTimersByTimeAsync(OPENAI_SPEECH_TIMEOUT_MS); + + const response = await responsePromise; + expect(response.status).toBe(504); + expect(await response.text()).toBe( + "The response could not be spoken. Read the visible text instead.", + ); + }); + + test("propagates browser disconnect while waiting for OpenAI", async () => { + const requestAbortController = new AbortController(); + const fetch = vi.fn( + (_input, init) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => + reject(new DOMException("aborted", "AbortError")), + ); + }), + ); + const handler = createOpenAISpeechHandler({ + environment: enabledEnvironment, + fetch, + }); + + const responsePromise = handler( + createRequest(undefined, { signal: requestAbortController.signal }), + ); + await vi.waitFor(() => expect(fetch).toHaveBeenCalledOnce()); + requestAbortController.abort(); + + const response = await responsePromise; + expect(fetch.mock.calls[0]?.[1]?.signal?.aborted).toBe(true); + expect(response.status).toBe(502); + }); + + test("cancels the OpenAI stream when browser playback stops reading", async () => { + const upstreamCancel = vi.fn(); + const fetch = vi.fn( + async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3])); + }, + cancel: upstreamCancel, + }), + { headers: { "content-type": "audio/mpeg" } }, + ), + ); + const handler = createOpenAISpeechHandler({ + environment: enabledEnvironment, + fetch, + }); + + const response = await handler(createRequest()); + const reader = response.body!.getReader(); + await reader.read(); + await reader.cancel("playback stopped"); + + expect(fetch.mock.calls[0]?.[1]?.signal?.aborted).toBe(true); + expect(upstreamCancel).toHaveBeenCalledWith("playback stopped"); + }); +}); diff --git a/apps/petrinaut-website/src/server/voice/openai-speech.ts b/apps/petrinaut-website/src/server/voice/openai-speech.ts new file mode 100644 index 00000000000..ad579abac90 --- /dev/null +++ b/apps/petrinaut-website/src/server/voice/openai-speech.ts @@ -0,0 +1,243 @@ +import { hashCanonicalSpeechText } from "../../canonical-speech-fingerprint"; +import { getOpenAIVoiceAvailability } from "./openai-voice-policy"; + +const OPENAI_SPEECH_ENDPOINT = "https://api.openai.com/v1/audio/speech"; +const MAX_REQUEST_BYTES = 32_768; +const MAX_SPEECH_CHARACTERS = 4_096; +const SPEECH_ERROR_MESSAGE = + "The response could not be spoken. Read the visible text instead."; +const timeoutError = new DOMException("Upstream timed out", "TimeoutError"); + +export const OPENAI_SPEECH_TIMEOUT_MS = 25_000; + +interface VoiceEnvironment { + readonly NODE_ENV?: string; + readonly OPENAI_VOICE_API_KEY?: string; + readonly PETRINAUT_OPENAI_VOICE_ENABLED?: string; + readonly VERCEL_ENV?: string; +} + +interface OpenAISpeechDependencies { + readonly environment: VoiceEnvironment; + readonly fetch: typeof globalThis.fetch; +} + +interface SpeechRequest { + readonly segmentId: string; + readonly text: string; +} + +const response = ( + body: BodyInit | null, + status: number, + headers?: HeadersInit, +): Response => { + const responseHeaders = new Headers(headers); + responseHeaders.set("cache-control", "no-store"); + return new Response(body, { headers: responseHeaders, status }); +}; + +const readRequestBody = async ( + request: Request, +): Promise => { + const declaredLengthHeader = request.headers.get("content-length"); + if (declaredLengthHeader !== null) { + const declaredLength = Number(declaredLengthHeader); + if (Number.isFinite(declaredLength) && declaredLength > MAX_REQUEST_BYTES) { + return response("The speech request is too large.", 413); + } + } + + if (!request.body) { + return ""; + } + + const reader = request.body.getReader(); + const chunks: Uint8Array[] = []; + let totalBytes = 0; + let requestComplete = false; + + while (!requestComplete) { + const { done, value } = await reader.read(); + if (done) { + requestComplete = true; + continue; + } + totalBytes += value.byteLength; + if (totalBytes > MAX_REQUEST_BYTES) { + await reader.cancel(); + return response("The speech request is too large.", 413); + } + chunks.push(value); + } + + const body = new Uint8Array(totalBytes); + let offset = 0; + for (const chunk of chunks) { + body.set(chunk, offset); + offset += chunk.byteLength; + } + return new TextDecoder().decode(body); +}; + +const isSpeechRequest = (value: unknown): value is SpeechRequest => { + if (typeof value !== "object" || value === null || Array.isArray(value)) { + return false; + } + + const record = value as Record; + if ( + Object.keys(record).length !== 2 || + typeof record.segmentId !== "string" || + typeof record.text !== "string" + ) { + return false; + } + + return ( + record.segmentId.length <= 1_200 && + /^canonical-speech:[^:]+:[^:]+:fnv1a32:[0-9a-f]{8}$/u.test( + record.segmentId, + ) && + record.segmentId.endsWith(`:${hashCanonicalSpeechText(record.text)}`) && + Boolean(record.text.trim()) && + Array.from(record.text).length <= MAX_SPEECH_CHARACTERS + ); +}; + +const proxyAudioStream = ( + upstreamBody: ReadableStream, + abortController: AbortController, + cleanup: () => void, +): ReadableStream => { + const reader = upstreamBody.getReader(); + let finished = false; + const finish = () => { + if (!finished) { + finished = true; + cleanup(); + } + }; + + return new ReadableStream({ + async cancel(reason) { + abortController.abort(reason); + try { + await reader.cancel(reason); + } finally { + finish(); + } + }, + async pull(controller) { + try { + const { done, value } = await reader.read(); + if (done) { + controller.close(); + finish(); + return; + } + controller.enqueue(value); + } catch (error) { + controller.error(error); + finish(); + } + }, + }); +}; + +export const createOpenAISpeechHandler = + ({ environment, fetch }: OpenAISpeechDependencies) => + async (request: Request): Promise => { + if (request.method !== "POST") { + return response("Method not allowed.", 405, { allow: "POST" }); + } + + if (request.headers.get("origin") !== new URL(request.url).origin) { + return response("Forbidden.", 403); + } + + const contentType = request.headers + .get("content-type") + ?.split(";", 1)[0] + ?.trim() + .toLowerCase(); + if (contentType !== "application/json") { + return response("The request must contain JSON.", 415); + } + + if (!getOpenAIVoiceAvailability(environment).available) { + return response("Not found.", 404); + } + + let parsedBody: unknown; + try { + const body = await readRequestBody(request); + if (body instanceof Response) { + return body; + } + parsedBody = JSON.parse(body); + } catch { + return response("The speech request is invalid.", 400); + } + + if (!isSpeechRequest(parsedBody)) { + return response("The speech request is invalid.", 400); + } + + const abortController = new AbortController(); + const abortForTimeout = () => abortController.abort(timeoutError); + const abortForRequest = () => abortController.abort(); + const timeout = globalThis.setTimeout( + abortForTimeout, + OPENAI_SPEECH_TIMEOUT_MS, + ); + request.signal.addEventListener("abort", abortForRequest, { once: true }); + const cleanup = () => { + globalThis.clearTimeout(timeout); + request.signal.removeEventListener("abort", abortForRequest); + }; + + try { + const upstreamResponse = await fetch(OPENAI_SPEECH_ENDPOINT, { + body: JSON.stringify({ + input: parsedBody.text, + model: "gpt-4o-mini-tts", + response_format: "mp3", + stream_format: "audio", + voice: "marin", + }), + headers: { + authorization: `Bearer ${environment.OPENAI_VOICE_API_KEY!.trim()}`, + "content-type": "application/json", + }, + method: "POST", + signal: abortController.signal, + }); + const upstreamContentType = upstreamResponse.headers + .get("content-type") + ?.split(";", 1)[0] + ?.trim() + .toLowerCase(); + if ( + !upstreamResponse.ok || + upstreamContentType !== "audio/mpeg" || + !upstreamResponse.body + ) { + await upstreamResponse.body?.cancel(); + cleanup(); + return response(SPEECH_ERROR_MESSAGE, 502); + } + + return response( + proxyAudioStream(upstreamResponse.body, abortController, cleanup), + 200, + { "content-type": "audio/mpeg" }, + ); + } catch { + cleanup(); + return response( + SPEECH_ERROR_MESSAGE, + abortController.signal.reason === timeoutError ? 504 : 502, + ); + } + }; diff --git a/apps/petrinaut-website/vercel.json b/apps/petrinaut-website/vercel.json index ee12cb859b6..068e54e05c2 100644 --- a/apps/petrinaut-website/vercel.json +++ b/apps/petrinaut-website/vercel.json @@ -64,6 +64,9 @@ }, "api/voice/realtime-call.ts": { "maxDuration": 30 + }, + "api/voice/speech.ts": { + "maxDuration": 30 } } } diff --git a/apps/petrinaut-website/vite.config.ts b/apps/petrinaut-website/vite.config.ts index ff28dedd11d..07cbf8cc429 100644 --- a/apps/petrinaut-website/vite.config.ts +++ b/apps/petrinaut-website/vite.config.ts @@ -26,6 +26,7 @@ const apiModules = [ ["/api/oembed", "/api/oembed.ts"], ["/api/voice/config", "/api/voice/config.ts"], ["/api/voice/realtime-call", "/api/voice/realtime-call.ts"], + ["/api/voice/speech", "/api/voice/speech.ts"], ] as const; // Plugin required to serve the Vercel fetch handlers in dev. In production, diff --git a/libs/@hashintel/brunch-agent/docs/adr/0009-openai-voice-ui-turn-shell.md b/libs/@hashintel/brunch-agent/docs/adr/0009-openai-voice-ui-turn-shell.md index b6b3db45e68..ed8625bd55d 100644 --- a/libs/@hashintel/brunch-agent/docs/adr/0009-openai-voice-ui-turn-shell.md +++ b/libs/@hashintel/brunch-agent/docs/adr/0009-openai-voice-ui-turn-shell.md @@ -73,6 +73,26 @@ production recovery or public availability. acoustic-pronunciation guarantee requires a new decision. Exact lexical input to Speech is the enforceable fidelity contract in this record. +## Voice PR 3 implementation evidence + +- The website selects only finalized assistant text parts and schema-validated + `brunch_ask.input.question` values. Segment identity combines the AI SDK message ID, text-part + index or tool-call ID, and an exact-text fingerprint. Reasoning, user/system text, partial text, + tool output, malformed asks, and other tools are excluded. +- The Brunch projection test fixes the structured-ask boundary: an awaiting `brunch_ask` emits no + duplicate plain-text question. If that contract changes, the projection must be corrected + rather than hiding duplicates with fuzzy text matching in the voice layer. +- The app-owned Speech edge forwards the exact selected text to OpenAI's dedicated Speech API with + fixed server policy: `gpt-4o-mini-tts`, the `marin` voice, MP3 response format, and no delivery + instruction. The Realtime session remains transcription-only. +- The turn controller receives chat status and canonical segments atomically, queues speech in + order, seeds pre-existing segments as already seen, and rejects stale playback generations. + The microphone cannot reopen between Brunch becoming ready and speech being queued; ending or + reconnecting cancels synthesis and playback before media teardown. +- The preview visibly discloses that spoken responses use an AI-generated OpenAI voice. Speech + failure keeps the canonical response visible, closes the microphone, and requires an explicit + recovery action. + ## Revisit condition Revisit if the unified OpenAI WebRTC initialization API cannot enforce server-owned transcription diff --git a/libs/@hashintel/petrinaut/docs/ai-assistant.md b/libs/@hashintel/petrinaut/docs/ai-assistant.md index 3f1ea025b9a..8b76dc57393 100644 --- a/libs/@hashintel/petrinaut/docs/ai-assistant.md +++ b/libs/@hashintel/petrinaut/docs/ai-assistant.md @@ -28,6 +28,13 @@ other follow-up that must not answer the pending question. If the host offers voice input, a finalized spoken turn is held while an existing response finishes and is submitted when the conversation is ready. +When the Brunch voice preview is enabled by the host, the additional control can accept finalized +microphone transcripts and speak finalized assistant responses. Live transcript fragments are +labelled **not sent** and do not enter the conversation. The microphone is off while Brunch is +working or a response is playing. Spoken responses use an AI-generated OpenAI voice, as disclosed +in the voice status panel. If speech fails, the response remains visible to read and the voice +control offers recovery instead of changing or regenerating the text. + **Clear AI chat** via the delete button in the top right of the panel: wipes the conversation, stops any in-flight stream, and tells the host app to forget the messages (if the host persists them). ## What the assistant can do From be2645736d9078e145dbd347cae9399f5b4e56ca Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Thu, 27 Aug 2026 14:04:48 +0200 Subject: [PATCH 02/10] Fix voice session and composer regressions Co-authored-by: Cursor --- .../voice-turn-controller.test.ts | 52 +++++++++++++++++++ .../voice-interview/voice-turn-controller.ts | 10 +++- 2 files changed, 60 insertions(+), 2 deletions(-) diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts index 325a553cbb1..8eaa0dfc17b 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts @@ -113,6 +113,7 @@ describe("VoiceTurnController", () => { const starting = harness.controller.start(); updateChatStatus(harness.controller, "streaming"); + expect(harness.controller.getSnapshot().phase).toBe("connecting"); await starting; expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( @@ -126,6 +127,29 @@ describe("VoiceTurnController", () => { expect(harness.controller.getSnapshot().phase).toBe("listening"); }); + test("preserves a Brunch error that occurs while voice is connecting", async () => { + const harness = createHarness(); + let finishConnection: (() => void) | undefined; + harness.session.connect.mockImplementationOnce( + () => + new Promise((resolve) => { + finishConnection = () => resolve(1); + }), + ); + + const starting = harness.controller.start(); + updateChatStatus(harness.controller, "error"); + finishConnection?.(); + await starting; + + expect(harness.session.disconnect).toHaveBeenCalledOnce(); + expect(harness.controller.getSnapshot()).toMatchObject({ + errorMessage: + "Brunch could not complete the current turn. Use the composer to retry.", + phase: "recoverable-error", + }); + }); + test("pauses an open microphone for a non-voice Brunch turn", async () => { const harness = createHarness(); await harness.controller.start(); @@ -440,6 +464,34 @@ describe("VoiceTurnController", () => { expect(harness.controller.getSnapshot().phase).toBe("listening"); }); + test("finishes an in-flight transcript when Brunch becomes busy", async () => { + const harness = createHarness(); + await harness.controller.start(); + harness.emit({ + connectionEpoch: 1, + itemId: "item-a", + type: "input-committed", + }); + harness.emit({ + key: key(1, "item-a"), + text: "The support", + type: "partial", + }); + + updateChatStatus(harness.controller, "streaming"); + + expect(harness.controller.getSnapshot().phase).toBe("transcribing"); + harness.emit({ + key: key(1, "item-a"), + text: "The support lead triages it.", + type: "completed", + }); + await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledOnce()); + expect(harness.submitText).toHaveBeenCalledWith( + expect.objectContaining({ text: "The support lead triages it." }), + ); + }); + test("rejects events from a stopped epoch after reconnect", async () => { const harness = createHarness(); await harness.controller.start(); diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts index 4958ef8f257..32d5ba54d65 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts @@ -251,6 +251,10 @@ export class VoiceTurnController { const errorMessage = this.#awaitingChatCycle ? "Brunch could not accept the voice turn. Use the composer to retry." : "Brunch could not complete the current turn. Use the composer to retry."; + ++this.#generation; + this.#activeEpoch = null; + this.#activeItemId = null; + this.#activeKey = null; this.#awaitingChatCycle = false; this.#sawBusyChatStatus = false; this.#speechQueue.length = 0; @@ -258,6 +262,7 @@ export class VoiceTurnController { this.#speechLoopGeneration = null; this.#playback.cancel(); this.#session.setMicrophoneEnabled(false); + void this.#session.disconnect(); this.#update({ errorMessage, phase: "recoverable-error" }); return; } @@ -270,8 +275,9 @@ export class VoiceTurnController { this.#session.setMicrophoneEnabled(false); if ( this.#speechLoopGeneration === null && - this.#snapshot.phase !== "idle" && - this.#snapshot.phase !== "recoverable-error" + (this.#snapshot.phase === "listening" || + this.#snapshot.phase === "delivering" || + this.#snapshot.phase === "waiting") ) { this.#update({ phase: "waiting" }); } From 90796d62462f5ccbcb8968d845142fa4a2929a02 Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Thu, 27 Aug 2026 16:14:21 +0200 Subject: [PATCH 03/10] Preserve speech failure after voice delivery Co-authored-by: Cursor --- .../voice-turn-controller.test.ts | 47 +++++++++++++++++++ .../voice-interview/voice-turn-controller.ts | 4 +- 2 files changed, 50 insertions(+), 1 deletion(-) diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts index 8eaa0dfc17b..f05d5f1d356 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts @@ -810,6 +810,53 @@ describe("VoiceTurnController", () => { ); }); + test("does not overwrite a speech failure when delivery resolves later", async () => { + const harness = createHarness(); + let finishDelivery: (() => void) | undefined; + const delivery = new Promise<{ kind: "message" }>((resolve) => { + finishDelivery = () => resolve({ kind: "message" }); + }); + harness.submitText.mockImplementationOnce(() => delivery); + harness.playback.play.mockRejectedValueOnce( + new Error( + "The response could not be spoken. Read the visible text instead.", + ), + ); + await harness.controller.start(); + harness.emit({ + key: key(1, "answer"), + text: "A finalized answer", + type: "completed", + }); + await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledOnce()); + harness.controller.updateChat({ + canonicalSegments: [], + status: "streaming", + }); + harness.controller.updateChat({ + canonicalSegments: [ + canonicalSegment("canonical-speech:failed:text%3A0:fnv1a32:12345678"), + ], + status: "ready", + }); + await vi.waitFor(() => + expect(harness.controller.getSnapshot().phase).toBe("recoverable-error"), + ); + + finishDelivery?.(); + await delivery; + await Promise.resolve(); + + expect(harness.controller.getSnapshot()).toMatchObject({ + errorMessage: + "The response could not be spoken. Read the visible text instead.", + phase: "recoverable-error", + }); + expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( + false, + ); + }); + test("cancels speech synchronously and rejects stale playback events when voice ends", async () => { const harness = createHarness(); let playbackEvents: { onPlaying?: () => void } | undefined; diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts index 32d5ba54d65..92b9149fd7c 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts @@ -305,7 +305,9 @@ export class VoiceTurnController { ) { return; } - this.#update({ phase: "waiting" }); + if (this.#snapshot.phase === "delivering") { + this.#update({ phase: "waiting" }); + } this.#settleListeningIfReady(); } catch { if (generation !== this.#generation) { From 8c5c82f14ccecdba947b2d6252c5953d0706d534 Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Mon, 31 Aug 2026 21:26:42 +0200 Subject: [PATCH 04/10] Preserve live transcription during chat updates Keep ready chat updates from reopening the microphone or starting queued speech before the active transcript is finalized. Co-authored-by: Cursor --- .../voice-turn-controller.test.ts | 39 +++++++++++++++++++ .../voice-interview/voice-turn-controller.ts | 8 +++- 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts index f05d5f1d356..ef048da7b3b 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts @@ -492,6 +492,45 @@ describe("VoiceTurnController", () => { ); }); + test("finishes an in-flight transcript when a ready chat update arrives", async () => { + const harness = createHarness(); + await harness.controller.start(); + harness.emit({ + connectionEpoch: 1, + itemId: "item-a", + type: "input-committed", + }); + harness.emit({ + key: key(1, "item-a"), + text: "The support", + type: "partial", + }); + + harness.controller.updateChat({ + canonicalSegments: [ + canonicalSegment("canonical-speech:ready:text%3A0:fnv1a32:12345678"), + ], + status: "ready", + }); + + expect(harness.controller.getSnapshot().phase).toBe("transcribing"); + expect(harness.playback.play).not.toHaveBeenCalled(); + expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( + false, + ); + + harness.emit({ + key: key(1, "item-a"), + text: "The support lead triages it.", + type: "completed", + }); + + await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledOnce()); + expect(harness.submitText).toHaveBeenCalledWith( + expect.objectContaining({ text: "The support lead triages it." }), + ); + }); + test("rejects events from a stopped epoch after reconnect", async () => { const harness = createHarness(); await harness.controller.start(); diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts index 92b9149fd7c..fa94ad18594 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts @@ -441,7 +441,8 @@ export class VoiceTurnController { if ( this.#speechLoopGeneration !== null || this.#speechQueue.length === 0 || - this.#activeEpoch === null + this.#activeEpoch === null || + this.#snapshot.phase === "transcribing" ) { return; } @@ -528,6 +529,11 @@ export class VoiceTurnController { return; } + if (this.#snapshot.phase === "transcribing") { + this.#session.setMicrophoneEnabled(false); + return; + } + if (this.#awaitingChatCycle) { if (!this.#sawBusyChatStatus || this.#chatStatus !== "ready") { this.#session.setMicrophoneEnabled(false); From 2dbf3843832cdf6c31ec9c7b27c478c7e1c688c8 Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Mon, 31 Aug 2026 22:09:35 +0200 Subject: [PATCH 05/10] Remove redundant voice transcription guard Keep the early transcribing return authoritative so intermediate stack branches type-check without changing runtime behavior. Co-authored-by: Cursor --- .../src/main/app/voice-interview/voice-turn-controller.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts index fa94ad18594..a81d7616f5e 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts @@ -545,10 +545,7 @@ export class VoiceTurnController { if (!this.#isChatReady()) { this.#session.setMicrophoneEnabled(false); - if ( - this.#snapshot.phase !== "transcribing" && - this.#snapshot.phase !== "delivering" - ) { + if (this.#snapshot.phase !== "delivering") { this.#update({ phase: "waiting" }); } return; From 785125e6237622cfdbaa73eaeaab90028218994d Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Mon, 31 Aug 2026 23:19:54 +0200 Subject: [PATCH 06/10] Resume queued speech after transcription Co-authored-by: Cursor --- .../voice-turn-controller.test.ts | 53 +++++++++++++++++-- .../voice-interview/voice-turn-controller.ts | 8 ++- 2 files changed, 54 insertions(+), 7 deletions(-) diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts index ef048da7b3b..e89bdebf380 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts @@ -228,7 +228,7 @@ describe("VoiceTurnController", () => { test("queues a late finalized transcript until Brunch is ready", async () => { const harness = createHarness(); await harness.controller.start(); - harness.controller.updateChatStatus("streaming"); + updateChatStatus(harness.controller, "streaming"); expect(harness.controller.getSnapshot().phase).toBe("waiting"); harness.emit({ @@ -249,7 +249,7 @@ describe("VoiceTurnController", () => { phase: "waiting", }); - harness.controller.updateChatStatus("ready"); + updateChatStatus(harness.controller, "ready"); await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledOnce()); expect(harness.submitText).toHaveBeenCalledWith( @@ -297,7 +297,7 @@ describe("VoiceTurnController", () => { test("retries a finalized transcript rejected by a concurrent Brunch turn", async () => { const harness = createHarness(); harness.submitText.mockImplementationOnce(async () => { - harness.controller.updateChatStatus("streaming"); + updateChatStatus(harness.controller, "streaming"); throw new Error("Brunch became busy"); }); await harness.controller.start(); @@ -313,7 +313,7 @@ describe("VoiceTurnController", () => { ); expect(harness.controller.getSnapshot().errorMessage).toBe(""); - harness.controller.updateChatStatus("ready"); + updateChatStatus(harness.controller, "ready"); await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledTimes(2)); expect(harness.submitText).toHaveBeenLastCalledWith( @@ -464,7 +464,7 @@ describe("VoiceTurnController", () => { expect(harness.controller.getSnapshot().phase).toBe("listening"); }); - test("finishes an in-flight transcript when Brunch becomes busy", async () => { + test("queues an in-flight transcript when Brunch becomes busy", async () => { const harness = createHarness(); await harness.controller.start(); harness.emit({ @@ -486,6 +486,15 @@ describe("VoiceTurnController", () => { text: "The support lead triages it.", type: "completed", }); + + expect(harness.submitText).not.toHaveBeenCalled(); + expect(harness.controller.getSnapshot()).toMatchObject({ + lastCommittedText: "The support lead triages it.", + phase: "waiting", + }); + + updateChatStatus(harness.controller, "ready"); + await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledOnce()); expect(harness.submitText).toHaveBeenCalledWith( expect.objectContaining({ text: "The support lead triages it." }), @@ -531,6 +540,40 @@ describe("VoiceTurnController", () => { ); }); + test("starts queued speech after an empty transcript finishes", async () => { + const harness = createHarness(); + const response = canonicalSegment( + "canonical-speech:queued:text%3A0:fnv1a32:12345678", + ); + await harness.controller.start(); + harness.emit({ + connectionEpoch: 1, + itemId: "empty-item", + type: "input-committed", + }); + harness.controller.updateChat({ + canonicalSegments: [response], + status: "ready", + }); + + expect(harness.controller.getSnapshot().phase).toBe("transcribing"); + expect(harness.playback.play).not.toHaveBeenCalled(); + + harness.emit({ + key: key(1, "empty-item"), + text: " ", + type: "completed", + }); + + await vi.waitFor(() => + expect(harness.playback.play).toHaveBeenCalledWith( + response, + expect.any(Object), + ), + ); + expect(harness.session.setMicrophoneEnabled).toHaveBeenCalledWith(false); + }); + test("rejects events from a stopped epoch after reconnect", async () => { const harness = createHarness(); await harness.controller.start(); diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts index a81d7616f5e..2ea4f7bc728 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts @@ -523,8 +523,7 @@ export class VoiceTurnController { if ( this.#activeEpoch === null || this.#snapshot.phase === "recoverable-error" || - this.#speechLoopGeneration !== null || - this.#speechQueue.length > 0 + this.#speechLoopGeneration !== null ) { return; } @@ -534,6 +533,11 @@ export class VoiceTurnController { return; } + if (this.#speechQueue.length > 0) { + this.#startSpeechQueueIfNeeded(); + return; + } + if (this.#awaitingChatCycle) { if (!this.#sawBusyChatStatus || this.#chatStatus !== "ready") { this.#session.setMicrophoneEnabled(false); From 1881cccfdd162c57601d8f3e806d2870dc30d709 Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Tue, 1 Sep 2026 00:11:42 +0200 Subject: [PATCH 07/10] Stop speech requests after browser aborts Co-authored-by: Cursor --- .../src/server/voice/openai-speech.test.ts | 51 +++++++++++++++++++ .../src/server/voice/openai-speech.ts | 24 +++++++-- 2 files changed, 71 insertions(+), 4 deletions(-) diff --git a/apps/petrinaut-website/src/server/voice/openai-speech.test.ts b/apps/petrinaut-website/src/server/voice/openai-speech.test.ts index 27410833978..2d19930d1c7 100644 --- a/apps/petrinaut-website/src/server/voice/openai-speech.test.ts +++ b/apps/petrinaut-website/src/server/voice/openai-speech.test.ts @@ -100,6 +100,57 @@ describe("OpenAI Speech handler", () => { expect(fetch).not.toHaveBeenCalled(); }); + test("does not contact OpenAI for a pre-aborted request", async () => { + const requestAbortController = new AbortController(); + requestAbortController.abort(); + const fetch = vi.fn(); + const handler = createOpenAISpeechHandler({ + environment: enabledEnvironment, + fetch, + }); + + const response = await handler( + createRequest(undefined, { signal: requestAbortController.signal }), + ); + + expect(response.status).toBe(502); + expect(fetch).not.toHaveBeenCalled(); + }); + + test("does not contact OpenAI when the request aborts while its body is read", async () => { + const requestAbortController = new AbortController(); + const encodedRequest = new TextEncoder().encode( + JSON.stringify(validSpeechRequest), + ); + let finishBody: (() => void) | undefined; + const body = new ReadableStream({ + start(controller) { + finishBody = () => { + controller.enqueue(encodedRequest); + controller.close(); + }; + }, + }); + const fetch = vi.fn(); + const handler = createOpenAISpeechHandler({ + environment: enabledEnvironment, + fetch, + }); + + const responsePromise = handler( + createRequest(body, { + duplex: "half", + signal: requestAbortController.signal, + } as RequestInit), + ); + requestAbortController.abort(); + finishBody?.(); + const response = await responsePromise; + + expect(response.status).toBe(502); + expect(fetch).not.toHaveBeenCalled(); + }); + test("streams audio for the exact canonical text with fixed server policy", async () => { const firstChunk = new Uint8Array([1, 2, 3]); const secondChunk = new Uint8Array([4, 5]); diff --git a/apps/petrinaut-website/src/server/voice/openai-speech.ts b/apps/petrinaut-website/src/server/voice/openai-speech.ts index ad579abac90..837e90b6de9 100644 --- a/apps/petrinaut-website/src/server/voice/openai-speech.ts +++ b/apps/petrinaut-website/src/server/voice/openai-speech.ts @@ -169,35 +169,51 @@ export const createOpenAISpeechHandler = return response("Not found.", 404); } + const abortController = new AbortController(); + const abortForRequest = () => abortController.abort(); + request.signal.addEventListener("abort", abortForRequest, { once: true }); + if (request.signal.aborted) { + abortForRequest(); + } + const removeRequestAbortListener = () => { + request.signal.removeEventListener("abort", abortForRequest); + }; + let parsedBody: unknown; try { + abortController.signal.throwIfAborted(); const body = await readRequestBody(request); + abortController.signal.throwIfAborted(); if (body instanceof Response) { + removeRequestAbortListener(); return body; } parsedBody = JSON.parse(body); } catch { + removeRequestAbortListener(); + if (abortController.signal.aborted) { + return response(SPEECH_ERROR_MESSAGE, 502); + } return response("The speech request is invalid.", 400); } if (!isSpeechRequest(parsedBody)) { + removeRequestAbortListener(); return response("The speech request is invalid.", 400); } - const abortController = new AbortController(); const abortForTimeout = () => abortController.abort(timeoutError); - const abortForRequest = () => abortController.abort(); const timeout = globalThis.setTimeout( abortForTimeout, OPENAI_SPEECH_TIMEOUT_MS, ); - request.signal.addEventListener("abort", abortForRequest, { once: true }); const cleanup = () => { globalThis.clearTimeout(timeout); - request.signal.removeEventListener("abort", abortForRequest); + removeRequestAbortListener(); }; try { + abortController.signal.throwIfAborted(); const upstreamResponse = await fetch(OPENAI_SPEECH_ENDPOINT, { body: JSON.stringify({ input: parsedBody.text, From 7db65320dcc8546d4ba3cba47b2c008a2a58cbd2 Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Tue, 1 Sep 2026 12:31:50 +0200 Subject: [PATCH 08/10] Show waiting after speech playback drains Move the voice status off playback while Brunch is still completing the active turn, without reopening the microphone early. Co-authored-by: Cursor --- .../voice-turn-controller.test.ts | 39 +++++++++++++++++++ .../voice-interview/voice-turn-controller.ts | 3 ++ 2 files changed, 42 insertions(+) diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts index e89bdebf380..74c86231364 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts @@ -810,6 +810,45 @@ describe("VoiceTurnController", () => { expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith(true); }); + test("shows waiting after speech drains while Brunch is still busy", async () => { + const harness = createHarness(); + let finishPlayback: (() => void) | undefined; + harness.playback.play.mockImplementationOnce(async (_segment, events) => { + events?.onPlaying?.(); + await new Promise((resolve) => { + finishPlayback = resolve; + }); + }); + await harness.controller.start(); + harness.emit({ + key: key(1, "answer"), + text: "A finalized answer", + type: "completed", + }); + await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledOnce()); + + harness.controller.updateChat({ + canonicalSegments: [ + canonicalSegment( + "canonical-speech:busy-response:text%3A0:fnv1a32:12345678", + ), + ], + status: "streaming", + }); + + expect(harness.controller.getSnapshot().phase).toBe("playing"); + finishPlayback?.(); + await vi.waitFor(() => + expect(harness.controller.getSnapshot().phase).toBe("waiting"), + ); + expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( + false, + ); + + updateChatStatus(harness.controller, "ready"); + expect(harness.controller.getSnapshot().phase).toBe("listening"); + }); + test("speaks finalized segments in order and reopens only after the queue drains", async () => { const harness = createHarness(); const playbackResolvers: Array<() => void> = []; diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts index 2ea4f7bc728..f3a27cc000a 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts @@ -541,6 +541,9 @@ export class VoiceTurnController { if (this.#awaitingChatCycle) { if (!this.#sawBusyChatStatus || this.#chatStatus !== "ready") { this.#session.setMicrophoneEnabled(false); + if (this.#snapshot.phase !== "waiting") { + this.#update({ phase: "waiting" }); + } return; } this.#awaitingChatCycle = false; From 241c0308f308900ba4978b7ac40b255314dea876 Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Tue, 1 Sep 2026 16:00:23 +0200 Subject: [PATCH 09/10] Fix voice turn status test calls Use the existing chat update helper after the controller API consolidation so the pending-transcript regression test exercises the current interface. Co-authored-by: Cursor --- .../main/app/voice-interview/voice-turn-controller.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts index 74c86231364..2b557ce2678 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts @@ -260,7 +260,7 @@ describe("VoiceTurnController", () => { test("preserves the first pending transcript while Brunch is busy", async () => { const harness = createHarness(); await harness.controller.start(); - harness.controller.updateChatStatus("streaming"); + updateChatStatus(harness.controller, "streaming"); harness.emit({ connectionEpoch: 1, @@ -283,7 +283,7 @@ describe("VoiceTurnController", () => { type: "completed", }); - harness.controller.updateChatStatus("ready"); + updateChatStatus(harness.controller, "ready"); await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledOnce()); expect(harness.submitText).toHaveBeenCalledWith( From b15adb5afbdd1e474283ab0118e34c115356a01b Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Tue, 1 Sep 2026 16:08:47 +0200 Subject: [PATCH 10/10] Preserve speech failures and active streams Keep recoverable speech failures latched across pending delivery races, and limit the upstream timeout to waiting for OpenAI's response so valid audio streams can finish. Co-authored-by: Cursor --- .../voice-turn-controller.test.ts | 85 +++++++++++++++++++ .../voice-interview/voice-turn-controller.ts | 9 +- .../src/server/voice/openai-speech.test.ts | 31 +++++++ .../src/server/voice/openai-speech.ts | 6 +- 4 files changed, 129 insertions(+), 2 deletions(-) diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts index 2b557ce2678..49158855858 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts @@ -978,6 +978,91 @@ describe("VoiceTurnController", () => { ); }); + test("does not flush a pending transcript after speech fails", async () => { + const harness = createHarness(); + harness.playback.play.mockRejectedValueOnce( + new Error( + "The response could not be spoken. Read the visible text instead.", + ), + ); + await harness.controller.start(); + updateChatStatus(harness.controller, "streaming"); + harness.emit({ + key: key(1, "pending-answer"), + text: "A pending finalized answer", + type: "completed", + }); + expect(harness.submitText).not.toHaveBeenCalled(); + + harness.controller.updateChat({ + canonicalSegments: [ + canonicalSegment( + "canonical-speech:pending-failed:text%3A0:fnv1a32:12345678", + ), + ], + status: "streaming", + }); + await vi.waitFor(() => + expect(harness.controller.getSnapshot().phase).toBe("recoverable-error"), + ); + + updateChatStatus(harness.controller, "ready"); + await Promise.resolve(); + + expect(harness.submitText).not.toHaveBeenCalled(); + expect(harness.controller.getSnapshot()).toMatchObject({ + errorMessage: + "The response could not be spoken. Read the visible text instead.", + phase: "recoverable-error", + }); + }); + + test("does not overwrite a speech failure when delivery rejects while Brunch is busy", async () => { + const harness = createHarness(); + let rejectDelivery: (() => void) | undefined; + const delivery = new Promise((_resolve, reject) => { + rejectDelivery = () => reject(new Error("Brunch became busy")); + }); + harness.submitText.mockImplementationOnce(() => delivery); + harness.playback.play.mockRejectedValueOnce( + new Error( + "The response could not be spoken. Read the visible text instead.", + ), + ); + await harness.controller.start(); + harness.emit({ + key: key(1, "in-flight-answer"), + text: "An in-flight finalized answer", + type: "completed", + }); + await vi.waitFor(() => expect(harness.submitText).toHaveBeenCalledOnce()); + + harness.controller.updateChat({ + canonicalSegments: [ + canonicalSegment( + "canonical-speech:in-flight-failed:text%3A0:fnv1a32:12345678", + ), + ], + status: "streaming", + }); + await vi.waitFor(() => + expect(harness.controller.getSnapshot().phase).toBe("recoverable-error"), + ); + + rejectDelivery?.(); + await expect(delivery).rejects.toThrow("Brunch became busy"); + await Promise.resolve(); + + expect(harness.controller.getSnapshot()).toMatchObject({ + errorMessage: + "The response could not be spoken. Read the visible text instead.", + phase: "recoverable-error", + }); + expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( + false, + ); + }); + test("cancels speech synchronously and rejects stale playback events when voice ends", async () => { const harness = createHarness(); let playbackEvents: { onPlaying?: () => void } | undefined; diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts index f3a27cc000a..b7ae5af42cf 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts @@ -231,6 +231,10 @@ export class VoiceTurnController { } } + if (this.#snapshot.phase === "recoverable-error") { + return; + } + if ( status === "ready" && !this.#awaitingChatCycle && @@ -310,7 +314,10 @@ export class VoiceTurnController { } this.#settleListeningIfReady(); } catch { - if (generation !== this.#generation) { + if ( + generation !== this.#generation || + this.#snapshot.phase === "recoverable-error" + ) { return; } if (!this.#isChatReady()) { diff --git a/apps/petrinaut-website/src/server/voice/openai-speech.test.ts b/apps/petrinaut-website/src/server/voice/openai-speech.test.ts index 2d19930d1c7..6b8fab97139 100644 --- a/apps/petrinaut-website/src/server/voice/openai-speech.test.ts +++ b/apps/petrinaut-website/src/server/voice/openai-speech.test.ts @@ -264,6 +264,37 @@ describe("OpenAI Speech handler", () => { ); }); + test("does not apply the response timeout to an active audio stream", async () => { + vi.useFakeTimers(); + const upstreamCancel = vi.fn(); + const fetch = vi.fn( + async () => + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2, 3])); + }, + cancel: upstreamCancel, + }), + { headers: { "content-type": "audio/mpeg" } }, + ), + ); + const handler = createOpenAISpeechHandler({ + environment: enabledEnvironment, + fetch, + }); + + const response = await handler(createRequest()); + const reader = response.body!.getReader(); + await reader.read(); + await vi.advanceTimersByTimeAsync(OPENAI_SPEECH_TIMEOUT_MS); + + expect(fetch.mock.calls[0]?.[1]?.signal?.aborted).toBe(false); + + await reader.cancel("playback stopped"); + expect(upstreamCancel).toHaveBeenCalledWith("playback stopped"); + }); + test("propagates browser disconnect while waiting for OpenAI", async () => { const requestAbortController = new AbortController(); const fetch = vi.fn( diff --git a/apps/petrinaut-website/src/server/voice/openai-speech.ts b/apps/petrinaut-website/src/server/voice/openai-speech.ts index 837e90b6de9..1f8639af452 100644 --- a/apps/petrinaut-website/src/server/voice/openai-speech.ts +++ b/apps/petrinaut-website/src/server/voice/openai-speech.ts @@ -207,8 +207,11 @@ export const createOpenAISpeechHandler = abortForTimeout, OPENAI_SPEECH_TIMEOUT_MS, ); - const cleanup = () => { + const clearSpeechTimeout = () => { globalThis.clearTimeout(timeout); + }; + const cleanup = () => { + clearSpeechTimeout(); removeRequestAbortListener(); }; @@ -244,6 +247,7 @@ export const createOpenAISpeechHandler = return response(SPEECH_ERROR_MESSAGE, 502); } + clearSpeechTimeout(); return response( proxyAudioStream(upstreamResponse.body, abortController, cleanup), 200,