Interruptions & cancellation

Interruptions (barge-in) let the user talk over the bot: the bot stops right away, in-flight work is cancelled, and the pipeline is ready for the new input. Making that work in JavaScript needs Python's asyncio cancellation semantics, which is what @pipedog/asyncio provides.

What happens on an interruptionCopy Anchor Link

When a user turn starts with interruptions enabled, the user aggregator broadcasts an InterruptionFrame upstream and downstream. Then:

Only the words actually spoken end up in the LLM context: the output transport pushes TTSTextFrames in sync with playback, and the assistant aggregator commits the spoken-so-far text when interrupted.

Triggering an interruption yourselfCopy Anchor Link

From inside a custom processor:

await this.broadcastInterruption();

From code holding the worker, queue an InterruptionWorkerFrame; the worker turns it into an InterruptionFrame sent through the whole pipeline:

await worker.queueFrame(new InterruptionWorkerFrame());

A processor sees the interruption in processFrame() (after super.processFrame() has already cancelled the work in progress), and can override protected async _startInterruption() (calling super._startInterruption() first) for cleanup that must happen at that moment.

Turns with no transcriptCopy Anchor Link

If the user interrupts but nothing is transcribed (a cough, say), the bot has stopped and has nothing to answer. By default the LLM then runs once, so the bot asks the user to repeat or picks up where it left off; emptyUserTurn on LLMUserAggregatorParams changes that (see Turns with no transcript).

Controlling interruptionsCopy Anchor Link

Interruptions are on by default. What starts one is the start strategy that opens the user's turn, and mute strategies decide whether the user is heard at all:

To turn interruptions off, set enableInterruptions: false on every start strategy, since any of them can start a turn:

const userParams = new LLMUserAggregatorParams({
  vadAnalyzer: new SileroVADAnalyzer(),
  userTurnStrategies: new UserTurnStrategies({
    start: [
      new VADUserTurnStartStrategy({ enableInterruptions: false }),
      new TranscriptionUserTurnStartStrategy({ enableInterruptions: false }),
    ],
  }),
});

Disabling interruptions is not muting. With interruptions off the user's turn still starts and is transcribed: the bot finishes speaking and then answers. A muted user's audio and transcripts are dropped at the user aggregator, so nothing they say over the bot reaches the context or gets an answer. The user aggregator also drops InterruptionFrames while the user is muted, so an interruption you queue on the worker doesn't get past it then; one broadcast by a processor after the aggregator still does.

With an STT service that detects turns itself, set shouldInterrupt: false on the service rather than on the strategies: the service hands the setting to the strategies it recommends. Speech-to-speech services detect interruptions on their side, and pipedog follows their speaking events and still flushes the local audio, so the bot goes quiet at once.

Why JavaScript needs helpCopy Anchor Link

Python delivers CancelledError into a coroutine, at the await it is suspended on. That is how an interruption stops a processor mid-work: its processing task is cancelled, and whatever it was awaiting raises.

JavaScript has no such thing. A promise cannot be interrupted from outside, and catch (e) catches everything, including what Python would consider a cancellation. @pipedog/asyncio emulates asyncio on top of promises and AsyncLocalStorage, which Bun, Node, Deno and Cloudflare Workers (nodejs_compat) all provide:

import {
  abortable,
  CancelledError,
  createTask,
  currentSignal,
  Event,
  gather,
  Queue,
  rethrowIfCancelled,
  sleep,
  TimeoutError,
  waitFor,
} from "@pipedog/asyncio";

const task = createTask(async () => {
  await sleep(10); // a cancellation point
  return "done";
}, "slow-task");

task.cancel(); // like asyncio: the pending sleep() rejects with CancelledError
try {
  await task;
} catch (e) {
  console.log(e instanceof CancelledError); // true
}

Cancellation pointsCopy Anchor Link

When a task is cancelled:

• if it is suspended on a cancellation point, that await rejects with CancelledError right away (the request is consumed, as in asyncio);
• otherwise the request stays pending, and the next cancellation point the task reaches throws it;
• task.signal, an AbortSignal, is aborted.

The cancellation points are the awaits on this package's primitives: sleep(), Event.wait(), Queue.get() / Queue.put(), waitFor(), gather(), wait(), abortable(), and awaiting a Task. Cancelling a task that awaits another task cancels the awaited one too, as in asyncio.

try {
  // Pass a function, so it runs as a child task that can be cancelled on timeout.
  await waitFor(() => sleep(5), 0.1);
} catch (e) {
  if (!(e instanceof TimeoutError)) throw e;
}

const [x, y] = await gather(
  async () => 1,
  async () => "two",
);

Always pass waitFor() a function rather than a promise, so it can run as a child task and be cancelled on timeout. Queues and events behave like their asyncio counterparts:

const queue = new Queue<string>();
const ready = new Event();

const consumer = createTask(async () => {
  await ready.wait();
  for (;;) {
    const item = await queue.get(); // a cancellation point
    console.log("got", item);
    queue.taskDone();
  }
});

ready.set();
await queue.put("hello");
await queue.join();
consumer.cancel();

rethrowIfCancelledCopy Anchor Link

Python's except Exception: does not catch CancelledError (it derives from BaseException). JavaScript's catch (e) does. A ported except Exception: that swallowed a cancellation would keep a cancelled task running, so every broad catch starts with rethrowIfCancelled(e):

async function callProvider(url: string): Promise<unknown> {
  try {
    // fetch is not a cancellation point. Give it the current task's signal
    // so cancelling the task aborts the request too.
    const response = await fetch(url, { signal: currentSignal() });
    return await response.json();
  } catch (e) {
    rethrowIfCancelled(e); // Python's `except Exception` never catches CancelledError
    console.error("provider failed", e);
    return null;
  }
}

Catching the cancellation on purpose (Python's except asyncio.CancelledError:) looks like if (e instanceof CancelledError) { cleanup(); throw e; }.

Foreign awaits: fetch, WebSockets, SDKsCopy Anchor Link

fetch, a WebSocket, or a provider SDK is not a cancellation point: a task cancelled while awaiting one keeps waiting until it settles. Two tools fix that:

• Pass currentSignal() (the current task's AbortSignal, or a never-aborting one outside tasks) to anything that accepts a signal, such as fetch(url, { signal: currentSignal() }), as in the sample above. Cancelling the task then aborts the request itself.
• Wrap the await in abortable(promise) to make it a cancellation point. The foreign operation keeps running in the background unless it was also given the signal.

async function withSdk(sdkCall: () => Promise<string>): Promise<string> {
  // An SDK that takes no AbortSignal: turn the await into a cancellation point.
  // The SDK call itself keeps running in the background if we are cancelled.
  return await abortable(sdkCall());
}

Tasks in processorsCopy Anchor Link

Processors (and every BaseObject) create tasks through their task manager with this.createTask(fn, name), and cancel them with await this.cancelTask(task) (which waits up to 1 second for the task to finish). Tasks created this way are tracked and cancelled when the pipeline cleans up, so nothing is left floating. Never fire and forget a bare promise.

import { rethrowIfCancelled, sleep, type Task } from "@pipedog/asyncio";
import { type Frame, InterruptionFrame, StartFrame, TextFrame } from "@pipedog/core/frames/frames";
import { type FrameDirection, FrameProcessor } from "@pipedog/core/processors/frame_processor";

/** Says "tick" every second until the pipeline ends. */
export class Ticker extends FrameProcessor {
  #tickTask: Task<unknown> | null = null;

  override async processFrame(frame: Frame, direction: FrameDirection): Promise<void> {
    await super.processFrame(frame, direction);

    if (frame instanceof StartFrame) {
      // Tracked by the task manager: cancelled at cleanup, never left floating.
      this.#tickTask = this.createTask(() => this.#tick(), "tick");
    } else if (frame instanceof InterruptionFrame) {
      console.log("interrupted: stop what we were saying");
    }

    await this.pushFrame(frame, direction);
  }

  async #tick(): Promise<void> {
    for (;;) {
      await sleep(1.0); // seconds, never raw setTimeout
      try {
        await this.pushFrame(new TextFrame({ text: "tick" }));
      } catch (e) {
        rethrowIfCancelled(e);
        await this.pushError("tick failed", { exception: e });
      }
    }
  }

  override async cleanup(): Promise<void> {
    await super.cleanup();
    if (this.#tickTask) {
      await this.cancelTask(this.#tickTask);
      this.#tickTask = null;
    }
  }
}

await task waits for the task as asyncio does: if your task is cancelled while it waits, task is cancelled too. To react to a task's end without waiting for it, use task.addDoneCallback(cb), task.promise.then(cb) or task.then(cb). These don't tie the two tasks together, so cancelling yours leaves task running until it is cancelled itself. (In 0.10.0 and earlier, task.then(cb) could leave the calling task impossible to cancel and hang cleanup().)

Rules of thumbCopy Anchor Link

1. Every broad catch begins with rethrowIfCancelled(e).
2. Use sleep(seconds) from @pipedog/asyncio, never a raw setTimeout.
3. Pass currentSignal() to fetch and other APIs that take an AbortSignal; wrap the rest in abortable().
4. Never leave a promise floating: use this.createTask() in processors, createTask() elsewhere.
5. Never return a Task (or any thenable) from an async function unless you mean to await it.

The asyncio API reference lists every export.