Unplayed response resumes while nonempty preflight is still awaiting final transcript (1.8.0)

[sherifsarhan]

Summary

On unmodified @livekit/agents@1.8.0, an already-paused response that has never played resumes after the false-interruption timeout even when nonempty PREFLIGHT_TRANSCRIPT input is still awaiting its final transcript.

The pending words are not yet a committed user turn, so the old response plays before the new input can replace it. This is distinct from ordinary noise without recognized words.

Reproducer and observed result

The self-contained test below runs a real AgentSession, STT pumping, adaptive filtering, reply scheduling, and real ParticipantAudioOutput pause/resume/capture methods. Provider transports and the native audio sink are deterministic; there is no room, network, model inference, app code, or credential requirement.

1. Commit First input.; its old response is still waiting for the first TTS frame.
2. Emit caller speech start, 300 ms speech activity, and a nonempty PREFLIGHT_TRANSCRIPT for Second input..
3. Emit VAD end, but do not emit a final transcript yet.
4. Deliver the old response's first audio frame. It is initially paused correctly.
5. Let the unchanged false-interruption timeout elapse, before the 6000 ms transcription timeout.
6. Actual: frame 1 reaches the audio sink while only the first user input is committed.
7. Expected: the unplayed old reply stays held or is safely retired while the nonempty provisional continuation remains unsettled. Provisional text must not itself commit as a user turn.

Two tests: the pending-preflight invariant fails; the control without any recognized words passes and resumes the old response normally. An output frame reaching this synthetic sink proves SDK forwarding, not telephone audibility.

Configuration

turnHandling: {
  turnDetection: 'stt',
  interruption: {
    mode: 'adaptive', minDuration: 300, minWords: 0,
    resumeFalseInterruption: true,
  },
  endpointing: { mode: 'fixed', minDelay: 0, maxDelay: 1500 },
  preemptiveGeneration: { enabled: true, preemptiveTts: false },
},
transcriptionTimeout: 6000,

Environment and command

• Exactly @livekit/agents@1.8.0, no patch-package or package modifications.
• @livekit/rtc-node@0.13.34, tsx@4.23.13, Node 25.6.1, npm 11.9.0, macOS.
• The test replaces adaptive detector construction/transport only and reads the SDK input origin to align scripted STT events. Those test-only internals are not proposed production APIs.
• The real participant-output methods are used with synthetic native-source fields. The fake sink intentionally captures SDK-forwarded frames immediately; it does not model RTP delivery.

Save the code as repro/pending-preflight.test.mts under an empty project:

npm install --save-exact @livekit/agents@1.8.0 @livekit/rtc-node@0.13.34 tsx@4.23.13
node --import tsx --test repro/pending-preflight.test.mts

Complete runnable reproducer

import assert from 'node:assert/strict';
import test, { type TestContext } from 'node:test';
import { setImmediate as settle } from 'node:timers/promises';
import { ReadableStream, type ReadableStreamDefaultController } from 'node:stream/web';
import { AudioFrame } from '@livekit/rtc-node';
import { asLanguageCode, initializeLogger, stt, voice, VAD, VADEventType,
  type VADEvent, type VADStream } from '@livekit/agents';
import { AgentActivity } from '../node_modules/@livekit/agents/dist/voice/agent_activity.js';
import { ParticipantAudioOutput } from '../node_modules/@livekit/agents/dist/voice/room_io/_output.js';
import { Future } from '../node_modules/@livekit/agents/dist/utils.js';
import { AdaptiveInterruptionDetector } from '../node_modules/@livekit/agents/dist/inference/interruption/interruption_detector.js';
import type { InterruptionStreamBase } from '../node_modules/@livekit/agents/dist/inference/interruption/interruption_stream.js';
import type { OverlappingSpeechEvent } from '../node_modules/@livekit/agents/dist/inference/interruption/types.js';

initializeLogger({ pretty: false, level: 'silent' });
async function until(predicate: () => boolean, label: string) {
  const deadline = performance.now() + 3000;
  while (!predicate() && performance.now() < deadline) await settle();
  assert.ok(predicate(), label);
}

// Real session/recognition/scheduling/output. Script only provider transports
// and native audio sink. No application helper, patch or model connection.
async function fixture(t: TestContext) {
  let vadEvents!: ReadableStreamDefaultController<VADEvent>;
  class ScriptedVAD extends VAD {
    label = 'synthetic-vad';
    constructor() { super({ updateInterval: 1 }); }
    stream(): VADStream {
      const source = new ReadableStream<VADEvent>({ start(c) { vadEvents = c; } });
      return { updateInputStream() {}, detachInputStream() {}, flush() {},
        close() { vadEvents.close(); }, [Symbol.asyncIterator]: () => source[Symbol.asyncIterator](),
      } as unknown as VADStream;
    }
  }
  class ScriptedSTT extends stt.STT {
    label = 'synthetic-stt';
    constructor() { super({ streaming: true, interimResults: true, alignedTranscript: 'chunk' }); }
    protected async _recognize(): Promise<stt.SpeechEvent> { throw new Error('No provider calls'); }
    stream(): stt.SpeechStream { throw new Error('Events supplied through sttNode'); }
  }
  const detector = new AdaptiveInterruptionDetector({ apiKey: 'synthetic', apiSecret: 'synthetic', baseUrl: 'http://127.0.0.1:1' });
  t.mock.method(detector, 'createStream', () => {
    let events!: ReadableStreamDefaultController<OverlappingSpeechEvent>;
    const source = new ReadableStream<OverlappingSpeechEvent>({ start(c) { events = c; } });
    return { stream: () => source, async pushFrame() {}, async close() { events.close(); } } as unknown as InterruptionStreamBase;
  });
  // Replace construction of the inference transport, not interruption logic.
  t.mock.method(AgentActivity.prototype as unknown as { resolveInterruptionDetector(): AdaptiveInterruptionDetector },
    'resolveInterruptionDetector', () => detector);

  class Output extends ParticipantAudioOutput {
    captured = 0;
    played: number[] = [];
    override async captureFrame(frame: AudioFrame) { this.captured++; await super.captureFrame(frame); }
  }
  // Native-source construction is unavailable in a no-room unit test. Preserve
  // real ParticipantAudioOutput pause/resume/capture/clear methods and fields.
  const output = Object.create(Output.prototype) as Output;
  Object.assign(output, new class extends voice.AudioOutput {
    constructor() { super(24000, undefined, { pause: true }); }
    clearBuffer() {}
  }());
  const resolved = () => { const f = new Future<void>(); f.resolve(); return f; };
  Object.assign(output, { captured: 0, played: [], startedFuture: resolved(),
    playbackEnabledFuture: resolved(), forwardingIdleFuture: resolved(),
    interruptedFuture: new Future<void>(), firstFrameEmitted: false, pushedDuration: 0,
    sourcePushedDuration: 0, sourceDiscardedDuration: 0, interruptionGeneration: 0, forwardingCount: 0,
    audioSource: { queuedDuration: 0, clearQueue() {}, async waitForPlayout() {},
      async captureFrame(frame: AudioFrame) { output.played.push(frame.data[0]!); } },
  });
  let sttEvents!: ReadableStreamDefaultController<stt.SpeechEvent>;
  const committed: string[] = [];
  const preflights: string[] = [];
  const resumed: boolean[] = [];
  let ttsCount = 0;
  let releaseFirst!: () => void;
  const firstTts = new Promise<void>(resolve => { releaseFirst = resolve; });
  class Agent extends voice.Agent {
    override async sttNode(audio: ReadableStream<AudioFrame>) {
      void (async () => { for await (const _ of audio) {} })();
      return new ReadableStream<stt.SpeechEvent>({ start(c) { sttEvents = c; } });
    }
    override async onUserTurnCompleted(_ctx: unknown, message: { textContent?: string }) {
      committed.push(message.textContent ?? '');
    }
    override async ttsNode(text: ReadableStream<string>) {
      for await (const _ of text) {}
      const index = ++ttsCount;
      return new ReadableStream<AudioFrame>({ async start(c) {
        if (index === 1) await firstTts;
        c.enqueue(new AudioFrame(new Int16Array(480).fill(index), 24000, 1, 480));
        c.close();
      } });
    }
  }
  const session = new voice.AgentSession({ stt: new ScriptedSTT(), vad: new ScriptedVAD(),
    llm: new voice.testing.FakeLLM([
      { input: 'First input.', content: 'Old response.' },
      { input: 'Second input.', content: 'New response.' },
    ]), userAwayTimeout: null, transcriptionTimeout: 6000,
    turnHandling: { turnDetection: 'stt',
      interruption: { mode: 'adaptive', minDuration: 300, minWords: 0, resumeFalseInterruption: true },
      endpointing: { mode: 'fixed', minDelay: 0, maxDelay: 1500 },
      preemptiveGeneration: { enabled: true, preemptiveTts: false } },
  });
  session.output.audio = output;
  let input!: ReadableStreamDefaultController<AudioFrame>;
  const inputStream = new ReadableStream<AudioFrame>({ start(c) { input = c; } });
  class Input extends voice.AudioInput { override get stream() { return inputStream; } }
  session.input.audio = new Input();
  session.on(voice.AgentSessionEventTypes.AgentFalseInterruption, e => resumed.push(e.resumed));
  session.on(voice.AgentSessionEventTypes.UserInputTranscribed, e => { if (!e.isFinal) preflights.push(e.transcript); });
  await session.start({ agent: new Agent({ instructions: 'Synthetic regression.' }) });
  input.enqueue(new AudioFrame(new Int16Array(72000), 24000, 1, 72000));
  await until(() => !!sttEvents && !!vadEvents, 'input pipelines started');
  // Read-only fixture alignment to the real SDK input origin; no state mutation.
  const recognition = (session as unknown as { activity: { audioRecognition: { inputStartedAt?: number } } }).activity.audioRecognition;
  await until(() => recognition.inputStartedAt !== undefined, 'input clock established');
  const origin = recognition.inputStartedAt!;
  const elapsed = () => (Date.now() - origin) / 1000;
  t.after(async () => { releaseFirst(); input.close(); await session.close(); });
  await session.say('Synthetic greeting.', { audio: new ReadableStream<AudioFrame>({ start(c) {
    c.enqueue(new AudioFrame(new Int16Array(480), 24000, 1, 480)); c.close();
  } }) });
  await new Promise(resolve => setTimeout(resolve, (session.sessionOptions.aecWarmupDuration ?? 0) + 20));
  output.played.length = 0; output.captured = 0;
  const raw = async (event: stt.SpeechEvent) => { sttEvents.enqueue(event); await settle(); };
  const transcript = (type: stt.SpeechEventType, text: string, startTime: number, endTime: number): stt.SpeechEvent => ({ type,
    alternatives: [{ text, startTime, endTime, confidence: 1, language: asLanguageCode('en') }] });
  const vad = async (type: VADEventType, duration = 0) => {
    vadEvents.enqueue({ type, samplesIndex: 0, timestamp: Date.now(), speechDuration: duration,
      silenceDuration: 0, frames: [], probability: 1, inferenceDuration: 0,
      speaking: type !== VADEventType.END_OF_SPEECH, rawAccumulatedSilence: 0, rawAccumulatedSpeech: duration });
    await settle();
  };
  const start = elapsed();
  await raw({ type: stt.SpeechEventType.START_OF_SPEECH });
  await new Promise(resolve => setTimeout(resolve, 20));
  await raw(transcript(stt.SpeechEventType.FINAL_TRANSCRIPT, 'First input.', start, elapsed()));
  await raw({ type: stt.SpeechEventType.END_OF_SPEECH });
  await until(() => ttsCount === 1, 'old response waits on first TTS frame');
  return { session, output, committed, preflights, resumed, releaseFirst,
    async continuation(withPreflight: boolean) {
      const start = elapsed();
      await vad(VADEventType.START_OF_SPEECH);
      await raw({ type: stt.SpeechEventType.START_OF_SPEECH });
      await new Promise(resolve => setTimeout(resolve, 300));
      await vad(VADEventType.INFERENCE_DONE, 300);
      if (withPreflight) await raw(transcript(stt.SpeechEventType.PREFLIGHT_TRANSCRIPT, 'Second input.', start, elapsed()));
      await vad(VADEventType.END_OF_SPEECH);
    },
  };
}

for (const withPreflight of [true, false]) {
  test(withPreflight ? 'unplayed response must not resume while nonempty preflight awaits final' : 'noise without recognized words still resumes normally',
    { timeout: 10000 }, async t => {
      const f = await fixture(t);
      await f.continuation(withPreflight);
      f.releaseFirst();
      await until(() => f.output.captured === 1, 'old frame buffered at real SDK output');
      assert.deepEqual(f.output.played, [], 'pending reply is initially paused');
      await new Promise(resolve => setTimeout(resolve, f.session.sessionOptions.turnHandling.interruption.falseInterruptionTimeout + 100));
      assert.deepEqual(f.committed, ['First input.'], 'preflight is not a committed caller turn');
      assert.deepEqual(f.preflights, withPreflight ? ['Second input.'] : []);
      assert.deepEqual(f.output.played, withPreflight ? [] : [1],
        'nonempty provisional input must remain distinct from noise with no recognized words');
      assert.deepEqual(f.resumed, withPreflight ? [] : [true]);
    });
}

Relationship to existing fixes / request

This is adjacent to #1909 and #2059, but this reproducer does not depend on a cancellation timeout, multiple model responses for the same interim, or permanent loss of the first frame. The frame successfully resumes; the problem is its authorization while later nonempty input remains provisional.

The published release already includes #2263 (pending-reply launch gating) and #2290 (adaptive interruptions across tools). Initial pausing works here. The later false-interruption resume is the failing boundary.

Disabling automatic resumption avoids this path but changes noise/backchannel behavior. Could the SDK preserve the distinction between (a) no recognized words and (b) nonempty provisional input awaiting a final for a never-played response? Alternatively, please identify the supported public boundary for safely holding/retiring that response without replacing scheduler logic.