Topics

Topics implement a publish-subscribe communication pattern between nodes. A node emits a topic to publish messages, and other nodes consume that topic to receive them.

Use topics for continuous, unidirectional data streams such as sensor readings, camera frames, state updates, or any information that flows over time. Multiple consumers can listen to the same topic simultaneously.

Topics also underpin two higher-level constructs: contract implementation, where a topic’s contract lives in a standalone contract document that producers implement, and pairing, where two node instances exchange topics 1:1 in both directions over a two-role contract. For a side-by-side decision guide across every mechanism, see Choosing a communication pattern.

Emitting a topic

A node that publishes messages declares its topics under interfaces.topics.emits in its peppy.json5. Each topic defines a name, a qos_profile, and a message_format:

Python
Rust

{
  peppy_schema: "node/v1",
  manifest: {
    name: "uvc_camera",
    tag: "v1",
  },
  interfaces: {
    topics: {
      emits: [
        {
          name: "video_stream",
          qos_profile: "sensor_data",
          message_format: {
            header: {
              $type: "object",
              stamp: "time",
              frame_id: "u32"
            },
            encoding: "string",
            width: "u32",
            height: "u32",
            frame: {
              $type: "array",
              $items: "u8"
            }
          }
        }
      ],
    },
  },
  execution: {
    language: "python",
    build_cmd: ["uv", "sync"],
    run_cmd: ["uv", "run", "uvc_camera"]
  },
}

{
  peppy_schema: "node/v1",
  manifest: {
    name: "uvc_camera",
    tag: "v1",
  },
  interfaces: {
    topics: {
      emits: [
        {
          name: "video_stream",
          qos_profile: "sensor_data",
          message_format: {
            header: {
              $type: "object",
              stamp: "time",
              frame_id: "u32"
            },
            encoding: "string",
            width: "u32",
            height: "u32",
            frame: {
              $type: "array",
              $items: "u8"
            }
          }
        }
      ],
    },
  },
  execution: {
    language: "rust",
    build_cmd: ["cargo", "build", "--release"],
    run_cmd: ["./target/release/uvc_camera"]
  },
}

The qos_profile sets the publisher’s delivery behavior for this topic; consumers always subscribe with the standard profile. Available profiles are:

sensor_data: optimized for high-frequency data where occasional drops are acceptable.
standard: balanced defaults suitable for most use cases. This is the default when qos_profile is omitted.
reliable: blocks on congestion instead of dropping, at the cost of higher latency.
critical: the same block-on-congestion guarantee as reliable, dispatched at the highest scheduling priority with express batching, for the lowest latency on safety-critical data.

Publishing messages

After running peppy node sync, the code generator creates a module for each emitted topic under peppygen.emitted_topics (Python) / peppygen::emitted_topics (Rust). Each module exposes declare_publisher(node_runner), which returns a TopicPublisher, and build_message(...), which serializes the message fields into a payload. Declare the publisher once, then publish each message on it:

Python
Rust

import asyncio
import time

from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.emitted_topics import video_stream


async def emit_frames(node_runner: NodeRunner):
    # Declare the publisher once; every publish below then reuses it.
    publisher = await video_stream.declare_publisher(node_runner)
    frame_id = 0
    while True:
        payload = video_stream.build_message(
            video_stream.MessageHeader(stamp=time.time(), frame_id=frame_id),
            "rgb8",
            640,
            480,
            bytes([1, 2, 3]),
        )
        await publisher.publish(payload)
        frame_id = (frame_id + 1) % (2**32)
        await asyncio.sleep(0.1)


async def setup(_params: Parameters, node_runner: NodeRunner) -> list[asyncio.Task]:
    return [asyncio.create_task(emit_frames(node_runner))]


def main():
    NodeBuilder().run(setup)


if __name__ == "__main__":
    main()

use peppygen::emitted_topics::video_stream;
use peppygen::{NodeBuilder, Parameters, Result};
use std::time::Duration;

fn main() -> Result<()> {
    NodeBuilder::new().run(|_args: Parameters, node_runner| async move {
        let node_runner_clone = node_runner.clone();
        tokio::spawn(async move {
            // Declare the publisher once; every publish below is then lock-free.
            let publisher = video_stream::declare_publisher(&node_runner_clone)
                .await
                .expect("failed to declare video_stream publisher");
            let mut frame_id = 0u32;
            loop {
                if let Ok(payload) = video_stream::build_message(
                    video_stream::MessageHeader {
                        stamp: std::time::SystemTime::now(),
                        frame_id,
                    },
                    "rgb8".to_owned(),
                    640,
                    480,
                    vec![1, 2, 3],
                ) {
                    let _ = publisher.publish(payload).await;
                }

                frame_id = frame_id.wrapping_add(1);
                tokio::time::sleep(Duration::from_secs_f64(0.1)).await;
            }
        });

        Ok(())
    })
}

build_message takes each field from the message_format in order (no node_runner argument) and returns the payload that publisher.publish sends; in Rust it returns a Result you unwrap, in Python it returns the payload directly and raises on invalid input. Nested objects become generated types whose fields match the object definition: a struct in Rust (video_stream::MessageHeader), a dataclass in Python (video_stream.MessageHeader).

A producer publishes a single stream regardless of how many consumers are listening, and it never knows about the bindings consumers use to address it. Starting the producer before any consumer (or after them) is equally valid; consumers attach when they show up.

For a topic with a simple message format:

{
  name: "message_stream",
  qos_profile: "sensor_data",
  message_format: {
    message: "string"
  }
}

build_message takes a single string argument:

Python
Rust

from peppygen.emitted_topics import message_stream

publisher = await message_stream.declare_publisher(node_runner)
payload = message_stream.build_message("hello world")
await publisher.publish(payload)

use peppygen::emitted_topics::message_stream;

let publisher = message_stream::declare_publisher(&node_runner).await?;
let payload = message_stream::build_message("hello world".to_owned())?;
publisher.publish(payload).await?;

Consuming a topic

A node that receives messages declares what it consumes under interfaces.topics.consumes. Dependencies are declared in manifest.depends_on and referenced by link_id in the interface:

Python
Rust

{
  peppy_schema: "node/v1",
  manifest: {
    name: "web_video_stream",
    tag: "v1",
    depends_on: {
      nodes: [
        { name: "uvc_camera", tag: "v1", link_id: "uvc_camera" },
      ]
    },
  },
  interfaces: {
    topics: {
      consumes: [
        {
          link_id: "uvc_camera",  // References depends_on.nodes[].link_id
          name: "video_stream",         // Topic name on that node
        },
      ],
    },
  },
  execution: {
    language: "python",
    build_cmd: ["uv", "sync"],
    run_cmd: ["uv", "run", "web_video_stream"]
  },
}

{
  peppy_schema: "node/v1",
  manifest: {
    name: "web_video_stream",
    tag: "v1",
    depends_on: {
      nodes: [
        { name: "uvc_camera", tag: "v1", link_id: "uvc_camera" },
      ]
    },
  },
  interfaces: {
    topics: {
      consumes: [
        {
          link_id: "uvc_camera",  // References depends_on.nodes[].link_id
          name: "video_stream",         // Topic name on that node
        },
      ],
    },
  },
  execution: {
    language: "rust",
    build_cmd: ["cargo", "build", "--release"],
    run_cmd: ["./target/release/web_video_stream"]
  },
}

By default, the node add and node sync commands require the target node to already be in the node stack so the proper interfaces can be generated. The exception is a dependency whose slot admits an empty bound set (cardinality: "zero_or_one" or "zero_or_more"): node add resolves it through the repository cache automatically when it is absent from the stack. If the target node is missing, you will see an error like:

Error: `web_video_stream:v1` depends on `uvc_camera:v1`, but it does not exist in the stack

Pass peppy node sync --include-repositories (-r) to let the daemon fall back to the repository cache for dependencies that aren’t in the stack.

Receiving messages

The code generator creates a module for each consumed topic under peppygen.consumed_topics (Python) / peppygen::consumed_topics (Rust). Every slot-backed module lives at <category>/<link_id>/<member>, keyed on the author’s link_id. The consumed module is therefore consumed_topics/<link_id>/<topic_name>, imported as from peppygen.consumed_topics.uvc_camera import video_stream (Python) / peppygen::consumed_topics::uvc_camera::video_stream (Rust); here the link_id is uvc_camera and the topic is video_stream. Call subscribe once to obtain a held Subscription, then await next for each message:

Python
Rust

import asyncio

from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.consumed_topics.uvc_camera import video_stream


async def receive_frames(node_runner: NodeRunner):
    # Same subscribe() as every cardinality: it covers the slot's complete
    # bound set, which for `one` is exactly one producer. The yielded
    # ProducerRef is constant here, so the loop ignores it.
    subscription = await video_stream.subscribe(node_runner)

    async for _producer, frame in subscription:
        print(f"frame: {frame.width}x{frame.height}")


async def setup(_params: Parameters, node_runner: NodeRunner) -> list[asyncio.Task]:
    return [asyncio.create_task(receive_frames(node_runner))]


def main():
    NodeBuilder().run(setup)


if __name__ == "__main__":
    main()

use peppygen::consumed_topics::uvc_camera::video_stream;
use peppygen::{NodeBuilder, Parameters, Result};

fn main() -> Result<()> {
    NodeBuilder::new().run(|_args: Parameters, node_runner| async move {
        // Same subscribe() as every cardinality: it covers the slot's complete
        // bound set, which for `one` is exactly one producer. The yielded
        // ProducerRef is constant here, so the loop ignores it.
        let mut subscription = video_stream::subscribe(&node_runner).await?;

        while let Some((_, frame)) = subscription.next().await? {
            println!("frame: {}x{}", frame.width, frame.height);
        }

        Ok(())
    })
}

subscribe takes only node_runner (the node runner reference) and returns a Subscription covering the slot’s complete bound producer set: one producer-pinned wire subscription per bound producer, merged client-side behind the single subscribe() API. The shape is identical for every cardinality; only the size of the bound set changes. Awaiting subscription.next() yields:

a (producer, message) pair for each message (Ok(Some((producer, message))) in Rust), where producer is a ProducerRef carrying the publisher’s full (core_node, instance_id) wire identity, so the consumer can tell messages from different bound producers apart, and message is the deserialized message, with fields matching the topic’s message_format.
None (Ok(None) in Rust) once the node is shutting down and no queued message remains, or when every source has closed.
an error if a received payload fails to deserialize: Err(..) in Rust, a raised exception in Python. The error names the message type that failed to decode, and neither tears down the subscription nor shrinks the bound set.

Each underlying subscription holds an in-order buffer, so a message published between next calls is kept, not dropped; loop on next and nothing is lost in the gap between iterations. Message order is preserved independently per producer (no total ordering across producers is promised), and ready producers are merged fairly so one busy producer cannot indefinitely starve another. ProducerRef is Eq + Hash, so it keys per-producer state directly; to follow a single producer, filter on the yielded producer.

There are no call-site producer-targeting arguments at all. Which producers reach a slot is decided entirely by application bindings (see Bindings and routing below): a slot receives messages only from the producer instances bound to it, and each resolved binding carries the producer’s full (core_node, instance_id) wire address, so the slot can never match a producer that merely shares the instance_id on another core node. The producer segments of a keyexpr are never wildcarded: a federated router forwards traffic only for the explicitly bound producers, and every subscriber stays fully pinned (and auditable) in the zenoh admin space. That same full identity is what next returns with every message, so a consumer can pass it straight back into any producer-targeting call (a ProducerRef yielded by the slot’s own subscription is a member of the slot’s bound set by construction).

Every generated consumed topic, service, and action module also exposes a bound-producer accessor for its slot, returning the runtime-resolved, immutable producer binding in application declaration order. The accessor’s return type encodes exactly the guarantee launch validation established for the slot’s cardinality, instead of restating it in comments:

// Generated for every consumed module of a `cardinality: "one"` slot,
// replacing bound_producers() for that slot.
pub fn bound_producer(node_runner: &crate::NodeRunner) -> &ProducerRef;

// Generated for every consumed module of a `cardinality: "zero_or_one"` slot:
// the same singular name returning an Option, `None` wherever the deployment
// wrote the slot vacant.
pub fn bound_producer(node_runner: &crate::NodeRunner) -> Option<&ProducerRef>;

// Generated for every consumed module of a `cardinality: "one_or_more"` slot:
// a never-empty ordered set whose first() is infallible.
pub fn bound_producers(node_runner: &crate::NodeRunner) -> NonEmptyProducers<'_>;

// Generated for every consumed module of a `cardinality: "zero_or_more"` slot:
// a plain, possibly empty slice, so the empty branch is forced by the type
// and is never dead code.
pub fn bound_producers(node_runner: &crate::NodeRunner) -> &[ProducerRef];

Python mirrors the split by name and annotation: a one slot generates bound_producer(node_runner) -> peppylib.ProducerRef, a zero_or_one slot bound_producer(node_runner) -> Optional[peppylib.ProducerRef], and the multi cardinalities generate bound_producers(node_runner) -> List[peppylib.ProducerRef] (documented never-empty for one_or_more).

An observer slot types its own source() / sources() accessor the same four ways against its own cardinality.

Every module sharing the slot’s link_id returns the same set in the same order. The set is fixed when the node starts and is not live discovery: a producer disconnecting does not shrink it, trigger discovery, or cause automatic rebinding. A cardinality change is therefore visible at compile time in the consuming node: the accessor’s type changes and every call site that relied on the old guarantee must be revisited. This is intended; a cardinality flip is a semantic change, not a rebinding.

Receiving messages continuously

To receive messages continuously, subscribe once and loop on next. The held subscription keeps every message in arrival order, so the loop never misses a message published between iterations. Each message is processed in a separate task so that the receive loop is not blocked:

Python
Rust

import asyncio
import sys

from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.consumed_topics.uvc_camera import video_stream


async def process_frame(producer, frame):
    print(
        f"got {frame.width}x{frame.height} frame "
        f"from {producer.core_node}/{producer.instance_id}"
    )


async def receive_frames(node_runner: NodeRunner):
    try:
        subscription = await video_stream.subscribe(node_runner)
    except Exception as e:
        print(f"Failed to subscribe: {e}", file=sys.stderr)
        return

    # The subscription is an async iterator; the loop ends when it closes.
    # Hold a reference to each task until it finishes, otherwise the event
    # loop may garbage-collect it mid-flight.
    tasks: set[asyncio.Task] = set()
    async for producer, frame in subscription:
        task = asyncio.create_task(process_frame(producer, frame))
        tasks.add(task)
        task.add_done_callback(tasks.discard)


async def setup(_params: Parameters, node_runner: NodeRunner) -> list[asyncio.Task]:
    return [asyncio.create_task(receive_frames(node_runner))]


def main():
    NodeBuilder().run(setup)


if __name__ == "__main__":
    main()

use std::sync::Arc;
use peppygen::consumed_topics::uvc_camera::video_stream;
use peppygen::{NodeBuilder, NodeRunner, Parameters, Result};

fn main() -> Result<()> {
    NodeBuilder::new().run(|_args: Parameters, node_runner| async move {
        tokio::spawn(receive_frames(node_runner));
        Ok(())
    })
}

async fn receive_frames(node_runner: Arc<NodeRunner>) {
    let mut subscription = match video_stream::subscribe(&node_runner).await {
        Ok(subscription) => subscription,
        Err(e) => {
            eprintln!("Failed to subscribe: {e}");
            return;
        }
    };

    loop {
        match subscription.next().await {
            Ok(Some((producer, frame))) => {
                tokio::spawn(async move {
                    println!(
                        "got {}x{} frame from {}/{}",
                        frame.width, frame.height, producer.core_node, producer.instance_id
                    );
                });
            }
            Ok(None) => break,
            Err(e) => {
                eprintln!("Error receiving frame: {e}");
                break;
            }
        }
    }
}

In Python the Subscription is also an async iterator: async for producer, message in subscription is equivalent to looping on next and stops cleanly once the subscription closes.

Consuming a multi-cardinality bound set

Nothing changes at the call site when the slot’s cardinality allows several producers: the same single subscribe() covers every bound producer, and the yielded ProducerRef tells the streams apart. ProducerRef is Eq + Hash, so it keys per-producer state directly:

Python
Rust

from peppygen.consumed_topics.camera import video_stream


async def count_frames(node_runner):
    # One subscription covering every producer bound to the `camera`
    # slot; the yielded producer tells per-camera streams apart.
    subscription = await video_stream.subscribe(node_runner)

    frames_per_camera: dict = {}
    async for producer, frame in subscription:
        frames_per_camera[producer] = frames_per_camera.get(producer, 0) + 1
        print(
            f"frame from {producer.instance_id}@{producer.core_node}: "
            f"{frame.width}x{frame.height} "
            f"(total from this camera: {frames_per_camera[producer]})"
        )

use peppygen::consumed_topics::camera::video_stream;
use peppygen::{NodeBuilder, Parameters, ProducerRef, Result};
use std::collections::HashMap;

fn main() -> Result<()> {
    NodeBuilder::new().run(|_args: Parameters, node_runner| async move {
        // One subscription covering every producer bound to the `camera`
        // slot. The shape is identical for every cardinality; only the
        // size of the bound set changes.
        // `one`: the subscription covers exactly one producer.
        // `one_or_more`: the merged subscription covers at least one producer.
        // `zero_or_one` written vacant, or `zero_or_more` bound to nothing:
        // the slot is empty; the subscription then yields nothing until the
        // node stops.
        let mut subscription = video_stream::subscribe(&node_runner).await?;

        let mut frames_per_camera: HashMap<ProducerRef, u64> = HashMap::new();
        while let Some((producer, frame)) = subscription.next().await? {
            *frames_per_camera.entry(producer.clone()).or_insert(0) += 1;
            println!(
                "frame from {}@{}: {}x{} (total from this camera: {})",
                producer.instance_id,
                producer.core_node,
                frame.width,
                frame.height,
                frames_per_camera[&producer],
            );
        }
        Ok(())
    })
}

To follow a single camera instead, filter on the yielded producer (bound_producers() lists the members in binding order).

Bindings and routing

Routing is entirely consumer-side. Producers publish unconditionally and are binding-agnostic; only the consumer’s depends_on slots and the application bindings that fill them decide which producer instance reaches which slot. (Pairing topics are the one exception: a pairing slot is routed by the pair itself, established with --link at instance start, not by a binding.)

A binding takes the form KEY: VALUE, where KEY is a link_id declared in the consumer’s depends_on and VALUE selects the slot’s target producer instance(s). Each bound target creates a private one-way channel from that producer instance to the consumer’s slot: a multi-cardinality slot expands into N independent application-selected bound edges sharing the same link_id. How many targets a slot takes is the slot’s declared cardinality; there is no wildcard fallback and no discovery, only explicitly bound producers.

In a launcher / stack config:

{
  source: { name: "consumer:v1" },
  instances: [{
    instance_id: "my_consumer",
    links: { uvc_camera: "my-camera-instance" },
  }],
}

or, when launching a single node during development:

peppy node run --link uvc_camera@my-camera-instance web_video_stream:v1

Dependency cardinality

Every depends_on.contracts, depends_on.nodes and depends_on.pairing_observers entry may declare a cardinality, constraining how many sources the application may bind to the slot: producers for the first two, observed pairings for the third.

depends_on: {
  contracts: [
    { name: "uvc_camera", tag: "v1", link_id: "camera", cardinality: "one_or_more" },
  ],
}

Cardinality	Valid unique targets	Slot kinds it is valid on	Omitted binding
`one` / omitted	exactly 1	producer, observer	error
`zero_or_one`	0 or 1	producer, observer	error: link it, or write it `{ vacant: "<why>" }`
`one_or_more`	1 or more	producer, observer	error
`zero_or_more`	0 or more	producer, observer	empty set

zero_or_one is the one cardinality a vacancy is legal on, on a producer slot and an observer slot alike: its empty state is written rather than omitted, so a slot nobody mentioned stays an error and “this rig deliberately has none” stays distinguishable from “whoever wrote this launcher forgot it”.

The binding value’s shape mirrors the slot’s cardinality. A scalar slot (one or zero_or_one) takes a scalar string only, exactly as above. A one_or_more or zero_or_more slot takes an array of instance ids; for zero_or_more the empty array [] is a valid definition, equivalent to omitting the binding entirely:

links: {
  camera: ["front_camera", "rear_camera"],
}

Any other combination is an error: an array on a scalar slot (single-element arrays included), a scalar on a multi slot, and an empty array on a one_or_more slot (cardinality unmet). The CLI mirrors the array form by repetition, which accumulates on a multi slot and stays a hard error on a scalar slot. A zero_or_more slot’s empty set has no flag spelling, omission is it; a zero_or_one slot’s is --vacant-link 'SLOT=<why>', because that one has to be said out loud:

peppy node run --link camera@front_camera --link camera@rear_camera web_video_stream:v1

Cardinality applies to a consumer’s dependency slot; it does not permit undeclared or unbound launched node instances. On a producer-binding slot it constrains the application-configured bound set at startup, not runtime availability: a producer disconnecting does not shrink the set, trigger discovery, or cause automatic rebinding. An observer slot differs in exactly that respect: its member set is delivered live by the daemon rather than frozen at startup, so what the set says about each member moves while the node runs. Its size does not, because the planner checks the count and the node re-checks it against the boot seed before setup, so each cardinality’s floor holds on every read there too. depends_on.pairings entries have no cardinality at all (a pairing is strictly 1:1, so there is no set to size; a cardinality key on a participant entry is a manifest error). A participant slot that may run with no peer says so with optional: true instead, and zero_or_one says the same thing for the two slot kinds that carry a cardinality: each is the node’s own statement that a deployment may leave the slot empty, and the deployment still has to write down that it did.

The consuming API is uniform across cardinalities in everything except the bound-producer accessor: topics subscribe to the slot’s complete bound set and yield the producing ProducerRef with every message, and services and actions require the caller to select one explicit, membership-checked member of that set, whatever its size. The accessor itself is cardinality-typed: one generates the singular, infallible bound_producer(), zero_or_one generates bound_producer() returning an Option, one_or_more generates bound_producers() returning a never-empty set whose first() needs no unwrap, and zero_or_more generates bound_producers() returning a possibly empty slice, so the empty branch is forced by the type. An observer slot types its accessor the same four ways: one generates the singular, infallible source(), zero_or_one generates source() returning an Option, one_or_more generates sources() returning a never-empty set, and zero_or_more generates sources() returning a possibly empty one. For an empty slot (a vacant zero_or_one, or a zero_or_more bound to nothing) the subscription simply yields nothing until the node stops.

Worked example: a slot that may bind nothing

A receiver whose greeter slot is declared zero_or_one: a rig that ships with a greeter links one, and a rig that ships without one says so. The manifest is the whole declaration:

Python
Rust

{
  peppy_schema: "node/v1",
  manifest: {
    name: "optional_receiver",
    tag: "v1",
    depends_on: {
      nodes: [
        {
          name: "hello_world_param",
          tag: "v1",
          link_id: "greeter",
          cardinality: "zero_or_one",
        },
      ]
    },
  },
  interfaces: {
    topics: {
      consumes: [
        {
          link_id: "greeter",
          name: "message_stream",
        }
      ],
    }
  },
  execution: {
    language: "python",
    build_cmd: [
      "uv",
      "sync"
    ],
    run_cmd: [
      "uv",
      "run",
      "optional_receiver"
    ]
  },
}

{
  peppy_schema: "node/v1",
  manifest: {
    name: "optional_receiver",
    tag: "v1",
    depends_on: {
      nodes: [
        {
          name: "hello_world_param",
          tag: "v1",
          link_id: "greeter",
          cardinality: "zero_or_one",
        },
      ]
    },
  },
  interfaces: {
    topics: {
      consumes: [
        {
          link_id: "greeter",
          name: "message_stream",
        }
      ],
    }
  },
  execution: {
    language: "rust",
    build_cmd: [
      "cargo",
      "build",
      "--release"
    ],
    run_cmd: [
      "./target/release/optional_receiver"
    ]
  },
}

The generated accessor is scalar and answers with the two cases the cardinality allows, so the node reads its own wiring without a set to interpret:

Python
Rust

import asyncio

from peppygen import NodeBuilder, NodeRunner
from peppygen.parameters import Parameters
from peppygen.consumed_topics.greeter import message_stream


async def setup(_params: Parameters, node_runner: NodeRunner) -> list[asyncio.Task]:
    # The `greeter` slot declares `cardinality: "zero_or_one"`, so its accessor
    # is `Optional`: a producer where the deployment linked one, `None` where it
    # wrote the slot vacant. There is no third case, and no empty list to
    # interpret.
    greeter = message_stream.bound_producer(node_runner)
    if greeter is None:
        print("no greeter bound: running without greetings")
        return []

    print(f"greeter bound: {greeter.instance_id}")
    return [asyncio.create_task(receive_messages(node_runner))]


async def receive_messages(node_runner: NodeRunner):
    subscription = await message_stream.subscribe(node_runner)
    async for producer, message in subscription:
        print(f"Received from {producer.instance_id}: {message.message}")


def main():
    NodeBuilder().run(setup)


if __name__ == "__main__":
    main()

use std::sync::Arc;

use peppygen::consumed_topics::greeter::message_stream;
use peppygen::{NodeBuilder, NodeRunner, Parameters, Result};

fn main() -> Result<()> {
    NodeBuilder::new().run(|_args: Parameters, node_runner| async move {
        // The `greeter` slot declares `cardinality: "zero_or_one"`, so its
        // accessor is an `Option`: `Some` where the deployment linked a
        // producer, `None` where it wrote the slot vacant. There is no third
        // case, and no empty set to interpret.
        match message_stream::bound_producer(&node_runner).cloned() {
            Some(greeter) => {
                println!("greeter bound: {}", greeter.instance_id);
                tokio::spawn(receive_messages(node_runner));
            }
            None => println!("no greeter bound: running without greetings"),
        }
        Ok(())
    })
}

async fn receive_messages(node_runner: Arc<NodeRunner>) {
    let mut subscription = match message_stream::subscribe(&node_runner).await {
        Ok(subscription) => subscription,
        Err(e) => {
            eprintln!("Failed to subscribe: {e}");
            return;
        }
    };

    loop {
        match subscription.next().await {
            Ok(Some((producer, message))) => {
                println!("Received from {}: {}", producer.instance_id, message.message)
            }
            Ok(None) => break,
            Err(e) => {
                eprintln!("Error receiving message: {e}");
                break;
            }
        }
    }
}

Both states are deployments of the same node. Bound:

peppy node run --link greeter@hello_world_param_1 optional_receiver:v1

Empty, with the reason recorded on the instance:

peppy node run --vacant-link 'greeter=this rig ships without a greeter' optional_receiver:v1

Leaving greeter out of the run entirely stays an error, which is what keeps the second command distinguishable from a launcher that forgot the slot.

Per-message routing

For every message a producer publishes, each consumer instance applies one rule on its own slots: the message is delivered to every slot whose bound set names the producer, and to no other slot. A producer named by no binding is ignored by the consumer entirely; there is no wildcard fallback. Wildcard-subscribe-and-filter-in-process is deliberately not offered: a wildcard expresses interest in every same-namespace producer of the contract, so a federated router would forward traffic from unbound producers across the mesh, and the bound set would stop being auditable from the zenoh admin space.

Worked example: `openarm01_backbone`

A consumer that wires two specific depth cameras to dedicated wrist slots:

{
  manifest: {
    name: "openarm01_backbone",
    tag: "v1",
    depends_on: {
      contracts: [
        { name: "depth_camera", tag: "v1", link_id: "wrist_left_camera" },
        { name: "depth_camera", tag: "v1", link_id: "wrist_right_camera" },
      ],
    },
  },
  // ...
}

And the launcher that binds it:

{
  deployments: [
    { source: { name: "depth_camera:v1" }, instances: [
      { instance_id: "left_cam" },
      { instance_id: "right_cam" },
    ]},
    { source: { name: "openarm01_backbone:v1" }, instances: [
      { instance_id: "backbone_inst_1", links: {
        wrist_left_camera: "left_cam",
        wrist_right_camera: "right_cam",
      }},
    ]},
  ],
}

Three contract statements follow from this manifest:

A frame from left_cam arrives only on the wrist_left_camera slot.
A frame from right_cam arrives only on the wrist_right_camera slot.
If the wrist_right_camera binding line were removed, the launch would be rejected: every declared slot except zero_or_more must be bound. A producer named by no binding is simply ignored, but a declared one or one_or_more slot with no binding is an error, not a silent gap.

Dedicated slots (one link_id per camera, as here) and one multi-cardinality slot (cameras: ["left_cam", "right_cam"]) are both valid designs. Dedicated slots give each producer a distinct role in code; a multi slot treats the producers as an N-of-a-kind set behind one API.

Validator rules

The launcher validator runs these checks before the stack starts:

Every KEY must name a declared slot, and every declared slot must resolve. A --link KEY@VALUE (or launcher links: entry) whose KEY matches no depends_on.{nodes,contracts} link_id is rejected; there are no free-form keys. In the other direction, a declared one or one_or_more slot the bindings leave out fails the launch with one error per unfulfilled slot; a zero_or_more slot with no binding resolves to the empty set.
The value’s shape must match the slot’s cardinality. A one slot takes a scalar only (an array is rejected, single-element and empty included), a multi slot takes an array only (a scalar is rejected), and an empty array meets only zero_or_more (on one_or_more it is a cardinality-unmet error). Repeated --link KEY@… flags carry no shape and are checked by count: more than one occurrence on a one slot is rejected. Duplicate targets within one slot are rejected rather than deduplicated.
Every target must satisfy the slot, checked per bound instance. For node slots, each target producer’s (node_name, node_tag) must equal the slot’s declared pair (BindingTargetMismatch otherwise). For contract slots, each target’s manifest.implements must include the slot’s (name, tag) (BindingContractNotImplemented otherwise); see Contract implementation for the implementation rules.
Stack-wide instance_id uniqueness. Every instance_id must be unique across the entire stack, not just within a (node_name, node_tag) group. The --link syntax names producers by instance_id, so reusing one across two different node kinds would make the binding ambiguous.
Bindings are stamped with the producer’s core_node. The --link KEY@instance_id syntax names producers by instance_id alone, but the wire addresses producers by the full (core_node, instance_id) pair, since instance_id is only unique within one stack. The validator stamps each target producer instance’s resolved core_node into the binding (the machine the launcher places that instance on, which is the launching daemon only in the single-machine case), so the runtime never matches on instance_id alone: a slot subscribes with both wire fields set. A producer on another core node that happens to share an instance_id can never feed a bound slot. Bound-set member order is the application declaration order (launcher array order / CLI flag occurrence order), preserved verbatim into the runtime configuration.

Message format types

The message_format supports the following field types (see the message format reference for the full type system, including aliases and nesting rules):

Type	Rust type	Python type
`"bool"`	`bool`	`bool`
`"u8"`	`u8`	`int`
`"u16"`	`u16`	`int`
`"u32"`	`u32`	`int`
`"u64"`	`u64`	`int`
`"i8"`	`i8`	`int`
`"i16"`	`i16`	`int`
`"i32"`	`i32`	`int`
`"i64"`	`i64`	`int`
`"f32"`	`f32`	`float`
`"f64"`	`f64`	`float`
`"string"`	`String`	`str`
`"bytes"`	`Vec<u8>`	`bytes`
`"time"`	`std::time::SystemTime`	`float` (Unix epoch seconds)

Fields can also use complex types:

Object: a nested struct in Rust, a generated dataclass in Python:

header: {
  $type: "object",
  stamp: "time",
  frame_id: "u32"
}

Array: a variable-length list (Vec<T> in Rust, list[T] in Python; an array of u8 maps to Vec<u8> / bytes):
```
frame: {
  $type: "array",
  $items: "u8"
}
```
Fixed-length array: an array with a known size ([T; N] in Rust, a list[T] that must hold exactly N items in Python; the u8 item type is the exception, arriving as bytes in Python regardless of $length):
```
position: {
  $type: "array",
  $items: "f32",
  $length: 3
}
```
Optional: a field that may be absent (Option<T> in Rust, T | None in Python):
```
error_msg: {
  $type: "string",
  $optional: true
}
```