# FrameProcessor

> FrameProcessor options, FrameProcessorSetup and the FrameProcessor class, from @pipedog/core/processors/frame_processor.

Source: https://polvallverdu.github.io/pipedog/docs/api/frame-processor/



```ts
import {
  FrameDirection,
  FrameProcessor,
  type FrameProcessorOptions,
  FrameProcessorSetup,
} from "@pipedog/core/processors/frame_processor";
```

See [Frame processors](https://polvallverdu.github.io/pipedog/docs/concepts/frame-processors.md) for a guide.

## FrameProcessorOptions [#frameprocessoroptions]

The options object every processor constructor accepts (and passes to `super(options)`).

| Option | Type | Required | Description |
|---|---|---|---|
| `enableDirectMode` | `boolean \| undefined` | no | Process frames immediately instead of using internal queues. |
| `metrics` | `FrameProcessorMetrics \| null \| undefined` | no | Optional metrics collector for this processor. |
| `name` | `string \| null \| undefined` | no | Custom name; defaults to `ClassName#count`. |
| `taskManager` | `BaseTaskManager \| null \| undefined` | no | Task manager for handling tasks. |

It extends `BaseObjectOptions` from `@pipedog/core/utils/base_object`:

| Option | Type | Required | Description |
|---|---|---|---|
| `name` | `string \| null \| undefined` | no | Custom name; defaults to `ClassName#count`. |
| `taskManager` | `BaseTaskManager \| null \| undefined` | no | Task manager for handling tasks. |

## FrameDirection [#framedirection]

```ts
export const FrameDirection = { DOWNSTREAM: 1, UPSTREAM: 2 } as const;
export type FrameDirection = (typeof FrameDirection)[keyof typeof FrameDirection];
```

## FrameProcessorSetup [#frameprocessorsetup]

Passed to `setup()` by the pipeline before any frame flows. Built from these options:

| Option | Type | Required | Description |
|---|---|---|---|
| `clock` | `BaseClock` | yes |  |
| `taskManager` | `BaseTaskManager` | yes |  |
| `pipelineWorker` | `PipelineWorker \| null` | yes |  |
| `audioInSampleRate` | `number \| undefined` | no |  |
| `audioOutSampleRate` | `number \| undefined` | no |  |
| `enableExpressiveConversation` | `boolean \| undefined` | no |  |
| `enableMetrics` | `boolean \| undefined` | no |  |
| `enableTracing` | `boolean \| undefined` | no |  |
| `enableUsageMetrics` | `boolean \| undefined` | no |  |
| `expressionPolicy` | `ExpressionPolicy \| null \| undefined` | no |  |
| `observer` | `BaseObserver \| null \| undefined` | no |  |
| `reportOnlyInitialTtfb` | `boolean \| undefined` | no |  |
| `tracingContext` | `TracingContext \| null \| undefined` | no |  |

## FrameProcessor [#frameprocessor]

The public members of the class (including those inherited from `BaseObject`):

| Option | Type | Required | Description |
|---|---|---|---|
| `_prev` | `FrameProcessor \| null` | yes |  |
| `_next` | `FrameProcessor \| null` | yes |  |
| `_enableDirectMode` | `boolean` | yes |  |
| `_setup` | `FrameProcessorSetup \| null` | yes |  |
| `_cancelling` | `boolean` | yes |  |
| `_metrics` | `FrameProcessorMetrics` | yes |  |
| `_isUsable` | `boolean` | yes |  |
| `isUsable` | `boolean` | yes | Whether this processor can still do its job. A processor stays usable through failures it might recover from, and becomes unusable once its work can no longer succeed: a provider has rejected its API key, model or voice, or it has failed enough times to stop trying. Services stop accepting work and stop reconnecting once this is false. Errors set this as they are reported, so an error handler reading `frame.processor.isUsable` sees the verdict that came with the error. |
| `setUsable` | `(isUsable: boolean) => Promise<void>` | yes | Set whether this processor can be given work. Call this to bring back a processor that became unusable, once whatever stopped it working has been dealt with. |
| `processors` | `FrameProcessor[]` | yes | The sub-processors contained within this processor. Only compound processors (e.g. pipelines and parallel pipelines) have sub-processors. |
| `entryProcessors` | `FrameProcessor[]` | yes | The entry processors for this processor: the first processors in a compound processor. Non-compound processors return an empty list. |
| `next` | `FrameProcessor \| null` | yes | The next processor, or null. |
| `previous` | `FrameProcessor \| null` | yes | The previous processor, or null. |
| `processorSetup` | `FrameProcessorSetup` | yes | The configuration this processor was set up with. |
| `metricsEnabled` | `boolean` | yes | Whether metrics collection is enabled. |
| `usageMetricsEnabled` | `boolean` | yes | Whether usage metrics collection is enabled. |
| `reportOnlyInitialTtfb` | `boolean` | yes | Whether only the initial TTFB should be reported. |
| `pipelineWorker` | `PipelineWorker` | yes | The PipelineWorker this processor is running in. Gives access to worker-scoped state, e.g. `this.pipelineWorker.appResources`. |
| `workerRunner` | `WorkerRunner` | yes | The WorkerRunner hosting this processor's worker. Use it to reach another worker by name, e.g. `this.workerRunner.getWorker("ui-jobs")`. |
| `processorsWithMetrics` | `() => FrameProcessor[]` | yes | Processors that can generate metrics, including nested ones. |
| `canGenerateMetrics` | `() => boolean` | yes | Whether this processor can generate metrics. |
| `setCoreMetricsData` | `(data: MetricsData) => void` | yes | Set core metrics data for this processor. |
| `startTtfbMetrics` | `(options?: { startTime?: number \| null; }) => Promise<void>` | yes | Start time-to-first-byte metrics collection. |
| `cancelTtfbMetrics` | `() => Promise<void>` | yes | Abandon the current time-to-first-byte measurement without reporting it. |
| `stopTtfbMetrics` | `(options?: { endTime?: number \| null; }) => Promise<void>` | yes | Stop time-to-first-byte metrics collection and push results. |
| `processTtfaMetrics` | `(frame: TTSAudioRawFrame) => Promise<void>` | yes | Scan a TTS audio frame for the first audible sample and push TTFA. Call for every audio frame until a measurement is produced. |
| `stopTtfatMetrics` | `(options?: { endTime?: number \| null; }) => Promise<void>` | yes | Stop time-to-first-answer-token metrics collection and push results. |
| `startProcessingMetrics` | `(options?: { startTime?: number \| null; }) => Promise<void>` | yes | Start processing metrics collection. |
| `stopProcessingMetrics` | `(options?: { endTime?: number \| null; }) => Promise<void>` | yes | Stop processing metrics collection and push results. |
| `startLlmUsageMetrics` | `(tokens: LLMTokenUsage) => Promise<void>` | yes | Start LLM usage metrics collection. |
| `startSttUsageMetrics` | `(usage: STTUsage) => Promise<void>` | yes | Start STT usage metrics collection. |
| `startTtsUsageMetrics` | `(text: string) => Promise<void>` | yes | Start TTS usage metrics collection. |
| `startTextAggregationMetrics` | `() => Promise<void>` | yes | Start text aggregation time metrics collection. |
| `stopTextAggregationMetrics` | `() => Promise<void>` | yes | Stop text aggregation time metrics collection and push results. |
| `stopAllMetrics` | `() => Promise<void>` | yes | Stop all active metrics collection. |
| `setup` | `(setup: FrameProcessorSetup) => Promise<void>` | yes | Set up the processor with required components. This is where a processor connects and does its other slow start-up work. A pipeline sets its processors up concurrently, so a resource shared with another processor needs guarding (see `utils/shared`). |
| `_setupStarted` | `() => void` | yes | Hold the StartFrame until `_setupFinished` is called. Called by the pipeline before setup. |
| `_setupFinished` | `() => void` | yes | Let this processor start, once its own setup() has returned. |
| `cleanup` | `() => Promise<void>` | yes | Release this processor's resources at teardown. This base implementation cancels only the processor's internal input/process tasks; tasks created via `createTask` are released by an override. Runs concurrently with every other processor's cleanup. |
| `link` | `(processor: FrameProcessor) => void` | yes | Link this processor to the next processor in the pipeline. |
| `getClock` | `() => BaseClock` | yes | The clock used by this processor. |
| `getEventLoop` | `() => EventLoop` | yes | The event loop used by this processor. |
| `queueFrame` | `(frame: Frame, direction?: FrameDirection, callback?: FrameCallback \| null) => Promise<void>` | yes | Queue a frame for processing. |
| `pauseProcessingFrames` | `() => Promise<void>` | yes | Pause processing of queued frames. |
| `pauseProcessingSystemFrames` | `() => Promise<void>` | yes | Pause processing of queued system frames. |
| `resumeProcessingFrames` | `() => Promise<void>` | yes | Resume processing of queued frames. |
| `resumeProcessingSystemFrames` | `() => Promise<void>` | yes | Resume processing of queued system frames. |
| `pauseProcessingAllFramesUntil` | `(ready: () => Promise<unknown>, options?: { timeout?: number; }) => Promise<void>` | yes | Hold frames arriving at this processor until `ready` resolves. Useful for a processor that cannot act on frames until some condition holds, such as one that establishes a connection in the background. Frames wait in the processor's queues and are delivered in order once the condition resolves, so nothing is lost. The frame being processed when this is called is unaffected: the pause takes hold from the next frame onwards. Both queues are held, so the pause is always lifted: when `ready` resolves, when `timeout` (seconds) elapses, or at teardown. |
| `processFrame` | `(frame: Frame, direction: FrameDirection) => Promise<void>` | yes | Process a frame. Subclasses override this and call `await super.processFrame(frame, direction)` first. |
| `_classifyError` | `(_exception: unknown) => ErrorCategory \| null` | yes | Classify an exception this processor knows the shape of. Override for providers that signal failures through SDK-specific errors rather than an HTTP status code. Return null to fall back to `classifyHttpException`. |
| `pushError` | `(errorMsg: string, options?: { exception?: unknown; category?: ErrorCategory \| null; forceTreatAsPermanent?: boolean; }) => Promise<void>` | yes | Create and push an ErrorFrame upstream, with context about which processor generated the error. |
| `pushErrorFrame` | `(error: ErrorFrame, options?: { forceTreatAsPermanent?: boolean; }) => Promise<void>` | yes | Push an error frame upstream. |
| `pushFrame` | `(frame: Frame, direction?: FrameDirection) => Promise<void>` | yes | Push a frame to the next (downstream) or previous (upstream) processor. |
| `broadcastInterruption` | `() => Promise<void>` | yes | Broadcast an InterruptionFrame both upstream and downstream. |
| `broadcastFrame` | `<O>(frameCls: new (options: O) => Frame, options?: O) => Promise<void>` | yes | Broadcast a frame of the specified class upstream and downstream. Creates two instances of the given frame class from the provided options (without deep-copying them) and pushes them upstream and downstream. |
| `broadcastFrameInstance` | `(frame: Frame) => Promise<void>` | yes | Broadcast a frame instance upstream and downstream. Creates two new frames shallow-copying all fields from the original frame except `id` and `name`, which get fresh values. Prefer `broadcastFrame()` when you are the creator of the frame. |
| `_startInterruption` | `() => Promise<void>` | yes | Start handling an interruption by cancelling current tasks. |
| `_cancelFrameInProgress` | `(frame: Frame) => Promise<boolean>` | yes | Stop processing a frame if it is the one in progress, leaving the queue as it is. Unlike `_startInterruption`, this drops nothing else: the frames queued behind the cancelled one, interruptible or not, are processed next. The cancelled frame's callback, if any, is not called. Call it from a system frame's processing, as interruptions are, so it runs on the input task and never overlaps an interruption. |
| `hasQueuedFrame` | `(frameType: FrameClass) => boolean` | yes | Whether a frame of the given type is waiting in the processing queue. |
| `_id` | `number` | yes |  |
| `_name` | `string` | yes |  |
| `_taskManager` | `BaseTaskManager \| null` | yes |  |
| `_eventHandlers` | `Map<string, EventHandler>` | yes |  |
| `_eventTasks` | `Set<[string, Task<unknown>]>` | yes |  |
| `id` | `number` | yes | The unique identifier for this object. |
| `name` | `string` | yes | The object's name, either custom-provided or auto-generated. |
| `taskManager` | `BaseTaskManager` | yes | The task manager. Throws if not initialized. |
| `createTask` | `<T>(coroutine: TaskFn<T>, name?: string \| null) => Task<T \| undefined>` | yes | Create a new task managed by this object's task manager. |
| `cancelTask` | `(task: Task<unknown>, timeout?: number \| null) => Promise<void>` | yes | Cancel a task managed by this object's task manager. A default timeout of 1 second avoids freezes caused by code that swallows cancellation. |
| `eventHandler` | `<F extends EventHandlerFn<this>>(eventName: string, handler: F) => F` | yes | Register a handler for an event. Equivalent of Python's `@obj.event_handler("on_x")` decorator; returns the handler. |
| `addEventHandler` | `(eventName: string, handler: EventHandlerFn<FrameProcessor>) => void` | yes | Add an event handler for the specified event. Can be sync or async. |
| `removeEventHandler` | `(eventName: string, handler: EventHandlerFn<FrameProcessor>) => void` | yes | Remove a previously added event handler. No-op if unknown. |
| `_registerEventHandler` | `(eventName: string, sync?: boolean) => void` | yes | Register an event handler type. |
| `_hasEventHandlers` | `(eventName: string) => boolean` | yes | Whether any handler is registered for `eventName`. Not in upstream: code that fires an event for every frame checks it first, so an event nobody listens to costs no await. |
| `_callEventHandler` | `(eventName: string, ...args: any[]) => Promise<void>` | yes |  |
| `_runHandler` | `(eventName: string, handler: EventHandlerFn, ...args: any[]) => Promise<void>` | yes |  |
| `toString` | `() => string` | yes |  |