@anthriq_dev/services

Node.js SDK for communicating with BXI service servers (Executor, Operator Registry, Pipeline Registry) over ZMQ.

Architecture

Node.js SDK (ZMQ DEALER) ──→ C++ Service Server (ZMQ ROUTER)
                                ├── Executor         (tcp://*:5560)
                                ├── Operator Registry (tcp://*:5561)
                                └── Pipeline Registry (tcp://*:5562)

Service-level Logs:
Node.js SDK (ZMQ SUB) ←── C++ Service Server (ZMQ PUB)
                                ├── Executor         (tcp://*:5570)
                                ├── Operator Registry (tcp://*:5571)
                                └── Pipeline Registry (tcp://*:5572)

Per-pipeline Logs (direct streaming, dynamic ports):
Node.js SDK (ZMQ SUB) ←── Pipeline PUB Socket (tcp://127.0.0.1:<dynamic>)
                                ├── Pipeline A (port returned in create response)
                                ├── Pipeline B (port returned in create response)
                                └── ...

Install

npm install @anthriq_dev/services

Requires zeromq native bindings (installed automatically).

Quick Start

Unified Client (all services)

import { BxiServicesClient } from "@anthriq_dev/services";

const services = new BxiServicesClient({
  onLog: (entry) => console.log(`[${entry.service}] ${entry.level}: ${entry.message}`),
});

await services.connect();

// Create and run a pipeline
const result = await services.executor.create({
  pipelineId: "eeg-pipeline",
  config: { /* pipeline JSON */ },
});
console.log(result.data); // { pipelineId, state: "created", nodeCount }

await services.executor.start({ pipelineId: "eeg-pipeline" });

// List installed operators
const ops = await services.operatorRegistry.list();
console.log(ops.data?.operators);

// List installed pipelines
const pipes = await services.pipelineRegistry.list();
console.log(pipes.data?.pipelines);

await services.executor.stop({ pipelineId: "eeg-pipeline" });
await services.executor.destroy({ pipelineId: "eeg-pipeline" });

await services.disconnect();

Individual Clients

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

const executor = new ExecutorClient({ endpoint: "tcp://localhost:5560" });
await executor.connect();

await executor.create({ pipelineId: "pipe1", config: pipelineJson });
await executor.start({ pipelineId: "pipe1" });
// ...
await executor.disconnect();

API Reference

BxiServicesClient

Unified client that manages connections to all three service servers.

const services = new BxiServicesClient({
  executorEndpoint: "tcp://localhost:5560",         // default
  operatorRegistryEndpoint: "tcp://localhost:5561", // default
  pipelineRegistryEndpoint: "tcp://localhost:5562", // default
  timeout: 10000,                                   // request timeout ms
  onLog: (entry) => { /* handle real-time logs */ },
  debug: false,
});

await services.connect();

services.executor           // ExecutorClient
services.operatorRegistry   // OperatorRegistryClient
services.pipelineRegistry   // PipelineRegistryClient

await services.disconnect();

ExecutorClient

Manages pipeline lifecycle.

| Method | Payload | Response | |--------|---------|----------| | create(payload) | { pipelineId, config, deviceId? } | { pipelineId, state, nodeCount, logPort, logFile } | | validate(payload) | { config: { jsonData } } | { valid, message?, type?, details? } | | start(payload) | { pipelineId } | { pipelineId, state: "running" } | | stop(payload) | { pipelineId } | { pipelineId, state: "stopped" } | | destroy(payload) | { pipelineId } | { pipelineId, state: "destroyed" } | | signal(payload) | { pipelineId, signal, nodeIds?, args? } | { delivered: boolean } | | list(payload?) | { status? } | { pipelines: [...], total } | | info(payload) | { pipelineId } | { pipelineId, state, nodeCount, nodes?, config? } |

const executor = new ExecutorClient({
  endpoint: "tcp://localhost:5560",
  // Optional: receive per-pipeline logs (node logs + pipeline lifecycle)
  onPipelineLog: (pipelineId, entry) => {
    console.log(`[${pipelineId}/${entry.source}] ${entry.level}: ${entry.message}`);
  },
});
await executor.connect();

// Full pipeline lifecycle
// create() auto-subscribes to pipeline's log socket when onPipelineLog is set
const { data } = await executor.create({
  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" },
    ],
  },
});

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

// Send signal to specific nodes
await executor.signal({
  pipelineId: "eeg-recording",
  signal: "set_gain",
  nodeIds: ["eeg"],
  args: { gain: 24 },
});

// List all pipelines
const { data: listData } = await executor.list();
console.log(listData?.pipelines);

await executor.stop({ pipelineId: "eeg-recording" });
await executor.destroy({ pipelineId: "eeg-recording" });
await executor.disconnect();

OperatorRegistryClient

Manages operator (node plugin) installation and discovery.

| Method | Payload | Response | |--------|---------|----------| | install(payload) | { name, version, platform?, force? } | { name, version, installPath, status } | | uninstall(payload) | { name, version } | { name, version, removed } | | list(payload?) | { remote?, page?, pageSize?, operator?, versions? } | { operators: [...], total, page, pageSize } | | info(payload) | { name, version, remote? } | OperatorMetadata | | status(payload) | { name, version } | { name, version, status } | | push(payload) | { name, version, tarPath?, localOnly? } | { name, version, pushed, installed } | | repair(payload?) | {} | { repaired, issues[] } |

const registry = new OperatorRegistryClient({ endpoint: "tcp://localhost:5561" });
await registry.connect();

// Install an operator
await registry.install({ name: "fft", version: "1.0.0" });

// List installed operators
const { data } = await registry.list();
for (const op of data?.operators ?? []) {
  console.log(`${op.name}@${op.version} -> ${op.installPath}`);
}

// List remote operators
const { data: remote } = await registry.list({ remote: true });

// Check status
const { data: status } = await registry.status({ name: "fft", version: "1.0.0" });
console.log(status?.status); // "INSTALLED"

await registry.disconnect();

PipelineRegistryClient

Manages pipeline definition installation and discovery.

| Method | Payload | Response | |--------|---------|----------| | install(payload) | { id, version, force? } | { id, version, installPath, status } | | uninstall(payload) | { id, version } | { id, version, removed } | | list(payload?) | { remote?, page?, pageSize? } | { pipelines: [...], total, page, pageSize } | | info(payload) | { id, version, remote?, extend? } | PipelineMetadata | | status(payload) | { id, version } | { id, version, status } | | push(payload) | { id, version, pipelineJson?, tarPath?, localOnly? } | { id, version, pushed, installed } | | pull(payload) | { id, version, force? } | { id, version, installPath } |

const registry = new PipelineRegistryClient({ endpoint: "tcp://localhost:5562" });
await registry.connect();

// Register a pipeline from JSON
await registry.push({
  id: "eeg-basic",
  version: "1.0.0",
  pipelineJson: {
    name: "Basic EEG Pipeline",
    nodes: [
      { id: "eeg", type: "node", name: "eegstream", version: "1.0.0" },
      { id: "ws", type: "node", name: "websocket", version: "1.0.0", config: { port: 8080 } },
    ],
    pipes: [{ id: "p1", source: "eeg", destination: "ws" }],
  },
});

// Get pipeline info with resolved operator details
const { data } = await registry.info({ id: "eeg-basic", version: "1.0.0", extend: true });

await registry.disconnect();

Real-Time Log Streaming

Service-level Logs

All service servers publish structured logs on their PUB sockets. Subscribe via onLog callback:

import { BxiServicesClient, type LogEntry } from "@anthriq_dev/services";

const services = new BxiServicesClient({
  onLog: (entry: LogEntry) => {
    // entry.service   — "executor", "operator_registry", "pipeline_registry"
    // entry.level     — "debug" | "info" | "warn" | "error"
    // entry.source    — component (e.g., "executor", "node_loader")
    // entry.message   — log message
    // entry.timestamp — unix ms
    // entry.data      — optional structured data
    console.log(`[${entry.service}/${entry.source}] ${entry.level}: ${entry.message}`);
  },
});

await services.connect();
// Service-level events stream in (pipeline created/destroyed, service lifecycle)

Per-Pipeline Logs (Direct Streaming)

Each pipeline owns its own ZMQ PUB socket for direct log streaming. This includes all node logs and pipeline lifecycle events. The SDK auto-subscribes when you pass onPipelineLog to ExecutorClient:

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

const executor = new ExecutorClient({
  endpoint: "tcp://localhost:5560",
  onPipelineLog: (pipelineId, entry) => {
    // entry.source — "node" for node logs, "pipeline" for pipeline logs
    // entry.data?.nodeId — present for node logs
    console.log(`[${pipelineId}] ${entry.level}: ${entry.message}`);
  },
});
await executor.connect();

// create() returns logPort + logFile, and auto-subscribes to log stream
const { data } = await executor.create({
  pipelineId: "eeg-pipeline",
  config: pipelineJson,
});
console.log(data.logPort);  // dynamic port for ZMQ PUB socket
console.log(data.logFile);  // path to log file on server

// Node logs stream to onPipelineLog as pipeline runs
await executor.start({ pipelineId: "eeg-pipeline" });

// destroy() auto-unsubscribes
await executor.destroy({ pipelineId: "eeg-pipeline" });

You can also manually subscribe to existing pipelines:

// Subscribe to a pipeline created before this client existed
await executor.subscribeToPipelineLogs("existing-pipeline", 6001);

// Unsubscribe manually
await executor.unsubscribeFromPipelineLogs("existing-pipeline");

Log File Persistence

All pipeline logs are also written to files asynchronously (non-blocking). The log file path is returned in the create() response and is located at:

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

Each line is a JSON log entry, same format as the ZMQ PUB messages.

Service Process Management

The SDK can check if C++ service servers are running and optionally spawn them as detached background processes:

import { ServiceProcessManager } from "@anthriq_dev/services";

const manager = new ServiceProcessManager({
  autoSpawn: true, // spawn servers that aren't running
  services: [
    {
      name: "executor",
      binaryPath: "/path/to/build/bin/executor_server",
      port: 5560,
      logPort: 5570,
    },
    {
      name: "operator_registry",
      binaryPath: "/path/to/build/bin/operator_registry_server",
      port: 5561,
      logPort: 5571,
      extraArgs: ["--base-path", "/opt/operators"],
    },
  ],
});

// Check health and spawn if needed
const statuses = await manager.ensureRunning();
for (const status of statuses) {
  console.log(`${status.name}: ${status.running ? "running" : status.error}`);
}

// Clean up spawned processes on shutdown
await manager.stopAll();

Wire Protocol

The SDK uses the same JSON protocol as the BXI Interface daemon.

Request (SDK → Service):

{
  "feature": "executor",
  "operation": "create",
  "requestId": "req_abc123",
  "payload": { "pipelineId": "pipe1", "config": {} }
}

Response (Service → SDK):

{
  "success": true,
  "feature": "executor",
  "operation": "create",
  "requestId": "req_abc123",
  "data": { "pipelineId": "pipe1", "state": "created", "nodeCount": 3 },
  "message": "Pipeline created"
}

Response Shape

All typed methods return SdkResponse<T>:

interface SdkResponse<T> {
  success: boolean;
  data?: T;       // present on success
  error?: {       // present on failure
    code: string;
    message: string;
    details?: unknown;
  };
}

const result = await executor.create({ pipelineId: "p1", config: {} });

if (result.success) {
  console.log(result.data.pipelineId);
  console.log(result.data.state);      // "created"
  console.log(result.data.nodeCount);
} else {
  console.error(result.error?.message);
}

Default Ports

| Service | ROUTER (req/res) | PUB (logs) | |---------|-----------------|------------| | Executor | 5560 | 5570 | | Operator Registry | 5561 | 5571 | | Pipeline Registry | 5562 | 5572 |

Development

npm run build    # Build with tsup
npm run test     # Run tests
npm run clean    # Clean dist/