diff --git a/examples/live/README.md b/examples/live/README.md index 8527616c31..e3a05216f2 100644 --- a/examples/live/README.md +++ b/examples/live/README.md @@ -56,7 +56,12 @@ pnpm tsn -T examples/live/transcript-grouper.ts ``` Audio playback requires `ffplay` on your `PATH`. To run without playback, add -`--no-audio`. +`--no-audio`. The example uses a 45-second demo budget after preparing the TTS +clips, including session startup. In a terminal, press Enter to finish sooner, +or Ctrl-C to cancel and stop playback. Both normal finish paths request closure +and wait for the server's confirmation. +The budget is an observation window, not a claim that the model finished its +answer. No path infers answer completion from a pause or transcript fragment. ## WebRTC with a backend sideband diff --git a/examples/live/transcript-demo.ts b/examples/live/transcript-demo.ts index c3471521e7..de1a7219dc 100644 --- a/examples/live/transcript-demo.ts +++ b/examples/live/transcript-demo.ts @@ -49,11 +49,19 @@ export async function prepareTranscriptDemo(client: OpenAI): Promise { * Send pre-generated PCM through the public Live WebSocket and group only its * public transcript events. This makes paid API requests; requires Live access. * No microphone, local TTS latency, internal API, or server turn events are used. + * The caller ends capture with finishSignal; transcript activity never ends it. + * This controls capture duration, not whether the model finished an answer. + * The result is returned only after the server confirms session.closed. + * Finishing before session startup rejects without returning an unconfirmed capture. */ export async function runTranscriptDemo( client: OpenAI, audio: readonly Buffer[], options: { + /** End capture when the caller or user decides. This is separate from operation failures. */ + finishSignal: AbortSignal; + /** Cancel capture and pending socket waits without returning a transcript. */ + signal?: AbortSignal; /** Optional transport options, for example a local opt-in compatibility header. */ websocketOptions?: LiveWSClientOptions | undefined; /** Interrupt the first answer instead of waiting for its projected completion. */ @@ -66,15 +74,20 @@ export async function runTranscriptDemo( onInputAudio?: (pcm: Buffer) => void; /** Observe incoming mono 24 kHz PCM independently of transcript grouping. */ onOutputAudio?: (pcm: Buffer) => void; - } = {}, + }, ): Promise { const [firstAudio, secondAudio] = audio; if (!firstAudio?.length || !secondAudio?.length) { throw new Error('Generate both TTS clips before opening Live.'); } + if (options.finishSignal.aborted) { + throw new Error('Live transcript capture stopped before session startup'); + } + options.signal?.throwIfAborted(); const live = new LiveWS(client, options.websocketOptions); const abort = new AbortController(); + const signal = options.signal ? AbortSignal.any([abort.signal, options.signal]) : abort.signal; const timeout = setTimeout( () => abort.abort(new Error('Live transcript demo exceeded 120 seconds')), 120_000, @@ -89,6 +102,7 @@ export async function runTranscriptDemo( }; let ready = false; let finishing = false; + let sessionClosed = false; let lastAudioAt = Number.NEGATIVE_INFINITY; let lastTextAt = Number.NEGATIVE_INFINITY; let assistantCharacters = 0; @@ -117,12 +131,13 @@ export async function runTranscriptDemo( live.on('error', (error) => abort.abort(error)); live.on('close', () => { - if (!finishing) { + if (!finishing || !sessionClosed) { abort.abort(new Error('Live transport closed before the demo completed')); } }); const onEvent = (event: ServerEvent) => { - if (event.type === 'session.started') { + // A late start must not turn an earlier finish into a successful capture. + if (event.type === 'session.started' && !options.finishSignal.aborted) { ready = true; } if (event.type === 'session.input_transcript.delta') { @@ -143,6 +158,12 @@ export async function runTranscriptDemo( } transcript.push(event); if (event.type === 'session.closed') { + if (!finishing) { + abort.abort(new Error('Live session ended before the demo completed')); + } else if (event.reason !== 'close_requested') { + abort.abort(new Error('Live session ended without confirming the requested close')); + } + sessionClosed = true; live.off('event', onEvent); } } catch (error) { @@ -169,7 +190,10 @@ export async function runTranscriptDemo( async function sendPCM(pcm: Buffer): Promise { const startedAt = performance.now(); for (let offset = 0; offset < pcm.length; offset += CHUNK_BYTES) { - abort.signal.throwIfAborted(); + signal.throwIfAborted(); + if (options.finishSignal.aborted) { + return; + } const chunk = pcm.subarray(offset, offset + CHUNK_BYTES); live.send({ type: 'session.input_audio.append', @@ -177,7 +201,7 @@ export async function runTranscriptDemo( }); options.onInputAudio?.(chunk); const sentMs = (Math.min(offset + CHUNK_BYTES, pcm.length) / (SAMPLE_RATE * 2)) * 1000; - await sleep(Math.max(0, startedAt + sentMs - performance.now()), undefined, { signal: abort.signal }); + await sleep(Math.max(0, startedAt + sentMs - performance.now()), undefined, { signal }); } } @@ -185,7 +209,8 @@ export async function runTranscriptDemo( const silence = Buffer.alloc(CHUNK_BYTES * 5); async function waitWithAudio(predicate: () => boolean, description: string): Promise { const deadline = performance.now() + 45_000; - while (!predicate()) { + while (!predicate() && !options.finishSignal.aborted) { + signal.throwIfAborted(); if (performance.now() >= deadline) { throw new Error( `Timed out waiting for ${description} (${assistantCharacters} assistant characters, ${closedAssistant} closed assistant segments, ${Math.round(performance.now() - lastTextAt)} ms since the last output transcript)`, @@ -199,8 +224,12 @@ export async function runTranscriptDemo( // The generated transport queues this until the socket opens. live.send({ type: 'session.start', session }); // oxlint-disable-next-line no-unmodified-loop-condition -- The session.started event listener sets ready. - while (!ready) { - await sleep(20, undefined, { signal: abort.signal }); + while (!ready && !options.finishSignal.aborted) { + await sleep(20, undefined, { signal }); + } + signal.throwIfAborted(); + if (!ready) { + throw new Error('Live transcript capture stopped before session startup'); } await sendPCM(firstAudio); if (options.interrupt) { @@ -211,27 +240,27 @@ export async function runTranscriptDemo( performance.now() - lastTextAt < 1000, 'the assistant to speak before interrupting', ); - result.interruptionSentDuringOutput = true; + result.interruptionSentDuringOutput = !options.finishSignal.aborted; } else { await waitWithAudio( - () => closedAssistant > 0 && performance.now() - lastTextAt >= 2500, + () => closedAssistant > 0 && performance.now() - Math.max(lastTextAt, lastAudioAt) >= 2500, 'the first answer to settle', ); } await sendPCM(secondAudio); - const beforeAnswer = assistantCharacters; - const beforeClose = closedAssistant; - await waitWithAudio( - () => - assistantCharacters > beforeAnswer && - closedAssistant > beforeClose && - performance.now() - lastTextAt >= 2500, - 'the final answer to settle', - ); + // Live has no public turn-final marker. Keep the duplex input running until + // the caller explicitly ends capture instead of guessing from transcript gaps. + while (!options.finishSignal.aborted) { + await sendPCM(silence); + } + signal.throwIfAborted(); finishing = true; live.send({ type: 'session.close' }); - // Give the public session.closed event a chance to finalize remaining text. - await sleep(300, undefined, { signal: abort.signal }); + // Confirm finalization before closing the transport or returning the transcript. + // oxlint-disable-next-line no-unmodified-loop-condition -- The session.closed event listener sets sessionClosed. + while (!sessionClosed) { + await sleep(20, undefined, { signal }); + } } finally { finishing = true; clearTimeout(timeout); @@ -242,6 +271,6 @@ export async function runTranscriptDemo( live.close(); } } - abort.signal.throwIfAborted(); + signal.throwIfAborted(); return result; } diff --git a/examples/live/transcript-grouper.ts b/examples/live/transcript-grouper.ts index 6e8a956efc..a261e059cf 100644 --- a/examples/live/transcript-grouper.ts +++ b/examples/live/transcript-grouper.ts @@ -1,3 +1,5 @@ +import { createInterface } from 'node:readline'; +import type { Interface } from 'node:readline'; import OpenAI from 'openai'; import { prepareTranscriptDemo, runTranscriptDemo } from './transcript-demo'; import { createPCMPlayback } from './transcript-playback'; @@ -6,28 +8,58 @@ async function main(): Promise { const playback = process.argv.includes('--no-audio') ? undefined : { user: createPCMPlayback(), assistant: createPCMPlayback() }; + const finish = new AbortController(); + const cancel = new AbortController(); + let input: Interface | undefined; + const stop = () => finish.abort(); + const interrupt = () => { + cancel.abort(); + void playback?.user.close(); + void playback?.assistant.close(); + }; try { const client = new OpenAI({ logLevel: 'off' }); console.log('All voices are AI-generated. Preparing both TTS clips before connecting to Live…'); const audio = await prepareTranscriptDemo(client); - console.log(playback ? 'Playing both voices through playAudio (ffplay).' : 'Audio playback disabled.'); + console.log(playback ? 'Playing both voices through ffplay.' : 'Audio playback disabled.'); console.log('TTS ready. Starting a picnic conversation, then interrupting with a breakfast request.'); + process.once('SIGINT', interrupt); + console.log('Capturing for up to 45 seconds, then closing the session.'); + if (process.stdin.isTTY) { + input = createInterface({ input: process.stdin, output: process.stdout }); + input.once('line', stop); + input.once('close', stop); + input.once('SIGINT', interrupt); + console.log('Press Enter to finish capturing the transcript and close the session sooner.'); + } const result = await runTranscriptDemo(client, audio, { + finishSignal: AbortSignal.any([finish.signal, AbortSignal.timeout(45_000)]), + signal: cancel.signal, interrupt: true, onInputAudio: (pcm) => playback?.user.write(pcm), onOutputAudio: (pcm) => playback?.assistant.write(pcm), onClosed: ({ segment, reason }) => console.log(`[${segment.speaker}; ${reason}] ${segment.text}`), }); await Promise.all([playback?.user.finish(), playback?.assistant.finish()]); + cancel.signal.throwIfAborted(); console.log( `${result.rawTranscriptEvents} raw transcript fragments → ${result.updateCount} bubble updates → ${result.segments.length} final bubbles.`, ); } catch (error) { - const message = error instanceof Error ? error.message : 'The Live example failed'; - const key = process.env['OPENAI_API_KEY']; - console.error(key ? message.split(key).join('[REDACTED]') : message); - process.exitCode = 1; + if (cancel.signal.aborted) { + process.exitCode = 130; + } else { + const message = error instanceof Error ? error.message : 'The Live example failed'; + const key = process.env['OPENAI_API_KEY']; + console.error(key ? message.split(key).join('[REDACTED]') : message); + process.exitCode = 1; + } } finally { + process.removeListener('SIGINT', interrupt); + input?.removeListener('line', stop); + input?.removeListener('close', stop); + input?.removeListener('SIGINT', interrupt); + input?.close(); await Promise.all([playback?.user.close(), playback?.assistant.close()]); } } diff --git a/examples/live/transcript-playback.ts b/examples/live/transcript-playback.ts index 9aefc13972..bc4e0575c2 100644 --- a/examples/live/transcript-playback.ts +++ b/examples/live/transcript-playback.ts @@ -1,7 +1,6 @@ -import { PassThrough } from 'node:stream'; -import { playAudio } from 'openai/helpers/audio'; +import { spawn } from 'node:child_process'; -/** Play the demo's mono, 24 kHz, 16-bit PCM using the SDK's existing ffplay helper. */ +/** Play the demo's mono, 24 kHz, 16-bit PCM and retain the child for cancellation. */ export function createPCMPlayback() { // A streaming WAV header lets playAudio recognize raw Live/TTS PCM. The // unknown lengths are resolved by EOF, so playback can start before completion. @@ -19,18 +18,30 @@ export function createPCMPlayback() { header.write('data', 36); header.writeUInt32LE(0xff_ff_ff_ff, 40); - const stream = new PassThrough(); - stream.write(header); + // We own the child so close() can stop queued playback even after its input has ended. + const player = spawn('ffplay', ['-autoexit', '-nodisp', '-i', 'pipe:0']); + player.stdout.resume(); + player.stderr.resume(); + const stream = player.stdin; let failure: Error | undefined; - // Attach rejection handling immediately, including when ffplay is missing. - const finished = (async () => { - try { - await playAudio(stream); - } catch (error) { - failure = error instanceof Error ? error : new Error('Audio playback failed'); - stream.destroy(); - } - })(); + // Record failures immediately and settle only when the child has exited. + // oxlint-disable-next-line promise/avoid-new -- Await child close even after spawn or stdin errors, so cleanup never leaves a player running. + const finished = new Promise((resolve) => { + player.once('error', (error) => { + failure ??= error; + }); + stream.once('error', (error) => { + failure ??= error; + player.kill(); + }); + player.once('close', (code) => { + if (code !== 0) { + failure ??= new Error(`ffplay process exited with code ${code}`); + } + resolve(); + }); + }); + stream.write(header); return { /** Queue a PCM chunk without blocking the duplex transport. */ @@ -54,6 +65,7 @@ export function createPCMPlayback() { /** Stop playback on teardown, including after a failed Live session. */ async close(): Promise { stream.destroy(); + player.kill(); await finished; }, }; diff --git a/tests/lib/live/transcript-demo-cli.test.ts b/tests/lib/live/transcript-demo-cli.test.ts new file mode 100644 index 0000000000..0a2dbe44d1 --- /dev/null +++ b/tests/lib/live/transcript-demo-cli.test.ts @@ -0,0 +1,205 @@ +import { spawn } from 'node:child_process'; +import { once } from 'node:events'; +import { createServer } from 'node:http'; +import path from 'node:path'; +import { describe, expect, it } from 'vitest'; +import { WebSocketServer } from 'ws'; +import { compiledFixture, compiledFixtureConfig } from '../../utils/compiled-fixtures'; + +const quote = (part: string) => `'${part.split("'").join("'\\''")}'`; + +async function runCLI(action: 'deadline' | 'signal' | 'terminal' | 'terminal-deadline' | 'terminal-enter') { + const server = createServer((request, response) => { + if (request.method !== 'POST' || request.url !== '/v1/audio/speech') { + response.writeHead(404).end(); + return; + } + response.writeHead(200, { 'content-type': 'application/octet-stream' }); + response.end(Buffer.alloc(1920)); + }); + const webSocket = new WebSocketServer({ server }); + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + const address = server.address(); + if (!address || typeof address === 'string') { + throw new Error('Expected local TCP address'); + } + const arguments_ = [ + '-r', + path.join(process.cwd(), 'node_modules/tsconfig-paths/register.js'), + compiledFixture('examples/live/transcript-grouper.ts'), + '--no-audio', + ]; + const terminal = action.startsWith('terminal'); + const deadline = action === 'deadline' || action === 'terminal-deadline'; + // script(1) supplies real terminal mode, so Node's readline processes Ctrl-C. + const child = spawn( + terminal ? '/usr/bin/script' : process.execPath, + terminal + ? ['-q', '-e', '-c', `exec ${[process.execPath, ...arguments_].map(quote).join(' ')}`, '/dev/null'] + : arguments_, + { + cwd: process.cwd(), + env: { + OPENAI_API_KEY: 'synthetic-live-cli-key', + OPENAI_BASE_URL: `http://127.0.0.1:${address.port}/v1`, + TS_NODE_PROJECT: compiledFixtureConfig(), + DISABLE_V8_COMPILE_CACHE: '1', + NODE_COMPILE_CACHE: process.env['NODE_COMPILE_CACHE'], + SystemRoot: process.env['SystemRoot'], + SHELL: '/bin/sh', + }, + stdio: [terminal ? 'pipe' : 'ignore', 'pipe', 'pipe'], + timeout: deadline ? 65_000 : 12_000, + }, + ); + if (!child.stdout || !child.stderr) { + throw new Error('Expected CLI stdout and stderr pipes'); + } + let stdout = ''; + let stderr = ''; + child.stdout.setEncoding('utf-8').on('data', (data: string) => (stdout += data)); + child.stderr.setEncoding('utf-8').on('data', (data: string) => (stderr += data)); + let receivedStart = false; + let receivedAudio = false; + let receivedClose = false; + let confirmedClose = false; + let startedAt = 0; + let closedAt = 0; + webSocket.on('connection', (socket) => { + socket.on('message', (data) => { + // SAFETY: this local peer reads serialized messages from the actual SDK. + const event = JSON.parse(data.toString()) as { type: string }; + if (event.type === 'session.start') { + receivedStart = true; + startedAt = performance.now(); + socket.send( + JSON.stringify({ type: 'session.started', session: { id: 'fixture', status: 'active' } }), + ); + } else if (event.type === 'session.input_audio.append' && !receivedAudio) { + receivedAudio = true; + if (deadline || action === 'terminal-enter') { + socket.send( + JSON.stringify({ + type: 'session.input_transcript.delta', + event_id: 'synthetic-input', + delta: 'A vegetarian picnic.', + start_ms: 0, + end_ms: 40, + }), + ); + } + if (action === 'signal') { + child.kill('SIGINT'); + } + if (action === 'terminal') { + child.stdin?.write('\u0003'); + } + if (action === 'terminal-enter') { + child.stdin?.write('\n'); + } + } else if (event.type === 'session.close') { + receivedClose = true; + closedAt = performance.now(); + socket.send(JSON.stringify({ type: 'session.closed', reason: 'close_requested' })); + confirmedClose = true; + } + }); + }); + try { + const [exitCode, signal] = await once(child, 'close'); + expect(stdout + stderr).not.toContain('synthetic-live-cli-key'); + return { + exitCode, + signal, + stdout, + receivedStart, + receivedAudio, + receivedClose, + confirmedClose, + captureMs: closedAt - startedAt, + }; + } finally { + child.kill(); + for (const socket of webSocket.clients) { + socket.terminate(); + } + const wsClosed = once(webSocket, 'close'); + webSocket.close(); + await wsClosed; + const httpClosed = once(server, 'close'); + server.close(); + await httpClosed; + } +} + +describe('Live transcript CLI capture and terminal input', () => { + it.concurrent('captures to its default deadline with stdin redirected from /dev/null', async () => { + const result = await runCLI('deadline'); + expect(result).toMatchObject({ + exitCode: 0, + signal: null, + receivedStart: true, + receivedAudio: true, + receivedClose: true, + confirmedClose: true, + }); + expect(result.captureMs).toBeGreaterThan(44_000); + expect(result.stdout).toContain('A vegetarian picnic.'); + expect(result.stdout).toContain('final bubbles.'); + }, 70_000); + + it.skipIf(process.platform !== 'linux').concurrent( + 'captures to its default deadline in an unattended terminal', + async () => { + const result = await runCLI('terminal-deadline'); + expect(result).toMatchObject({ + exitCode: 0, + signal: null, + receivedStart: true, + receivedAudio: true, + receivedClose: true, + confirmedClose: true, + }); + expect(result.captureMs).toBeGreaterThan(44_000); + expect(result.stdout).toContain('A vegetarian picnic.'); + expect(result.stdout).toContain('final bubbles.'); + }, + 70_000, + ); + + it.skipIf(process.platform !== 'linux').concurrent( + 'allows Enter to finish before the terminal deadline', + async () => { + const result = await runCLI('terminal-enter'); + expect(result).toMatchObject({ + exitCode: 0, + signal: null, + receivedAudio: true, + receivedClose: true, + confirmedClose: true, + }); + expect(result.captureMs).toBeLessThan(10_000); + expect(result.stdout).toContain('final bubbles.'); + }, + ); + + // Windows subprocess.kill('SIGINT') terminates unconditionally, unlike a real console Ctrl-C. + it.skipIf(process.platform === 'win32').concurrent( + 'cancels with exit 130 on process SIGINT during capture', + async () => { + const result = await runCLI('signal'); + expect(result).toMatchObject({ exitCode: 130, signal: null, receivedAudio: true }); + expect(result.stdout).not.toContain('final bubbles.'); + }, + ); + + it.skipIf(process.platform !== 'linux').concurrent( + 'cancels readline Ctrl-C in an actual terminal', + async () => { + const result = await runCLI('terminal'); + expect(result).toMatchObject({ exitCode: 130, signal: null, receivedAudio: true }); + expect(result.stdout).not.toContain('final bubbles.'); + }, + ); +}); diff --git a/tests/lib/live/transcript-demo.test.ts b/tests/lib/live/transcript-demo.test.ts new file mode 100644 index 0000000000..87939144a6 --- /dev/null +++ b/tests/lib/live/transcript-demo.test.ts @@ -0,0 +1,521 @@ +import { once } from 'node:events'; +import { expect, it } from 'vitest'; +import { WebSocketServer } from 'ws'; + +import OpenAI from 'openai'; +import { runTranscriptDemo } from '../../../examples/live/transcript-demo'; + +const firstUser = 'Plan our picnic.'; +const secondUser = 'Make it breakfast.'; +const firstAnswer = 'There are sandwiches, fruit and a vegetarian salad.'; +const finalAnswer = 'Bring breakfast rolls and coffee.'; +const acknowledgment = 'I understand the change and will think about a breakfast picnic.'; + +it.concurrent.each([ + { timing: 'early', interrupt: false }, + { timing: 'early', interrupt: true }, + { timing: 'delayed', interrupt: false }, + { timing: 'delayed', interrupt: true }, + { timing: 'overlap', interrupt: false }, + { timing: 'overlap', interrupt: true }, + { timing: 'stale', interrupt: true }, + { timing: 'no-answer', interrupt: true }, + { timing: 'straddling-stale-only', interrupt: true }, + { timing: 'reused-user', interrupt: true }, + { timing: 'delayed-first-input', interrupt: true }, + { timing: 'long-acknowledgment', interrupt: true }, + { timing: 'audio-tail', interrupt: false }, + { timing: 'delayed-finalization', interrupt: false }, + { timing: 'eof-finalization', interrupt: false }, + { timing: 'premature-finalization', interrupt: false }, + { timing: 'premature-finalization-open', interrupt: false }, + { timing: 'error', interrupt: false }, + { timing: 'eof', interrupt: true }, +])('handles a $timing answer (interrupt=$interrupt)', async ({ timing, interrupt }) => { + const expectsAnswer = ![ + 'no-answer', + 'straddling-stale-only', + 'error', + 'eof', + 'eof-finalization', + 'premature-finalization', + 'premature-finalization-open', + ].includes(timing); + const receivesAnswer = expectsAnswer || timing === 'eof-finalization'; + const server = new WebSocketServer({ port: 0, host: '127.0.0.1' }); + await once(server, 'listening'); + const address = server.address(); + if (!address || typeof address === 'string') { + throw new Error('Expected an ephemeral TCP address'); + } + const baseURL = `http://127.0.0.1:${address.port}/v1`; + const client = new OpenAI({ apiKey: 'local-test-key', baseURL }); + const timers: ReturnType[] = []; + let secondChunks = 0; + let answerAtChunk = 0; + let sentSessionClose = false; + let sentSessionClosed = false; + let forcedPrematureClose = false; + let audioFrames = 0; + let answerSent = false; + const finish = new AbortController(); + const finishAfterExpectedInputAndAnswer = (): void => { + if (secondChunks === 40 && answerSent && timing !== 'audio-tail') { + finish.abort(); + } + }; + + server.on('connection', (socket) => { + const startedAt = performance.now(); + const now = () => Math.round(performance.now() - startedAt); + const sendText = (speaker: 'input' | 'output', text: string, start: number, end: number) => + socket.send( + JSON.stringify({ + type: `session.${speaker}_transcript.delta`, + event_id: `${speaker}-${now()}`, + delta: text, + start_ms: start, + end_ms: end, + }), + ); + const sendAudio = () => { + audioFrames += 1; + socket.send( + JSON.stringify({ + type: 'session.output_audio.delta', + event_id: `audio-${now()}`, + delta: Buffer.alloc(1920).toString('base64'), + }), + ); + }; + const startSession = () => { + if (timing === 'premature-finalization' || timing === 'premature-finalization-open') { + socket.send( + JSON.stringify({ type: 'session.closed', reason: 'close_requested', session: { id: 'fixture' } }), + ); + if (timing === 'premature-finalization-open') { + timers.push( + setTimeout(() => { + forcedPrematureClose = true; + socket.close(); + }, 3000), + ); + } else { + socket.close(); + } + return; + } + socket.send(JSON.stringify({ type: 'session.started', session: { id: 'fixture', status: 'active' } })); + }; + let heardFirst = false; + let firstEnd = 0; + let firstMark = 0; + const answerDelay = + new Map([ + ['delayed', 2050], + ['overlap', 2050], + ['delayed-first-input', 4300], + ['long-acknowledgment', 4300], + ]).get(timing) ?? 720; + const firstInputFragmented = ['reused-user', 'delayed-first-input'].includes(timing); + socket.on('message', (raw) => { + const event: { type: string; audio: string } = JSON.parse(raw.toString()); + if (event.type === 'session.start') { + startSession(); + } + if (event.type === 'session.input_audio.append') { + const [marker] = Buffer.from(event.audio, 'base64'); + if (marker === 1 && !heardFirst) { + heardFirst = true; + const mark = now(); + firstMark = mark; + sendText('input', firstInputFragmented ? 'Plan our ' : firstUser, Math.max(0, mark - 20), mark); + timers.push( + setTimeout(() => { + firstEnd = now(); + sendText('output', firstAnswer, firstEnd - 50, firstEnd); + sendAudio(); + if (timing === 'reused-user') { + // A delayed user fragment leaves the same user bubble active + // across the start of the second clip. + sendText('input', 'picnic.', firstEnd + 1, firstEnd + 2); + } + }, 720), + ); + } + if (marker === 2) { + secondChunks += 1; + if (secondChunks === 1) { + const mark = now(); + if (timing === 'delayed-first-input') { + sendText('input', 'picnic.', firstMark, firstMark + 10); + sendText('output', ' Plus lemonade.', firstEnd + 1, firstEnd + 40); + sendAudio(); + } else if (timing !== 'overlap') { + sendText('input', secondUser, Math.max(0, mark - 20), mark); + } + if (timing === 'stale' || timing === 'no-answer' || timing === 'straddling-stale-only') { + timers.push( + setTimeout(() => { + // This older fragment is delivered after the new user input. + sendText( + 'output', + ' Plus lemonade.', + firstEnd - 50, + timing === 'straddling-stale-only' ? mark + 500 : firstEnd, + ); + sendAudio(); + }, 140), + ); + } + if (timing === 'long-acknowledgment') { + timers.push( + setTimeout(() => { + const ackEnd = now(); + sendText('output', acknowledgment, ackEnd - 200, ackEnd); + sendAudio(); + }, 400), + ); + } + if (receivesAnswer) { + timers.push( + setTimeout(() => { + answerAtChunk = secondChunks; + const answerEnd = now(); + if (timing === 'overlap') { + // The answer occurred during the user's long second interval, + // even though the peer delivers both transcripts afterwards. + sendText('input', secondUser, Math.max(0, mark - 20), answerEnd); + sendText('output', finalAnswer, mark + 650, mark + 1000); + } else { + if (timing === 'delayed-first-input') { + sendText('input', secondUser, mark, mark + 1600); + } + sendText('output', finalAnswer, answerEnd - 50, answerEnd); + } + sendAudio(); + answerSent = true; + finishAfterExpectedInputAndAnswer(); + }, answerDelay), + ); + } + if (timing === 'audio-tail') { + for (const delay of [1800, 2400, 3200, 4000, 4800]) { + timers.push(setTimeout(sendAudio, delay)); + } + timers.push(setTimeout(() => finish.abort(), 5200)); + } + // Allow a real settled answer, then close the peer to bound a failed + // demo without waiting for its 45-second production timeout. + timers.push( + setTimeout( + () => { + if (timing === 'error') { + socket.send( + JSON.stringify({ + type: 'error', + error: { type: 'invalid_request_error', code: 'fixture', message: 'Synthetic error' }, + }), + ); + } else { + socket.close(); + } + }, + timing === 'audio-tail' ? 9000 : answerDelay + 3700, + ), + ); + } + finishAfterExpectedInputAndAnswer(); + } + } + if (event.type === 'session.close') { + sentSessionClose = true; + const finalize = () => { + if (timing !== 'eof-finalization') { + sentSessionClosed = true; + socket.send( + JSON.stringify({ + type: 'session.closed', + reason: 'close_requested', + session: { id: 'fixture' }, + }), + ); + } + socket.close(); + }; + if (timing === 'delayed-finalization') { + timers.push(setTimeout(finalize, 650)); + } else { + finalize(); + } + } + }); + }); + + try { + const result = runTranscriptDemo(client, [Buffer.alloc(1920, 1), Buffer.alloc(1920 * 40, 2)], { + finishSignal: finish.signal, + interrupt, + }); + if (expectsAnswer) { + const demo = await result; + expect(sentSessionClose).toBe(true); + expect(sentSessionClosed).toBe(true); + expect(secondChunks).toBe(40); + if (['delayed', 'overlap', 'delayed-first-input', 'long-acknowledgment'].includes(timing)) { + expect(answerAtChunk).toBe(secondChunks); + } else { + expect(answerAtChunk).toBeLessThan(secondChunks); + } + expect(demo.rawUserText).toBe(firstUser + secondUser); + expect(demo.interruptionSentDuringOutput).toBe(interrupt); + if (timing === 'delayed-first-input' || timing === 'long-acknowledgment') { + expect(answerSent).toBe(true); + const assistant = demo.segments + .filter(({ segment }) => segment.speaker === 'assistant') + .map(({ segment }) => segment.text) + .join(''); + expect(assistant).toContain(finalAnswer); + return; + } + if (timing === 'audio-tail') { + expect(audioFrames).toBe(7); + } + expect(demo.segments.map(({ segment }) => [segment.speaker, segment.text])).toEqual([ + ['user', timing === 'reused-user' ? 'Plan our ' : firstUser], + ['assistant', firstAnswer], + ['user', (timing === 'reused-user' ? 'picnic.' : '') + secondUser], + ['assistant', (timing === 'stale' ? ' Plus lemonade.' : '') + finalAnswer], + ]); + for (const [index, { segment }] of demo.segments.entries()) { + expect(segment.previousId).toBe(demo.segments[index - 1]?.segment.id ?? null); + } + } else { + await expect(result).rejects.toBeInstanceOf(Error); + expect(sentSessionClose).toBe(timing === 'eof-finalization'); + expect(forcedPrematureClose).toBe(false); + } + } finally { + for (const timer of timers) { + clearTimeout(timer); + } + const closed = once(server, 'close'); + for (const peer of server.clients) { + peer.terminate(); + } + server.close(); + await closed; + } +}); + +it.concurrent.each(['close_requested', 'expired', 'content', 'remote_hangup', 'connection_lost'])( + 'finishes during the first input and checks the session.closed reason: %s', + async (reason) => { + const server = new WebSocketServer({ port: 0, host: '127.0.0.1' }); + await once(server, 'listening'); + const address = server.address(); + if (!address || typeof address === 'string') { + throw new Error('Expected an ephemeral TCP address'); + } + const client = new OpenAI({ apiKey: 'local-test-key', baseURL: `http://127.0.0.1:${address.port}/v1` }); + const finish = new AbortController(); + const markers: number[] = []; + let sawSessionClose = false; + let sawSessionClosed = false; + const timers: ReturnType[] = []; + server.on('connection', (socket) => { + socket.on('message', (raw) => { + const event: { type: string; audio: string } = JSON.parse(raw.toString()); + if (event.type === 'session.start') { + socket.send( + JSON.stringify({ type: 'session.started', session: { id: 'early', status: 'active' } }), + ); + } else if (event.type === 'session.input_audio.append') { + const [marker] = Buffer.from(event.audio, 'base64'); + expect(marker).toBeDefined(); + if (marker !== undefined) { + markers.push(marker); + } + finish.abort(); + } else if (event.type === 'session.close') { + sawSessionClose = true; + timers.push( + setTimeout(() => { + sawSessionClosed = true; + socket.send( + JSON.stringify({ + type: 'session.closed', + reason, + session: { id: 'early', status: 'active' }, + }), + ); + socket.close(); + }, 200), + ); + } + }); + }); + try { + const result = runTranscriptDemo(client, [Buffer.alloc(1920 * 40, 1), Buffer.alloc(1920 * 40, 2)], { + interrupt: true, + finishSignal: finish.signal, + }); + await (reason === 'close_requested' + ? expect(result).resolves.toMatchObject({ interruptionSentDuringOutput: false }) + : expect(result).rejects.toThrow()); + expect(markers).toEqual([1]); + expect(sawSessionClose).toBe(true); + expect(sawSessionClosed).toBe(true); + } finally { + for (const timer of timers) { + clearTimeout(timer); + } + const closed = once(server, 'close'); + for (const socket of server.clients) { + socket.terminate(); + } + server.close(); + await closed; + } + }, +); + +it.concurrent.each(['before-opening', 'during-startup', 'before-delayed-start'])( + 'stops capture %s without sending an unstarted session.close', + async (stage) => { + const server = new WebSocketServer({ port: 0, host: '127.0.0.1' }); + await once(server, 'listening'); + const address = server.address(); + if (!address || typeof address === 'string') { + throw new Error('Expected an ephemeral TCP address'); + } + const client = new OpenAI({ apiKey: 'local-test-key', baseURL: `http://127.0.0.1:${address.port}/v1` }); + const finish = new AbortController(); + const events: string[] = []; + let connections = 0; + let startupTimedOut = false; + const timers: ReturnType[] = []; + if (stage === 'before-opening') { + finish.abort('This caller reason must not enter the error'); + } + server.on('connection', (socket) => { + connections += 1; + socket.on('message', (raw) => { + const event: { type: string } = JSON.parse(raw.toString()); + events.push(event.type); + if (event.type === 'session.start') { + finish.abort('This caller reason must not enter the error'); + if (stage === 'before-delayed-start') { + socket.send( + JSON.stringify({ type: 'session.started', session: { id: 'fixture', status: 'active' } }), + ); + } + // An early stop must not wait for this error or send a command which + // requires an accepted startup. + timers.push( + setTimeout(() => { + startupTimedOut = true; + socket.send(JSON.stringify({ type: 'error', message: 'Synthetic startup timeout' })); + }, 1200), + ); + } else if (event.type === 'session.close') { + socket.send(JSON.stringify({ type: 'session.closed', reason: 'close_requested' })); + } + }); + }); + try { + await expect( + runTranscriptDemo(client, [Buffer.alloc(1920, 1), Buffer.alloc(1920, 2)], { + finishSignal: finish.signal, + }), + ).rejects.toThrow('Live transcript capture stopped before session startup'); + expect(startupTimedOut).toBe(false); + expect(events).toEqual(stage === 'before-opening' ? [] : ['session.start']); + expect(connections).toBe(stage === 'before-opening' ? 0 : 1); + } finally { + for (const timer of timers) { + clearTimeout(timer); + } + const closed = once(server, 'close'); + for (const socket of server.clients) { + socket.terminate(); + } + server.close(); + await closed; + } + }, +); + +it.concurrent.each(['startup', 'input', 'finalization'])( + 'cancels %s promptly without publishing an unconfirmed transcript', + async (stage) => { + const server = new WebSocketServer({ port: 0, host: '127.0.0.1' }); + await once(server, 'listening'); + const address = server.address(); + if (!address || typeof address === 'string') { + throw new Error('Expected an ephemeral TCP address'); + } + + const finish = new AbortController(); + const cancel = new AbortController(); + const events: string[] = []; + const connected = once(server, 'connection'); + server.once('connection', (socket) => { + socket.on('message', (message) => { + const event = JSON.parse(message.toString()); + events.push(event.type); + if (event.type === 'session.start') { + if (stage === 'startup') { + cancel.abort(); + } else { + socket.send( + JSON.stringify({ type: 'session.started', session: { id: 'fixture', status: 'active' } }), + ); + } + } else if (event.type === 'session.input_audio.append') { + if (stage === 'input') { + cancel.abort(); + } else { + finish.abort(); + } + } else if (event.type === 'session.close') { + // A silent peer must not hold cancellation until the 120 s demo deadline. + cancel.abort(); + } + }); + }); + const client = new OpenAI({ apiKey: 'local-test-key', baseURL: `http://127.0.0.1:${address.port}/v1` }); + try { + const demo = runTranscriptDemo(client, [Buffer.alloc(48_000), Buffer.alloc(48_000)], { + finishSignal: finish.signal, + signal: cancel.signal, + interrupt: true, + }); + const [socket] = await connected; + const closed = once(socket, 'close'); + await expect(demo).rejects.toThrow(); + await closed; + const expected = ['session.start']; + if (stage !== 'startup') { + expected.push('session.input_audio.append'); + } + if (stage === 'finalization') { + expected.push('session.close'); + } + // Extra input frames can already be in flight before the server's abort reaches the client. + expect(new Set(events)).toEqual(new Set(expected)); + expect(events[0]).toBe('session.start'); + if (stage === 'finalization') { + expect(events.slice(-1)).toEqual(['session.close']); + } + } finally { + const closed = once(server, 'close'); + for (const socket of server.clients) { + socket.terminate(); + } + server.close(); + await closed; + } + }, + 2000, +); diff --git a/tests/lib/live/transcript-playback.test.ts b/tests/lib/live/transcript-playback.test.ts index f833d0822a..3326c745a2 100644 --- a/tests/lib/live/transcript-playback.test.ts +++ b/tests/lib/live/transcript-playback.test.ts @@ -1,70 +1,97 @@ -import type { PassThrough } from 'node:stream'; -import { beforeEach, describe, expect, it, vi } from 'vitest'; -import { playAudio } from 'openai/helpers/audio'; +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { expect, it, vi } from 'vitest'; import { createPCMPlayback } from '../../../examples/live/transcript-playback'; -// oxlint-disable-next-line anti-slop/no-module-mocking -- Control player failure and stream lifetime without starting ffplay or accessing audio hardware. -vi.mock('openai/helpers/audio', () => ({ playAudio: vi.fn() })); - -const chunks: Buffer[] = []; -let source: PassThrough | undefined; - -beforeEach(() => { - chunks.length = 0; - source = undefined; - vi.mocked(playAudio).mockReset(); - vi.mocked(playAudio).mockImplementation(async (input) => { - // SAFETY: createPCMPlayback supplies its owned PassThrough to the mocked player; retain that stream to check playback completion and destruction. - source = input as PassThrough; - for await (const chunk of source) { - chunks.push(chunk); +it.for(['before-eof', 'after-eof', 'complete', 'missing', 'nonzero'])( + 'handles %s playback without leaving a child running', + { timeout: 5000 }, + async (stage, { skip }) => { + // The controlled player is a POSIX shell script, which Windows cannot execute. + if (process.platform === 'win32' && stage !== 'missing') { + skip(); } - }); -}); - -describe('Live example PCM playback', () => { - it('streams a mono 24 kHz WAV header and unchanged PCM through the existing helper', async () => { - const playback = createPCMPlayback(); - const pcm = Buffer.from([0, 0, 255, 127]); - playback.write(pcm); - await vi.waitFor(() => expect(chunks.length).toBeGreaterThan(0)); - expect(source?.readableEnded).toBe(false); - await playback.finish(); - - const audio = Buffer.concat(chunks); - expect(audio.toString('ascii', 0, 4)).toBe('RIFF'); - expect(audio.readUInt32LE(4)).toBe(0xff_ff_ff_ff); - expect(audio.toString('ascii', 8, 16)).toBe('WAVEfmt '); - expect(audio.readUInt32LE(16)).toBe(16); - expect(audio.readUInt16LE(20)).toBe(1); - expect(audio.readUInt16LE(22)).toBe(1); - expect(audio.readUInt32LE(24)).toBe(24_000); - expect(audio.readUInt32LE(28)).toBe(48_000); - expect(audio.readUInt16LE(32)).toBe(2); - expect(audio.readUInt16LE(34)).toBe(16); - expect(audio.toString('ascii', 36, 40)).toBe('data'); - expect(audio.readUInt32LE(40)).toBe(0xff_ff_ff_ff); - expect(audio.subarray(44)).toEqual(pcm); - expect(playAudio).toHaveBeenCalledTimes(1); - expect(() => playback.write(pcm)).toThrow('after closing'); - await playback.close(); - }); - - it('destroys an unfinished stream on teardown and allows repeated close', async () => { - const playback = createPCMPlayback(); - await playback.close(); - await playback.close(); - expect(source?.destroyed).toBe(true); - expect(() => playback.write(Buffer.alloc(2))).toThrow(); - }); - - it('handles a missing or failed player immediately and reports it on write and finish', async () => { - const error = new Error('ffplay was not found'); - vi.mocked(playAudio).mockRejectedValueOnce(error); - const playback = createPCMPlayback(); - await Promise.resolve(); - expect(() => playback.write(Buffer.alloc(2))).toThrow(error); - await expect(playback.finish()).rejects.toBe(error); - await playback.close(); - }); -}); + const dir = await mkdtemp(path.join(tmpdir(), 'transcript-player-')); + const ready = path.join(dir, 'ready'); + const eof = path.join(dir, 'eof'); + const oldPath = process.env['PATH']; + let player: ReturnType | undefined; + if (stage !== 'missing') { + await writeFile( + path.join(dir, 'ffplay'), + `#!/bin/sh\nexec "${process.execPath}" "${path.join(dir, 'player.mjs')}"\n`, + { mode: 0o700 }, + ); + await writeFile( + path.join(dir, 'player.mjs'), + ` + import fs from 'node:fs'; + fs.writeFileSync(${JSON.stringify(ready)}, String(process.pid)); + const chunks = []; + process.stdin.on('data', (chunk) => chunks.push(chunk)); + process.stdin.on('end', () => { + fs.writeFileSync(${JSON.stringify(eof)}, Buffer.concat(chunks)); + setTimeout(() => process.exit(${stage === 'nonzero' ? 21 : 0}), + ${stage === 'after-eof' ? 2400 : 40}); + }); + `, + ); + } + // The example must find the controlled executable, never a machine's real audio player. + process.env['PATH'] = dir; + try { + player = createPCMPlayback(); + if (stage === 'missing') { + await expect(player.finish()).rejects.toThrow(); + expect(() => player?.write(Buffer.alloc(2))).toThrow(); + return; + } + player.write(Buffer.alloc(480, 3)); + const pid = Number(await vi.waitFor(() => readFile(ready, 'utf-8'))); + if (stage === 'before-eof') { + expect(await readFile(eof).catch(() => {})).toBeUndefined(); + await player.close(); + } else { + const result = player.finish().then( + () => 'completed', + () => 'failed', + ); + const input = await vi.waitFor(() => readFile(eof)); + expect(input.toString('ascii', 0, 4)).toBe('RIFF'); + expect(input.readUInt32LE(4)).toBe(0xff_ff_ff_ff); + expect(input.toString('ascii', 8, 16)).toBe('WAVEfmt '); + expect(input.readUInt32LE(16)).toBe(16); + expect(input.readUInt16LE(20)).toBe(1); + expect(input.readUInt16LE(22)).toBe(1); + expect(input.readUInt32LE(24)).toBe(24_000); + expect(input.readUInt32LE(28)).toBe(48_000); + expect(input.readUInt16LE(32)).toBe(2); + expect(input.readUInt16LE(34)).toBe(16); + expect(input.toString('ascii', 36, 40)).toBe('data'); + expect(input.readUInt32LE(40)).toBe(0xff_ff_ff_ff); + expect(input.subarray(44)).toEqual(Buffer.alloc(480, 3)); + if (stage === 'after-eof') { + const start = performance.now(); + await player.close(); + expect(performance.now() - start).toBeLessThan(900); + expect(await result).toBe('failed'); + } else { + expect(await result).toBe(stage === 'complete' ? 'completed' : 'failed'); + } + } + await player.close(); + await player.close(); + expect(() => player?.write(Buffer.alloc(2))).toThrow(); + expect(() => process.kill(pid, 0)).toThrow(); + } finally { + await player?.close(); + if (oldPath === undefined) { + delete process.env['PATH']; + } else { + process.env['PATH'] = oldPath; + } + await rm(dir, { recursive: true, force: true }); + } + }, +); diff --git a/tests/live/transcript-grouper.live.test.ts b/tests/live/transcript-grouper.live.test.ts index 9c33cfc982..ee0fc0ff63 100644 --- a/tests/live/transcript-grouper.live.test.ts +++ b/tests/live/transcript-grouper.live.test.ts @@ -21,6 +21,9 @@ describeLive('Transcript grouper with TTS input over the public Live API', () => 'groups a real conversation (interrupt: %s)', async (interrupt) => { const result = await runTranscriptDemo(client, audio, { + // This test controls its observation window; it does not infer a final + // answer from live speech. Assert what was actually captured below. + finishSignal: AbortSignal.timeout(45_000), interrupt, websocketOptions: process.env['OPENAI_LIVE_COMPATIBILITY_HEADER'] ? { headers: { 'OpenAI-Alpha': process.env['OPENAI_LIVE_COMPATIBILITY_HEADER'] } }