> For the complete documentation index, see [llms.txt](https://docs.anthriq.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.anthriq.com/bxi-studio/developer/overview/backend-services/pipelines.md).

# Run pipelines

A pipeline is a processing graph that runs **inside the executor server**, not in your process. You describe it as JSON, hand it to `ExecutorClient`, and the executor loads each node, wires them together, and runs the data through. Your script controls the lifecycle and reads the output; it never touches the samples in between.

## Understand the model

A pipeline is two lists: the **nodes** that do the work, and the **pipes** that connect them.

```mermaid
flowchart LR
  eeg["eegstream<br/><small>source</small>"] --> fft["fft<br/><small>processor</small>"]
  fft --> bp["bandpower<br/><small>processor</small>"]
  bp --> ws["websocket<br/><small>sink</small>"]
```

Each node is a compiled operator loaded from a shared library, and falls into one of three roles:

| Role      | Takes input | Produces output | Examples                                                       |
| --------- | ----------- | --------------- | -------------------------------------------------------------- |
| Source    | No          | Yes             | `eegstream`, `nidaqstream`, `csvfilestream`, `signalgenerator` |
| Processor | Yes         | Yes             | `fft`, `bandpower`, `firfilter`, `iirfilter`                   |
| Sink      | Yes         | No              | `websocket`, `csvrecorder`, `logger`                           |

A pipe names a `source` node and a `destination` node by id. Every pipeline needs at least one source and one sink; a graph of processors alone has nothing to run on and nowhere to put the result.

Nodes declare their parameters, inputs, and outputs in a manifest, which is what [Manage operators and pipelines](/bxi-studio/developer/overview/backend-services/registries.md) installs and lists. The executor validates a definition against those manifests, so a misspelled parameter or a pipe between incompatible nodes is caught at `validatePipeline` rather than at run time.

### Why it runs out of process

The executor owns the graph for the same reason the bridge owns the device: the work is C++, it must keep running whether or not your script is alive, and several clients can watch one pipeline. Your process sends lifecycle commands and receives logs.

That has a consequence worth knowing early — a pipeline **outlives the client that created it**. Disconnecting does not stop it. Call `destroyPipeline` or it keeps running.

## Write a pipeline definition

A definition is a plain object with `nodes` and `pipes`. This one takes live EXG from a device, runs an FFT, reduces it to band powers, and serves the result over a WebSocket:

```json
{
  "nodes": [
    {
      "id": "eeg",
      "module_path": "/path/to/nodes/libeegstream.dylib",
      "config": {
        "host": "localhost",
        "port": 9000,
        "streamId": 3,
        "sourceId": "0x2100",
        "zmqAddress": "ipc:///tmp/bxi-interface.sock",
        "deviceId": "anthriq-instinct"
      }
    },
    {
      "id": "fft",
      "module_path": "/path/to/nodes/libfft.dylib",
      "config": { "resolution": 0.25, "numberOfSegments": 2 }
    },
    {
      "id": "bp",
      "module_path": "/path/to/nodes/libbandpower.dylib",
      "config": { "normalize": true }
    },
    {
      "id": "ws",
      "module_path": "/path/to/nodes/libwebsocket.dylib",
      "config": { "port": 9500 }
    }
  ],
  "pipes": [
    { "source": "eeg", "destination": "fft" },
    { "source": "fft", "destination": "bp" },
    { "source": "bp", "destination": "ws" }
  ]
}
```

| Field         | Meaning                                                               |
| ------------- | --------------------------------------------------------------------- |
| `id`          | Name for this node within the pipeline; pipes and signals refer to it |
| `module_path` | Absolute path to the operator's shared library                        |
| `config`      | Parameters for that operator, as its manifest declares them           |

`module_path` carries the platform's own extension — `.dylib` on macOS, `.so` on Linux, `.dll` on Windows. Build the path at run time rather than hardcoding one, or the same definition fails on another machine.

> **Note:** The `eegstream` node reaches the device through the same BXI interface bridge the SDK uses, at `zmqAddress`. It does not open its own transport, so a pipeline and an ordinary [EXG stream](/bxi-studio/developer/overview/features/exg.md) can read one device at once.

### Swap the source to run without hardware

Replacing the source node with `signalgenerator` gives the same graph on synthetic data, which is how the processing half gets tested without a device attached:

```json
{
  "id": "gen",
  "module_path": "/path/to/nodes/libsignalgenerator.dylib",
  "config": { "sampleRate": 250, "channels": 4 }
}
```

Nothing else in the definition changes. Because the pipes name nodes by id, repointing `{ "source": "gen", "destination": "fft" }` is the whole edit.

## Method reference

| Method             | Payload                                   | Response data                                                             |
| ------------------ | ----------------------------------------- | ------------------------------------------------------------------------- |
| `createPipeline`   | `{ pipelineId, config, deviceId? }`       | `{ pipelineId, state, nodeCount, logEndpoint, logFile }`                  |
| `validatePipeline` | `{ config }`                              | `{ valid, message?, type?, details? }`                                    |
| `startPipeline`    | `{ pipelineId }`                          | `{ pipelineId, state }`                                                   |
| `stopPipeline`     | `{ pipelineId }`                          | `{ pipelineId, state }`                                                   |
| `destroyPipeline`  | `{ pipelineId }`                          | `{ pipelineId, state }`                                                   |
| `signalPipeline`   | `{ pipelineId, signal, nodeIds?, args? }` | `{ delivered }`                                                           |
| `listPipelines`    | `{ status? }`                             | `{ pipelines, total }`                                                    |
| `getPipelineInfo`  | `{ pipelineId }`                          | `{ pipelineId, state, nodeCount, logEndpoint, logFile, nodes?, config? }` |

A pipeline is in one of five states: `created`, `running`, `stopped`, `destroyed`, or `error`.

Each method also has a shorter alias — `create`, `validate`, `start`, `stop`, `destroy`, `signal`, `list`, `info` — kept for backwards compatibility with existing frontend code. They are marked deprecated and forward to the methods above. Use the full names in new code.

## Create and run

{% tabs %}
{% tab title="Node.js" %}

```typescript
import { ExecutorClient } from "@anthriq_dev/services";

const executor = new ExecutorClient({ endpoint: "ipc:///tmp/bxi-executor.sock" });
await executor.connect();

const created = await executor.createPipeline({
  pipelineId: "eeg-recording",
  config: {
    nodes: [
      { id: "eeg", type: "eegstream", config: { channels: 8 } },
      { id: "fft", type: "fft", config: { windowSize: 256 } },
      { id: "ws", type: "websocket", config: { port: 8080 } },
    ],
    pipes: [
      { source: "eeg", destination: "fft" },
      { source: "fft", destination: "ws" },
    ],
  },
});

console.log(created.data?.nodeCount, created.data?.logEndpoint);

await executor.startPipeline({ pipelineId: "eeg-recording" });
```

{% endtab %}

{% tab title="Python" %}

```python
from anthriq_services import ExecutorClient

async with ExecutorClient("ipc:///tmp/bxi-executor.sock") as executor:
    created = await executor.create_pipeline(
        pipeline_id="exg-recording",
        config={
            "nodes": [
                {"id": "eeg", "type": "eegstream", "config": {"channels": 8}},
                {"id": "fft", "type": "fft", "config": {"windowSize": 256}},
                {"id": "ws", "type": "websocket", "config": {"port": 8080}},
            ],
            "pipes": [
                {"source": "eeg", "destination": "fft"},
                {"source": "fft", "destination": "ws"},
            ],
        },
    )

    print(created.data["nodeCount"], created.data["logEndpoint"])

    await executor.start_pipeline("exg-recording")
```

{% endtab %}
{% endtabs %}

`startPipeline` then returns `{ pipelineId: 'eeg-recording', state: 'running' }`.

`createPipeline` instantiates the graph without running it; `startPipeline` begins execution. Pass `deviceId` to bind the pipeline to a specific device when more than one is connected.

## Validate first

`validatePipeline` checks a configuration without instantiating it, which is how to surface errors in a user interface before committing.

{% tabs %}
{% tab title="Node.js" %}

```typescript
// jsonData must be a JSON string. Passing the object itself reports
// "Pipeline missing or invalid 'nodes' array" rather than validating it.
const result = await executor.validatePipeline({
  config: { jsonData: JSON.stringify(pipelineConfig) },
});

if (!result.data?.valid) {
  console.error(result.data?.message, result.data?.details);
}
```

{% endtab %}

{% tab title="Python" %}

```python
# jsonData must be a JSON string, not the dict itself.
result = await executor.validate_pipeline({"jsonData": json.dumps(pipeline_config)})

if not result.data["valid"]:
    print(result.data.get("message"), result.data.get("details"))
```

{% endtab %}
{% endtabs %}

`jsonData` must be a **JSON string**. Handing it the configuration object directly does not raise — the call succeeds and reports the pipeline as invalid with `Pipeline missing or invalid 'nodes' array`, which reads like a problem with your graph rather than with the argument.

> **Note:** A `validatePipeline` call that fails validation still returns `success: true`. The verdict is `data.valid`; `success` only says the request reached the executor and came back.

## Signal running nodes

{% tabs %}
{% tab title="Node.js" %}

```typescript
await executor.signalPipeline({
  pipelineId: "eeg-recording",
  signal: "set_gain",
  nodeIds: ["eeg"],
  args: { gain: 24 },
});
```

{% endtab %}

{% tab title="Python" %}

```python
await executor.signal_pipeline(
    pipeline_id="exg-recording",
    signal="set_gain",
    node_ids=["eeg"],
    args={"gain": 24},
)
```

{% endtab %}
{% endtabs %}

Omit `nodeIds` to broadcast to every node.

The response reports only whether the signal was delivered, not what the node did with it. Watch the pipeline log stream to confirm the effect.

Which signals a node accepts is declared in its operator metadata. Read it with `operatorRegistry.info()`. See [Manage operators and pipelines](/bxi-studio/developer/overview/backend-services/registries.md).

## Query pipelines

{% tabs %}
{% tab title="Node.js" %}

```typescript
const all = await executor.listPipelines();
const running = await executor.listPipelines({ status: "running" });

const info = await executor.getPipelineInfo({ pipelineId: "eeg-recording" });
for (const node of info.data?.nodes ?? []) {
  console.log(node.id, node.type, node.state);
}
```

{% endtab %}

{% tab title="Python" %}

```python
everything = await executor.list_pipelines()
running = await executor.list_pipelines(status="running")

info = await executor.get_pipeline_info("exg-recording")
for node in info.data.get("nodes", []):
    print(node["id"], node["type"], node["state"])
```

{% endtab %}
{% endtabs %}

A node stuck in `created` while the others run is the usual sign of a failed start; the pipeline log stream carries the reason.

`getPipelineInfo` returns per-node state and the original configuration, which `listPipelines` omits.

## Stream per-pipeline logs

Each pipeline owns a ZeroMQ PUB socket carrying its node logs and lifecycle events. Pass `onPipelineLog` and the client subscribes on `createPipeline` and unsubscribes on `destroyPipeline`.

{% tabs %}
{% tab title="Node.js" %}

```typescript
const executor = new ExecutorClient({
  endpoint: "ipc:///tmp/bxi-executor.sock",
  onPipelineLog: (pipelineId, entry) => {
    console.log(`[${pipelineId}/${entry.source}] ${entry.level}: ${entry.message}`);
  },
});
```

{% endtab %}

{% tab title="Python" %}

```python
def on_log(pipeline_id: str, entry: dict) -> None:
    print(f"[{pipeline_id}/{entry.get('source')}] {entry['level']}: {entry['message']}")

executor = ExecutorClient("ipc:///tmp/bxi-executor.sock", on_pipeline_log=on_log)
await executor.connect()
```

{% endtab %}
{% endtabs %}

`entry.source` is `"node"` for node logs and `"pipeline"` for lifecycle events. Node logs carry the node identifier in `entry.data?.nodeId`.

Subscribe manually to a pipeline this client did not create:

{% tabs %}
{% tab title="Node.js" %}

```typescript
await executor.subscribeToPipelineLogs("existing-pipeline", "ipc:///tmp/bxi-pipeline-existing.sock");
await executor.unsubscribeFromPipelineLogs("existing-pipeline");
```

{% endtab %}

{% tab title="Python" %}

```python
await executor.subscribe_to_pipeline_logs(
    "existing-pipeline", "ipc:///tmp/bxi-pipeline-existing.sock"
)
await executor.unsubscribe_from_pipeline_logs("existing-pipeline")
```

{% endtab %}
{% endtabs %}

The second argument is the socket address from `logEndpoint`, not a port number.

Pipeline logs are also written to disk, at the path returned as `logFile`:

```
<tmp>/bxi/pipelines/<pipelineId>.log
```

Each line is one JSON entry in the same format as the socket messages.

## Destroy pipelines

{% tabs %}
{% tab title="Node.js" %}

```typescript
await executor.stopPipeline({ pipelineId: "eeg-recording" });
await executor.destroyPipeline({ pipelineId: "eeg-recording" });
await executor.disconnect();
```

{% endtab %}

{% tab title="Python" %}

```python
await executor.stop_pipeline("exg-recording")
await executor.destroy_pipeline("exg-recording")
await executor.disconnect()
```

{% endtab %}
{% endtabs %}

> **Warning:** A pipeline that is not destroyed keeps its nodes, sockets, and device claims allocated in the executor server, which outlives the SDK process. Destroy every pipeline you create, including on the error path.

## Complete example

{% tabs %}
{% tab title="Node.js" %}

```typescript
import { ExecutorClient } from "@anthriq_dev/services";

async function runPipeline(pipelineId: string, config: Record<string, unknown>) {
  const executor = new ExecutorClient({
    endpoint: "ipc:///tmp/bxi-executor.sock",
    onPipelineLog: (id, entry) => console.log(`[${id}] ${entry.level}: ${entry.message}`),
  });

  await executor.connect();

  try {
    const validation = await executor.validatePipeline({ config: { jsonData: config } });
    if (!validation.data?.valid) {
      throw new Error(`Invalid pipeline: ${validation.data?.message}`);
    }

    const created = await executor.createPipeline({ pipelineId, config });
    if (!created.success) throw new Error(created.error?.message);

    await executor.startPipeline({ pipelineId });
    await new Promise((resolve) => setTimeout(resolve, 30000));
    await executor.stopPipeline({ pipelineId });
  } finally {
    await executor.destroyPipeline({ pipelineId });
    await executor.disconnect();
  }
}
```

{% endtab %}

{% tab title="Python" %}

```python
import asyncio
from anthriq_services import ExecutorClient

async def run_pipeline(pipeline_id: str, config: dict, duration_s: float) -> None:
    executor = ExecutorClient(
        "ipc:///tmp/bxi-executor.sock",
        on_pipeline_log=lambda pid, entry: print(f"[{pid}] {entry['level']}: {entry['message']}"),
    )
    await executor.connect()

    try:
        validation = await executor.validate_pipeline(config)
        if not validation.data["valid"]:
            raise RuntimeError(f"Invalid pipeline: {validation.data.get('message')}")

        created = await executor.create_pipeline(pipeline_id=pipeline_id, config=config)
        if not created.success:
            raise RuntimeError(created.error.message)

        await executor.start_pipeline(pipeline_id)
        await asyncio.sleep(duration_s)
        await executor.stop_pipeline(pipeline_id)
    finally:
        await executor.destroy_pipeline(pipeline_id)
        await executor.disconnect()
```

{% endtab %}
{% endtabs %}

## Next steps

* [Manage operators and pipelines](/bxi-studio/developer/overview/backend-services/registries.md)
* [Record and play back](/bxi-studio/developer/overview/backend-services/recording.md)