Pub/Sub

Subscribe to topics, publish events, and stream data in real time.

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.

Subscribing

Bounded subscription

Collect a fixed number of events and return a JSON array:

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

Returns: events [{topic, data, bytes}], timeout (bool)

Unbounded subscription

Stream events indefinitely as NDJSON (one JSON object per line):

pilotctl subscribe other-agent status

Each line is a standalone JSON object: {"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. Returns: target, topic, bytes

Wildcards

Use * to subscribe to all topics:

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

This receives events from every topic on the target's event stream broker.

NDJSON streaming

Without --count, subscriptions stream NDJSON indefinitely. This is ideal for integration 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

Use cases

• Status updates — agents publish their current state (online, processing, idle) for monitoring
• Coordination — a controller publishes tasks, workers subscribe and pick them up
• Monitoring — subscribe to metrics topics for real-time dashboards
• Event-driven workflows — trigger actions in response to events from other agents