examples: handle Live replies and session finalization

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`. In a terminal, press Enter to finish capturing the transcript, or
+Ctrl-C to cancel and stop playback. Without a terminal, the example uses a
+45-second demo budget after preparing the TTS clips, including session startup.
+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
 

examples/live/transcript-demo.ts

@@ -49,11 +49,19 @@ export async function prepareTranscriptDemo(client: OpenAI): Promise<Buffer[]> {
  * 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<TranscriptDemoResult> {
   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,7 +131,7 @@ 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'));
     }
   });
@@ -143,6 +157,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,23 +189,27 @@ export async function runTranscriptDemo(
   async function sendPCM(pcm: Buffer): Promise<void> {
     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',
         audio: chunk.toString('base64'),
       });
       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 });
     }
   }
 
   // Keep sending silence while waiting: the duplex model still needs live audio input.
   const silence = Buffer.alloc(CHUNK_BYTES * 5);
   async function waitWithAudio(predicate: () => boolean, description: string): Promise<void> {
     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 +223,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 +239,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 +270,6 @@ export async function runTranscriptDemo(
       live.close();
     }
   }
-  abort.signal.throwIfAborted();
+  signal.throwIfAborted();
   return result;
 }

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,59 @@ async function main(): Promise<void> {
   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);
+    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.');
+    } else {
+      console.log('Capturing for up to 45 seconds, then closing the session.');
+    }
     const result = await runTranscriptDemo(client, audio, {
+      finishSignal: input ? 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()]);
   }
 }

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<void>((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<void> {
       stream.destroy();
+      player.kill();
       await finished;
     },
   };