LemonCore.
View Source
Process-safe PubSub for cross-app event communication.
The Bus provides a simple publish/subscribe mechanism for broadcasting events across the Lemon umbrella apps.
Topic Contract
Platform-contract topics. Their payloads are typed by LemonCore.Events, and changing one
after publication is a semver-major change:
"run:<run_id>"- Events for a specific run"session:<session_key>"- Events for a specific session, including everything the router forwards from that session's run topics"cron"- Cron/automation events"exec_approvals"- Execution approval request/resolution events"system"- System-wide events (config reload, secret changes, talk mode)"goals"- Durable goal lifecycle events"routing_feedback"- Finalized run feedback samples for router-owned model selection
Topics whose publisher and every subscriber live in one app are not platform contract.
They are listed here only so this is a complete map of what is on the bus: "nodes" and
"presence" (lemon_control_plane), "run_graph:<run_id>" and
"parent_question:<request_id>" (coding_agent), and "sim:<id>", "sim:<id>:decisions",
"sim:lobby", "arena:<domain>:league" plus the hosted-game and philosopher-chat topics
(lemon_sim / lemon_sim_ui).
"channels" and "logs" were listed here as stable topics until Phase 3.1, but no module
has ever published to either. "channels" is reachable only through the control-plane
event-injection methods; "logs" is reachable by nothing.
The full catalog, including every publisher and subscriber, is in
docs/platform/bus-events.md in the Lemon repository.
Examples
# Subscribe to run events
LemonCore.Bus.subscribe("run:abc-123")
# Receive events in the subscribing process
receive do
%LemonCore.Event{type: :delta, payload: payload} ->
IO.puts("Received delta: #{inspect(payload)}")
end
# Broadcast an event
event = LemonCore.Event.new(:delta, %{text: "Hello"})
LemonCore.Bus.broadcast("run:abc-123", event)Backends
phoenix_pubsub is an optional dependency. When it is available (the case
for the Lemon runtime, and for anything running Phoenix) the Bus is a thin
wrapper over Phoenix.PubSub and broadcasts reach the whole cluster.
When it is not, the Bus falls back to a Registry with duplicate keys,
started by LemonCore.Application under the same supervision tree. The four
operations behave identically on the local node; the fallback does not
cross node boundaries. Distributed deployments should depend on
phoenix_pubsub explicitly.
Summary
Functions
Which backend is in use: :pubsub or :registry.
Broadcast an event to all subscribers of a topic.
Broadcast a typed event, checking the payload against LemonCore.Events.
Broadcast an event from the calling process (excluding self).
Whether phoenix_pubsub backs the Bus, as opposed to the Registry fallback.
Build a run topic from a run_id.
Whether the active backend's process is running, so a broadcast would be delivered.
Build a session topic from a session_key.
Subscribe the calling process to a topic.
Unsubscribe the calling process from a topic.
Functions
@spec backend() :: :pubsub | :registry
Which backend is in use: :pubsub or :registry.
Detected at runtime on first use and memoized, because phoenix_pubsub being
compiled alongside lemon_core does not mean it is on the code path of every
app that depends on lemon_core — optional dependencies are not inherited
transitively. config :lemon_core, :bus_backend, :registry forces the
fallback even where Phoenix.PubSub is available, which is how tests
exercise it.
@spec broadcast(topic :: binary(), event :: LemonCore.Event.t() | term()) :: :ok
Broadcast an event to all subscribers of a topic.
The event can be a LemonCore.Event struct or any term.
Returns :ok on success.
@spec broadcast_event( topic :: binary(), type :: atom(), payload :: term(), meta :: map() | nil ) :: :ok
Broadcast a typed event, checking the payload against LemonCore.Events.
This is the publishing path for contract topics. When the event type is registered and
the payload is not its struct, the mismatch raises in :dev and :test and is passed
through untouched in :prod — a malformed payload should fail a developer's test run,
not a user's agent run.
Bus.broadcast_event(Bus.run_topic(run_id), :run_started, %Events.RunStarted{...})Unregistered types are broadcast as-is, so app-internal topics can use this function too.
@spec broadcast_from(topic :: binary(), event :: LemonCore.Event.t() | term()) :: :ok
Broadcast an event from the calling process (excluding self).
@spec pubsub?() :: boolean()
Whether phoenix_pubsub backs the Bus, as opposed to the Registry fallback.
Build a run topic from a run_id.
@spec running?() :: boolean()
Whether the active backend's process is running, so a broadcast would be delivered.
Publishers that are optional about eventing (they emit only as a side effect of some
other durable write) should gate on this rather than on Process.whereis/1 of a
particular backend — checking for LemonCore.PubSub specifically makes the publisher
silently inert under the Registry fallback.
Build a session topic from a session_key.
@spec subscribe(topic :: binary()) :: :ok
Subscribe the calling process to a topic.
Returns :ok on success.
@spec unsubscribe(topic :: binary()) :: :ok
Unsubscribe the calling process from a topic.
Returns :ok on success.