Adora

Agentic Dataflow-Oriented Robotic Architecture – a 100% Rust framework for building real-time robotics and AI applications.

Why Adora?

Performance

• 10-17x faster than ROS2 Python – 100% Rust internals with zero-copy shared memory IPC for messages >4KB, flat latency from 4KB to 4MB payloads
• Apache Arrow native – columnar memory format end-to-end with zero serialization overhead; shared across all language bindings

Developer Experience

• Single CLI, full lifecycle – adora run for local dev, adora up/start for distributed prod, plus build, logs, monitoring, record/replay all from one tool
• Declarative YAML dataflows – define pipelines as directed graphs, connect nodes through typed inputs/outputs, optional type annotations with static validation
• Multi-language nodes – write nodes in Rust, Python, C, or C++ with native APIs (not wrappers); mix languages freely in one dataflow
• Reusable modules – compose sub-graphs as standalone YAML files with typed inputs/outputs, parameters, and nested composition
• Hot reload – live-reload Python operators without restarting the dataflow

Production Readiness

• Fault tolerance – per-node restart policies (never/on-failure/always), exponential backoff, health monitoring, circuit breakers with configurable input timeouts
• Distributed by default – local shared memory between co-located nodes, automatic Zenoh pub-sub for cross-machine communication, SSH-based cluster management with label scheduling
• Configurable queue policies – drop_oldest (default) or backpressure per input, with metrics on dropped messages
• OpenTelemetry – built-in structured logging with rotation/routing, metrics, distributed tracing

Debugging and Observability

• Record/replay – capture dataflow messages to .adorec files, replay offline at any speed with node substitution
• Topic inspection – topic echo to print live data, topic hz TUI for frequency analysis, topic info for schema and bandwidth
• Resource monitoring – adora top TUI showing per-node CPU, memory, queue depth, network I/O across all machines
• Log aggregation – subscribe to adora/logs to receive structured log messages from all nodes without extra wiring
• Trace inspection – trace list and trace view for viewing coordinator spans without external infrastructure

Ecosystem

• Communication patterns – built-in service (request/reply), action (goal/feedback/result), and streaming (session/segment/chunk) patterns via well-known metadata keys
• ROS2 bridge – bidirectional interop with ROS2 topics, services, and actions
• In-process operators – lightweight functions that run inside a shared runtime, avoiding per-node process overhead

Next Steps

• Install Adora
• Quick Start tutorial
• Architecture overview
• Dataflow YAML reference
• Type annotations
• Communication patterns

Installation

From crates.io (recommended)

cargo install adora-cli           # CLI (adora command)
pip install adora-rs              # Python node/operator API

From source

git clone https://github.com/dora-rs/adora.git
cd adora
cargo build --release -p adora-cli
PATH=$PATH:$(pwd)/target/release

# Python API (requires maturin >= 1.8: pip install maturin)
# Must run from the package directory for dependency resolution
cd apis/python/node && maturin develop --uv && cd ../../..

Platform installers

macOS / Linux:

curl --proto '=https' --tlsv1.2 -LsSf \
  https://github.com/dora-rs/adora/releases/latest/download/adora-cli-installer.sh | sh

Windows:

powershell -ExecutionPolicy ByPass -c "irm https://github.com/dora-rs/adora/releases/latest/download/adora-cli-installer.ps1 | iex"

Build features

cargo install adora-cli --features redb-backend

Verify

adora --version
adora status

Getting Started with Python

This guide walks you through writing Python nodes and operators for adora dataflows.

Prerequisites

cargo install adora-cli    # CLI (adora command)
pip install adora-rs       # Python node/operator API

The adora-rs package includes pyarrow as a dependency.

Building from source (instead of pip install adora-rs):

pip install maturin  # requires >= 1.8
cd apis/python/node && maturin develop --uv && cd ../../..

Hello World: Sender and Receiver

Create three files:

sender.py – sends 100 numbered messages:

import pyarrow as pa
from adora import Node

node = Node()
for i in range(100):
    node.send_output("message", pa.array([i]))

receiver.py – receives and prints messages:

from adora import Node

node = Node()
for event in node:
    if event["type"] == "INPUT":
        values = event["value"].to_pylist()
        print(f"Received {event['id']}: {values}")
    elif event["type"] == "STOP":
        break

dataflow.yml – connects sender to receiver:

nodes:
  - id: sender
    path: sender.py
    outputs:
      - message

  - id: receiver
    path: receiver.py
    inputs:
      message: sender/message

Run it:

adora run dataflow.yml

Events

Every call to node.next() or iteration over for event in node returns an event dictionary:

Handle events by checking event["type"]:

for event in node:
    match event["type"]:
        case "INPUT":
            process(event["id"], event["value"])
        case "INPUT_CLOSED":
            print(f"Input {event['id']} closed")
        case "STOP":
            break

Working with Arrow Data

All data flows through adora as Apache Arrow arrays. Common patterns:

import pyarrow as pa

# Simple values
node.send_output("count", pa.array([42]))
node.send_output("names", pa.array(["alice", "bob"]))

# Read values back
values = event["value"].to_pylist()  # [42] or ["alice", "bob"]

# Structured data
struct = pa.StructArray.from_arrays(
    [pa.array([1.5]), pa.array(["hello"])],
    names=["x", "y"],
)
node.send_output("point", struct)

# Raw bytes (images, serialized data, etc.)
node.send_output("frame", pa.array(raw_bytes))

Operators

Operators are lightweight alternatives to nodes. They run inside the adora runtime process (no separate OS process), making them faster for simple transformations.

Define an Operator class with an on_event method:

# doubler_op.py
import pyarrow as pa
from adora import AdoraStatus

class Operator:
    def on_event(self, event, send_output) -> AdoraStatus:
        if event["type"] == "INPUT":
            value = event["value"].to_pylist()[0]
            send_output("doubled", pa.array([value * 2]), event["metadata"])
        return AdoraStatus.CONTINUE

Reference it in YAML with operator instead of path:

nodes:
  - id: timer
    path: adora/timer/millis/500
    outputs:
      - tick

  - id: doubler
    operator:
      python: doubler_op.py
      inputs:
        tick: timer/tick
      outputs:
        - doubled

When to use operators vs nodes:

Async Nodes

For nodes that need async I/O (HTTP calls, database queries, etc.), use recv_async():

import asyncio
from adora import Node

async def main():
    node = Node()
    for _ in range(50):
        event = await node.recv_async()
        if event["type"] == "STOP":
            break
        # Do async work here
        result = await fetch_data(event["value"])
        node.send_output("result", result)

asyncio.run(main())

See examples/python-async for a complete example.

Logging

Use node.log() for structured logging that integrates with adora logs:

node.log("info", "Processing item", {"count": str(i)})

Or use Python’s standard logging module – adora captures stdout/stderr automatically:

import logging
logging.info("Processing item %d", i)

See examples/python-logging for logging module integration.

Timers

Built-in timer nodes generate periodic ticks without writing any code:

nodes:
  - id: tick-source
    path: adora/timer/millis/100    # tick every 100ms
    outputs:
      - tick

  - id: my-node
    path: my_node.py
    inputs:
      tick: tick-source/tick

Also available: adora/timer/hz/30 for 30 Hz.

Next Steps

• Python API Reference – full API docs for Node, Operator, DataflowBuilder, CUDA
• Communication Patterns – service (request/reply) and action (goal/feedback/result) patterns
• Examples – python-dataflow, python-async, python-drain, python-concurrent-rw, python-multiple-arrays
• Distributed Deployment – running across multiple machines with adora up

Adora Architecture

Comprehensive architecture reference for Adora (AI-Dora, Agentic Dataflow-Oriented Robotic Architecture) — a 100% Rust framework for real-time robotics and AI applications.

Overview and Design Philosophy

Adora is built on four core principles:

1. Dataflow-oriented: Applications are directed graphs of nodes connected by typed data channels. Nodes declare inputs and outputs; the framework handles routing, scheduling, and lifecycle.
2. Zero-copy performance: Messages above 4 KiB use shared memory with 128-byte aligned buffers and atomic coordination, achieving 10-17x lower latency than ROS2.
3. Multi-language: First-class support for Rust, Python (PyO3), C, and C++ nodes — all sharing the same Apache Arrow data format.
4. Four-layer stack: Message protocol, core libraries, daemon/runtime execution, and CLI/coordinator orchestration.

Architecture Stack

┌─────────────────────────────────────────────────┐
│  CLI (adora)          Coordinator (orchestrator) │  Layer 4: Orchestration
├─────────────────────────────────────────────────┤
│  Daemon (per-machine)    Runtime (operators)     │  Layer 3: Execution
├─────────────────────────────────────────────────┤
│  adora-core    shared-memory-server    Node API  │  Layer 2: Core Libraries
├─────────────────────────────────────────────────┤
│  adora-message (protocol + Arrow types)          │  Layer 1: Protocol
└─────────────────────────────────────────────────┘

Workspace Structure

Rust edition 2024, MSRV 1.85.0, workspace version 0.1.0. All crates share the workspace version.

Binaries (7)

Core Libraries (6)

Extension Libraries (5)

API Crates (9)

Component Architecture

CLI

The adora command provides three command groups:

Lifecycle (run, up, down, build, start, stop, restart):

• adora run executes a dataflow locally without coordinator/daemon (single-machine shortcut)
• adora up / adora down manage coordinator + daemon infrastructure
• adora start / adora stop control dataflows on a running coordinator

Monitoring (list, logs, inspect, topic, node, record, replay, trace):

• Real-time inspection with adora inspect top
• Topic subscription and data inspection
• Recording and replay via .adorec files

Setup (status, new, graph, system, completion, self):

• Project scaffolding, dataflow visualization, self-update

Coordinator

The coordinator is an Axum-based WebSocket server that orchestrates distributed deployments.

                          ┌──────────────────┐
                          │   Coordinator     │
            WS /api/control  │  ┌────────────┐  │  WS /api/daemon
   CLI ◄──────────────────►  │  │   State    │  │ ◄──────────────────► Daemon(s)
                          │  │   Store    │  │
                          │  └────────────┘  │
                          │  /api/artifacts  │
                          │  /health         │
                          └──────────────────┘

WebSocket routes:

• /api/control — CLI control plane (build, start, stop, list, logs, topic subscribe)
• /api/daemon — Daemon registration and event stream
• /api/artifacts/{build_id}/{node_id} — Binary artifact downloads
• /health — Health check endpoint

State management: In-memory by default, optional persistent storage via redb backend.

Daemon

The daemon runs one per machine and manages the lifecycle of all nodes on that machine.

┌──────────────────────────────────────────────────────┐
│                     Daemon                           │
│                                                      │
│  ┌──────────┐  ┌───────────┐  ┌──────────────────┐  │
│  │ Event    │  │ Spawner   │  │ Node Comm        │  │
│  │ Loop     │──│ (nodes)   │  │ ┌──────────────┐ │  │
│  │          │  └───────────┘  │ │ TCP listener │ │  │
│  │ Sources: │  ┌───────────┐  │ │ Shmem server │ │  │
│  │ • Coord  │  │ Fault     │  │ │ Unix socket  │ │  │
│  │ • Nodes  │──│ Tolerance │  │ └──────────────┘ │  │
│  │ • Zenoh  │  └───────────┘  └──────────────────┘  │
│  │ • Timers │                                        │
│  └──────────┘                                        │
│                                                      │
│  ┌──────────────────────────────────────────────┐    │
│  │ Running Dataflows                            │    │
│  │  ├─ Node A (process) ◄──► TCP/Shmem          │    │
│  │  ├─ Node B (process) ◄──► TCP/Shmem          │    │
│  │  └─ Runtime (operators) ◄──► TCP/Shmem       │    │
│  └──────────────────────────────────────────────┘    │
└──────────────────────────────────────────────────────┘

Event loop (Daemon::run_inner()): Async Tokio event loop merging:

• Coordinator commands (WebSocket)
• Node events (TCP/shared memory)
• Inter-daemon events (Zenoh)
• Heartbeat (5s interval), metrics collection (2s), health checks (5s default)

Node spawning:

1. Create working directory for the node
2. Set up communication channel (TCP, shmem, or Unix domain socket)
3. Serialize NodeConfig to environment variable
4. Spawn process with sanitized environment (blocks LD_PRELOAD, DYLD_INSERT_LIBRARIES, etc.)
5. Monitor via ProcessHandle

Runtime

The runtime executes in-process operators (Python, shared library, WASM) in a dedicated process.

┌──────────────────────────────┐
│          Runtime             │
│                              │
│  ┌────────────────────────┐  │
│  │ Operator Runner        │  │
│  │ (separate thread)      │  │
│  │                        │  │
│  │ SharedLibrary → dlopen │  │
│  │ Python → PyO3          │  │
│  │ Wasm → (planned)       │  │
│  └──────────┬─────────────┘  │
│             │ flume(2)       │
│  ┌──────────▼─────────────┐  │
│  │ Event Merge Loop       │  │
│  │ ├─ OperatorEvent       │  │
│  │ └─ DaemonEvent         │  │
│  └────────────────────────┘  │
└──────────────────────────────┘

• Single-threaded Tokio runtime
• Operator runs in a separate thread, communicates via flume::bounded(2) channel
• Input queue size per data ID configurable (default: 10)

Nodes

Nodes are standalone processes that communicate with the daemon.

Lifecycle:

1. Node starts, reads NodeConfig from environment
2. Registers with daemon via DaemonRequest::Register
3. Subscribes to events via DaemonRequest::Subscribe
4. Processes events in a loop (NextEvent → handle → SendMessage)
5. Reports drop tokens for shared memory cleanup
6. Signals completion via OutputsDone

Communication Protocols

CLI to Coordinator (WebSocket)

JSON-RPC-like message format:

// Request (client → server)
{"id": "uuid", "method": "control", "params": {...}}

// Response (server → client)
{"id": "uuid", "result": {...}}
// or
{"id": "uuid", "error": "message"}

// Event (fire-and-forget, either direction)
{"event": "log", "payload": {...}}

Key control methods: Build, Start, Stop, List, Logs, TopicSubscribe, TopicUnsubscribe, Reload, Restart, Destroy.

Coordinator to Daemon (WebSocket)

Daemon events (daemon → coordinator): BuildResult, SpawnResult, AllNodesReady, AllNodesFinished, Heartbeat, StatusReport, Log, NodeMetrics, Exit.

Coordinator commands (coordinator → daemon): Build, Spawn, AllNodesReady, StopDataflow, ReloadDataflow, Logs, Destroy, Heartbeat.

Daemon to Node (Local)

Three transport options, configured via LocalCommunicationConfig:

TCP (default):

• Binds 127.0.0.1:0 (ephemeral port), TCP_NODELAY enabled
• Frame format: [8-byte u64 LE length][bincode payload]
• Max message: 64 MiB, read timeout: 30s

Shared Memory (zero-copy):

• Four 4 KiB regions per node: control, events, drop tokens, events-close
• Used for messages >= 4096 bytes (ZERO_COPY_THRESHOLD)
• Atomic synchronization with acquire/release ordering

Unix Domain Socket (Unix only):

• Socket at /tmp/{dataflow_id}/{node_id}.sock
• Permissions: 0o700
• Same bincode frame format as TCP

Node → Daemon requests: Register, Subscribe, SendMessage, CloseOutputs, OutputsDone, NextEvent, ReportDropTokens, SubscribeDrop, NodeConfig.

Daemon → Node replies: Result, PreparedMessage, NextEvents, NextDropEvents, NodeConfig, Empty.

Node events: Stop, Reload, Input, InputClosed, InputRecovered, NodeRestarted, AllInputsClosed.

Daemon to Daemon (Zenoh)

Topic pattern:

adora/{network_id}/{dataflow_id}/output/{node_id}/{output_id}

Default network_id is "default".

InterDaemonEvent:

• Output { dataflow_id, node_id, output_id, metadata, data } — data message
• OutputClosed { dataflow_id, node_id, output_id } — stream end

Message Types and Wire Formats

Timestamped Wrapper

All inter-component messages are wrapped in a timestamp:

#![allow(unused)]
fn main() {
pub struct Timestamped<T> {
    pub inner: T,
    pub timestamp: uhlc::Timestamp,  // hybrid logical clock
}
}

DataMessage

Transport abstraction for payloads:

#![allow(unused)]
fn main() {
pub enum DataMessage {
    Vec(AVec<u8, ConstAlign<128>>),    // inline, 128-byte aligned
    SharedMemory {
        shared_memory_id: String,
        len: usize,
        drop_token: DropToken,          // UUIDv7, tracks lifetime
    },
}
}

LogMessage

#![allow(unused)]
fn main() {
pub struct LogMessage {
    pub build_id: Option<BuildId>,
    pub dataflow_id: Option<DataflowId>,
    pub node_id: Option<NodeId>,
    pub daemon_id: Option<DaemonId>,
    pub level: LogLevelOrStdout,       // Stdout | LogLevel(Error/Warn/Info/Debug/Trace)
    pub target: Option<String>,
    pub module_path: Option<String>,
    pub file: Option<String>,
    pub line: Option<u32>,
    pub message: String,
    pub timestamp: DateTime<Utc>,
    pub fields: Option<BTreeMap<String, String>>,
}
}

NodeError

#![allow(unused)]
fn main() {
pub struct NodeError {
    pub timestamp: uhlc::Timestamp,
    pub cause: NodeErrorCause,         // GraceDuration | Cascading | FailedToSpawn | Other
    pub exit_status: NodeExitStatus,   // Success | IoError | ExitCode | Signal | Unknown
}
}

Data Format and Metadata

Apache Arrow

All data payloads use Apache Arrow columnar format with 128-byte alignment. Arrow type information is carried in every message via ArrowTypeInfo:

#![allow(unused)]
fn main() {
pub struct ArrowTypeInfo {
    pub data_type: DataType,           // Arrow DataType
    pub len: usize,
    pub null_count: usize,
    pub validity: Option<Vec<u8>>,     // null bitmap
    pub offset: usize,
    pub buffer_offsets: Vec<BufferOffset>,
    pub child_data: Vec<ArrowTypeInfo>,  // recursive for nested types
}
}

Metadata

Every message carries structured metadata:

#![allow(unused)]
fn main() {
pub struct Metadata {
    metadata_version: u16,
    timestamp: uhlc::Timestamp,
    pub type_info: ArrowTypeInfo,
    pub parameters: MetadataParameters,   // BTreeMap<String, Parameter>
}
}

Parameter Types

#![allow(unused)]
fn main() {
pub enum Parameter {
    Bool(bool),
    Integer(i64),
    String(String),
    ListInt(Vec<i64>),
    Float(f64),
    ListFloat(Vec<f64>),
    ListString(Vec<String>),
    Timestamp(DateTime<Utc>),
}
}

Well-Known Metadata Keys

Zero-Copy Shared Memory

Architecture

┌────────────────────────────────────────────────────┐
│              Shared Memory Region                  │
│                                                    │
│  ┌──────────┐ ┌──────────┐ ┌──────┐ ┌────┐ ┌────┐│
│  │ Server   │ │ Client   │ │Discon│ │Len │ │Data││
│  │ Event    │ │ Event    │ │(bool)│ │(u64)│ │    ││
│  └──────────┘ └──────────┘ └──────┘ └────┘ └────┘│
│  (raw_sync_2)  (raw_sync_2) AtomicBool AtomicU64  │
└────────────────────────────────────────────────────┘

ShmemChannel

#![allow(unused)]
fn main() {
pub struct ShmemChannel {
    memory: Shmem,
    server_event: Box<dyn EventImpl>,
    client_event: Box<dyn EventImpl>,
    disconnect_offset: usize,
    len_offset: usize,
    data_offset: usize,
    server: bool,
}
}

Synchronization Protocol

Send (write → release store length → signal event → check disconnect):

1. Copy data to shared memory buffer
2. Store message length with Release ordering (publishes data)
3. Signal event to wake receiver
4. Check disconnect flag with Acquire ordering

Receive (wait event → check disconnect → acquire load length → read data):

1. Wait for event signal
2. Check disconnect flag with Acquire ordering
3. Load message length with Acquire ordering (ensures all writes visible)
4. Read and deserialize data from buffer

Thresholds and Limits

DropToken Lifecycle

1. Sender allocates shared memory, generates DropToken (UUIDv7)
2. Sender transmits DataMessage::SharedMemory { shared_memory_id, len, drop_token }
3. Receiver processes data, returns drop_token via ReportDropTokens
4. Sender receives confirmed token, returns memory to cache for reuse

Dataflow Specification

YAML Format

nodes:
  # Standard node (executable)
  - id: my-node
    build: cargo build --release
    path: target/release/my-node
    inputs:
      tick: adora/timer/millis/100
      data: other-node/output
    outputs:
      - result
    restart_policy: on-failure
    max_restarts: 3
    restart_delay: 1.0
    env:
      DEBUG: true

  # Single operator (Python)
  - id: processor
    operator:
      python: process.py
      inputs:
        image: camera/frame
      outputs:
        - detection

  # Multi-operator runtime
  - id: pipeline
    operators:
      - id: stage1
        python: stage1.py
        inputs:
          data: source/output
        outputs:
          - intermediate
      - id: stage2
        shared-library: target/release/libstage2.so
        inputs:
          data: stage1/intermediate
        outputs:
          - final

  # ROS2 bridge
  - id: ros-input
    ros2:
      topic: /robot/state
      message_type: sensor_msgs/JointState
      direction: subscribe
      qos:
        reliable: true
    outputs:
      - joints

Descriptor Structs

#![allow(unused)]
fn main() {
pub struct Descriptor {
    pub nodes: Vec<Node>,
    pub communication: CommunicationConfig,
    pub deploy: Option<Deploy>,
    pub debug: Debug,
    pub health_check_interval: Option<f64>,  // default 5.0s
}
}

Node types (mutually exclusive fields):

• path — standard executable/script
• operator — single in-process operator
• operators — multiple in-process operators
• custom — legacy configuration
• ros2 — declarative ROS2 bridge

Timer Nodes

Built-in timer nodes generate periodic ticks:

• adora/timer/millis/<N> — every N milliseconds
• adora/timer/secs/<N> — every N seconds

Operator Sources

#![allow(unused)]
fn main() {
pub enum OperatorSource {
    SharedLibrary(String),   // .so/.dll path
    Python(PythonSource),    // Python module
    Wasm(String),            // WebAssembly (planned)
}
}

Deploy Configuration

#![allow(unused)]
fn main() {
pub struct Deploy {
    pub machine: Option<String>,
    pub working_dir: Option<PathBuf>,
    pub labels: BTreeMap<String, String>,
    pub distribute: DistributeStrategy,  // Local | Scp | Http
}
}

Fault Tolerance

Restart Policies

#![allow(unused)]
fn main() {
pub enum RestartPolicy {
    Never,       // default
    OnFailure,   // restart on non-zero exit
    Always,      // restart unless user-stopped or inputs closed
}
}

Configuration fields per node:

• max_restarts — 0 = unlimited
• restart_delay — initial backoff in seconds (doubles each attempt)
• max_restart_delay — caps exponential backoff
• restart_window — reset counter after N seconds (enables “N restarts per M seconds”)
• health_check_timeout — kill node if no activity within this duration

Health Monitoring

• Heartbeat interval: 5 seconds (daemon → coordinator)
• Health check interval: 5 seconds (configurable per dataflow)
• Metrics collection: 2-second interval (CPU, memory, disk, pending messages)

Circuit Breaker

Per-input timeout detection with automatic recovery:

1. Input configured with input_timeout: <seconds>
2. If no data arrives within timeout → InputClosed event sent to node
3. Node marks input as degraded, can use cached last-known value
4. When upstream recovers → InputRecovered event, circuit breaker re-opens
5. Node status transitions: Running → Degraded → Running

Cascading Error Tracking

#![allow(unused)]
fn main() {
pub struct CascadingErrorCauses {
    pub caused_by: BTreeMap<NodeId, NodeId>,
}
}

Tracks which node failure caused downstream failures, enabling root-cause analysis.

Fault Tolerance Metrics

#![allow(unused)]
fn main() {
pub struct FaultToleranceSnapshot {
    pub restarts: u64,
    pub health_check_kills: u64,
    pub input_timeouts: u64,
    pub circuit_breaker_recoveries: u64,
}
}

Reported per daemon via heartbeat events. Visible via adora inspect top.

Distributed Deployment

Multi-Daemon Architecture

  ┌──────────┐       Zenoh        ┌──────────┐
  │ Daemon A │◄──────────────────►│ Daemon B │
  │ Machine 1│    pub/sub         │ Machine 2│
  │          │                    │          │
  │ Node 1   │                    │ Node 3   │
  │ Node 2   │                    │ Node 4   │
  └────┬─────┘                    └────┬─────┘
       │ WS                            │ WS
       └──────────┐  ┌────────────────┘
                  ▼  ▼
             ┌──────────┐
             │Coordinator│
             │  :6013    │
             └──────────┘

Zenoh Topic Naming

adora/{network_id}/{dataflow_id}/output/{node_id}/{output_id}

• network_id isolates separate Adora clusters (default: "default")
• Zenoh router port: 7447, peer port: 5456
• Routing mode: linkstate

Build Distribution

Three strategies via DistributeStrategy:

• Local — each daemon builds from source (default)
• Scp — CLI pushes built binaries via SSH/SCP
• Http — daemons pull from coordinator’s /api/artifacts endpoint

Machine Labels

Nodes can target specific machines via labels:

_unstable_deploy:
  labels:
    gpu: "true"
    arch: "arm64"

Recording and Replay

.adorec Binary Format

[HEADER]
├─ MAGIC: 8 bytes ("ADORAREC")
├─ version: u16 LE (currently 1)
├─ start_nanos: u64 LE (Unix epoch nanoseconds)
├─ dataflow_id: 16 bytes (UUID)
├─ yaml_len: u32 LE
└─ descriptor_yaml: [u8; yaml_len]

[ENTRIES] (repeated)
├─ record_len: u32 LE
├─ node_id_len: u16 LE
├─ node_id: [u8; node_id_len]
├─ output_id_len: u16 LE
├─ output_id: [u8; output_id_len]
├─ timestamp_offset_nanos: u64 LE
├─ event_bytes_len: u32 LE
└─ event_bytes: [u8; event_bytes_len]    (bincode InterDaemonEvent)

[FOOTER] (optional, written on clean finish)
├─ FOOTER_MAGIC: 8 bytes ("ADORAEND")
├─ total_messages: u64 LE
└─ total_bytes: u64 LE

Writer/Reader API

#![allow(unused)]
fn main() {
pub struct RecordingWriter<W: Write> { /* ... */ }
impl<W: Write> RecordingWriter<W> {
    pub fn new(inner: W, header: &RecordingHeader) -> Result<Self>;
    pub fn write_entry(&mut self, entry: &RecordEntry) -> Result<()>;
    pub fn finish(self) -> Result<RecordingFooter>;
}

pub struct RecordingReader<R: Read> { /* ... */ }
impl<R: Read> RecordingReader<R> {
    pub fn open(inner: R) -> Result<Self>;
    pub fn header(&self) -> &RecordingHeader;
    pub fn next_entry(&mut self) -> Result<Option<RecordEntry>>;
}
}

Extensions

Telemetry

Distributed Tracing (adora-tracing):

• OpenTelemetry with OTLP exporter (compatible with Jaeger, Zipkin, Tempo)
• Context propagation across nodes
• Setup: set_up_tracing(name: &str)

Metrics (adora-metrics):

• System metrics via sysinfo (CPU, memory, disk)
• OpenTelemetry meter with OTLP exporter
• Async process observer: run_metrics_monitor(meter_id)

ROS2 Bridge

Declarative YAML-based ROS2 integration supporting:

Topics — subscribe (ROS2 → Adora) or publish (Adora → ROS2):

ros2:
  topic: /camera/image
  message_type: sensor_msgs/Image
  direction: subscribe

Services — client or server role:

ros2:
  service: /add_two_ints
  service_type: example_interfaces/AddTwoInts
  role: client

Actions — goal/feedback/result lifecycle:

ros2:
  action: /fibonacci
  action_type: example_interfaces/Fibonacci
  role: client

QoS configuration:

qos:
  reliable: true
  durability: transient_local
  keep_last: 10

Download

File download utility for fetching operator/node binaries from HTTP URLs. Sanitizes filenames, sets executable permissions on Unix.

Key Constants and Defaults

Identifiers and Data Structures

ID Types

Authentication

#![allow(unused)]
fn main() {
pub struct AuthToken(String);  // 64 hex chars (32 bytes)
}

• Generated via cryptographically random bytes
• Stored at <working_dir>/.adora-token
• Constant-time comparison to prevent timing attacks
• Applied to all WebSocket routes

Node Status

#![allow(unused)]
fn main() {
pub enum NodeStatus {
    Running,     // healthy
    Restarting,  // restart in progress
    Degraded,    // circuit breaker open (input timeout)
    Failed,      // terminal failure
}
}

Serialization Summary

Dataflow YAML Specification

Dataflows are defined in YAML files. Each file describes a graph of nodes, their inputs/outputs, and execution parameters.

A JSON Schema is available at the repo root (adora-schema.json) for editor autocompletion and validation.

Quick Start

nodes:
  - id: sender
    path: sender.py
    outputs:
      - message

  - id: receiver
    path: receiver.py
    inputs:
      message: sender/message

Run with adora run dataflow.yml (local mode) or adora up && adora start dataflow.yml (networked mode).

Editor Setup

Add a schema comment at the top of your YAML file for VS Code autocompletion (requires the YAML extension):

# yaml-language-server: $schema=https://raw.githubusercontent.com/dora-rs/adora/main/adora-schema.json
nodes:
  - id: my-node
    # ... autocompletion works here

Root-Level Fields

Node Configuration

Every node requires an id. All other fields are optional (though most nodes need at least path or operator/operators).

Identity

Source

A node’s executable comes from a local path, a git repository, a module reference, or is implicit (operator/ROS2 nodes).

Example with git source:

- id: rust-node
  git: https://github.com/dora-rs/adora.git
  branch: main
  build: cargo build -p example-node --release
  path: target/release/example-node

Data I/O

Inputs

Inputs subscribe to another node’s output using the format <node-id>/<output-id>:

inputs:
  # Short form
  image: camera/frames
  tick: adora/timer/millis/100

  # Long form with options
  sensor_data:
    source: sensor/frames
    queue_size: 10
    queue_policy: drop_oldest
    input_timeout: 5.0

  # Lossless input (blocks sender when full)
  commands:
    source: controller/cmd
    queue_size: 100
    queue_policy: backpressure

Built-in Timers

Timers are virtual nodes that emit ticks at fixed intervals:

inputs:
  tick: adora/timer/millis/100   # every 100ms
  slow: adora/timer/millis/1000  # every 1s
  fast: adora/timer/hz/30        # 30 Hz (~33ms)

Built-in Log Aggregation

Subscribe to structured log messages from all (or filtered) nodes:

inputs:
  all_logs: adora/logs               # all nodes, all levels
  errors:   adora/logs/error         # error+ from all nodes
  sensor:   adora/logs/info/sensor   # info+ from specific node

Each message arrives as a JSON-encoded LogMessage string. See Logging for details.

Outputs

A list of output identifiers the node produces:

outputs:
  - processed_image
  - metadata

Type Annotations

Optional type annotations for inputs and outputs. Types are never required – unannotated ports remain fully dynamic.

- id: camera
  path: camera.py
  outputs:
    - image
    - depth
  output_types:
    image: std/media/v1/Image
    depth: std/media/v1/Image

- id: detector
  path: detect.py
  inputs:
    image: camera/image
  input_types:
    image: std/media/v1/Image
  outputs:
    - bbox
  output_types:
    bbox: std/vision/v1/BoundingBox

Type URNs use the format std/<category>/v<version>/<TypeName> and support parameters (e.g. std/media/v1/AudioFrame[sample_type=f32]). See the Type Annotations Guide for the full standard type library, parameterized types, compatibility rules, and user-defined types.

Run adora validate <file> to check type annotations statically. For runtime checking, set ADORA_RUNTIME_TYPE_CHECK=warn or error:

adora validate dataflow.yml
ADORA_RUNTIME_TYPE_CHECK=warn adora run dataflow.yml

Types also appear on adora graph edge labels when annotated.

Module Parameters

When using module:, pass configuration values via params::

- id: fast_pipeline
  module: modules/transform.module.yml
  inputs:
    data: sender/value
  params:
    speed: "2.0"
    mode: turbo

Inside the module, params are available as $PARAM_<UPPERCASE_KEY> in args: and as environment variables. See the Modules Guide for full documentation.

Environment

env:
  MY_VAR: "value"          # string
  DEBUG: true               # boolean
  PORT: 8080                # integer
  RATE: 1.5                 # float
  FROM_HOST:
    __adora_env: HOST_VAR   # read from host environment at runtime

Environment variables apply to both build commands and node execution. Values support $VAR expansion syntax.

Logging

Example:

- id: sensor
  path: ./sensor
  min_log_level: info
  send_stdout_as: raw_output
  send_logs_as: log_entries
  max_log_size: "100MB"
  max_rotated_files: 3
  outputs:
    - data
    - raw_output
    - log_entries

When using send_stdout_as or send_logs_as, include the output name in the outputs list so downstream nodes can subscribe to it.

For a complete guide to all logging features, see Logging.

Fault Tolerance

Restart policies:

• never (default): no automatic restart
• on-failure: restart only on non-zero exit code
• always: restart on any exit, except when stopped by user or all inputs closed with success

Example with exponential backoff:

- id: sensor
  path: ./sensor
  restart_policy: on-failure
  max_restarts: 5
  restart_delay: 1.0         # 1s, 2s, 4s, 8s, 16s
  max_restart_delay: 30.0    # capped at 30s
  restart_window: 300.0      # 5 restarts per 5 minutes
  health_check_timeout: 30.0

Deployment

Assign nodes to specific machines using _unstable_deploy:

- id: camera-driver
  _unstable_deploy:
    machine: robot-arm
  path: ./target/debug/camera
  outputs:
    - frames

- id: ml-inference
  _unstable_deploy:
    machine: gpu-server
    labels:
      gpu: "true"
    distribute: scp
  path: ./target/debug/inference
  inputs:
    frames: camera-driver/frames

When nodes are on different machines, communication automatically switches from shared memory to Zenoh pub/sub.

Operator Nodes

Operators run in-process inside a shared runtime (no separate process). Use operator for a single operator or operators for multiple.

Single Operator

The id field is optional for single operators (defaults to the node id):

- id: detector
  operator:
    python: detect.py
    build: pip install -r requirements.txt
    inputs:
      image: camera/frames
    outputs:
      - bbox

Multiple Operators

Each operator in operators requires a unique id:

- id: runtime-node
  operators:
    - id: preprocessor
      shared-library: ../../target/debug/libpreprocess
      inputs:
        raw: sensor/data
      outputs:
        - processed
    - id: analyzer
      shared-library: ../../target/debug/libanalyze
      inputs:
        data: runtime-node/preprocessor/processed
      outputs:
        - result

Operator Source Types

Operators also support inputs, outputs, build, send_stdout_as, send_logs_as, min_log_level, max_log_size, and max_rotated_files with the same semantics as node-level fields.

ROS2 Bridge

Declare a node as a ROS2 bridge to automatically convert between ROS2 DDS messages and Adora’s Arrow format. No custom code needed.

Single Topic

- id: camera_bridge
  ros2:
    topic: /camera/image_raw
    message_type: sensor_msgs/Image
    direction: subscribe
  outputs:
    - image

Multiple Topics

- id: robot_bridge
  ros2:
    topics:
      - topic: /camera/image_raw
        message_type: sensor_msgs/Image
        direction: subscribe
        output: image
      - topic: /cmd_vel
        message_type: geometry_msgs/Twist
        direction: publish
        input: velocity
    qos:
      reliable: true
  inputs:
    velocity: planner/cmd_vel
  outputs:
    - image

Service Bridge

- id: add_service
  ros2:
    service: /add_two_ints
    service_type: example_interfaces/AddTwoInts
    role: server
  inputs:
    request: client_node/request
  outputs:
    - response

Action Bridge

- id: nav_action
  ros2:
    action: /navigate
    action_type: nav2_msgs/NavigateToPose
    role: client
  inputs:
    goal: planner/goal
  outputs:
    - feedback
    - result

QoS Configuration

QoS can be set at the bridge level (applies to all topics) or per-topic:

Other ROS2 Fields

Debug

_unstable_debug:
  publish_all_messages_to_zenoh: true

Required for adora topic echo, adora topic hz, and adora topic info commands.

Communication Patterns

Adora supports four communication patterns built on top of the dataflow:

• Topic (default): pub/sub dataflow
• Service: request/reply via request_id metadata
• Action: goal/feedback/result via goal_id/goal_status metadata, with cancellation support
• Streaming: session/segment/chunk via session_id/segment_id/seq/fin/flush metadata, with queue flush for interruption

See Communication Patterns for details and examples.

Full Example

health_check_interval: 10.0

_unstable_debug:
  publish_all_messages_to_zenoh: true

nodes:
  - id: webcam
    operator:
      python: webcam.py
      inputs:
        tick: adora/timer/millis/100
      outputs:
        - image

  - id: detector
    operator:
      python: detect.py
      build: pip install ultralytics
      inputs:
        image: webcam/image
      outputs:
        - bbox

  - id: plotter
    operator:
      python: plot.py
      inputs:
        image: webcam/image
        bbox: detector/bbox

  - id: logger
    path: ./logger
    inputs:
      bbox: detector/bbox
    send_stdout_as: logs
    min_log_level: info
    restart_policy: on-failure
    max_restarts: 3
    outputs:
      - logs

Type Annotations

Optional type annotations on dataflow inputs and outputs. Types are never required – unannotated ports remain fully dynamic. Type checking runs at build time and validate time (no runtime overhead by default).

Quick Start

nodes:
  - id: camera
    path: camera.py
    outputs:
      - image
    output_types:
      image: std/media/v1/Image

  - id: detector
    path: detect.py
    inputs:
      image: camera/image
    input_types:
      image: std/media/v1/Image
    outputs:
      - bbox
    output_types:
      bbox: std/vision/v1/BoundingBox

Validate with:

adora validate dataflow.yml

# Fail with non-zero exit code on warnings (for CI)
adora validate --strict-types dataflow.yml

# Type checks also run during build
adora build dataflow.yml --strict-types

You can also set strict_types: true at the top level of the YAML to enable strict mode without the CLI flag:

strict_types: true
nodes:
  # ...

Type URN Format

Type URNs follow the pattern std/<category>/v<version>/<TypeName>:

std/core/v1/Float32
std/media/v1/Image
std/vision/v1/BoundingBox

Parameterized Types

Some struct types accept parameters to distinguish variants:

std/media/v1/AudioFrame[sample_type=f32]
std/media/v1/AudioFrame[sample_type=f32,channels=2]

Matching rules:

• Same base + same params -> compatible
• Same base + one side unparameterized -> compatible (wildcard)
• Same base + different param values -> mismatch

# These are compatible (wildcard):
output_types:
  audio: std/media/v1/AudioFrame[sample_type=f32]
input_types:
  audio: std/media/v1/AudioFrame

# These are a mismatch:
output_types:
  audio: std/media/v1/AudioFrame[sample_type=f32]
input_types:
  audio: std/media/v1/AudioFrame[sample_type=i16]

Standard Type Library

std/core/v1

std/math/v1

std/control/v1

std/media/v1

std/vision/v1

Validation Rules

adora validate and adora build check:

1. Key existence: output_types keys must appear in outputs, input_types keys must appear in inputs
2. URN resolution: All type URNs must exist in the standard or user-defined type library. Typos get “did you mean?” suggestions.
3. Edge compatibility: Connected edges must have compatible types (exact match, implicit widening, or user-defined rules)
4. Timer auto-typing: Timer inputs (adora/timer/*) are automatically typed as std/core/v1/UInt64
5. Type inference: When only the upstream side annotates a type, it is inferred on the downstream input and reported
6. Parameterized types: Parameter mismatches are detected (see above)
7. Metadata patterns: output_metadata keys and pattern shorthands are validated (see below)
8. Schema compatibility: Struct types are checked at the field level – missing fields or wrong field types are flagged

All checks produce warnings (non-fatal by default). Use --strict-types to treat warnings as errors for CI pipelines.

Type warnings:
  - node "camera": output_types key "framez" not found in outputs list
  - node "detector": unknown type "std/vision/v1/BoundingBx" on output "bbox"
    (did you mean "std/vision/v1/BoundingBox"?)
  - node "detector": type mismatch on input "image": upstream camera/image
    declares "std/core/v1/Bytes", but expected "std/media/v1/Image"

Inferred types:
  inferred std/core/v1/Float64 on processor/reading (from sensor/reading)

Type Compatibility Rules

Beyond exact matching, the type checker supports implicit widening conversions:

Widening is transitive up to depth 3 (e.g. UInt8 -> UInt32 -> UInt64 works, but chains of 4+ do not).

User-Defined Compatibility Rules

Add custom rules in the dataflow YAML:

type_rules:
  - from: myproject/SensorV1
    to: myproject/SensorV2

nodes:
  # ...

Metadata Patterns

Nodes that implement communication patterns (services, actions) can declare required metadata keys on their outputs.

Explicit metadata

- id: server
  path: server.py
  outputs:
    - response
  output_metadata:
    response: [request_id]

Pattern shorthand

Use the pattern field to auto-imply required metadata keys:

- id: server
  path: server.py
  pattern: service-server
  outputs:
    - response

User-Defined Types

Projects can define custom types in a types/ directory next to the dataflow. The directory structure determines the URN prefix:

project/
  dataflow.yml
  types/
    myproject/
      sensors/
        v1.yml    # URN prefix: myproject/sensors/v1

Type YAML files use the same format as the standard library:

types:
  MySensor:
    arrow: Struct
    description: Custom sensor reading
    fields:
      - name: temperature
        type: Float32
      - name: humidity
        type: Float32

This creates the URN myproject/sensors/v1/MySensor.

The std/ prefix is reserved and cannot be used for user types.

User types are loaded automatically by adora validate and adora build when a types/ directory exists.

Runtime Type Checking

In addition to static validation, Adora supports optional runtime type checking on send_output(). When enabled, the actual Arrow data type is compared against the declared output_types at send time.

Enable via environment variable:

# Warn on mismatches (log and continue)
ADORA_RUNTIME_TYPE_CHECK=warn adora run dataflow.yml

# Error on mismatches (node returns error)
ADORA_RUNTIME_TYPE_CHECK=error adora run dataflow.yml

Valid values: 1, warn, true (warn mode), error (error mode). Unset or any other value disables checking (zero overhead).

Scope:

• Validates output_types on the sender side (send_output() calls). input_types are checked statically by adora validate but not enforced at runtime
• Covers all languages that send Arrow arrays (Rust, Python, C++ Arrow path)
• Raw byte sends (send_output_bytes, C nodes) are untyped and skip checking
• Complex types (Struct-based: Image, Vector3, etc.) are skipped – only primitive types, String, Bytes, and Bool are validated at runtime

Graph Visualization

When outputs have type annotations, adora graph shows the type on edge labels:

adora graph dataflow.yml --open

Edges display as output_name [TypeName] (e.g. image [Image]).

Operators

Operators support the same output_types, input_types, output_metadata, and pattern fields:

- id: runtime-node
  operators:
    - id: preprocessor
      python: preprocess.py
      inputs:
        raw: sensor/data
      input_types:
        raw: std/core/v1/Bytes
      outputs:
        - processed
      output_types:
        processed: std/media/v1/Image

Modules (Reusable Sub-Dataflows)

Modules let you define reusable sub-graphs of nodes in separate YAML files and compose them into larger dataflows. Modules are expanded at compile time – the runtime never sees them.

Quick Start

Module file (modules/transform_module.yml):

module:
  name: transform_pipeline
  inputs: [raw_data]
  outputs: [filtered]

nodes:
  - id: doubler
    path: doubler.py
    inputs:
      data: _mod/raw_data
    outputs:
      - doubled

  - id: filter
    path: filter_even.py
    inputs:
      data: doubler/doubled
    outputs:
      - filtered

Dataflow file (dataflow.yml):

nodes:
  - id: sender
    path: sender.py
    outputs:
      - value

  - id: pipeline
    module: modules/transform_module.yml
    inputs:
      raw_data: sender/value

  - id: receiver
    path: receiver.py
    inputs:
      filtered: pipeline/filtered

After expansion, pipeline becomes two nodes: pipeline.doubler and pipeline.filter, with all wiring resolved automatically.

Module Definition File

A module file has two sections:

module: header

nodes: list

Standard node definitions, with one special syntax: _mod/port_name references a module input port. When expanded, _mod/port_name is replaced with whatever the parent wired to that port.

module:
  name: my_module
  inputs: [camera_feed]
  outputs: [detections]

nodes:
  - id: detector
    path: detect.py
    inputs:
      image: _mod/camera_feed    # resolved to parent's wiring
    outputs:
      - detections

Module-level build

Modules can have a top-level build: command that runs before any inner node builds:

module:
  name: ml_pipeline
  inputs: [image]
  outputs: [result]

build: pip install -r requirements.txt

nodes:
  - id: model
    path: model.py
    inputs:
      image: _mod/image
    outputs:
      - result

Using Modules

Reference a module in a dataflow node using the module: field instead of path::

- id: nav_stack
  module: modules/navigation.module.yml
  inputs:
    goal_pose: localization/goal

The module node’s inputs: map wires parent outputs to module input ports. External nodes reference module outputs as <module_id>/<output_name> (e.g., nav_stack/cmd_vel).

Parameters

Pass configuration values to modules via params::

- id: fast_pipeline
  module: modules/transform_module.yml
  inputs:
    raw_data: sender/value
  params:
    speed: "2.0"
    mode: turbo

Inside the module, reference params in args: using $PARAM_<UPPERCASE_KEY>:

nodes:
  - id: processor
    path: processor.py
    args: --speed $PARAM_SPEED --mode $PARAM_MODE
    inputs:
      data: _mod/raw_data
    outputs:
      - result

Parameters are also injected as environment variables (PARAM_SPEED, PARAM_MODE) into every node inside the module.

Expansion Rules

1. Load the module YAML file and validate its header
2. Prefix all internal node IDs with {module_id}. (e.g., nav_stack.planner)
3. Replace _mod/port_name references with the actual sources from the parent’s input map
4. Rewrite internal cross-references (e.g., planner/path becomes nav_stack.planner/path)
5. Map module-declared outputs to internal node outputs, so nav_stack/cmd_vel resolves to nav_stack.controller/cmd_vel
6. Replace the module node with the expanded flat nodes
7. Substitute params: values in args: fields and inject as env vars

Use adora expand to see the result:

adora expand dataflow.yml

Nested Modules

Modules can reference other modules. The expansion is recursive with a depth limit of 8 levels:

# outer_module.yml
module:
  name: outer
  inputs: [data]
  outputs: [result]

nodes:
  - id: inner
    module: inner_module.yml
    inputs:
      raw: _mod/data

  - id: postprocess
    path: postprocess.py
    inputs:
      data: inner/processed
    outputs:
      - result

After expansion, node IDs are fully qualified: outer.inner.some_node.

Optional Inputs

Declare inputs as optional when a module should work with or without certain connections:

module:
  name: flexible_processor
  inputs: [data]
  inputs_optional: [config]
  outputs: [result]

nodes:
  - id: processor
    path: processor.py
    inputs:
      data: _mod/data
      config: _mod/config    # silently dropped if not wired
    outputs:
      - result

When the parent doesn’t wire config, the input is simply omitted from the expanded node.

Visualization

adora graph renders module boundaries as Mermaid subgraphs, making it easy to see which nodes came from which module:

adora graph dataflow.yml --open

Validation

Validate a standalone module file without a full dataflow:

adora expand --module modules/transform_module.yml

This checks:

• Valid YAML structure
• Module header is present with name, inputs, outputs
• All _mod/ references correspond to declared inputs or optional inputs
• No duplicate node IDs
• Internal wiring is consistent

Security

• Path confinement: Module file paths must resolve within the dataflow’s base directory. Absolute paths and directory traversal (../) outside the base are rejected.
• File size limit: Module files are capped at 1 MB.
• Depth limit: Recursive nesting is capped at 8 levels.
• Param key validation: Parameter keys must be alphanumeric with underscores only.

Example

See examples/module-dataflow/ for a complete working example with a sender, transform module (doubler + filter), and receiver.

adora run examples/module-dataflow/dataflow.yml

Communication Patterns

Adora is a dataflow framework based on pub/sub message passing. On top of basic topics, the framework supports service (request/reply), action (goal/feedback/result), and streaming (session/segment/chunk) patterns using well-known metadata keys. No changes to the daemon, coordinator, or YAML syntax are required – the patterns are implemented as conventions at the node API level.

1. Topic (pub/sub)

The default pattern. A node publishes data on an output, and any node that subscribes to that output receives it.

nodes:
  - id: publisher
    outputs:
      - data
  - id: subscriber
    inputs:
      data: publisher/data

Use when: streaming sensor data, periodic status, fire-and-forget events.

2. Service (request/reply)

A client sends a request and expects exactly one response, correlated by a request_id metadata key.

Well-known metadata keys

YAML

nodes:
  - id: client
    inputs:
      tick: adora/timer/millis/500
      response: server/response
    outputs:
      - request

  - id: server
    inputs:
      request: client/request
    outputs:
      - response

Node API helpers

#![allow(unused)]
fn main() {
// Client: send request with auto-generated request_id
let rid = node.send_service_request("request".into(), params, data)?;

// Server: pass through metadata.parameters (includes request_id)
node.send_service_response("response".into(), metadata.parameters, result)?;
}

The server MUST pass through the request_id from the incoming request’s metadata parameters into the response. The client matches responses to requests using this key.

Example: examples/service-example/

3. Action (goal/feedback/result)

A client sends a goal and receives periodic feedback plus a final result. Actions support cancellation.

Well-known metadata keys

Goal status values:

YAML

nodes:
  - id: client
    inputs:
      tick: adora/timer/millis/2000
      feedback: server/feedback
      result: server/result
    outputs:
      - goal
      - cancel

  - id: server
    inputs:
      goal: client/goal
      cancel: client/cancel
    outputs:
      - feedback
      - result

Cancel pattern

The client sends a message on the cancel output with goal_id in the metadata. The server checks for cancel requests between processing steps and sends a result with goal_status = "canceled".

Example: examples/action-example/

4. Streaming (session/segment/chunk)

For real-time pipelines (voice, video, sensor streams) where a user can interrupt mid-stream and queued data must be discarded.

Well-known metadata keys

YAML

nodes:
  - id: asr
    inputs:
      mic: mic-source/audio
    outputs:
      - text

  - id: llm
    inputs:
      text: asr/text
    outputs:
      - tokens

  - id: tts
    inputs:
      tokens: llm/tokens
    outputs:
      - audio

Node API

#![allow(unused)]
fn main() {
use adora_node_api::{StreamSegment, AdoraNode};

let mut seg = StreamSegment::new();

// Send chunks with auto-incrementing seq (e.g. inside an ASR node)
node.send_stream_chunk("text".into(), &mut seg, false, chunk_data)?;
// Mark final chunk of a segment
node.send_stream_chunk("text".into(), &mut seg, true, last_chunk)?;

// On user interruption: flush downstream queues and start a new segment.
// The prior segment ends without a fin=true signal -- old data is discarded.
let flush_params = seg.flush();
node.send_output("text".into(), flush_params, empty_data)?;
}

Queue flush behavior

When a message arrives with flush: true in its metadata, the receiver’s input queue is cleared of all older messages before the flush message is delivered. This enables instant interruption in voice pipelines – when the user speaks over TTS output, the ASR node sends a new segment with flush: true, and the TTS node immediately discards any queued audio chunks from the previous response.

Note: flush discards all queued messages on the input regardless of session_id. Do not multiplex independent sessions on a single input when using flush.

Python

# Streaming metadata is a plain dict
params = {
    "session_id": session_id,
    "segment_id": 1,
    "seq": 0,
    "fin": False,
    "flush": True,  # flush older queued messages
}
node.send_output("text", data, metadata={"parameters": params})

5. Choosing a pattern

6. Important details

• goal_status matching is case-sensitive. Always use the exact lowercase values: "succeeded", "aborted", "canceled". The ROS2 bridge defaults to Aborted for unrecognised values.

7. Python compatibility

Python nodes use the same metadata conventions. Parameters are plain dicts with string keys:

import uuid

# Service client (uuid7 for time-ordered IDs, matching Rust API)
params = {"request_id": str(uuid.uuid7())}
node.send_output("request", data, metadata={"parameters": params})

# Service server -- pass through parameters
node.send_output("response", result, metadata=event["metadata"])

Note: uuid.uuid7() requires Python 3.13+. On older versions, use the uuid_utils package or uuid.uuid4() (random v4 also works for correlation, but loses time-ordering).

Rust API Reference

This document covers the two main Rust crates for building Adora dataflow components:

• adora-node-api – for standalone node executables
• adora-operator-api – for in-process operators managed by the Adora runtime

Node API (adora-node-api)

Add to your Cargo.toml:

[dependencies]
adora-node-api = { workspace = true }

AdoraNode

The primary struct for sending outputs and retrieving node information. Obtained through one of the initialization functions below.

Initialization

#![allow(unused)]
fn main() {
// Recommended: auto-detect environment (daemon, testing, or interactive).
pub fn init_from_env() -> NodeResult<(Self, EventStream)>

// Same as init_from_env but errors instead of falling back to interactive mode.
pub fn init_from_env_force() -> NodeResult<(Self, EventStream)>

// For dynamic nodes: connect to the daemon by node ID.
pub fn init_from_node_id(node_id: NodeId) -> NodeResult<(Self, EventStream)>

// Try init_from_env first; fall back to init_from_node_id.
pub fn init_flexible(node_id: NodeId) -> NodeResult<(Self, EventStream)>

// Standalone interactive mode (prompts for inputs on the terminal).
pub fn init_interactive() -> NodeResult<(Self, EventStream)>

// Integration test mode with synthetic inputs/outputs.
pub fn init_testing(
    input: TestingInput,
    output: TestingOutput,
    options: TestingOptions,
) -> NodeResult<(Self, EventStream)>
}

init_from_env is the recommended entry point. It checks, in order:

1. Thread-local testing state set by setup_integration_testing
2. ADORA_NODE_CONFIG environment variable (set by the daemon)
3. ADORA_TEST_WITH_INPUTS environment variable (file-based integration testing)
4. Interactive terminal fallback (only if stdin is a TTY)

Sending Outputs

All send methods silently ignore output IDs not declared in the dataflow YAML.

#![allow(unused)]
fn main() {
// Send an Arrow array. Copies data into shared memory when beneficial.
pub fn send_output(
    &mut self,
    output_id: DataId,
    parameters: MetadataParameters,
    data: impl Array,
) -> NodeResult<()>

// Send raw bytes. Copies into shared memory when beneficial.
pub fn send_output_bytes(
    &mut self,
    output_id: DataId,
    parameters: MetadataParameters,
    data_len: usize,
    data: &[u8],
) -> NodeResult<()>

// Send raw bytes via a closure for zero-copy writing.
pub fn send_output_raw<F>(
    &mut self,
    output_id: DataId,
    parameters: MetadataParameters,
    data_len: usize,
    data: F,
) -> NodeResult<()>
where
    F: FnOnce(&mut [u8])

// Send raw bytes with explicit Arrow type information.
pub fn send_typed_output<F>(
    &mut self,
    output_id: DataId,
    type_info: ArrowTypeInfo,
    parameters: MetadataParameters,
    data_len: usize,
    data: F,
) -> NodeResult<()>
where
    F: FnOnce(&mut [u8])

// Send a pre-allocated DataSample with type information.
pub fn send_output_sample(
    &mut self,
    output_id: DataId,
    type_info: ArrowTypeInfo,
    parameters: MetadataParameters,
    sample: Option<DataSample>,
) -> NodeResult<()>

// Report output IDs as closed. No further sends allowed for those IDs.
pub fn close_outputs(&mut self, outputs_ids: Vec<DataId>) -> NodeResult<()>
}

Service, Action, and Streaming Helpers

Higher-level methods for the communication patterns. These use well-known metadata keys to correlate requests, goals, responses, and streaming segments.

#![allow(unused)]
fn main() {
// Generate a unique, time-ordered ID (UUID v7) for correlation.
pub fn new_request_id() -> String
pub fn new_goal_id() -> String   // alias for new_request_id

// Send a service request. Injects a `request_id` into parameters and returns it.
pub fn send_service_request(
    &mut self,
    output_id: DataId,
    parameters: MetadataParameters,
    data: impl Array,
) -> NodeResult<String>

// Send a service response. Semantic alias for send_output.
// Caller must pass through the request_id from the incoming request's metadata.
pub fn send_service_response(
    &mut self,
    output_id: DataId,
    parameters: MetadataParameters,
    data: impl Array,
) -> NodeResult<()>
}

Service example (client sends request, server replies):

#![allow(unused)]
fn main() {
// Client: auto-generates and injects request_id
let rid = node.send_service_request("request".into(), params, data)?;

// Server: pass through metadata.parameters (includes request_id)
node.send_service_response("response".into(), metadata.parameters, result)?;
}

Action example (client sends goal, server streams feedback + result):

#![allow(unused)]
fn main() {
use adora_node_api::{GOAL_ID, GOAL_STATUS, GOAL_STATUS_SUCCEEDED, Parameter};

// Client: generate goal_id, attach to params
let goal_id = AdoraNode::new_goal_id();
params.insert(GOAL_ID.to_string(), Parameter::String(goal_id));
node.send_output("goal".into(), params, data)?;

// Server: extract goal_id, send feedback/result with goal_status
let gid = get_string_param(&metadata.parameters, GOAL_ID);
}

Streaming example (real-time voice/video pipeline with interruption):

#![allow(unused)]
fn main() {
use adora_node_api::StreamSegment;

// Create a streaming segment builder (auto-generates session_id)
let mut seg = StreamSegment::new();

// Send chunks with auto-incrementing seq
node.send_stream_chunk("text".into(), &mut seg, false, chunk_data)?;
// Mark final chunk of a segment
node.send_stream_chunk("text".into(), &mut seg, true, last_chunk)?;

// On user interruption: flush downstream queues and start a new segment
let flush_params = seg.flush();
node.send_output("text".into(), flush_params, empty_data)?;
}

See patterns.md for the full guide and examples/service-example and examples/action-example for working code.

Data Allocation

#![allow(unused)]
fn main() {
// Allocate a DataSample of the given size.
// Uses shared memory for data >= ZERO_COPY_THRESHOLD (4096 bytes).
pub fn allocate_data_sample(&mut self, data_len: usize) -> NodeResult<DataSample>
}

Node Information

#![allow(unused)]
fn main() {
// Node ID from the dataflow YAML.
pub fn id(&self) -> &NodeId

// Unique identifier for this dataflow run.
pub fn dataflow_id(&self) -> &DataflowId

// Input/output configuration for this node.
pub fn node_config(&self) -> &NodeRunConfig

// True if this node was restarted after a previous exit or failure.
pub fn is_restart(&self) -> bool

// Number of times this node has been restarted (0 on first run).
pub fn restart_count(&self) -> u32

// Parsed dataflow YAML descriptor.
pub fn dataflow_descriptor(&self) -> NodeResult<&Descriptor>
}

Logging

Rust nodes have two ways to emit structured logs. Both produce identical structured log entries in the daemon.

Option 1: Node API (recommended for most cases)

All log methods emit structured JSONL to stdout, which the daemon parses automatically. Works with min_log_level filtering, send_logs_as routing, and adora/logs subscribers.

#![allow(unused)]
fn main() {
// General structured log. Level: "error", "warn", "info", "debug", "trace".
pub fn log(&self, level: &str, message: &str, target: Option<&str>)

// Structured log with additional key-value fields.
pub fn log_with_fields(
    &self,
    level: &str,
    message: &str,
    target: Option<&str>,
    fields: Option<&BTreeMap<String, String>>,
)

// Convenience methods (no target parameter).
pub fn log_error(&self, message: &str)
pub fn log_warn(&self, message: &str)
pub fn log_info(&self, message: &str)
pub fn log_debug(&self, message: &str)
pub fn log_trace(&self, message: &str)
}

Option 2: Rust tracing crate

When adora’s tracing subscriber is initialized (via init_tracing() or the default feature), tracing::info!() etc. output structured JSON to stdout that the daemon parses identically:

#![allow(unused)]
fn main() {
tracing::info!("Sensor started");
tracing::warn!(sensor_id = "temp-01", "High temperature");
}

Use tracing when you want ecosystem integration (spans, instrumentation, OpenTelemetry). Use node.log_*() when you want explicit control or structured fields as BTreeMap.

EventStream

Asynchronous iterator over incoming events destined for this node. Implements the futures::Stream trait.

The event stream closes itself after a Stop event is received. Nodes should exit once the stream ends.

#![allow(unused)]
fn main() {
// Block until the next event arrives. Returns None when the stream closes.
// Uses an internal EventScheduler that may reorder events for fairness.
pub fn recv(&mut self) -> Option<Event>

// Block with a timeout. Returns an Event::Error on timeout.
pub fn recv_timeout(&mut self, dur: Duration) -> Option<Event>

// Async receive with EventScheduler reordering.
pub async fn recv_async(&mut self) -> Option<Event>

// Async receive with a timeout. Returns Event::Error on timeout.
pub async fn recv_async_timeout(&mut self, dur: Duration) -> Option<Event>

// Non-blocking receive. Returns TryRecvError::Empty if nothing is ready.
pub fn try_recv(&mut self) -> Result<Event, TryRecvError>

// Drain all buffered events without blocking.
// Returns Some(Vec::new()) if nothing is ready; None if the stream is closed.
pub fn drain(&mut self) -> Option<Vec<Event>>

// True if no events are buffered in the scheduler or receiver.
pub fn is_empty(&self) -> bool

// Returns and resets accumulated drop counts per input ID.
// For `drop_oldest` inputs, drops happen at `queue_size`.
// For `backpressure` inputs, drops happen at 10x `queue_size` (hard safety cap).
pub fn drain_drop_counts(&mut self) -> HashMap<DataId, u64>
}

EventStream also implements futures::Stream<Item = Event>, so it can be used with StreamExt::next() and other combinators. Unlike recv/recv_async, the Stream implementation does not use the EventScheduler, preserving chronological event order.

Event

Represents an incoming event. This enum is #[non_exhaustive] – ignore unknown variants to stay forward-compatible.

#![allow(unused)]
fn main() {
#[non_exhaustive]
pub enum Event {
    // An input was received from another node.
    Input {
        id: DataId,           // input ID from the YAML (not the sender's output ID)
        metadata: Metadata,   // timestamp and type information
        data: ArrowData,      // Apache Arrow data
    },

    // The sender mapped to this input exited; no more data will arrive.
    InputClosed { id: DataId },

    // A previously closed input recovered (e.g., upstream node came back after timeout).
    InputRecovered { id: DataId },

    // An upstream node has restarted. Useful for resetting caches or state.
    NodeRestarted { id: NodeId },

    // The event stream is about to close. See StopCause for the reason.
    Stop(StopCause),

    // Instructs the node to reload an operator (used internally by the runtime).
    Reload { operator_id: Option<OperatorId> },

    // An unexpected internal error. Log it for debugging.
    Error(String),
}
}

StopCause

#![allow(unused)]
fn main() {
#[non_exhaustive]
pub enum StopCause {
    // Explicit stop via `adora stop` or Ctrl-C. Exit promptly or be killed.
    Manual,

    // All inputs were closed (upstream nodes exited). Only sent if the node has inputs.
    AllInputsClosed,
}
}

Supporting Types

DataSample

A data region suitable for sending as an output message. Uses shared memory for data >= ZERO_COPY_THRESHOLD to enable zero-copy transfer.

Implements Deref<Target = [u8]> and DerefMut for reading and writing the underlying bytes.

Metadata and MetadataParameters

#![allow(unused)]
fn main() {
// Full metadata attached to every input event.
pub struct Metadata {
    // Contains timestamp, Arrow type info, and user-defined parameters.
}

// User-controlled metadata fields attached when sending outputs.
// Type alias for BTreeMap<String, Parameter>.
// Default is empty. Pass metadata.parameters from an input to forward metadata.
pub type MetadataParameters = BTreeMap<String, Parameter>;

// A single metadata parameter value.
pub enum Parameter {
    Bool(bool), Integer(i64), Float(f64), String(String),
    ListInt(Vec<i64>), ListFloat(Vec<f64>), ListString(Vec<String>),
    Timestamp(DateTime<Utc>),
}

// Extract typed parameters, returning None if missing or wrong type.
pub fn get_string_param<'a>(params: &'a MetadataParameters, key: &str) -> Option<&'a str>
pub fn get_integer_param(params: &MetadataParameters, key: &str) -> Option<i64>
pub fn get_bool_param(params: &MetadataParameters, key: &str) -> Option<bool>
}

Well-known metadata keys (for communication patterns):

All constants are re-exported from adora_node_api.

Identity Types

#![allow(unused)]
fn main() {
// Unique identifier for a running dataflow instance (UUID v4).
pub struct DataflowId(/* ... */);

// Node identifier, as defined in the dataflow YAML.
pub struct NodeId(/* ... */);

// Input/output identifier, as defined in the dataflow YAML.
pub struct DataId(/* ... */);
}

Error Types

#![allow(unused)]
fn main() {
#[derive(Debug, Error)]
pub enum NodeError {
    Init(String),        // config parsing, env vars, daemon handshake
    Connection(String),  // daemon connection lost
    Output(String),      // send or close failure
    Data(String),        // allocation or descriptor parsing
    Internal(eyre::Report),  // catch-all for unexpected errors
}

pub type NodeResult<T> = Result<T, NodeError>;
}

TryRecvError

#![allow(unused)]
fn main() {
pub enum TryRecvError {
    Empty,   // no event available right now
    Closed,  // event stream has been closed
}
}

ZERO_COPY_THRESHOLD

#![allow(unused)]
fn main() {
pub const ZERO_COPY_THRESHOLD: usize = 4096;
}

Messages smaller than this threshold are sent via TCP. Messages at or above this size use shared memory for zero-copy transfer.

ArrowData

#![allow(unused)]
fn main() {
// Wrapper around arrow::array::ArrayRef. Implements Deref to the inner ArrayRef.
pub struct ArrowData(pub arrow::array::ArrayRef);
}

Data from Event::Input arrives as ArrowData. Use TryFrom conversions or Arrow APIs to extract typed values.

InputTracker

Helper for tracking input health and caching the last received value per input. Useful for graceful degradation when upstream nodes time out.

#![allow(unused)]
fn main() {
pub struct InputTracker { /* ... */ }

impl InputTracker {
    pub fn new() -> Self

    // Update state from an event. Returns true if the event was relevant.
    pub fn process_event(&mut self, event: &Event) -> bool

    // Current state of an input (Healthy or Closed), if tracked.
    pub fn state(&self, id: &DataId) -> Option<InputState>

    // True if the input is currently closed.
    pub fn is_closed(&self, id: &DataId) -> bool

    // Last received value for an input. Available even when closed.
    pub fn last_value(&self, id: &DataId) -> Option<&ArrowData>

    // All inputs currently in Closed state.
    pub fn closed_inputs(&self) -> Vec<&DataId>

    // True if any tracked input is closed.
    pub fn any_closed(&self) -> bool
}

pub enum InputState {
    Healthy,  // receiving data normally
    Closed,   // upstream exited or timed out
}
}

Integration Testing

The integration_testing module provides tools for testing nodes without a running daemon.

setup_integration_testing

Sets up thread-local state so that the next call to AdoraNode::init_from_env on the same thread initializes in test mode.

#![allow(unused)]
fn main() {
pub fn setup_integration_testing(
    input: TestingInput,
    output: TestingOutput,
    options: TestingOptions,
)
}

TestingInput

#![allow(unused)]
fn main() {
pub enum TestingInput {
    // Load events from a JSON file (must deserialize to IntegrationTestInput).
    FromJsonFile(PathBuf),

    // Provide events directly.
    Input(IntegrationTestInput),
}
}

TestingOutput

#![allow(unused)]
fn main() {
pub enum TestingOutput {
    // Write outputs to a JSONL file (created or overwritten).
    ToFile(PathBuf),

    // Write outputs as JSONL to any writer.
    ToWriter(Box<dyn std::io::Write + Send>),

    // Send each output as a JSON object to a flume channel.
    ToChannel(flume::Sender<serde_json::Map<String, serde_json::Value>>),
}
}

TestingOptions

#![allow(unused)]
fn main() {
#[derive(Debug, Clone, Default)]
pub struct TestingOptions {
    // Skip time offsets in outputs for deterministic comparison.
    pub skip_output_time_offsets: bool,
}
}

Environment Variable Testing

Nodes using init_from_env also support file-based testing via environment variables:

Operator API (adora-operator-api)

Operators are in-process components managed by the Adora runtime. They are compiled as shared libraries (.so/.dylib/.dll) and loaded by the runtime.

Add to your Cargo.toml:

[dependencies]
adora-operator-api = { workspace = true }

[lib]
crate-type = ["cdylib"]

AdoraOperator Trait

#![allow(unused)]
fn main() {
pub trait AdoraOperator: Default {
    fn on_event(
        &mut self,
        event: &Event,
        output_sender: &mut AdoraOutputSender,
    ) -> Result<AdoraStatus, String>;
}
}

Implement this trait to define your operator’s behavior. The runtime calls on_event for each incoming event. Return AdoraStatus to control execution flow.

Event (Operator)

The operator Event enum is simpler than the node Event and uses &str for IDs.

#![allow(unused)]
fn main() {
#[non_exhaustive]
pub enum Event<'a> {
    // An input was received.
    Input { id: &'a str, data: ArrowData },

    // Failed to parse the input data as an Arrow array.
    InputParseError { id: &'a str, error: String },

    // An input was closed by the sender.
    InputClosed { id: &'a str },

    // The operator should stop.
    Stop,
}
}

AdoraOutputSender

#![allow(unused)]
fn main() {
pub struct AdoraOutputSender<'a>(/* ... */);

impl AdoraOutputSender<'_> {
    // Send an output. `id` is the output ID from your dataflow YAML.
    pub fn send(&mut self, id: String, data: impl Array) -> Result<(), String>
}
}

AdoraStatus

Returned from on_event to control the operator lifecycle.

#![allow(unused)]
fn main() {
pub enum AdoraStatus {
    Continue,  // keep running, wait for the next event
    Stop,      // stop this operator
    StopAll,   // stop the entire dataflow
}
}

register_operator! Macro

Generates the FFI entry points required by the Adora runtime to load and call your operator.

#![allow(unused)]
fn main() {
use adora_operator_api::register_operator;

register_operator!(MyOperator);
}

This must be called exactly once per crate, at the top level, with the type that implements AdoraOperator.

Quick Start Example: Node

A minimal node that receives tick inputs and sends a random number as output.

use adora_node_api::{AdoraNode, Event, IntoArrow, adora_core::config::DataId};

fn main() -> eyre::Result<()> {
    let (mut node, mut events) = AdoraNode::init_from_env()?;

    let output = DataId::from("random".to_owned());

    while let Some(event) = events.recv() {
        match event {
            Event::Input { id, metadata, data } => {
                if id.as_str() == "tick" {
                    let value: u64 = fastrand::u64(..);
                    node.send_output(
                        output.clone(),
                        metadata.parameters,
                        value.into_arrow(),
                    )?;
                }
            }
            Event::Stop(_) => {}
            _ => {}
        }
    }

    Ok(())
}

Corresponding dataflow YAML:

nodes:
  - id: timer
    path: adora/timer/millis/100
    outputs:
      - tick

  - id: my-node
    path: ./target/debug/my-node
    inputs:
      tick: timer/tick
    outputs:
      - random

  - id: sink
    path: ./target/debug/sink
    inputs:
      data: my-node/random

Quick Start Example: Operator

A minimal operator that counts ticks and forwards formatted messages.

#![allow(unused)]
#![warn(unsafe_op_in_unsafe_fn)]

fn main() {
use adora_operator_api::{
    AdoraOperator, AdoraOutputSender, AdoraStatus, Event, IntoArrow, register_operator,
};

register_operator!(MyOperator);

#[derive(Debug, Default)]
struct MyOperator {
    ticks: usize,
}

impl AdoraOperator for MyOperator {
    fn on_event(
        &mut self,
        event: &Event,
        output_sender: &mut AdoraOutputSender,
    ) -> Result<AdoraStatus, String> {
        match event {
            Event::Input { id, data } => match *id {
                "tick" => {
                    self.ticks += 1;
                    let msg = format!("tick count: {}", self.ticks);
                    output_sender.send("status".into(), msg.into_arrow())?;
                }
                other => eprintln!("ignoring unexpected input {other}"),
            },
            Event::InputClosed { id } => {
                if *id == "tick" {
                    return Ok(AdoraStatus::Stop);
                }
            }
            Event::Stop => {}
            other => {
                eprintln!("received unknown event {other:?}");
            }
        }

        Ok(AdoraStatus::Continue)
    }
}
}

Corresponding dataflow YAML:

nodes:
  - id: timer
    path: adora/timer/millis/500
    outputs:
      - tick

  - id: runtime-node
    operator:
      shared_library: ./target/debug/libmy_operator
      inputs:
        tick: timer/tick
      outputs:
        - status

Python API Reference

This document covers the Python APIs for building adora nodes, operators, and dataflows. Install with:

pip install adora-rs

Table of Contents

• Node API
  • Node class
  • Event dictionary
  • AdoraStatus enum
• Operator API
  • Operator class (user-defined)
  • send_output callback
• DataflowBuilder
  • DataflowBuilder class
  • Node class (builder)
  • Output class (builder)
  • Operator class (builder)
• CUDA Module
• Quick Start Example
• DataflowBuilder Example

Node API

from adora import Node

The Node class is the primary interface for custom nodes. It connects to a running dataflow, receives input events, and sends outputs.

Node class

__init__(node_id=None)

Create a new node and connect to the running dataflow.

# Standard: node ID is read from environment variables set by the daemon
node = Node()

# Dynamic: connect to a running dataflow by explicit node ID
node = Node(node_id="my-dynamic-node")

Parameters:

• node_id (str, optional) – Explicit node ID for dynamic nodes. When omitted, the node reads its identity from environment variables set by the adora daemon.

Raises: RuntimeError if the node cannot connect to the dataflow.

next(timeout=None)

Retrieve the next event from the event stream. Blocks until an event is available or the timeout expires.

event = node.next()              # block indefinitely
event = node.next(timeout=2.0)   # block up to 2 seconds

Parameters:

• timeout (float, optional) – Maximum wait time in seconds.

Returns: dict – An event dictionary, or None if all senders have been dropped or the timeout expired.

drain()

Retrieve all buffered events without blocking.

events = node.drain()
for event in events:
    print(event["type"])

Returns: list[dict] – A list of event dictionaries. Returns an empty list if no events are buffered.

try_recv()

Non-blocking receive. Returns the next buffered event if one is available.

event = node.try_recv()
if event is not None:
    print(event["type"])

Returns: dict | None – An event dictionary, or None if no event is buffered.

recv_async(timeout=None)

Asynchronous receive. For use with asyncio.

event = await node.recv_async()
event = await node.recv_async(timeout=5.0)

Parameters:

• timeout (float, optional) – Maximum wait time in seconds. Returns an error if the timeout is reached.

Returns: dict | None – An event dictionary, or None if all senders have been dropped.

Note: This method is experimental. The pyo3 async (Rust-Python FFI) integration is still in development.

is_empty()

Check whether there are any buffered events in the event stream.

if not node.is_empty():
    event = node.try_recv()

Returns: bool

send_output(output_id, data, metadata=None)

Send data on an output channel.

import pyarrow as pa

# Send raw bytes
node.send_output("status", b"OK")

# Send an Apache Arrow array (zero-copy capable)
node.send_output("values", pa.array([1, 2, 3]))

# Send with metadata
node.send_output("image", pa.array(pixels), {"camera_id": "front"})

Parameters:

• output_id (str) – The output name as declared in the dataflow YAML.
• data (bytes | pyarrow.Array) – The payload. Use bytes for simple data or pyarrow.Array for zero-copy shared-memory transport.
• metadata (dict, optional) – Key-value pairs attached to the message. Supported value types: bool, int, float, str, list[int], list[float], list[str], datetime.datetime.

Raises: RuntimeError if data is neither bytes nor a pyarrow.Array.

Service, action, and streaming patterns

Python nodes use the same metadata key conventions as Rust for communication patterns. Parameters are plain dicts with string keys.

Well-known metadata keys:

Service client example:

import uuid

# Send a request with a unique request_id
request_id = str(uuid.uuid7())  # Python 3.13+; use uuid_utils or uuid.uuid4() on older versions
node.send_output("request", data, {"request_id": request_id})

Service server example:

# Pass through the metadata (includes request_id) from the incoming request
node.send_output("response", result, event["metadata"])

Action client example:

goal_id = str(uuid.uuid7())
node.send_output("goal", data, {"goal_id": goal_id})

Streaming example (flush downstream queues on user interruption):

params = {
    "session_id": session_id,
    "segment_id": 1,
    "seq": 0,
    "fin": False,
    "flush": True,
}
node.send_output("text", data, metadata={"parameters": params})

See patterns.md for the full guide.

Logging

Python nodes can log using either Python’s built-in logging module (recommended) or the explicit node API.

Python logging module (auto-bridged):

When Node() is created, it automatically installs a handler that routes Python’s logging module through the adora daemon. No configuration needed:

import logging
from adora import Node

node = Node()  # Installs the logging bridge

logging.info("Sensor initialized")       # -> structured "info" log entry
logging.warning("High temperature")      # -> structured "warn" log entry
logging.debug("Raw bytes: %s", data)     # -> structured "debug" log entry

These log entries are captured with full metadata (level, message, file path, line number) and work with min_log_level filtering, send_logs_as routing, and adora/logs subscribers.

Note: Do not call logging.basicConfig() before creating Node(). The constructor sets up the bridge; calling basicConfig() first may install a conflicting handler.

Explicit node API:

log(level, message, target=None, fields=None)

Emit a structured log message with optional target and key-value fields.

node.log("info", "Processing frame", target="vision")
node.log("error", "Sensor timeout", fields={"sensor": "lidar", "retry": "3"})

Parameters:

• level (str) – Log level: "error", "warn", "info", "debug", or "trace".
• message (str) – The log message.
• target (str, optional) – Target module or subsystem name.
• fields (dict[str, str], optional) – Structured key-value context fields.

Works with the daemon’s min_log_level filtering, send_logs_as routing, and adora/logs subscribers.

log_error(message), log_warn(message), log_info(message), log_debug(message), log_trace(message)

Convenience methods for common log levels:

node.log_error("Connection failed")
node.log_warn("Temperature elevated")
node.log_info("Sensor initialized")
node.log_debug("Raw bytes received")
node.log_trace("Entering loop iteration")

Each is equivalent to node.log(level, message).

When to use which:

dataflow_descriptor()

Return the full dataflow descriptor (the parsed dataflow YAML) as a Python dictionary.

descriptor = node.dataflow_descriptor()
print(descriptor["nodes"])

Returns: dict

node_config()

Return the configuration block for this node from the dataflow descriptor.

config = node.node_config()
model_path = config.get("model", "default.pt")

Returns: dict

dataflow_id()

Return the unique identifier of the running dataflow.

print(node.dataflow_id())  # e.g. "a1b2c3d4-..."

Returns: str

is_restart()

Check whether this node was restarted after a previous exit or failure. Useful for deciding whether to restore saved state or start fresh.

if node.is_restart():
    restore_checkpoint()

Returns: bool

restart_count()

Return how many times this node has been restarted. Returns 0 on the first run, 1 after the first restart, and so on.

print(f"Restart #{node.restart_count()}")

Returns: int

merge_external_events(subscription)

Merge a ROS2 subscription stream into the node’s main event loop. After calling this method, ROS2 messages arrive as events with kind set to "external".

from adora import Node, Ros2Context, Ros2Node, Ros2NodeOptions, Ros2Topic

node = Node()
ros2_context = Ros2Context()
ros2_node = ros2_context.new_node("listener", Ros2NodeOptions())
topic = Ros2Topic("/chatter", "std_msgs/String", ros2_node)
subscription = ros2_node.create_subscription(topic)

node.merge_external_events(subscription)

for event in node:
    if event["kind"] == "external":
        print("ROS2:", event["value"])
    elif event["type"] == "INPUT":
        print("Adora:", event["id"])

Parameters:

• subscription (adora.Ros2Subscription) – A ROS2 subscription created via the adora ROS2 bridge.

Iteration support

The Node class implements __iter__ and __next__, so you can iterate directly:

for event in node:
    match event["type"]:
        case "INPUT":
            process(event["value"])
        case "STOP":
            break

The iterator calls next() with no timeout on each iteration. It yields None when the event stream is closed, which terminates the loop.

Event dictionary

Events are returned as plain Python dictionaries. The structure depends on the event type.

INPUT

An input message arrived from another node.

{
    "type": "INPUT",
    "id": "camera_image",          # input ID as declared in the dataflow YAML
    "kind": "adora",               # "adora" for dataflow events, "external" for ROS2
    "value": <pyarrow.Array>,      # the payload as an Apache Arrow array
    "metadata": {
        "timestamp": datetime,     # UTC-aware datetime.datetime
        "open_telemetry_context": "...",  # tracing context (if enabled)
        ...                        # any user-supplied metadata
    },
}

Access the data:

values = event["value"].to_pylist()     # convert to Python list
array = event["value"].to_numpy()       # convert to NumPy array

INPUT_CLOSED

An input channel was closed (the upstream node finished).

{
    "type": "INPUT_CLOSED",
    "id": "camera_image",
    "kind": "adora",
}

STOP

The dataflow is shutting down.

{
    "type": "STOP",
    "id": "MANUAL" | "ALL_INPUTS_CLOSED",   # stop cause
    "kind": "adora",
}

ERROR

An error occurred in the runtime.

{
    "type": "ERROR",
    "error": "description of the error",
    "kind": "adora",
}

External (ROS2)

When using merge_external_events, ROS2 messages arrive as:

{
    "kind": "external",
    "value": <pyarrow.Array>,   # the ROS2 message as an Arrow array
}

AdoraStatus enum

Used as the return value from operator on_event methods to control the event loop.

from adora import AdoraStatus

Operator API

Operators run inside the adora runtime process (no separate OS process). They are defined as a Python class named Operator with an on_event method.

Operator class (user-defined)

Create a Python file with an Operator class:

from adora import AdoraStatus

class Operator:
    def __init__(self):
        # Initialize state here
        self.count = 0

    def on_event(self, adora_event, send_output) -> AdoraStatus:
        if adora_event["type"] == "INPUT":
            self.count += 1
            # Process the input and optionally send output
            send_output("result", b"processed", adora_event["metadata"])
        return AdoraStatus.CONTINUE

Methods:

• __init__(self) – Called once when the operator is loaded. Initialize any state or models here.
• on_event(self, adora_event, send_output) -> AdoraStatus – Called for every incoming event. Must return an AdoraStatus value.

Parameters of on_event:

• adora_event (dict) – An event dictionary.
• send_output (callable) – Callback to send output data (see below).

The runtime also sets self.dataflow_descriptor on the operator instance with the parsed dataflow YAML as a dictionary.

send_output callback

The send_output callback is passed to on_event for sending data from an operator.

send_output(output_id, data, metadata=None)

Parameters:

• output_id (str) – The output name as declared in the dataflow YAML.
• data (bytes | pyarrow.Array) – The payload.
• metadata (dict, optional) – Metadata to attach. Pass adora_event["metadata"] to propagate tracing context.

Example:

import pyarrow as pa
from adora import AdoraStatus

class Operator:
    def on_event(self, adora_event, send_output) -> AdoraStatus:
        if adora_event["type"] == "INPUT":
            result = pa.array([42], type=pa.int64())
            send_output("output", result, adora_event["metadata"])
        return AdoraStatus.CONTINUE

DataflowBuilder

from adora.builder import DataflowBuilder, Node, Operator, Output

Build dataflow YAML programmatically in Python.

DataflowBuilder class

__init__(name="adora-dataflow")

Create a new dataflow builder.

flow = DataflowBuilder("my-robot")

Parameters:

• name (str, optional) – Name of the dataflow. Defaults to "adora-dataflow".

add_node(id, **kwargs) -> Node

Add a node to the dataflow. Returns a Node object for further configuration.

sender = flow.add_node("sender")

Parameters:

• id (str) – Unique node identifier.
• **kwargs – Additional node configuration passed through to the YAML.

Returns: Node (builder)

to_yaml(path=None) -> str | None

Generate the YAML representation of the dataflow. If path is given, writes to file and returns None. Otherwise returns the YAML string.

# Write to file
flow.to_yaml("dataflow.yml")

# Get as string
yaml_str = flow.to_yaml()

Parameters:

• path (str, optional) – File path to write the YAML.

Returns: str | None

Context manager

DataflowBuilder supports the with statement:

with DataflowBuilder("my-flow") as flow:
    flow.add_node("sender").path("sender.py")
    flow.to_yaml("dataflow.yml")

Node class (builder)

Returned by DataflowBuilder.add_node(). All setter methods return self for chaining.

path(path) -> Node

Set the path to the node’s executable or script.

node.path("my_node.py")

args(args) -> Node

Set command-line arguments for the node.

node.args("--verbose --port 8080")

env(env) -> Node

Set environment variables for the node.

node.env({"MODEL_PATH": "/models/yolo.pt"})

build(command) -> Node

Set the build command for the node (run before starting).

node.build("pip install -r requirements.txt")

git(url, branch=None, tag=None, rev=None) -> Node

Set a Git repository as the source for the node.

node.git("https://github.com/org/repo.git", branch="main")

add_operator(operator) -> Node

Attach an Operator to this node.

op = Operator("detector", python="object_detection.py")
node.add_operator(op)

add_output(output_id) -> Output

Declare an output on this node and return an Output reference for use as an input source.

output = sender.add_output("data")

add_input(input_id, source, queue_size=None, queue_policy=None) -> Node

Subscribe this node to an output from another node.

# Using an Output object
output = sender.add_output("data")
receiver.add_input("data", output)

# Using a string reference
receiver.add_input("tick", "adora/timer/millis/100")

# With a custom queue size
receiver.add_input("images", camera_output, queue_size=2)

# Lossless input (blocks sender when full)
receiver.add_input("commands", cmd_output, queue_size=100, queue_policy="backpressure")

Parameters:

• input_id (str) – Name of the input on this node.
• source (str | Output) – Either a string ("node_id/output_id") or an Output object.
• queue_size (int, optional) – Maximum number of buffered messages for this input.
• queue_policy (str, optional) – "drop_oldest" (default) or "backpressure" (buffers up to 10x queue_size before dropping).

to_dict() -> dict

Return the dictionary representation of the node for YAML serialization.

Output class (builder)

Returned by Node.add_output(). Represents a reference to a node’s output, used as a source in add_input().

output = sender.add_output("data")
receiver.add_input("sensor_data", output)
str(output)  # "sender/data"

Operator class (builder)

Defines an operator for embedding in a node’s YAML configuration.

__init__(id, name=None, description=None, build=None, python=None, shared_library=None, send_stdout_as=None)

op = Operator(
    id="detector",
    python="object_detection.py",
    send_stdout_as="detection_text",
)

Parameters:

• id (str) – Unique operator identifier.
• name (str, optional) – Display name.
• description (str, optional) – Human-readable description.
• build (str, optional) – Build command to run before loading.
• python (str, optional) – Path to the Python operator file.
• shared_library (str, optional) – Path to a shared library operator.
• send_stdout_as (str, optional) – Route the operator’s stdout as an output with this ID.

to_dict() -> dict

Return the dictionary representation for YAML serialization.

CUDA Module

from adora.cuda import torch_to_ipc_buffer, ipc_buffer_to_ipc_handle, open_ipc_handle

Utilities for zero-copy GPU tensor sharing between nodes via CUDA IPC. Requires PyTorch with CUDA and Numba with CUDA support.

torch_to_ipc_buffer(tensor) -> tuple[pyarrow.Array, dict]

Convert a PyTorch CUDA tensor into an Arrow array containing the CUDA IPC handle, plus a metadata dictionary. Send both through the dataflow to share GPU memory without copying.

import torch
import pyarrow as pa
from adora import Node
from adora.cuda import torch_to_ipc_buffer

node = Node()
tensor = torch.randn(1024, 768, device="cuda")
ipc_buffer, metadata = torch_to_ipc_buffer(tensor)
node.send_output("gpu_data", ipc_buffer, metadata)

Parameters:

• tensor (torch.Tensor) – A CUDA tensor.

Returns: tuple[pyarrow.Array, dict] – The IPC handle as an int8 Arrow array, and metadata with shape, strides, dtype, size, offset, and source info.

ipc_buffer_to_ipc_handle(handle_buffer, metadata) -> IpcHandle

Reconstruct a CUDA IPC handle from a received Arrow buffer and metadata.

from adora.cuda import ipc_buffer_to_ipc_handle

event = node.next()
ipc_handle = ipc_buffer_to_ipc_handle(event["value"], event["metadata"])

Parameters:

• handle_buffer (pyarrow.Array) – The Arrow array from event["value"].
• metadata (dict) – The metadata from event["metadata"].

Returns: numba.cuda.cudadrv.driver.IpcHandle

open_ipc_handle(ipc_handle, metadata) -> ContextManager[torch.Tensor]

Open a CUDA IPC handle and yield a PyTorch tensor. Use as a context manager to ensure proper cleanup.

from adora.cuda import ipc_buffer_to_ipc_handle, open_ipc_handle

event = node.next()
ipc_handle = ipc_buffer_to_ipc_handle(event["value"], event["metadata"])

with open_ipc_handle(ipc_handle, event["metadata"]) as tensor:
    result = tensor * 2  # use the GPU tensor directly

Parameters:

• ipc_handle (IpcHandle) – Handle from ipc_buffer_to_ipc_handle.
• metadata (dict) – The metadata dictionary with shape, strides, and dtype info.

Returns: Context manager yielding a torch.Tensor on CUDA.

Quick Start Example

A complete node that receives images, processes them, and sends results:

#!/usr/bin/env python3
"""Example node: receives messages, transforms them, and sends output."""

import logging

import pyarrow as pa
from adora import Node


def main():
    node = Node()

    for event in node:
        if event["type"] == "INPUT":
            input_id = event["id"]

            if input_id == "message":
                values = event["value"].to_pylist()
                number = values[0]

                # Create a struct array with multiple fields
                result = pa.StructArray.from_arrays(
                    [
                        pa.array([number * 2]),
                        pa.array([f"Message #{number}"]),
                    ],
                    names=["doubled", "description"],
                )
                node.send_output("transformed", result)
                logging.info("Transformed message %d", number)

        elif event["type"] == "STOP":
            logging.info("Node stopping")
            break


if __name__ == "__main__":
    main()

Run with:

adora run dataflow.yml

DataflowBuilder Example

Build a dataflow programmatically instead of writing YAML by hand:

#!/usr/bin/env python3
"""Build a simple sender -> receiver dataflow."""

from adora.builder import DataflowBuilder, Operator

flow = DataflowBuilder("example-flow")

# Add a timer-driven sender node
sender = flow.add_node("sender")
sender.path("sender.py")
tick_output = sender.add_output("message")

# Add a receiver that subscribes to the sender
receiver = flow.add_node("receiver")
receiver.path("receiver.py")
receiver.add_input("message", tick_output)

# Add a node with a timer input
timed_node = flow.add_node("periodic")
timed_node.path("periodic.py")
timed_node.add_input("tick", "adora/timer/millis/100")

# Add a node with an operator
runtime_node = flow.add_node("runtime-node")
op = Operator("detector", python="object_detection.py")
runtime_node.add_operator(op)
runtime_node.add_input("image", "camera/image")

# Write or print the YAML
flow.to_yaml("dataflow.yml")
print(flow.to_yaml())

C API Reference

This document covers the two C APIs provided by the Adora framework: the Node API for standalone C processes and the Operator API for shared-library operators loaded by the Adora runtime.

Table of Contents

• Node API (adora-node-api-c)
  • Initialization
  • Event Loop
  • Event Inspection
  • Output
  • Logging
  • Enums
• Operator API (adora-operator-api-c)
  • Lifecycle Functions
  • Event Handling
  • Input Reading
  • Output Sending
  • Memory Management
  • Structs
  • Enums
• Node Example
• Operator Example
• Building and Linking

Node API (adora-node-api-c)

Header: apis/c/node/node_api.h Crate: adora-node-api-c (builds as staticlib)

The Node API is used by standalone C executables that participate in an Adora dataflow as external processes. The daemon spawns the process and sets environment variables that the node reads during initialization.

Initialization

init_adora_context_from_env

void *init_adora_context_from_env();

Initializes an Adora node context from environment variables set by the daemon. Returns an opaque pointer to the context on success, or NULL on failure.

The returned pointer must be passed to all subsequent Node API calls that expect a context argument. When the node is finished, free it with free_adora_context.

free_adora_context

void free_adora_context(void *adora_context);

Frees a context previously created by init_adora_context_from_env. Each context must be freed exactly once. After freeing, the pointer must not be used again.

Event Loop

adora_next_event

void *adora_next_event(void *adora_context);

Blocks until the next event is available for this node. Returns an opaque pointer to the event, or NULL when all event streams have closed (indicating the node should exit).

The returned pointer must not be dereferenced directly. Use the read_adora_* functions to extract the event type and payload. Free the event with free_adora_event when done.

free_adora_event

void free_adora_event(void *adora_event);

Frees an event previously returned by adora_next_event. Each event must be freed exactly once. After freeing, the event pointer and all derived pointers (from read_adora_input_id, read_adora_input_data) become invalid.

Event Inspection

read_adora_event_type

enum AdoraEventType read_adora_event_type(void *adora_event);

Returns the type of the given event. See AdoraEventType for possible values.

read_adora_input_id

void read_adora_input_id(void *adora_event, char **out_ptr, size_t *out_len);

Reads the input ID from an AdoraEventType_Input event. Writes the string start pointer to *out_ptr and its byte length to *out_len. The string is valid UTF-8 but not null-terminated; use out_len to determine its bounds.

If the event is not an input event, sets *out_ptr = NULL and *out_len = 0.

The returned pointer borrows from the event. It becomes invalid after free_adora_event is called.

read_adora_input_data

void read_adora_input_data(void *adora_event, char **out_ptr, size_t *out_len);

Reads the raw data bytes from an AdoraEventType_Input event. Writes the data start pointer to *out_ptr and its byte length to *out_len.

Sets *out_ptr = NULL and *out_len = 0 if the event is not an input event or the input carries no data.

Currently only UInt8 Arrow arrays are supported. Other Arrow data types will cause a runtime panic. Future versions will use the Arrow C Data Interface for full type support.

The returned pointer borrows from the event. It becomes invalid after free_adora_event is called.

read_adora_input_timestamp

unsigned long long read_adora_input_timestamp(void *adora_event);

Returns the hybrid logical clock timestamp from an input event’s metadata as a uint64 value. Returns 0 if the event is not an input event.

Output

adora_send_output

int adora_send_output(
    void *adora_context,
    const char *id_ptr,
    size_t id_len,
    const char *data_ptr,
    size_t data_len
);

Sends output data to all downstream subscribers. The output ID (id_ptr/id_len) must be a valid UTF-8 string matching one of the node’s declared outputs in the dataflow YAML. The data (data_ptr/data_len) is sent as raw bytes (UInt8 Arrow array).

Returns 0 on success, -1 on error. Errors are logged via tracing.

Returns -1 immediately if any pointer argument is NULL.

Logging

adora_log

int adora_log(
    void *adora_context,
    const char *level_ptr,
    size_t level_len,
    const char *msg_ptr,
    size_t msg_len
);

Sends a structured log message through the Adora logging pipeline. Both level and msg must be valid UTF-8 strings.

Valid log levels: "error", "warn", "info", "debug", "trace".

Returns 0 on success, -1 on error. Returns -1 immediately if any pointer argument is NULL.

Enums

AdoraEventType

enum AdoraEventType {
    AdoraEventType_Stop,        // Graceful shutdown requested
    AdoraEventType_Input,       // New input data available
    AdoraEventType_InputClosed, // An input stream was closed
    AdoraEventType_Error,       // An error occurred
    AdoraEventType_Unknown,     // Unrecognized event type
};

Operator API (adora-operator-api-c)

Headers: apis/c/operator/operator_api.h, apis/c/operator/operator_types.h Crate: adora-operator-api-c

The Operator API is used by shared libraries (.so/.dylib/.dll) loaded into the Adora runtime process. Unlike nodes, operators do not have their own main function. Instead, they export three functions that the runtime calls at the appropriate lifecycle points.

The operator_types.h header is auto-generated by safer-ffi and defines all C-compatible struct and enum types.

Lifecycle Functions

adora_init_operator

AdoraInitResult_t adora_init_operator(void);

Called once when the runtime loads the operator. Allocate and initialize any operator state, then return it via the operator_context field. The runtime passes this pointer back on every subsequent call.

Return an AdoraInitResult_t with .result.error = NULL on success.

adora_drop_operator

AdoraResult_t adora_drop_operator(void *operator_context);

Called once when the operator is being unloaded. Free all resources associated with operator_context.

Return an AdoraResult_t with .error = NULL on success.

Event Handling

adora_on_event

OnEventResult_t adora_on_event(
    RawEvent_t *event,
    const SendOutput_t *send_output,
    void *operator_context
);

Called by the runtime each time an event arrives for this operator. Inspect the event fields to determine the event type:

Use send_output to emit data to downstream nodes (see adora_send_operator_output). Return an OnEventResult_t with the appropriate AdoraStatus_t to control the operator lifecycle.

Input Reading

adora_read_input_id

char *adora_read_input_id(const Input_t *input);

Returns a newly allocated null-terminated string containing the input ID. The caller must free it with adora_free_input_id.

adora_read_data

Vec_uint8_t adora_read_data(Input_t *input);

Reads the input data as a byte array. Consumes the underlying Arrow array from the input (the data can only be read once per event). Returns a Vec_uint8_t with .ptr = NULL if the input has no data or the data has already been consumed.

The caller must free the returned data with adora_free_data.

Output Sending

adora_send_operator_output

AdoraResult_t adora_send_operator_output(
    const SendOutput_t *send_output,
    const char *id,
    const uint8_t *data_ptr,
    size_t data_len
);

Sends output data to downstream subscribers. The id must be a null-terminated string matching one of the operator’s declared outputs. The data (data_ptr/data_len) is converted to a UInt8 Arrow array internally.

Returns an AdoraResult_t with .error = NULL on success.

Memory Management

The Operator API allocates memory that the caller must free using the corresponding functions:

void adora_free_input_id(char *input_id);
void adora_free_data(Vec_uint8_t data);

Failing to call these functions will leak memory. Do not use free() on these allocations – they are allocated by the Rust runtime and must be freed through the API.

Structs

Vec_uint8_t

typedef struct Vec_uint8 {
    uint8_t *ptr;
    size_t len;
    size_t cap;
} Vec_uint8_t;

A Rust-allocated byte vector. Access len bytes starting at ptr. Do not modify cap. Free with adora_free_data.

AdoraResult_t

typedef struct AdoraResult {
    Vec_uint8_t *error;  // NULL on success, points to error string on failure
} AdoraResult_t;

Generic result type. A NULL error pointer indicates success. When non-NULL, the error pointer contains a UTF-8 error message.

AdoraInitResult_t

typedef struct AdoraInitResult {
    AdoraResult_t result;
    void *operator_context;  // opaque pointer to operator state
} AdoraInitResult_t;

Returned by adora_init_operator. On success, result.error is NULL and operator_context holds the operator state pointer.

OnEventResult_t

typedef struct OnEventResult {
    AdoraResult_t result;
    AdoraStatus_t status;
} OnEventResult_t;

Returned by adora_on_event. Contains both an error/success result and a status code controlling the operator lifecycle.

RawEvent_t

typedef struct RawEvent {
    Input_t *input;           // non-NULL when this is an input event
    Vec_uint8_t input_closed; // non-empty when an input stream closed
    bool stop;                // true when shutdown is requested
    Vec_uint8_t error;        // non-empty on error
} RawEvent_t;

Represents an event delivered to the operator. Multiple fields may be set simultaneously; check them in order of priority.

Input_t

typedef struct Input Input_t;  // opaque

Opaque type representing an input event’s data. Use adora_read_input_id and adora_read_data to extract its contents.

Output_t

typedef struct Output Output_t;  // opaque

Opaque type used internally by adora_send_operator_output. Not created directly by user code.

SendOutput_t

typedef struct SendOutput {
    ArcDynFn1_AdoraResult_Output_t send_output;
} SendOutput_t;

Callback handle passed to adora_on_event. Pass it to adora_send_operator_output to emit data. Do not store it beyond the scope of the current adora_on_event call.

Metadata_t

typedef struct Metadata {
    Vec_uint8_t open_telemetry_context;
} Metadata_t;

Event metadata containing an OpenTelemetry trace context string.

Operator Enums

AdoraStatus_t

enum AdoraStatus {
    ADORA_STATUS_CONTINUE = 0,  // Keep running
    ADORA_STATUS_STOP     = 1,  // Stop this operator
    ADORA_STATUS_STOP_ALL = 2,  // Stop the entire dataflow
};
typedef uint8_t AdoraStatus_t;

Returned in OnEventResult_t to control operator lifecycle after processing an event.

Node Example

A complete C node that receives timer ticks and sends output messages:

#include <stdio.h>
#include <string.h>
#include "node_api.h"

int main() {
    void *ctx = init_adora_context_from_env();
    if (ctx == NULL) {
        fprintf(stderr, "failed to init adora context\n");
        return 1;
    }

    for (int i = 0; i < 100; i++) {
        void *event = adora_next_event(ctx);
        if (event == NULL)
            break;  // all streams closed

        enum AdoraEventType ty = read_adora_event_type(event);

        if (ty == AdoraEventType_Input) {
            char *id;
            size_t id_len;
            read_adora_input_id(event, &id, &id_len);

            // Send a response
            char out_id[] = "message";
            char out_data[64];
            int out_len = snprintf(out_data, sizeof(out_data),
                                   "iteration %d", i);

            adora_send_output(ctx, out_id, strlen(out_id),
                              out_data, out_len);
        } else if (ty == AdoraEventType_Stop) {
            free_adora_event(event);
            break;
        }

        free_adora_event(event);
    }

    free_adora_context(ctx);
    return 0;
}

Dataflow YAML for the node:

nodes:
  - id: c_node
    path: build/c_node
    inputs:
      timer: adora/timer/millis/100
    outputs:
      - message

Operator Example

A complete C operator that reads input, maintains state, and sends output:

#include "operator_api.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

AdoraInitResult_t adora_init_operator(void) {
    // Allocate operator state (a simple counter)
    int *counter = (int *)calloc(1, sizeof(int));

    AdoraInitResult_t result = {.operator_context = counter};
    return result;
}

AdoraResult_t adora_drop_operator(void *operator_context) {
    free(operator_context);
    AdoraResult_t result = {.error = NULL};
    return result;
}

OnEventResult_t adora_on_event(
    RawEvent_t *event,
    const SendOutput_t *send_output,
    void *operator_context)
{
    OnEventResult_t result = {.status = ADORA_STATUS_CONTINUE};
    int *counter = (int *)operator_context;

    if (event->input != NULL) {
        char *id = adora_read_input_id(event->input);
        Vec_uint8_t data = adora_read_data(event->input);

        if (data.ptr != NULL) {
            *counter += 1;
            printf("received input '%s', counter: %d\n", id, *counter);

            // Send counter value as string
            char buf[64];
            int len = snprintf(buf, sizeof(buf), "count=%d", *counter);
            result.result = adora_send_operator_output(
                send_output, "counter", (uint8_t *)buf, len);

            adora_free_data(data);
        }

        adora_free_input_id(id);
    }

    if (event->stop) {
        result.status = ADORA_STATUS_STOP;
    }

    return result;
}

Dataflow YAML for the operator:

nodes:
  - id: runtime-node
    operators:
      - id: c_operator
        shared-library: build/operator
        inputs:
          data: source_node/output
        outputs:
          - counter

Building and Linking

Node (static library)

C nodes link against adora-node-api-c, which builds as a static library.

Step 1: Build the static library

cargo build -p adora-node-api-c --release

This produces target/release/libadora_node_api_c.a (or .lib on Windows).

Step 2: Compile and link

clang node.c -ladora_node_api_c -L ../../target/release -o build/c_node <FLAGS>

Platform-specific linker flags:

On Windows, add the .exe extension to the output file.

Operator (shared library)

C operators are compiled into shared libraries that the Adora runtime loads at startup.

Step 1: Compile to object file

clang -c operator.c -o build/operator.o -fdeclspec -fPIC

Omit -fPIC on Windows.

Step 2: Link as shared library

# Linux
clang -shared build/operator.o -o build/liboperator.so

# macOS
clang -shared build/operator.o -o build/liboperator.dylib

# Windows
clang -shared build/operator.o -o build/operator.dll

Step 3: Reference in dataflow YAML

operators:
  - id: c_operator
    shared-library: build/operator   # without lib prefix or extension
    inputs:
      data: source/output
    outputs:
      - result

The shared-library path omits the platform-specific prefix (lib) and extension (.so/.dylib/.dll). The runtime resolves the correct file for the current platform.

Include Paths

The Node API header is at apis/c/node/node_api.h. The Operator API headers are at apis/c/operator/operator_api.h and apis/c/operator/operator_types.h. Adjust your include paths accordingly:

# Node
clang -I path/to/adora/apis/c/node node.c ...

# Operator
clang -I path/to/adora/apis/c/operator operator.c ...

C++ Compatibility

Both headers include extern "C" guards (in the operator headers) or use C-compatible declarations (in the node header), so they can be included directly from C++ source files.

C++ API Reference

Adora provides C++ bindings for both standalone nodes and in-process operators via CXX (Rust-C++ interop). The CXX bridge generates type-safe C++ headers from Rust definitions – no raw FFI or manual extern "C" declarations are needed.

Two crates provide the C++ surface:

Generated headers: adora-node-api.h and adora-operator-api.h.

Node API (adora-node-api-cxx)

Initialization

#include "adora-node-api.h"

// Initialize a node from environment variables set by the Adora daemon.
// Returns an AdoraNode struct containing the event stream and output sender.
// Throws on failure.
AdoraNode init_adora_node();

AdoraNode

Returned by init_adora_node(). Owns the event stream and the output sender for the lifetime of the node.

struct AdoraNode {
    rust::Box<Events>        events;       // event stream (blocking receiver)
    rust::Box<OutputSender>  send_output;  // output sender
};

Events

Opaque Rust type exposed to C++. Provides blocking iteration over the node’s incoming events.

// Member function -- call on the boxed object directly.
rust::Box<AdoraEvent> Events::next();

// Free function form -- equivalent to events->next().
rust::Box<AdoraEvent> next_event(rust::Box<Events>& events);

Both forms block until the next event arrives and return an owned AdoraEvent.

AdoraEvent

Opaque Rust type. Inspect its kind with event_type(), then downcast with event_as_input() or event_as_arrow_input().

// Determine the event kind.
AdoraEventType event_type(const rust::Box<AdoraEvent>& event);

// Downcast to a raw-byte input. Throws if the event is not Input.
AdoraInput event_as_input(rust::Box<AdoraEvent> event);

// Downcast to an Arrow FFI input (writes Arrow C Data Interface structs).
// out_array and out_schema must point to valid ArrowArray / ArrowSchema structs.
// Returns AdoraResult with empty error on success.
AdoraResult event_as_arrow_input(
    rust::Box<AdoraEvent> event,
    uint8_t* out_array,
    uint8_t* out_schema);

// Same as above, but also returns the input ID and metadata.
ArrowInputInfo event_as_arrow_input_with_info(
    rust::Box<AdoraEvent> event,
    uint8_t* out_array,
    uint8_t* out_schema);

AdoraEventType

enum class AdoraEventType : uint8_t {
    Stop,             // graceful shutdown requested
    Input,            // new data arrived on an input
    InputClosed,      // a single input was closed
    Error,            // an error occurred
    Unknown,          // unrecognized event variant
    AllInputsClosed,  // all inputs closed (stream ended)
};

AdoraInput

Returned by event_as_input(). Contains raw bytes.

struct AdoraInput {
    rust::String     id;    // input identifier (e.g. "tick", "image")
    rust::Vec<uint8_t> data;  // raw payload bytes
};

ArrowInputInfo

Returned by event_as_arrow_input_with_info(). Contains the input ID, metadata, and an error string.

struct ArrowInputInfo {
    rust::String       id;        // input identifier
    rust::Box<Metadata> metadata; // attached metadata
    rust::String       error;     // empty on success
};

AdoraResult

Returned by output-sending functions. Check the error field – empty means success.

struct AdoraResult {
    rust::String error;  // empty string on success
};

OutputSender

Opaque Rust type. All methods take rust::Box<OutputSender>& as the first argument (the sender from AdoraNode::send_output).

send_output

Send raw bytes on a named output.

AdoraResult send_output(
    rust::Box<OutputSender>& sender,
    rust::String id,
    rust::Slice<const uint8_t> data);

send_output_with_metadata

Send raw bytes with attached metadata.

AdoraResult send_output_with_metadata(
    rust::Box<OutputSender>& sender,
    rust::String id,
    rust::Slice<const uint8_t> data,
    rust::Box<Metadata> metadata);

send_arrow_output

Send an Arrow array via the C Data Interface. The pointers must reference valid ArrowArray and ArrowSchema structs. Ownership of the Arrow data transfers to Rust on success.

AdoraResult send_arrow_output(
    rust::Box<OutputSender>& sender,
    rust::String id,
    uint8_t* array_ptr,
    uint8_t* schema_ptr);

// Overload with metadata (same C++ name via cxx_name attribute).
AdoraResult send_arrow_output(
    rust::Box<OutputSender>& sender,
    rust::String id,
    uint8_t* array_ptr,
    uint8_t* schema_ptr,
    rust::Box<Metadata> metadata);

log_message

Send a log message through the Adora logging system.

AdoraResult log_message(
    const rust::Box<OutputSender>& sender,
    rust::String level,    // e.g. "info", "warn", "error"
    rust::String message);

Metadata

Opaque Rust type for attaching typed key-value pairs to outputs.

Construction

rust::Box<Metadata> new_metadata();

Reading

uint64_t     Metadata::timestamp() const;

bool         Metadata::get_bool(const rust::Str key) const;        // throws on missing/wrong type
int64_t      Metadata::get_int(const rust::Str key) const;
double       Metadata::get_float(const rust::Str key) const;
rust::String Metadata::get_str(const rust::Str key) const;

rust::Vec<int64_t>      Metadata::get_list_int(const rust::Str key) const;
rust::Vec<double>       Metadata::get_list_float(const rust::Str key) const;
rust::Vec<rust::String> Metadata::get_list_string(const rust::Str key) const;

int64_t      Metadata::get_timestamp(const rust::Str key) const;   // nanoseconds since epoch
rust::String Metadata::get_json(const rust::Str key) const;        // single value as JSON string

Writing

All setters throw on failure.

void Metadata::set_bool(const rust::Str key, bool value);
void Metadata::set_int(const rust::Str key, int64_t value);
void Metadata::set_float(const rust::Str key, double value);
void Metadata::set_string(const rust::Str key, rust::String value);

void Metadata::set_list_int(const rust::Str key, rust::Vec<int64_t> value);
void Metadata::set_list_float(const rust::Str key, rust::Vec<double> value);
void Metadata::set_list_string(const rust::Str key, rust::Vec<rust::String> value);

void Metadata::set_timestamp(const rust::Str key, int64_t nanos);  // nanoseconds since epoch

Introspection

MetadataValueType Metadata::type(const rust::Str key) const;  // throws if key missing
rust::String      Metadata::to_json() const;                   // full metadata as JSON
rust::Vec<rust::String> Metadata::list_keys() const;

MetadataValueType

enum class MetadataValueType : uint8_t {
    Bool,
    Integer,
    Float,
    String,
    ListInt,
    ListFloat,
    ListString,
    Timestamp,
};

Service, Action, and Streaming Patterns

C++ nodes can implement communication patterns using the metadata API. The well-known metadata keys are:

// Service server: pass through request_id from input metadata
auto input_metadata = event_as_arrow_input_with_info(event);
send_output_with_metadata(sender, "response", result, std::move(input_metadata.metadata));

// Action server: set goal_id and goal_status on result
auto meta = new_metadata();
meta->set_string("goal_id", goal_id);
meta->set_string("goal_status", "succeeded");
send_output_with_metadata(sender, "result", result_data, std::move(meta));

CombinedEvents (ROS2 integration)

When using the optional ros2-bridge feature, node events and ROS2 subscription events can be merged into a single stream.

// Convert Adora events into a combined stream.
CombinedEvents adora_events_into_combined(rust::Box<Events> events);

// Create an empty combined stream (for ROS2-only nodes).
CombinedEvents empty_combined_events();

CombinedEvents struct

struct CombinedEvents {
    rust::Box<MergedEvents> events;

    CombinedEvent next();  // blocking -- returns the next merged event
};

CombinedEvent struct

struct CombinedEvent {
    rust::Box<MergedAdoraEvent> event;

    bool is_adora() const;  // true if this is a standard Adora event
};

// Downcast a combined event back to an AdoraEvent. Throws if not an Adora event.
rust::Box<AdoraEvent> downcast_adora(CombinedEvent event);

ROS2 subscriptions add their own events to the merged stream. Use subscription->matches(event) and subscription->downcast(event) to handle ROS2-specific events (see the ROS2 Bridge docs).

Operator API (adora-operator-api-cxx)

Operators are shared libraries loaded by the Adora runtime. The C++ side implements two functions that the CXX bridge calls into.

Required C++ interface

You must provide a header operator.h and an implementation file. The header declares an Operator class and two free functions:

// operator.h
#pragma once
#include <memory>
#include "adora-operator-api.h"

class Operator {
public:
    Operator();
    // Add any state your operator needs.
};

std::unique_ptr<Operator> new_operator();

AdoraOnInputResult on_input(
    Operator& op,
    rust::Str id,
    rust::Slice<const uint8_t> data,
    OutputSender& output_sender);

• new_operator() – called once at startup; returns the operator instance.
• on_input() – called for every input event; process data and optionally send outputs.

OutputSender (operator)

Available inside on_input(). Sends data on a named output.

AdoraSendOutputResult send_output(
    OutputSender& sender,
    rust::Str id,
    rust::Slice<const uint8_t> data);

Result types

struct AdoraOnInputResult {
    rust::String error;  // empty on success
    bool         stop;   // true to request graceful shutdown
};

struct AdoraSendOutputResult {
    rust::String error;  // empty on success
};

Quick Start: Node Example

A minimal node that receives timer ticks and sends a counter.

#include "adora-node-api.h"
#include <iostream>
#include <vector>

int main() {
    auto adora_node = init_adora_node();
    unsigned char counter = 0;

    for (;;) {
        auto event = next_event(adora_node.events);
        auto ty = event_type(event);

        if (ty == AdoraEventType::AllInputsClosed) {
            break;
        }
        if (ty == AdoraEventType::Stop) {
            break;
        }
        if (ty == AdoraEventType::Input) {
            auto input = event_as_input(std::move(event));
            counter += 1;

            std::cout << "Input: " << std::string(input.id)
                      << " counter=" << (int)counter << std::endl;

            std::vector<unsigned char> out{counter};
            rust::Slice<const uint8_t> slice{out.data(), out.size()};
            auto result = send_output(adora_node.send_output, "counter", slice);
            if (!result.error.empty()) {
                std::cerr << "Send error: " << std::string(result.error) << std::endl;
                return 1;
            }
        }
    }
    return 0;
}

Dataflow YAML:

nodes:
  - id: cxx-node
    path: build/my_node
    inputs:
      tick: adora/timer/millis/300
    outputs:
      - counter

Quick Start: Arrow Node Example

A node that receives and sends Arrow arrays via the C Data Interface, with metadata.

#include "adora-node-api.h"
#include <arrow/api.h>
#include <arrow/c/bridge.h>
#include <iostream>

int main() {
    auto adora_node = init_adora_node();

    for (int i = 0; i < 10; i++) {
        auto event = adora_node.events->next();
        auto ty = event_type(event);

        if (ty == AdoraEventType::AllInputsClosed || ty == AdoraEventType::Stop) {
            break;
        }
        if (ty == AdoraEventType::Input) {
            // Receive Arrow input with metadata
            struct ArrowArray c_array;
            struct ArrowSchema c_schema;
            auto info = event_as_arrow_input_with_info(
                std::move(event),
                reinterpret_cast<uint8_t*>(&c_array),
                reinterpret_cast<uint8_t*>(&c_schema));

            if (!info.error.empty()) {
                std::cerr << std::string(info.error) << std::endl;
                continue;
            }

            std::cout << "Input: " << std::string(info.id)
                      << " ts=" << info.metadata->timestamp() << std::endl;

            auto imported = arrow::ImportArray(&c_array, &c_schema);
            auto array = imported.ValueOrDie();
            std::cout << "Arrow: " << array->ToString() << std::endl;

            // Build an output Arrow array
            arrow::Int32Builder builder;
            builder.Append(i * 10);
            std::shared_ptr<arrow::Array> out_array;
            builder.Finish(&out_array);

            // Export and send with metadata
            struct ArrowArray out_c_array;
            struct ArrowSchema out_c_schema;
            arrow::ExportArray(*out_array, &out_c_array, &out_c_schema);

            auto meta = new_metadata();
            meta->set_string("source", "cpp-arrow-node");
            meta->set_int("iteration", i);

            auto result = send_arrow_output(
                adora_node.send_output, "counter",
                reinterpret_cast<uint8_t*>(&out_c_array),
                reinterpret_cast<uint8_t*>(&out_c_schema),
                std::move(meta));

            if (!result.error.empty()) {
                std::cerr << "Send error: " << std::string(result.error) << std::endl;
            }
        }
    }
    return 0;
}

Quick Start: Operator Example

A minimal operator shared library.

// operator.cc
#include "operator.h"
#include <iostream>
#include <vector>

Operator::Operator() {}

std::unique_ptr<Operator> new_operator() {
    return std::make_unique<Operator>();
}

AdoraOnInputResult on_input(
    Operator& op,
    rust::Str id,
    rust::Slice<const uint8_t> data,
    OutputSender& output_sender)
{
    op.counter += 1;

    std::vector<unsigned char> out{op.counter};
    rust::Slice<const uint8_t> slice{out.data(), out.size()};
    auto send_result = send_output(output_sender, rust::Str("status"), slice);

    return AdoraOnInputResult{send_result.error, false};
}

Dataflow YAML:

nodes:
  - id: runtime-node
    operators:
      - id: my-operator
        shared-library: build/my_operator
        inputs:
          data: some-node/output
        outputs:
          - status

Build Integration (CMake)

The recommended build approach uses CMake with the DoraTargets.cmake helper (see examples/cmake-dataflow/).

Project structure

my-project/
  CMakeLists.txt
  DoraTargets.cmake       # copied from examples/cmake-dataflow/
  node/main.cc
  operator/operator.h
  operator/operator.cc
  dataflow.yml

CMakeLists.txt

cmake_minimum_required(VERSION 3.21)
project(my-dataflow LANGUAGES C CXX)

set(CMAKE_CXX_STANDARD 20)
set(CMAKE_CXX_FLAGS "-fPIC")

include(DoraTargets.cmake)
link_directories(${adora_link_dirs})

# Standalone node (executable)
add_executable(my_node node/main.cc ${node_bridge})
add_dependencies(my_node Adora_cxx)
target_include_directories(my_node PRIVATE ${adora_cxx_include_dir})
target_link_libraries(my_node adora_node_api_cxx)

# Operator (shared library)
add_library(my_operator SHARED
    operator/operator.cc ${operator_bridge})
add_dependencies(my_operator Adora_cxx)
target_include_directories(my_operator PRIVATE
    ${adora_cxx_include_dir} ${adora_c_include_dir}
    ${CMAKE_CURRENT_SOURCE_DIR}/operator)
target_link_libraries(my_operator adora_operator_api_cxx)

install(TARGETS my_node DESTINATION ${CMAKE_CURRENT_SOURCE_DIR}/bin)
install(TARGETS my_operator DESTINATION ${CMAKE_CURRENT_SOURCE_DIR}/lib)

What DoraTargets.cmake provides

Build steps

# Option A: Build against local Adora source
mkdir build && cd build
cmake .. -DDORA_ROOT_DIR=/path/to/adora
cmake --build .

# Option B: Build against Adora from GitHub (cloned automatically)
mkdir build && cd build
cmake ..
cmake --build .

Requirements

• C++20 compiler
• Rust toolchain (for building the Adora static libraries via Cargo)
• CMake 3.21+
• For Arrow integration: Apache Arrow C++ library

CXX Bridge Notes

• All Rust opaque types (Events, OutputSender, AdoraEvent, Metadata, MergedEvents, MergedAdoraEvent) are accessed through rust::Box<T>.
• rust::String, rust::Vec<T>, and rust::Slice<const T> are CXX bridge types that interoperate with their C++ standard library counterparts. See the CXX type reference.
• Functions that return Result<T> in Rust throw C++ exceptions on the error path.
• Arrow FFI functions (event_as_arrow_input, send_arrow_output) are unsafe on the Rust side. The caller must pass valid pointers to ArrowArray / ArrowSchema structs cast to uint8_t*.
• The node library is a static archive (staticlib). Link it into your executable with -ladora_node_api_cxx.
• The operator library is also a static archive. Link it into your shared library with -ladora_operator_api_cxx.

Adora CLI Reference

Adora (AI-Dora, Dataflow-Oriented Robotic Architecture) is a 100% Rust framework for building real-time robotics and AI applications. This document covers the adora CLI from both an end-user and developer perspective.

Table of Contents

• Quick Start
• Installation
• Core Concepts
• Dataflow Descriptor
• Command Reference
  • Lifecycle
  • Monitoring
  • Debugging
  • Setup
  • Utility
  • Self-Management
• Environment Variables
• Architecture Guide
• Writing Nodes
• Writing Operators
• Distributed Deployments – see also Distributed Deployment Guide for cluster management, scheduling, and operations
• Troubleshooting
• Debugging and Observability – standalone guide covering record/replay, topic inspection, log analysis, and resource monitoring
• API References: Rust | Python | C | C++

Quick Start

# Create a new project
adora new my-robot --kind dataflow --lang rust

# Run locally (no coordinator/daemon needed)
adora run dataflow.yml

# Or use coordinator/daemon for production
adora up
adora start dataflow.yml --attach
# Ctrl-C to stop
adora down

Installation

From crates.io (recommended)

cargo install adora-cli

From source

cargo install --path binaries/cli --locked

Verify

adora --version
adora status

Core Concepts

Dataflow

A dataflow is a directed graph of nodes connected by typed data channels. Nodes produce outputs that other nodes consume as inputs. The framework handles data routing, serialization (Apache Arrow), and lifecycle management.

Execution Modes

Component Roles

CLI  -->  Coordinator  -->  Daemon(s)  -->  Nodes / Operators
              (control plane)  (per machine)    (user code)

• CLI: User interface. Sends commands, displays logs.
• Coordinator: Orchestrates dataflow lifecycle across machines.
• Daemon: Spawns node processes, manages IPC, collects metrics.
• Node: A standalone process that produces and consumes Arrow data.
• Operator: In-process code running inside a shared runtime (lower latency than nodes).

Data Format

All data flows through the system as Apache Arrow columnar arrays. This enables zero-copy shared memory transfer between co-located nodes and zero-serialization overhead.

Dataflow Descriptor

Dataflows are defined in YAML files. Here is the complete schema:

Minimal Example

nodes:
  - id: sender
    path: sender.py
    outputs:
      - message

  - id: receiver
    path: receiver.py
    inputs:
      message: sender/message

Full Schema

# Dataflow-level settings
health_check_interval: 5.0    # health check sweep interval in seconds (default: 5.0)

nodes:
  - id: my-node                 # unique identifier (required)
    name: "My Node"             # human-readable name (optional)
    description: "..."          # description (optional)

    # --- Source (pick one) ---
    path: ./target/debug/my-node          # local executable
    # path: https://example.com/node.zip  # download from URL
    # git: https://github.com/org/repo.git  # build from git
    #   branch: main            # git branch (mutually exclusive with tag/rev)
    #   tag: v1.0               # git tag
    #   rev: abc123             # git commit hash

    # --- Build ---
    build: cargo build -p my-node   # shell command to build (optional)

    # --- Inputs ---
    inputs:
      # Short form: source_node/output_id
      tick: adora/timer/millis/100
      data: other-node/output

      # Long form with options
      sensor_data:
        source: sensor/frames
        queue_size: 10            # input buffer size (default: 10)
        queue_policy: drop_oldest # or "backpressure" (buffers up to 10x queue_size)
        input_timeout: 5.0        # circuit breaker timeout in seconds

    # --- Outputs ---
    outputs:
      - processed
      - status

    # --- Environment ---
    env:
      MY_VAR: "value"
      FROM_ENV:
        __adora_env: HOST_VAR     # read from host environment
    args: "--verbose"             # command-line arguments

    # --- Fault tolerance ---
    restart_policy: on-failure    # never (default) | on-failure | always
    max_restarts: 5               # 0 = unlimited
    restart_delay: 1.0            # initial backoff in seconds
    max_restart_delay: 30.0       # backoff cap in seconds
    restart_window: 300.0         # reset counter after N seconds
    health_check_timeout: 30.0    # kill if no activity for N seconds

    # --- Logging ---
    min_log_level: info           # source-level filter (daemon-side)
    send_stdout_as: raw_output    # route raw stdout as data output
    send_logs_as: log_entries     # route structured logs as data output
    max_log_size: "50MB"          # rotate log files at this size
    max_rotated_files: 5          # number of rotated files to keep (1-100)

    # --- Deployment ---
    _unstable_deploy:
      machine: A                  # target machine/daemon ID

# Debug settings
_unstable_debug:
  publish_all_messages_to_zenoh: true   # required for topic echo/hz/info

Built-in Timer Nodes

Timers are virtual nodes that emit ticks at fixed intervals:

inputs:
  tick: adora/timer/millis/100   # every 100ms
  slow: adora/timer/millis/1000  # every 1s
  fast: adora/timer/hz/30        # 30 Hz (~33ms)

Operator Nodes

Operators run in-process inside a shared runtime (no separate process):

nodes:
  # Single operator (shorthand)
  - id: detector
    operator:
      python: detect.py
      build: pip install -r requirements.txt
      inputs:
        image: camera/frames
      outputs:
        - bbox

  # Multiple operators sharing a runtime
  - id: runtime-node
    operators:
      - id: preprocessor
        shared-library: ../../target/debug/libpreprocess
        inputs:
          raw: sensor/data
        outputs:
          - processed
      - id: analyzer
        shared-library: ../../target/debug/libanalyze
        inputs:
          data: runtime-node/preprocessor/processed
        outputs:
          - result

Distributed Deployment

Assign nodes to specific machines using _unstable_deploy:

nodes:
  - id: camera-driver
    _unstable_deploy:
      machine: robot-arm
    path: ./target/debug/camera
    outputs:
      - frames

  - id: ml-inference
    _unstable_deploy:
      machine: gpu-server
    path: ./target/debug/inference
    inputs:
      frames: camera-driver/frames
    outputs:
      - predictions

When nodes are on different machines, communication automatically switches from shared memory to Zenoh pub/sub.

Command Reference

Lifecycle Commands

adora run

Run a dataflow locally without coordinator or daemon. Best for development and testing.

adora run <PATH> [OPTIONS]

Examples:

# Basic run
adora run dataflow.yml

# Stop after 10 seconds, only show warnings
adora run dataflow.yml --stop-after 10s --log-level warn

# Python dataflow with uv
adora run dataflow.yml --uv

# Debug one node, silence others
adora run dataflow.yml --log-level warn --log-filter "sensor=debug"

# JSON output for CI pipelines
adora run dataflow.yml --log-format json --stop-after 30s 2>test.json

adora up

Start coordinator and daemon in local mode.

adora up

Spawns adora coordinator and adora daemon as background processes. Waits for both to be ready before returning. Idempotent: if already running, does nothing.

adora down (alias: adora destroy)

Tear down coordinator and daemon. Stops all running dataflows first.

adora down [OPTIONS]

adora build

Run build commands defined in the dataflow descriptor.

adora build <PATH> [OPTIONS]

Type checking: After expanding modules, build runs the same type checks as validate. Warnings are printed by default; use --strict-types (or set strict_types: true in the YAML) to fail the build on type mismatches. User-defined types in a types/ directory next to the dataflow are loaded automatically.

Build strategy: If nodes have _unstable_deploy sections and a coordinator is reachable, builds are distributed to target machines. Otherwise, builds run locally.

Git sources: Nodes with a git: field are cloned/updated before building. The build command runs from the git repository root.

adora start

Start a dataflow on a running coordinator.

adora start <PATH> [OPTIONS]

If neither --attach nor --detach is specified: attaches if running in a TTY, detaches otherwise.

Attach mode: Streams logs, handles Ctrl-C gracefully (first = stop, second = force kill).

Hot reload: Watches Python operator source files. On change, sends a reload request to the coordinator which propagates to the daemon.

adora stop

Stop a running dataflow.

adora stop [UUID_OR_NAME] [OPTIONS]

If no identifier is given and running in a TTY, presents an interactive picker.

Stop sequence: Send Event::Stop -> wait grace duration -> SIGTERM -> hard kill.

adora restart

Restart a running dataflow (stop + re-start with stored descriptor). No YAML path needed – the coordinator retains the original descriptor.

adora restart [UUID] [OPTIONS]

Examples:

# Restart by name
adora restart --name my-app

# Restart by UUID with forced stop
adora restart a1b2c3d4-... --force

adora record

Record dataflow messages to an .adorec file for offline replay. See Debugging Guide for full workflows.

adora record <DATAFLOW_YAML> [OPTIONS]

Default mode injects a record node into the dataflow. --proxy mode requires a running dataflow and publish_all_messages_to_zenoh: true.

adora replay

Replay a recorded .adorec file by replacing source nodes with replay nodes. See Debugging Guide for full workflows.

adora replay <FILE> [OPTIONS]

Monitoring Commands

adora list (alias: adora ps)

List running dataflows with metrics.

adora list [OPTIONS]

Output columns: UUID, Name, Status, Nodes, CPU, Memory

adora logs

Show and follow logs of a dataflow and node.

adora logs [UUID_OR_NAME] [NODE] [OPTIONS]

Filter pipeline: Read/Parse -> Time filters -> Grep -> Tail -> Display

Examples:

# Follow all nodes live
adora logs my-dataflow --all-nodes --follow

# Last 50 errors from a specific node
adora logs my-dataflow sensor --level error --tail 50

# Search logs from last 5 minutes
adora logs my-dataflow --all-nodes --since 5m --grep "timeout"

# Read local files (no coordinator needed)
adora logs --local --all-nodes --tail 100

# Post-mortem analysis: errors in time window
adora logs --local sensor --since 1h --until 30m --level error

Duration formats: 30 (seconds), 30s, 5m, 1h, 2d

adora inspect top (alias: adora top)

Real-time TUI monitor for node resource usage (like top).

adora inspect top [OPTIONS]
adora top [OPTIONS]

Requires an interactive terminal (unless --once is used).

Columns: NODE, STATUS, DATAFLOW, PID, CPU%, MEMORY (MB), RESTARTS, QUEUE, NET TX, NET RX, I/O READ (MB/s), I/O WRITE (MB/s)

• STATUS: Running, Restarting, Degraded (broken inputs), or Failed
• RESTARTS: Current restart count per node
• QUEUE: Pending messages in the node’s input queue
• NET TX/RX: Cumulative cross-daemon network bytes sent/received via Zenoh

CPU values are per-core (can exceed 100% with multiple cores). Metrics come from daemons, so this works for distributed deployments.

Scripting example:

# JSON snapshot for CI/monitoring pipelines
adora top --once | jq '.[].cpu_usage'

adora topic list

List all topics (outputs) in a running dataflow.

adora topic list [OPTIONS]

adora topic echo

Subscribe to topics and display messages in real-time.

adora topic echo [OPTIONS] [DATA...]

Requires _unstable_debug.publish_all_messages_to_zenoh: true in the descriptor.

adora topic hz

Measure topic publish frequency with a TUI dashboard.

adora topic hz [OPTIONS] [DATA...]

Requires an interactive terminal. Displays: Avg (ms), Avg (Hz), Min (ms), Max (ms), Std (ms), plus a rate sparkline and histogram for the selected topic.

adora topic info

Show detailed metadata of a single topic.

adora topic info [OPTIONS] DATA

Subscribes to the topic for the specified duration and reports: type (Arrow schema), publisher, subscribers, message count, bandwidth.

adora node

Manage and inspect dataflow nodes.

adora node list

adora node list [OPTIONS]

Lists nodes in a running dataflow with their status, CPU, memory, and restart count.

Columns: NODE, STATUS, PID, CPU%, MEMORY (MB), RESTARTS, DATAFLOW

adora node info

Show detailed information about a specific node including status, inputs, outputs, and metrics.

adora node info <NODE> [OPTIONS]

adora node restart

Restart a single node within a running dataflow. The daemon stops the node process and respawns it.

adora node restart <NODE> [OPTIONS]

adora node stop

Stop a single node within a running dataflow without stopping the entire dataflow.

adora node stop <NODE> [OPTIONS]

adora topic pub

Publish JSON data to a topic in a running dataflow. Requires publish_all_messages_to_zenoh: true.

adora topic pub <TOPIC> [DATA] [OPTIONS]

Examples:

# Publish a single value
adora topic pub -d my-app sensor/threshold '[42]'

# Publish from file, 10 times
adora topic pub -d my-app sensor/config --file config.json --count 10

adora param

Manage runtime parameters for nodes. Parameters are persisted in the coordinator store and optionally forwarded to running nodes.

adora param list

List all runtime parameters for a node.

adora param list <NODE> [OPTIONS]

adora param get

Get a single runtime parameter value.

adora param get <NODE> <KEY> [OPTIONS]

adora param set

Set a runtime parameter. The value is JSON. The parameter is stored in the coordinator and forwarded to the node if it is running.

adora param set <NODE> <KEY> <VALUE> [OPTIONS]

Examples:

# Set a numeric parameter
adora param set -d my-app sensor threshold 42

# Set a string parameter
adora param set -d my-app camera resolution '"1080p"'

# Set a complex parameter
adora param set -d my-app detector config '{"confidence": 0.8, "nms": 0.5}'

adora param delete

Delete a runtime parameter.

adora param delete <NODE> <KEY> [OPTIONS]

adora doctor

Diagnose environment, coordinator/daemon connectivity, and optionally validate a dataflow YAML.

adora doctor [OPTIONS]

Checks performed:

1. Coordinator reachability
2. Daemon connectivity
3. Active dataflow status
4. Dataflow YAML validation (if --dataflow provided)

Examples:

# Basic health check
adora doctor

# Check environment + validate a dataflow
adora doctor --dataflow dataflow.yml

adora trace list

List recent traces captured by the coordinator. The coordinator captures spans from adora_coordinator and adora_core crates in-memory (up to 4096 spans). No external tracing infrastructure required.

adora trace list [OPTIONS]

Output columns: TRACE ID (first 12 chars), ROOT SPAN, SPANS, STARTED, DURATION

Example:

adora trace list

TRACE ID      ROOT SPAN          SPANS  STARTED              DURATION
a1b2c3d4e5f6  spawn_dataflow     12     2026-03-01 10:30:05  1.234s
f8e7d6c5b4a3  build_dataflow     5      2026-03-01 10:29:58  0.500s

adora trace view

View spans for a specific trace as an indented tree. Supports prefix matching on trace IDs.

adora trace view <TRACE_ID> [OPTIONS]

Example:

adora trace view a1b2c3d4

spawn_dataflow [INFO 1.234s] {build_id="abc", session_id="def"}
  build_dataflow [INFO 0.500s]
    download_node [DEBUG 0.200s] {url="..."}
  start_inner [INFO 0.734s]
    spawn_node [INFO 0.100s] {node_id="camera"}
    spawn_node [INFO 0.080s] {node_id="detector"}

Trace IDs are prefix-matched: if the prefix uniquely identifies a trace, it resolves automatically. If ambiguous, you’ll be prompted to use a longer prefix.

Setup Commands

adora status (alias: adora check)

Check system health and connectivity.

adora status [OPTIONS]

Reports coordinator connectivity, daemon status, and active dataflow count.

adora new

Generate a new project or node from templates.

adora new <NAME> [OPTIONS]

adora expand

Expand module references in a dataflow and print the resulting flat YAML. Useful for debugging module composition.

adora expand <PATH> [OPTIONS]

Examples:

# Expand a dataflow with modules
adora expand dataflow.yml

# Validate a module file
adora expand --module modules/navigation.module.yml

See the Modules Guide for full documentation on module composition.

adora graph

Visualize a dataflow as a graph.

adora graph <PATH> [OPTIONS]

Without --mermaid, generates an interactive HTML file using mermaid.js. When outputs have type annotations, edge labels include the type name (e.g. image [Image]).

# Generate HTML
adora graph dataflow.yml --open

# Generate Mermaid for GitHub markdown
adora graph dataflow.yml --mermaid

adora validate

Validate a dataflow YAML file and check type annotations.

adora validate <PATH> [OPTIONS]

Checks:

1. Key existence: output_types/input_types keys exist in the corresponding outputs/inputs lists
2. URN resolution: All type URNs resolve in the standard or user-defined type library
3. Edge compatibility: Connected edges have compatible types (exact match, widening, or user-defined rules)
4. Parameterized types: Parameter mismatches (e.g. AudioFrame[sample_type=f32] vs AudioFrame[sample_type=i16])
5. Timer auto-typing: Timer inputs are automatically typed as std/core/v1/UInt64
6. Type inference: When only upstream annotates a type, it is inferred on the downstream input
7. Metadata patterns: output_metadata keys and pattern shorthands are validated
8. Schema compatibility: Struct types are checked at the field level (missing/wrong fields)

User-defined types in a types/ directory next to the dataflow are loaded automatically.

# Validate with warnings
adora validate dataflow.yml

# Strict mode for CI (exit 1 on warnings)
adora validate --strict-types dataflow.yml

See the Type Annotations Guide for the full type library and usage details.

Utility Commands

adora completion

Generate shell completion scripts.

adora completion [SHELL]

Shell is auto-detected if omitted. Supported: bash, zsh, fish, elvish, powershell.

# Bash
eval "$(adora completion bash)"
echo 'eval "$(adora completion bash)"' >> ~/.bashrc

# Zsh
eval "$(adora completion zsh)"
echo 'eval "$(adora completion zsh)"' >> ~/.zshrc

# Fish
adora completion fish > ~/.config/fish/completions/adora.fish

adora system

System management commands.

adora system status [OPTIONS]

Currently provides status as a subcommand (equivalent to adora status).

Self-Management Commands

adora self update

Check for and install CLI updates.

adora self update [--check-only]

Downloads from GitHub releases (dora-rs/adora).

adora self uninstall

Remove the CLI from the system.

adora self uninstall [--force]

Without --force, prompts for confirmation (requires a TTY). Tries uv pip uninstall first, then pip uninstall, then binary self-delete.

Environment Variables

All environment variables serve as fallbacks. CLI flags always take precedence.

# Set defaults for a development session
export ADORA_COORDINATOR_ADDR=192.168.1.10
export ADORA_LOG_LEVEL=info
export ADORA_LOG_FORMAT=compact

Architecture Guide

This section is for developers who want to understand the framework internals, extend it, or debug issues.

Communication Stack

                    ┌─────────────────────────────────────┐
                    │           CLI (adora)                │
                    │   WebSocket (JSON request/reply)     │
                    └─────────────┬───────────────────────┘
                                  │
                    ┌─────────────▼───────────────────────┐
                    │        Coordinator                   │
                    │   WebSocket control + daemon mgmt    │
                    │   State: InMemoryStore | RedbStore   │
                    └──┬──────────────────────────────┬───┘
                       │                              │
          ┌────────────▼──────────┐     ┌─────────────▼──────────┐
          │     Daemon A          │     │     Daemon B           │
          │  (machine: robot)     │     │  (machine: gpu-server) │
          │                       │     │                        │
          │  ┌─────┐  ┌─────┐    │     │  ┌──────┐  ┌───────┐  │
          │  │Node1│  │Node2│    │     │  │Node3 │  │Node4  │  │
          │  └──┬──┘  └──┬──┘    │     │  └──┬───┘  └───┬───┘  │
          │     │shmem    │shmem  │     │     │shmem      │shmem │
          │     └────┬────┘       │     │     └─────┬─────┘      │
          └──────────┼────────────┘     └───────────┼────────────┘
                     │                              │
                     └──────── Zenoh pub/sub ────────┘
                              (cross-machine)

Protocol Layers

Coordinator Internals

The coordinator is an event-driven async server:

Event Sources:
  - CLI WebSocket connections (ControlRequest)
  - Daemon WebSocket connections (DaemonEvent)
  - Heartbeat timer (3s interval)
  - External events (for embedding)

Event Loop:
  merge_all(cli_events, daemon_events, heartbeat, external)
    -> handle_event()
    -> update state
    -> persist to store (if redb)
    -> send replies

Key types:

#![allow(unused)]
fn main() {
// State
RunningDataflow { uuid, name, descriptor, daemons, node_metrics, ... }
RunningBuild    { build_id, errors, log_subscribers, pending_results, ... }
DaemonConnection { sender, pending_replies, last_heartbeat }

// Store trait
trait CoordinatorStore: Send + Sync {
    fn put_dataflow(&self, record: &DataflowRecord) -> Result<()>;
    fn get_dataflow(&self, uuid: &Uuid) -> Result<Option<DataflowRecord>>;
    fn list_dataflows(&self) -> Result<Vec<DataflowRecord>>;
    // ... daemon and build methods
}
}

Store backends:

• memory (default): In-memory, lost on restart.
• redb: Persistent to disk (~/.adora/coordinator.redb). Survives crashes. Requires redb-backend feature.

adora coordinator --store redb
adora coordinator --store redb:/custom/path.redb

Daemon Internals

The daemon manages node processes on a single machine:

Per Node:
  1. Build (if build command specified)
  2. Spawn process with ADORA_NODE_CONFIG env var
  3. Node registers via TCP/shmem handshake
  4. Route inputs/outputs between nodes
  5. Collect metrics (CPU, memory, I/O)
  6. Handle restart policy on exit
  7. Forward logs to coordinator

Communication:
  - Shared memory for messages > 4KB (zero-copy)
  - TCP for control messages and small data
  - flume channels for internal event routing

Metrics collection:

#![allow(unused)]
fn main() {
struct NodeMetrics {
    pid: u32,
    cpu_usage: f32,      // per-core percentage
    memory_mb: f64,
    disk_read_mb_s: Option<f64>,
    disk_write_mb_s: Option<f64>,
    status: NodeStatus,  // Running | Restarting | Degraded | Failed
    restart_count: u32,
    pending_messages: u64,
}
}

Message Types

All inter-component messages are defined in libraries/message/:

#![allow(unused)]
fn main() {
// Node identification
struct NodeId(String);      // [a-zA-Z0-9_.-]
struct DataId(String);      // same validation
type DataflowId = uuid::Uuid;

// Data metadata
struct Metadata {
    timestamp: uhlc::Timestamp,    // hybrid logical clock
    type_info: ArrowTypeInfo,      // Arrow schema
    parameters: MetadataParameters, // custom key-value pairs
}

// Node events (daemon -> node)
enum NodeEvent {
    Stop,
    Reload { operator_id },
    Input { id, metadata, data },
    InputClosed { id },
    InputRecovered { id },
    NodeRestarted { id },
    AllInputsClosed,
}
}

Timestamping

Adora uses a Unified Hybrid Logical Clock (UHLC) for distributed causality. Every message carries a uhlc::Timestamp that preserves causal ordering across machines without synchronized clocks.

Zero-Copy Shared Memory

For large messages (> 4KB), the daemon uses shared memory regions:

1. Sender node requests a shared memory slot from daemon
2. Daemon allocates a region and returns the ID
3. Sender writes Arrow data directly into shared memory
4. Daemon notifies receiver node of the region ID
5. Receiver reads directly from shared memory (zero-copy)
6. Receiver sends a drop token when done

This achieves 10-17x lower latency than ROS2 for large payloads.

Writing Nodes

Rust Node

use adora_node_api::{AdoraNode, Event, IntoArrow};
use adora_core::config::DataId;

fn main() -> eyre::Result<()> {
    let (mut node, mut events) = AdoraNode::init_from_env()?;

    let output = DataId::from("result".to_owned());

    while let Some(event) = events.recv() {
        match event {
            Event::Input { id, metadata, data } => {
                // Process input data (Arrow array)
                let result: u64 = 42;
                node.send_output(
                    output.clone(),
                    metadata.parameters,
                    result.into_arrow(),
                )?;
            }
            Event::Stop(_) => break,
            Event::InputClosed { id } => {
                eprintln!("input {id} closed");
            }
            Event::InputRecovered { id } => {
                eprintln!("input {id} recovered");
            }
            _ => {}
        }
    }
    Ok(())
}

Cargo.toml:

[dependencies]
adora-node-api = { workspace = true }
eyre = "0.6"

Python Node

import pyarrow as pa
from adora import Node

node = Node()

for event in node:
    if event["type"] == "INPUT":
        # event["value"] is a PyArrow array
        values = event["value"].to_pylist()
        result = pa.array([sum(values)])
        node.send_output("result", result)
    elif event["type"] == "STOP":
        break

C Node

#include "node_api.h"

int main() {
    void *ctx = init_adora_context_from_env();
    // ... event loop using adora_next_event / adora_send_output
    free_adora_context(ctx);
    return 0;
}

Node Logging

Nodes can emit structured logs:

Rust:

#![allow(unused)]
fn main() {
// Via tracing (recommended)
tracing::info!("processing frame {}", frame_id);

// Via node API
node.log_info("processing complete");
node.log_with_fields("info", "reading", None, Some(&fields));
}

Python:

import logging
logging.info("processing frame %d", frame_id)

# Or via node API
node.log("info", "processing complete")

Writing Operators

Operators run in-process inside a shared runtime, avoiding process spawn overhead.

Rust Operator

#![allow(unused)]
fn main() {
use adora_operator_api::{register_operator, AdoraOperator, AdoraOutputSender, AdoraStatus, Event};

#[register_operator]
#[derive(Default)]
pub struct MyOperator {
    counter: u32,
}

impl AdoraOperator for MyOperator {
    fn on_event(
        &mut self,
        event: &Event,
        output_sender: &mut AdoraOutputSender,
    ) -> Result<AdoraStatus, String> {
        match event {
            Event::Input { id, data } => {
                self.counter += 1;
                output_sender.send(
                    "count".to_string(),
                    arrow::array::UInt32Array::from(vec![self.counter]),
                )?;
                Ok(AdoraStatus::Continue)
            }
            Event::Stop => Ok(AdoraStatus::Stop),
            _ => Ok(AdoraStatus::Continue),
        }
    }
}
}

Cargo.toml:

[lib]
crate-type = ["cdylib"]

[dependencies]
adora-operator-api = { workspace = true }
arrow = "53"

Python Operator

nodes:
  - id: my-node
    operator:
      python: my_operator.py
      inputs:
        data: source/output
      outputs:
        - result

# my_operator.py
class Operator:
    def __init__(self):
        self.counter = 0

    def on_event(self, event, send_output):
        if event["type"] == "INPUT":
            self.counter += 1
            send_output("result", pa.array([self.counter]))

Distributed Deployments

Setup

# Machine A (coordinator + daemon)
adora up

# Machine B (daemon only, pointing to coordinator on Machine A)
adora daemon --interface 0.0.0.0 --coordinator-addr 192.168.1.10 --machine-id B

# Machine C (same)
adora daemon --interface 0.0.0.0 --coordinator-addr 192.168.1.10 --machine-id C

Dataflow with Machine Assignment

nodes:
  - id: camera
    _unstable_deploy:
      machine: robot
    path: ./camera-driver
    outputs:
      - frames

  - id: inference
    _unstable_deploy:
      machine: gpu-server
    path: ./ml-model
    inputs:
      frames: camera/frames
    outputs:
      - predictions

  - id: actuator
    _unstable_deploy:
      machine: robot
    path: ./actuator-driver
    inputs:
      commands: inference/predictions

Build and Start

# From any machine with coordinator access
adora build dataflow.yml       # distributed build on target machines
adora start dataflow.yml --name my-robot --attach

Monitor

# Resource usage across all machines
adora top

# Logs from any node regardless of machine
adora logs my-robot inference --follow

# List all dataflows
adora list

Coordinator Persistence

For production, use the redb store backend so the coordinator survives restarts:

adora coordinator --store redb

State is persisted to ~/.adora/coordinator.redb. On restart, stale dataflows are marked as failed and the coordinator resumes normal operation.

For managed cluster deployments (cluster.yml, SSH-based lifecycle, label scheduling, systemd services, rolling upgrades), see the Distributed Deployment Guide.

Troubleshooting

For a comprehensive debugging guide covering record/replay workflows, topic inspection, resource monitoring, and end-to-end debugging scenarios, see Debugging and Observability Guide.

Common Issues

“Could not connect to adora-coordinator”

• Run adora up first, or check ADORA_COORDINATOR_ADDR/ADORA_COORDINATOR_PORT
• Verify with adora status

“publish_all_messages_to_zenoh not enabled”

• Use --debug flag: adora start dataflow.yml --debug or adora run dataflow.yml --debug
• Or add to your dataflow YAML:

  _unstable_debug:
    publish_all_messages_to_zenoh: true
  

• Required for topic echo, topic hz, topic info

“adora top requires an interactive terminal”

• These TUI commands need a real terminal (not piped output)
• Same applies to topic hz

Node not receiving inputs

• Check that output names match: source_node/output_id
• Verify the source node lists the output in its outputs: array
• Check adora topic list for available topics

Logs not appearing

• Check --log-level setting (default stdout shows everything)
• Check min_log_level in YAML (filters at source)
• For distributed: verify coordinator/daemon connectivity

Build fails with git source

• Verify git: URL is accessible
• Check that branch, tag, or rev exists
• Build command runs from the git repo root, not the dataflow directory

Debug Workflow

# 1. Full environment diagnosis
adora doctor --dataflow dataflow.yml

# 2. Start with verbose logging and debug topics
adora run dataflow.yml --log-level trace --debug

# 3. Inspect a specific node
adora node info -d my-dataflow problem-node

# 4. Monitor specific node logs
adora logs my-dataflow problem-node --follow --level debug

# 5. Check resource usage
adora top

# 6. Inspect topic data
adora topic echo -d my-dataflow problem-node/output

# 7. Publish test data to a topic
adora topic pub -d my-dataflow problem-node/input '[1, 2, 3]'

# 8. Measure frequencies
adora topic hz -d my-dataflow --window 5

# 9. View/modify runtime parameters
adora param list -d my-dataflow problem-node
adora param set -d my-dataflow problem-node threshold 42

# 10. Restart a misbehaving node without stopping the dataflow
adora node restart -d my-dataflow problem-node

# 11. View coordinator traces (no external infra needed)
adora trace list
adora trace view <trace-id-prefix>

# 12. Visualize dataflow graph
adora graph dataflow.yml --open

Log File Locations

out/
  <dataflow-uuid>/
    log_<node-id>.jsonl          # current log
    log_<node-id>.1.jsonl        # rotated (previous)
    log_<node-id>.2.jsonl        # rotated (older)

Read directly with:

adora logs --local --all-nodes
adora logs --local <node-name> --tail 50

Logging

Adora provides a structured logging system for real-time robotics and AI dataflows. Logs are captured per-node as structured JSONL files, forwarded to the coordinator for live streaming, and optionally routed through the dataflow graph as data messages.

Which Logging Approach Should I Use?

Start here if you’re unsure which approach fits your use case.

Language-Specific Quick Start

Python – the simplest path is Python’s built-in logging module:

import logging
from adora import Node

node = Node()  # Automatically bridges Python logging -> adora

logging.info("Sensor started")       # Captured as structured "info" log
logging.warning("High temp: 42C")    # Captured as structured "warn" log
print("raw debug output")            # Captured as "stdout" level

When Node() is created, it installs a handler that routes all Python logging calls through Rust’s tracing system. The daemon parses these as structured log entries with level, message, file, and line number. No extra configuration needed.

You can also use the explicit API for structured fields:

node.log_info("Reading acquired")
node.log("info", "Reading acquired", fields={"sensor_id": "temp-01"})

Rust – use the node API convenience methods:

#![allow(unused)]
fn main() {
let (node, mut events) = AdoraNode::init_from_env()?;

// Convenience methods (recommended for most cases)
node.log_info("Sensor started");
node.log_warn("High temperature");

// With structured fields
let mut fields = BTreeMap::new();
fields.insert("sensor_id".into(), "temp-01".into());
node.log_with_fields("info", "Reading acquired", None, Some(&fields));
}

Alternatively, Rust nodes can use the tracing crate. When adora’s tracing subscriber is initialized (via init_tracing()), tracing::info!() etc. output structured JSON to stdout, which the daemon parses automatically:

#![allow(unused)]
fn main() {
// Also works -- parsed as structured logs by the daemon
tracing::info!("Sensor started");
tracing::warn!(sensor_id = "temp-01", "High temperature");
}

Use node.log_*() when you want explicit control over the log format. Use tracing::*!() when you want ecosystem integration (spans, instrumentation, OpenTelemetry). Both produce identical structured log entries in the daemon.

C – use the adora_log() function:

adora_log(ctx, "info", 4, "Sensor started", 14);

C++ – use the log_message() function:

log_message(node.send_output, "info", "Sensor started");

Features at a Glance

Log File Format

Each node produces a JSONL file (one JSON object per line) at:

<working_dir>/out/<dataflow_uuid>/log_<node_id>.jsonl

Each line has this structure:

{
  "timestamp": "2024-01-15T10:30:00.123Z",
  "level": "info",
  "node_id": "sensor",
  "message": "Starting sensor...",
  "target": "sensor::module",
  "fields": { "key": "value" }
}

How Node Output Becomes Log Entries

The daemon captures each line of stdout/stderr from a node process and attempts to parse it as a structured log message (JSON with level, message, timestamp, and optional fields). If parsing succeeds, the structured fields are preserved. If parsing fails, the raw line becomes a "stdout"-level entry.

This means nodes using Rust’s tracing or log crate with JSON output get full structured logging automatically. Nodes that simply println! produce "stdout"-level entries.

Viewing Logs: adora run

When running a dataflow with adora run, logs from all nodes are displayed in real-time on the terminal.

Flags

adora run dataflow.yml [OPTIONS]

Log Levels

From most to least verbose:

Setting --log-level info hides stdout, trace, and debug messages. The stdout level is a special catch-all that passes everything.

Level Filtering Logic

The level filter uses LogLevelOrStdout::passes():

Message level    Filter level    Displayed?
─────────────    ────────────    ──────────
stdout           stdout          yes
stdout           info            no       (stdout only passes stdout filter)
info             stdout          yes      (any log level passes stdout filter)
debug            info            no       (debug is more verbose than info)
error            info            yes      (error is less verbose than info)

Per-Node Overrides

The --log-filter flag lets you set different levels for different nodes:

adora run dataflow.yml --log-level info --log-filter "sensor=debug,planner=warn"

This shows info and above for all nodes, except sensor (shows debug and above) and planner (shows warn and above).

Format: "node1=level,node2=level" (comma-separated name=level pairs).

Output Formats

Pretty (default) – colored, human-readable:

10:30:00 INFO   sensor: Starting sensor...

10:30:01 INFO   [adora]: spawning node processor

10:30:01 stdout sensor: raw output line

• Timestamp in local timezone (HH:MM:SS)
• Level colored: ERROR (red), WARN (yellow), INFO (green), DEBUG (blue), TRACE (dimmed), stdout (italic dimmed blue)
• Node name in bold with a unique color based on the name
• System messages prefixed with [adora]
• Lifecycle messages (spawning, node finished, stopping) get visual separation with blank lines

Json – full LogMessage struct as JSON, one per line:

{"build_id":null,"dataflow_id":"abc-123","node_id":"sensor","level":"INFO","message":"Starting...","timestamp":"2024-01-15T10:30:00Z",...}

Useful for piping to jq or ingesting into log aggregation systems.

Compact – minimal, no color:

10:30:00 INFO sensor: Starting sensor...

Useful for CI/CD environments and log files.

Viewing Logs: adora logs

Read historical logs or stream live logs from a running dataflow.

Basic Usage

# Read logs for a specific node (via coordinator)
adora logs <dataflow_uuid> <node_name>

# Read local log files directly
adora logs --local <node_name>
adora logs --local --all-nodes

# Stream live logs
adora logs <dataflow_uuid> <node_name> --follow
adora logs --local <node_name> --follow

Flags

Time Filters

--since and --until accept duration strings relative to now:

# Logs from the last 5 minutes
adora logs --local sensor --since 5m

# Logs from 1 hour ago to 30 minutes ago
adora logs --local sensor --since 1h --until 30m

# Last 10 errors from the past hour
adora logs --local sensor --since 1h --level error --tail 10

Supported duration formats: 30 (seconds), 30s, 5m, 1h, 2d.

Text Search

--grep performs case-insensitive substring matching against:

• The log message text
• The node ID
• The module target

# Find all timeout-related messages
adora logs --local --all-nodes --grep "timeout"

# Find errors from a specific module
adora logs --local sensor --grep "camera::driver" --level error

Filter Pipeline

All filters are applied in this order:

Read/Parse -> Time Filters -> Grep -> Tail -> Display

When --since, --until, or --grep are used in coordinator mode, the CLI fetches all logs from the server (ignoring --tail server-side) and applies all filters client-side. This ensures correct results when combining filters.

Local vs Coordinator Mode

Local mode (--local) reads JSONL files directly from the out/ directory in the current working directory. No coordinator or daemon needs to be running. If --all-nodes is used or no node name is given, all log files are merged and sorted by timestamp.

Coordinator mode (default) connects to a running coordinator via WebSocket. The coordinator reads log files from the daemon’s working directory and streams them back. This works for both local and distributed deployments.

Follow Mode

Local follow (--local --follow): Polls log files every 200ms for new content. New lines are parsed, filtered by --grep, and printed. Time/tail filters only apply to the initial historical output.

Coordinator follow (--follow): Opens a WebSocket subscription to the coordinator. The coordinator forwards log messages from the daemon in real-time. Level filtering is applied server-side for efficiency. --grep and --since are applied client-side on the stream.

Environment Variables

All environment variables serve as fallbacks – CLI flags always take precedence.

Example:

# Set defaults for a development session
export ADORA_LOG_LEVEL=info
export ADORA_LOG_FORMAT=pretty
export ADORA_LOG_FILTER="sensor=debug"

# These are equivalent:
adora run dataflow.yml
adora run dataflow.yml --log-level info --log-format pretty --log-filter "sensor=debug"

# CLI flag overrides env var:
adora run dataflow.yml --log-level debug   # overrides ADORA_LOG_LEVEL=info

YAML Configuration

min_log_level

Filter logs at the source (daemon-side) before they reach log files, the coordinator, or send_logs_as routing.

nodes:
  - id: noisy-sensor
    path: ./target/debug/sensor
    min_log_level: info    # suppress debug/trace/stdout from this node

Valid values: error, warn, info, debug, trace, stdout.

When set, the daemon drops log messages below this level immediately after parsing. This reduces disk I/O, network traffic, and log file size. The filtering uses the same passes() logic as the CLI display filter.

send_stdout_as

Route raw stdout/stderr lines as dataflow output messages.

nodes:
  - id: legacy-node
    path: ./legacy-script.py
    send_stdout_as: raw_output
    outputs:
      - raw_output
      - data

  - id: log-consumer
    inputs:
      logs: legacy-node/raw_output

Each stdout/stderr line is sent as an Arrow-encoded string. This is useful for integrating legacy nodes that output data on stdout (e.g., Python scripts using print()).

Both send_stdout_as and normal log file writing happen – stdout routing does not suppress log files.

send_logs_as

Route parsed structured log entries as dataflow output messages.

nodes:
  - id: sensor
    path: ./target/debug/sensor
    send_logs_as: log_entries
    outputs:
      - data
      - log_entries

  - id: log-aggregator
    inputs:
      sensor_logs: sensor/log_entries

Unlike send_stdout_as, this only sends lines that were successfully parsed as structured logs (not raw stdout). Each entry is serialized as a full JSON LogMessage string. The min_log_level filter applies before routing – suppressed messages are not sent.

Use this to build log aggregation, alerting, or monitoring nodes within the dataflow itself.

adora/logs – Automatic Log Aggregation

Subscribe to logs from all nodes with a single input line – no manual wiring needed:

nodes:
  - id: sensor
    path: sensor.py
    inputs:
      tick: adora/timer/millis/200
    outputs:
      - reading

  - id: processor
    path: processor.py
    inputs:
      reading: sensor/reading
    outputs:
      - result

  - id: log-viewer
    path: log_viewer.py
    inputs:
      logs: adora/logs              # all nodes, all levels
      errors: adora/logs/error      # only error+ from all nodes
      sensor: adora/logs/info/sensor  # info+ from one node

The adora/logs virtual input works like adora/timer – the daemon handles subscription internally. Each log message arrives as a JSON-encoded LogMessage string in an Arrow array. To prevent infinite loops, a node never receives its own log messages.

Syntax:

Levels: stdout, error, warn, info, debug, trace.

When to use adora/logs vs send_logs_as:

See examples/log-aggregator/ for a complete working example.

max_log_size

Enable size-based log file rotation.

nodes:
  - id: sensor
    path: ./target/debug/sensor
    max_log_size: "50MB"

When the active log file exceeds the configured size, the daemon:

1. Flushes and closes the current file
2. Renames existing rotated files: .4.jsonl -> .5.jsonl, .3.jsonl -> .4.jsonl, etc.
3. Renames the current file: log_sensor.jsonl -> log_sensor.1.jsonl
4. Creates a fresh log_sensor.jsonl
5. Deletes any file beyond the rotation limit (default 5, configurable via max_rotated_files)

Naming convention:

log_sensor.jsonl       # current (active)
log_sensor.1.jsonl     # previous
log_sensor.2.jsonl     # older
log_sensor.3.jsonl
log_sensor.4.jsonl
log_sensor.5.jsonl     # oldest (deleted on next rotation)

Maximum disk usage per node: max_log_size * (1 + max_rotated_files) (1 active + N rotated).

Without max_log_size, log files grow unbounded. For long-running dataflows, always set this.

The adora logs --local command automatically reads all rotated files for a node and merges them in chronological order (oldest rotated file first, current file last).

max_rotated_files

Control how many rotated log files to keep (default: 5, range: 1-100).

nodes:
  - id: sensor
    path: ./target/debug/sensor
    max_log_size: "50MB"
    max_rotated_files: 10    # keep 10 rotated files instead of 5

With max_rotated_files: 10 and max_log_size: "50MB", maximum disk usage is 50MB * 11 = 550MB per node. Lower values save disk space; higher values preserve more history.

Runtime Node Restrictions

For runtime nodes (operators), only one of each logging field is allowed per runtime:

# OK -- single operator
nodes:
  - id: runtime-node
    operator:
      python: process.py
      send_logs_as: logs
      min_log_level: info
      max_log_size: "100MB"

# ERROR -- multiple operators with conflicting configs
nodes:
  - id: runtime-node
    operators:
      - id: op1
        python: a.py
        send_logs_as: logs1
      - id: op2
        python: b.py
        send_logs_as: logs2    # Error: multiple send_logs_as

When a single operator in a runtime sets these fields, the output name is prefixed with the operator ID (e.g., op1/logs).

Node Log API

Nodes can emit structured log messages programmatically using the node API. These are equivalent to writing JSON-formatted log lines to stdout – the daemon parses them identically.

Rust

#![allow(unused)]
fn main() {
use adora_node_api::AdoraNode;
use std::collections::BTreeMap;

let (node, mut events) = AdoraNode::init_from_env()?;

// General log with level string and optional target
node.log("info", "sensor initialized", Some("sensor::init"));

// Convenience methods (no target parameter)
node.log_error("connection failed");
node.log_warn("temperature elevated");
node.log_info("reading acquired");
node.log_debug("raw bytes received");
node.log_trace("entering loop iteration");

// Structured fields (key-value context preserved through send_logs_as)
let mut fields = BTreeMap::new();
fields.insert("sensor_id".to_string(), "temp-01".to_string());
fields.insert("reading".to_string(), "42.5".to_string());
node.log_with_fields("info", "reading acquired", None, Some(&fields));
}

The level parameter accepts "error", "warn" (or "warning"), "info", "debug", "trace". Unknown levels default to "info". Fields are capped at 60 KB total to match the downstream 64 KB parse limit.

Python

Python nodes have three ways to log, all producing structured log entries:

from adora import Node
import logging

node = Node()

# Option 1: Python's logging module (recommended -- auto-bridged by Node())
logging.info("sensor initialized")
logging.warning("temperature elevated")
logging.debug("raw bytes: %s", data)

# Option 2: Explicit adora API with level string
node.log("info", "sensor initialized", target="sensor.init")
node.log("info", "reading acquired", fields={"sensor_id": "temp-01", "reading": "42.5"})

# Option 3: Convenience methods
node.log_error("connection failed")
node.log_warn("temperature elevated")
node.log_info("reading acquired")
node.log_debug("raw bytes received")
node.log_trace("entering loop iteration")

# This also works but produces "stdout"-level entries (no structure):
print("raw output")

How the Python logging bridge works: When Node() is created, it installs a custom logging.Handler that routes all Python logging calls through Rust’s tracing system. The daemon parses these as structured log entries with level, message, file path, and line number. This happens automatically – no configuration needed.

Common pitfall: Do not call logging.basicConfig() before creating Node(). The node constructor sets up the logging bridge; calling basicConfig() first may install a conflicting handler. If you need custom formatters, configure them after Node() creation.

C

#include "node_api.h"

void *ctx = init_adora_context_from_env();
const char *level = "info";
const char *msg = "sensor initialized";
adora_log(ctx, level, strlen(level), msg, strlen(msg));

C++

// Via the cxx bridge
auto node = init_adora_node();
log_message(node.send_output, "info", "sensor initialized");

Log Utilities Library (adora-log-utils)

The adora-log-utils crate provides parsing, merging, filtering, and formatting utilities for working with LogMessage entries in custom sink nodes. Use it when building nodes that consume log data via send_logs_as.

API

#![allow(unused)]
fn main() {
use adora_log_utils;

// Parse a LogMessage from JSON (as received from send_logs_as)
let log = adora_log_utils::parse_log(json_str)?;

// Parse directly from Arrow input data (convenience for event handlers)
let log = adora_log_utils::parse_log_from_arrow(&data)?;

// Merge multiple log streams into a single timeline
let merged = adora_log_utils::merge_by_timestamp(vec![stream_a, stream_b]);

// Filter by minimum level
let errors = adora_log_utils::filter_by_level(&logs, &min_level);

// Format as JSON (one line, no trailing newline)
let json = adora_log_utils::format_json(&log);

// Format as compact single-line: "<timestamp> <node> <LEVEL>: <message>"
let compact = adora_log_utils::format_compact(&log);

// Format as pretty: "[<timestamp>][<LEVEL>][<node>] <message>"
let pretty = adora_log_utils::format_pretty(&log);
}

Dependency

Add to your sink node’s Cargo.toml:

[dependencies]
adora-log-utils = { workspace = true }

Log Sink Examples

Three example sink nodes demonstrate how to consume logs routed via send_logs_as and forward them to external destinations.

File Sink (examples/log-sink-file/)

Merges log streams from multiple nodes into a single JSONL file. Useful for unified log collection.

nodes:
  - id: sensor
    path: sensor.py
    send_logs_as: log_entries
    inputs:
      tick: adora/timer/millis/200
    outputs:
      - reading
      - log_entries

  - id: processor
    path: processor.py
    send_logs_as: log_entries
    inputs:
      reading: sensor/reading
    outputs:
      - result
      - log_entries

  - id: file_sink
    path: log-sink-file
    inputs:
      sensor_logs: sensor/log_entries
      processor_logs: processor/log_entries
    env:
      LOG_FILE: "./combined.jsonl"

The file sink reads LOG_FILE from the environment (default ./combined.jsonl), parses each incoming Arrow message with adora_log_utils::parse_log_from_arrow(), formats it as JSON, and appends it to the file.

TCP Sink (examples/log-sink-tcp/)

Forwards log entries over a TCP socket to a remote log collector. Useful for embedded systems that lack local filesystems and need to stream logs off-device.

nodes:
  - id: source
    path: source.py
    send_logs_as: log_entries
    inputs:
      tick: adora/timer/millis/500
    outputs:
      - data
      - log_entries

  - id: tcp_sink
    path: log-sink-tcp
    inputs:
      logs: source/log_entries
    env:
      SINK_ADDR: "127.0.0.1:9876"

The TCP sink reads SINK_ADDR from the environment (default 127.0.0.1:9876), connects to the server on startup, and sends each log entry as a JSON line. It reconnects automatically on write failure.

Alert Router (examples/log-sink-alert/)

Splits incoming log entries by severity. All logs are forwarded to the all_logs output; only error and warn logs are forwarded to the alerts output. This enables downstream nodes to handle alerts differently (e.g., trigger notifications, write to a dedicated file).

nodes:
  - id: source
    path: my_node.py
    send_stdout_as: log_entries
    inputs:
      tick: adora/timer/millis/200
    outputs:
      - log_entries

  - id: alert_router
    path: log-sink-alert
    inputs:
      logs: source/log_entries
    outputs:
      - all_logs
      - alerts

The source node uses send_stdout_as to route its stdout lines as Arrow string data. The router parses each log entry with adora_log_utils::parse_log_from_arrow(), checks the level, and uses node.send_output() to forward data to the appropriate outputs. Nodes using the node API can alternatively use send_logs_as to route structured logs from node.log().

Building a Custom Sink

To build your own sink node, follow this pattern:

use adora_node_api::{AdoraNode, Event};

fn main() -> eyre::Result<()> {
    let (_node, mut events) = AdoraNode::init_from_env()?;

    while let Some(event) = events.recv() {
        match event {
            Event::Input { data, .. } => {
                let log = adora_log_utils::parse_log_from_arrow(&data)?;
                // Process the log entry: write to file, send over network, etc.
                let json = adora_log_utils::format_json(&log);
                println!("{json}");
            }
            Event::Stop(_) => break,
            _ => {}
        }
    }
    Ok(())
}

How the Daemon Processes Logs

Understanding the internal pipeline helps with debugging and tuning. For each node, the daemon runs a dedicated async task that processes log lines in order:

Node Process (stdout/stderr)
    |
    v
[1] Capture: lines buffered in mpsc channel (capacity 100)
    |
    v
[2] send_stdout_as: raw line -> Arrow data -> dataflow output
    |
    v
[3] Parse: try JSON structured log, fall back to Stdout-level
    |
    v
[4] min_log_level filter: drop messages below threshold
    |
    v
[5] send_logs_as: LogMessage -> JSON -> Arrow data -> dataflow output
    |
    v
[6] Write JSONL: compact format to log file, track bytes written
    |
    v
[7] Rotation check: if bytes_written >= max_log_size, rotate files
    |
    v
[8] Forward: send LogMessage to display channel (unless ADORA_QUIET)
    |
    v
[9] Sync: fsync log file to disk

Key details:

• Step 2 happens before parsing, so send_stdout_as captures every line including non-structured output
• Step 4 happens before Steps 5-8, so min_log_level suppresses messages from all downstream processing
• Step 5 only fires for successfully parsed structured logs (Step 3 success path)
• Step 8 sends to either a flume channel (adora run direct mode) or the coordinator (distributed mode)
• Step 9 calls sync_all() after every write, ensuring durability at the cost of some I/O overhead

Structured Log Parsing

When a node emits JSON-formatted log output (e.g., from tracing-subscriber with JSON formatting), the daemon extracts:

• level: log severity
• message: the log text
• target: module path
• timestamp: when the log was emitted
• fields: arbitrary key-value pairs
• build_id, dataflow_id, node_id, daemon_id: extracted from fields as fallback

The daemon also sets dataflow_id, node_id, and daemon_id on all messages to ensure they are always present in the log file.

Coordinator Log Streaming Protocol

When a daemon runs under a coordinator (distributed mode), log forwarding works via WebSocket:

1. Daemon -> Coordinator: Each LogMessage is wrapped in DaemonEvent::Log(message) and sent over the daemon’s WebSocket connection
2. Coordinator storage: The coordinator stores/forwards logs
3. CLI subscription: The CLI sends ControlRequest::LogSubscribe { dataflow_id, level } over its WebSocket connection
4. Server-side filtering: The coordinator only forwards messages where msg_level <= subscription_level. This reduces network traffic for filtered subscriptions
5. CLI receive: Messages arrive as serialized LogMessage structs

The --level flag maps to log::LevelFilter:

• stdout -> LevelFilter::Trace (most permissive, receives everything)
• info -> LevelFilter::Info (receives Error, Warn, Info)
• etc.

Complete YAML Reference

nodes:
  - id: sensor
    path: ./target/debug/sensor
    outputs:
      - data
      - raw_output       # for send_stdout_as
      - log_entries       # for send_logs_as

    # Source-level log filtering (daemon-side)
    min_log_level: info          # suppress debug/trace/stdout

    # Route stdout to dataflow
    send_stdout_as: raw_output   # every stdout line becomes a data message

    # Route structured logs to dataflow
    send_logs_as: log_entries    # parsed log entries become data messages

    # Log file rotation
    max_log_size: "50MB"         # rotate when file exceeds 50MB
    max_rotated_files: 5         # keep 5 rotated files (default, range 1-100)

    inputs:
      tick: adora/timer/millis/100

Complete Example

The examples/python-logging/ directory contains a runnable three-node pipeline that exercises every logging feature:

sensor (noisy, high-volume) --> processor (structured logs) --> monitor (log aggregator)

Dataflow configuration highlights:

nodes:
  - id: sensor
    path: sensor.py
    min_log_level: info       # suppress debug noise at source
    max_log_size: "1KB"       # small for demo (triggers rotation quickly)
    inputs:
      tick: adora/timer/millis/50
    outputs:
      - reading

  - id: processor
    path: processor.py
    send_logs_as: log_entries  # route structured logs as data
    inputs:
      reading: sensor/reading
    outputs:
      - result
      - log_entries

  - id: monitor
    path: monitor.py
    inputs:
      logs: processor/log_entries
      reading: sensor/reading

What each node demonstrates:

• sensor – Mixes print() (raw stdout), logging.info(), logging.debug(), and logging.warning(). With min_log_level: info, debug messages are dropped by the daemon before reaching log files. With max_log_size: "1KB", log rotation kicks in after a few seconds.
• processor – Uses send_logs_as: log_entries to route its structured log entries as dataflow data. Raw print() output is not routed (only parsed structured entries are).
• monitor – Subscribes to processor/log_entries and counts warnings/errors, demonstrating in-dataflow log aggregation.

Direct mode (adora run – single process, good for quick testing):

# Basic run
adora run examples/python-logging/dataflow.yml --stop-after 5s

# Only warnings and above
adora run examples/python-logging/dataflow.yml --log-level warn --stop-after 5s

# Per-node overrides
adora run examples/python-logging/dataflow.yml --log-filter "monitor=debug,sensor=warn" --stop-after 5s

# JSON output for machine parsing
adora run examples/python-logging/dataflow.yml --log-format json --stop-after 3s

# Environment variable control
ADORA_LOG_LEVEL=warn adora run examples/python-logging/dataflow.yml --stop-after 5s

Distributed mode (adora up + adora start – coordinator/daemon architecture, required for multi-machine deployments):