# PipelineWorker & WorkerRunner

> PipelineParams, PipelineWorker options, WorkerRunner options and runTest options.

Source: https://polvallverdu.github.io/pipedog/docs/api/pipeline-worker/



```ts
import { Pipeline } from "@pipedog/core/pipeline/pipeline";
import { PipelineParams, PipelineWorker } from "@pipedog/core/pipeline/worker";
import { WorkerRunner } from "@pipedog/core/workers/runner";
```

See [Pipelines & workers](https://polvallverdu.github.io/pipedog/docs/concepts/pipelines-and-workers.md) for a guide.

## PipelineParams [#pipelineparams]

`new PipelineParams(options)`: pipeline-wide parameters, passed to every processor through
`FrameProcessorSetup`.

| Option | Type | Required | Description |
|---|---|---|---|
| `audioInSampleRate` | `number` | yes |  |
| `audioOutSampleRate` | `number` | yes |  |
| `enableExpressiveConversation` | `boolean` | yes |  |
| `expressionPolicy` | `ExpressionPolicy \| null` | yes |  |
| `enableHeartbeats` | `boolean` | yes |  |
| `enableMetrics` | `boolean` | yes |  |
| `enableUsageMetrics` | `boolean` | yes |  |
| `heartbeatsPeriodSecs` | `number` | yes |  |
| `heartbeatsMonitorSecs` | `number` | yes |  |
| `reportOnlyInitialTtfb` | `boolean` | yes |  |
| `sendInitialEmptyMetrics` | `boolean` | yes |  |
| `startMetadata` | `Record<string, unknown>` | yes |  |

## PipelineWorkerOptions [#pipelineworkeroptions]

`new PipelineWorker(pipeline, options)`.

| Option | Type | Required | Description |
|---|---|---|---|
| `active` | `boolean \| undefined` | no | Whether the worker starts active. Forwarded to `BaseWorker`. |
| `additionalSpanAttributes` | `Record<string, OtelAttributeValue> \| null \| undefined` | no | Optional attributes to propagate as OpenTelemetry conversation span attributes. |
| `appResources` | `unknown` | no | Optional application-defined bag of anything your application code may want to share across this session (DB handles, HTTP clients, etc.), passed by reference. Pipecat passes it through untouched and exposes it on the worker itself as `worker.appResources` and passes it to tool handlers as `FunctionCallParams.appResources`. The framework never copies or clears this object; the caller retains their handle and can read any mutations after the worker finishes. |
| `bridged` | `readonly string[] \| null \| undefined` | no | Bridge configuration. `null` means the pipeline is not bridged. An empty array `[]` wraps the pipeline with bus edge processors that accept frames from all bridges. An array of names like `["voice"]` accepts only frames from those bridges. The bus comes from `attach()` (called by the runner). |
| `cancelOnIdleTimeout` | `boolean \| undefined` | no | Whether reaching the idle timeout should cancel the pipeline worker. When false, the idle event still fires `on_idle_timeout` but the worker is left alone (and `cancelRunnerOnIdleTimeout` is ignored too: opting out of local cancellation also opts out of the runner-wide cancel). |
| `cancelRunnerOnIdleTimeout` | `boolean \| undefined` | no | When `cancelOnIdleTimeout` is also true, whether reaching the idle timeout should also cancel the entire `WorkerRunner`. The worker is always cancelled first; when this is true the worker also emits a `BusCancelMessage` so the runner broadcasts cancellation to every other root worker. Defaults to true so a multi-worker bot's helpers shut down with the main pipeline; set to false for a sidecar `PipelineWorker` that should self-cancel on idle without bringing down its peers. |
| `cancelTimeoutSecs` | `number \| undefined` | no | Timeout (in seconds) to wait for cancellation to happen cleanly. |
| `capabilities` | `BotCapabilities \| null \| undefined` | no | What the bot does, for fields the pipeline can't show. Its known fields replace the ones derived from the pipeline's transports and parameters. See `capabilities`. |
| `checkDanglingTasks` | `boolean \| undefined` | no | Whether to warn about tasks left running when the worker finishes. Only applies when the worker owns its task manager; otherwise the runner reports dangling tasks. |
| `clock` | `BaseClock \| null \| undefined` | no | Clock implementation for timing operations. |
| `conversationId` | `string \| null \| undefined` | no | Optional custom ID for the conversation. |
| `enableTracing` | `boolean \| undefined` | no | Whether to enable tracing. |
| `enableTurnTracking` | `boolean \| undefined` | no | Whether to enable turn tracking. |
| `handleFlushFrame` | `boolean \| null \| undefined` | no | Whether this worker answers a flush probe, bouncing it at the sink and completing it at the source. Defaults to whether the pipeline is unbridged, so a worker wired into someone else's topology leaves the probe to travel on and be completed by the pipeline that owns the transport. A bridged worker with no such peer never completes a flush and every `flushPipeline` call waits out its timeout. |
| `enableRtvi` | `boolean \| null \| undefined` | no | Whether to automatically add RTVI support to the pipeline. `null`, the default, adds it unless the pipeline is bridged: a bridged worker has no client of its own, and its RTVI would report every frame a second time as it crosses the bridge. |
| `excludeFrames` | `readonly FrameClass[] \| null \| undefined` | no | When `bridged` is set, extra frame types that should not cross the bus (lifecycle frames are always excluded). |
| `idleTimeoutFrames` | `readonly FrameClass[] \| undefined` | no | The frames that should trigger an idle timeout if not received within `idleTimeoutSecs`. The default pairs the VAD-only `UserSpeakingFrame` with the turn and transcription frames a provider-driven pipeline reports instead. |
| `idleTimeoutSecs` | `number \| null \| undefined` | no | Timeout (in seconds) to consider pipeline idle, or null. If a pipeline is idle the pipeline worker will be cancelled automatically. |
| `name` | `string \| null \| undefined` | no | Optional worker name (used for worker-style addressing on the bus). |
| `observers` | `BaseObserver[] \| null \| undefined` | no | Observers for monitoring pipeline execution. |
| `processorUnusablePolicy` | `ProcessorUnusablePolicy \| undefined` | no | What to do when a processor reports an error that leaves it unable to do its job, such as a service whose API key was rejected. Defaults to `ProcessorUnusablePolicy.CONTINUE`, leaving the decision to `on_pipeline_error` handlers. |
| `params` | `PipelineParams \| null \| undefined` | no | Configuration parameters for the pipeline. |
| `rtviProcessor` | `RTVIProcessor \| null \| undefined` | no | The RTVI processor to add if RTVI is enabled. |
| `rtviObserverParams` | `RTVIObserverParams \| null \| undefined` | no | The RTVI observer parameters to use if RTVI is enabled. |
| `setupTimeoutSecs` | `number \| undefined` | no | Timeout (in seconds) to wait for every processor to be set up. Processors connect while they are set up, so one that blocks connecting never lets the pipeline start, and the worker is torn down instead of waiting forever. The `StartFrame` is pushed while processors are still being set up, and each processor holds it only until its own setup is done, so processors ahead of a slow one start without it. |
| `startTimeoutSecs` | `number \| undefined` | no | Timeout (in seconds) to wait for the `StartFrame` to reach the end of the pipeline, counted from when every processor is set up. A processor that blocks while handling it never lets the pipeline start, so the worker is torn down instead of waiting forever. |
| `taskManager` | `BaseTaskManager \| null \| undefined` | no | Optional task manager for handling tasks. |

## PipelineOptions [#pipelineoptions]

`new Pipeline(processors, options)`.

| Option | Type | Required | Description |
|---|---|---|---|
| `source` | `FrameProcessor \| null \| undefined` | no | An optional pipeline source processor. |
| `sink` | `FrameProcessor \| null \| undefined` | no | An optional pipeline sink processor. |

## WorkerRunnerOptions [#workerrunneroptions]

`new WorkerRunner(options)`.

| Option | Type | Required | Description |
|---|---|---|---|
| `name` | `string \| null \| undefined` | no | Optional name for the runner instance. Defaults to a UUID-based name. Must be unique across runners in a distributed setup. |
| `bus` | `WorkerBus \| null \| undefined` | no | Optional `WorkerBus`. Defaults to a new in-process `AsyncQueueBus`. |
| `handleSigint` | `boolean \| undefined` | no | Whether to automatically handle SIGINT signals. |
| `handleSigterm` | `boolean \| undefined` | no | Whether to automatically handle SIGTERM signals. |
| `forceGc` | `boolean \| undefined` | no | Whether to force garbage collection once every worker has been torn down. |
| `checkDanglingTasks` | `boolean \| undefined` | no | Whether to warn about tasks left running on the shared task manager once every worker has finished. |
| `taskManager` | `BaseTaskManager \| null \| undefined` | no | Optional task manager for handling tasks. |

`runner.run(options)`:

| Option | Type | Required | Description |
|---|---|---|---|
| `autoEnd` | `boolean \| undefined` | no | When true (the default), the runner ends once every root worker has finished. When false, the runner blocks until `end()` or `cancel()` is called. |

## RunTestOptions [#runtestoptions]

`runTest(processor, options)` from `@pipedog/core/tests/utils`.

| Option | Type | Required | Description |
|---|---|---|---|
| `enableRtvi` | `boolean \| undefined` | no | Whether RTVI should be enabled in this test. |
| `expectedDownFrames` | `readonly FrameType[] \| null \| undefined` | no | Expected frame types flowing downstream (optional). |
| `expectedUpFrames` | `readonly FrameType[] \| null \| undefined` | no | Expected frame types flowing upstream (optional). |
| `framesToSend` | `readonly Frame[]` | yes | Sequence of frames to send through the processor. |
| `framesToSendDirection` | `FrameDirection \| undefined` | no | Direction to send `framesToSend`. Downstream frames are pushed from the beginning of the pipeline, upstream frames from the end. Defaults to DOWNSTREAM. |
| `ignoreStart` | `boolean \| undefined` | no | Whether to ignore StartFrames in frame validation. |
| `observers` | `BaseObserver[] \| null \| undefined` | no | Optional list of observers to attach to the pipeline. |
| `pipelineParams` | `PipelineParams \| null \| undefined` | no | Optional pipeline parameters. |
| `sendEndFrame` | `boolean \| undefined` | no | Whether to send an EndFrame at the end of the test. |
| `startTimeout` | `number \| undefined` | no | How long to wait, in seconds, for the pipeline to start before giving up. |