Event pub/sub that connects actors without coupling.
Two event layers:
- Layer A: Domain events (emitted by your application)
- Layer B: Runtime events (emitted by Cyclium internals)
Built on Phoenix.PubSub. The consuming app provides the PubSub process
via Cyclium.Supervisor opts.
Runtime event payloads are atom-keyed maps (%{episode_id: ..., actor_id: ...}), and PubSub delivers the term unchanged, on any node — match
payload[:actor_id], not payload["actor_id"]. The exception is what lies
inside trigger_ref: it is the episode's stored JSON map, so its keys are
strings. Read it with subject/2.
Summary
Functions
Publish to both the global topic and the event-specific topic, so both broad and targeted subscribers receive the event.
The part of an episode's trigger_ref that lifecycle broadcasts carry.
Publish an event to the bus. All subscribers receive {:bus, event_type, payload}.
Broadcast a cancel request for an episode to whichever node runs it. Backs
Cyclium.Episodes.request_cancel/2.
Read a value from an episode event's trigger subject.
Subscribe the calling process to all bus events.
Messages arrive as {:bus, event_type, payload}.
Subscribe to a specific event type topic.
Messages arrive as {:bus, event_type, payload}.
Subscribe the calling process to cancel requests for one episode.
Functions
Publish to both the global topic and the event-specific topic, so both broad and targeted subscribers receive the event.
The part of an episode's trigger_ref that lifecycle broadcasts carry.
Every Cyclium.Bus.subscribe/0 subscriber in the cluster receives every
lifecycle event, so the conversation turn's message, the principal map and
any history are dropped before broadcast; the subject payload and the
correlation ids stay, so subject/2 keeps working. Read the full
trigger_ref from the episode row when a consumer needs it.
Publish an event to the bus. All subscribers receive {:bus, event_type, payload}.
Broadcast a cancel request for an episode to whichever node runs it. Backs
Cyclium.Episodes.request_cancel/2.
Read a value from an episode event's trigger subject.
Every episode lifecycle event — episode.queued, episode.started,
episode.completed, episode.failed, episode.canceled and
episode.interrupted — carries the run's expectation_id and trigger_ref,
the subject the run is about, whichever path ends the run: the runner, a
crashed or exited task, a recovery sweep, a replayed workflow step, a cancel,
or a queued run shed at capacity. (episode.dropped names no episode — the
run was never created — so it carries expectation_id only.) This reads
trigger_ref["payload"][key] so a subscriber can tell which subject an event
names without hand-rolling the nested access or a DB read. Returns nil when the payload has no trigger_ref, no "payload", or
no such key.
def handle_info({:bus, "episode.completed", payload}, socket) do
case Cyclium.Bus.subject(payload, "thing_id") do
nil -> {:noreply, socket}
thing_id -> {:noreply, refresh_row(socket, thing_id)}
end
endSubscribe the calling process to all bus events.
Messages arrive as {:bus, event_type, payload}.
Subscribe to a specific event type topic.
Messages arrive as {:bus, event_type, payload}.
Subscribe the calling process to cancel requests for one episode.
A running EpisodeTask subscribes so a "stop" reaches it on whatever node
runs the episode (PubSub is cluster-wide), landing as {:cyclium_cancel, reason} handled at the next step boundary.