# Interruptions & cancellation > How the bot stops when the user speaks, and how @pipedog/asyncio gives JavaScript Python's asyncio cancellation. Source: https://polvallverdu.github.io/pipedog/docs/concepts/interruptions-and-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 interruption [#what-happens-on-an-interruption] When a user turn starts with interruptions enabled, the user aggregator **broadcasts an `InterruptionFrame`** upstream and downstream. Then: **Every processor cancels its current work.** `InterruptionFrame` is a system frame, so each processor handles it immediately instead of waiting behind queued frames. The processor cancels the task processing its current data/control frame and discards queued ones. Frames with `interruptible = false` (like `FunctionCallResultFrame` and `EndFrame`) are kept and still processed. **The LLM stops generating**: the in-flight completion is cancelled mid-stream. **TTS clears its buffers** and drops pending output. **The transport flushes unplayed audio**, so the bot goes silent within about one audio write. Only the words actually spoken end up in the LLM context: the output transport pushes `TTSTextFrame`s in sync with playback, and the assistant aggregator commits the spoken-so-far text when interrupted. ### Triggering an interruption yourself [#triggering-an-interruption-yourself] From inside a custom processor: ```ts await this.broadcastInterruption(); ``` From code holding the worker, queue an `InterruptionWorkerFrame`; the worker turns it into an `InterruptionFrame` sent through the whole pipeline: ```ts 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 transcript [#turns-with-no-transcript] 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](https://polvallverdu.github.io/pipedog/docs/concepts/turn-strategies/#turns-with-no-transcript)). ## Controlling interruptions [#controlling-interruptions] Interruptions are on by default. What starts one is the [start strategy](https://polvallverdu.github.io/pipedog/docs/concepts/turn-strategies.md) that opens the user's turn, and [mute strategies](https://polvallverdu.github.io/pipedog/docs/concepts/user-muting.md) decide whether the user is heard at all: | Goal | How | | -------------------------------------------------- | ---------------------------------------------------------------------------------------------------------- | | Stop the bot as soon as the user speaks | the default (`VADUserTurnStartStrategy`) | | Never interrupt, but answer after the bot finishes | `enableInterruptions: false` on the start strategies | | Ignore short acknowledgements like "okay" | `MinWordsUserTurnStartStrategy` | | Tell interruptions from backchannels with a model | `KrispVivaIPUserTurnStartStrategy` (bring your own Krisp SDK) | | Discard what the user says while the bot talks | `AlwaysUserMuteStrategy` | | Protect only the greeting, or tool calls | `MuteUntilFirstBotCompleteUserMuteStrategy`, `FirstSpeechUserMuteStrategy`, `FunctionCallUserMuteStrategy` | | Keep a provider's turn detection from interrupting | `shouldInterrupt: false` on the STT service | To turn interruptions off, set `enableInterruptions: false` on every start strategy, since any of them can start a turn: ```ts 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 `InterruptionFrame`s 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 help [#why-javascript-needs-help] 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: | Python | `@pipedog/asyncio` | | -------------------------------------------------------------- | ----------------------------------------------------- | | `asyncio.create_task(coro())` | `createTask(() => coro())` | | `task.cancel()` / `await task` | `task.cancel()` / `await task` (a `Task` is thenable) | | `asyncio.sleep(s)` | `sleep(s)` (seconds) | | `asyncio.wait_for(coro(), timeout=t)` | `waitFor(() => coro(), t)` | | `asyncio.gather(a(), b())` | `gather(() => a(), () => b())` | | `asyncio.Event`, `Queue`, `PriorityQueue`, `Lock`, `Semaphore` | same names | | `asyncio.CancelledError`, `asyncio.TimeoutError` | `CancelledError`, `TimeoutError` | | `asyncio.current_task()` | `currentTask()` | | `async with lock:` | `await lock.run(async () => { ... })` | ```ts 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 points [#cancellation-points] 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. ```ts 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: ```ts const queue = new Queue(); 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(); ``` ## rethrowIfCancelled [#rethrowifcancelled] 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)`**: ```ts async function callProvider(url: string): Promise { 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, SDKs [#foreign-awaits-fetch-websockets-sdks] `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. ```ts async function withSdk(sdkCall: () => Promise): Promise { // 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 processors [#tasks-in-processors] 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. ```ts twoslash 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 | null = null; override async processFrame(frame: Frame, direction: FrameDirection): Promise { 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 { 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 { 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 thumb [#rules-of-thumb] 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](https://polvallverdu.github.io/pipedog/docs/api/asyncio.md) lists every export.