From 947cafec532d19c0d266ba27cc0b44a36cfd65d4 Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Wed, 2 Sep 2026 16:39:46 +0200 Subject: [PATCH 1/6] Submit only completed Voice transcripts as interview answers Realtime previously generated a continue_interview tool call after every semantic VAD commit, so silence or noise could produce model-invented answers. Disable inbound response creation and tools, route completed gpt-4o-transcribe transcripts to the Brunch bridge keyed by connection epoch, item ID, and content index, submit each accepted transcript exactly once, and surface empty or failed input as a recoverable not-heard notice. Amp-Thread-ID: https://ampcode.com/threads/T-01a0626d-0c66-725e-9b03-afe14d779f2d Co-authored-by: Amp --- apps/petrinaut-website/README.md | 14 +- .../openai-realtime-session.test.ts | 126 ++-- .../openai-realtime-session.ts | 151 +---- .../realtime-brunch-bridge.test.ts | 539 ++++++++++++------ .../voice-interview/realtime-brunch-bridge.ts | 276 +++------ .../voice-preview.integration.test.ts | 128 +++-- .../voice-session-state.test.ts | 1 + .../voice-turn-controller.test.ts | 93 ++- .../voice-interview/voice-turn-controller.ts | 34 +- .../server/voice/openai-realtime-call.test.ts | 6 +- .../server/voice/openai-voice-policy.test.ts | 40 +- .../src/server/voice/openai-voice-policy.ts | 36 +- 12 files changed, 775 insertions(+), 669 deletions(-) diff --git a/apps/petrinaut-website/README.md b/apps/petrinaut-website/README.md index 2d47a9d81bc..c781a8d1636 100644 --- a/apps/petrinaut-website/README.md +++ b/apps/petrinaut-website/README.md @@ -113,10 +113,16 @@ 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. +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, the session exposes no tools, and +the model never infers or paraphrases what the speaker said. The browser bridge +normalizes whitespace, submits each 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 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..63b686792ab 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 @@ -276,36 +276,39 @@ describe("OpenAIRealtimeSession", () => { ); }); - test("parses streamed tool arguments and the completed GA response output", async () => { + 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":"Ignored"}', - call_id: "call-ignored", - item_id: "item-ignored", + arguments: '{"answer":"hi"}', + call_id: "call-legacy", + item_id: "item-legacy", output_index: 0, - response_id: "response-tool", + response_id: "response-legacy", type: "response.function_call_arguments.done", }); channel.receive({ - call_id: "call-1", - delta: '{"answer":"Approved"}', - item_id: "item-function", + call_id: "call-legacy", + delta: '{"answer":"hi"}', + item_id: "item-legacy", output_index: 0, - response_id: "response-tool", + response_id: "response-legacy", type: "response.function_call_arguments.delta", }); + expect(harness.events).toEqual([]); + channel.receive({ response: { - id: "response-tool", + id: "response-legacy", output: [ { - arguments: '{"answer":"Approved"}', - call_id: "call-1", - id: "item-function", + arguments: '{"answer":"hi"}', + call_id: "call-legacy", + id: "item-legacy", name: "continue_interview", + status: "completed", type: "function_call", }, ], @@ -315,74 +318,17 @@ describe("OpenAIRealtimeSession", () => { }); expect(harness.events).toEqual([ - { - callId: "call-1", - connectionEpoch: 1, - delta: '{"answer":"Approved"}', - itemId: "item-function", - responseId: "response-tool", - type: "tool-arguments-delta", - }, - { - arguments: '{"answer":"Approved"}', - callId: "call-1", - connectionEpoch: 1, - itemId: "item-function", - name: "continue_interview", - responseId: "response-tool", - type: "tool-arguments-done", - }, - { - connectionEpoch: 1, - responseId: "response-tool", - status: "completed", - type: "response-terminal", - }, + expect.objectContaining({ code: "invalid-response", type: "error" }), ]); - - 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?"] }), - }, - }); - expect(responseCreate).toMatchObject({ - type: "response.create", - response: { - instructions: - "Speak only the response_text strings supplied by Petrinaut, in array order and verbatim. Deliver them as a warm, calm, curious, confident, concise, and professionally neutral expert interviewer, at a measured conversational pace with natural emphasis. Never sound robotic, fawning, rushed, overenthusiastic, or patronizing. Do not add, remove, paraphrase, acknowledge, or explain anything.", - output_modalities: ["audio"], - parallel_tool_calls: false, - tool_choice: "none", - tools: [], - }, - }); - - 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."], - }), - }, - }); + 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 +1075,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 +1093,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 +1111,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 +1175,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..5f3dbe4c160 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; @@ -473,37 +462,11 @@ export class OpenAIRealtimeSession { } 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 +645,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 +658,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, @@ -857,10 +821,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 +1072,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, @@ -1350,26 +1279,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, 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..e27bed528c6 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,7 @@ describe("RealtimeBrunchBridge", () => { ); }); - test("completes a function call without speech when preparation is cancelled", async () => { + 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 +401,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 +445,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 +494,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 +505,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 +542,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 +625,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); - expect(harness.session.completeFunctionCall).toHaveBeenCalledWith( - "call-1", - [unrelated.text], + 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" }, + ]); + + 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 +724,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 +795,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 +821,81 @@ 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, "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 +903,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 +967,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 +1013,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 +1047,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..bba0cd3199d 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 = { @@ -222,20 +233,18 @@ 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.#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,9 +261,7 @@ export class RealtimeBrunchBridge { ++this.#generation; this.#activeEpoch = null; this.#activeSubmission = null; - this.#argumentDeltas.clear(); this.#preparedContextCache.clear(); - this.#terminalResponseIds.clear(); } public cancelPendingSpeech(): void { @@ -296,7 +303,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,91 +324,46 @@ export class RealtimeBrunchBridge { ): void { ++this.#generation; this.#activeSubmission = null; - this.#argumentDeltas.clear(); this.#emit({ code, message, type: "error" }); } #handleSessionEvent(event: OpenAIRealtimeSessionEvent): void { - if ( - !("connectionEpoch" in event) || - event.connectionEpoch !== this.#activeEpoch - ) { + // 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" - ) { - return; - } - - const responseKey = `${event.connectionEpoch}:${event.responseId}`; - if (this.#terminalResponseIds.has(responseKey)) { - return; - } - const callKey = `${event.connectionEpoch}:${event.callId}`; - if (this.#processedCalls.has(callKey)) { + const transcriptId = transcriptIdentity(event.key); + if (this.#processedTranscripts.has(transcriptId)) { + this.#emit({ reason: "duplicate", type: "transcript-rejected" }); 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, - }); - } + this.#processedTranscripts.add(transcriptId); + 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); + const answer = normalizeTranscript(event.text); + if (!answer) { + this.#emit({ reason: "empty", type: "transcript-rejected" }); 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); + if (Array.from(answer).length > ANSWER_LIMIT) { + this.#emit({ reason: "too-long", type: "transcript-rejected" }); return; } - - const answer = parseContinueInterviewArguments(event.arguments); const question = latestPendingQuestion(this.#chat.canonicalSegments); if ( - !answer || + this.#activeSubmission || !this.#chat.canAcceptInterviewAnswer || (!question && this.#chat.status !== "ready") ) { - this.#fail(INVALID_BRIDGE_EVENT); + this.#emit({ reason: "unavailable", type: "transcript-rejected" }); return; } @@ -410,51 +372,33 @@ export class RealtimeBrunchBridge { 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 +412,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 +446,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,10 +473,7 @@ export class RealtimeBrunchBridge { ); } - #prepareAndDeliver( - source: InterviewSpeechSource, - delivery: SpeechDelivery, - ): void { + #prepareAndDeliver(source: InterviewSpeechSource): void { this.#emit({ type: "speech-delivery-pending" }); const generation = this.#generation; const speechGeneration = this.#speechGeneration; @@ -569,7 +499,6 @@ export class RealtimeBrunchBridge { }, source: questionOnlySource, }).text, - delivery, ); return; } @@ -591,7 +520,6 @@ export class RealtimeBrunchBridge { }, source, }).text, - delivery, ); return; } @@ -604,36 +532,20 @@ 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 (barge-in, pause, or replay) while preparing; + // the caller owns the next utterance. return; } 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..4458902f796 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 duplex audio", async () => { const responseMessagesSnapshot = structuredClone(responseMessages); const diagnostics: VoiceDiagnosticEvent[] = []; const reportDiagnostic = (event: VoiceDiagnosticEvent) => @@ -293,43 +293,69 @@ describe("controlled voice preview", () => { output: "interrupted", }); + // 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", + 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 +422,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 +586,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,7 +632,7 @@ describe("controlled voice preview", () => { turn_detection: { type: "semantic_vad", eagerness: "low", - create_response: true, + create_response: false, interrupt_response: true, }, }, 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..349542a1755 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 @@ -477,7 +477,6 @@ describe("VoiceTurnController", () => { harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-started", }); expect(harness.controller.getSnapshot()).toMatchObject({ @@ -489,11 +488,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 +511,6 @@ describe("VoiceTurnController", () => { await harness.controller.start(); harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-started", }); @@ -529,11 +525,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 +548,6 @@ describe("VoiceTurnController", () => { await harness.controller.start(); harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-started", }); @@ -565,11 +558,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 +582,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 +648,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 +715,84 @@ 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(); + harness.controller.updateChat({ + canAcceptInterviewAnswer: true, + canonicalSegments: [question("ask-1")], + 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({ + connection: "connected", + currentQuestion: "What happens after approval?", + input: "listening", + inputNotice: "not-heard", + lastAnswerDelivery: "none", + lastCommittedText: "", + output: "idle", + partialText: "", + }); + 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", + }); + }); + + 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 +1037,6 @@ describe("VoiceTurnController", () => { await harness.controller.start(); harness.emitBridge({ answer: "The supervisor approves it.", - callId: "call-1", type: "submission-started", }); harness.emitBridge({ @@ -1041,7 +1105,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..71ae6e98849 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; @@ -102,6 +110,7 @@ const initialSnapshot: VoiceTurnSnapshot = { errorMessage: "", errorRequestId: "", input: "paused", + inputNotice: "none", lastAnswerDelivery: "none", lastCommittedText: "", microphoneEnabled: false, @@ -446,6 +455,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,6 +463,16 @@ export class VoiceTurnController { }); return; } + if (event.type === "transcript-rejected") { + // Nothing was submitted, so the interview is unchanged. Silence, noise, + // or a transcription failure is surfaced as a recoverable notice; other + // rejections (duplicate redelivery, overlap with an in-flight answer) + // are diagnostics only. + if (event.reason === "empty" || event.reason === "failed") { + this.#update({ inputNotice: "not-heard", partialText: "" }); + } + return; + } if (event.type === "submission-accepted") { this.#answeredQuestionId = this.#submittingQuestionId; this.#submittingQuestionId = null; @@ -541,9 +561,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 +584,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..db3a13e2512 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 @@ -131,15 +131,15 @@ 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, + create_response: false, interrupt_response: true, }, }, 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..b8906acbd50 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 @@ -47,17 +47,17 @@ describe("OpenAI voice policy", () => { }); test("owns the trusted GPT-Realtime-2 duplex session policy", () => { - expect(OPENAI_REALTIME_POLICY_VERSION).toBe("brunch-control-plane-v1"); + expect(OPENAI_REALTIME_POLICY_VERSION).toBe("brunch-control-plane-v2"); 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,7 +87,7 @@ After the tool result arrives, speak only its response_text strings, in array or turn_detection: { type: "semantic_vad", eagerness: "low", - create_response: true, + create_response: false, interrupt_response: true, }, }, @@ -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..0d9d26e143a 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-v2"; 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,32 @@ 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, 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,7 +70,7 @@ export const createOpenAIRealtimeSession = () => ({ turn_detection: { type: "semantic_vad" as const, eagerness: "low" as const, - create_response: true, + create_response: false, interrupt_response: true, }, }, From c4bcf19dbf273942cf9708376e005ece07e76e56 Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Wed, 2 Sep 2026 17:53:22 +0200 Subject: [PATCH 2/6] Prevent Voice playback from becoming user input Suspend microphone capture during assistant speech and discard overlapping transcripts so speaker echo cannot answer interview questions. Co-authored-by: Cursor --- apps/petrinaut-website/README.md | 26 ++--- .../openai-realtime-session.test.ts | 95 +++++++++++++++---- .../openai-realtime-session.ts | 66 +++++++++---- .../voice-interview/realtime-brunch-bridge.ts | 2 +- .../voice-preview.integration.test.ts | 19 ++-- .../voice-turn-controller.test.ts | 16 +--- .../server/voice/openai-realtime-call.test.ts | 4 +- .../server/voice/openai-voice-policy.test.ts | 6 +- .../src/server/voice/openai-voice-policy.ts | 7 +- .../@hashintel/petrinaut/docs/ai-assistant.md | 8 +- 10 files changed, 167 insertions(+), 82 deletions(-) diff --git a/apps/petrinaut-website/README.md b/apps/petrinaut-website/README.md index c781a8d1636..2e3cbd952ff 100644 --- a/apps/petrinaut-website/README.md +++ b/apps/petrinaut-website/README.md @@ -110,14 +110,17 @@ 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 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, the session exposes no tools, and -the model never infers or paraphrases what the speaker said. The browser bridge -normalizes whitespace, submits each completed transcript exactly once (keyed by +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, @@ -135,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 63b686792ab..3ce2b50b1bf 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,27 +253,82 @@ 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.events).toEqual([ + { + connectionEpoch: 1, + responseId: "response-canonical", + type: "output-started", + }, + { + connectionEpoch: 1, + responseId: "response-canonical", + type: "output-stopped", + }, + ]); + }); + + 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(harness.localTracks[0]!.enabled).toBe(false); + + channel.receive({ + response_id: "response-canonical", + type: "output_audio_buffer.stopped", + }); 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", - }, - ]), - ); + channel.receive({ + response: { + id: "response-canonical", + output: [], + status: "completed", + }, + type: "response.done", + }); + + 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 () => { 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 5f3dbe4c160..5cd364fa6d2 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 @@ -258,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[] = []; @@ -279,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; @@ -449,16 +451,8 @@ 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 { @@ -790,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, @@ -1264,6 +1256,7 @@ export class OpenAIRealtimeSession { return; } this.#speakingResponseId = responseId; + this.#syncMicrophoneTrack(); this.#emit({ connectionEpoch, responseId, type: "output-started" }); return; } @@ -1287,6 +1280,26 @@ export class OpenAIRealtimeSession { const contentIndex = nonNegativeInteger(event.content_index); if (!itemId || contentIndex === null) return; const key = { connectionEpoch, contentIndex, itemId }; + const overlapsPlayback = + this.#speakingResponseId !== null || + this.#playbackOverlappingInputItemIds.has(itemId); + if (overlapsPlayback) { + this.#playbackOverlappingInputItemIds.add(itemId); + 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"); @@ -1341,6 +1354,7 @@ export class OpenAIRealtimeSession { this.#authorizedResponseIds.delete(responseId); if (this.#speakingResponseId === responseId) { this.#speakingResponseId = null; + this.#syncMicrophoneTrack(); } } @@ -1522,6 +1536,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); @@ -1622,6 +1652,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(); @@ -1633,6 +1664,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.ts b/apps/petrinaut-website/src/main/app/voice-interview/realtime-brunch-bridge.ts index bba0cd3199d..89af30aeb0e 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 @@ -532,7 +532,7 @@ export class RealtimeBrunchBridge { source, }); if (speechGeneration !== this.#speechGeneration) { - // Speech was cancelled (barge-in, pause, or replay) while preparing; + // Speech was cancelled (new input, pause, or replay) while preparing; // the caller owns the next utterance. return; } 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 4458902f796..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 completed user transcript 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,21 +276,21 @@ 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 @@ -321,6 +321,11 @@ describe("controlled voice preview", () => { await Promise.resolve(); expect(submitInterviewAnswer).not.toHaveBeenCalled(); + dataChannel.receive({ + audio_start_ms: 300, + item_id: "user-item", + type: "input_audio_buffer.speech_started", + }); dataChannel.receive({ content_index: 0, delta: "The supervisor", @@ -633,7 +638,7 @@ describe("controlled voice preview", () => { type: "semantic_vad", eagerness: "low", create_response: false, - interrupt_response: true, + interrupt_response: false, }, }, }, 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 349542a1755..26652d2b5a5 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 @@ -84,7 +84,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 +109,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 +123,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 () => { 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 db3a13e2512..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 () => @@ -140,7 +140,7 @@ describe("OpenAI Realtime call handler", () => { type: "semantic_vad", eagerness: "low", create_response: false, - interrupt_response: true, + 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 b8906acbd50..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,8 +46,8 @@ 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-v2"); + 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", @@ -88,7 +88,7 @@ When Petrinaut supplies response_text, speak only those strings, in array order type: "semantic_vad", eagerness: "low", create_response: false, - interrupt_response: true, + interrupt_response: false, }, }, output: { voice: "marin" }, 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 0d9d26e143a..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-v2"; +export const OPENAI_REALTIME_POLICY_VERSION = "brunch-control-plane-v3"; interface VoiceEnvironment { readonly NODE_ENV?: string; @@ -45,7 +45,8 @@ When Petrinaut supplies response_text, speak only those strings, in array order /** * 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, and no tool exists for the model to invent an answer through. + * 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. */ @@ -71,7 +72,7 @@ export const createOpenAIRealtimeSession = () => ({ type: "semantic_vad" as const, eagerness: "low" as const, create_response: false, - interrupt_response: true, + 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 From a63a7123cb32c30535eef5b03aef22e38ec38804 Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Wed, 2 Sep 2026 19:06:24 +0200 Subject: [PATCH 3/6] Format Realtime transcription failure handling Co-authored-by: Cursor --- .../src/main/app/voice-interview/openai-realtime-session.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 5cd364fa6d2..245e40a8388 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 @@ -1292,8 +1292,7 @@ export class OpenAIRealtimeSession { ) { this.#finishTranscription( itemId, - event.type === - "conversation.item.input_audio_transcription.failed" + event.type === "conversation.item.input_audio_transcription.failed" ? "invalid-response" : undefined, ); From 25f968aad6d230d3089ef75460ba19af9362546a Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Wed, 2 Sep 2026 21:52:24 +0200 Subject: [PATCH 4/6] Preserve Voice input started before playback Classify playback overlap when speech begins so late transcript events remain authoritative for user turns that predate assistant audio. Co-authored-by: Cursor --- .../openai-realtime-session.test.ts | 45 +++++++++++++++++++ .../openai-realtime-session.ts | 5 +-- 2 files changed, 46 insertions(+), 4 deletions(-) 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 3ce2b50b1bf..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 @@ -284,6 +284,51 @@ describe("OpenAIRealtimeSession", () => { ]); }); + 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({ + audio_start_ms: 120, + item_id: "item-user", + type: "input_audio_buffer.speech_started", + }); + channel.receive({ + response_id: "response-canonical", + type: "output_audio_buffer.started", + }); + channel.receive({ + content_index: 0, + item_id: "item-user", + transcript: "The supervisor approves it.", + type: "conversation.item.input_audio_transcription.completed", + }); + + expect(harness.events).toEqual([ + { + connectionEpoch: 1, + itemId: "item-user", + type: "input-speech-started", + }, + { + connectionEpoch: 1, + responseId: "response-canonical", + type: "output-started", + }, + { + key: { connectionEpoch: 1, contentIndex: 0, itemId: "item-user" }, + text: "The supervisor approves it.", + type: "completed", + }, + ]); + }); + test("restores only the requested microphone state after assistant playback", async () => { const harness = createHarness(); await harness.session.connect(); 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 245e40a8388..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 @@ -1280,11 +1280,8 @@ export class OpenAIRealtimeSession { const contentIndex = nonNegativeInteger(event.content_index); if (!itemId || contentIndex === null) return; const key = { connectionEpoch, contentIndex, itemId }; - const overlapsPlayback = - this.#speakingResponseId !== null || - this.#playbackOverlappingInputItemIds.has(itemId); + const overlapsPlayback = this.#playbackOverlappingInputItemIds.has(itemId); if (overlapsPlayback) { - this.#playbackOverlappingInputItemIds.add(itemId); if ( event.type === "conversation.item.input_audio_transcription.completed" || From 5ed5a6174a73ad6e3b29027ef704b897ef2e42fb Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Wed, 2 Sep 2026 22:11:46 +0200 Subject: [PATCH 5/6] Recover Voice speech after rejected input Restore canceled pending questions after unusable transcripts, release replay guards on rejection, and classify in-flight silence as unavailable. Co-authored-by: Cursor --- .../realtime-brunch-bridge.test.ts | 50 +++++++++++++++++++ .../voice-interview/realtime-brunch-bridge.ts | 47 +++++++++++++---- .../voice-turn-controller.test.ts | 10 +++- .../voice-interview/voice-turn-controller.ts | 12 ++++- 4 files changed, 106 insertions(+), 13 deletions(-) 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 e27bed528c6..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 @@ -392,6 +392,47 @@ describe("RealtimeBrunchBridge", () => { ); }); + 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?"); @@ -865,6 +906,15 @@ describe("RealtimeBrunchBridge", () => { await vi.waitFor(() => expect(harness.submitInterviewAnswer).toHaveBeenCalledOnce(), ); + 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(); 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 89af30aeb0e..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 @@ -212,7 +212,9 @@ export class RealtimeBrunchBridge { canonicalSegments: [], status: "ready", }; + #cancelledSpeechSource: InterviewSpeechSource | null = null; #generation = 0; + #pendingSpeechSource: InterviewSpeechSource | null = null; #speechGeneration = 0; public constructor({ @@ -237,6 +239,8 @@ export class RealtimeBrunchBridge { } this.#activeEpoch = connectionEpoch; this.#activeSubmission = null; + this.#cancelledSpeechSource = null; + this.#pendingSpeechSource = null; this.#seenSegmentIds.clear(); for (const segment of this.#chat.canonicalSegments) { this.#seenSegmentIds.add(segment.id); @@ -261,11 +265,26 @@ export class RealtimeBrunchBridge { ++this.#generation; this.#activeEpoch = null; this.#activeSubmission = null; + this.#cancelledSpeechSource = null; + this.#pendingSpeechSource = null; this.#preparedContextCache.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 { @@ -324,6 +343,8 @@ export class RealtimeBrunchBridge { ): void { ++this.#generation; this.#activeSubmission = null; + this.#cancelledSpeechSource = null; + this.#pendingSpeechSource = null; this.#emit({ code, message, type: "error" }); } @@ -343,6 +364,15 @@ export class RealtimeBrunchBridge { return; } 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; + } if (event.type === "transcription-failed") { this.#emit({ reason: "failed", type: "transcript-rejected" }); return; @@ -357,17 +387,9 @@ export class RealtimeBrunchBridge { this.#emit({ reason: "too-long", type: "transcript-rejected" }); return; } - 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 generation = this.#generation; + this.#cancelledSpeechSource = null; this.#activeSubmission = { baselineSegmentIds: new Set( this.#chat.canonicalSegments.map(({ id }) => id), @@ -474,9 +496,11 @@ export class RealtimeBrunchBridge { } #prepareAndDeliver(source: InterviewSpeechSource): void { - this.#emit({ type: "speech-delivery-pending" }); 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; @@ -500,6 +524,7 @@ export class RealtimeBrunchBridge { source: questionOnlySource, }).text, ); + this.#pendingSpeechSource = null; return; } @@ -521,6 +546,7 @@ export class RealtimeBrunchBridge { source, }).text, ); + this.#pendingSpeechSource = null; return; } void this.#session.prepareInterviewSpeech(request).then((preparation) => { @@ -536,6 +562,7 @@ export class RealtimeBrunchBridge { // the caller owns the next utterance. return; } + this.#pendingSpeechSource = null; if (preparation.kind === "prepared") { this.#preparedContextCache.set(request.cacheKey, preparation.context); } 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 26652d2b5a5..2cd99b86471 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) => { @@ -707,9 +708,11 @@ describe("VoiceTurnController", () => { "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: [question("ask-1")], + canonicalSegments: [...source.fullResponseSegments], status: "ready", }); await harness.controller.start(); @@ -727,8 +730,10 @@ describe("VoiceTurnController", () => { harness.emitBridge({ reason, type: "transcript-rejected" }); expect(harness.controller.getSnapshot()).toMatchObject({ + canReadFullResponse: true, + canRepeatQuestion: true, connection: "connected", - currentQuestion: "What happens after approval?", + currentQuestion: "Who approves release?", input: "listening", inputNotice: "not-heard", lastAnswerDelivery: "none", @@ -736,6 +741,7 @@ describe("VoiceTurnController", () => { output: "idle", partialText: "", }); + expect(harness.bridge.restoreCancelledSpeech).toHaveBeenCalledOnce(); expect(harness.submitText).not.toHaveBeenCalled(); harness.emitSession({ 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 71ae6e98849..1d1d0a937b3 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 @@ -72,6 +72,7 @@ interface RealtimeSession { interface RealtimeBridge { cancelPendingSpeech(): void; + restoreCancelledSpeech(): void; start(connectionEpoch: number): void; stop(): void; subscribe(listener: (event: RealtimeBrunchBridgeEvent) => void): () => void; @@ -468,15 +469,24 @@ export class VoiceTurnController { // or a transcription failure is surfaced as a recoverable notice; other // rejections (duplicate redelivery, overlap with an in-flight answer) // are diagnostics only. + 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({}); } + 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") { From 935aa9f02a5ac635a50eb8bc130edb3e258af8e4 Mon Sep 17 00:00:00 2001 From: Kostandin Angjellari Date: Wed, 2 Sep 2026 22:26:11 +0200 Subject: [PATCH 6/6] Keep diagnostic Voice rejections passive Leave active turns untouched for duplicate and unavailable transcripts, and defer cancelled speech restoration while Voice mode is paused. Co-authored-by: Cursor --- .../voice-turn-controller.test.ts | 55 +++++++++++++++++++ .../voice-interview/voice-turn-controller.ts | 15 +++-- 2 files changed, 66 insertions(+), 4 deletions(-) diff --git a/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts b/apps/petrinaut-website/src/main/app/voice-interview/voice-turn-controller.test.ts index 2cd99b86471..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 @@ -767,8 +767,63 @@ describe("VoiceTurnController", () => { 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(); 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 1d1d0a937b3..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 @@ -465,10 +465,15 @@ 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; other - // rejections (duplicate redelivery, overlap with an in-flight answer) - // are diagnostics only. + // or a transcription failure is surfaced as a recoverable notice. this.#inputTurnPending = false; this.#transcriptItemId = null; this.#transcriptKey = null; @@ -477,7 +482,9 @@ export class VoiceTurnController { } else { this.#update({}); } - this.#bridge.restoreCancelledSpeech(); + if (this.#snapshot.input !== "paused") { + this.#bridge.restoreCancelledSpeech(); + } return; } if (event.type === "submission-accepted") {