diff --git a/apps/petrinaut-website/README.md b/apps/petrinaut-website/README.md index 2d47a9d81bc..2e3cbd952ff 100644 --- a/apps/petrinaut-website/README.md +++ b/apps/petrinaut-website/README.md @@ -110,13 +110,22 @@ 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 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 shared -composer path with pending-`brunch_ask` correlation. +Realtime is the disposable media plane: it carries microphone and remote audio +and detects complete turns. Microphone capture is suspended during assistant +playback so speaker echo cannot become an interview answer, then restored to +the user's requested mute state when playback ends. Brunch remains the control +plane and sole authority for questions, captures, state, completion, and durable +history. The completed `gpt-4o-transcribe` transcript of the speaker's audio is +the only source of user answers: semantic VAD commits the audio without asking +the model to respond or interrupt playback, the session exposes no tools, and +the model never infers or paraphrases what the speaker said. The media session +drops playback-overlapping transcripts. The browser bridge normalizes +whitespace, submits each remaining completed transcript exactly once (keyed by +connection, item, and content index), ignores empty, duplicate, stale, or failed +transcripts, and submits the accepted text through Petrinaut's shared composer +path with pending-`brunch_ask` correlation and voice provenance. Silence, noise, +or a transcription failure creates no user message; the session returns to +listening and reports a recoverable "didn't catch that" notice. The experimental **Approach D** design waits for the correlated Brunch turn, then gives Realtime two bounded roles. First, an out-of-band, text-only request @@ -129,9 +138,10 @@ preparation is unavailable, invalid, or times out, the bridge supplies the canonical context and question instead. Brunch's canonical transcript and exact question are never rewritten and remain authoritative. Preparation, provisional transcription, and Realtime audio are ephemeral and are not -persisted. The microphone stays 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. +persisted. The microphone stays active while Brunch is working, but capture is +suspended while the interviewer speaks. Wait for playback to finish before +answering; Voice resumes listening automatically without changing Brunch +history. The Brunch deployment must allow the website origin through its `BRUNCH_PETRINAUT_ORIGINS` setting. Denying microphone permission leaves the 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 3798423848c..9f260841a96 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 @@ -232,7 +232,7 @@ describe("OpenAIRealtimeSession", () => { expect(harness.peers[0]!.close).toHaveBeenCalledOnce(); }); - test("keeps the microphone active through playback and reports automatic interruption", async () => { + test("does not expose audio detected during assistant playback as user speech", async () => { const harness = createHarness(); await harness.session.connect(); harness.session.setMicrophoneEnabled(true); @@ -253,136 +253,182 @@ describe("OpenAIRealtimeSession", () => { item_id: "item-user", type: "input_audio_buffer.speech_started", }); + channel.receive({ + response_id: "response-canonical", + type: "output_audio_buffer.stopped", + }); + channel.receive({ + content_index: 0, + item_id: "item-user", + transcript: "Each bit.", + type: "conversation.item.input_audio_transcription.completed", + }); + channel.receive({ + content_index: 0, + item_id: "item-user", + transcript: "Each bit.", + type: "conversation.item.input_audio_transcription.completed", + }); - 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", - }, - ]), - ); + expect(harness.events).toEqual([ + { + connectionEpoch: 1, + responseId: "response-canonical", + type: "output-started", + }, + { + connectionEpoch: 1, + responseId: "response-canonical", + type: "output-stopped", + }, + ]); }); - test("parses streamed tool arguments and the completed GA response output", async () => { + test("keeps a user turn that started before assistant playback", async () => { const harness = createHarness(); await harness.session.connect(); + harness.session.setMicrophoneEnabled(true); + harness.session.speakCanonical([ + canonicalSegment("ask-1", "What happens next?"), + ]); const channel = harness.channels[0]!; + authorizeLatestSpeechResponse(channel, "response-canonical"); 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", + audio_start_ms: 120, + item_id: "item-user", + type: "input_audio_buffer.speech_started", }); 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", + response_id: "response-canonical", + type: "output_audio_buffer.started", }); 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", + content_index: 0, + item_id: "item-user", + transcript: "The supervisor approves it.", + type: "conversation.item.input_audio_transcription.completed", }); expect(harness.events).toEqual([ { - callId: "call-1", connectionEpoch: 1, - delta: '{"answer":"Approved"}', - itemId: "item-function", - responseId: "response-tool", - type: "tool-arguments-delta", + itemId: "item-user", + type: "input-speech-started", }, { - arguments: '{"answer":"Approved"}', - callId: "call-1", connectionEpoch: 1, - itemId: "item-function", - name: "continue_interview", - responseId: "response-tool", - type: "tool-arguments-done", + responseId: "response-canonical", + type: "output-started", }, { - connectionEpoch: 1, - responseId: "response-tool", - status: "completed", - type: "response-terminal", + key: { connectionEpoch: 1, contentIndex: 0, itemId: "item-user" }, + text: "The supervisor approves it.", + type: "completed", }, ]); + }); - harness.session.completeFunctionCall("call-1", ["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?"] }), - }, + test("restores only the requested microphone state after assistant playback", async () => { + const harness = createHarness(); + await harness.session.connect(); + harness.session.setMicrophoneEnabled(true); + harness.session.speakCanonical([ + canonicalSegment("ask-1", "What happens next?"), + ]); + const channel = harness.channels[0]!; + authorizeLatestSpeechResponse(channel, "response-canonical"); + + channel.receive({ + response_id: "response-canonical", + type: "output_audio_buffer.started", }); - expect(responseCreate).toMatchObject({ - type: "response.create", + expect(harness.localTracks[0]!.enabled).toBe(false); + + channel.receive({ + response_id: "response-canonical", + type: "output_audio_buffer.stopped", + }); + expect(harness.localTracks[0]!.enabled).toBe(true); + channel.receive({ 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: [], + id: "response-canonical", + output: [], + status: "completed", }, + type: "response.done", }); - const responseCreateCount = sentEvents(channel).filter( - ({ type }) => type === "response.create", - ).length; - harness.session.completeFunctionCall( - "call-2", - ["Response cancelled before speech."], - { speakResponse: false }, - ); - expect(sentEvents(channel).at(-1)).toEqual({ - type: "conversation.item.create", - item: { - type: "function_call_output", - call_id: "call-2", - output: JSON.stringify({ - response_text: ["Response cancelled before speech."], - }), + harness.session.speakCanonical([ + canonicalSegment("ask-2", "And after that?"), + ]); + authorizeLatestSpeechResponse(channel, "response-canonical-2"); + channel.receive({ + response_id: "response-canonical-2", + type: "output_audio_buffer.started", + }); + harness.session.setMicrophoneEnabled(false); + channel.receive({ + response_id: "response-canonical-2", + type: "output_audio_buffer.stopped", + }); + + expect(harness.localTracks[0]!.enabled).toBe(false); + }); + + test("never surfaces model-generated function-call arguments as user speech", async () => { + const harness = createHarness(); + await harness.session.connect(); + const channel = harness.channels[0]!; + + channel.receive({ + arguments: '{"answer":"hi"}', + call_id: "call-legacy", + item_id: "item-legacy", + output_index: 0, + response_id: "response-legacy", + type: "response.function_call_arguments.done", + }); + channel.receive({ + call_id: "call-legacy", + delta: '{"answer":"hi"}', + item_id: "item-legacy", + output_index: 0, + response_id: "response-legacy", + type: "response.function_call_arguments.delta", + }); + expect(harness.events).toEqual([]); + + channel.receive({ + response: { + id: "response-legacy", + output: [ + { + arguments: '{"answer":"hi"}', + call_id: "call-legacy", + id: "item-legacy", + name: "continue_interview", + status: "completed", + type: "function_call", + }, + ], + status: "completed", }, + type: "response.done", }); + + expect(harness.events).toEqual([ + expect.objectContaining({ code: "invalid-response", type: "error" }), + ]); + expect(JSON.stringify(harness.events)).not.toContain('"hi"'); expect( - sentEvents(channel).filter(({ type }) => type === "response.create"), - ).toHaveLength(responseCreateCount); + sentEvents(channel).some( + ({ item }) => + (item as { type?: unknown } | undefined)?.type === + "function_call_output", + ), + ).toBe(false); + expect(harness.localTracks[0]!.stop).toHaveBeenCalledOnce(); }); test("renders prepared strings through the verbatim out-of-band audio response", async () => { @@ -1129,7 +1175,7 @@ describe("OpenAIRealtimeSession", () => { expect(harness.peers[0]!.close).toHaveBeenCalledOnce(); }); - test("treats transcripts as display-only and never closes capture", async () => { + test("emits keyed transcripts verbatim and never closes capture", async () => { const harness = createHarness(); await harness.session.connect(); harness.session.setMicrophoneEnabled(true); @@ -1147,6 +1193,12 @@ describe("OpenAIRealtimeSession", () => { transcript: "The supervisor approves it.", type: "conversation.item.input_audio_transcription.completed", }); + channel.receive({ + content_index: 0, + item_id: "item-silence", + transcript: "", + type: "conversation.item.input_audio_transcription.completed", + }); expect(harness.events).toEqual([ { @@ -1159,6 +1211,11 @@ describe("OpenAIRealtimeSession", () => { text: "The supervisor approves it.", type: "completed", }, + { + key: { connectionEpoch: 1, contentIndex: 0, itemId: "item-silence" }, + text: "", + type: "completed", + }, ]); expect(harness.localTracks[0]!.enabled).toBe(true); }); @@ -1218,13 +1275,10 @@ describe("OpenAIRealtimeSession", () => { await expect(harness.session.connect()).resolves.toBe(2); firstChannel.receive({ - arguments: '{"answer":"Stale"}', - call_id: "call-stale", + content_index: 0, item_id: "item-stale", - name: "continue_interview", - output_index: 0, - response_id: "response-stale", - type: "response.function_call_arguments.done", + transcript: "Stale answer", + type: "conversation.item.input_audio_transcription.completed", }); expect(harness.events).toEqual([]); 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 0475308cf52..6bbd9e36be6 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 @@ -19,13 +19,11 @@ export interface OpenAIRealtimeTranscriptKey { readonly itemId: string; } -interface RealtimeToolEventIdentity { - readonly callId: string; - readonly connectionEpoch: number; - readonly itemId: string; - readonly responseId: string; -} - +/** + * Events the browser control plane consumes. Completed `input_audio_transcription` + * transcripts (`type: "completed"`) are the only source of the user's words: + * the session never exposes model-generated output as user speech. + */ export type OpenAIRealtimeSessionEvent = | { readonly key: OpenAIRealtimeTranscriptKey; @@ -68,15 +66,6 @@ export type OpenAIRealtimeSessionEvent = 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; @@ -269,6 +258,7 @@ export class OpenAIRealtimeSession { PendingInterviewPreparation >(); readonly #pendingSpeechRequests = new Map(); + readonly #playbackOverlappingInputItemIds = new Set(); readonly #preparationResponseIds = new Map(); readonly #remoteStreams = new Set(); readonly #responseQueue: SerializedResponseRequest[] = []; @@ -290,6 +280,7 @@ export class OpenAIRealtimeSession { #meterHasSample = false; #meterLevel = 0; #meterSamples: Uint8Array | null = null; + #microphoneRequested = false; #microphoneTrack: MediaStreamTrack | null = null; #peerConnection: RTCPeerConnection | null = null; #remoteAudio: RemoteAudio | null = null; @@ -460,50 +451,16 @@ export class OpenAIRealtimeSession { } public setMicrophoneEnabled(enabled: boolean): void { - if (!this.#microphoneTrack) { - return; - } - const isEnabled = enabled && this.#connected; - this.#microphoneTrack.enabled = isEnabled; - if (isEnabled) { - this.#startMeter(); - } else { - this.#stopMeter(); - } + this.#microphoneRequested = enabled && this.#connected; + this.#syncMicrophoneTrack(); } public speakCanonical(segments: CanonicalSpeechSegment[]): void { - this.#requestSpeech(this.#canonicalResponseText(segments), true); + this.#requestSpeech(this.#canonicalResponseText(segments)); } public speakPrepared(responseText: readonly string[]): void { - this.#requestSpeech(this.#validResponseText(responseText), true); - } - - public completeFunctionCall( - callId: string, - responseTextInput: readonly string[], - { - speakResponse = true, - }: { - readonly speakResponse?: boolean; - } = {}, - ): void { - if (!callId) { - throw new VoiceError("speech", "invalid-response", ""); - } - const responseText = this.#validResponseText(responseTextInput); - this.#send({ - type: "conversation.item.create", - item: { - type: "function_call_output", - call_id: callId, - output: JSON.stringify({ response_text: responseText }), - }, - }); - if (speakResponse) { - this.#requestSpeech(responseText, false); - } + this.#requestSpeech(this.#validResponseText(responseText)); } public prepareInterviewSpeech( @@ -682,7 +639,12 @@ export class OpenAIRealtimeSession { return responseText; } - #requestSpeech(responseText: string[], outOfBand: boolean): void { + /** + * Every spoken response is out of band (`conversation: "none"`) with tools + * disabled, so Realtime only ever renders text Petrinaut supplied and never + * continues the conversation on its own. + */ + #requestSpeech(responseText: string[]): void { const speechRequestId = `canonical-${this.#activeEpoch}-${++this.#speechRequestSequence}`; this.#pendingSpeechRequests.set(speechRequestId, { requestId: @@ -690,23 +652,19 @@ export class OpenAIRealtimeSession { startedAt: this.#now(), }); const response = { - ...(outOfBand - ? { - conversation: "none", - input: [ - { - type: "message", - role: "system", - content: [ - { - type: "input_text", - text: JSON.stringify({ response_text: responseText }), - }, - ], - }, - ], - } - : {}), + 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, @@ -826,23 +784,21 @@ export class OpenAIRealtimeSession { if (parsed.type === "input_audio_buffer.speech_started") { const itemId = nonEmptyString(parsed.item_id); if (!itemId || nonNegativeInteger(parsed.audio_start_ms) === null) return; + if (this.#speakingResponseId) { + this.#playbackOverlappingInputItemIds.add(itemId); + 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; + if (this.#playbackOverlappingInputItemIds.has(itemId)) return; this.#emit({ connectionEpoch, itemId, @@ -857,10 +813,6 @@ export class OpenAIRealtimeSession { 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" || @@ -1112,44 +1064,13 @@ export class OpenAIRealtimeSession { 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)) - ) { + // The session exposes no tools, so a function call means the provider + // ignored the policy. Its arguments are model output, never the user's + // words, and must never reach the interview. + if (output.some((item) => asRecord(item)?.type === "function_call")) { 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, @@ -1335,6 +1256,7 @@ export class OpenAIRealtimeSession { return; } this.#speakingResponseId = responseId; + this.#syncMicrophoneTrack(); this.#emit({ connectionEpoch, responseId, type: "output-started" }); return; } @@ -1350,26 +1272,6 @@ export class OpenAIRealtimeSession { } } - #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, @@ -1378,6 +1280,22 @@ export class OpenAIRealtimeSession { const contentIndex = nonNegativeInteger(event.content_index); if (!itemId || contentIndex === null) return; const key = { connectionEpoch, contentIndex, itemId }; + const overlapsPlayback = this.#playbackOverlappingInputItemIds.has(itemId); + if (overlapsPlayback) { + if ( + event.type === + "conversation.item.input_audio_transcription.completed" || + event.type === "conversation.item.input_audio_transcription.failed" + ) { + this.#finishTranscription( + itemId, + event.type === "conversation.item.input_audio_transcription.failed" + ? "invalid-response" + : undefined, + ); + } + return; + } this.#startTranscription(itemId); if (event.type === "conversation.item.input_audio_transcription.failed") { this.#finishTranscription(itemId, "invalid-response"); @@ -1432,6 +1350,7 @@ export class OpenAIRealtimeSession { this.#authorizedResponseIds.delete(responseId); if (this.#speakingResponseId === responseId) { this.#speakingResponseId = null; + this.#syncMicrophoneTrack(); } } @@ -1613,6 +1532,22 @@ export class OpenAIRealtimeSession { this.#meterFrame = this.#dependencies.requestAnimationFrame(sample); } + #syncMicrophoneTrack(): void { + if (!this.#microphoneTrack) { + return; + } + const enabled = + this.#microphoneRequested && + this.#connected && + this.#speakingResponseId === null; + this.#microphoneTrack.enabled = enabled; + if (enabled) { + this.#startMeter(); + } else { + this.#stopMeter(); + } + } + #stopMeter(): void { if (this.#meterFrame === null) return; this.#dependencies.cancelAnimationFrame(this.#meterFrame); @@ -1713,6 +1648,7 @@ export class OpenAIRealtimeSession { this.#pendingClientEvents.clear(); this.#pendingPreparations.clear(); this.#pendingSpeechRequests.clear(); + this.#playbackOverlappingInputItemIds.clear(); this.#preparationResponseIds.clear(); this.#responseQueue.length = 0; this.#speechTimings.clear(); @@ -1724,6 +1660,7 @@ export class OpenAIRealtimeSession { this.#waitingForResponseTerminal = false; this.#activeEpoch = null; this.#connected = false; + this.#microphoneRequested = false; this.#connectedAt = null; this.#connectionRequestId = null; this.#abortController?.abort(); 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 index 37c1f796a22..a6160e5360a 100644 --- 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 @@ -15,6 +15,7 @@ import type { InterviewSpeechPreparationRequest, InterviewSpeechPreparationResult, OpenAIRealtimeSessionEvent, + OpenAIRealtimeTranscriptKey, } from "./openai-realtime-session"; const segment = ( @@ -58,7 +59,6 @@ const speechSource = ({ const createHarness = () => { let listener: ((event: OpenAIRealtimeSessionEvent) => void) | undefined; const session = { - completeFunctionCall: vi.fn(), prepareInterviewSpeech: vi.fn( async ( request: InterviewSpeechPreparationRequest, @@ -102,42 +102,39 @@ const createHarness = () => { }; }; -const toolDelta = ( +const transcriptKey = ( connectionEpoch: number, - delta: string, -): Extract => ({ - callId: "call-1", - connectionEpoch, - delta, - itemId: "function-item-1", - responseId: "response-1", - type: "tool-arguments-delta", -}); + itemId = "user-item-1", + contentIndex = 0, +): OpenAIRealtimeTranscriptKey => ({ connectionEpoch, contentIndex, itemId }); -const toolDone = ( +const transcriptCompleted = ( 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", + text = "The supervisor approves it.", + itemId = "user-item-1", +): OpenAIRealtimeSessionEvent => ({ + key: transcriptKey(connectionEpoch, itemId), + text, + type: "completed", }); -const responseTerminal = ( +const transcriptFailed = ( connectionEpoch: number, - status: "cancelled" | "completed" | "failed" | "incomplete", - responseId = "response-1", -): Extract => ({ - connectionEpoch, - responseId, - status, - type: "response-terminal", + itemId = "user-item-1", +): Extract => ({ + key: transcriptKey(connectionEpoch, itemId), + type: "transcription-failed", }); +const submitTranscript = ( + harness: ReturnType, + connectionEpoch = 1, + text = "The supervisor approves it.", + itemId = "user-item-1", +) => { + harness.emit(transcriptCompleted(connectionEpoch, text, itemId)); +}; + describe("RealtimeBrunchBridge", () => { test("prepares only context and appends the exact canonical question", async () => { const harness = createHarness(); @@ -395,7 +392,48 @@ describe("RealtimeBrunchBridge", () => { ); }); - test("completes a function call without speech when preparation is cancelled", async () => { + test("restores pending speech after rejected input", async () => { + const harness = createHarness(); + const context = segment("context", "Complete context.", "assistant-text"); + const question = segment("ask-current", "Exact question?"); + const source = speechSource({ context: [context], question }); + let finishCancelledPreparation: + | ((result: InterviewSpeechPreparationResult) => void) + | undefined; + harness.session.prepareInterviewSpeech.mockImplementationOnce( + () => + new Promise((resolve) => { + finishCancelledPreparation = resolve; + }), + ); + harness.bridge.updateChat({ + automaticSource: source, + canAcceptInterviewAnswer: true, + canonicalSegments: [...source.fullResponseSegments], + status: "ready", + }); + harness.bridge.start(4); + + harness.bridge.cancelPendingSpeech(); + harness.bridge.restoreCancelledSpeech(); + + await vi.waitFor(() => + expect(harness.session.speakPrepared).toHaveBeenCalledWith([ + "Prepared concise context.", + question.text, + ]), + ); + finishCancelledPreparation?.({ + context: "Late cancelled context.", + kind: "prepared", + sourceSegmentIds: [context.id], + }); + await Promise.resolve(); + + expect(harness.session.speakPrepared).toHaveBeenCalledOnce(); + }); + + test("speaks nothing when preparation is cancelled after a transcript submission", async () => { const harness = createHarness(); const question = segment("ask-current", "What happens after approval?"); harness.bridge.updateChat({ @@ -404,7 +442,8 @@ describe("RealtimeBrunchBridge", () => { status: "ready", }); harness.bridge.start(7); - harness.emit(toolDone(7)); + harness.session.speakCanonical.mockClear(); + submitTranscript(harness, 7); await vi.waitFor(() => expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), ); @@ -447,21 +486,23 @@ describe("RealtimeBrunchBridge", () => { expect(harness.session.prepareInterviewSpeech).toHaveBeenCalledOnce(), ); + expect(harness.events).toContainEqual({ + segments: source.fullResponseSegments, + type: "canonical-response-ready", + }); + harness.bridge.cancelPendingSpeech(); finishPreparation?.({ context: "Prepared concise context.", kind: "prepared", sourceSegmentIds: [acknowledgement.id], }); + await Promise.resolve(); + await Promise.resolve(); - await vi.waitFor(() => - expect(harness.session.completeFunctionCall).toHaveBeenCalledWith( - "call-1", - [acknowledgement.text, nextQuestion.text], - { speakResponse: false }, - ), - ); expect(harness.session.speakPrepared).not.toHaveBeenCalled(); + expect(harness.session.speakCanonical).not.toHaveBeenCalled(); + expect(harness.events.some((event) => event.type === "error")).toBe(false); }); test("speaks the current canonical turn without replaying history", () => { @@ -494,7 +535,7 @@ describe("RealtimeBrunchBridge", () => { ]); }); - test("streams and validates one tool call, preserves ask correlation, and waits for canonical Brunch output", async () => { + test("submits the completed transcript verbatim, 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({ @@ -505,18 +546,21 @@ describe("RealtimeBrunchBridge", () => { 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)); + harness.emit({ + key: transcriptKey(7), + text: "The supervisor", + type: "partial", + }); + expect(harness.submitInterviewAnswer).not.toHaveBeenCalled(); + submitTranscript(harness, 7); await vi.waitFor(() => expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), ); expect(harness.submitInterviewAnswer).toHaveBeenCalledWith({ - id: createRealtimeSubmissionId(7, "call-1"), + id: createRealtimeSubmissionId(transcriptKey(7)), text: "The supervisor approves it.", }); - expect(harness.session.completeFunctionCall).not.toHaveBeenCalled(); harness.bridge.updateChat({ canAcceptInterviewAnswer: false, @@ -539,67 +583,81 @@ describe("RealtimeBrunchBridge", () => { status: "ready", }); - await vi.waitFor(() => - expect(harness.session.completeFunctionCall).toHaveBeenCalledWith( - "call-1", - [acknowledgement.text, nextQuestion.text], - ), - ); - expect(harness.events.map(({ type }) => type)).toEqual([ - "submission-started", - "submission-accepted", - "canonical-response-ready", + expect(harness.session.speakCanonical).toHaveBeenCalledWith([ + acknowledgement, + nextQuestion, + ]); + expect(harness.events).toEqual([ + { answer: "The supervisor approves it.", type: "submission-started" }, + { answer: "The supervisor approves it.", type: "submission-accepted" }, + { + segments: [acknowledgement, nextQuestion], + type: "canonical-response-ready", + }, ]); }); - test("uses the first spoken turn to start Brunch when no question exists", async () => { + test("delivers Brunch responses to transcript submissions through prepared speech", async () => { const harness = createHarness(); - harness.submitInterviewAnswer.mockResolvedValueOnce({ - kind: "message", - messageId: "message-kickoff", - }); + const question = segment("ask-current", "What happens after approval?"); harness.bridge.updateChat({ canAcceptInterviewAnswer: true, - canonicalSegments: [], + canonicalSegments: [question], status: "ready", }); harness.bridge.start(7); - - harness.emit(toolDone(7, '{"answer":"Battery charger workflow"}')); - + harness.session.speakCanonical.mockClear(); + submitTranscript(harness, 7); await vi.waitFor(() => - expect(harness.submitInterviewAnswer).toHaveBeenCalledWith({ - id: createRealtimeSubmissionId(7, "call-1"), - text: "Battery charger workflow", - }), + expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), ); harness.bridge.updateChat({ canAcceptInterviewAnswer: false, - canonicalSegments: [], - status: "submitted", + canonicalSegments: [question], + status: "streaming", }); - const firstQuestion = segment( - "ask-first", - "What starts the battery charger workflow?", + + const acknowledgement = segment( + "acknowledgement", + "Thanks. I have recorded that.", + "assistant-text", ); + const nextQuestion = segment("ask-next", "Who is informed next?"); + const source = speechSource({ + context: [acknowledgement], + question: nextQuestion, + }); harness.bridge.updateChat({ + automaticSource: source, canAcceptInterviewAnswer: true, - canonicalSegments: [firstQuestion], + canonicalSegments: [question, ...source.fullResponseSegments], status: "ready", }); - expect(harness.session.completeFunctionCall).toHaveBeenCalledWith( - "call-1", - [firstQuestion.text], + await vi.waitFor(() => + expect(harness.session.speakPrepared).toHaveBeenCalledOnce(), + ); + expect(harness.session.prepareInterviewSpeech).toHaveBeenCalledOnce(); + expect(harness.session.prepareInterviewSpeech).toHaveBeenCalledWith( + expect.objectContaining({ + contextText: [acknowledgement.text], + sourceSegmentIds: [acknowledgement.id], + }), ); + expect(harness.session.speakPrepared).toHaveBeenCalledWith([ + "Prepared concise context.", + nextQuestion.text, + ]); + expect(harness.session.speakCanonical).not.toHaveBeenCalled(); expect(harness.events.map(({ type }) => type)).toEqual([ "submission-started", "submission-accepted", + "speech-delivery-pending", "canonical-response-ready", ]); }); - test("requires a correlated Brunch busy cycle before accepting new canonical segments", async () => { + test("falls back to exact canonical content once when preparation fails after a transcript submission", async () => { const harness = createHarness(); const question = segment("ask-current", "What happens after approval?"); harness.bridge.updateChat({ @@ -608,42 +666,98 @@ describe("RealtimeBrunchBridge", () => { status: "ready", }); harness.bridge.start(7); - harness.emit(toolDone(7)); + harness.session.speakCanonical.mockClear(); + submitTranscript(harness, 7); await vi.waitFor(() => expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), ); - const unrelated = segment( - "unrelated", - "An unrelated canonical update.", + harness.bridge.updateChat({ + canAcceptInterviewAnswer: false, + canonicalSegments: [question], + status: "streaming", + }); + + harness.session.prepareInterviewSpeech.mockResolvedValueOnce({ + kind: "fallback", + reason: "timeout", + sourceSegmentIds: ["acknowledgement"], + }); + const acknowledgement = segment( + "acknowledgement", + "Thanks. I have recorded that.", "assistant-text", ); - + const nextQuestion = segment("ask-next", "Who is informed next?"); + const source = speechSource({ + context: [acknowledgement], + question: nextQuestion, + }); harness.bridge.updateChat({ + automaticSource: source, canAcceptInterviewAnswer: true, - canonicalSegments: [question, unrelated], + canonicalSegments: [question, ...source.fullResponseSegments], status: "ready", }); - expect(harness.session.completeFunctionCall).not.toHaveBeenCalled(); + await vi.waitFor(() => + expect(harness.session.speakPrepared).toHaveBeenCalledOnce(), + ); + expect(harness.session.speakPrepared).toHaveBeenCalledWith([ + acknowledgement.text, + nextQuestion.text, + ]); + expect(harness.session.speakCanonical).not.toHaveBeenCalled(); + expect(harness.events.some((event) => event.type === "error")).toBe(false); + }); + test("normalizes transcript whitespace before submitting", async () => { + const harness = createHarness(); harness.bridge.updateChat({ - canAcceptInterviewAnswer: false, - canonicalSegments: [question, unrelated], - status: "submitted", + canAcceptInterviewAnswer: true, + canonicalSegments: [segment("ask-current", "Question")], + status: "ready", }); + harness.bridge.start(3); + + submitTranscript(harness, 3, " The supervisor\napproves it. "); + + await vi.waitFor(() => + expect(harness.submitInterviewAnswer).toHaveBeenCalledWith({ + id: createRealtimeSubmissionId(transcriptKey(3)), + text: "The supervisor approves it.", + }), + ); + }); + + test("does not submit empty or whitespace-only transcripts", async () => { + const harness = createHarness(); harness.bridge.updateChat({ canAcceptInterviewAnswer: true, - canonicalSegments: [question, unrelated], + canonicalSegments: [segment("ask-current", "Question")], status: "ready", }); + harness.bridge.start(3); + + submitTranscript(harness, 3, "", "silent-item"); + submitTranscript(harness, 3, " \n\t ", "noise-item"); + await Promise.resolve(); + + expect(harness.submitInterviewAnswer).not.toHaveBeenCalled(); + expect(harness.events).toEqual([ + { reason: "empty", type: "transcript-rejected" }, + { reason: "empty", type: "transcript-rejected" }, + ]); - expect(harness.session.completeFunctionCall).toHaveBeenCalledWith( - "call-1", - [unrelated.text], + submitTranscript(harness, 3, "A real answer.", "spoken-item"); + await vi.waitFor(() => + expect(harness.submitInterviewAnswer).toHaveBeenCalledWith({ + id: createRealtimeSubmissionId(transcriptKey(3, "spoken-item")), + text: "A real answer.", + }), ); }); - test("rejects streamed arguments whose response or item identity changes", async () => { + test("submits duplicate completion events exactly once", async () => { const harness = createHarness(); harness.bridge.updateChat({ canAcceptInterviewAnswer: true, @@ -651,24 +765,70 @@ describe("RealtimeBrunchBridge", () => { status: "ready", }); harness.bridge.start(3); - harness.emit(toolDelta(3, '{"answer":"Answer"}')); - harness.emit({ - ...toolDone(3, '{"answer":"Answer"}'), - responseId: "response-2", + submitTranscript(harness, 3); + submitTranscript(harness, 3); + await vi.waitFor(() => + expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), + ); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: false, + canonicalSegments: [segment("ask-current", "Question")], + status: "submitted", + }); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [ + segment("ask-current", "Question"), + segment("ask-next", "Next question"), + ], + status: "ready", + }); + submitTranscript(harness, 3); + await Promise.resolve(); + + expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(); + expect(harness.events).toContainEqual({ + reason: "duplicate", + type: "transcript-rejected", + }); + expect( + harness.events.filter( + (event) => + event.type === "transcript-rejected" && event.reason === "duplicate", + ), + ).toHaveLength(2); + expect(harness.events.some((event) => event.type === "error")).toBe(false); + }); + + test("ignores stale connection-epoch transcripts", async () => { + const harness = createHarness(); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [segment("ask-current", "Question")], + status: "ready", }); + harness.bridge.start(1); + harness.bridge.stop(); + harness.bridge.start(2); + + submitTranscript(harness, 1, "Stale answer"); + harness.emit(transcriptFailed(1, "stale-failed")); await Promise.resolve(); expect(harness.submitInterviewAnswer).not.toHaveBeenCalled(); - expect(harness.events).toEqual([ - expect.objectContaining({ - code: "interview-correlation", - type: "error", + expect(harness.events).toEqual([]); + + submitTranscript(harness, 2, "Current answer"); + await vi.waitFor(() => + expect(harness.submitInterviewAnswer).toHaveBeenCalledWith({ + id: createRealtimeSubmissionId(transcriptKey(2)), + text: "Current answer", }), - ]); + ); }); - test("rejects concurrent argument streams before either can submit", async () => { + test("returns to listening after a failed transcription", async () => { const harness = createHarness(); harness.bridge.updateChat({ canAcceptInterviewAnswer: true, @@ -676,21 +836,25 @@ describe("RealtimeBrunchBridge", () => { 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", - }); + harness.emit(transcriptFailed(3, "failed-item")); + await Promise.resolve(); expect(harness.submitInterviewAnswer).not.toHaveBeenCalled(); expect(harness.events).toEqual([ - expect.objectContaining({ type: "error" }), + { reason: "failed", type: "transcript-rejected" }, ]); + + submitTranscript(harness, 3, "Retried answer.", "retry-item"); + await vi.waitFor(() => + expect(harness.submitInterviewAnswer).toHaveBeenCalledWith({ + id: createRealtimeSubmissionId(transcriptKey(3, "retry-item")), + text: "Retried answer.", + }), + ); }); - test("discards a cancelled argument stream without poisoning the next answer", async () => { + test("never submits fabricated continue_interview arguments", async () => { const harness = createHarness(); harness.bridge.updateChat({ canAcceptInterviewAnswer: true, @@ -698,36 +862,90 @@ describe("RealtimeBrunchBridge", () => { 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", - }); + callId: "call-1", + connectionEpoch: 3, + delta: '{"answer":"hi"}', + itemId: "function-item-1", + responseId: "response-1", + type: "tool-arguments-delta", + } as unknown as OpenAIRealtimeSessionEvent); harness.emit({ - ...toolDone(3, '{"answer":"Accepted"}'), - callId: "call-2", - itemId: "function-item-2", - responseId: "response-2", + arguments: '{"answer":"hi"}', + callId: "call-1", + connectionEpoch: 3, + itemId: "function-item-1", + name: "continue_interview", + responseId: "response-1", + type: "tool-arguments-done", + } as unknown as OpenAIRealtimeSessionEvent); + harness.emit({ + connectionEpoch: 3, + responseId: "response-1", + status: "completed", + type: "response-terminal", + }); + await Promise.resolve(); + + expect(harness.submitInterviewAnswer).not.toHaveBeenCalled(); + expect(harness.events).toEqual([]); + }); + + test("ignores transcripts while an answer is in flight or Brunch cannot accept input", async () => { + const harness = createHarness(); + const question = segment("ask-current", "Question"); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [question], + status: "ready", }); + harness.bridge.start(3); + submitTranscript(harness, 3, "First answer.", "first-item"); await vi.waitFor(() => expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), ); - expect(harness.submitInterviewAnswer).toHaveBeenCalledWith({ - id: createRealtimeSubmissionId(3, "call-2"), - text: "Accepted", + submitTranscript(harness, 3, "", "silent-item"); + harness.emit(transcriptFailed(3, "failed-item")); + await Promise.resolve(); + + expect(harness.events.slice(-2)).toEqual([ + { reason: "unavailable", type: "transcript-rejected" }, + { reason: "unavailable", type: "transcript-rejected" }, + ]); + + submitTranscript(harness, 3, "Overlapping answer.", "second-item"); + await Promise.resolve(); + + expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(); + expect(harness.events.at(-1)).toEqual({ + reason: "unavailable", + type: "transcript-rejected", }); - expect(harness.events).not.toContainEqual( - expect.objectContaining({ type: "error" }), - ); + + harness.bridge.updateChat({ + canAcceptInterviewAnswer: false, + canonicalSegments: [question], + status: "streaming", + }); + harness.bridge.updateChat({ + canAcceptInterviewAnswer: false, + canonicalSegments: [question, segment("ask-next", "Next question")], + status: "ready", + }); + submitTranscript(harness, 3, "Too early.", "third-item"); + await Promise.resolve(); + + expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(); + expect(harness.events.at(-1)).toEqual({ + reason: "unavailable", + type: "transcript-rejected", + }); + expect(harness.events.some((event) => event.type === "error")).toBe(false); }); - test("rejects an unfinished argument stream from a completed response", () => { + test("rejects transcripts over the answer limit", async () => { const harness = createHarness(); harness.bridge.updateChat({ canAcceptInterviewAnswer: true, @@ -735,20 +953,63 @@ describe("RealtimeBrunchBridge", () => { status: "ready", }); harness.bridge.start(3); - harness.emit(toolDelta(3, '{"answer":"Incomplete')); - harness.emit(responseTerminal(3, "completed")); + submitTranscript(harness, 3, "a".repeat(32_001)); + await Promise.resolve(); expect(harness.submitInterviewAnswer).not.toHaveBeenCalled(); expect(harness.events).toEqual([ - expect.objectContaining({ - code: "interview-correlation", - type: "error", + { reason: "too-long", type: "transcript-rejected" }, + ]); + }); + + 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); + + submitTranscript(harness, 7, "Battery charger workflow"); + + await vi.waitFor(() => + expect(harness.submitInterviewAnswer).toHaveBeenCalledWith({ + id: createRealtimeSubmissionId(transcriptKey(7)), + 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.speakCanonical).toHaveBeenCalledWith([ + firstQuestion, + ]); + expect(harness.events.map(({ type }) => type)).toEqual([ + "submission-started", + "submission-accepted", + "canonical-response-ready", ]); }); - test("rejects duplicate, stale, overlapping, and malformed calls without another Brunch submission", async () => { + 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({ @@ -756,47 +1017,38 @@ describe("RealtimeBrunchBridge", () => { 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)); + harness.bridge.start(7); + harness.session.speakCanonical.mockClear(); + submitTranscript(harness, 7); await vi.waitFor(() => expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), ); + const unrelated = segment( + "unrelated", + "An unrelated canonical update.", + "assistant-text", + ); - harness.emit({ - ...toolDone(2, '{"answer":"Overlapping"}'), - callId: "call-2", - itemId: "function-item-2", + harness.bridge.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [question, unrelated], + status: "ready", }); - expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(); - expect(harness.events.at(-1)).toMatchObject({ type: "error" }); - }); + expect(harness.session.speakCanonical).not.toHaveBeenCalled(); - 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: false, + canonicalSegments: [question, unrelated], + status: "submitted", + }); harness.bridge.updateChat({ canAcceptInterviewAnswer: true, - canonicalSegments: [segment("ask-current", "Question")], + canonicalSegments: [question, unrelated], 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" }), - ]); + expect(harness.session.speakCanonical).toHaveBeenCalledWith([unrelated]); }); test("rejects a composer result that does not match the pending brunch_ask", async () => { @@ -811,16 +1063,18 @@ describe("RealtimeBrunchBridge", () => { status: "ready", }); harness.bridge.start(5); + harness.session.speakCanonical.mockClear(); - harness.emit(toolDone(5)); + submitTranscript(harness, 5); await vi.waitFor(() => expect(harness.events.at(-1)).toMatchObject({ type: "error" }), ); - expect(harness.session.completeFunctionCall).not.toHaveBeenCalled(); + expect(harness.session.speakCanonical).not.toHaveBeenCalled(); + expect(harness.session.speakPrepared).not.toHaveBeenCalled(); }); - test("speaks new canonical text turns without creating a Realtime tool result", () => { + test("speaks new canonical text turns without a Realtime tool result", () => { const harness = createHarness(); const question = segment("ask-current", "Question"); harness.bridge.updateChat({ @@ -843,6 +1097,5 @@ describe("RealtimeBrunchBridge", () => { }); 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 index 5bf988dbfe0..737bdfc1903 100644 --- 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 @@ -8,6 +8,7 @@ import type { InterviewSpeechPreparationRequest, InterviewSpeechPreparationResult, OpenAIRealtimeSessionEvent, + OpenAIRealtimeTranscriptKey, } from "./openai-realtime-session"; type ChatStatus = "ready" | "submitted" | "streaming" | "error"; @@ -20,11 +21,6 @@ interface ChatUpdate { } interface RealtimeBridgeSession { - completeFunctionCall( - callId: string, - responseText: readonly string[], - options?: { readonly speakResponse?: boolean }, - ): void; prepareInterviewSpeech( request: InterviewSpeechPreparationRequest, ): Promise; @@ -51,41 +47,45 @@ interface RealtimeBrunchBridgeDependencies { interface ActiveSubmission { readonly baselineSegmentIds: ReadonlySet; - readonly callId: string; readonly epoch: number; readonly pendingQuestionId: string | null; + readonly transcriptId: string; correlated: boolean; sawBusyChatStatus: boolean; } -interface ArgumentStream { - readonly chunks: string[]; - readonly itemId: string; - readonly responseId: string; -} - -type SpeechDelivery = - | { readonly kind: "automatic" } - | { readonly callId: string; readonly kind: "function-call" }; - export type RealtimeBridgeErrorCode = | "interview-correlation" | "interview-response" | "interview-submission"; +/** + * Why a completed user transcript was not submitted to Brunch. + * + * - `empty`: the transcript contained no words (silence or noise). + * - `failed`: Realtime reported that transcription of the audio failed. + * - `duplicate`: this item and content index were already handled. + * - `unavailable`: an answer is already in flight or Brunch cannot accept + * input right now. + * - `too-long`: the transcript exceeds the interview answer limit. + */ +export type RealtimeTranscriptRejectionReason = + | "duplicate" + | "empty" + | "failed" + | "too-long" + | "unavailable"; + 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"; } @@ -96,6 +96,10 @@ export type RealtimeBrunchBridgeEvent = } | { readonly type: "speech-delivery-pending"; + } + | { + readonly reason: RealtimeTranscriptRejectionReason; + readonly type: "transcript-rejected"; }; export interface PreparedInterviewSpeech { @@ -138,10 +142,24 @@ 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; +const transcriptIdentity = ({ + connectionEpoch, + contentIndex, + itemId, +}: OpenAIRealtimeTranscriptKey): string => + `${connectionEpoch}:${encodeURIComponent(itemId)}:${contentIndex}`; + +/** + * Stable submission ID for one completed user transcript, derived from the + * connection epoch, Realtime item ID, and content index. The bridge submits + * each identity at most once, and downstream consumers can de-duplicate on it. + */ export const createRealtimeSubmissionId = ( - connectionEpoch: number, - callId: string, -): string => `voice-realtime:${connectionEpoch}:${encodeURIComponent(callId)}`; + key: OpenAIRealtimeTranscriptKey, +): string => `voice-realtime:${transcriptIdentity(key)}`; + +const normalizeTranscript = (text: string): string => + text.trim().replace(/\s+/gu, " "); const latestPendingQuestion = ( segments: CanonicalSpeechSegment[], @@ -164,36 +182,29 @@ const preparationCacheKey = ( return `speech-preparation:${hashCanonicalSpeechText(sourceIdentity)}`; }; -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; - } -}; - +/** + * Connects the Realtime session to the Brunch interview. + * + * Inbound: the completed `gpt-4o-transcribe` transcript of the user's audio is + * the only source of user answers. Realtime never generates a response of its + * own between turns, so nothing the model infers can become a user message. + * Each transcript identity (connection epoch, item ID, content index) is + * submitted at most once; empty, failed, stale, duplicate, or ill-timed + * transcripts are dropped and reported through `transcript-rejected`. + * + * Outbound: Brunch responses are spoken through the prepared-speech path + * (concise context plus the exact canonical question) or verbatim canonical + * speech, both of which are out-of-band Realtime responses without tools. + */ export class RealtimeBrunchBridge { - readonly #argumentDeltas = new Map(); readonly #listeners = new Set(); readonly #preparedContextCache = new Map(); - readonly #processedCalls = new Set(); + readonly #processedTranscripts = 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 = { @@ -201,7 +212,9 @@ export class RealtimeBrunchBridge { canonicalSegments: [], status: "ready", }; + #cancelledSpeechSource: InterviewSpeechSource | null = null; #generation = 0; + #pendingSpeechSource: InterviewSpeechSource | null = null; #speechGeneration = 0; public constructor({ @@ -222,20 +235,20 @@ export class RealtimeBrunchBridge { ++this.#generation; if (this.#activeEpoch !== connectionEpoch) { this.#preparedContextCache.clear(); + this.#processedTranscripts.clear(); } this.#activeEpoch = connectionEpoch; this.#activeSubmission = null; - this.#argumentDeltas.clear(); - this.#processedCalls.clear(); + this.#cancelledSpeechSource = null; + this.#pendingSpeechSource = null; this.#seenSegmentIds.clear(); - this.#terminalResponseIds.clear(); for (const segment of this.#chat.canonicalSegments) { this.#seenSegmentIds.add(segment.id); } const source = this.#chat.automaticSource; if (source) { - this.#prepareAndDeliver(source, { kind: "automatic" }); + this.#prepareAndDeliver(source); } else { const question = latestPendingQuestion(this.#chat.canonicalSegments); if (!question) { @@ -252,13 +265,26 @@ export class RealtimeBrunchBridge { ++this.#generation; this.#activeEpoch = null; this.#activeSubmission = null; - this.#argumentDeltas.clear(); + this.#cancelledSpeechSource = null; + this.#pendingSpeechSource = null; this.#preparedContextCache.clear(); - this.#terminalResponseIds.clear(); } public cancelPendingSpeech(): void { ++this.#speechGeneration; + if (this.#pendingSpeechSource) { + this.#cancelledSpeechSource = this.#pendingSpeechSource; + this.#pendingSpeechSource = null; + } + } + + public restoreCancelledSpeech(): void { + const source = this.#cancelledSpeechSource; + if (!source || this.#activeEpoch === null) { + return; + } + this.#cancelledSpeechSource = null; + this.#prepareAndDeliver(source); } public updateChat(update: ChatUpdate): void { @@ -296,7 +322,7 @@ export class RealtimeBrunchBridge { } const source = update.automaticSource; if (source && this.#sourceMatchesSegments(source, newSegments)) { - this.#prepareAndDeliver(source, { kind: "automatic" }); + this.#prepareAndDeliver(source); } else { this.#session.speakCanonical(newSegments); } @@ -317,144 +343,84 @@ export class RealtimeBrunchBridge { ): void { ++this.#generation; this.#activeSubmission = null; - this.#argumentDeltas.clear(); + this.#cancelledSpeechSource = null; + this.#pendingSpeechSource = null; this.#emit({ code, message, type: "error" }); } #handleSessionEvent(event: OpenAIRealtimeSessionEvent): void { - if ( - !("connectionEpoch" in event) || - event.connectionEpoch !== this.#activeEpoch - ) { + // Only completed user transcripts and transcription failures can affect the + // interview. Everything else (including any unexpected legacy tool events) + // is ignored, so model-generated content can never become a user answer. + if (event.type !== "completed" && event.type !== "transcription-failed") { return; } - if (event.type === "response-terminal") { - this.#handleResponseTerminal(event); + if (event.key.connectionEpoch !== this.#activeEpoch) { return; } - if ( - event.type !== "tool-arguments-delta" && - event.type !== "tool-arguments-done" - ) { + const transcriptId = transcriptIdentity(event.key); + if (this.#processedTranscripts.has(transcriptId)) { + this.#emit({ reason: "duplicate", type: "transcript-rejected" }); return; } - - const responseKey = `${event.connectionEpoch}:${event.responseId}`; - if (this.#terminalResponseIds.has(responseKey)) { + this.#processedTranscripts.add(transcriptId); + const question = latestPendingQuestion(this.#chat.canonicalSegments); + if ( + this.#activeSubmission || + !this.#chat.canAcceptInterviewAnswer || + (!question && this.#chat.status !== "ready") + ) { + this.#emit({ reason: "unavailable", type: "transcript-rejected" }); 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, - }); - } + if (event.type === "transcription-failed") { + this.#emit({ reason: "failed", type: "transcript-rejected" }); 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); + const answer = normalizeTranscript(event.text); + if (!answer) { + this.#emit({ reason: "empty", type: "transcript-rejected" }); 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); + if (Array.from(answer).length > ANSWER_LIMIT) { + this.#emit({ reason: "too-long", type: "transcript-rejected" }); return; } const generation = this.#generation; + this.#cancelledSpeechSource = null; this.#activeSubmission = { baselineSegmentIds: new Set( this.#chat.canonicalSegments.map(({ id }) => id), ), - callId: event.callId, correlated: false, - epoch: event.connectionEpoch, + epoch: event.key.connectionEpoch, pendingQuestionId: question?.partId ?? null, sawBusyChatStatus: false, + transcriptId, }; - 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); + this.#emit({ answer, type: "submission-started" }); + void this.#submit(event.key, transcriptId, answer, generation); } async #submit( - event: Extract, + key: OpenAIRealtimeTranscriptKey, + transcriptId: string, answer: string, generation: number, ): Promise { try { const result = await this.#submitInterviewAnswer({ - id: createRealtimeSubmissionId(event.connectionEpoch, event.callId), + id: createRealtimeSubmissionId(key), text: answer, }); const active = this.#activeSubmission; if ( generation !== this.#generation || !active || - active.callId !== event.callId || - active.epoch !== event.connectionEpoch + active.transcriptId !== transcriptId || + active.epoch !== key.connectionEpoch ) { return; } @@ -468,11 +434,7 @@ export class RealtimeBrunchBridge { return; } active.correlated = true; - this.#emit({ - answer, - callId: event.callId, - type: "submission-accepted", - }); + this.#emit({ answer, type: "submission-accepted" }); this.#completeCorrelatedSubmission(); } catch { if (generation === this.#generation) { @@ -506,23 +468,16 @@ export class RealtimeBrunchBridge { this.#activeSubmission = null; const source = this.#chat.automaticSource; if (source && this.#sourceMatchesSegments(source, responseSegments)) { - this.#prepareAndDeliver(source, { - callId: active.callId, - kind: "function-call", - }); + this.#prepareAndDeliver(source); } else { try { - this.#session.completeFunctionCall( - active.callId, - responseSegments.map(({ text }) => text), - ); + this.#session.speakCanonical(responseSegments); } catch { this.#fail(INVALID_BRIDGE_EVENT); return; } } this.#emit({ - callId: active.callId, segments: responseSegments, type: "canonical-response-ready", }); @@ -540,13 +495,12 @@ export class RealtimeBrunchBridge { ); } - #prepareAndDeliver( - source: InterviewSpeechSource, - delivery: SpeechDelivery, - ): void { - this.#emit({ type: "speech-delivery-pending" }); + #prepareAndDeliver(source: InterviewSpeechSource): void { const generation = this.#generation; const speechGeneration = this.#speechGeneration; + this.#cancelledSpeechSource = null; + this.#pendingSpeechSource = source; + this.#emit({ type: "speech-delivery-pending" }); const questionWordCount = source.questionSegment ? spokenWordCount(source.questionSegment.text) : 0; @@ -569,8 +523,8 @@ export class RealtimeBrunchBridge { }, source: questionOnlySource, }).text, - delivery, ); + this.#pendingSpeechSource = null; return; } @@ -591,8 +545,8 @@ export class RealtimeBrunchBridge { }, source, }).text, - delivery, ); + this.#pendingSpeechSource = null; return; } void this.#session.prepareInterviewSpeech(request).then((preparation) => { @@ -604,36 +558,21 @@ export class RealtimeBrunchBridge { source, }); if (speechGeneration !== this.#speechGeneration) { - if (delivery.kind === "function-call") { - try { - this.#session.completeFunctionCall( - delivery.callId, - source.fullResponseSegments.map(({ text }) => text), - { speakResponse: false }, - ); - } catch { - this.#fail(INVALID_BRIDGE_EVENT); - } - } + // Speech was cancelled (new input, pause, or replay) while preparing; + // the caller owns the next utterance. return; } + this.#pendingSpeechSource = null; if (preparation.kind === "prepared") { this.#preparedContextCache.set(request.cacheKey, preparation.context); } - this.#deliverPreparedSpeech(preparedSpeech.text, delivery); + this.#deliverPreparedSpeech(preparedSpeech.text); }); } - #deliverPreparedSpeech( - responseText: readonly string[], - delivery: SpeechDelivery, - ): void { + #deliverPreparedSpeech(responseText: readonly string[]): void { try { - if (delivery.kind === "function-call") { - this.#session.completeFunctionCall(delivery.callId, responseText); - } else { - this.#session.speakPrepared(responseText); - } + this.#session.speakPrepared(responseText); } catch { this.#fail(INVALID_BRIDGE_EVENT); } diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-preview.integration.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-preview.integration.test.ts index a45eee6542a..7b857490f34 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-preview.integration.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-preview.integration.test.ts @@ -143,7 +143,7 @@ const fallbackMessages = [ ] satisfies PetrinautAiMessage[]; describe("controlled voice preview", () => { - test("bridges one Realtime tool call through Brunch and back to canonical duplex audio", async () => { + test("bridges one completed user transcript through Brunch and back to canonical half-duplex audio", async () => { const responseMessagesSnapshot = structuredClone(responseMessages); const diagnostics: VoiceDiagnosticEvent[] = []; const reportDiagnostic = (event: VoiceDiagnosticEvent) => @@ -276,60 +276,91 @@ describe("controlled voice preview", () => { type: "output_audio_buffer.started", }); dataChannel.receive({ - audio_start_ms: 300, - item_id: "user-item", - type: "input_audio_buffer.speech_started", + response_id: "response-initial-question", + type: "output_audio_buffer.stopped", }); dataChannel.receive({ response: { id: "response-initial-question", - status: "cancelled", + output: [], + status: "completed", }, type: "response.done", }); expect(controller.getSnapshot()).toMatchObject({ input: "listening", microphoneEnabled: true, - output: "interrupted", + output: "idle", }); + // Silence or noise: Realtime completes an empty transcript. Nothing is + // submitted and the session keeps listening with a recoverable notice. + dataChannel.receive({ + content_index: 0, + item_id: "noise-item", + transcript: " ", + type: "conversation.item.input_audio_transcription.completed", + }); + await Promise.resolve(); + expect(submitInterviewAnswer).not.toHaveBeenCalled(); + expect(controller.getSnapshot()).toMatchObject({ + input: "listening", + inputNotice: "not-heard", + lastCommittedText: "", + }); + + // A legacy model-generated tool argument stream must never become speech. dataChannel.receive({ call_id: "call-1", - delta: `{"answer":"${spokenAnswer}"}`, + delta: '{"answer":"hi"}', item_id: "function-item-1", output_index: 0, response_id: "response-tool-1", type: "response.function_call_arguments.delta", }); + await Promise.resolve(); + expect(submitInterviewAnswer).not.toHaveBeenCalled(); + dataChannel.receive({ - response: { - id: "response-tool-1", - output: [ - { - arguments: `{"answer":"${spokenAnswer}"}`, - call_id: "call-1", - id: "function-item-1", - name: "continue_interview", - status: "completed", - type: "function_call", - }, - ], - status: "completed", - }, - type: "response.done", + audio_start_ms: 300, + item_id: "user-item", + type: "input_audio_buffer.speech_started", + }); + dataChannel.receive({ + content_index: 0, + delta: "The supervisor", + item_id: "user-item", + type: "conversation.item.input_audio_transcription.delta", + }); + expect(controller.getSnapshot().partialText).toBe("The supervisor"); + dataChannel.receive({ + content_index: 0, + item_id: "user-item", + transcript: spokenAnswer, + type: "conversation.item.input_audio_transcription.completed", + }); + dataChannel.receive({ + content_index: 0, + item_id: "user-item", + transcript: spokenAnswer, + type: "conversation.item.input_audio_transcription.completed", }); await vi.waitFor(() => expect(submitInterviewAnswer).toHaveBeenCalledWith({ - id: "voice-realtime:1:call-1", + id: "voice-realtime:1:user-item:0", text: spokenAnswer, }), ); + expect(submitInterviewAnswer).toHaveBeenCalledOnce(); expect(controller.getSnapshot()).toMatchObject({ input: "submitting", + inputNotice: "none", lastAnswerDelivery: "delivered", + lastCommittedText: spokenAnswer, microphoneEnabled: true, output: "waiting-for-tool", + partialText: "", }); const pendingSpeech = selectInterviewSpeech(initialMessages); @@ -396,35 +427,32 @@ describe("controlled voice preview", () => { }); await vi.waitFor(() => - expect(sentEvents(dataChannel)).toContainEqual( - expect.objectContaining({ type: "conversation.item.create" }), - ), + expect(sentEvents(dataChannel).at(-1)).toMatchObject({ + response: { metadata: { petrinaut_kind: "canonical-speech" } }, + type: "response.create", + }), ); - const [functionOutput, responseCreate] = sentEvents(dataChannel).slice(-2); - expect(functionOutput).toEqual({ - type: "conversation.item.create", - item: { - type: "function_call_output", - call_id: "call-1", - output: JSON.stringify({ - response_text: [preparedReply, canonicalQuestion], - }), - }, - }); + const responseCreate = sentEvents(dataChannel).at(-1)!; expect(responseCreate).toMatchObject({ type: "response.create", response: { + conversation: "none", output_modalities: ["audio"], tool_choice: "none", tools: [], }, }); - expect(JSON.stringify([functionOutput, responseCreate])).not.toContain( - toolMetadata, - ); - expect(JSON.stringify([functionOutput, responseCreate])).not.toContain( - toolError, + const speechResponse = responseCreate.response as { + input: Array<{ content: Array<{ text: string }> }>; + }; + expect(JSON.parse(speechResponse.input[0]!.content[0]!.text)).toEqual({ + response_text: [preparedReply, canonicalQuestion], + }); + expect(JSON.stringify(responseCreate)).not.toContain(toolMetadata); + expect(JSON.stringify(responseCreate)).not.toContain(toolError); + expect(sentEvents(dataChannel)).not.toContainEqual( + expect.objectContaining({ type: "conversation.item.create" }), ); authorizeLatestSpeechResponse(dataChannel, "response-canonical-reply"); @@ -563,6 +591,35 @@ describe("controlled voice preview", () => { ).toEqual({ response_text: [fallbackReply, fallbackQuestion] }); expect(responseMessages).toEqual(responseMessagesSnapshot); + // A fabricated `continue_interview` call from Realtime fails closed: the + // session disconnects rather than letting the argument become an answer. + dataChannel.receive({ + response: { + id: "response-fabricated-tool", + output: [ + { + arguments: '{"answer":"one"}', + call_id: "call-fabricated", + id: "function-item-fabricated", + name: "continue_interview", + status: "completed", + type: "function_call", + }, + ], + status: "completed", + }, + type: "response.done", + }); + await Promise.resolve(); + expect(submitInterviewAnswer).toHaveBeenCalledOnce(); + expect(controller.getSnapshot()).toMatchObject({ + connection: "error", + errorCode: "invalid-response", + }); + expect(sentEvents(dataChannel)).not.toContainEqual( + expect.objectContaining({ type: "conversation.item.create" }), + ); + expect(browserRequests).toEqual([ { path: "/api/voice/realtime-call", @@ -580,8 +637,8 @@ describe("controlled voice preview", () => { turn_detection: { type: "semantic_vad", eagerness: "low", - create_response: true, - interrupt_response: true, + create_response: false, + interrupt_response: false, }, }, }, diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-session-state.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-session-state.test.ts index c24b89af036..a0f878c29d5 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-session-state.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-session-state.test.ts @@ -14,6 +14,7 @@ const listeningSnapshot = { errorMessage: "", errorRequestId: "", input: "listening", + inputNotice: "none", lastAnswerDelivery: "none", lastCommittedText: "", microphoneEnabled: true, diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts index b2a730ea3d9..513a768a77a 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts @@ -33,6 +33,7 @@ const createHarness = () => { }; const bridge = { cancelPendingSpeech: vi.fn(), + restoreCancelledSpeech: vi.fn(), start: vi.fn(), stop: vi.fn(), subscribe: vi.fn((listener: (event: RealtimeBrunchBridgeEvent) => void) => { @@ -84,7 +85,7 @@ const speechSource = (): InterviewSpeechSource => { }; describe("VoiceTurnController", () => { - test("opens a continuous microphone before starting canonical question speech", async () => { + test("requests microphone input before starting canonical question speech", async () => { const harness = createHarness(); const order: string[] = []; harness.session.setMicrophoneEnabled.mockImplementation((enabled) => { @@ -109,7 +110,7 @@ describe("VoiceTurnController", () => { }); }); - test("keeps capture active while the interviewer speaks and interrupts automatically", async () => { + test("tracks playback without changing the user's requested microphone state", async () => { const harness = createHarness(); await harness.controller.start(); @@ -123,18 +124,6 @@ describe("VoiceTurnController", () => { microphoneEnabled: true, output: "speaking", }); - - harness.emitSession({ - connectionEpoch: 1, - itemId: "item-user", - type: "input-speech-started", - }); - expect(harness.controller.getSnapshot()).toMatchObject({ - microphoneEnabled: true, - output: "interrupted", - }); - expect(harness.session.cancelOutput).not.toHaveBeenCalled(); - expect(harness.bridge.cancelPendingSpeech).toHaveBeenCalledOnce(); }); test("invalidates pending preparation before pausing or cancelling paused output", async () => { @@ -477,7 +466,6 @@ describe("VoiceTurnController", () => { harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-started", }); expect(harness.controller.getSnapshot()).toMatchObject({ @@ -489,11 +477,9 @@ describe("VoiceTurnController", () => { }); harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-accepted", }); harness.emitBridge({ - callId: "call-1", segments: [question("ask-2", "Who acts next?")], type: "canonical-response-ready", }); @@ -514,7 +500,6 @@ describe("VoiceTurnController", () => { await harness.controller.start(); harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-started", }); @@ -529,11 +514,9 @@ describe("VoiceTurnController", () => { harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-accepted", }); harness.emitBridge({ - callId: "call-1", segments: [question("ask-2", "Who acts next?")], type: "canonical-response-ready", }); @@ -554,7 +537,6 @@ describe("VoiceTurnController", () => { await harness.controller.start(); harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-started", }); @@ -565,11 +547,9 @@ describe("VoiceTurnController", () => { }); harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-accepted", }); harness.emitBridge({ - callId: "call-1", segments: [question("ask-2", "Who acts next?")], type: "canonical-response-ready", }); @@ -591,13 +571,11 @@ describe("VoiceTurnController", () => { await harness.controller.start(); harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-started", }); harness.controller.pause(); harness.emitBridge({ - callId: "call-1", segments: [question("ask-2", "Who acts next?")], type: "canonical-response-ready", }); @@ -659,11 +637,9 @@ describe("VoiceTurnController", () => { }); harness.emitBridge({ answer: "First answer", - callId: "call-1", type: "submission-started", }); harness.emitBridge({ - callId: "call-1", segments: [question("ask-2", "Who acts next?")], type: "canonical-response-ready", }); @@ -728,6 +704,144 @@ describe("VoiceTurnController", () => { expect(harness.controller.getSnapshot().partialText).toBe("Current answer"); }); + test.each(["empty", "failed"] as const)( + "returns to listening with a recoverable notice when a transcript is rejected as %s", + async (reason) => { + const harness = createHarness(); + const source = speechSource(); + harness.controller.updateChat({ + automaticSource: source, + canAcceptInterviewAnswer: true, + canonicalSegments: [...source.fullResponseSegments], + status: "ready", + }); + await harness.controller.start(); + harness.emitSession({ + connectionEpoch: 1, + itemId: "item-1", + type: "input-speech-started", + }); + harness.emitSession({ + key: { connectionEpoch: 1, contentIndex: 0, itemId: "item-1" }, + text: "um", + type: "partial", + }); + + harness.emitBridge({ reason, type: "transcript-rejected" }); + + expect(harness.controller.getSnapshot()).toMatchObject({ + canReadFullResponse: true, + canRepeatQuestion: true, + connection: "connected", + currentQuestion: "Who approves release?", + input: "listening", + inputNotice: "not-heard", + lastAnswerDelivery: "none", + lastCommittedText: "", + output: "idle", + partialText: "", + }); + expect(harness.bridge.restoreCancelledSpeech).toHaveBeenCalledOnce(); + expect(harness.submitText).not.toHaveBeenCalled(); + + harness.emitSession({ + connectionEpoch: 1, + itemId: "item-2", + type: "input-speech-started", + }); + expect(harness.controller.getSnapshot().inputNotice).toBe("none"); + }, + ); + + test("keeps the current state when a transcript is rejected as duplicate or unavailable", async () => { + const harness = createHarness(); + await harness.controller.start(); + harness.emitBridge({ answer: "First answer", type: "submission-started" }); + + harness.emitBridge({ reason: "unavailable", type: "transcript-rejected" }); + harness.emitBridge({ reason: "duplicate", type: "transcript-rejected" }); + + expect(harness.controller.getSnapshot()).toMatchObject({ + input: "submitting", + inputNotice: "none", + lastAnswerDelivery: "pending", + lastCommittedText: "First answer", + }); + expect(harness.bridge.restoreCancelledSpeech).not.toHaveBeenCalled(); + }); + + test.each(["duplicate", "unavailable"] as const)( + "keeps the active input turn when a transcript is rejected as %s", + async (reason) => { + const harness = createHarness(); + await harness.controller.start(); + harness.emitSession({ + connectionEpoch: 1, + itemId: "item-1", + type: "input-speech-started", + }); + harness.emitSession({ + key: { connectionEpoch: 1, contentIndex: 0, itemId: "item-1" }, + text: "Current ", + type: "partial", + }); + + harness.emitBridge({ reason, type: "transcript-rejected" }); + harness.emitSession({ + key: { connectionEpoch: 1, contentIndex: 0, itemId: "item-1" }, + text: "answer", + type: "completed", + }); + + expect(harness.controller.getSnapshot()).toMatchObject({ + input: "listening", + inputNotice: "none", + partialText: "answer", + }); + expect(harness.bridge.restoreCancelledSpeech).not.toHaveBeenCalled(); + }, + ); + + test.each(["empty", "failed"] as const)( + "does not restore rejected speech while paused for %s input", + async (reason) => { + const harness = createHarness(); + await harness.controller.start(); + harness.emitSession({ + connectionEpoch: 1, + itemId: "item-1", + type: "input-speech-started", + }); + harness.controller.pause(); + + harness.emitBridge({ reason, type: "transcript-rejected" }); + + expect(harness.controller.getSnapshot()).toMatchObject({ + input: "paused", + inputNotice: "not-heard", + }); + expect(harness.bridge.restoreCancelledSpeech).not.toHaveBeenCalled(); + }, + ); + + test("clears the not-heard notice when an answer is submitted while paused", async () => { + const harness = createHarness(); + await harness.controller.start(); + harness.emitBridge({ reason: "empty", type: "transcript-rejected" }); + harness.controller.pause(); + expect(harness.controller.getSnapshot()).toMatchObject({ + input: "paused", + inputNotice: "not-heard", + }); + + harness.emitBridge({ answer: "Answer", type: "submission-started" }); + + expect(harness.controller.getSnapshot()).toMatchObject({ + input: "paused", + inputNotice: "none", + }); + }); + test("pauses and resumes input without ending the connection", async () => { const harness = createHarness(); await harness.controller.start(); @@ -972,7 +1086,6 @@ describe("VoiceTurnController", () => { await harness.controller.start(); harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-started", }); harness.emitBridge({ @@ -1041,7 +1154,6 @@ describe("VoiceTurnController", () => { await bridgeFailure.controller.start(); bridgeFailure.emitBridge({ answer: "Pending answer", - callId: "call-1", type: "submission-started", }); bridgeFailure.emitBridge({ diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts index 666f9745ffc..171d86fa120 100644 --- a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts +++ b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.ts @@ -22,6 +22,13 @@ export type VoiceOutputState = | "speaking" | "interrupted"; export type VoiceAnswerDelivery = "none" | "pending" | "delivered" | "failed"; +/** + * Recoverable notice about the user's last utterance. `not-heard` means the + * audio produced no usable transcript (silence, noise, or a transcription + * failure): nothing was submitted and the session is listening again. It clears + * as soon as the user starts speaking. + */ +export type VoiceInputNotice = "none" | "not-heard"; export interface VoiceTurnSnapshot { readonly canReadFullResponse: boolean; @@ -33,6 +40,7 @@ export interface VoiceTurnSnapshot { readonly errorMessage: string; readonly errorRequestId: string; readonly input: VoiceInputState; + readonly inputNotice: VoiceInputNotice; readonly lastAnswerDelivery: VoiceAnswerDelivery; readonly lastCommittedText: string; readonly microphoneEnabled: boolean; @@ -64,6 +72,7 @@ interface RealtimeSession { interface RealtimeBridge { cancelPendingSpeech(): void; + restoreCancelledSpeech(): void; start(connectionEpoch: number): void; stop(): void; subscribe(listener: (event: RealtimeBrunchBridgeEvent) => void): () => void; @@ -102,6 +111,7 @@ const initialSnapshot: VoiceTurnSnapshot = { errorMessage: "", errorRequestId: "", input: "paused", + inputNotice: "none", lastAnswerDelivery: "none", lastCommittedText: "", microphoneEnabled: false, @@ -446,6 +456,7 @@ export class VoiceTurnController { this.#transcriptKey = null; this.#update({ input: paused ? "paused" : "submitting", + inputNotice: "none", lastAnswerDelivery: "pending", lastCommittedText: event.answer, output: "waiting-for-tool", @@ -453,10 +464,36 @@ export class VoiceTurnController { }); return; } + if (event.type === "transcript-rejected") { + // Duplicate redelivery and overlap with an in-flight answer are + // diagnostics only. They may refer to an older transcript, so must not + // disturb the current input turn or restore its cancelled speech. + if (event.reason === "duplicate" || event.reason === "unavailable") { + return; + } + + // Nothing was submitted, so the interview is unchanged. Silence, noise, + // or a transcription failure is surfaced as a recoverable notice. + this.#inputTurnPending = false; + this.#transcriptItemId = null; + this.#transcriptKey = null; + if (event.reason === "empty" || event.reason === "failed") { + this.#update({ inputNotice: "not-heard", partialText: "" }); + } else { + this.#update({}); + } + if (this.#snapshot.input !== "paused") { + this.#bridge.restoreCancelledSpeech(); + } + return; + } if (event.type === "submission-accepted") { this.#answeredQuestionId = this.#submittingQuestionId; this.#submittingQuestionId = null; - this.#update({ lastAnswerDelivery: "delivered" }); + this.#update({ + inputNotice: "none", + lastAnswerDelivery: "delivered", + }); return; } if (event.type === "speech-delivery-pending") { @@ -541,9 +578,13 @@ export class VoiceTurnController { this.#transcriptItemId = event.itemId; this.#transcriptKey = null; if (this.#snapshot.output !== "idle") { - this.#update({ output: "interrupted", partialText: "" }); + this.#update({ + inputNotice: "none", + output: "interrupted", + partialText: "", + }); } else { - this.#update({ partialText: "" }); + this.#update({ inputNotice: "none", partialText: "" }); } return; } @@ -560,11 +601,7 @@ export class VoiceTurnController { } return; } - if ( - event.type === "input-speech-stopped" || - event.type === "tool-arguments-delta" || - event.type === "tool-arguments-done" - ) { + if (event.type === "input-speech-stopped") { return; } diff --git a/apps/petrinaut-website/src/server/voice/openai-realtime-call.test.ts b/apps/petrinaut-website/src/server/voice/openai-realtime-call.test.ts index 38188a178c3..c7f4f94c284 100644 --- a/apps/petrinaut-website/src/server/voice/openai-realtime-call.test.ts +++ b/apps/petrinaut-website/src/server/voice/openai-realtime-call.test.ts @@ -86,7 +86,7 @@ describe("OpenAI Realtime call handler", () => { expect(fetch).not.toHaveBeenCalled(); }); - test("forwards only the SDP and server-owned duplex Realtime policy", async () => { + test("forwards only the SDP and server-owned half-duplex Realtime policy", async () => { const reportDiagnostic = vi.fn(); const fetch = vi.fn( async () => @@ -131,16 +131,16 @@ describe("OpenAI Realtime call handler", () => { type: "realtime", model: "gpt-realtime-2", output_modalities: ["audio"], - tool_choice: "required", - tools: [{ name: "continue_interview", type: "function" }], + tool_choice: "none", + tools: [], audio: { input: { transcription: { model: "gpt-4o-transcribe", language: "en" }, turn_detection: { type: "semantic_vad", eagerness: "low", - create_response: true, - interrupt_response: true, + create_response: false, + interrupt_response: false, }, }, }, diff --git a/apps/petrinaut-website/src/server/voice/openai-voice-policy.test.ts b/apps/petrinaut-website/src/server/voice/openai-voice-policy.test.ts index 1cdd3f0fead..523b8c85f17 100644 --- a/apps/petrinaut-website/src/server/voice/openai-voice-policy.test.ts +++ b/apps/petrinaut-website/src/server/voice/openai-voice-policy.test.ts @@ -46,18 +46,18 @@ describe("OpenAI voice policy", () => { ).toEqual({ available: true, connectionTimeoutMs: 15_000 }); }); - test("owns the trusted GPT-Realtime-2 duplex session policy", () => { - expect(OPENAI_REALTIME_POLICY_VERSION).toBe("brunch-control-plane-v1"); + test("owns the trusted GPT-Realtime-2 half-duplex session policy", () => { + expect(OPENAI_REALTIME_POLICY_VERSION).toBe("brunch-control-plane-v3"); expect(createOpenAIRealtimeSession()).toEqual({ type: "realtime", model: "gpt-realtime-2", output_modalities: ["audio"], reasoning: { effort: "low" }, parallel_tool_calls: false, - tool_choice: "required", + tool_choice: "none", instructions: `# Role and objective -You are the realtime voice of an expert interviewer for process-model elicitation. The person speaking is the domain expert. Listen attentively, submit each complete spoken answer to Brunch, and deliver Brunch's next interview turn. +You are the realtime voice of an expert interviewer for process-model elicitation. The person speaking is the domain expert. Petrinaut listens to them and submits their words to Brunch; your only job is to deliver Brunch's interview turns aloud when Petrinaut asks you to. # Personality and delivery @@ -65,29 +65,16 @@ Sound warm, calm, curious, confident, concise, and professionally neutral. Speak # Authority -Brunch is the sole authority for interview state, questions, captures, completion, and business decisions. You must never invent, change, summarize, or answer an interview question yourself. +Brunch is the sole authority for interview state, questions, captures, completion, and business decisions. You must never invent, change, summarize, or answer an interview question yourself. You must never restate, guess, or fill in what the speaker said. # Turn handling -After semantic turn detection finds that the user has finished a complete spoken answer, call continue_interview exactly once with that answer. Do not speak, emit a preamble, or emit conversational text before calling the tool. +Never respond on your own after the speaker stops talking. Petrinaut transcribes their words and decides what happens next. Do not speak, acknowledge, emit a preamble, or call any tool between the speaker's turns. # Canonical output -After the tool result arrives, speak only its response_text strings, in array order and verbatim. Do not add, remove, paraphrase, acknowledge, or explain anything. Never call another tool while speaking a tool result.`, - tools: [ - { - type: "function", - name: "continue_interview", - description: - "Submit the user's complete spoken answer to the authoritative Brunch interview.", - parameters: { - type: "object", - additionalProperties: false, - properties: { answer: { type: "string" } }, - required: ["answer"], - }, - }, - ], +When Petrinaut supplies response_text, speak only those strings, in array order and verbatim. Do not add, remove, paraphrase, acknowledge, or explain anything.`, + tools: [], audio: { input: { noise_reduction: { type: "far_field" }, @@ -100,8 +87,8 @@ After the tool result arrives, speak only its response_text strings, in array or turn_detection: { type: "semantic_vad", eagerness: "low", - create_response: true, - interrupt_response: true, + create_response: false, + interrupt_response: false, }, }, output: { voice: "marin" }, @@ -109,12 +96,17 @@ After the tool result arrives, speak only its response_text strings, in array or }); }); - test("allows no provider-owned interview decisions or unrestricted tools", () => { - const serializedPolicy = JSON.stringify(createOpenAIRealtimeSession()); + test("lets Realtime neither answer for the user nor call tools between turns", () => { + const policy = createOpenAIRealtimeSession(); + const serializedPolicy = JSON.stringify(policy); expect(serializedPolicy).not.toContain("response.create"); expect(serializedPolicy).not.toContain("gpt-realtime-1.5"); + expect(serializedPolicy).not.toContain("continue_interview"); expect(serializedPolicy).not.toContain('"tool_choice":"auto"'); - expect(createOpenAIRealtimeSession().tools).toHaveLength(1); + expect(serializedPolicy).not.toContain('"tool_choice":"required"'); + expect(policy.tools).toHaveLength(0); + expect(policy.audio.input.turn_detection.create_response).toBe(false); + expect(policy.audio.input.transcription.model).toBe("gpt-4o-transcribe"); }); }); diff --git a/apps/petrinaut-website/src/server/voice/openai-voice-policy.ts b/apps/petrinaut-website/src/server/voice/openai-voice-policy.ts index fc401fc0c7b..321d114f9f4 100644 --- a/apps/petrinaut-website/src/server/voice/openai-voice-policy.ts +++ b/apps/petrinaut-website/src/server/voice/openai-voice-policy.ts @@ -1,5 +1,5 @@ export const OPENAI_REALTIME_CONNECTION_TIMEOUT_MS = 15_000; -export const OPENAI_REALTIME_POLICY_VERSION = "brunch-control-plane-v1"; +export const OPENAI_REALTIME_POLICY_VERSION = "brunch-control-plane-v3"; interface VoiceEnvironment { readonly NODE_ENV?: string; @@ -24,7 +24,7 @@ export const getOpenAIVoiceAvailability = (environment: VoiceEnvironment) => ({ const REALTIME_INSTRUCTIONS = `# Role and objective -You are the realtime voice of an expert interviewer for process-model elicitation. The person speaking is the domain expert. Listen attentively, submit each complete spoken answer to Brunch, and deliver Brunch's next interview turn. +You are the realtime voice of an expert interviewer for process-model elicitation. The person speaking is the domain expert. Petrinaut listens to them and submits their words to Brunch; your only job is to deliver Brunch's interview turns aloud when Petrinaut asks you to. # Personality and delivery @@ -32,38 +32,33 @@ Sound warm, calm, curious, confident, concise, and professionally neutral. Speak # Authority -Brunch is the sole authority for interview state, questions, captures, completion, and business decisions. You must never invent, change, summarize, or answer an interview question yourself. +Brunch is the sole authority for interview state, questions, captures, completion, and business decisions. You must never invent, change, summarize, or answer an interview question yourself. You must never restate, guess, or fill in what the speaker said. # Turn handling -After semantic turn detection finds that the user has finished a complete spoken answer, call continue_interview exactly once with that answer. Do not speak, emit a preamble, or emit conversational text before calling the tool. +Never respond on your own after the speaker stops talking. Petrinaut transcribes their words and decides what happens next. Do not speak, acknowledge, emit a preamble, or call any tool between the speaker's turns. # Canonical output -After the tool result arrives, speak only its response_text strings, in array order and verbatim. Do not add, remove, paraphrase, acknowledge, or explain anything. Never call another tool while speaking a tool result.`; +When Petrinaut supplies response_text, speak only those strings, in array order and verbatim. Do not add, remove, paraphrase, acknowledge, or explain anything.`; +/** + * The completed `gpt-4o-transcribe` transcript is the only source of the + * user's answer. Semantic VAD therefore only commits audio and never creates a + * response or interrupts playback, and no tool exists for the model to invent + * an answer through. + * Canonical speech and speech preparation are requested explicitly, out of + * band, with tools disabled at the response level. + */ export const createOpenAIRealtimeSession = () => ({ type: "realtime" as const, model: "gpt-realtime-2", output_modalities: ["audio"] as const, reasoning: { effort: "low" as const }, parallel_tool_calls: false, - tool_choice: "required" as const, + tool_choice: "none" as const, instructions: REALTIME_INSTRUCTIONS, - tools: [ - { - type: "function" as const, - name: "continue_interview", - description: - "Submit the user's complete spoken answer to the authoritative Brunch interview.", - parameters: { - type: "object" as const, - additionalProperties: false, - properties: { answer: { type: "string" as const } }, - required: ["answer"] as const, - }, - }, - ], + tools: [] as const, audio: { input: { noise_reduction: { type: "far_field" as const }, @@ -76,8 +71,8 @@ export const createOpenAIRealtimeSession = () => ({ turn_detection: { type: "semantic_vad" as const, eagerness: "low" as const, - create_response: true, - interrupt_response: true, + create_response: false, + interrupt_response: false, }, }, output: { voice: "marin" as const }, diff --git a/libs/@hashintel/petrinaut/docs/ai-assistant.md b/libs/@hashintel/petrinaut/docs/ai-assistant.md index 6fca195ecfa..53939f99f3a 100644 --- a/libs/@hashintel/petrinaut/docs/ai-assistant.md +++ b/libs/@hashintel/petrinaut/docs/ai-assistant.md @@ -62,10 +62,10 @@ Realtime audio are ephemeral. Finalized spoken user messages carry a small **Voi the words themselves, and the exact inline answer completed by speech carries the same chip, so Voice provenance remains visible without duplicating an answer. -The microphone stays on while the interviewer speaks, so speaking naturally interrupts the audio -and starts listening to you; you do not need to select an interrupt action. Semantic voice detection -finishes each answer automatically after a natural pause and is tuned to allow longer thinking -pauses. There is no required done-speaking action. +Microphone capture pauses while the interviewer speaks so speaker echo cannot be submitted as your +answer. Wait for playback to finish; Voice resumes listening automatically in your previous mute +state. Semantic voice detection finishes each answer after a natural pause and is tuned to allow +longer thinking pauses. There is no required done-speaking action. Every session control lives in the dock: **Show transcription in chat** on the left, and on the right **Mute microphone** (**Unmute microphone** once muted) beside **End voice mode**. Muting stops