Add Adaptive Interruption (bargein)

.changeset/config.json

@@ -8,13 +8,7 @@
   ],
   "commit": false,
   "ignore": ["livekit-agents-examples"],
-  "fixed": [
-    [
-      "@livekit/agents",
-      "@livekit/agents-plugin-*",
-      "@livekit/agents-plugins-test"
-    ]
-  ],
+  "fixed": [["@livekit/agents", "@livekit/agents-plugin-*", "@livekit/agents-plugins-test"]],
   "access": "public",
   "baseBranch": "main",
   "updateInternalDependencies": "patch",

.changeset/metal-teeth-buy.md

@@ -0,0 +1,6 @@
+---
+'@livekit/agents': minor
+---
+
+- Add adaptive interruption handling
+- Add remote session event handler

.github/workflows/test.yml

@@ -46,11 +46,11 @@ jobs:
       - name: Test agents
         if: steps.filter.outputs.agents-or-tests == 'true' || github.event_name == 'push'
         run: pnpm test agents
-      - name: Test examples
-        if: (steps.filter.outputs.examples == 'true' || github.event_name == 'push') && secrets.OPENAI_API_KEY != ''
-        env:
-          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
-        run: pnpm test:examples
+      # - name: Test examples
+      #   if: (steps.filter.outputs.examples == 'true' || github.event_name == 'push')
+      #   env:
+      #     OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
+      #   run: pnpm test:examples
       # TODO (AJS-83) Re-enable once plugins are refactored with abort controllers
       # - name: Test all plugins
       #   if: steps.filter.outputs.agents-or-tests == 'true' || github.event_name != 'pull_request'

agents/package.json

@@ -46,9 +46,10 @@
     "zod": "^3.25.76"
   },
   "dependencies": {
+    "@bufbuild/protobuf": "^1.10.0",
     "@ffmpeg-installer/ffmpeg": "^1.1.0",
     "@livekit/mutex": "^1.1.1",
-    "@livekit/protocol": "^1.43.0",
+    "@livekit/protocol": "^1.45.1",
     "@livekit/typed-emitter": "^3.0.0",
     "@opentelemetry/api": "^1.9.0",
     "@opentelemetry/api-logs": "^0.54.0",
@@ -69,6 +70,7 @@
     "heap-js": "^2.6.0",
     "json-schema": "^0.4.0",
     "livekit-server-sdk": "^2.14.1",
+    "ofetch": "^1.5.1",
     "openai": "^6.8.1",
     "pidusage": "^4.0.1",
     "pino": "^8.19.0",

agents/src/cli.ts

@@ -148,6 +148,7 @@ export const runApp = (opts: ServerOptions) => {
       opts.apiSecret = globalOptions.apiSecret || opts.apiSecret;
       opts.logLevel = commandOptions.logLevel;
       opts.workerToken = globalOptions.workerToken || opts.workerToken;
+      process.env.LIVEKIT_DEV_MODE = '1';
       runServer({
         opts,
         production: false,
@@ -169,6 +170,7 @@ export const runApp = (opts: ServerOptions) => {
       opts.apiSecret = globalOptions.apiSecret || opts.apiSecret;
       opts.logLevel = commandOptions.logLevel;
       opts.workerToken = globalOptions.workerToken || opts.workerToken;
+      process.env.LIVEKIT_DEV_MODE = '1';
       runServer({
         opts,
         production: false,

agents/src/constants.ts

@@ -7,3 +7,17 @@ export const TOPIC_TRANSCRIPTION = 'lk.transcription';
 export const ATTRIBUTE_TRANSCRIPTION_SEGMENT_ID = 'lk.segment_id';
 export const ATTRIBUTE_PUBLISH_ON_BEHALF = 'lk.publish_on_behalf';
 export const TOPIC_CHAT = 'lk.chat';
+
+export const ATTRIBUTE_AGENT_STATE = 'lk.agent.state';
+export const ATTRIBUTE_AGENT_NAME = 'lk.agent.name';
+
+// TODO(eval): export const ATTRIBUTE_SIMULATOR = 'lk.simulator';
+
+export const TOPIC_CLIENT_EVENTS = 'lk.agent.events';
+export const RPC_GET_SESSION_STATE = 'lk.agent.get_session_state';
+export const RPC_GET_CHAT_HISTORY = 'lk.agent.get_chat_history';
+export const RPC_GET_AGENT_INFO = 'lk.agent.get_agent_info';
+export const RPC_SEND_MESSAGE = 'lk.agent.send_message';
+export const TOPIC_AGENT_REQUEST = 'lk.agent.request';
+export const TOPIC_AGENT_RESPONSE = 'lk.agent.response';
+export const TOPIC_SESSION_MESSAGES = 'lk.agent.session';

agents/src/inference/interruption/defaults.ts

@@ -0,0 +1,51 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import type { ApiConnectOptions } from './interruption_stream.js';
+import type { InterruptionOptions } from './types.js';
+
+export const MIN_INTERRUPTION_DURATION_IN_S = 0.025 * 2; // 25ms per frame, 2 consecutive frames
+export const THRESHOLD = 0.5;
+export const MAX_AUDIO_DURATION_IN_S = 3.0;
+export const AUDIO_PREFIX_DURATION_IN_S = 0.5;
+export const DETECTION_INTERVAL_IN_S = 0.1;
+export const REMOTE_INFERENCE_TIMEOUT_IN_S = 0.7;
+export const SAMPLE_RATE = 16000;
+export const FRAMES_PER_SECOND = 40;
+export const FRAME_DURATION_IN_S = 0.025; // 25ms per frame
+
+export const apiConnectDefaults: ApiConnectOptions = {
+  maxRetries: 3,
+  retryInterval: 2_000,
+  timeout: 10_000,
+} as const;
+
+/**
+ * Calculate the retry interval using exponential backoff with jitter.
+ * Matches the Python implementation's _interval_for_retry behavior.
+ */
+export function intervalForRetry(
+  attempt: number,
+  baseInterval: number = apiConnectDefaults.retryInterval,
+): number {
+  // Exponential backoff: baseInterval * 2^attempt with some jitter
+  const exponentialDelay = baseInterval * Math.pow(2, attempt);
+  // Add jitter (0-25% of the delay)
+  const jitter = exponentialDelay * Math.random() * 0.25;
+  return exponentialDelay + jitter;
+}
+
+// baseUrl and useProxy are resolved dynamically in the constructor
+// to respect LIVEKIT_REMOTE_EOT_URL environment variable
+export const interruptionOptionDefaults: Omit<InterruptionOptions, 'baseUrl' | 'useProxy'> = {
+  sampleRate: SAMPLE_RATE,
+  threshold: THRESHOLD,
+  minFrames: Math.ceil(MIN_INTERRUPTION_DURATION_IN_S * FRAMES_PER_SECOND),
+  maxAudioDurationInS: MAX_AUDIO_DURATION_IN_S,
+  audioPrefixDurationInS: AUDIO_PREFIX_DURATION_IN_S,
+  detectionIntervalInS: DETECTION_INTERVAL_IN_S,
+  inferenceTimeout: REMOTE_INFERENCE_TIMEOUT_IN_S * 1_000,
+  apiKey: process.env.LIVEKIT_API_KEY || '',
+  apiSecret: process.env.LIVEKIT_API_SECRET || '',
+  minInterruptionDurationInS: MIN_INTERRUPTION_DURATION_IN_S,
+} as const;

agents/src/inference/interruption/errors.ts

@@ -0,0 +1,25 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+/**
+ * Error thrown during interruption detection.
+ */
+export class InterruptionDetectionError extends Error {
+  readonly type = 'interruption_detection_error' as const;
+
+  readonly timestamp: number;
+  readonly label: string;
+  readonly recoverable: boolean;
+
+  constructor(message: string, timestamp: number, label: string, recoverable: boolean) {
+    super(message);
+    this.name = 'InterruptionDetectionError';
+    this.timestamp = timestamp;
+    this.label = label;
+    this.recoverable = recoverable;
+  }
+
+  toString(): string {
+    return `${this.name}: ${this.message} (label=${this.label}, timestamp=${this.timestamp}, recoverable=${this.recoverable})`;
+  }
+}

agents/src/inference/interruption/http_transport.ts

@@ -0,0 +1,206 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import { FetchError, ofetch } from 'ofetch';
+import { TransformStream } from 'stream/web';
+import { z } from 'zod';
+import { APIConnectionError, APIError, APIStatusError, isAPIError } from '../../_exceptions.js';
+import { log } from '../../log.js';
+import { createAccessToken } from '../utils.js';
+import { InterruptionCacheEntry } from './interruption_cache_entry.js';
+import type { OverlappingSpeechEvent } from './types.js';
+import type { BoundedCache } from './utils.js';
+
+export interface PostOptions {
+  baseUrl: string;
+  token: string;
+  signal?: AbortSignal;
+  timeout?: number;
+  maxRetries?: number;
+}
+
+export interface PredictOptions {
+  threshold: number;
+  minFrames: number;
+}
+
+export const predictEndpointResponseSchema = z.object({
+  created_at: z.number(),
+  is_bargein: z.boolean(),
+  probabilities: z.array(z.number()),
+});
+
+export type PredictEndpointResponse = z.infer<typeof predictEndpointResponseSchema>;
+
+export interface PredictResponse {
+  createdAt: number;
+  isBargein: boolean;
+  probabilities: number[];
+  predictionDurationInS: number;
+}
+
+export async function predictHTTP(
+  data: Int16Array,
+  predictOptions: PredictOptions,
+  options: PostOptions,
+): Promise<PredictResponse> {
+  const createdAt = performance.now();
+  const url = new URL(`/bargein`, options.baseUrl);
+  url.searchParams.append('threshold', predictOptions.threshold.toString());
+  url.searchParams.append('min_frames', predictOptions.minFrames.toFixed());
+  url.searchParams.append('created_at', createdAt.toFixed());
+
+  try {
+    const response = await ofetch(url.toString(), {
+      retry: 0,
+      headers: {
+        'Content-Type': 'application/octet-stream',
+        Authorization: `Bearer ${options.token}`,
+      },
+      signal: options.signal,
+      timeout: options.timeout,
+      method: 'POST',
+      body: data,
+    });
+    const { created_at, is_bargein, probabilities } = predictEndpointResponseSchema.parse(response);
+
+    return {
+      createdAt: created_at,
+      isBargein: is_bargein,
+      probabilities,
+      predictionDurationInS: (performance.now() - createdAt) / 1000,
+    };
+  } catch (err) {
+    if (isAPIError(err)) throw err;
+    if (err instanceof FetchError) {
+      if (err.statusCode) {
+        throw new APIStatusError({
+          message: `error during interruption prediction: ${err.message}`,
+          options: { statusCode: err.statusCode, body: err.data },
+        });
+      }
+      if (
+        err.cause instanceof Error &&
+        (err.cause.name === 'TimeoutError' || err.cause.name === 'AbortError')
+      ) {
+        throw new APIStatusError({
+          message: `interruption inference timeout: ${err.message}`,
+          options: { statusCode: 408, retryable: false },
+        });
+      }
+      throw new APIConnectionError({
+        message: `interruption inference connection error: ${err.message}`,
+      });
+    }
+    throw new APIError(`error during interruption prediction: ${err}`);
+  }
+}
+
+export interface HttpTransportOptions {
+  baseUrl: string;
+  apiKey: string;
+  apiSecret: string;
+  threshold: number;
+  minFrames: number;
+  timeout: number;
+  maxRetries?: number;
+}
+
+export interface HttpTransportState {
+  overlapSpeechStarted: boolean;
+  overlapSpeechStartedAt: number | undefined;
+  cache: BoundedCache<number, InterruptionCacheEntry>;
+}
+
+/**
+ * Creates an HTTP transport TransformStream for interruption detection.
+ *
+ * This transport receives Int16Array audio slices and outputs InterruptionEvents.
+ * Each audio slice triggers an HTTP POST request.
+ *
+ * @param options - Transport options object. This is read on each request, so mutations
+ *                  to threshold/minFrames will be picked up dynamically.
+ */
+export function createHttpTransport(
+  options: HttpTransportOptions,
+  getState: () => HttpTransportState,
+  setState: (partial: Partial<HttpTransportState>) => void,
+  updateUserSpeakingSpan?: (entry: InterruptionCacheEntry) => void,
+  getAndResetNumRequests?: () => number,
+): TransformStream<Int16Array | OverlappingSpeechEvent, OverlappingSpeechEvent> {
+  const logger = log();
+
+  return new TransformStream<Int16Array | OverlappingSpeechEvent, OverlappingSpeechEvent>(
+    {
+      async transform(chunk, controller) {
+        if (!(chunk instanceof Int16Array)) {
+          controller.enqueue(chunk);
+          return;
+        }
+
+        const state = getState();
+        const overlapSpeechStartedAt = state.overlapSpeechStartedAt;
+        if (overlapSpeechStartedAt === undefined || !state.overlapSpeechStarted) return;
+
+        try {
+          const resp = await predictHTTP(
+            chunk,
+            { threshold: options.threshold, minFrames: options.minFrames },
+            {
+              baseUrl: options.baseUrl,
+              timeout: options.timeout,
+              maxRetries: options.maxRetries,
+              token: await createAccessToken(options.apiKey, options.apiSecret),
+            },
+          );
+
+          const { createdAt, isBargein, probabilities, predictionDurationInS } = resp;
+          const entry = state.cache.setOrUpdate(
+            createdAt,
+            () => new InterruptionCacheEntry({ createdAt }),
+            {
+              probabilities,
+              isInterruption: isBargein,
+              speechInput: chunk,
+              totalDurationInS: (performance.now() - createdAt) / 1000,
+              detectionDelayInS: (Date.now() - overlapSpeechStartedAt) / 1000,
+              predictionDurationInS,
+            },
+          );
+
+          if (state.overlapSpeechStarted && entry.isInterruption) {
+            if (updateUserSpeakingSpan) {
+              updateUserSpeakingSpan(entry);
+            }
+            const event: OverlappingSpeechEvent = {
+              type: 'overlapping_speech',
+              detectedAt: Date.now(),
+              overlapStartedAt: overlapSpeechStartedAt,
+              isInterruption: entry.isInterruption,
+              speechInput: entry.speechInput,
+              probabilities: entry.probabilities,
+              totalDurationInS: entry.totalDurationInS,
+              predictionDurationInS: entry.predictionDurationInS,
+              detectionDelayInS: entry.detectionDelayInS,
+              probability: entry.probability,
+              numRequests: getAndResetNumRequests?.() ?? 0,
+            };
+            logger.debug(
+              {
+                detectionDelayInS: entry.detectionDelayInS,
+                totalDurationInS: entry.totalDurationInS,
+              },
+              'interruption detected',
+            );
+            setState({ overlapSpeechStarted: false });
+            controller.enqueue(event);
+          }
+        } catch (err) {
+          controller.error(err);
+        }
+      },
+    },
+    { highWaterMark: 2 },
+    { highWaterMark: 2 },
+  );
+}