Raxol.Core.Runtime.EmitBus (Raxol v2.6.1)

View Source

Lightweight, package-neutral pub/sub for TEA runtime events.

This is the keystone seam for the harness event stream. The Dispatcher publishes one neutral event map at each of its two model-fold sites (process_app_update/3 and process_command_result/3); any process that subscribes by session_id receives those maps as {:emit_bus, session_id, event} messages.

The bus deliberately knows nothing about the agent contract. It carries a plain map so the main raxol package never depends on raxol_agent (a back-dep would be circular). The raxol_agent side subscribes via Raxol.Agent.EmitBridge, translates each neutral map into a Raxol.Agent.Contract.Event, and re-emits it through Raxol.Agent.SessionStreamer. Producer (Dispatcher) and consumer (agent surfaces) meet only at this map shape.

Neutral event shape

%{
  session_id: term(),
  family: :loop | :meta,
  type: atom(),
  tier: :ephemeral | :durable,
  turn_id: String.t() | nil,
  payload: map(),
  ts: integer()   # System.system_time(:microsecond)
}

Transport

A :duplicate-keyed Registry keyed by session_id. It is started as part of the application supervision tree (see Raxol.Application). When the registry is not running (e.g. a minimal/headless boot), subscribe/1, unsubscribe/1, and publish/1 degrade to :ok no-ops rather than crash — matching the resilience of Dispatcher.broadcast/2.

Summary

Types

event()
family()
tier()

Functions

build(session_id, type, tier, payload, opts \\ [])

Build a neutral event map.

child_spec(opts)

Child spec for the backing registry. Add Raxol.Core.Runtime.EmitBus to a supervision tree to start it.

publish(event)

Publish a neutral event to every subscriber of its session_id.

registry_name()

The registered name of the backing registry.

subscribe(session_id)

Subscribe the calling process to events for session_id.

unsubscribe(session_id)

Unsubscribe the calling process from session_id.

Types

event()

@type event() :: %{
  session_id: term(),
  family: family(),
  type: atom(),
  tier: tier(),
  turn_id: String.t() | nil,
  payload: map(),
  ts: integer()
}

family()

@type family() :: :loop | :meta

tier()

@type tier() :: :ephemeral | :durable

Functions

build(session_id, type, tier, payload, opts \\ [])

@spec build(term(), atom(), tier(), map(), keyword()) :: event()

Build a neutral event map.

Options: :family (default :loop) and :turn_id (default nil). ts is stamped at build time.

child_spec(opts)

@spec child_spec(term()) :: Supervisor.child_spec()

Child spec for the backing registry. Add Raxol.Core.Runtime.EmitBus to a supervision tree to start it.

publish(event)

@spec publish(event()) :: :ok

Publish a neutral event to every subscriber of its session_id.

No-ops when the registry is not running.

registry_name()

@spec registry_name() :: atom()

The registered name of the backing registry.

subscribe(session_id)

@spec subscribe(term()) :: :ok

Subscribe the calling process to events for session_id.

Events arrive as {:emit_bus, session_id, event} messages. No-ops when the registry is not running.

unsubscribe(session_id)

@spec unsubscribe(term()) :: :ok

Unsubscribe the calling process from session_id.