Reproduction script for STTv2 #sendTask stream teardown TypeError (livekit/agents-js)

repro-sttv2-teardown.ts

/**
 * Reproduction script for the STTv2 #sendTask stream teardown TypeError.
 *
 * This replicates the exact iterator + Promise.race pattern from
 * plugins/deepgram/src/stt_v2.ts lines 278-327, showing that the
 * original guard `!('value' in result)` fails to detect a finished
 * iterator, causing a TypeError when the code tries to read
 * `.data.buffer` on `undefined`.
 *
 * Usage (from repo root, after `pnpm install`):
 *   npx tsx repro-sttv2-teardown.ts
 *
 * Expected output:
 *   [OLD CODE] TypeError: Cannot read properties of undefined (reading 'data')
 *   [NEW CODE] Stream teardown completed cleanly
 */

import { AsyncIterableQueue } from '@livekit/agents';

// Stand-in for AudioFrame — just needs a `.data.buffer` property
// to trigger the same crash path as the real code.
interface FakeAudioFrame {
  data: { buffer: ArrayBuffer };
}

const FLUSH_SENTINEL = Symbol('FLUSH_SENTINEL');

// Simulate the reconnect event that never fires (normal teardown path)
const reconnectEvent = {
  wait: () => new Promise<{ abort: true }>(() => {}), // never resolves
};

/**
 * Replicates #sendTask from stt_v2.ts with the OLD guard:
 *   `if (hasEnded && !('value' in result))`
 */
async function sendTaskOld(input: AsyncIterableQueue<FakeAudioFrame | typeof FLUSH_SENTINEL>) {
  let hasEnded = false;
  const iterator = input[Symbol.asyncIterator]();

  while (true) {
    const nextPromise = iterator.next();
    const abortPromise = reconnectEvent.wait().then(() => ({ abort: true }) as const);
    const result = await Promise.race([nextPromise, abortPromise]);

    if ('abort' in result || result.done) {
      if (!('abort' in result) && result.done) {
        hasEnded = true;
      } else {
        break;
      }
    }

    // OLD CODE — bug is here
    if (hasEnded && !('value' in result)) {
      // Stream ended with no data to process — flush below
    } else if ('value' in result) {
      const data = result.value;

      if (data === FLUSH_SENTINEL) {
        // flush path (not relevant to bug)
      } else {
        // This is the crash: data is undefined, so .data throws TypeError
        void (data as FakeAudioFrame).data.buffer;
      }
    }

    if (hasEnded) break;
  }
}

/**
 * Replicates #sendTask from stt_v2.ts with the NEW guard:
 *   `if (hasEnded && result.value === undefined)`
 */
async function sendTaskNew(input: AsyncIterableQueue<FakeAudioFrame | typeof FLUSH_SENTINEL>) {
  let hasEnded = false;
  const iterator = input[Symbol.asyncIterator]();

  while (true) {
    const nextPromise = iterator.next();
    const abortPromise = reconnectEvent.wait().then(() => ({ abort: true }) as const);
    const result = await Promise.race([nextPromise, abortPromise]);

    if ('abort' in result || result.done) {
      if (!('abort' in result) && result.done) {
        hasEnded = true;
      } else {
        break;
      }
    }

    // NEW CODE — fix
    if (hasEnded && result.value === undefined) {
      // Stream ended with no data to process — flush below
    } else if ('value' in result) {
      const data = result.value;

      if (data === FLUSH_SENTINEL) {
        // flush path
      } else {
        void (data as FakeAudioFrame).data.buffer;
      }
    }

    if (hasEnded) break;
  }
}

async function main() {
  // --- OLD CODE: crashes ---
  const input1 = new AsyncIterableQueue<FakeAudioFrame | typeof FLUSH_SENTINEL>();
  input1.put({ data: { buffer: new ArrayBuffer(320) } });
  input1.close();

  try {
    await sendTaskOld(input1);
    console.log('[OLD CODE] No error (unexpected)');
  } catch (e) {
    console.log(`[OLD CODE] ${e}`);
  }

  // --- NEW CODE: clean teardown ---
  const input2 = new AsyncIterableQueue<FakeAudioFrame | typeof FLUSH_SENTINEL>();
  input2.put({ data: { buffer: new ArrayBuffer(320) } });
  input2.close();

  try {
    await sendTaskNew(input2);
    console.log('[NEW CODE] Stream teardown completed cleanly');
  } catch (e) {
    console.log(`[NEW CODE] ${e} (unexpected)`);
  }
}

main();