fix: support adaptive interruption without word timestamps

.changeset/fuzzy-dogs-interrupt.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Support adaptive interruption for STT providers without aligned word timestamps.

agents/etc/agents.api.md

@@ -531,7 +531,7 @@ export class AgentSession<UserData = UnknownUserData> extends AgentSession_base
         force?: boolean;
     }): Future<void, Error>;
     // (undocumented)
-    get interruptionDetection(): "adaptive" | "vad" | undefined;
+    get interruptionDetection(): "vad" | "adaptive" | undefined;
     // @internal (undocumented)
     readonly _keytermDetector: KeytermDetector;
     get keyterms(): string[];
@@ -5929,7 +5929,7 @@ export function resolveExpressiveOptions(expr: ExpressiveOptions, options: {
 // Warning: (ae-missing-release-tag) "RimeModels" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal)
 //
 // @public (undocumented)
-type RimeModels = 'rime/arcana' | 'rime/coda' | 'rime/mistv2' | 'rime/mistv3' | 'rime/mist';
+type RimeModels = 'rime/coda' | 'rime/mistv2' | 'rime/mistv3' | 'rime/mist';
 
 // Warning: (ae-missing-release-tag) "RimeOptions" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal)
 //
@@ -6643,6 +6643,7 @@ interface SpeechData {
 interface SpeechEvent {
     // (undocumented)
     alternatives?: [SpeechData, ...SpeechData[]];
+    createdAt?: number;
     // (undocumented)
     recognitionUsage?: RecognitionUsage;
     // (undocumented)
@@ -9027,7 +9028,7 @@ export const zipFunctionCallsAndOutputs: (event: FunctionToolsExecutedEvent) =>
 // src/llm/tool_context.ts:746:3 - (ae-unresolved-link) The @link reference could not be resolved: The reference is ambiguous because "ToolFlag" has more than one declaration; you need to add a TSDoc member reference selector
 // src/metrics/base.ts:194:3 - (ae-forgotten-export) The symbol "RealtimeModelMetricsInputTokenDetails" needs to be exported by the entry point index.d.ts
 // src/metrics/base.ts:198:3 - (ae-forgotten-export) The symbol "RealtimeModelMetricsOutputTokenDetails" needs to be exported by the entry point index.d.ts
-// src/stt/stt.ts:358:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "STT"
+// src/stt/stt.ts:361:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "STT"
 // src/utils.ts:501:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "cancelled"
 // src/voice/agent_session.ts:380:3 - (ae-unresolved-link) The @link reference could not be resolved: This type of declaration is not supported yet by the resolver
 // src/voice/agent_session.ts:988:5 - (ae-forgotten-export) The symbol "RecordingOptions" needs to be exported by the entry point index.d.ts

agents/src/inference/interruption/interruption_failover.test.ts

@@ -27,6 +27,7 @@ import { MockWebSocket } from './_mock_ws.js';
 import { apiConnectDefaults } from './defaults.js';
 import { AdaptiveInterruptionDetector } from './interruption_detector.js';
 import { InterruptionStreamBase, InterruptionStreamSentinel } from './interruption_stream.js';
+import type { OverlappingSpeechEvent } from './types.js';
 
 // ---------------------------------------------------------------------------
 // Mock `ws` so the WebSocket transport can be driven deterministically.
@@ -340,7 +341,7 @@ describe('interruption updateOptions reconnect', () => {
 
 function createHooks(): RecognitionHooks {
   return {
-    onInterruption: vi.fn(),
+    onOverlapSpeech: vi.fn(),
     onBackchannelConfirmed: vi.fn(),
     onStartOfSpeech: vi.fn(),
     onVADInferenceDone: vi.fn(),
@@ -355,6 +356,65 @@ function createHooks(): RecognitionHooks {
 }
 
 describe('interruption failover error emission', () => {
+  it('routes overlap verdicts through recognition hooks', async () => {
+    const event: OverlappingSpeechEvent = {
+      type: 'overlapping_speech',
+      detectedAt: Date.now(),
+      isInterruption: true,
+      totalDurationInS: 0,
+      predictionDurationInS: 0,
+      detectionDelayInS: 0,
+      probability: 1,
+      numRequests: 1,
+    };
+    const hooks = createHooks();
+    const eventStream = new ReadableStream<OverlappingSpeechEvent>({
+      start(controller) {
+        controller.enqueue(event);
+        controller.close();
+      },
+    });
+    const detectorStream = {
+      stream: () => eventStream,
+      pushFrame: async () => {},
+      close: async () => {},
+    };
+    const detector = {
+      label: 'mock-detector',
+      createStream: () => detectorStream,
+      emitError: vi.fn(),
+    };
+    const recognition = new AudioRecognition({
+      recognitionHooks: hooks,
+      interruptionDetection: detector as unknown as AdaptiveInterruptionDetector,
+    });
+    const abortController = new AbortController();
+    const task = (
+      recognition as unknown as {
+        createInterruptionTask: (
+          detector: AdaptiveInterruptionDetector,
+          signal: AbortSignal,
+        ) => Promise<void>;
+      }
+    ).createInterruptionTask(
+      detector as unknown as AdaptiveInterruptionDetector,
+      abortController.signal,
+    );
+
+    const startedAt = performance.now();
+    while (
+      !vi.mocked(hooks.onOverlapSpeech).mock.calls.length &&
+      performance.now() - startedAt < 2_000
+    ) {
+      await sleep(5);
+    }
+    abortController.abort();
+    await task;
+
+    expect(hooks.onOverlapSpeech).toHaveBeenCalledOnce();
+    expect(hooks.onOverlapSpeech).toHaveBeenCalledWith(event);
+  });
+
   it('emits exactly one unrecoverable error for a non-retryable transport failure', async () => {
     // Mirrors how ws_transport constructs the connection-rejected error (retryable forced off).
     const transportError = new APIStatusError({

agents/src/stt/stt.test.ts

@@ -0,0 +1,54 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import { describe, expect, it } from 'vitest';
+import type { AudioBuffer } from '../utils.js';
+import { STT, type SpeechEvent, SpeechEventType, SpeechStream } from './stt.js';
+
+class UntimestampedSTT extends STT {
+  label = 'untimestamped-stt';
+
+  constructor() {
+    super({ streaming: true, interimResults: false });
+  }
+
+  protected async _recognize(_frame: AudioBuffer): Promise<SpeechEvent> {
+    return { type: SpeechEventType.FINAL_TRANSCRIPT };
+  }
+
+  stream(): SpeechStream {
+    return new UntimestampedSpeechStream(this);
+  }
+}
+
+class UntimestampedSpeechStream extends SpeechStream {
+  label = 'untimestamped-speech-stream';
+
+  protected async run(): Promise<void> {
+    this.queue.put({ type: SpeechEventType.START_OF_SPEECH });
+  }
+}
+
+describe('STT event timestamps', () => {
+  it('timestamps non-streaming recognition results', async () => {
+    const before = Date.now();
+    const event = await new UntimestampedSTT().recognize([]);
+
+    expect(event.createdAt).toBeGreaterThanOrEqual(before);
+    expect(event.createdAt).toBeLessThanOrEqual(Date.now());
+  });
+
+  it('timestamps events before a speech stream delivers them', async () => {
+    const stream = new UntimestampedSTT().stream();
+    const before = Date.now();
+
+    try {
+      const { value } = await stream.next();
+
+      expect(value?.createdAt).toBeGreaterThanOrEqual(before);
+      expect(value?.createdAt).toBeLessThanOrEqual(Date.now());
+    } finally {
+      stream.close();
+    }
+  });
+});

agents/src/stt/stt.ts

@@ -117,6 +117,8 @@ export interface SpeechEvent {
   alternatives?: [SpeechData, ...SpeechData[]];
   requestId?: string;
   recognitionUsage?: RecognitionUsage;
+  /** Wall-clock time when this event was created, in milliseconds. STT boundaries populate it. */
+  createdAt?: number;
 }
 
 /**
@@ -211,6 +213,7 @@ export abstract class STT extends (EventEmitter as new () => TypedEmitter<STTCal
   async recognize(frame: AudioBuffer, abortSignal?: AbortSignal): Promise<SpeechEvent> {
     const startTime = process.hrtime.bigint();
     const event = await this._recognize(frame, abortSignal);
+    event.createdAt ??= Date.now();
     const durationMs = Number((process.hrtime.bigint() - startTime) / BigInt(1000000));
     this.emit('metrics_collected', {
       type: 'stt_metrics',
@@ -434,6 +437,7 @@ export abstract class SpeechStream implements AsyncIterableIterator<SpeechEvent>
 
   protected async monitorMetrics() {
     for await (const event of this.queue) {
+      event.createdAt ??= Date.now();
       if (!this.output.closed) {
         try {
           this.output.put(event);

agents/src/voice/agent.test.ts

@@ -800,10 +800,11 @@ describe('Agent', () => {
 
     it.each([
       ['word', true],
+      [false, true],
       [false, false],
     ] as const)(
-      'should use the STT alignment claim to resolve automatic adaptive interruption (%s)',
-      (alignedTranscript, expectedAdaptive) => {
+      'should resolve automatic adaptive interruption independently of STT alignment (%s, streaming=%s)',
+      (alignedTranscript, streaming) => {
         const previousDevMode = process.env.LIVEKIT_DEV_MODE;
         const previousApiKey = process.env.LIVEKIT_API_KEY;
         const previousApiSecret = process.env.LIVEKIT_API_SECRET;
@@ -815,7 +816,7 @@ describe('Agent', () => {
           const activity = Object.create(AgentActivity.prototype) as any;
           activity.agent = {
             _llm: {},
-            _stt: { capabilities: { alignedTranscript, streaming: true } },
+            _stt: { capabilities: { alignedTranscript, streaming } },
             _vad: {},
             turnHandling: { interruption: { enabled: true }, turnDetection: 'stt' },
           };
@@ -825,11 +826,10 @@ describe('Agent', () => {
             turnDetection: 'stt',
           };
           activity.logger = { warn: vi.fn() };
-          activity.onInterruptionOverlappingSpeech = vi.fn();
           activity.onInterruptionMetricsCollected = vi.fn();
           activity.onInterruptionError = vi.fn();
 
-          expect(activity.resolveInterruptionDetector() !== undefined).toBe(expectedAdaptive);
+          expect(activity.resolveInterruptionDetector()).toBeDefined();
         } finally {
           if (previousDevMode === undefined) delete process.env.LIVEKIT_DEV_MODE;
           else process.env.LIVEKIT_DEV_MODE = previousDevMode;
@@ -841,6 +841,49 @@ describe('Agent', () => {
       },
     );
 
+    it('should allow explicit adaptive interruption with an unaligned STT', () => {
+      const previousDevMode = process.env.LIVEKIT_DEV_MODE;
+      const previousRemoteEotUrl = process.env.LIVEKIT_REMOTE_EOT_URL;
+      const previousApiKey = process.env.LIVEKIT_API_KEY;
+      const previousApiSecret = process.env.LIVEKIT_API_SECRET;
+      delete process.env.LIVEKIT_DEV_MODE;
+      delete process.env.LIVEKIT_REMOTE_EOT_URL;
+      process.env.LIVEKIT_API_KEY = 'fake';
+      process.env.LIVEKIT_API_SECRET = 'fake-secret';
+
+      try {
+        const activity = Object.create(AgentActivity.prototype) as any;
+        activity.agent = {
+          _llm: {},
+          _stt: { capabilities: { alignedTranscript: false, streaming: true } },
+          _vad: {},
+          turnHandling: {
+            interruption: { enabled: true, mode: 'adaptive' },
+            turnDetection: 'stt',
+          },
+        };
+        activity.agentSession = {
+          _textOnly: false,
+          interruptionDetection: undefined,
+          turnDetection: 'stt',
+        };
+        activity.logger = { warn: vi.fn() };
+        activity.onInterruptionMetricsCollected = vi.fn();
+        activity.onInterruptionError = vi.fn();
+
+        expect(activity.resolveInterruptionDetector()).toBeDefined();
+      } finally {
+        if (previousDevMode !== undefined) process.env.LIVEKIT_DEV_MODE = previousDevMode;
+        if (previousRemoteEotUrl !== undefined) {
+          process.env.LIVEKIT_REMOTE_EOT_URL = previousRemoteEotUrl;
+        }
+        if (previousApiKey === undefined) delete process.env.LIVEKIT_API_KEY;
+        else process.env.LIVEKIT_API_KEY = previousApiKey;
+        if (previousApiSecret === undefined) delete process.env.LIVEKIT_API_SECRET;
+        else process.env.LIVEKIT_API_SECRET = previousApiSecret;
+      }
+    });
+
     it('should keep automatic adaptive interruption off in plain production', () => {
       const previousDevMode = process.env.LIVEKIT_DEV_MODE;
       const previousRemoteEotUrl = process.env.LIVEKIT_REMOTE_EOT_URL;

agents/src/voice/agent_activity.test.ts

@@ -455,6 +455,11 @@ describe('AgentActivity - mainTask', () => {
   it('does not deadlock cancelling a paused speech whose generation never finishes', async () => {
     const handle = SpeechHandle.create({ allowInterruptions: true });
     handle._authorizeGeneration();
+    const onEndOfAgentSpeech = (
+      AgentActivity.prototype as unknown as {
+        onEndOfAgentSpeech: (endedAt: number, options?: { paused?: boolean }) => Promise<void>;
+      }
+    ).onEndOfAgentSpeech;
 
     const fakeActivity = {
       cancelSpeechPauseTask: undefined,
@@ -471,6 +476,7 @@ describe('AgentActivity - mainTask', () => {
       audioRecognition: {
         onEndOfAgentSpeech: vi.fn(async () => {}),
       },
+      onEndOfAgentSpeech,
       agentSession: {
         sessionOptions: {
           turnHandling: {
@@ -808,6 +814,11 @@ describe('AgentActivity - speech completion', () => {
         done: () => true,
       },
       audioRecognition,
+      onEndOfAgentSpeech: (
+        AgentActivity.prototype as unknown as {
+          onEndOfAgentSpeech: (endedAt: number) => Promise<void>;
+        }
+      ).onEndOfAgentSpeech,
       agentSession: {
         agentState: 'speaking',
         _updateAgentState: vi.fn((state: string) => {