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..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 @@ -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,8 @@ describe("VoiceTurnController", () => { const harness = createHarness(); const starting = harness.controller.start(); - harness.controller.updateChatStatus("streaming"); + updateChatStatus(harness.controller, "streaming"); + expect(harness.controller.getSnapshot().phase).toBe("connecting"); await starting; expect(harness.session.setMicrophoneEnabled).toHaveBeenLastCalledWith( @@ -71,24 +121,47 @@ 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"); }); + 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(); - 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 +217,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); }); @@ -155,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({ @@ -176,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( @@ -187,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, @@ -210,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( @@ -224,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(); @@ -240,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( @@ -305,8 +378,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 +445,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,11 +459,121 @@ 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"); }); + test("queues 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", + }); + + 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." }), + ); + }); + + 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("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(); @@ -524,8 +707,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 +723,376 @@ 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("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> = []; + 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("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("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; + 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..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 @@ -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,79 @@ 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); } + } + + if (this.#snapshot.phase === "recoverable-error") { 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.#generation; + this.#activeEpoch = null; + this.#activeItemId = null; + this.#activeKey = null; 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", - }); + void this.#session.disconnect(); + 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 === "listening" || + this.#snapshot.phase === "delivering" || + this.#snapshot.phase === "waiting") + ) { + this.#update({ phase: "waiting" }); + } + return; + } + this.#settleListeningIfReady(); } async #deliver(input: SubmitTextInput): Promise { @@ -259,10 +309,15 @@ export class VoiceTurnController { ) { return; } - this.#update({ phase: "waiting" }); - this.#reopenListeningIfReady(); + if (this.#snapshot.phase === "delivering") { + this.#update({ phase: "waiting" }); + } + this.#settleListeningIfReady(); } catch { - if (generation !== this.#generation) { + if ( + generation !== this.#generation || + this.#snapshot.phase === "recoverable-error" + ) { return; } if (!this.#isChatReady()) { @@ -299,6 +354,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 +417,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 +444,127 @@ 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 || + this.#snapshot.phase === "transcribing" ) { 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 + ) { + return; + } + + if (this.#snapshot.phase === "transcribing") { + this.#session.setMicrophoneEnabled(false); + return; + } + + if (this.#speechQueue.length > 0) { + this.#startSpeechQueueIfNeeded(); + return; + } + + 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; + this.#sawBusyChatStatus = false; + } + + if (!this.#isChatReady()) { + this.#session.setMicrophoneEnabled(false); + if (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..6b8fab97139 --- /dev/null +++ b/apps/petrinaut-website/src/server/voice/openai-speech.test.ts @@ -0,0 +1,351 @@ +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("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]); + 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("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( + (_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..1f8639af452 --- /dev/null +++ b/apps/petrinaut-website/src/server/voice/openai-speech.ts @@ -0,0 +1,263 @@ +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); + } + + 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 abortForTimeout = () => abortController.abort(timeoutError); + const timeout = globalThis.setTimeout( + abortForTimeout, + OPENAI_SPEECH_TIMEOUT_MS, + ); + const clearSpeechTimeout = () => { + globalThis.clearTimeout(timeout); + }; + const cleanup = () => { + clearSpeechTimeout(); + removeRequestAbortListener(); + }; + + try { + abortController.signal.throwIfAborted(); + 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); + } + + clearSpeechTimeout(); + 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