feat(eot): allow disabling local fallback

Author: rosetta-livekit-bot[bot]

.changeset/tiny-clouds-wait.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': minor
+---
+
+Allow cloud turn detectors to disable fallback to the local model.

agents/src/inference/eot/base.ts

@@ -373,6 +373,12 @@ export class BaseStreamingTurnDetectorStream {
     this._audioInputNumChannels = undefined;
   }
 
+  /** Stop accepting audio and discard anything buffered for a transport that
+   * can no longer consume it. */
+  protected _discardAudioInput(): void {
+    void this._audioChannel.abort(new Error('turn detection disabled')).catch(() => {});
+  }
+
   // endregion
 
   // region: results

agents/src/inference/eot/detector.test.ts

@@ -371,7 +371,7 @@ describe('DetectorViewAfterFallback', () => {
     });
     // server defaults arrive, then the cloud session fails
     stream.thresholdsOptions._updateDefaults({ ...SERVER_THRESHOLDS }, SERVER_DEFAULT_THRESHOLD);
-    stream._fallBackToLocal(new APIConnectionError({ message: 'boom' }));
+    stream._fallbackToLocal(new APIConnectionError({ message: 'boom' }));
     await waitFor(() => stream.model === 'turn-detector-v1-mini');
 
     // Both the stream and the detector (sharing one ThresholdOptions) reflect it.
@@ -408,7 +408,7 @@ describe('WarningDedupe', () => {
     const stream = makeStreamWithTransport(transport);
     await waitFor(() => stream.model === 'turn-detector-v1-mini');
     // Trigger a second fallback path directly.
-    stream._fallBackToLocal(new APIConnectionError({ message: 'boom2' }));
+    stream._fallbackToLocal(new APIConnectionError({ message: 'boom2' }));
     // Across both invocations only one warning was emitted — tracked by
     // the `warnedCloudFailure` flag staying flipped after the first call.
     expect(stream.warnedCloudFailure).toBe(true);
@@ -702,7 +702,7 @@ describe('ThresholdRescaleOnFallback', () => {
     const transport = new ScriptedTransport({ runBehavior: 'idle' });
     const stream = makeStreamWithTransport(transport, { userThreshold: 0.5 });
     stream.thresholdsOptions._updateDefaults({ ...SERVER_THRESHOLDS }, SERVER_DEFAULT_THRESHOLD);
-    stream._fallBackToLocal(new APIConnectionError({ message: 'boom' }));
+    stream._fallbackToLocal(new APIConnectionError({ message: 'boom' }));
     await waitFor(() => stream.model === 'turn-detector-v1-mini');
     expect(stream.isFallback).toBe(true);
     expect(await stream.unlikelyThreshold('en')).toBeCloseTo(
@@ -715,7 +715,7 @@ describe('ThresholdRescaleOnFallback', () => {
     const transport = new ScriptedTransport({ runBehavior: 'idle' });
     const stream = makeStreamWithTransport(transport);
     stream.thresholdsOptions._updateDefaults({ ...SERVER_THRESHOLDS }, SERVER_DEFAULT_THRESHOLD);
-    stream._fallBackToLocal(new APIConnectionError({ message: 'boom' }));
+    stream._fallbackToLocal(new APIConnectionError({ message: 'boom' }));
     await waitFor(() => stream.model === 'turn-detector-v1-mini');
     // ratio 1.0 → local table unchanged
     expect(await stream.unlikelyThreshold('en')).toBeCloseTo(LOCAL_LANGUAGES.en!);
@@ -726,7 +726,7 @@ describe('ThresholdRescaleOnFallback', () => {
     const transport = new ScriptedTransport({ runBehavior: 'idle' });
     const stream = makeStreamWithTransport(transport, { userThreshold: { en: 0.55, ja: 0.25 } });
     stream.thresholdsOptions._updateDefaults({ ...SERVER_THRESHOLDS }, SERVER_DEFAULT_THRESHOLD);
-    stream._fallBackToLocal(new APIConnectionError({ message: 'boom' }));
+    stream._fallbackToLocal(new APIConnectionError({ message: 'boom' }));
     await waitFor(() => stream.model === 'turn-detector-v1-mini');
     expect(stream.isFallback).toBe(true);
     expect(await stream.unlikelyThreshold('en')).toBeCloseTo(

agents/src/inference/eot/detector.ts

@@ -44,6 +44,13 @@ export interface TurnDetectorOptions {
   apiSecret?: string;
   /** Sample rate (Hz). Defaults to 16000. */
   sampleRate?: number;
+  /**
+   * Whether a `v1` detector may degrade to the local `v1-mini` model when the
+   * gateway fails. `false` keeps it cloud-only, so the mini weights (~138 MB,
+   * resident for the process lifetime) are never loaded and turns commit on
+   * the endpointing delay instead.
+   */
+  localFallback?: boolean;
   connOptions?: APIConnectOptions;
   /**
    * Inference executor that runs the local `turn-detector-v1-mini` model in the
@@ -59,6 +66,7 @@ export class TurnDetector extends BaseStreamingTurnDetector {
   protected _model: TurnDetectorModel;
   protected _cloudOpts: CloudTransportOptions | undefined;
   protected _executor: InferenceExecutor | undefined;
+  protected _localFallback: boolean;
 
   constructor(opts: TurnDetectorOptions = {}) {
     // auto = caller didn't pin a version; missing cloud creds warn-and-
@@ -118,6 +126,13 @@ export class TurnDetector extends BaseStreamingTurnDetector {
     super(detectorOpts);
     this._model = resolvedModel;
     this._cloudOpts = cloudOpts;
+    this._localFallback = opts.localFallback ?? true;
+    if (!this._localFallback && resolvedModel === 'turn-detector-v1-mini') {
+      log().warn(
+        { model: resolvedModel },
+        'localFallback=false has no effect on the local model, which runs locally by design',
+      );
+    }
     this._warnThresholdOverride();
     // Default to the current job's shared inference executor. `getJobContext`
     // throws outside a job (tests, standalone) — degrade to `undefined`
@@ -208,6 +223,7 @@ export class TurnDetector extends BaseStreamingTurnDetector {
       opts: this._opts,
       cloudOpts,
       model: this._model,
+      localFallback: this._localFallback,
       executor: this._executor,
     });
     this._streams.add(stream);
@@ -220,6 +236,7 @@ export interface TurnDetectorStreamImplArgs {
   opts: BaseStreamingTurnDetectorOptions;
   cloudOpts: CloudTransportOptions | undefined;
   model: TurnDetectorModel;
+  localFallback?: boolean;
   /** Shared inference executor for the `turn-detector-v1-mini` (local) model
    * (undefined degrades to a positive-default prediction). */
   executor?: InferenceExecutor;
@@ -239,7 +256,9 @@ export class TurnDetectorStreamImpl extends BaseStreamingTurnDetectorStream {
   protected _model: TurnDetectorModel;
   protected _cloudOpts: CloudTransportOptions | undefined;
   protected _executor: InferenceExecutor | undefined;
+  protected _localFallback: boolean;
   protected _isFallback = false;
+  protected _isDegraded = false;
   protected _warnedCloudFailure = false;
   protected _warnedLocalFailure = false;
   private _detLogger = log();
@@ -258,6 +277,7 @@ export class TurnDetectorStreamImpl extends BaseStreamingTurnDetectorStream {
     this._model = args.model;
     this._cloudOpts = args.cloudOpts;
     this._executor = args.executor;
+    this._localFallback = args.localFallback ?? true;
   }
 
   /** This stream's *current* model name (flips to `'turn-detector-v1-mini'`
@@ -271,6 +291,11 @@ export class TurnDetectorStreamImpl extends BaseStreamingTurnDetectorStream {
     return this._isFallback;
   }
 
+  /** True once the cloud transport is gone with no local model to replace it. */
+  get isDegraded(): boolean {
+    return this._isDegraded;
+  }
+
   /** @internal Test-visible. */
   get warnedCloudFailure(): boolean {
     return this._warnedCloudFailure;
@@ -287,7 +312,12 @@ export class TurnDetectorStreamImpl extends BaseStreamingTurnDetectorStream {
   /** @internal Test-visible: same logic as the path taken when `_run` catches
    * a cloud transport error. Tests call this directly to verify the warning
    * dedupe across multiple invocations on the same stream. */
-  _fallBackToLocal(reason: Error): void {
+  _fallbackToLocal(reason: Error): boolean {
+    if (!this._localFallback) {
+      this._emitDefaultForInflight();
+      return false;
+    }
+
     if (!this._warnedCloudFailure) {
       this._detLogger.warn(
         { reason: reason.message },
@@ -313,6 +343,22 @@ export class TurnDetectorStreamImpl extends BaseStreamingTurnDetectorStream {
     this._transport.attach(this);
     this._model = 'turn-detector-v1-mini';
     this._isFallback = true;
+    return true;
+  }
+
+  protected _degrade(reason: Error): void {
+    this._detLogger.warn(
+      { reason: reason.message },
+      'cloud turn detector failed; local fallback is disabled, so turn detection is off for ' +
+        'the rest of this stream and turns commit on the endpointing delay',
+    );
+    this._isDegraded = true;
+    try {
+      this._transport.detach();
+    } catch {
+      // ignore detach errors while terminating the failed transport
+    }
+    this._discardAudioInput();
   }
 
   /** @internal Test-visible: same logic as the path taken when `_run` sees a
@@ -366,8 +412,11 @@ export class TurnDetectorStreamImpl extends BaseStreamingTurnDetectorStream {
         }
         const e = err instanceof Error ? err : new Error(String(err));
         if (this._model === 'turn-detector-v1') {
-          this._fallBackToLocal(e);
-          continue;
+          if (this._fallbackToLocal(e)) {
+            continue;
+          }
+          this._degrade(e);
+          return;
         }
         this._onLocalFailure(e);
         return;
@@ -377,13 +426,9 @@ export class TurnDetectorStreamImpl extends BaseStreamingTurnDetectorStream {
 
   protected override _onPredictTimeout(): void {
     if (this._model === 'turn-detector-v1') {
-      // Signal the swap BEFORE mutating model/transport state. The
-      // race in `_raceWithSwap` is rejected with `SwapAbortError`
-      // immediately, so the main loop exits through the
-      // SwapAbortError branch and never consults `_model` for a
-      // classification that would race with the assignment below.
-      this._signalSwap();
-      this._fallBackToLocal(new Error('predict_end_of_turn'));
+      if (this._fallbackToLocal(new Error('predict_end_of_turn'))) {
+        this._signalSwap();
+      }
     }
   }
 }