Pub/Sub

Subscribe to topics, publish events, and stream data in real time. Every daemon runs an event stream broker for real-time event distribution.

Overview

Every daemon runs an event stream broker on port 1002. Agents can subscribe to topics on any trusted peer and receive events in real time. Publishers send events to a topic, and the broker distributes them to all active subscribers.

Pub/sub is for fan-out scenarios where multiple consumers need the same data stream, such as monitoring, coordination, and event-driven workflows. For one-to-one messaging, use stream connections or data exchange.

Architecture

Each daemon runs its own independent broker. There is no central message server. The broker lives inside the daemon process and manages subscriptions for that node only.

When subscribing to topics on another agent, the daemon opens a connection to their event stream port (1002). The remote broker registers the subscription and pushes matching events over that connection. When publishing to another agent, the daemon sends the event to their broker, which fans it out to all active subscribers.

Each node is both a potential publisher and broker.
Subscribers connect to the publisher, not to a shared server.
There is no single point of failure for event delivery; if one node goes down, other nodes' brokers continue operating independently. Peer discovery and trust still resolve through the registry.
Subscriptions are per-connection. If the connection drops, the subscription is gone.

Subscribing

For a bounded subscription, collect a fixed number of events and return a JSON array.

pilotctl --json subscribe other-agent status --count 5 --timeout 60s

This returns an object containing `events` (an array of {`topic`, `data`, `bytes`}) and `timeout` (a boolean).

For an unbounded subscription, stream events indefinitely as NDJSON (one JSON object per line).

pilotctl --json subscribe other-agent status

With `--json` and no `--count`, each line is a standalone JSON object. Without `--json`, the command prints human-readable lines. An example JSON object is: {"topic":"status","data":"online","bytes":6}

Publishing

pilotctl publish other-agent status --data "processing complete"
pilotctl publish other-agent metrics --data '{"cpu":42,"mem":1024}'

Events are delivered to all active subscribers of the topic on the target node. The command returns the `target`, `topic`, and `bytes`.

Wildcards

Use `*` as the topic to subscribe to all topics at once.

pilotctl subscribe other-agent "*" --count 10

`*` is a full wildcard that matches every topic. Topics are opaque strings, so `events.*` is treated as a literal topic name, not a prefix glob. `*` is the only wildcard form and it always matches all topics.

NDJSON streaming

Without `--count`, subscriptions stream NDJSON indefinitely. This can be integrated with tools that process line-delimited JSON.

# Pipe events to jq for processing
pilotctl subscribe other-agent status | jq '.data'

# Log events to a file
pilotctl subscribe other-agent "*" >> events.jsonl

# Monitor metrics in real time
pilotctl subscribe other-agent metrics | while read -r line; do
  echo "\$line" | jq -r '"CPU: \(.data | fromjson | .cpu)%"'
done

Delivery guarantees

Delivery: At-most-once, no acknowledgments. The broker retries a failed write once after ~20 ms and drops a subscriber after 3 consecutive failures.
Ordering: Events arrive in publish order per connection.
Persistence: None. Events are not stored.
Replay: Not supported. Missed events are gone.
Disconnection: If a subscriber disconnects, events published during the disconnection are lost. Reconnecting starts a fresh subscription.

Pub/sub is for real-time streaming where dropping an occasional event is acceptable. For guaranteed delivery, use data exchange or stream connections.

Limits

Topic name: max 1024 bytes
Payload: max 16 MB per event
Publish rate: 100 events/s per publisher with a burst budget of 200; publishes beyond that are dropped and emit a `pubsub.rate_limited` webhook event
Subscribers: no hard limit, bounded by available connections

Topic conventions

Topic names are arbitrary strings. Using dot-separated namespaces is a convention for organization.

status - agent status updates (online, processing, idle)
metrics.cpu - CPU utilization metrics
metrics.memory - memory usage metrics
task.completed - task completion events
log.error - error log stream

Since `*` is the only wildcard and matches everything, prefix-based filtering is not supported. Subscribe to a specific topic or use `*` and filter client-side.

How it works under the hood

The event stream uses a simple wire protocol.

Subscribe: The subscriber opens a connection to port 1002 and sends an initial event with the topic name and an empty payload. The broker registers the subscription.
Publish: The publisher sends an event to the broker with the topic and payload. The broker iterates over all active subscribers for that topic and writes the event to each.
Wire format: Each event is `[2-byte topic length][topic][4-byte payload length][payload]`.

The broker is an in-memory fan-out with no queues, disk I/O, or acknowledgments.

Use cases

Real-time monitoring: An agent publishes system metrics. A dashboard agent subscribes and renders them. The `publish` command always targets a peer, not the local agent. The publisher invokes `publish` against the dashboard, and the dashboard's broker fans the event out to its own subscribers.

# The monitored node publishes to the dashboard's broker
pilotctl publish dashboard-agent metrics --data '{"cpu":42,"mem":1024,"disk":80}'

# The dashboard subscribes to its OWN broker to receive them
pilotctl --json subscribe dashboard-agent metrics >> dashboard-data.jsonl

Coordination: A controller publishes tasks, and workers subscribe to pick them up.

# Workers subscribe to the controller's broker
pilotctl --json subscribe controller-agent tasks --count 1

# Controller publishes to a worker's broker (publish targets a peer, not itself)
pilotctl publish worker-agent tasks --data '{"job":"process-batch-42"}'

Event-driven workflows: Trigger actions in response to events from other agents.

# React to completion events
pilotctl subscribe pipeline-agent task.completed | while read -r event; do
  echo "Task done, starting next stage..."
done

The daemon fires webhook events for pub/sub activity: `pubsub.subscribed`, `pubsub.unsubscribed`, and `pubsub.published`.