diff --git a/apps/petrinaut-website/README.md b/apps/petrinaut-website/README.md index 319df516d0a..8951b4136b6 100644 --- a/apps/petrinaut-website/README.md +++ b/apps/petrinaut-website/README.md @@ -61,16 +61,16 @@ 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 | 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. | +| 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 to create Realtime WebRTC calls. | +| `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. @@ -80,23 +80,33 @@ 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 real `VITE_BRUNCH_CHAT_ENDPOINT`, `PETRINAUT_OPENAI_VOICE_ENABLED=true`, and a dedicated `OPENAI_VOICE_API_KEY`. The browser sends its SDP offer to this app; -the server initializes an OpenAI transcription-only Realtime session and keeps -the provider key, model, language, and vocabulary policy private. The session -uses `gpt-live-transcribe`'s default server VAD because OpenAI's unified call -currently times out when explicit turn detection is included during setup. - -Only finalized transcripts enter the existing Petrinaut composer and Brunch AI -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 server initializes a trusted `gpt-realtime-2` audio-input/audio-output +session through OpenAI's unified Realtime call endpoint. The provider key, +model, instructions, tools, language and vocabulary policy stay server-side. +The session uses semantic VAD with low eagerness so natural thinking pauses are +less likely to end an answer early. + +Realtime is the disposable media plane: it carries continuous microphone and +remote audio, detects complete turns and handles barge-in. Brunch remains the +control plane and sole authority for interview questions, captures, state, +completion and durable history. The browser bridge accepts only the configured +`continue_interview` function, validates and serializes its arguments, rejects +duplicate or stale calls and submits the answer through Petrinaut's existing +composer path with pending-`brunch_ask` correlation. + +The bridge waits for the correlated Brunch turn before returning canonical +speech segments to Realtime. It then requests audio with tools disabled and +instructs Realtime to speak only those segments. Generated audio is not treated +as a verbatim record: canonical Brunch text remains visible and authoritative. +Partial input transcripts are display-only and never submit or disable capture. + +The microphone remains active while the interviewer speaks and while Brunch is +working. Speaking over assistant audio interrupts playback automatically; +WebRTC truncates provider-side unheard audio without changing Brunch history. +**Pause** temporarily turns off input, while **End interview**, failure and +reconnect release or replace all media and invalidate events from old +connections. There is no separate Speech API, MP3 playback queue, required +**Done speaking** action or manual **Interrupt and speak** action. The Brunch deployment must allow the website origin through its `BRUNCH_PETRINAUT_ORIGINS` setting. Starting voice input requests browser @@ -108,19 +118,18 @@ permission, microphone device, interrupted request, network, timeout, invalid response, and unavailable/disabled failures. Permission and device failures identify what to fix; network, timeout, and interrupted requests offer a reconnect; invalid responses include a diagnostic reference for an operator; -and unavailable voice leaves the text composer as the fallback. Speech failures +and unavailable voice leaves the text composer as the fallback. Voice failures always leave the canonical response visible to read. -Realtime connection and Speech requests carry a random `x-request-id` through -the browser and server route, and the existing Brunch transport sends the same -header on each chat request so Brunch's privacy-safe request inspection can -correlate that boundary. Browser and server diagnostics report operation, -stage, outcome, duration, request ID, and—where applicable—status or sanitized -error code. Voice responses also expose privacy-safe `Server-Timing` metrics. -These diagnostics never record audio, SDP, transcript or prompt contents, -canonical speech text, credentials, or provider response bodies. This -controlled-preview evidence does not enable production: production remains -unconditionally disabled by the server policy. +Realtime connection, transcription and canonical speech timings use random +request IDs, and the existing Brunch transport provides its own request +correlation. Browser and server diagnostics report only operation, stage, +outcome, duration, request ID and—where applicable—status or a sanitized error +code. Voice responses also expose privacy-safe `Server-Timing` metrics. These +diagnostics never record audio, SDP, transcript or prompt contents, canonical +speech text, credentials or provider response bodies. This controlled-preview +evidence does not enable production: production remains unconditionally +disabled by the server policy. ## Testing the API against the built output diff --git a/apps/petrinaut-website/api/voice/speech.ts b/apps/petrinaut-website/api/voice/speech.ts deleted file mode 100644 index 62e4a580e18..00000000000 --- a/apps/petrinaut-website/api/voice/speech.ts +++ /dev/null @@ -1,14 +0,0 @@ -import { createOpenAISpeechHandler } from "../../src/server/voice/openai-speech"; -import { reportVoiceDiagnostic } from "../../src/voice-diagnostics"; - -declare const process: { - env: Record; -}; - -export default { - fetch: createOpenAISpeechHandler({ - environment: process.env, - fetch: globalThis.fetch.bind(globalThis), - reportDiagnostic: reportVoiceDiagnostic, - }), -}; diff --git a/apps/petrinaut-website/src/main/app/voice-interview/openai-realtime-session.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/openai-realtime-session.test.ts index 926d553cade..2e486daae5d 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/openai-realtime-session.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/openai-realtime-session.test.ts @@ -5,20 +5,21 @@ import { type OpenAIRealtimeSessionEvent, } from "./openai-realtime-session"; +import type { CanonicalSpeechSegment } from "./canonical-speech"; + class FakeDataChannel extends EventTarget { public readyState: RTCDataChannelState = "connecting"; - public readonly close = vi.fn(() => { this.readyState = "closed"; }); public readonly send = vi.fn(); - public open() { + public open(): void { this.readyState = "open"; this.dispatchEvent(new Event("open")); } - public receive(payload: unknown) { + public receive(payload: unknown): void { const event = new Event("message"); Object.defineProperty(event, "data", { value: typeof payload === "string" ? payload : JSON.stringify(payload), @@ -27,37 +28,36 @@ class FakeDataChannel extends EventTarget { } } +const canonicalSegment = ( + id: string, + text: string, +): CanonicalSpeechSegment => ({ + contentHash: "fnv1a32:12345678", + id, + messageId: `message-${id}`, + partId: id, + source: "brunch-ask", + text, +}); + const createHarness = ({ connectionTimeoutMs = 15_000, - createAudioContext, }: { readonly connectionTimeoutMs?: number; - readonly createAudioContext?: () => AudioContext; } = {}) => { let requestNumber = 0; - const animationFrames: FrameRequestCallback[] = []; - const analyser = { - fftSize: 0, - getByteTimeDomainData: vi.fn((data: Uint8Array) => { - data.fill(160); - }), - }; - const mediaSource = { connect: vi.fn() }; - const tracks: Array<{ enabled: boolean; stop: ReturnType }> = - []; - const trackEnabledWhenMeterConnected: boolean[] = []; - const audioContext = { - close: vi.fn(async () => undefined), - createAnalyser: vi.fn(() => analyser), - createMediaStreamSource: vi.fn(() => { - trackEnabledWhenMeterConnected.push(tracks.at(-1)?.enabled ?? true); - return mediaSource; - }), - resume: vi.fn(async () => undefined), - state: "suspended" as AudioContextState, - }; - const cancelAnimationFrame = vi.fn(); const channels: FakeDataChannel[] = []; + const localTracks: Array<{ + enabled: boolean; + kind: string; + stop: ReturnType; + }> = []; + const remoteAudios: Array<{ + autoplay: boolean; + pause: ReturnType; + play: ReturnType; + srcObject: MediaStream | null; + }> = []; const peers: Array<{ addTrack: ReturnType; close: ReturnType; @@ -65,87 +65,117 @@ const createHarness = ({ createDataChannel: ReturnType; createOffer: ReturnType; onconnectionstatechange: (() => void) | null; + ontrack: ((event: RTCTrackEvent) => void) | null; setLocalDescription: ReturnType; setRemoteDescription: ReturnType Promise>>; }> = []; - const fetch = vi.fn( - async () => - new Response("v=0\r\no=OpenAI answer", { - headers: { "content-type": "application/sdp" }, - }), - ); const getUserMedia = vi.fn(async () => { - const track = { enabled: true, stop: vi.fn() }; - tracks.push(track); + const track = { enabled: true, kind: "audio", stop: vi.fn() }; + localTracks.push(track); return { getAudioTracks: () => [track], getTracks: () => [track], } as unknown as MediaStream; }); - const reportDiagnostic = vi.fn(); - const createPeerConnection = () => { - const channel = new FakeDataChannel(); - channels.push(channel); - const peer = { - addTrack: vi.fn(), - close: vi.fn(), - connectionState: "new" as RTCPeerConnectionState, - createDataChannel: vi.fn(() => channel), - createOffer: vi.fn(async () => ({ - type: "offer", - sdp: "v=0\r\no=browser offer", - })), - onconnectionstatechange: null as (() => void) | null, - setLocalDescription: vi.fn(async () => undefined), - setRemoteDescription: vi.fn(async () => { - channel.open(); + const fetch = vi.fn( + async () => + new Response("v=0\r\no=OpenAI answer", { + headers: { "content-type": "application/sdp" }, }), - }; - peers.push(peer); - return peer as unknown as RTCPeerConnection; - }; + ); + const reportDiagnostic = vi.fn(); const session = new OpenAIRealtimeSession({ - cancelAnimationFrame, + cancelAnimationFrame: vi.fn(), connectionTimeoutMs, - createAudioContext: - createAudioContext ?? (() => audioContext as unknown as AudioContext), + createAudioContext: () => + ({ + close: vi.fn(async () => undefined), + createAnalyser: vi.fn(() => ({ + fftSize: 0, + getByteTimeDomainData: vi.fn((samples: Uint8Array) => + samples.fill(140), + ), + })), + createMediaStreamSource: vi.fn(() => ({ connect: vi.fn() })), + resume: vi.fn(async () => undefined), + state: "running", + }) as unknown as AudioContext, + createPeerConnection: () => { + const channel = new FakeDataChannel(); + channels.push(channel); + const peer = { + addTrack: vi.fn(), + close: vi.fn(), + connectionState: "new" as RTCPeerConnectionState, + createDataChannel: vi.fn(() => channel), + createOffer: vi.fn(async () => ({ + sdp: "v=0\r\no=browser offer", + type: "offer" as RTCSdpType, + })), + onconnectionstatechange: null as (() => void) | null, + ontrack: null as ((event: RTCTrackEvent) => void) | null, + setLocalDescription: vi.fn(async () => undefined), + setRemoteDescription: vi.fn(async () => channel.open()), + }; + peers.push(peer); + return peer as unknown as RTCPeerConnection; + }, + createRemoteAudio: () => { + const audio = { + autoplay: false, + pause: vi.fn(), + play: vi.fn(async () => undefined), + srcObject: null as MediaStream | null, + }; + remoteAudios.push(audio); + return audio; + }, createRequestId: () => `voice-request-${++requestNumber}`, - createPeerConnection, fetch, getUserMedia, now: () => 100, reportDiagnostic, - requestAnimationFrame: (callback) => { - animationFrames.push(callback); - return animationFrames.length; - }, + requestAnimationFrame: vi.fn(() => 1), }); const events: OpenAIRealtimeSessionEvent[] = []; session.subscribe((event) => events.push(event)); return { - analyser, - animationFrames, - audioContext, - cancelAnimationFrame, channels, events, fetch, getUserMedia, + localTracks, peers, + remoteAudios, reportDiagnostic, session, - trackEnabledWhenMeterConnected, - tracks, }; }; +const sentEvents = (channel: FakeDataChannel): Record[] => + channel.send.mock.calls.map(([payload]) => JSON.parse(payload as string)); + +const authorizeLatestSpeechResponse = ( + channel: FakeDataChannel, + responseId: string, +): void => { + const responseCreate = sentEvents(channel).findLast( + ({ type }) => type === "response.create", + )!; + const response = responseCreate.response as Record; + channel.receive({ + type: "response.created", + response: { id: responseId, metadata: response.metadata }, + }); +}; + describe("OpenAIRealtimeSession", () => { afterEach(() => { vi.useRealTimers(); }); - test("negotiates through the app server and starts with the microphone closed", async () => { + test("negotiates duplex WebRTC, attaches remote audio, and cleans all media", async () => { const harness = createHarness(); await expect(harness.session.connect()).resolves.toBe(1); @@ -157,360 +187,704 @@ describe("OpenAIRealtimeSession", () => { noiseSuppression: true, }, }); - expect(harness.tracks[0]!.enabled).toBe(false); - expect(harness.trackEnabledWhenMeterConnected).toEqual([false]); + expect(harness.localTracks[0]!.enabled).toBe(false); expect(harness.fetch).toHaveBeenCalledWith( "/api/voice/realtime-call", expect.objectContaining({ body: "v=0\r\no=browser offer", - headers: { - "content-type": "application/sdp", - "x-request-id": "voice-request-1", - }, method: "POST", }), ); expect(JSON.stringify(harness.fetch.mock.calls)).not.toContain( "authorization", ); - expect(harness.peers[0]!.setRemoteDescription).toHaveBeenCalledWith({ - sdp: "v=0\r\no=OpenAI answer", - type: "answer", - }); - harness.session.setMicrophoneEnabled(true); - expect(harness.tracks[0]!.enabled).toBe(true); - expect(harness.reportDiagnostic).toHaveBeenCalledWith({ - durationMs: 0, - operation: "connection", - outcome: "success", - requestId: "voice-request-1", - stage: "browser", + const remoteTrack = { kind: "audio", stop: vi.fn() }; + const remoteStream = { + getTracks: () => [remoteTrack], + } as unknown as MediaStream; + harness.peers[0]!.ontrack?.({ + streams: [remoteStream], + track: remoteTrack, + } as unknown as RTCTrackEvent); + expect(harness.remoteAudios[0]).toMatchObject({ + autoplay: true, + srcObject: remoteStream, }); + expect(harness.remoteAudios[0]!.play).toHaveBeenCalledOnce(); + + await harness.session.disconnect(); + expect(harness.remoteAudios[0]!.pause).toHaveBeenCalledOnce(); + expect(harness.remoteAudios[0]!.srcObject).toBeNull(); + expect(remoteTrack.stop).toHaveBeenCalledOnce(); + expect(harness.localTracks[0]!.stop).toHaveBeenCalledOnce(); + expect(harness.peers[0]!.close).toHaveBeenCalledOnce(); }); - test("reports real input level only while the microphone track is enabled", async () => { + test("keeps the microphone active through playback and reports automatic interruption", async () => { const harness = createHarness(); await harness.session.connect(); - - expect(harness.animationFrames).toHaveLength(0); harness.session.setMicrophoneEnabled(true); - expect(harness.animationFrames).toHaveLength(1); - harness.animationFrames.shift()?.(0); + harness.session.speakCanonical([ + canonicalSegment("ask-1", "What happens next?"), + ]); + const channel = harness.channels[0]!; + authorizeLatestSpeechResponse(channel, "response-canonical"); - expect(harness.events.at(-1)).toMatchObject({ - type: "microphone-level", + channel.receive({ + event_id: "event-1", + response_id: "response-canonical", + type: "output_audio_buffer.started", + }); + channel.receive({ + audio_start_ms: 120, + event_id: "event-2", + item_id: "item-user", + type: "input_audio_buffer.speech_started", }); - expect((harness.events.at(-1) as { level: number }).level).toBeGreaterThan( - 0, + + expect(harness.localTracks[0]!.enabled).toBe(true); + expect(harness.events).toEqual( + expect.arrayContaining([ + { + connectionEpoch: 1, + responseId: "response-canonical", + type: "output-started", + }, + { + connectionEpoch: 1, + itemId: "item-user", + type: "input-speech-started", + }, + { + connectionEpoch: 1, + responseId: "response-canonical", + type: "output-interrupted", + }, + ]), ); + }); - harness.session.setMicrophoneEnabled(false); - expect(harness.tracks[0]!.enabled).toBe(false); - expect(harness.cancelAnimationFrame).toHaveBeenCalled(); - expect(harness.events.at(-1)).toEqual({ - level: 0, - type: "microphone-level", + test("parses streamed tool arguments and the completed GA response output", async () => { + const harness = createHarness(); + await harness.session.connect(); + const channel = harness.channels[0]!; + + channel.receive({ + arguments: '{"answer":"Ignored"}', + call_id: "call-ignored", + item_id: "item-ignored", + output_index: 0, + response_id: "response-tool", + type: "response.function_call_arguments.done", + }); + channel.receive({ + call_id: "call-1", + delta: '{"answer":"Approved"}', + item_id: "item-function", + output_index: 0, + response_id: "response-tool", + type: "response.function_call_arguments.delta", + }); + channel.receive({ + response: { + id: "response-tool", + output: [ + { + arguments: '{"answer":"Approved"}', + call_id: "call-1", + id: "item-function", + name: "continue_interview", + type: "function_call", + }, + ], + status: "completed", + }, + type: "response.done", + }); + + expect(harness.events).toEqual([ + { + callId: "call-1", + connectionEpoch: 1, + delta: '{"answer":"Approved"}', + itemId: "item-function", + responseId: "response-tool", + type: "tool-arguments-delta", + }, + { + arguments: '{"answer":"Approved"}', + callId: "call-1", + connectionEpoch: 1, + itemId: "item-function", + name: "continue_interview", + responseId: "response-tool", + type: "tool-arguments-done", + }, + { + connectionEpoch: 1, + responseId: "response-tool", + status: "completed", + type: "response-terminal", + }, + ]); + + harness.session.completeFunctionCall("call-1", [ + canonicalSegment("ask-2", "Who acts next?"), + ]); + const [functionOutput, responseCreate] = sentEvents(channel).slice(-2); + expect(functionOutput).toEqual({ + type: "conversation.item.create", + item: { + type: "function_call_output", + call_id: "call-1", + output: JSON.stringify({ response_text: ["Who acts next?"] }), + }, + }); + expect(responseCreate).toMatchObject({ + type: "response.create", + response: { + instructions: + "Speak only the response_text strings supplied by Petrinaut, in array order and verbatim. Deliver them as a warm, calm, curious, confident, concise, and professionally neutral expert interviewer, at a measured conversational pace with natural emphasis. Never sound robotic, fawning, rushed, overenthusiastic, or patronizing. Do not add, remove, paraphrase, acknowledge, or explain anything.", + output_modalities: ["audio"], + parallel_tool_calls: false, + tool_choice: "none", + tools: [], + }, }); }); - test("quantizes input levels and skips unchanged meter frames", async () => { + test("queues canonical speech behind an active Realtime response", async () => { const harness = createHarness(); - harness.analyser.getByteTimeDomainData.mockImplementation( - (data: Uint8Array) => { - data.fill(200); + await harness.session.connect(); + const channel = harness.channels[0]!; + channel.receive({ + response: { id: "response-active" }, + type: "response.created", + }); + + harness.session.speakCanonical([ + canonicalSegment("question", "Canonical question"), + ]); + + expect(sentEvents(channel)).toEqual([]); + channel.receive({ + response: { + id: "response-active", + output: [], + status: "completed", }, - ); + type: "response.done", + }); + + expect(sentEvents(channel)).toHaveLength(1); + expect(sentEvents(channel)[0]).toMatchObject({ + type: "response.create", + response: { + metadata: { petrinaut_kind: "canonical-speech" }, + }, + }); + expect(harness.events).toContainEqual({ + connectionEpoch: 1, + responseId: "response-active", + status: "completed", + type: "response-terminal", + }); + }); + + test("cancels canonical speech before the response starts", async () => { + const harness = createHarness(); await harness.session.connect(); - harness.session.setMicrophoneEnabled(true); + const channel = harness.channels[0]!; + harness.session.speakCanonical([ + canonicalSegment("question", "Canonical question"), + ]); + const responseCreate = sentEvents(channel)[0]!; - const levels = () => - harness.events - .filter((event) => event.type === "microphone-level") - .map((event) => (event as { level: number }).level); + harness.session.cancelOutput(); - harness.animationFrames.shift()?.(0); - harness.animationFrames.shift()?.(0); - harness.animationFrames.shift()?.(0); + expect( + sentEvents(channel).filter(({ type }) => type === "response.cancel"), + ).toEqual([]); - expect(levels()).toEqual([0.56]); + channel.receive({ + response: { + id: "response-canonical", + metadata: (responseCreate.response as Record).metadata, + }, + type: "response.created", + }); - harness.analyser.getByteTimeDomainData.mockImplementation( - (data: Uint8Array) => { - data.fill(160); + expect(sentEvents(channel).slice(-2)).toEqual([ + expect.objectContaining({ + response_id: "response-canonical", + type: "response.cancel", + }), + { type: "output_audio_buffer.clear" }, + ]); + + channel.receive({ + response_id: "response-canonical", + type: "output_audio_buffer.started", + }); + channel.receive({ + response: { + id: "response-canonical", + output: [], + status: "cancelled", }, - ); - harness.animationFrames.shift()?.(0); + type: "response.done", + }); - expect(levels()).toEqual([0.56, 0.25]); + expect(harness.events).not.toContainEqual( + expect.objectContaining({ type: "output-started" }), + ); + expect(harness.events).not.toContainEqual( + expect.objectContaining({ type: "error" }), + ); + expect(harness.localTracks[0]!.stop).not.toHaveBeenCalled(); }); - test("resumes a suspended input meter before waiting for microphone access", async () => { + test("retries a correlated canonical response after the active response ends", async () => { const harness = createHarness(); - const track = { enabled: true, stop: vi.fn() }; - const stream = { - getAudioTracks: () => [track], - getTracks: () => [track], - } as unknown as MediaStream; - let resolveMedia: ((mediaStream: MediaStream) => void) | undefined; - harness.getUserMedia.mockImplementationOnce( - () => - new Promise((resolve) => { - resolveMedia = resolve; - }), + await harness.session.connect(); + const channel = harness.channels[0]!; + harness.session.speakCanonical([ + canonicalSegment("question", "Canonical question"), + ]); + const firstCreate = sentEvents(channel)[0]!; + + channel.receive({ + error: { + code: "conversation_already_has_active_response", + event_id: firstCreate.event_id, + message: "private provider detail", + type: "invalid_request_error", + }, + type: "error", + }); + channel.receive({ + response: { id: "response-active" }, + type: "response.created", + }); + channel.receive({ + response: { + id: "response-active", + output: [], + status: "completed", + }, + type: "response.done", + }); + + const responseCreates = sentEvents(channel).filter( + ({ type }) => type === "response.create", ); + expect(responseCreates).toHaveLength(2); + expect(responseCreates[1]?.response).toEqual(firstCreate.response); + expect(harness.events).not.toContainEqual( + expect.objectContaining({ type: "error" }), + ); + expect(harness.localTracks[0]!.stop).not.toHaveBeenCalled(); + }); - const connection = harness.session.connect(); - const resumeCallsBeforeMedia = - harness.audioContext.resume.mock.calls.length; - const sourceCallsBeforeMedia = - harness.audioContext.createMediaStreamSource.mock.calls.length; - resolveMedia?.(stream); - await connection; + test("retries canonical speech when the active response ends before the correlated error arrives", async () => { + const harness = createHarness(); + await harness.session.connect(); + const channel = harness.channels[0]!; + harness.session.speakCanonical([ + canonicalSegment("question", "Canonical question"), + ]); + const firstCreate = sentEvents(channel)[0]!; - expect(resumeCallsBeforeMedia).toBe(1); - expect(sourceCallsBeforeMedia).toBe(0); - expect(harness.audioContext.createMediaStreamSource).toHaveBeenCalledWith( - stream, + channel.receive({ + response: { id: "response-active" }, + type: "response.created", + }); + channel.receive({ + response: { + id: "response-active", + output: [], + status: "completed", + }, + type: "response.done", + }); + channel.receive({ + error: { + code: "conversation_already_has_active_response", + event_id: firstCreate.event_id, + message: "private provider detail", + type: "invalid_request_error", + }, + type: "error", + }); + + const responseCreates = sentEvents(channel).filter( + ({ type }) => type === "response.create", + ); + expect(responseCreates).toHaveLength(2); + expect(responseCreates[1]?.response).toEqual(firstCreate.response); + expect(harness.events).not.toContainEqual( + expect.objectContaining({ type: "error" }), ); }); - test("connects without metering when audio context construction throws", async () => { - const harness = createHarness({ - createAudioContext: () => { - throw new Error("AudioContext unavailable"); + test("ignores a correlated cancel for an already-finished response", async () => { + const harness = createHarness(); + await harness.session.connect(); + const channel = harness.channels[0]!; + harness.session.speakCanonical([ + canonicalSegment("question", "Canonical question"), + ]); + const responseCreate = sentEvents(channel)[0]!; + channel.receive({ + response: { + id: "response-canonical", + metadata: (responseCreate.response as Record).metadata, }, + type: "response.created", + }); + channel.receive({ + response_id: "response-canonical", + type: "output_audio_buffer.started", }); - await expect(harness.session.connect()).resolves.toBe(1); - harness.session.setMicrophoneEnabled(true); + harness.session.cancelOutput(); + const cancelEvent = sentEvents(channel).findLast( + ({ type }) => type === "response.cancel", + )!; + expect(cancelEvent).toMatchObject({ + response_id: "response-canonical", + type: "response.cancel", + }); + channel.receive({ + error: { + code: "response_cancel_not_active", + event_id: cancelEvent.event_id, + message: "private provider detail", + type: "invalid_request_error", + }, + type: "error", + }); - expect(harness.fetch).toHaveBeenCalledOnce(); - expect(harness.peers[0]!.addTrack).toHaveBeenCalledOnce(); - expect(harness.tracks[0]!.enabled).toBe(true); - expect(harness.animationFrames).toHaveLength(0); - expect(harness.events).toEqual([]); + expect(harness.events).not.toContainEqual( + expect.objectContaining({ type: "error" }), + ); + expect(harness.localTracks[0]!.stop).not.toHaveBeenCalled(); }); - test("connects without metering when meter initialization throws", async () => { + test("ignores a late correlated cancel error after response completion", async () => { const harness = createHarness(); - harness.audioContext.createMediaStreamSource.mockImplementationOnce(() => { - throw new Error("Media stream source unavailable"); + await harness.session.connect(); + const channel = harness.channels[0]!; + harness.session.speakCanonical([ + canonicalSegment("question", "Canonical question"), + ]); + const responseCreate = sentEvents(channel)[0]!; + channel.receive({ + response: { + id: "response-canonical", + metadata: (responseCreate.response as Record).metadata, + }, + type: "response.created", + }); + channel.receive({ + response_id: "response-canonical", + type: "output_audio_buffer.started", }); - await expect(harness.session.connect()).resolves.toBe(1); - harness.session.setMicrophoneEnabled(true); + harness.session.cancelOutput(); + const cancelEvent = sentEvents(channel).findLast( + ({ type }) => type === "response.cancel", + )!; + channel.receive({ + response: { + id: "response-canonical", + output: [], + status: "cancelled", + }, + type: "response.done", + }); + channel.receive({ + error: { + code: "response_cancel_not_active", + event_id: cancelEvent.event_id, + message: "private provider detail", + type: "invalid_request_error", + }, + type: "error", + }); - expect(harness.fetch).toHaveBeenCalledOnce(); - expect(harness.peers[0]!.addTrack).toHaveBeenCalledOnce(); - expect(harness.tracks[0]!.enabled).toBe(true); - expect(harness.audioContext.close).toHaveBeenCalledOnce(); - expect(harness.animationFrames).toHaveLength(0); - expect(harness.events).toEqual([]); + expect(harness.events).not.toContainEqual( + expect.objectContaining({ type: "error" }), + ); + expect(harness.localTracks[0]!.stop).not.toHaveBeenCalled(); }); - test("emits only strict input transcription events with stable source identity", async () => { + test("rejects malformed completed function calls without exposing provider data", async () => { const harness = createHarness(); await harness.session.connect(); - harness.session.setMicrophoneEnabled(true); - harness.channels[0]!.receive("not-json"); - harness.channels[0]!.receive({ type: "response.created" }); harness.channels[0]!.receive({ - type: "conversation.item.input_audio_transcription.delta", - item_id: "item-a", - content_index: "0", - delta: "ignored", + response: { + id: "response-tool", + output: [ + { + arguments: { private: "provider arguments" }, + call_id: "call-1", + id: "item-function", + name: "continue_interview", + type: "function_call", + }, + ], + status: "completed", + }, + type: "response.done", }); - harness.channels[0]!.receive({ - type: "conversation.item.input_audio_transcription.delta", - item_id: "item-a", - content_index: 0, - delta: "The support lead", + + expect(harness.events.at(-1)).toMatchObject({ + code: "invalid-response", + type: "error", }); + expect(JSON.stringify(harness.events)).not.toContain("provider arguments"); + expect(harness.localTracks[0]!.stop).toHaveBeenCalledOnce(); + }); + + test("rejects multiple function calls before emitting either one", async () => { + const harness = createHarness(); + await harness.session.connect(); + harness.channels[0]!.receive({ - type: "conversation.item.input_audio_transcription.completed", - item_id: "item-a", - content_index: 0, - transcript: "The support lead triages it.", + response: { + id: "response-tool", + output: ["call-1", "call-2"].map((callId) => ({ + arguments: '{"answer":"Approved"}', + call_id: callId, + id: `item-${callId}`, + name: "continue_interview", + status: "completed", + type: "function_call", + })), + status: "completed", + }, + type: "response.done", }); expect(harness.events).toEqual([ - { - key: { connectionEpoch: 1, contentIndex: 0, itemId: "item-a" }, - text: "The support lead", - type: "partial", - }, - { - key: { connectionEpoch: 1, contentIndex: 0, itemId: "item-a" }, - text: "The support lead triages it.", - type: "completed", - }, + expect.objectContaining({ code: "invalid-response", type: "error" }), ]); - expect(harness.tracks[0]!.enabled).toBe(false); - expect(harness.reportDiagnostic).toHaveBeenLastCalledWith({ - durationMs: 0, - operation: "transcription", - outcome: "success", - requestId: "voice-request-2", - stage: "browser", - }); }); - test("surfaces failed input transcription as a recoverable error", async () => { + test("rejects a tool call from a canonical speech response", async () => { const harness = createHarness(); await harness.session.connect(); + harness.session.speakCanonical([ + canonicalSegment("question", "Canonical question"), + ]); + const responseCreate = sentEvents(harness.channels[0]!).at(-1)!; + const response = responseCreate.response as Record; harness.channels[0]!.receive({ - type: "conversation.item.input_audio_transcription.failed", - item_id: "item-a", - content_index: 0, - error: { message: "private provider diagnostic" }, + response: { + id: "response-canonical", + metadata: response.metadata, + }, + type: "response.created", + }); + harness.channels[0]!.receive({ + response: { + id: "response-canonical", + output: [ + { + arguments: '{"answer":"Invented overlap"}', + call_id: "call-not-allowed", + id: "function-item-1", + name: "continue_interview", + type: "function_call", + }, + ], + status: "completed", + }, + type: "response.done", }); expect(harness.events).toEqual([ - { + expect.objectContaining({ code: "invalid-response", - message: - "The transcription service returned an invalid response. Try again; if it continues, give the diagnostic reference to an operator.", - requestId: "voice-request-1", type: "error", - }, + }), ]); - expect(JSON.stringify(harness.events)).not.toContain( - "private provider diagnostic", - ); - expect(harness.tracks[0]!.stop).toHaveBeenCalledOnce(); - expect(harness.channels[0]!.close).toHaveBeenCalledOnce(); expect(harness.peers[0]!.close).toHaveBeenCalledOnce(); - expect(harness.reportDiagnostic).toHaveBeenLastCalledWith({ - durationMs: 0, - errorCode: "invalid-response", - operation: "transcription", - outcome: "failure", - requestId: "voice-request-1", - stage: "browser", + }); + + test("treats transcripts as display-only and never closes capture", async () => { + const harness = createHarness(); + await harness.session.connect(); + harness.session.setMicrophoneEnabled(true); + const channel = harness.channels[0]!; + + channel.receive({ + content_index: 0, + delta: "The supervisor", + item_id: "item-user", + type: "conversation.item.input_audio_transcription.delta", + }); + channel.receive({ + content_index: 0, + item_id: "item-user", + transcript: "The supervisor approves it.", + type: "conversation.item.input_audio_transcription.completed", }); + + expect(harness.events).toEqual([ + { + key: { connectionEpoch: 1, contentIndex: 0, itemId: "item-user" }, + text: "The supervisor", + type: "partial", + }, + { + key: { connectionEpoch: 1, contentIndex: 0, itemId: "item-user" }, + text: "The supervisor approves it.", + type: "completed", + }, + ]); + expect(harness.localTracks[0]!.enabled).toBe(true); }); - test("surfaces OpenAI data-channel errors without exposing diagnostics", async () => { + test("keeps the duplex session alive when optional input transcription fails", async () => { const harness = createHarness(); await harness.session.connect(); + harness.session.setMicrophoneEnabled(true); harness.channels[0]!.receive({ - type: "error", - error: { message: "private provider diagnostic" }, + content_index: 0, + error: { message: "private provider detail" }, + item_id: "item-user", + type: "conversation.item.input_audio_transcription.failed", }); expect(harness.events).toEqual([ { - code: "invalid-response", - message: - "The transcription service returned an invalid response. Try again; if it continues, give the diagnostic reference to an operator.", - requestId: "voice-request-1", - type: "error", + key: { connectionEpoch: 1, contentIndex: 0, itemId: "item-user" }, + type: "transcription-failed", }, ]); + expect(harness.localTracks[0]!.enabled).toBe(true); expect(JSON.stringify(harness.events)).not.toContain( - "private provider diagnostic", + "private provider detail", ); - expect(harness.tracks[0]!.stop).toHaveBeenCalledOnce(); - expect(harness.channels[0]!.close).toHaveBeenCalledOnce(); - expect(harness.peers[0]!.close).toHaveBeenCalledOnce(); }); - test("closes the microphone when VAD commits an input item", async () => { + test("fails closed if Realtime tries to play non-canonical audio", async () => { const harness = createHarness(); await harness.session.connect(); - harness.session.setMicrophoneEnabled(true); harness.channels[0]!.receive({ - type: "input_audio_buffer.committed", - item_id: "item-a", + event_id: "event-unauthorized", + response_id: "response-unauthorized", + type: "output_audio_buffer.started", }); - expect(harness.tracks[0]!.enabled).toBe(false); - expect(harness.events).toEqual([ - { connectionEpoch: 1, itemId: "item-a", type: "input-committed" }, - ]); + const [cancelEvent, clearEvent] = sentEvents(harness.channels[0]!); + expect(cancelEvent).toMatchObject({ + response_id: "response-unauthorized", + type: "response.cancel", + }); + expect(typeof cancelEvent?.event_id).toBe("string"); + expect(clearEvent).toEqual({ type: "output_audio_buffer.clear" }); + expect(harness.events.at(-1)).toMatchObject({ + code: "invalid-response", + type: "error", + }); + expect(harness.localTracks[0]!.stop).toHaveBeenCalledOnce(); }); - test("disposes all WebRTC resources and rejects stale events after reconnect", async () => { + test("rejects stale events and cleans the first epoch during reconnect", async () => { const harness = createHarness(); await harness.session.connect(); const firstChannel = harness.channels[0]!; await expect(harness.session.connect()).resolves.toBe(2); - - expect(harness.tracks[0]!.stop).toHaveBeenCalledOnce(); - expect(firstChannel.close).toHaveBeenCalledOnce(); - expect(harness.peers[0]!.close).toHaveBeenCalledOnce(); firstChannel.receive({ - type: "conversation.item.input_audio_transcription.completed", - item_id: "stale-item", - content_index: 0, - transcript: "Stale transcript", + arguments: '{"answer":"Stale"}', + call_id: "call-stale", + item_id: "item-stale", + name: "continue_interview", + output_index: 0, + response_id: "response-stale", + type: "response.function_call_arguments.done", }); - expect(harness.events).toEqual([]); - await harness.session.disconnect(); - expect(harness.tracks[1]!.stop).toHaveBeenCalledOnce(); - expect(harness.channels[1]!.close).toHaveBeenCalledOnce(); - expect(harness.peers[1]!.close).toHaveBeenCalledOnce(); + expect(harness.events).toEqual([]); + expect(harness.localTracks[0]!.stop).toHaveBeenCalledOnce(); + expect(harness.peers[0]!.close).toHaveBeenCalledOnce(); }); - test("reports microphone permission denial without leaking browser diagnostics", async () => { + test("sanitizes provider errors and releases media", async () => { const harness = createHarness(); - harness.getUserMedia.mockRejectedValueOnce( - new DOMException("private browser detail", "NotAllowedError"), - ); + await harness.session.connect(); - await expect(harness.session.connect()).rejects.toThrow( - "Allow microphone access in your browser settings, then reconnect voice input.", - ); - expect(harness.reportDiagnostic).toHaveBeenCalledWith({ - durationMs: 0, - errorCode: "microphone-permission", - operation: "connection", - outcome: "failure", - requestId: "voice-request-1", - stage: "browser", + harness.channels[0]!.receive({ + error: { message: "private provider body" }, + type: "error", + }); + + expect(harness.events.at(-1)).toMatchObject({ + code: "invalid-response", + type: "error", }); + expect(JSON.stringify(harness.events)).not.toContain( + "private provider body", + ); + expect(harness.localTracks[0]!.stop).toHaveBeenCalledOnce(); }); - test("distinguishes microphone device failures from permission denial", async () => { - const harness = createHarness(); - harness.getUserMedia.mockRejectedValueOnce( - new DOMException("private device detail", "NotFoundError"), + test("classifies microphone permission and network failures", async () => { + const permissionFailure = createHarness(); + permissionFailure.getUserMedia.mockRejectedValueOnce( + new DOMException("private browser detail", "NotAllowedError"), ); - await expect(harness.session.connect()).rejects.toMatchObject({ - code: "microphone-device", - message: - "No usable microphone was found. Connect or select one, then reconnect voice input.", + await expect(permissionFailure.session.connect()).rejects.toMatchObject({ + code: "microphone-permission", requestId: "voice-request-1", }); - expect(JSON.stringify(harness.reportDiagnostic.mock.calls)).not.toContain( - "private device detail", - ); - }); + expect( + JSON.stringify(permissionFailure.reportDiagnostic.mock.calls), + ).not.toContain("private browser detail"); - test("classifies browser network failures without exposing thrown details", async () => { - const harness = createHarness(); - harness.fetch.mockRejectedValueOnce( - new Error("private SDP and credential diagnostics"), + const networkFailure = createHarness(); + networkFailure.fetch.mockRejectedValueOnce( + new Error("private network detail"), ); - await expect(harness.session.connect()).rejects.toMatchObject({ + await expect(networkFailure.session.connect()).rejects.toMatchObject({ code: "network", - message: - "The voice connection could not be reached. Check your connection, then reconnect voice input.", requestId: "voice-request-1", }); - expect(harness.tracks[0]!.stop).toHaveBeenCalledOnce(); - expect(JSON.stringify(harness.reportDiagnostic.mock.calls)).not.toContain( - "private SDP and credential diagnostics", + expect(networkFailure.localTracks[0]!.stop).toHaveBeenCalledOnce(); + expect( + JSON.stringify(networkFailure.reportDiagnostic.mock.calls), + ).not.toContain("private network detail"); + + const bodyFailure = createHarness(); + bodyFailure.fetch.mockResolvedValueOnce( + new Response( + new ReadableStream({ + start(controller) { + controller.error(new Error("private response stream detail")); + }, + }), + { headers: { "content-type": "application/sdp" } }, + ), ); + + await expect(bodyFailure.session.connect()).rejects.toMatchObject({ + code: "network", + requestId: "voice-request-1", + }); + expect( + JSON.stringify(bodyFailure.reportDiagnostic.mock.calls), + ).not.toContain("private response stream detail"); }); test("classifies a data channel that closes during startup as a network failure", async () => { @@ -531,9 +905,8 @@ describe("OpenAIRealtimeSession", () => { }), ); await vi.waitFor(() => - expect(harness.peers[0]?.setRemoteDescription).toHaveBeenCalledOnce(), + expect(harness.peers[0]!.setRemoteDescription).toHaveBeenCalledOnce(), ); - await Promise.resolve(); harness.channels[0]!.dispatchEvent(new Event("close")); @@ -541,8 +914,7 @@ describe("OpenAIRealtimeSession", () => { code: "network", requestId: "voice-request-1", }); - expect(harness.tracks[0]!.stop).toHaveBeenCalledOnce(); - expect(harness.channels[0]!.close).toHaveBeenCalledOnce(); + expect(harness.localTracks[0]!.stop).toHaveBeenCalledOnce(); expect(harness.peers[0]!.close).toHaveBeenCalledOnce(); }); @@ -574,7 +946,7 @@ describe("OpenAIRealtimeSession", () => { ); await connection; - expect(harness.tracks[0]!.stop).toHaveBeenCalledOnce(); + expect(harness.localTracks[0]!.stop).toHaveBeenCalledOnce(); expect(JSON.stringify(harness.reportDiagnostic.mock.calls)).not.toContain( "private provider diagnostic", ); @@ -673,60 +1045,53 @@ describe("OpenAIRealtimeSession", () => { await connection; }); - test("sanitizes unexpected browser startup failures", async () => { - const harness = createHarness(); + test("sanitizes invalid SDP and browser application failures", async () => { + const invalidAnswer = createHarness(); + invalidAnswer.fetch.mockResolvedValueOnce( + new Response("private invalid answer", { + headers: { "content-type": "application/sdp" }, + }), + ); + + await expect(invalidAnswer.session.connect()).rejects.toMatchObject({ + code: "invalid-response", + requestId: "voice-request-1", + }); + expect( + JSON.stringify(invalidAnswer.reportDiagnostic.mock.calls), + ).not.toContain("private invalid answer"); + + const applicationFailure = createHarness(); let resolveFetch: ((response: Response) => void) | undefined; - harness.fetch.mockImplementationOnce( + applicationFailure.fetch.mockImplementationOnce( () => new Promise((resolve) => { resolveFetch = resolve; }), ); - const connection = harness.session.connect(); - await vi.waitFor(() => expect(harness.fetch).toHaveBeenCalledOnce()); - harness.peers[0]!.setRemoteDescription.mockRejectedValueOnce( - new Error("private browser and SDP diagnostics"), + const connection = applicationFailure.session.connect(); + await vi.waitFor(() => + expect(applicationFailure.fetch).toHaveBeenCalledOnce(), + ); + applicationFailure.peers[0]!.setRemoteDescription.mockRejectedValueOnce( + new Error("private SDP application failure"), ); resolveFetch?.( - new Response("v=0\r\no=private provider answer", { + new Response("v=0\r\no=OpenAI answer", { headers: { "content-type": "application/sdp" }, }), ); await expect(connection).rejects.toMatchObject({ code: "invalid-response", - message: - "The voice connection returned an invalid response. Try again; if it continues, give the diagnostic reference to an operator.", requestId: "voice-request-1", }); - expect(JSON.stringify(harness.reportDiagnostic.mock.calls)).not.toContain( - "private browser and SDP diagnostics", - ); + expect( + JSON.stringify(applicationFailure.reportDiagnostic.mock.calls), + ).not.toContain("private SDP application failure"); }); - test("surfaces disabled voice as unavailable without reading the response body", async () => { - const harness = createHarness(); - const serverRequestId = "00000000-0000-4000-8000-000000000021"; - harness.fetch.mockResolvedValueOnce( - new Response("private provider response", { - headers: { - "x-petrinaut-voice-error": "unavailable", - "x-request-id": serverRequestId, - }, - status: 404, - }), - ); - - await expect(harness.session.connect()).rejects.toMatchObject({ - code: "unavailable", - requestId: serverRequestId, - }); - expect(JSON.stringify(harness.reportDiagnostic.mock.calls)).not.toContain( - "private provider response", - ); - }); - - test("classifies an explicit startup abort and cleans media resources", async () => { + test("classifies explicit disconnect during startup as aborted", async () => { const harness = createHarness(); harness.fetch.mockImplementationOnce( (_input, init) => @@ -736,166 +1101,45 @@ describe("OpenAIRealtimeSession", () => { ); }), ); - const connection = harness.session .connect() .catch((error: unknown) => error); await vi.waitFor(() => expect(harness.fetch).toHaveBeenCalledOnce()); - await harness.session.disconnect(); - - await expect(connection).resolves.toMatchObject({ - code: "request-aborted", - requestId: "voice-request-1", - }); - expect(harness.tracks[0]!.stop).toHaveBeenCalledOnce(); - expect(harness.channels[0]!.close).toHaveBeenCalledOnce(); - expect(harness.peers[0]!.close).toHaveBeenCalledOnce(); - }); - - test("classifies disconnect while reading the SDP answer as aborted", async () => { - const harness = createHarness(); - let rejectAnswerRead: ((reason?: unknown) => void) | undefined; - harness.fetch.mockResolvedValueOnce({ - headers: new Headers({ "content-type": "application/sdp" }), - ok: true, - text: () => - new Promise((_resolve, reject) => { - rejectAnswerRead = reject; - }), - } as Response); - const connection = harness.session - .connect() - .catch((error: unknown) => error); - await vi.waitFor(() => expect(rejectAnswerRead).toBeTypeOf("function")); await harness.session.disconnect(); - rejectAnswerRead?.(new Error("private response read failure")); await expect(connection).resolves.toMatchObject({ code: "request-aborted", requestId: "voice-request-1", }); - expect(harness.reportDiagnostic).toHaveBeenCalledWith( - expect.objectContaining({ - errorCode: "request-aborted", - outcome: "aborted", - }), + expect(harness.localTracks[0]!.stop).toHaveBeenCalledOnce(); + expect(JSON.stringify(harness.reportDiagnostic.mock.calls)).not.toContain( + "private abort detail", ); }); - test("classifies disconnect while applying the SDP answer as aborted", async () => { + test("closes all media when the peer connection fails", async () => { const harness = createHarness(); - let resolveFetch: ((response: Response) => void) | undefined; - let rejectRemoteDescription: ((reason?: unknown) => void) | undefined; - harness.fetch.mockImplementationOnce( - () => - new Promise((resolve) => { - resolveFetch = resolve; - }), - ); - const connection = harness.session - .connect() - .catch((error: unknown) => error); - await vi.waitFor(() => expect(harness.fetch).toHaveBeenCalledOnce()); - const remoteDescription = new Promise((_resolve, reject) => { - rejectRemoteDescription = reject; - }); - harness.peers[0]!.setRemoteDescription.mockReturnValueOnce( - remoteDescription, - ); - resolveFetch?.( - new Response("v=0\r\no=OpenAI answer", { - headers: { "content-type": "application/sdp" }, - }), - ); - await vi.waitFor(() => - expect(harness.peers[0]!.setRemoteDescription).toHaveBeenCalledOnce(), - ); + await harness.session.connect(); + harness.session.setMicrophoneEnabled(true); - await harness.session.disconnect(); - rejectRemoteDescription?.(new Error("private SDP application failure")); + harness.peers[0]!.connectionState = "failed"; + harness.peers[0]!.onconnectionstatechange?.(); - await expect(connection).resolves.toMatchObject({ - code: "request-aborted", - requestId: "voice-request-1", + expect(harness.events.at(-1)).toMatchObject({ + code: "network", + type: "error", }); - expect(harness.reportDiagnostic).toHaveBeenCalledWith( - expect.objectContaining({ - errorCode: "request-aborted", - outcome: "aborted", - }), - ); - }); - - test("times out startup and cleans up resources", async () => { - vi.useFakeTimers(); - const harness = createHarness(); - harness.fetch.mockImplementationOnce( - (_input, init) => - new Promise((_resolve, reject) => { - init?.signal?.addEventListener("abort", () => - reject(new DOMException("aborted", "AbortError")), - ); - }), - ); - - const connection = expect(harness.session.connect()).rejects.toThrow( - "The voice connection timed out. Check your connection, then reconnect voice input.", - ); - await vi.advanceTimersByTimeAsync(15_000); - - await connection; - expect(harness.tracks[0]!.stop).toHaveBeenCalledOnce(); + expect(harness.localTracks[0]!.enabled).toBe(false); + expect(harness.localTracks[0]!.stop).toHaveBeenCalledOnce(); expect(harness.channels[0]!.close).toHaveBeenCalledOnce(); expect(harness.peers[0]!.close).toHaveBeenCalledOnce(); - expect(harness.reportDiagnostic).toHaveBeenCalledWith( - expect.objectContaining({ - errorCode: "timeout", - outcome: "failure", - requestId: "voice-request-1", - }), - ); - }); - - test("times out when abort occurs before waiting for the data channel", async () => { - vi.useFakeTimers(); - const harness = createHarness(); - let resolveAnswer: ((answer: string) => void) | undefined; - harness.fetch.mockResolvedValueOnce({ - headers: new Headers({ "content-type": "application/sdp" }), - ok: true, - text: () => - new Promise((resolve) => { - resolveAnswer = resolve; - }), - } as Response); - let settled = false; - const connection = harness.session.connect().then( - () => new Error("Connection unexpectedly succeeded."), - (error: unknown) => { - settled = true; - return error; - }, - ); - await vi.waitFor(() => expect(harness.fetch).toHaveBeenCalledOnce()); - harness.peers[0]!.setRemoteDescription.mockResolvedValueOnce(undefined); - - await vi.advanceTimersByTimeAsync(15_000); - resolveAnswer?.("v=0\r\no=late OpenAI answer"); - - await vi.waitFor(() => expect(settled).toBe(true)); - await expect(connection).resolves.toMatchObject({ - code: "timeout", - message: - "The voice connection timed out. Check your connection, then reconnect voice input.", - requestId: "voice-request-1", - }); }); - test("times out a stalled permission prompt and stops a late media stream", async () => { + test("times out a stalled microphone permission prompt and stops late media", async () => { vi.useFakeTimers(); const harness = createHarness(); - const lateTrack = { enabled: true, stop: vi.fn() }; + const lateTrack = { enabled: true, kind: "audio", stop: vi.fn() }; const lateStream = { getAudioTracks: () => [lateTrack], getTracks: () => [lateTrack], @@ -907,93 +1151,17 @@ describe("OpenAIRealtimeSession", () => { resolveMedia = resolve; }), ); - let settled = false; - const connection = harness.session.connect().then( - () => new Error("Connection unexpectedly succeeded."), - (error: unknown) => { - settled = true; - return error; - }, - ); + const connection = harness.session + .connect() + .catch((error: unknown) => error); await vi.advanceTimersByTimeAsync(15_000); - const settledAtTimeout = settled; resolveMedia?.(lateStream); - const error = await connection; - expect(settledAtTimeout).toBe(true); - expect(error).toMatchObject({ + await expect(connection).resolves.toMatchObject({ code: "timeout", - message: - "The voice connection timed out. Check your connection, then reconnect voice input.", requestId: "voice-request-1", }); expect(lateTrack.stop).toHaveBeenCalledOnce(); }); - - test("closes the microphone and emits a recoverable connection error", async () => { - const harness = createHarness(); - await harness.session.connect(); - harness.session.setMicrophoneEnabled(true); - - harness.peers[0]!.connectionState = "failed"; - harness.peers[0]!.onconnectionstatechange?.(); - - expect(harness.tracks[0]!.enabled).toBe(false); - expect(harness.tracks[0]!.stop).toHaveBeenCalledOnce(); - expect(harness.channels[0]!.close).toHaveBeenCalledOnce(); - expect(harness.peers[0]!.close).toHaveBeenCalledOnce(); - expect(harness.events).toEqual([ - { - code: "network", - message: - "The voice connection could not be reached. Check your connection, then reconnect voice input.", - requestId: "voice-request-1", - type: "error", - }, - ]); - expect(harness.reportDiagnostic).toHaveBeenLastCalledWith({ - durationMs: 0, - errorCode: "network", - operation: "connection", - outcome: "failure", - requestId: "voice-request-1", - stage: "browser", - }); - }); - - test("fails closed on a malformed completed provider transcript", async () => { - const harness = createHarness(); - await harness.session.connect(); - harness.session.setMicrophoneEnabled(true); - - harness.channels[0]!.receive({ - type: "conversation.item.input_audio_transcription.completed", - item_id: "item-a", - content_index: 0, - transcript: { private: "provider response body" }, - }); - - expect(harness.tracks[0]!.stop).toHaveBeenCalledOnce(); - expect(harness.events).toEqual([ - { - code: "invalid-response", - message: - "The transcription service returned an invalid response. Try again; if it continues, give the diagnostic reference to an operator.", - requestId: "voice-request-2", - type: "error", - }, - ]); - expect(harness.reportDiagnostic).toHaveBeenLastCalledWith({ - durationMs: 0, - errorCode: "invalid-response", - operation: "transcription", - outcome: "failure", - requestId: "voice-request-2", - stage: "browser", - }); - expect(JSON.stringify(harness.reportDiagnostic.mock.calls)).not.toContain( - "provider response body", - ); - }); }); diff --git a/apps/petrinaut-website/src/main/app/voice-interview/openai-realtime-session.ts b/apps/petrinaut-website/src/main/app/voice-interview/openai-realtime-session.ts index a27585f89e4..44139b9d740 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/openai-realtime-session.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/openai-realtime-session.ts @@ -11,24 +11,72 @@ import { type VoiceOperation, } from "../../../voice-diagnostics"; +import type { CanonicalSpeechSegment } from "./canonical-speech"; + export interface OpenAIRealtimeTranscriptKey { readonly connectionEpoch: number; readonly contentIndex: number; readonly itemId: string; } +interface RealtimeToolEventIdentity { + readonly callId: string; + readonly connectionEpoch: number; + readonly itemId: string; + readonly responseId: string; +} + export type OpenAIRealtimeSessionEvent = - | { - readonly connectionEpoch: number; - readonly itemId: string; - readonly type: "input-committed"; - } | { readonly key: OpenAIRealtimeTranscriptKey; readonly text: string; readonly type: "partial" | "completed"; } + | { + readonly key: OpenAIRealtimeTranscriptKey; + readonly type: "transcription-failed"; + } | { readonly level: number; readonly type: "microphone-level" } + | { + readonly connectionEpoch: number; + readonly itemId: string; + readonly type: "input-speech-started"; + } + | { + readonly connectionEpoch: number; + readonly itemId: string; + readonly type: "input-speech-stopped"; + } + | { + readonly connectionEpoch: number; + readonly responseId: string; + readonly type: "output-started"; + } + | { + readonly connectionEpoch: number; + readonly responseId: string; + readonly type: "output-stopped"; + } + | { + readonly connectionEpoch: number; + readonly responseId: string; + readonly type: "output-interrupted"; + } + | { + readonly connectionEpoch: number; + readonly responseId: string; + readonly status: "cancelled" | "completed" | "failed" | "incomplete"; + readonly type: "response-terminal"; + } + | (RealtimeToolEventIdentity & { + readonly delta: string; + readonly type: "tool-arguments-delta"; + }) + | (RealtimeToolEventIdentity & { + readonly arguments: string; + readonly name: string; + readonly type: "tool-arguments-done"; + }) | { readonly code: VoiceErrorCode; readonly message: string; @@ -36,10 +84,18 @@ export type OpenAIRealtimeSessionEvent = readonly type: "error"; }; +interface RemoteAudio { + autoplay: boolean; + srcObject: HTMLMediaElement["srcObject"]; + pause(): void; + play(): Promise; +} + interface OpenAIRealtimeSessionDependencies { readonly cancelAnimationFrame: (handle: number) => void; readonly connectionTimeoutMs: number; readonly createAudioContext: () => AudioContext; + readonly createRemoteAudio: () => RemoteAudio; readonly createRequestId?: () => string; readonly createPeerConnection: () => RTCPeerConnection; readonly fetch: typeof globalThis.fetch; @@ -51,23 +107,48 @@ interface OpenAIRealtimeSessionDependencies { readonly requestAnimationFrame: (callback: FrameRequestCallback) => number; } -type SessionListener = (event: OpenAIRealtimeSessionEvent) => void; - -interface TranscriptionTiming { +interface RequestTiming { readonly requestId: string; readonly startedAt: number; } +interface CanonicalSpeechRequest { + readonly response: Record; + readonly speechRequestId: string; +} + +type PendingClientEvent = + | { + readonly kind: "response-cancel"; + readonly responseId: string; + } + | { + readonly kind: "response-create"; + readonly request: CanonicalSpeechRequest; + readonly responseTerminalSequence: number; + }; + +type SessionListener = (event: OpenAIRealtimeSessionEvent) => void; + +const CANONICAL_RESPONSE_INSTRUCTIONS = + "Speak only the response_text strings supplied by Petrinaut, in array order and verbatim. Deliver them as a warm, calm, curious, confident, concise, and professionally neutral expert interviewer, at a measured conversational pace with natural emphasis. Never sound robotic, fawning, rushed, overenthusiastic, or patronizing. Do not add, remove, paraphrase, acknowledge, or explain anything."; +const MAX_CANONICAL_SEGMENTS = 64; + const asRecord = (value: unknown): Record | null => - typeof value === "object" && value !== null + typeof value === "object" && value !== null && !Array.isArray(value) ? (value as Record) : null; +const nonEmptyString = (value: unknown): string | null => + typeof value === "string" && value ? value : null; + +const nonNegativeInteger = (value: unknown): number | null => + Number.isInteger(value) && (value as number) >= 0 ? (value as number) : null; + const parseRealtimeEvent = (value: unknown): Record | null => { if (typeof value !== "string") { return null; } - try { return asRecord(JSON.parse(value)); } catch { @@ -88,7 +169,6 @@ const waitForAbort = ( if (signal.aborted) { return Promise.reject(signal.reason); } - return new Promise((resolve, reject) => { const handleAbort = () => reject(signal.reason); signal.addEventListener("abort", handleAbort, { once: true }); @@ -107,26 +187,44 @@ const waitForAbort = ( export class OpenAIRealtimeSession { readonly #dependencies: OpenAIRealtimeSessionDependencies; + readonly #activeResponseIds = new Set(); readonly #listeners = new Set(); - #analyser: AnalyserNode | null = null; - #audioContext: AudioContext | null = null; + readonly #authorizedResponseIds = new Set(); + readonly #cancelledCanonicalResponseIds = new Set(); + readonly #cancelledSpeechRequestIds = new Set(); + readonly #canonicalResponseIds = new Set(); + readonly #canonicalSpeechQueue: CanonicalSpeechRequest[] = []; + readonly #completedResponseCancelEventIds = new Set(); + readonly #pendingClientEvents = new Map(); + readonly #pendingSpeechRequests = new Map(); + readonly #remoteStreams = new Set(); + readonly #speechTimings = new Map(); + readonly #transcriptionTimings = new Map(); #abortController: AbortController | null = null; #activeEpoch: number | null = null; + #analyser: AnalyserNode | null = null; + #audioContext: AudioContext | null = null; #connected = false; #connectedAt: number | null = null; + #clientEventSequence = 0; #connectionRequestId: string | null = null; #dataChannel: RTCDataChannel | null = null; #epoch = 0; #mediaStream: MediaStream | null = null; + #messageListener: ((event: MessageEvent) => void) | null = null; #meterFrame: number | null = null; #meterHasSample = false; #meterLevel = 0; #meterSamples: Uint8Array | null = null; - #messageListener: ((event: MessageEvent) => void) | null = null; #microphoneTrack: MediaStreamTrack | null = null; #peerConnection: RTCPeerConnection | null = null; + #remoteAudio: RemoteAudio | null = null; + #responseCreateEventId: string | null = null; + #responseTerminalSequence = 0; + #speakingResponseId: string | null = null; + #speechRequestSequence = 0; #unexpectedCloseListener: (() => void) | null = null; - readonly #transcriptionTimings = new Map(); + #waitingForResponseTerminal = false; public constructor(dependencies: OpenAIRealtimeSessionDependencies) { this.#dependencies = dependencies; @@ -139,7 +237,6 @@ export class OpenAIRealtimeSession { public async connect(): Promise { this.#releaseResources(); - const requestId = this.#dependencies.createRequestId?.() ?? createVoiceRequestId(); const startedAt = this.#now(); @@ -152,91 +249,28 @@ export class OpenAIRealtimeSession { "Connection timed out", "TimeoutError", ); - const timeout = globalThis.setTimeout(() => { - abortController.abort(timeoutError); - }, this.#dependencies.connectionTimeoutMs); + const timeout = globalThis.setTimeout( + () => abortController.abort(timeoutError), + this.#dependencies.connectionTimeoutMs, + ); try { - let audioContext: AudioContext | null = null; - try { - audioContext = this.#dependencies.createAudioContext(); - this.#audioContext = audioContext; - if (audioContext.state === "suspended") { - try { - void audioContext.resume().catch(() => undefined); - } catch { - // Input metering is optional and must not block voice connection. - } - } - } catch { - // Input metering is optional and must not block voice connection. - } - - let mediaStream: MediaStream; - try { - const mediaStreamPromise = this.#dependencies.getUserMedia({ - audio: { - autoGainControl: true, - echoCancellation: true, - noiseSuppression: true, - }, - }); - void mediaStreamPromise.then( - (lateMediaStream) => { - if (this.#activeEpoch !== connectionEpoch) { - stopStream(lateMediaStream); - } - }, - () => undefined, - ); - mediaStream = await waitForAbort( - mediaStreamPromise, - abortController.signal, - ); - } catch (error) { - if ( - error instanceof DOMException && - (error.name === "NotAllowedError" || error.name === "SecurityError") - ) { - throw new VoiceError( - "connection", - "microphone-permission", - requestId, - ); - } - if ( - error instanceof DOMException && - [ - "DevicesNotFoundError", - "NotFoundError", - "NotReadableError", - "OverconstrainedError", - "TrackStartError", - ].includes(error.name) - ) { - throw new VoiceError("connection", "microphone-device", requestId); - } - if (abortController.signal.aborted) { - throw error; - } - throw new VoiceError("connection", "microphone-device", requestId); - } - - if (this.#activeEpoch !== connectionEpoch) { - stopStream(mediaStream); - throw new VoiceError("connection", "request-aborted", requestId); - } + this.#initializeOptionalMeter(); + const mediaStream = await this.#getMediaStream( + abortController.signal, + requestId, + connectionEpoch, + ); this.#mediaStream = mediaStream; - const [microphoneTrack] = mediaStream.getAudioTracks(); if (!microphoneTrack) { throw new VoiceError("connection", "microphone-device", requestId); } microphoneTrack.enabled = false; this.#microphoneTrack = microphoneTrack; - if (audioContext) { + if (this.#audioContext) { try { - this.#initializeMeter(audioContext, mediaStream); + this.#initializeMeter(this.#audioContext, mediaStream); } catch { this.#releaseMeterResources(); } @@ -244,6 +278,27 @@ export class OpenAIRealtimeSession { const peerConnection = this.#dependencies.createPeerConnection(); this.#peerConnection = peerConnection; + this.#remoteAudio = this.#dependencies.createRemoteAudio(); + this.#remoteAudio.autoplay = true; + peerConnection.ontrack = (event) => { + if ( + this.#activeEpoch !== connectionEpoch || + event.track.kind !== "audio" + ) { + return; + } + const [stream] = event.streams; + if (!stream || !this.#remoteAudio) { + return; + } + this.#remoteStreams.add(stream); + this.#remoteAudio.srcObject = stream; + try { + void this.#remoteAudio.play().catch(() => undefined); + } catch { + // Autoplay remains enabled; the media element will retry on audio. + } + }; peerConnection.addTrack(microphoneTrack, mediaStream); peerConnection.onconnectionstatechange = () => { if ( @@ -256,21 +311,7 @@ export class OpenAIRealtimeSession { const dataChannel = peerConnection.createDataChannel("oai-events"); this.#dataChannel = dataChannel; - const messageListener = (event: MessageEvent) => { - if (this.#activeEpoch === connectionEpoch) { - this.#handleMessage(event, connectionEpoch); - } - }; - const unexpectedCloseListener = () => { - if (this.#activeEpoch === connectionEpoch && this.#connected) { - this.#handleConnectionFailure("network", "connection"); - } - }; - this.#messageListener = messageListener; - this.#unexpectedCloseListener = unexpectedCloseListener; - dataChannel.addEventListener("message", messageListener); - dataChannel.addEventListener("close", unexpectedCloseListener); - dataChannel.addEventListener("error", unexpectedCloseListener); + this.#installDataChannelListeners(dataChannel, connectionEpoch); const offer = await peerConnection.createOffer(); await peerConnection.setLocalDescription(offer); @@ -278,61 +319,25 @@ export class OpenAIRealtimeSession { if (!offerSdp) { throw new VoiceError("connection", "invalid-response", requestId); } - - let response: Response; - try { - response = await this.#dependencies.fetch("/api/voice/realtime-call", { - body: offerSdp, - headers: { - "content-type": "application/sdp", - [VOICE_REQUEST_ID_HEADER]: requestId, - }, - method: "POST", - signal: abortController.signal, - }); - } catch (error) { - if (abortController.signal.aborted) { - throw error; - } - throw new VoiceError("connection", "network", requestId); - } - if (!response.ok) { - await response.body?.cancel(); - throw voiceErrorFromResponse(response, "connection", requestId); - } - const contentType = response.headers - .get("content-type") - ?.split(";", 1)[0] - ?.trim() - .toLowerCase(); - if (contentType !== "application/sdp") { - await response.body?.cancel(); - throw new VoiceError("connection", "invalid-response", requestId); - } + const response = await this.#requestRealtimeCall( + offerSdp, + abortController.signal, + requestId, + ); let answerSdp: string; try { answerSdp = await response.text(); } catch (error) { - if (abortController.signal.aborted) { - throw error; - } + if (abortController.signal.aborted) throw error; throw new VoiceError("connection", "network", requestId); } if (!answerSdp.trim() || !answerSdp.trimStart().startsWith("v=0")) { throw new VoiceError("connection", "invalid-response", requestId); } - - try { - await peerConnection.setRemoteDescription({ - sdp: answerSdp, - type: "answer", - }); - } catch (error) { - if (abortController.signal.aborted) { - throw error; - } - throw new VoiceError("connection", "invalid-response", requestId); - } + await peerConnection.setRemoteDescription({ + sdp: answerSdp, + type: "answer", + }); await this.#waitForDataChannelOpen( dataChannel, abortController.signal, @@ -347,14 +352,15 @@ export class OpenAIRealtimeSession { this.#connected = true; this.#connectedAt = this.#now(); - this.#reportDiagnostic("connection", requestId, startedAt, undefined); + this.#reportDiagnostic("connection", requestId, startedAt); return connectionEpoch; } catch (error) { + const abortReason: unknown = abortController.signal.reason; const voiceError = - abortController.signal.reason === timeoutError + abortReason === timeoutError ? new VoiceError("connection", "timeout", requestId) - : abortController.signal.reason instanceof VoiceError - ? abortController.signal.reason + : abortReason instanceof VoiceError + ? abortReason : error instanceof VoiceError ? error : abortController.signal.aborted || @@ -380,37 +386,642 @@ export class OpenAIRealtimeSession { } public setMicrophoneEnabled(enabled: boolean): void { - if (this.#microphoneTrack) { - const isEnabled = enabled && this.#connected; - this.#microphoneTrack.enabled = isEnabled; - if (isEnabled) { - this.#startMeter(); - } else { - this.#stopMeter(); + if (!this.#microphoneTrack) { + return; + } + const isEnabled = enabled && this.#connected; + this.#microphoneTrack.enabled = isEnabled; + if (isEnabled) { + this.#startMeter(); + } else { + this.#stopMeter(); + } + } + + public speakCanonical(segments: CanonicalSpeechSegment[]): void { + this.#requestCanonicalSpeech(segments, true); + } + + public completeFunctionCall( + callId: string, + segments: CanonicalSpeechSegment[], + ): void { + if (!callId) { + throw new VoiceError("speech", "invalid-response", ""); + } + const responseText = this.#canonicalResponseText(segments); + this.#send({ + type: "conversation.item.create", + item: { + type: "function_call_output", + call_id: callId, + output: JSON.stringify({ response_text: responseText }), + }, + }); + this.#requestCanonicalSpeech(segments, false); + } + + public cancelOutput(): void { + if ( + !this.#connected || + this.#dataChannel?.readyState !== "open" + ) { + return; + } + + for (const request of this.#canonicalSpeechQueue.splice(0)) { + this.#cancelPendingSpeechRequest(request.speechRequestId); + } + + if (this.#responseCreateEventId !== null) { + const pendingEvent = this.#pendingClientEvents.get( + this.#responseCreateEventId, + ); + if (pendingEvent?.kind === "response-create") { + this.#cancelledSpeechRequestIds.add( + pendingEvent.request.speechRequestId, + ); } } + + for (const responseId of this.#canonicalResponseIds) { + if ( + this.#activeResponseIds.has(responseId) && + !this.#cancelledCanonicalResponseIds.has(responseId) + ) { + this.#cancelledCanonicalResponseIds.add(responseId); + this.#cancelOutputResponse(responseId); + } + } + } + + #cancelOutputResponse(responseId: string): void { + this.#cancelResponse(responseId); + this.#send({ type: "output_audio_buffer.clear" }); } public async disconnect(): Promise { this.#releaseResources(); } - #emit(event: OpenAIRealtimeSessionEvent): void { - for (const listener of this.#listeners) { - listener(event); + #canonicalResponseText(segments: CanonicalSpeechSegment[]): string[] { + const responseText = segments + .slice(0, MAX_CANONICAL_SEGMENTS) + .map(({ text }) => text.trim()) + .filter(Boolean); + if (responseText.length === 0 || responseText.length !== segments.length) { + throw new VoiceError("speech", "invalid-response", ""); + } + return responseText; + } + + #requestCanonicalSpeech( + segments: CanonicalSpeechSegment[], + outOfBand: boolean, + ): void { + const responseText = this.#canonicalResponseText(segments); + const speechRequestId = `canonical-${this.#activeEpoch}-${++this.#speechRequestSequence}`; + this.#pendingSpeechRequests.set(speechRequestId, { + requestId: + this.#dependencies.createRequestId?.() ?? createVoiceRequestId(), + startedAt: this.#now(), + }); + const response = { + ...(outOfBand + ? { + conversation: "none", + input: [ + { + type: "message", + role: "system", + content: [ + { + type: "input_text", + text: JSON.stringify({ response_text: responseText }), + }, + ], + }, + ], + } + : {}), + instructions: CANONICAL_RESPONSE_INSTRUCTIONS, + output_modalities: ["audio"], + parallel_tool_calls: false, + tool_choice: "none", + tools: [], + metadata: { + petrinaut_kind: "canonical-speech", + petrinaut_request_id: speechRequestId, + }, + }; + const request = { response, speechRequestId }; + this.#canonicalSpeechQueue.push(request); + try { + this.#sendNextCanonicalSpeech(); + } catch (error) { + const queuedRequestIndex = this.#canonicalSpeechQueue.indexOf(request); + if (queuedRequestIndex >= 0) { + this.#canonicalSpeechQueue.splice(queuedRequestIndex, 1); + } + this.#pendingSpeechRequests.delete(speechRequestId); + throw error; + } + } + + #cancelResponse(responseId: string): void { + const eventId = this.#createClientEventId(); + this.#pendingClientEvents.set(eventId, { + kind: "response-cancel", + responseId, + }); + try { + this.#send({ + event_id: eventId, + response_id: responseId, + type: "response.cancel", + }); + } catch (error) { + this.#pendingClientEvents.delete(eventId); + throw error; + } + } + + #createClientEventId(): string { + return `petrinaut-${this.#activeEpoch}-${++this.#clientEventSequence}`; + } + + #sendNextCanonicalSpeech(): void { + if ( + this.#activeResponseIds.size > 0 || + this.#responseCreateEventId !== null || + this.#waitingForResponseTerminal + ) { + return; + } + const request = this.#canonicalSpeechQueue.shift(); + if (!request) { + return; + } + + const eventId = this.#createClientEventId(); + this.#responseCreateEventId = eventId; + this.#pendingClientEvents.set(eventId, { + kind: "response-create", + request, + responseTerminalSequence: this.#responseTerminalSequence, + }); + try { + this.#send({ + event_id: eventId, + response: request.response, + type: "response.create", + }); + } catch (error) { + this.#responseCreateEventId = null; + this.#pendingClientEvents.delete(eventId); + throw error; + } + } + + #send(event: Record): void { + if (!this.#connected || this.#dataChannel?.readyState !== "open") { + throw new VoiceError("connection", "network", ""); + } + this.#dataChannel.send(JSON.stringify(event)); + } + + #handleMessage(event: MessageEvent, connectionEpoch: number): void { + const parsed = parseRealtimeEvent(event.data); + if (!parsed || typeof parsed.type !== "string") { + return; + } + if (parsed.type === "error") { + this.#handleProviderError(parsed); + return; + } + if (parsed.type === "response.created") { + this.#handleResponseCreated(parsed); + return; + } + if (parsed.type === "response.done") { + this.#handleResponseDone(parsed, connectionEpoch); + return; + } + if (parsed.type === "input_audio_buffer.committed") { + const itemId = nonEmptyString(parsed.item_id); + if (itemId) this.#startTranscription(itemId); + return; + } + if (parsed.type === "input_audio_buffer.speech_started") { + const itemId = nonEmptyString(parsed.item_id); + if (!itemId || nonNegativeInteger(parsed.audio_start_ms) === null) return; + this.#emit({ + connectionEpoch, + itemId, + type: "input-speech-started", + }); + if (this.#speakingResponseId) { + this.#emit({ + connectionEpoch, + responseId: this.#speakingResponseId, + type: "output-interrupted", + }); + } + return; + } + if (parsed.type === "input_audio_buffer.speech_stopped") { + const itemId = nonEmptyString(parsed.item_id); + if (!itemId || nonNegativeInteger(parsed.audio_end_ms) === null) return; + this.#emit({ + connectionEpoch, + itemId, + type: "input-speech-stopped", + }); + return; + } + if ( + parsed.type === "output_audio_buffer.started" || + parsed.type === "output_audio_buffer.stopped" + ) { + this.#handleOutputBufferEvent(parsed, connectionEpoch); + return; + } + if (parsed.type === "response.function_call_arguments.delta") { + this.#handleToolEvent(parsed, connectionEpoch); + return; + } + if ( + parsed.type === "conversation.item.input_audio_transcription.delta" || + parsed.type === "conversation.item.input_audio_transcription.completed" || + parsed.type === "conversation.item.input_audio_transcription.failed" + ) { + this.#handleTranscriptEvent(parsed, connectionEpoch); + } + } + + #handleResponseCreated(event: Record): void { + const response = asRecord(event.response); + const responseId = nonEmptyString(response?.id); + if (!responseId) { + return; + } + this.#activeResponseIds.add(responseId); + const metadata = asRecord(response?.metadata); + const speechRequestId = nonEmptyString(metadata?.petrinaut_request_id); + if (metadata?.petrinaut_kind !== "canonical-speech" || !speechRequestId) { + return; + } + this.#completeResponseCreateEvent(speechRequestId); + this.#canonicalResponseIds.add(responseId); + if (this.#cancelledSpeechRequestIds.delete(speechRequestId)) { + this.#cancelPendingSpeechRequest(speechRequestId); + this.#cancelledCanonicalResponseIds.add(responseId); + this.#cancelOutputResponse(responseId); + return; + } + const timing = this.#pendingSpeechRequests.get(speechRequestId); + if (!timing) { + return; + } + this.#pendingSpeechRequests.delete(speechRequestId); + this.#authorizedResponseIds.add(responseId); + this.#speechTimings.set(responseId, timing); + } + + #completeResponseCreateEvent(speechRequestId: string): void { + if (!this.#responseCreateEventId) { + return; + } + const pendingEvent = this.#pendingClientEvents.get( + this.#responseCreateEventId, + ); + if ( + pendingEvent?.kind !== "response-create" || + pendingEvent.request.speechRequestId !== speechRequestId + ) { + return; } + this.#pendingClientEvents.delete(this.#responseCreateEventId); + this.#responseCreateEventId = null; + } + + #handleProviderError(event: Record): void { + const providerError = asRecord(event.error); + const errorType = nonEmptyString(providerError?.type); + const errorCode = nonEmptyString(providerError?.code); + const sourceEventId = nonEmptyString(providerError?.event_id); + const pendingEvent = sourceEventId + ? this.#pendingClientEvents.get(sourceEventId) + : undefined; + const completedResponseCancel = + sourceEventId !== null && + this.#completedResponseCancelEventIds.has(sourceEventId); + + if ( + errorType === "invalid_request_error" && + errorCode === "response_cancel_not_active" && + sourceEventId && + (pendingEvent?.kind === "response-cancel" || completedResponseCancel) + ) { + this.#pendingClientEvents.delete(sourceEventId); + this.#completedResponseCancelEventIds.delete(sourceEventId); + return; + } + + if ( + errorType === "invalid_request_error" && + errorCode === "conversation_already_has_active_response" && + pendingEvent?.kind === "response-create" && + sourceEventId + ) { + this.#pendingClientEvents.delete(sourceEventId); + if (this.#responseCreateEventId === sourceEventId) { + this.#responseCreateEventId = null; + } + if ( + this.#cancelledSpeechRequestIds.delete( + pendingEvent.request.speechRequestId, + ) + ) { + this.#cancelPendingSpeechRequest(pendingEvent.request.speechRequestId); + return; + } + this.#canonicalSpeechQueue.unshift(pendingEvent.request); + this.#waitingForResponseTerminal = + this.#responseTerminalSequence === + pendingEvent.responseTerminalSequence; + this.#resumeCanonicalSpeechQueue(); + return; + } + + this.#handleConnectionFailure("invalid-response", "connection"); + } + + #handleResponseDone( + event: Record, + connectionEpoch: number, + ): void { + const response = asRecord(event.response); + const responseId = nonEmptyString(response?.id); + const status = response?.status; + if (!response || !responseId || typeof status !== "string") { + return; + } + if ( + status !== "completed" && + status !== "cancelled" && + status !== "failed" && + status !== "incomplete" + ) { + this.#handleConnectionFailure("invalid-response", "connection"); + return; + } + this.#responseTerminalSequence += 1; + this.#activeResponseIds.delete(responseId); + this.#clearResponseCancelEvents(responseId); + this.#waitingForResponseTerminal = false; + + if (this.#cancelledCanonicalResponseIds.delete(responseId)) { + this.#emit({ + connectionEpoch, + responseId, + status, + type: "response-terminal", + }); + this.#finishSpeech(responseId, "request-aborted"); + this.#resumeCanonicalSpeechQueue(); + return; + } + + if (status === "completed") { + const output = response.output; + if (!Array.isArray(output)) { + this.#handleConnectionFailure("invalid-response", "connection"); + return; + } + const functionCalls = output + .map(asRecord) + .filter( + (item): item is Record => + item?.type === "function_call", + ); + if ( + functionCalls.length > 1 || + (functionCalls.length > 0 && this.#canonicalResponseIds.has(responseId)) + ) { + this.#handleConnectionFailure("invalid-response", "connection"); + return; + } + for (const item of functionCalls) { + const argumentsJson = nonEmptyString(item.arguments); + const callId = nonEmptyString(item.call_id); + const itemId = nonEmptyString(item.id); + const name = nonEmptyString(item.name); + if ( + !argumentsJson || + !callId || + !itemId || + !name || + (item.status !== undefined && item.status !== "completed") + ) { + this.#handleConnectionFailure("invalid-response", "connection"); + return; + } + this.#emit({ + arguments: argumentsJson, + callId, + connectionEpoch, + itemId, + name, + responseId, + type: "tool-arguments-done", + }); + } + this.#emit({ + connectionEpoch, + responseId, + status, + type: "response-terminal", + }); + this.#resumeCanonicalSpeechQueue(); + return; + } + if (status === "cancelled") { + if (this.#speakingResponseId === responseId) { + this.#emit({ + connectionEpoch, + responseId, + type: "output-interrupted", + }); + } + this.#emit({ + connectionEpoch, + responseId, + status, + type: "response-terminal", + }); + this.#finishSpeech(responseId, "request-aborted"); + this.#resumeCanonicalSpeechQueue(); + return; + } + this.#emit({ + connectionEpoch, + responseId, + status, + type: "response-terminal", + }); + if (this.#authorizedResponseIds.has(responseId)) { + this.#finishSpeech(responseId, "invalid-response"); + } + this.#handleConnectionFailure("invalid-response", "connection"); + } + + #clearResponseCancelEvents(responseId: string): void { + for (const [eventId, pendingEvent] of this.#pendingClientEvents) { + if ( + pendingEvent.kind === "response-cancel" && + pendingEvent.responseId === responseId + ) { + this.#pendingClientEvents.delete(eventId); + this.#completedResponseCancelEventIds.add(eventId); + } + } + } + + #resumeCanonicalSpeechQueue(): void { + try { + this.#sendNextCanonicalSpeech(); + } catch { + this.#handleConnectionFailure("network", "speech"); + } + } + + #handleOutputBufferEvent( + event: Record, + connectionEpoch: number, + ): void { + const responseId = nonEmptyString(event.response_id); + if (!responseId) return; + if (event.type === "output_audio_buffer.started") { + if (this.#cancelledCanonicalResponseIds.has(responseId)) { + this.#send({ type: "output_audio_buffer.clear" }); + return; + } + if (!this.#authorizedResponseIds.has(responseId)) { + this.#cancelOutputResponse(responseId); + this.#handleConnectionFailure("invalid-response", "connection"); + return; + } + this.#speakingResponseId = responseId; + this.#emit({ connectionEpoch, responseId, type: "output-started" }); + return; + } + const wasSpeaking = this.#speakingResponseId === responseId; + this.#finishSpeech( + responseId, + this.#cancelledCanonicalResponseIds.has(responseId) + ? "request-aborted" + : undefined, + ); + if (wasSpeaking) { + this.#emit({ connectionEpoch, responseId, type: "output-stopped" }); + } + } + + #handleToolEvent( + event: Record, + connectionEpoch: number, + ): void { + const callId = nonEmptyString(event.call_id); + const itemId = nonEmptyString(event.item_id); + const responseId = nonEmptyString(event.response_id); + const outputIndex = nonNegativeInteger(event.output_index); + if (!callId || !itemId || !responseId || outputIndex === null) return; + if (typeof event.delta !== "string") return; + this.#emit({ + callId, + connectionEpoch, + delta: event.delta, + itemId, + responseId, + type: "tool-arguments-delta", + }); + } + + #handleTranscriptEvent( + event: Record, + connectionEpoch: number, + ): void { + const itemId = nonEmptyString(event.item_id); + const contentIndex = nonNegativeInteger(event.content_index); + if (!itemId || contentIndex === null) return; + const key = { connectionEpoch, contentIndex, itemId }; + this.#startTranscription(itemId); + if (event.type === "conversation.item.input_audio_transcription.failed") { + this.#finishTranscription(itemId, "invalid-response"); + this.#emit({ key, type: "transcription-failed" }); + return; + } + const text = + event.type === "conversation.item.input_audio_transcription.delta" + ? event.delta + : event.transcript; + if (typeof text !== "string") return; + if ( + event.type === "conversation.item.input_audio_transcription.completed" + ) { + this.#finishTranscription(itemId); + } + this.#emit({ + key, + text, + type: + event.type === "conversation.item.input_audio_transcription.delta" + ? "partial" + : "completed", + }); + } + + #cancelPendingSpeechRequest(speechRequestId: string): void { + const timing = this.#pendingSpeechRequests.get(speechRequestId); + if (!timing) { + return; + } + this.#pendingSpeechRequests.delete(speechRequestId); + this.#reportDiagnostic( + "speech", + timing.requestId, + timing.startedAt, + "request-aborted", + ); + } + + #finishSpeech(responseId: string, errorCode?: VoiceErrorCode): void { + const timing = this.#speechTimings.get(responseId); + if (timing) { + this.#speechTimings.delete(responseId); + this.#reportDiagnostic( + "speech", + timing.requestId, + timing.startedAt, + errorCode, + ); + } + this.#authorizedResponseIds.delete(responseId); + if (this.#speakingResponseId === responseId) { + this.#speakingResponseId = null; + } + } + + #emit(event: OpenAIRealtimeSessionEvent): void { + for (const listener of this.#listeners) listener(event); } #handleConnectionFailure( code: VoiceErrorCode, operation: VoiceOperation, ): void { - const transcriptionTiming = - operation === "transcription" - ? this.#transcriptionTimings.values().next().value - : undefined; const requestId = - transcriptionTiming?.requestId ?? this.#connectionRequestId ?? this.#dependencies.createRequestId?.() ?? createVoiceRequestId(); @@ -418,33 +1029,12 @@ export class OpenAIRealtimeSession { this.#abortController?.abort(new VoiceError(operation, code, requestId)); return; } - if (operation === "transcription") { - if (this.#transcriptionTimings.size === 0) { - this.#reportDiagnostic( - operation, - requestId, - this.#connectedAt ?? this.#now(), - code, - ); - } else { - for (const timing of this.#transcriptionTimings.values()) { - this.#reportDiagnostic( - operation, - timing.requestId, - timing.startedAt, - code, - ); - } - this.#transcriptionTimings.clear(); - } - } else { - this.#reportDiagnostic( - operation, - requestId, - this.#connectedAt ?? this.#now(), - code, - ); - } + this.#reportDiagnostic( + operation, + requestId, + this.#connectedAt ?? this.#now(), + code, + ); this.#releaseResources(); this.#emit({ code, @@ -454,10 +1044,115 @@ export class OpenAIRealtimeSession { }); } - #initializeMeter(audioContext: AudioContext, mediaStream: MediaStream): void { + #initializeOptionalMeter(): void { + try { + const audioContext = this.#dependencies.createAudioContext(); + this.#audioContext = audioContext; + if (audioContext.state === "suspended") { + try { + void audioContext.resume().catch(() => undefined); + } catch { + // Input metering is optional and must not block connection. + } + } + } catch { + // Input metering is optional and must not block connection. + } + } + + async #getMediaStream( + signal: AbortSignal, + requestId: string, + connectionEpoch: number, + ): Promise { + try { + const promise = this.#dependencies.getUserMedia({ + audio: { + autoGainControl: true, + echoCancellation: true, + noiseSuppression: true, + }, + }); + void promise.then( + (lateStream) => { + if (this.#activeEpoch !== connectionEpoch) stopStream(lateStream); + }, + () => undefined, + ); + return await waitForAbort(promise, signal); + } catch (error) { + if ( + error instanceof DOMException && + (error.name === "NotAllowedError" || error.name === "SecurityError") + ) { + throw new VoiceError("connection", "microphone-permission", requestId); + } + if (signal.aborted) throw error; + throw new VoiceError("connection", "microphone-device", requestId); + } + } + + async #requestRealtimeCall( + offerSdp: string, + signal: AbortSignal, + requestId: string, + ): Promise { + let response: Response; + try { + response = await this.#dependencies.fetch("/api/voice/realtime-call", { + body: offerSdp, + headers: { + "content-type": "application/sdp", + [VOICE_REQUEST_ID_HEADER]: requestId, + }, + method: "POST", + signal, + }); + } catch (error) { + if (signal.aborted) throw error; + throw new VoiceError("connection", "network", requestId); + } + if (!response.ok) { + await response.body?.cancel(); + throw voiceErrorFromResponse(response, "connection", requestId); + } + const contentType = response.headers + .get("content-type") + ?.split(";", 1)[0] + ?.trim() + .toLowerCase(); + if (contentType !== "application/sdp") { + await response.body?.cancel(); + throw new VoiceError("connection", "invalid-response", requestId); + } + return response; + } + + #installDataChannelListeners( + dataChannel: RTCDataChannel, + connectionEpoch: number, + ): void { + const messageListener = (event: MessageEvent) => { + if (this.#activeEpoch === connectionEpoch) { + this.#handleMessage(event, connectionEpoch); + } + }; + const unexpectedCloseListener = () => { + if (this.#activeEpoch === connectionEpoch && this.#connected) { + this.#handleConnectionFailure("network", "connection"); + } + }; + this.#messageListener = messageListener; + this.#unexpectedCloseListener = unexpectedCloseListener; + dataChannel.addEventListener("message", messageListener); + dataChannel.addEventListener("close", unexpectedCloseListener); + dataChannel.addEventListener("error", unexpectedCloseListener); + } + + #initializeMeter(audioContext: AudioContext, stream: MediaStream): void { const analyser = audioContext.createAnalyser(); analyser.fftSize = 256; - audioContext.createMediaStreamSource(mediaStream).connect(analyser); + audioContext.createMediaStreamSource(stream).connect(analyser); this.#analyser = analyser; this.#meterSamples = new Uint8Array(analyser.fftSize); } @@ -466,7 +1161,6 @@ export class OpenAIRealtimeSession { if (this.#meterFrame !== null || !this.#analyser || !this.#meterSamples) { return; } - const sample = () => { if ( !this.#microphoneTrack?.enabled || @@ -482,9 +1176,6 @@ export class OpenAIRealtimeSession { const normalized = (value - 128) / 128; squaredTotal += normalized * normalized; } - // The meter only drives a five-bar waveform, so two decimals is all the - // resolution a listener can use. Quantizing and skipping repeats keeps - // an every-animation-frame sample from re-rendering the interview. const level = Math.round( Math.min(1, Math.sqrt(squaredTotal / this.#meterSamples.length)) * @@ -497,14 +1188,11 @@ export class OpenAIRealtimeSession { this.#meterHasSample = true; this.#meterFrame = this.#dependencies.requestAnimationFrame(sample); }; - this.#meterFrame = this.#dependencies.requestAnimationFrame(sample); } #stopMeter(): void { - if (this.#meterFrame === null) { - return; - } + if (this.#meterFrame === null) return; this.#dependencies.cancelAnimationFrame(this.#meterFrame); this.#meterFrame = null; this.#meterLevel = 0; @@ -529,78 +1217,26 @@ export class OpenAIRealtimeSession { } } - #handleMessage(event: MessageEvent, connectionEpoch: number): void { - const parsed = parseRealtimeEvent(event.data); - if (!parsed || typeof parsed.type !== "string") { - return; - } - - if ( - parsed.type === "error" || - parsed.type === "conversation.item.input_audio_transcription.failed" - ) { - this.#handleConnectionFailure("invalid-response", "transcription"); - return; - } - - if (parsed.type === "input_audio_buffer.committed") { - if (typeof parsed.item_id !== "string" || !parsed.item_id) { - return; - } - this.#startTranscription(parsed.item_id); - this.setMicrophoneEnabled(false); - this.#emit({ - connectionEpoch, - itemId: parsed.item_id, - type: "input-committed", + #startTranscription(itemId: string): void { + if (!this.#transcriptionTimings.has(itemId)) { + this.#transcriptionTimings.set(itemId, { + requestId: + this.#dependencies.createRequestId?.() ?? createVoiceRequestId(), + startedAt: this.#now(), }); - return; - } - - const transcriptEventType = - parsed.type === "conversation.item.input_audio_transcription.delta" - ? "partial" - : parsed.type === - "conversation.item.input_audio_transcription.completed" - ? "completed" - : null; - if (transcriptEventType === null) { - return; - } - if ( - typeof parsed.item_id !== "string" || - !parsed.item_id || - !Number.isInteger(parsed.content_index) || - (parsed.content_index as number) < 0 - ) { - if (transcriptEventType === "completed") { - this.#handleConnectionFailure("invalid-response", "transcription"); - } - return; } + } - this.#startTranscription(parsed.item_id); - const text = - transcriptEventType === "partial" ? parsed.delta : parsed.transcript; - if (typeof text !== "string") { - if (transcriptEventType === "completed") { - this.#handleConnectionFailure("invalid-response", "transcription"); - } - return; - } - if (transcriptEventType === "completed") { - this.setMicrophoneEnabled(false); - this.#finishTranscription(parsed.item_id); - } - this.#emit({ - key: { - connectionEpoch, - contentIndex: parsed.content_index as number, - itemId: parsed.item_id, - }, - text, - type: transcriptEventType, - }); + #finishTranscription(itemId: string, errorCode?: VoiceErrorCode): void { + const timing = this.#transcriptionTimings.get(itemId); + if (!timing) return; + this.#transcriptionTimings.delete(itemId); + this.#reportDiagnostic( + "transcription", + timing.requestId, + timing.startedAt, + errorCode, + ); } #now(): number { @@ -623,25 +1259,6 @@ export class OpenAIRealtimeSession { }); } - #startTranscription(itemId: string): void { - if (!this.#transcriptionTimings.has(itemId)) { - this.#transcriptionTimings.set(itemId, { - requestId: - this.#dependencies.createRequestId?.() ?? createVoiceRequestId(), - startedAt: this.#now(), - }); - } - } - - #finishTranscription(itemId: string): void { - const timing = this.#transcriptionTimings.get(itemId); - if (!timing) { - return; - } - this.#transcriptionTimings.delete(itemId); - this.#reportDiagnostic("transcription", timing.requestId, timing.startedAt); - } - #releaseResources(): void { for (const timing of this.#transcriptionTimings.values()) { this.#reportDiagnostic( @@ -651,7 +1268,32 @@ export class OpenAIRealtimeSession { "request-aborted", ); } + for (const timing of [ + ...this.#pendingSpeechRequests.values(), + ...this.#speechTimings.values(), + ]) { + this.#reportDiagnostic( + "speech", + timing.requestId, + timing.startedAt, + "request-aborted", + ); + } this.#transcriptionTimings.clear(); + this.#activeResponseIds.clear(); + this.#cancelledCanonicalResponseIds.clear(); + this.#cancelledSpeechRequestIds.clear(); + this.#canonicalSpeechQueue.length = 0; + this.#completedResponseCancelEventIds.clear(); + this.#pendingClientEvents.clear(); + this.#pendingSpeechRequests.clear(); + this.#speechTimings.clear(); + this.#authorizedResponseIds.clear(); + this.#canonicalResponseIds.clear(); + this.#responseCreateEventId = null; + this.#responseTerminalSequence = 0; + this.#speakingResponseId = null; + this.#waitingForResponseTerminal = false; this.#activeEpoch = null; this.#connected = false; this.#connectedAt = null; @@ -680,13 +1322,19 @@ export class OpenAIRealtimeSession { if (this.#peerConnection) { this.#peerConnection.onconnectionstatechange = null; + this.#peerConnection.ontrack = null; this.#peerConnection.close(); this.#peerConnection = null; } + if (this.#remoteAudio) { + this.#remoteAudio.pause(); + this.#remoteAudio.srcObject = null; + this.#remoteAudio = null; + } + for (const stream of this.#remoteStreams) stopStream(stream); + this.#remoteStreams.clear(); if (this.#mediaStream) { - if (this.#microphoneTrack) { - this.#microphoneTrack.enabled = false; - } + if (this.#microphoneTrack) this.#microphoneTrack.enabled = false; stopStream(this.#mediaStream); this.#mediaStream = null; } @@ -698,35 +1346,25 @@ export class OpenAIRealtimeSession { signal: AbortSignal, requestId: string, ): Promise { - if (signal.aborted) { - return Promise.reject(signal.reason); - } - if (dataChannel.readyState === "open") { - return Promise.resolve(); - } + if (signal.aborted) return Promise.reject(signal.reason); + if (dataChannel.readyState === "open") return Promise.resolve(); if ( dataChannel.readyState === "closing" || dataChannel.readyState === "closed" ) { return Promise.reject(new VoiceError("connection", "network", requestId)); } - return new Promise((resolve, reject) => { const handleResult = (event: Event) => { dataChannel.removeEventListener("open", handleResult); dataChannel.removeEventListener("error", handleResult); dataChannel.removeEventListener("close", handleResult); signal.removeEventListener("abort", handleResult); - - if (event.type === "open") { - resolve(); - } else if (event.type === "abort") { - reject(signal.reason); - } else { - reject(new VoiceError("connection", "network", requestId)); - } + if (event.type === "open") resolve(); + else if (event.type === "abort") { + reject(new DOMException("Connection aborted", "AbortError")); + } else reject(new VoiceError("connection", "network", requestId)); }; - dataChannel.addEventListener("open", handleResult); dataChannel.addEventListener("error", handleResult); dataChannel.addEventListener("close", handleResult); diff --git a/apps/petrinaut-website/src/main/app/voice-interview/realtime-brunch-bridge.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/realtime-brunch-bridge.test.ts new file mode 100644 index 00000000000..a723f2dfe5a --- /dev/null +++ b/apps/petrinaut-website/src/main/app/voice-interview/realtime-brunch-bridge.test.ts @@ -0,0 +1,480 @@ +import { describe, expect, test, vi } from "vitest"; + +import { + createRealtimeSubmissionId, + RealtimeBrunchBridge, + type RealtimeBrunchBridgeEvent, +} from "./realtime-brunch-bridge"; + +import type { CanonicalSpeechSegment } from "./canonical-speech"; +import type { OpenAIRealtimeSessionEvent } from "./openai-realtime-session"; + +const segment = ( + id: string, + text: string, + source: CanonicalSpeechSegment["source"] = "brunch-ask", +): CanonicalSpeechSegment => ({ + contentHash: "fnv1a32:12345678", + id, + messageId: `message-${id}`, + partId: id, + source, + text, +}); + +const createHarness = () => { + let listener: ((event: OpenAIRealtimeSessionEvent) => void) | undefined; + const session = { + completeFunctionCall: vi.fn(), + speakCanonical: vi.fn(), + subscribe: vi.fn((next: (event: OpenAIRealtimeSessionEvent) => void) => { + listener = next; + return () => { + listener = undefined; + }; + }), + }; + const submitInterviewAnswer = vi.fn( + async (): Promise< + | { kind: "interactive-tool"; toolCallId: string } + | { kind: "message"; messageId: string } + > => ({ + kind: "interactive-tool", + toolCallId: "ask-current", + }), + ); + const bridge = new RealtimeBrunchBridge({ + session, + submitInterviewAnswer, + }); + const events: RealtimeBrunchBridgeEvent[] = []; + bridge.subscribe((event) => events.push(event)); + + return { + bridge, + emit: (event: OpenAIRealtimeSessionEvent) => listener?.(event), + events, + session, + submitInterviewAnswer, + }; +}; + +const toolDelta = ( + connectionEpoch: number, + delta: string, +): Extract => ({ + callId: "call-1", + connectionEpoch, + delta, + itemId: "function-item-1", + responseId: "response-1", + type: "tool-arguments-delta", +}); + +const toolDone = ( + connectionEpoch: number, + argumentsJson = '{"answer":"The supervisor approves it."}', +): Extract => ({ + arguments: argumentsJson, + callId: "call-1", + connectionEpoch, + itemId: "function-item-1", + name: "continue_interview", + responseId: "response-1", + type: "tool-arguments-done", +}); + +const responseTerminal = ( + connectionEpoch: number, + status: "cancelled" | "completed" | "failed" | "incomplete", + responseId = "response-1", +): Extract => ({ + connectionEpoch, + responseId, + status, + type: "response-terminal", +}); + +describe("RealtimeBrunchBridge", () => { + test("speaks the current canonical turn without replaying history", () => { + const harness = createHarness(); + const historical = segment( + "history", + "Do not replay this.", + "assistant-text", + ); + const preamble = { + ...segment("preamble", "Thanks. One more question.", "assistant-text"), + messageId: "message-current-turn", + }; + const question = { + ...segment("ask-current", "What happens after approval?"), + messageId: "message-current-turn", + }; + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [historical, preamble, question], + status: "ready", + }); + + harness.bridge.start(4); + + expect(harness.session.speakCanonical).toHaveBeenCalledOnce(); + expect(harness.session.speakCanonical).toHaveBeenCalledWith([ + preamble, + question, + ]); + }); + + test("streams and validates one tool call, preserves ask correlation, and waits for canonical Brunch output", async () => { + const harness = createHarness(); + const question = segment("ask-current", "What happens after approval?"); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [question], + status: "ready", + }); + harness.bridge.start(7); + harness.session.speakCanonical.mockClear(); + + harness.emit(toolDelta(7, '{"answer":"The supervisor')); + harness.emit(toolDelta(7, ' approves it."}')); + harness.emit(toolDone(7)); + + await vi.waitFor(() => + expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), + ); + expect(harness.submitInterviewAnswer).toHaveBeenCalledWith({ + id: createRealtimeSubmissionId(7, "call-1"), + text: "The supervisor approves it.", + }); + expect(harness.session.completeFunctionCall).not.toHaveBeenCalled(); + + harness.bridge.updateChat({ + canAcceptInterviewAnswer: false, + canonicalSegments: [question], + status: "streaming", + }); + const acknowledgement = segment( + "acknowledgement", + "Thanks. I have recorded that.", + "assistant-text", + ); + const nextQuestion = segment( + "ask-next", + "Who is informed next?", + "brunch-ask", + ); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [question, acknowledgement, nextQuestion], + status: "ready", + }); + + await vi.waitFor(() => + expect(harness.session.completeFunctionCall).toHaveBeenCalledWith( + "call-1", + [acknowledgement, nextQuestion], + ), + ); + expect(harness.events.map(({ type }) => type)).toEqual([ + "submission-started", + "submission-accepted", + "canonical-response-ready", + ]); + }); + + test("uses the first spoken turn to start Brunch when no question exists", async () => { + const harness = createHarness(); + harness.submitInterviewAnswer.mockResolvedValueOnce({ + kind: "message", + messageId: "message-kickoff", + }); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [], + status: "ready", + }); + harness.bridge.start(7); + + harness.emit(toolDone(7, '{"answer":"Battery charger workflow"}')); + + await vi.waitFor(() => + expect(harness.submitInterviewAnswer).toHaveBeenCalledWith({ + id: createRealtimeSubmissionId(7, "call-1"), + text: "Battery charger workflow", + }), + ); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: false, + canonicalSegments: [], + status: "submitted", + }); + const firstQuestion = segment( + "ask-first", + "What starts the battery charger workflow?", + ); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [firstQuestion], + status: "ready", + }); + + expect(harness.session.completeFunctionCall).toHaveBeenCalledWith( + "call-1", + [firstQuestion], + ); + expect(harness.events.map(({ type }) => type)).toEqual([ + "submission-started", + "submission-accepted", + "canonical-response-ready", + ]); + }); + + test("requires a correlated Brunch busy cycle before accepting new canonical segments", async () => { + const harness = createHarness(); + const question = segment("ask-current", "What happens after approval?"); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [question], + status: "ready", + }); + harness.bridge.start(7); + harness.emit(toolDone(7)); + await vi.waitFor(() => + expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), + ); + const unrelated = segment( + "unrelated", + "An unrelated canonical update.", + "assistant-text", + ); + + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [question, unrelated], + status: "ready", + }); + + expect(harness.session.completeFunctionCall).not.toHaveBeenCalled(); + + harness.bridge.updateChat({ + canAcceptInterviewAnswer: false, + canonicalSegments: [question, unrelated], + status: "submitted", + }); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [question, unrelated], + status: "ready", + }); + + expect(harness.session.completeFunctionCall).toHaveBeenCalledWith( + "call-1", + [unrelated], + ); + }); + + test("rejects streamed arguments whose response or item identity changes", async () => { + const harness = createHarness(); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [segment("ask-current", "Question")], + status: "ready", + }); + harness.bridge.start(3); + harness.emit(toolDelta(3, '{"answer":"Answer"}')); + + harness.emit({ + ...toolDone(3, '{"answer":"Answer"}'), + responseId: "response-2", + }); + await Promise.resolve(); + + expect(harness.submitInterviewAnswer).not.toHaveBeenCalled(); + expect(harness.events).toEqual([ + expect.objectContaining({ + code: "interview-correlation", + type: "error", + }), + ]); + }); + + test("rejects concurrent argument streams before either can submit", async () => { + const harness = createHarness(); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [segment("ask-current", "Question")], + status: "ready", + }); + harness.bridge.start(3); + harness.emit(toolDelta(3, '{"answer":"First"}')); + + harness.emit({ + ...toolDelta(3, '{"answer":"Second"}'), + callId: "call-2", + itemId: "function-item-2", + }); + + expect(harness.submitInterviewAnswer).not.toHaveBeenCalled(); + expect(harness.events).toEqual([ + expect.objectContaining({ type: "error" }), + ]); + }); + + test("discards a cancelled argument stream without poisoning the next answer", async () => { + const harness = createHarness(); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [segment("ask-current", "Question")], + status: "ready", + }); + harness.bridge.start(3); + harness.emit(toolDelta(3, '{"answer":"Cancelled"}')); + harness.emit(responseTerminal(3, "cancelled")); + harness.emit(toolDone(3, '{"answer":"Cancelled"}')); + + harness.emit({ + ...toolDelta(3, '{"answer":"Accepted"}'), + callId: "call-2", + itemId: "function-item-2", + responseId: "response-2", + }); + harness.emit({ + ...toolDone(3, '{"answer":"Accepted"}'), + callId: "call-2", + itemId: "function-item-2", + responseId: "response-2", + }); + + await vi.waitFor(() => + expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), + ); + expect(harness.submitInterviewAnswer).toHaveBeenCalledWith({ + id: createRealtimeSubmissionId(3, "call-2"), + text: "Accepted", + }); + expect(harness.events).not.toContainEqual( + expect.objectContaining({ type: "error" }), + ); + }); + + test("rejects an unfinished argument stream from a completed response", () => { + const harness = createHarness(); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [segment("ask-current", "Question")], + status: "ready", + }); + harness.bridge.start(3); + harness.emit(toolDelta(3, '{"answer":"Incomplete')); + + harness.emit(responseTerminal(3, "completed")); + + expect(harness.submitInterviewAnswer).not.toHaveBeenCalled(); + expect(harness.events).toEqual([ + expect.objectContaining({ + code: "interview-correlation", + type: "error", + }), + ]); + }); + + test("rejects duplicate, stale, overlapping, and malformed calls without another Brunch submission", async () => { + const harness = createHarness(); + const question = segment("ask-current", "What happens after approval?"); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [question], + status: "ready", + }); + harness.bridge.start(2); + + harness.emit(toolDone(1)); + harness.emit(toolDelta(2, '{"answer":"The supervisor approves it."}')); + harness.emit(toolDone(2)); + harness.emit(toolDone(2)); + await vi.waitFor(() => + expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), + ); + + harness.emit({ + ...toolDone(2, '{"answer":"Overlapping"}'), + callId: "call-2", + itemId: "function-item-2", + }); + + expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(); + expect(harness.events.at(-1)).toMatchObject({ type: "error" }); + }); + + test.each([ + ["wrong tool", { ...toolDone(3), name: "invent_question" }], + ["invalid JSON", toolDone(3, "not-json")], + ["extra property", toolDone(3, '{"answer":"Valid","extra":true}')], + ["empty answer", toolDone(3, '{"answer":" "}')], + ])("rejects %s arguments", async (_label, event) => { + const harness = createHarness(); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [segment("ask-current", "Question")], + status: "ready", + }); + harness.bridge.start(3); + + harness.emit(event); + await Promise.resolve(); + + expect(harness.submitInterviewAnswer).not.toHaveBeenCalled(); + expect(harness.events).toEqual([ + expect.objectContaining({ type: "error" }), + ]); + }); + + test("rejects a composer result that does not match the pending brunch_ask", async () => { + const harness = createHarness(); + harness.submitInterviewAnswer.mockResolvedValueOnce({ + kind: "interactive-tool", + toolCallId: "another-ask", + }); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [segment("ask-current", "Question")], + status: "ready", + }); + harness.bridge.start(5); + + harness.emit(toolDone(5)); + + await vi.waitFor(() => + expect(harness.events.at(-1)).toMatchObject({ type: "error" }), + ); + expect(harness.session.completeFunctionCall).not.toHaveBeenCalled(); + }); + + test("speaks new canonical text turns without creating a Realtime tool result", () => { + const harness = createHarness(); + const question = segment("ask-current", "Question"); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [question], + status: "ready", + }); + harness.bridge.start(8); + harness.session.speakCanonical.mockClear(); + const response = segment( + "typed-response", + "Canonical response", + "assistant-text", + ); + + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [question, response], + status: "ready", + }); + + expect(harness.session.speakCanonical).toHaveBeenCalledWith([response]); + expect(harness.session.completeFunctionCall).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/petrinaut-website/src/main/app/voice-interview/realtime-brunch-bridge.ts b/apps/petrinaut-website/src/main/app/voice-interview/realtime-brunch-bridge.ts new file mode 100644 index 00000000000..1a84fa74c62 --- /dev/null +++ b/apps/petrinaut-website/src/main/app/voice-interview/realtime-brunch-bridge.ts @@ -0,0 +1,427 @@ +import type { CanonicalSpeechSegment } from "./canonical-speech"; +import type { OpenAIRealtimeSessionEvent } from "./openai-realtime-session"; + +type ChatStatus = "ready" | "submitted" | "streaming" | "error"; + +interface ChatUpdate { + readonly canAcceptInterviewAnswer: boolean; + readonly canonicalSegments: CanonicalSpeechSegment[]; + readonly status: ChatStatus; +} + +interface RealtimeBridgeSession { + completeFunctionCall( + callId: string, + segments: CanonicalSpeechSegment[], + ): void; + speakCanonical(segments: CanonicalSpeechSegment[]): void; + subscribe(listener: (event: OpenAIRealtimeSessionEvent) => void): () => void; +} + +interface SubmitInterviewAnswerInput { + readonly id: string; + readonly text: string; +} + +type SubmitInterviewAnswerResult = + | { readonly kind: "interactive-tool"; readonly toolCallId: string } + | { readonly kind: "message"; readonly messageId: string }; + +interface RealtimeBrunchBridgeDependencies { + readonly session: RealtimeBridgeSession; + readonly submitInterviewAnswer: ( + input: SubmitInterviewAnswerInput, + ) => Promise; +} + +interface ActiveSubmission { + readonly baselineSegmentIds: ReadonlySet; + readonly callId: string; + readonly epoch: number; + readonly pendingQuestionId: string | null; + correlated: boolean; + sawBusyChatStatus: boolean; +} + +interface ArgumentStream { + readonly chunks: string[]; + readonly itemId: string; + readonly responseId: string; +} + +export type RealtimeBridgeErrorCode = + | "interview-correlation" + | "interview-response" + | "interview-submission"; + +export type RealtimeBrunchBridgeEvent = + | { + readonly answer: string; + readonly callId: string; + readonly type: "submission-started"; + } + | { + readonly answer: string; + readonly callId: string; + readonly type: "submission-accepted"; + } + | { + readonly callId: string; + readonly segments: CanonicalSpeechSegment[]; + readonly type: "canonical-response-ready"; + } + | { + readonly code: RealtimeBridgeErrorCode; + readonly message: string; + readonly type: "error"; + }; + +type BridgeListener = (event: RealtimeBrunchBridgeEvent) => void; + +const INVALID_BRIDGE_EVENT = + "The voice response could not be matched to the interview. Reconnect voice or use text instead."; +const ANSWER_LIMIT = 32_000; + +export const createRealtimeSubmissionId = ( + connectionEpoch: number, + callId: string, +): string => `voice-realtime:${connectionEpoch}:${encodeURIComponent(callId)}`; + +const latestPendingQuestion = ( + segments: CanonicalSpeechSegment[], +): CanonicalSpeechSegment | undefined => + segments.findLast(({ source }) => source === "brunch-ask"); + +const parseContinueInterviewArguments = ( + argumentsJson: string, +): string | null => { + try { + const value: unknown = JSON.parse(argumentsJson); + if (typeof value !== "object" || value === null || Array.isArray(value)) { + return null; + } + const record = value as Record; + if (Object.keys(record).length !== 1 || typeof record.answer !== "string") { + return null; + } + const answer = record.answer.trim(); + return answer && Array.from(answer).length <= ANSWER_LIMIT ? answer : null; + } catch { + return null; + } +}; + +export class RealtimeBrunchBridge { + readonly #argumentDeltas = new Map(); + readonly #listeners = new Set(); + readonly #processedCalls = new Set(); + readonly #session: RealtimeBridgeSession; + readonly #submitInterviewAnswer: ( + input: SubmitInterviewAnswerInput, + ) => Promise; + readonly #seenSegmentIds = new Set(); + readonly #terminalResponseIds = new Set(); + #activeEpoch: number | null = null; + #activeSubmission: ActiveSubmission | null = null; + #chat: ChatUpdate = { + canAcceptInterviewAnswer: false, + canonicalSegments: [], + status: "ready", + }; + #generation = 0; + + public constructor({ + session, + submitInterviewAnswer, + }: RealtimeBrunchBridgeDependencies) { + this.#session = session; + this.#submitInterviewAnswer = submitInterviewAnswer; + session.subscribe((event) => this.#handleSessionEvent(event)); + } + + public subscribe(listener: BridgeListener): () => void { + this.#listeners.add(listener); + return () => this.#listeners.delete(listener); + } + + public start(connectionEpoch: number): void { + ++this.#generation; + this.#activeEpoch = connectionEpoch; + this.#activeSubmission = null; + this.#argumentDeltas.clear(); + this.#processedCalls.clear(); + this.#seenSegmentIds.clear(); + this.#terminalResponseIds.clear(); + for (const segment of this.#chat.canonicalSegments) { + this.#seenSegmentIds.add(segment.id); + } + + const question = latestPendingQuestion(this.#chat.canonicalSegments); + if (question) { + this.#session.speakCanonical( + this.#chat.canonicalSegments.filter( + ({ messageId }) => messageId === question.messageId, + ), + ); + } + } + + public stop(): void { + ++this.#generation; + this.#activeEpoch = null; + this.#activeSubmission = null; + this.#argumentDeltas.clear(); + this.#terminalResponseIds.clear(); + } + + public updateChat(update: ChatUpdate): void { + this.#chat = update; + if (this.#activeEpoch === null) { + return; + } + if (update.status === "error") { + this.#fail( + "The interview could not complete that turn. Use the composer to retry.", + "interview-response", + ); + return; + } + if (this.#activeSubmission) { + if (update.status === "submitted" || update.status === "streaming") { + this.#activeSubmission.sawBusyChatStatus = true; + } + this.#completeCorrelatedSubmission(); + return; + } + if (update.status !== "ready") { + return; + } + + const newSegments = update.canonicalSegments.filter( + ({ id }) => !this.#seenSegmentIds.has(id), + ); + if (newSegments.length === 0) { + return; + } + try { + this.#session.speakCanonical(newSegments); + for (const segment of newSegments) { + this.#seenSegmentIds.add(segment.id); + } + } catch { + this.#fail(INVALID_BRIDGE_EVENT); + } + } + + #emit(event: RealtimeBrunchBridgeEvent): void { + for (const listener of this.#listeners) { + listener(event); + } + } + + #fail( + message: string, + code: RealtimeBridgeErrorCode = "interview-correlation", + ): void { + ++this.#generation; + this.#activeSubmission = null; + this.#argumentDeltas.clear(); + this.#emit({ code, message, type: "error" }); + } + + #handleSessionEvent(event: OpenAIRealtimeSessionEvent): void { + if ( + !("connectionEpoch" in event) || + event.connectionEpoch !== this.#activeEpoch + ) { + return; + } + if (event.type === "response-terminal") { + this.#handleResponseTerminal(event); + return; + } + if ( + event.type !== "tool-arguments-delta" && + event.type !== "tool-arguments-done" + ) { + return; + } + + const responseKey = `${event.connectionEpoch}:${event.responseId}`; + if (this.#terminalResponseIds.has(responseKey)) { + return; + } + const callKey = `${event.connectionEpoch}:${event.callId}`; + if (this.#processedCalls.has(callKey)) { + return; + } + if (event.type === "tool-arguments-delta") { + const stream = this.#argumentDeltas.get(callKey); + if (!stream && this.#argumentDeltas.size > 0) { + this.#processedCalls.add(callKey); + this.#fail(INVALID_BRIDGE_EVENT); + return; + } + if ( + stream && + (stream.itemId !== event.itemId || + stream.responseId !== event.responseId) + ) { + this.#processedCalls.add(callKey); + this.#fail(INVALID_BRIDGE_EVENT); + return; + } + if (stream) { + stream.chunks.push(event.delta); + } else { + this.#argumentDeltas.set(callKey, { + chunks: [event.delta], + itemId: event.itemId, + responseId: event.responseId, + }); + } + return; + } + + this.#processedCalls.add(callKey); + const stream = this.#argumentDeltas.get(callKey); + if (!stream && this.#argumentDeltas.size > 0) { + this.#fail(INVALID_BRIDGE_EVENT); + return; + } + this.#argumentDeltas.delete(callKey); + if ( + this.#activeSubmission || + event.name !== "continue_interview" || + (stream !== undefined && + (stream.itemId !== event.itemId || + stream.responseId !== event.responseId || + stream.chunks.join("") !== event.arguments)) + ) { + this.#fail(INVALID_BRIDGE_EVENT); + return; + } + + const answer = parseContinueInterviewArguments(event.arguments); + const question = latestPendingQuestion(this.#chat.canonicalSegments); + if ( + !answer || + !this.#chat.canAcceptInterviewAnswer || + (!question && this.#chat.status !== "ready") + ) { + this.#fail(INVALID_BRIDGE_EVENT); + return; + } + + const generation = this.#generation; + this.#activeSubmission = { + baselineSegmentIds: new Set( + this.#chat.canonicalSegments.map(({ id }) => id), + ), + callId: event.callId, + correlated: false, + epoch: event.connectionEpoch, + pendingQuestionId: question?.partId ?? null, + sawBusyChatStatus: false, + }; + this.#emit({ answer, callId: event.callId, type: "submission-started" }); + void this.#submit(event, answer, generation); + } + + #handleResponseTerminal( + event: Extract, + ): void { + const responseKey = `${event.connectionEpoch}:${event.responseId}`; + const matchingStreams = [...this.#argumentDeltas].filter( + ([, stream]) => stream.responseId === event.responseId, + ); + if (event.status === "completed" && matchingStreams.length > 0) { + this.#fail(INVALID_BRIDGE_EVENT); + return; + } + + for (const [callKey] of matchingStreams) { + this.#argumentDeltas.delete(callKey); + this.#processedCalls.add(callKey); + } + this.#terminalResponseIds.add(responseKey); + } + + async #submit( + event: Extract, + answer: string, + generation: number, + ): Promise { + try { + const result = await this.#submitInterviewAnswer({ + id: createRealtimeSubmissionId(event.connectionEpoch, event.callId), + text: answer, + }); + const active = this.#activeSubmission; + if ( + generation !== this.#generation || + !active || + active.callId !== event.callId || + active.epoch !== event.connectionEpoch + ) { + return; + } + const resultMatchesSubmission = + active.pendingQuestionId === null + ? result.kind === "message" + : result.kind === "interactive-tool" && + result.toolCallId === active.pendingQuestionId; + if (!resultMatchesSubmission) { + this.#fail(INVALID_BRIDGE_EVENT); + return; + } + active.correlated = true; + this.#emit({ + answer, + callId: event.callId, + type: "submission-accepted", + }); + this.#completeCorrelatedSubmission(); + } catch { + if (generation === this.#generation) { + this.#fail( + "The interview could not accept that answer. Use the composer to retry.", + "interview-submission", + ); + } + } + } + + #completeCorrelatedSubmission(): void { + const active = this.#activeSubmission; + if ( + !active?.correlated || + !active.sawBusyChatStatus || + this.#chat.status !== "ready" + ) { + return; + } + const responseSegments = this.#chat.canonicalSegments.filter( + ({ id }) => !active.baselineSegmentIds.has(id), + ); + if (responseSegments.length === 0) { + return; + } + + try { + this.#session.completeFunctionCall(active.callId, responseSegments); + } catch { + this.#fail(INVALID_BRIDGE_EVENT); + return; + } + for (const segment of responseSegments) { + this.#seenSegmentIds.add(segment.id); + } + this.#activeSubmission = null; + this.#emit({ + callId: active.callId, + segments: responseSegments, + type: "canonical-response-ready", + }); + } +} 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 deleted file mode 100644 index 3e6158449a9..00000000000 --- a/apps/petrinaut-website/src/main/app/voice-interview/speech-playback-controller.test.ts +++ /dev/null @@ -1,326 +0,0 @@ -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 reportDiagnostic = vi.fn(); - const revokeObjectURL = vi.fn(); - const controller = new SpeechPlaybackController({ - createAudio, - createObjectURL, - createRequestId: () => "voice-speech-request", - fetch, - now: () => 100, - reportDiagnostic, - revokeObjectURL, - }); - return { - audio, - controller, - createAudio, - createObjectURL, - fetch, - reportDiagnostic, - 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", - "x-request-id": "voice-speech-request", - }, - 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", - ); - expect(harness.reportDiagnostic.mock.calls).toEqual([ - [ - { - durationMs: 0, - operation: "speech", - outcome: "success", - requestId: "voice-speech-request", - stage: "browser", - }, - ], - [ - { - durationMs: 0, - operation: "speech", - outcome: "success", - requestId: "voice-speech-request", - stage: "playback", - }, - ], - ]); - expect(JSON.stringify(harness.reportDiagnostic.mock.calls)).not.toContain( - segment.text, - ); - }); - - 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 speech service returned an invalid response. Read the visible response instead.", - ); - await expect(harness.controller.play(segment)).rejects.toThrow( - "The speech service returned an invalid response. Read the visible response instead.", - ); - await expect(harness.controller.play(segment)).rejects.toThrow( - "The speech service returned an invalid response. Read the visible response instead.", - ); - expect(harness.createAudio).not.toHaveBeenCalled(); - }); - - test("preserves an upstream response-body abort as a request abort", async () => { - const response = new Response(new Uint8Array([1]), { - headers: { "content-type": "audio/mpeg" }, - }); - vi.spyOn(response, "blob").mockRejectedValue( - new DOMException("upstream aborted", "AbortError"), - ); - const fetch = vi.fn(async () => response); - const harness = createHarness(fetch); - - await expect(harness.controller.play(segment)).rejects.toMatchObject({ - code: "request-aborted", - requestId: "voice-speech-request", - }); - expect(harness.createAudio).not.toHaveBeenCalled(); - expect(harness.reportDiagnostic).toHaveBeenLastCalledWith( - expect.objectContaining({ - errorCode: "request-aborted", - outcome: "aborted", - requestId: "voice-speech-request", - stage: "browser", - }), - ); - }); - - test("rejects untrusted server-only diagnostics and request references", async () => { - const fetch = vi.fn( - async () => - new Response("private provider response", { - headers: { - "x-petrinaut-voice-error": "microphone-permission", - "x-request-id": "private transcript used as a request id", - }, - status: 502, - }), - ); - const harness = createHarness(fetch); - - await expect(harness.controller.play(segment)).rejects.toMatchObject({ - code: "invalid-response", - requestId: "voice-speech-request", - }); - expect(JSON.stringify(harness.reportDiagnostic.mock.calls)).not.toContain( - "private transcript used as a request id", - ); - expect(JSON.stringify(harness.reportDiagnostic.mock.calls)).not.toContain( - "private provider response", - ); - }); - - 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 speech service returned an invalid response. Read the visible response 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(); - expect(harness.reportDiagnostic).toHaveBeenCalledWith( - expect.objectContaining({ - errorCode: "request-aborted", - outcome: "aborted", - requestId: "voice-speech-request", - stage: "browser", - }), - ); - }); - - 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", - ); - expect(harness.reportDiagnostic).toHaveBeenLastCalledWith( - expect.objectContaining({ - errorCode: "request-aborted", - outcome: "aborted", - stage: "playback", - }), - ); - }); - - 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 speech service returned an invalid response. Read the visible response 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 speech service returned an invalid response. Read the visible response instead.", - ); - expect(secondHarness.revokeObjectURL).toHaveBeenCalledOnce(); - }); - - test("revokes the object URL when the audio element cannot be created", async () => { - const harness = createHarness(); - harness.createAudio.mockImplementationOnce(() => { - throw new Error("audio construction failed"); - }); - - await expect(harness.controller.play(segment)).rejects.toMatchObject({ - code: "invalid-response", - requestId: "voice-speech-request", - }); - - expect(harness.revokeObjectURL).toHaveBeenCalledWith( - "blob:canonical-speech", - ); - }); - - test("classifies browser network failures without leaking diagnostics", async () => { - const fetch = vi.fn(async () => { - throw new Error("private browser network detail"); - }); - const harness = createHarness(fetch); - - await expect(harness.controller.play(segment)).rejects.toMatchObject({ - code: "network", - message: - "The speech service could not be reached. Read the visible response instead.", - requestId: "voice-speech-request", - }); - expect(JSON.stringify(harness.reportDiagnostic.mock.calls)).not.toContain( - "private browser network detail", - ); - }); -}); 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 deleted file mode 100644 index 5d29d0d6be9..00000000000 --- a/apps/petrinaut-website/src/main/app/voice-interview/speech-playback-controller.ts +++ /dev/null @@ -1,279 +0,0 @@ -import { - createVoiceRequestId, - VoiceError, - VOICE_REQUEST_ID_HEADER, - voiceDiagnosticOutcome, - voiceDurationMs, - voiceErrorFromResponse, - type VoiceDiagnosticReporter, - type VoiceErrorCode, -} from "../../../voice-diagnostics"; -import { - hashCanonicalSpeechText, - type CanonicalSpeechSegment, -} from "./canonical-speech"; - -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 createRequestId?: () => string; - readonly fetch: typeof globalThis.fetch; - readonly now?: () => number; - readonly reportDiagnostic?: VoiceDiagnosticReporter; - readonly revokeObjectURL: (url: string) => void; -} - -interface SpeechPlaybackEvents { - readonly onPlaying?: () => void; -} - -interface ActiveAudio { - cancel(reason: DOMException): void; - readonly generation: number; -} - -const fallbackError = (requestId: string): VoiceError => - new VoiceError("speech", "invalid-response", requestId); -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(); - const requestId = - this.#dependencies.createRequestId?.() ?? createVoiceRequestId(); - const requestStartedAt = this.#now(); - let requestReported = false; - let playbackStartedAt: number | null = null; - if ( - segment.contentHash !== hashCanonicalSpeechText(segment.text) || - !segment.id.endsWith(`:${segment.contentHash}`) - ) { - this.#reportDiagnostic( - "browser", - requestId, - requestStartedAt, - "invalid-response", - ); - throw fallbackError(requestId); - } - const generation = this.#generation; - const abortController = new AbortController(); - this.#abortController = abortController; - - try { - let response: Response; - try { - 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", - [VOICE_REQUEST_ID_HEADER]: requestId, - }, - method: "POST", - signal: abortController.signal, - }), - abortController.signal, - ); - } catch (error) { - if (isAbortError(error)) { - throw error; - } - throw new VoiceError("speech", "network", requestId); - } - if (!response.ok) { - await response.body?.cancel(); - throw voiceErrorFromResponse(response, "speech", requestId); - } - const contentType = response.headers - .get("content-type") - ?.split(";", 1)[0] - ?.trim() - .toLowerCase(); - if (contentType !== "audio/mpeg") { - await response.body?.cancel(); - throw fallbackError(requestId); - } - - let blob: Blob; - try { - blob = await waitForAbort(response.blob(), abortController.signal); - } catch (error) { - if (isAbortError(error)) { - if (abortController.signal.aborted) { - throw error; - } - throw new VoiceError("speech", "request-aborted", requestId); - } - throw new VoiceError("speech", "network", requestId); - } - if (generation !== this.#generation || blob.size === 0) { - throw generation === this.#generation - ? fallbackError(requestId) - : abortError(); - } - this.#reportDiagnostic("browser", requestId, requestStartedAt); - requestReported = true; - playbackStartedAt = this.#now(); - - const objectUrl = this.#dependencies.createObjectURL(blob); - let audio: SpeechAudio; - try { - audio = this.#dependencies.createAudio(objectUrl); - } catch { - this.#dependencies.revokeObjectURL(objectUrl); - throw fallbackError(requestId); - } - 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(() => { - this.#reportDiagnostic("playback", requestId, playbackStartedAt!); - resolve(); - }); - const handleError = () => - settle(() => reject(fallbackError(requestId))); - 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) { - const errorCode = isAbortError(error) - ? "request-aborted" - : error instanceof VoiceError - ? error.code - : "invalid-response"; - if (!requestReported) { - this.#reportDiagnostic( - "browser", - requestId, - requestStartedAt, - errorCode, - ); - } else if (playbackStartedAt !== null) { - this.#reportDiagnostic( - "playback", - requestId, - playbackStartedAt, - errorCode, - ); - } - if (isAbortError(error)) { - throw error; - } - throw error instanceof VoiceError ? error : fallbackError(requestId); - } finally { - if (this.#abortController === abortController) { - this.#abortController = null; - } - } - } - - #now(): number { - return this.#dependencies.now?.() ?? performance.now(); - } - - #reportDiagnostic( - stage: "browser" | "playback", - requestId: string, - startedAt: number, - errorCode?: VoiceErrorCode, - ): void { - this.#dependencies.reportDiagnostic?.({ - durationMs: voiceDurationMs(startedAt, this.#now()), - ...(errorCode === undefined ? {} : { errorCode }), - operation: "speech", - outcome: voiceDiagnosticOutcome(errorCode), - requestId, - stage, - }); - } - - 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 8bc0dc29fb9..7ee6d71f1d9 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 @@ -10,7 +10,7 @@ import { } from "@testing-library/react"; import { StrictMode, useState } from "react"; import { renderToStaticMarkup } from "react-dom/server"; -import { afterEach, describe, expect, test, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { OpenAIRealtimeSession } from "./openai-realtime-session"; import { @@ -28,17 +28,19 @@ import type { PetrinautAiInterviewStageContext } from "@hashintel/petrinaut/ui"; const snapshot = { canReviseLastAnswer: false, + connection: "connected", currentQuestion: "What happens after approval?", errorCode: null, errorMessage: "", errorRequestId: "", + input: "listening", lastAnswerDelivery: "none" as const, lastCommittedText: "", microphoneEnabled: true, microphoneLevel: 0.24, + output: "idle", partialText: "The request goes to", - phase: "listening" as const, -}; +} satisfies VoiceTurnSnapshot; const config = { available: true as const, connectionTimeoutMs: 15_000 }; @@ -53,15 +55,12 @@ const viewProps = ( onCheckMicrophone: vi.fn(), onConsentChange: vi.fn(), onCorrectionChange: vi.fn(), - onDoneSpeaking: vi.fn(), onEdit: vi.fn(), onEnd: vi.fn(), onExpand: vi.fn(), - onInterrupt: vi.fn(), onMinimize: vi.fn(), onPause: vi.fn(), onReconnect: vi.fn(), - onRedo: vi.fn(), onResume: vi.fn(), onStart: vi.fn(), onSubmitCorrection: vi.fn(), @@ -156,10 +155,27 @@ const stubUnavailableMicrophone = () => { return getUserMedia; }; +beforeEach(() => { + const values = new Map(); + Object.defineProperty(window, "localStorage", { + configurable: true, + value: { + clear: () => values.clear(), + getItem: (key: string) => values.get(key) ?? null, + key: (index: number) => [...values.keys()][index] ?? null, + get length() { + return values.size; + }, + removeItem: (key: string) => values.delete(key), + setItem: (key: string, value: string) => values.set(key, value), + } satisfies Storage, + }); +}); + afterEach(() => { cleanup(); - vi.unstubAllGlobals(); window.localStorage.clear(); + vi.unstubAllGlobals(); }); describe("voice interview stage", () => { @@ -233,7 +249,7 @@ describe("voice interview stage", () => { expect(html).toContain("Voice interview"); expect(html).toContain("Talk through your process with AI"); - expect(html).toContain("transcribed by OpenAI"); + expect(html).toContain("OpenAI processes live audio"); expect(html).toContain("keeps finalized answers"); expect(html).toContain("not the audio"); expect(html.indexOf("Start interview")).toBeLessThan( @@ -259,14 +275,16 @@ describe("voice interview stage", () => { {...viewProps({ snapshot: { ...snapshot, + connection: "error", errorCode: "microphone-permission", errorMessage: "Allow microphone access in your browser settings, then reconnect voice input.", errorRequestId: "voice-request-permission", + input: "paused", microphoneEnabled: false, microphoneLevel: 0, + output: "idle", partialText: "", - phase: "recoverable-error", }, })} />, @@ -522,13 +540,32 @@ describe("voice interview stage", () => { expect(html).toContain("Live transcript"); expect(html).toContain("Listening"); expect(html).toContain("Microphone input level: Medium"); - expect(html).toContain('aria-label="Done speaking"'); + expect(html).not.toContain('aria-label="Done speaking"'); + expect(html).not.toContain('aria-label="Interrupt and speak"'); expect(html).toContain("motionReduce:vis_hidden"); expect(html).toContain("pos_relative"); expect(html).not.toContain("pos_fixed"); expect(html).toContain('aria-live="polite"'); }); + test("invites a spoken kickoff when no interview question exists", () => { + render( + , + ); + + expect( + screen.getByText("Tell me about the process you want to model."), + ).not.toBeNull(); + }); + test("centers a circular microphone and waveform without visible level copy", () => { render(); @@ -560,10 +597,7 @@ describe("voice interview stage", () => { ...snapshot, lastAnswerDelivery: "delivered", lastCommittedText: "The shift lead assigns an owner.", - microphoneEnabled: false, - microphoneLevel: 0, partialText: "", - phase: "waiting", }, })} />, @@ -580,12 +614,11 @@ describe("voice interview stage", () => { {...viewProps({ snapshot: { ...snapshot, + input: "submitting", lastAnswerDelivery: "pending", lastCommittedText: "The shift lead assigns an owner.", - microphoneEnabled: false, - microphoneLevel: 0, + output: "waiting-for-tool", partialText: "", - phase: "delivering", }, })} />, @@ -602,15 +635,17 @@ describe("voice interview stage", () => { {...viewProps({ snapshot: { ...snapshot, + connection: "error", errorCode: null, errorMessage: "The interview could not accept that answer. Use the composer to retry.", + input: "paused", lastAnswerDelivery: "failed", lastCommittedText: "The shift lead assigns an owner.", microphoneEnabled: false, microphoneLevel: 0, + output: "idle", partialText: "", - phase: "recoverable-error", }, })} />, @@ -624,31 +659,26 @@ describe("voice interview stage", () => { test("uses voice-app icon controls while listening", () => { render(); - for (const name of ["Use text instead", "Done speaking", "Pause"]) { + for (const name of ["Use text instead", "Pause"]) { const button = screen.getByRole("button", { name }); expect(button.querySelector("svg")).not.toBeNull(); expect(button.parentElement?.getAttribute("data-scope")).toBe("tooltip"); } - + expect(screen.queryByRole("button", { name: "Done speaking" })).toBeNull(); expect( - screen - .getByRole("button", { name: "Done speaking" }) - .textContent.replaceAll("\u200B", "") - .trim(), - ).toBe(""); + screen.queryByRole("button", { name: "Interrupt and speak" }), + ).toBeNull(); }); - test("orders the full listening actions as keyboard, done speaking, then pause", () => { + test("orders the full listening actions as keyboard then pause", () => { const html = renderToStaticMarkup( , ); expect(html.indexOf('aria-label="Use text instead"')).toBeLessThan( - html.indexOf('aria-label="Done speaking"'), - ); - expect(html.indexOf('aria-label="Done speaking"')).toBeLessThan( html.indexOf('aria-label="Pause"'), ); + expect(html).not.toContain('aria-label="Done speaking"'); }); test("offers only resume and keyboard actions while paused", () => { @@ -657,10 +687,10 @@ describe("voice interview stage", () => { {...viewProps({ snapshot: { ...snapshot, + input: "paused", microphoneEnabled: false, microphoneLevel: 0, partialText: "", - phase: "paused", }, })} />, @@ -680,28 +710,34 @@ describe("voice interview stage", () => { ).toBeNull(); }); - test("shows the waveform only while the microphone is listening", () => { + test("shows the waveform whenever the continuous microphone is active", () => { const rendered = render(); expect(screen.getByTestId("voice-waveform")).not.toBeNull(); - for (const phase of ["paused", "playing", "waiting"] as const) { - rendered.rerender( - , - ); - expect(screen.queryByTestId("voice-waveform")).toBeNull(); - expect(screen.queryByText(/Microphone input level:/u)).toBeNull(); - } + rendered.rerender( + , + ); + expect(screen.getByTestId("voice-waveform")).not.toBeNull(); + expect(screen.getByText("Interviewer speaking")).not.toBeNull(); + + rendered.rerender( + , + ); + expect(screen.queryByTestId("voice-waveform")).toBeNull(); + expect(screen.queryByText(/Microphone input level:/u)).toBeNull(); }); test("keeps reconnect visible and makes secondary recovery icon-only", () => { @@ -710,14 +746,16 @@ describe("voice interview stage", () => { {...viewProps({ snapshot: { ...snapshot, + connection: "error", errorCode: "microphone-permission", errorMessage: "Allow microphone access in your browser settings, then reconnect voice input.", errorRequestId: "voice-request-permission", + input: "paused", microphoneEnabled: false, microphoneLevel: 0, + output: "idle", partialText: "", - phase: "recoverable-error", }, })} />, @@ -746,12 +784,14 @@ describe("voice interview stage", () => { {...viewProps({ snapshot: { ...snapshot, + connection: "error", errorCode, errorMessage, + input: "paused", microphoneEnabled: false, microphoneLevel: 0, + output: "idle", partialText: "", - phase: "recoverable-error", }, })} /> @@ -807,12 +847,7 @@ describe("voice interview stage", () => { test("renders icons for the listening controls", () => { render(); - for (const name of [ - "Minimize voice interview", - "End interview", - "Done speaking", - "Pause", - ]) { + for (const name of ["Minimize voice interview", "End interview", "Pause"]) { expect( screen.getByRole("button", { name }).querySelector("svg"), ).not.toBeNull(); @@ -825,10 +860,10 @@ describe("voice interview stage", () => { {...viewProps({ snapshot: { ...snapshot, + input: "paused", microphoneEnabled: false, microphoneLevel: 0, partialText: "", - phase: "waiting", }, })} />, @@ -839,17 +874,14 @@ describe("voice interview stage", () => { ); }); - test("renders committed repair actions separately from pause, minimize, and end", () => { + test("renders committed edit separately from pause, minimize, and end", () => { const html = renderToStaticMarkup( , @@ -858,7 +890,6 @@ describe("voice interview stage", () => { for (const name of [ "Minimize voice interview", "End interview", - "Redo answer", "Edit text", "Use text instead", ]) { @@ -874,20 +905,12 @@ describe("voice interview stage", () => { ...snapshot, canReviseLastAnswer: false, lastCommittedText: "The shift lead approves it.", - microphoneEnabled: false, - microphoneLevel: 0, partialText: "", - phase: "waiting", }, })} />, ); - expect( - screen - .getByRole("button", { name: "Redo answer" }) - .hasAttribute("disabled"), - ).toBe(true); expect( screen .getByRole("button", { name: "Edit text" }) @@ -907,11 +930,6 @@ describe("voice interview stage", () => { />, ); - expect( - screen - .getByRole("button", { name: "Redo answer" }) - .hasAttribute("disabled"), - ).toBe(false); expect( screen .getByRole("button", { name: "Edit text" }) @@ -919,24 +937,24 @@ describe("voice interview stage", () => { ).toBe(false); }); - test("offers deterministic interrupt instead of listening during playback", () => { + test("keeps listening and offers no manual interrupt during playback", () => { const html = renderToStaticMarkup( , ); - expect(html).toContain("Microphone off · Interviewer speaking"); - expect(html).toContain('aria-label="Interrupt and speak"'); - expect(html).not.toContain(">Pause<"); + expect(html).toContain( + "Microphone on · Interviewer speaking · Speak to interrupt", + ); + expect(html).not.toContain('aria-label="Interrupt and speak"'); + expect(html).toContain('aria-label="Pause"'); }); test("uses a detached bottom mini bar with independent expand, type, pause, and end controls", () => { @@ -957,9 +975,6 @@ describe("voice interview stage", () => { expect(screen.getByText("Listening")).not.toBeNull(); expect(screen.queryByText("Microphone on · Listening")).toBeNull(); expect(screen.getByText("What happens after approval?")).not.toBeNull(); - expect( - screen.getByRole("button", { name: "Done speaking" }), - ).not.toBeNull(); expect(screen.getByRole("button", { name: "Pause" })).not.toBeNull(); expect( screen.getByRole("button", { name: "Use text instead" }), @@ -968,12 +983,7 @@ describe("voice interview stage", () => { screen.getByRole("button", { name: "End interview" }), ).not.toBeNull(); - for (const name of [ - "Done speaking", - "Pause", - "Use text instead", - "End interview", - ]) { + for (const name of ["Pause", "Use text instead", "End interview"]) { const button = screen.getByRole("button", { name }); expect(button.querySelector("svg")).not.toBeNull(); expect(button.parentElement?.getAttribute("data-scope")).toBe("tooltip"); @@ -993,7 +1003,11 @@ describe("voice interview stage", () => { const rendered = render( , @@ -1017,25 +1031,23 @@ describe("voice interview stage", () => { rendered.rerender( , ); + expect(screen.getByRole("button", { name: "Pause" })).not.toBeNull(); expect( - screen.getByRole("button", { name: "Interrupt and speak" }), + screen.getByRole("button", { name: "Pause" }).querySelector("svg"), ).not.toBeNull(); - expect(screen.queryByRole("button", { name: "Pause" })).toBeNull(); expect( screen - .getByRole("button", { name: "Interrupt and speak" }) - .querySelector("svg"), - ).not.toBeNull(); - expect( - screen - .getByRole("button", { name: "Interrupt and speak" }) + .getByRole("button", { name: "Pause" }) .parentElement?.getAttribute("data-scope"), ).toBe("tooltip"); + expect( + screen.queryByRole("button", { name: "Interrupt and speak" }), + ).toBeNull(); }); test("announces compact question and provisional transcript context", () => { 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 b89eeb13591..0eb0ef6f7bf 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 @@ -9,8 +9,6 @@ import { useSyncExternalStore, } from "react"; import { - FaArrowRotateLeft, - FaCheck, FaCircleCheck, FaCircleNotch, FaKeyboard, @@ -36,11 +34,10 @@ import { selectInterviewCoverage, } from "./interview-coverage"; import { OpenAIRealtimeSession } from "./openai-realtime-session"; -import { SpeechPlaybackController } from "./speech-playback-controller"; +import { RealtimeBrunchBridge } from "./realtime-brunch-bridge"; import { VoiceTurnController, type VoiceLatencyEvent, - type VoiceTurnPhase, type VoiceTurnSnapshot, } from "./voice-turn-controller"; @@ -496,29 +493,22 @@ const liveRegionStyle = css({ }); const statusText = (snapshot: VoiceTurnSnapshot): string => { - switch (snapshot.phase) { - case "idle": - return "Microphone off · Interview not started"; - case "connecting": - return "Microphone off · Joining the interview"; - case "listening": - return "Microphone on · Listening"; - case "paused": - return "Microphone off · Paused"; - case "transcribing": - return "Microphone off · Finishing your answer"; - case "delivering": - return "Microphone off · Answer recorded"; - case "waiting": - return "Microphone off · Writing that down"; - case "synthesizing": - return "Microphone off · Preparing the next question"; - case "playing": - return "Microphone off · Interviewer speaking"; - case "recoverable-error": { - return `Microphone off · ${snapshot.errorMessage}`; - } - } + if (snapshot.connection === "idle") + return "Microphone off · Interview not started"; + if (snapshot.connection === "connecting") + return "Microphone off · Joining the interview"; + if (snapshot.connection === "error") + return `Microphone off · ${snapshot.errorMessage}`; + if (snapshot.input === "paused") return "Microphone off · Paused"; + if (snapshot.output === "speaking") + return "Microphone on · Interviewer speaking · Speak to interrupt"; + if (snapshot.input === "submitting") + return "Microphone on · Answer recorded · Writing that down"; + if (snapshot.output === "waiting-for-tool") + return "Microphone on · Preparing the next question"; + if (snapshot.output === "interrupted") + return "Microphone on · Listening after interruption"; + return "Microphone on · Listening"; }; type RecoveryErrorFamily = "connection" | "interview" | "microphone"; @@ -540,36 +530,23 @@ const recoveryErrorFamily = ( }; const shortStatusText = (snapshot: VoiceTurnSnapshot): string => { - switch (snapshot.phase) { - case "idle": - return "Ready"; - case "connecting": - return "Connecting"; - case "listening": - return "Listening"; - case "paused": - return "Paused"; - case "transcribing": - return "Finishing answer"; - case "delivering": - return "Answer recorded"; - case "waiting": - return "Writing that down"; - case "synthesizing": - return "Preparing next question"; - case "playing": - return "Interviewer speaking"; - case "recoverable-error": { - switch (recoveryErrorFamily(snapshot.errorCode)) { - case "microphone": - return "Microphone unavailable"; - case "connection": - return "Connection paused"; - case "interview": - return "Interview paused"; - } + if (snapshot.connection === "idle") return "Ready"; + if (snapshot.connection === "connecting") return "Connecting"; + if (snapshot.connection === "error") { + switch (recoveryErrorFamily(snapshot.errorCode)) { + case "microphone": + return "Microphone unavailable"; + case "connection": + return "Connection paused"; + case "interview": + return "Interview paused"; } } + if (snapshot.input === "paused") return "Paused"; + if (snapshot.output === "speaking") return "Interviewer speaking"; + if (snapshot.input === "submitting") return "Writing that down"; + if (snapshot.output === "waiting-for-tool") return "Preparing next question"; + return "Listening"; }; const inputLevelText = (level: number): string => @@ -605,44 +582,31 @@ const Meter = ({ snapshot }: { snapshot: VoiceTurnSnapshot }) => { ); }; -const focalIcon = (phase: VoiceTurnPhase): ReactNode => { - switch (phase) { - case "listening": - return