Subscriptions & Presence

This describes how processes subscribe to events from a running agent (or filesystem) in Sagents, and how presence makes those subscriptions resilient to producer restarts.

Mental Model

Per-agent events are delivered directly from each AgentServer to its subscribers via Sagents.Publisher. A subscriber registers itself by calling the producer's subscribe API (GenServer.call), and from then on every event is delivered by plain send/2 from the producer to the subscriber pid. There is no Phoenix.PubSub topic in the path — the producer's named process is the rendezvous point.

Each producer maintains its own subscriber list, partitioned into channels. For an AgentServer, two channels exist:

FileSystemServer exposes a single :main channel for file change events.

The producer monitors each subscriber, so subscriber departure cleans up the producer's bookkeeping automatically. To detect producer death, the subscriber side uses its own monitor on the producer pid (see Subscriber-Side APIs).

Phoenix.PubSub is used for only two narrow purposes:

1. Agent-discovery presence — agents that come online publish their presence on a known topic so subscribers can discover them and attach.
2. Application-level viewer presence — optional, for hosts that want the agent to shut down when no one is watching.

These are explained below.

Subscriber-Side APIs

Direct: AgentServer.subscribe/1

The raw producer-side API:

{:ok, server_pid, monitor_ref} = AgentServer.subscribe("conversation-123")
{:ok, server_pid, monitor_ref} = AgentServer.subscribe("conversation-123", :debug)

# Returns {:error, :process_not_found} if no AgentServer is running
# under that agent_id.

Events arrive at the subscriber as {:agent, event} (main) or {:agent, {:debug, event}} (debug).

def handle_info({:agent, event}, socket) do
  {:noreply, handle_agent_event(event, socket)}
end

The producer monitors the subscriber, so a dying subscriber is removed automatically. It does not install the reverse monitor for you — if you want to detect producer death (e.g., to re-subscribe after a crash), call Process.monitor(server_pid) on the returned pid yourself.

AgentServer.unsubscribe/1 always returns :ok, even if the producer is no longer running.

FileSystemServer.subscribe/1 and unsubscribe/1 follow the same shape — events arrive as {:file_system, event}.

Recommended: Sagents.Subscriber

For long-lived hosts (LiveViews, GenServer bridges) that need crash-safe subscriptions across an agent's full lifecycle, use Sagents.Subscriber. It provides consumer-side bookkeeping over the raw API:

• Stores subscription metadata in a host-owned subs() map.
• Installs Process.monitor(server_pid) on your behalf, so the host receives :DOWN if the producer crashes.
• Tracks a :pending state when subscribing to an agent that isn't running yet — the subscription auto-upgrades to :subscribed once the agent appears (see Agent-Discovery Presence).
• Provides handle_publisher_down/3 and handle_presence_diff/3 helpers that the host's handle_info/2 clauses delegate to.

API surface:

# Bookkeeping primitives (all take and return a subs map)
subs = Subscriber.subscribe_to_agent(subs, agent_id)            # :main channel
subs = Subscriber.subscribe_to_agent(subs, agent_id, :debug)
subs = Subscriber.subscribe_to_filesystem(subs, scope_key)
subs = Subscriber.unsubscribe_from_agent(subs, agent_id)
subs = Subscriber.unsubscribe_from_filesystem(subs, scope_key)

# Recovery handlers (call from host's handle_info/2)
{:matched, subs} | :no_match = Subscriber.handle_publisher_down(subs, ref, reason)
subs = Subscriber.handle_presence_diff(subs, Subscriber.presence_topic(), payload)

# Topic to subscribe to for agent-arrival events
topic = Subscriber.presence_topic()

The subs() map is opaque to callers — store it in your socket assigns or GenServer state and pass it back into the next call.

AgentServer.subscribe/1 is enough for short-lived or one-off subscribers. Reach for Sagents.Subscriber when the host must survive an agent restart or be ready to subscribe before the agent exists.

Agent-Discovery Presence

When an AgentServer starts with the presence_module: option, it tracks itself on the constant topic "agent_server:presence" and updates its presence metadata on every status change. This is the mechanism that lets subscribers learn an agent has come online so they can attach (or re-attach after a restart).

AgentServer.start_link(
  agent: agent,
  pubsub: {Phoenix.PubSub, MyApp.PubSub},  # required when presence_module is set
  presence_module: MyApp.Presence
)

Subscribers consume agent arrivals by subscribing to Sagents.Subscriber.presence_topic() (which returns the same constant string) via Phoenix.PubSub.subscribe/2, and then forwarding each presence_diff broadcast to Sagents.Subscriber.handle_presence_diff/3. For each entry in payload.joins, the subscriber upgrades any matching :pending subscriptions to :subscribed and re-installs the producer monitor.

This is what makes the load path safe: a LiveView opening a conversation can call Subscriber.subscribe_to_agent(subs, agent_id) even if the agent isn't running yet — the sub is recorded as :pending, and the next time the agent starts, the presence_diff fulfillment closes the loop.

Worked LiveView example

defmodule MyAppWeb.ChatLive do
  use MyAppWeb, :live_view
  alias Sagents.Subscriber

  @impl true
  def mount(%{"id" => conversation_id}, _session, socket) do
    agent_id = "conversation-#{conversation_id}"

    socket =
      if connected?(socket) do
        # Subscribe to agent-arrival presence first, so we never miss a join.
        Phoenix.PubSub.subscribe(MyApp.PubSub, Subscriber.presence_topic())

        # Subscribe to the agent itself. If it isn't running yet, this is
        # recorded as :pending and resolved by the next presence_diff.
        subs = Subscriber.subscribe_to_agent(%{}, agent_id)
        assign(socket, sagents_subs: subs)
      else
        assign(socket, sagents_subs: %{})
      end

    {:ok, assign(socket, agent_id: agent_id)}
  end

  # Producer crashed — Sagents.Subscriber flips the entry back to :pending.
  @impl true
  def handle_info({:DOWN, ref, :process, _pid, reason}, socket) do
    subs = socket.assigns.sagents_subs

    socket =
      case Subscriber.handle_publisher_down(subs, ref, reason) do
        {:matched, new_subs} -> assign(socket, sagents_subs: new_subs)
        :no_match -> socket
      end

    {:noreply, socket}
  end

  # Agent came online — Sagents.Subscriber upgrades :pending → :subscribed.
  @impl true
  def handle_info(
        %Phoenix.Socket.Broadcast{event: "presence_diff", payload: payload},
        socket
      ) do
    new_subs =
      Subscriber.handle_presence_diff(
        socket.assigns.sagents_subs,
        Subscriber.presence_topic(),
        payload
      )

    {:noreply, assign(socket, sagents_subs: new_subs)}
  end

  # Actual agent events.
  @impl true
  def handle_info({:agent, event}, socket) do
    {:noreply, handle_agent_event(event, socket)}
  end
end

The two recovery clauses are six lines of dispatch. Apps that ship a generator (like the demo) typically wrap them in a thin AgentLiveHelpers module so the LiveView reads as one-liners. See mix sagents.gen.live_helpers for the template.

Multiple Subscribers

Multiple processes can subscribe to the same agent on the same channel. Each gets its own monitor ref and receives every event delivered on that channel:

# Tab 1
{:ok, _pid, _ref} = AgentServer.subscribe("conversation-123")

# Tab 2 (same or different user)
{:ok, _pid, _ref} = AgentServer.subscribe("conversation-123")

# Both receive every {:agent, event} delivered on :main.

This enables multiple browser tabs on the same conversation, shared conversations between users, and admin dashboards monitoring user conversations.

Event Reference

All events are wrapped as {:agent, event} on :main, or {:agent, {:debug, event}} on :debug.

Status events

{:agent, {:status_changed, status, data}}

def handle_agent_event({:status_changed, status, data}, socket) do
  case {status, data} do
    {:running, nil} ->
      socket |> assign(status: :running) |> assign(streaming_content: "")

    {:idle, nil} ->
      assign(socket, status: :idle)

    {:interrupted, interrupt_data} ->
      socket
      |> assign(status: :interrupted, interrupt: interrupt_data)
      |> push_event("show_approval_modal", %{})

    {:error, reason} ->
      socket
      |> assign(status: :error)
      |> put_flash(:error, "Agent error: #{inspect(reason)}")

    {:cancelled, nil} ->
      assign(socket, status: :cancelled)
  end
end

LLM message events

{:agent, {:llm_deltas, [%MessageDelta{}, ...]}}   # streaming tokens
{:agent, {:llm_message, %Message{}}}              # complete message
{:agent, {:llm_token_usage, %TokenUsage{}}}       # token usage info

TODO events

{:agent, {:todos_updated, [%Todo{}, ...]}}

The TODO list is a complete snapshot, not a diff — replace the local list on each event.

Tool events

{:agent, {:tool_call_identified, tool_info}}
# tool_info contains: %{call_id, name, display_text, arguments}

{:agent, {:tool_execution_started, tool_info}}

{:agent, {:tool_execution_completed, call_id, %ToolResult{}}}

{:agent, {:tool_execution_failed, call_id, error}}

Display message events

{:agent, {:display_message_saved, display_message}}
{:agent, {:display_message_updated, display_message}}

Fired when a display message has been persisted to the database (requires DisplayMessagePersistence to be configured).

Shutdown event

{:agent, {:agent_shutdown, %{reason: reason, metadata: map}}}

Debug events

Available on the :debug channel after AgentServer.subscribe/2.

{:agent, {:debug, {:agent_state_update, %State{}}}}
{:agent, {:debug, {:middleware_action, module, data}}}
{:agent, {:debug, {:llm_error, error}}}
{:agent, {:debug, {:subagent, subagent_id, event}}}

The sagents_live_debugger package consumes these events to provide a real-time debugging dashboard covering agent state, middleware actions, and sub-agent hierarchies.

Publishing Custom Events From Middleware

Middleware can publish events on either channel via the AgentServer API:

defmodule MyMiddleware do
  @behaviour Sagents.Middleware

  def after_model(state, _config) do
    AgentServer.publish_event_from(
      state.agent_id,
      {:my_custom_event, %{data: "something"}}
    )

    AgentServer.publish_debug_event_from(
      state.agent_id,
      {:middleware_action, __MODULE__, {:custom_action, "details"}}
    )

    {:ok, state}
  end
end

Both functions are non-blocking casts. With zero subscribers on the target channel, the broadcast is a no-op.

Viewer-Presence Shutdown

This is a separate, optional mechanism: it lets the AgentServer shut itself down when the application says no one is watching. It is unrelated to the Agent-Discovery Presence layer above.

Pass presence_tracking: at start time, supplying the application's presence module and the topic the host LiveView (or other client) tracks its viewers on:

AgentServer.start_link(
  agent: agent,
  pubsub: {Phoenix.PubSub, MyApp.PubSub},
  presence_tracking: [
    enabled: true,
    presence_module: MyApp.Presence,
    topic: "conversation:#{conversation_id}",
    check_delay: 1_000  # ms before shutdown after presence drops to zero
  ]
)

The agent then:

1. Subscribes to the supplied topic via Phoenix.PubSub to receive presence_diff broadcasts.
2. On each broadcast, checks whether presence_module.list(topic) is empty.
3. If status is :idle and the list is empty, schedules :shutdown_no_viewers after check_delay.
4. Subscribers receive {:agent, {:agent_shutdown, %{reason: :no_viewers}}} and the process terminates.

The host's side: track and release on the switch

Sagents.Presence.track/4 tracks self(), and a subscriber process outlives the conversation it is showing. A LiveView that opens conversation A, then switches to B, is the same process throughout: the entry on A stays until that process dies. So the release belongs on the conversation switch, not on mount/3 plus terminate/2.

Phoenix.Presence does reap an entry when the tracked process dies, but that is a backstop, not the mechanism. It is also why the mistake is easy to miss: closing the tab cleans up perfectly, so a session that tracked five conversations and released none looks fine from the outside while five agents sit past the check that exists to release them.

The generated helpers do this for you. Open a conversation through AgentLiveHelpers.load_conversation/3, or through enter_conversation/3 when the caller already holds the conversation and no database read is wanted:

# The load path: reads the conversation, subscribes, tracks the viewer, and
# releases whatever entry this socket held before.
{:ok, socket} =
  AgentLiveHelpers.load_conversation(socket, conversation_id,
    scope: socket.assigns.current_scope,
    user_id: socket.assigns.current_scope.user.id
  )

# A conversation the caller just created or was handed.
socket =
  AgentLiveHelpers.enter_conversation(socket, conversation,
    user_id: socket.assigns.current_scope.user.id
  )

# Leaving without entering another: a reset, a deleted thread, a closed panel.
socket = AgentLiveHelpers.reset_conversation(socket)

Every path that opens a conversation goes through one funnel that leaves the previous one first, which is what keeps an entry from being taken without the previous one being handed back.

Two rules keep it that way:

• Never assign :conversation_id directly. A path that assigns it and then navigates can short-circuit the host's own same-id guard, so load_conversation/3 never runs for that conversation, and the hand-rolled track_conversation_viewer/3 call such paths end up making is the one nothing releases.
• Never call Coordinator.track_conversation_viewer/3 or untrack_conversation_viewer/2 yourself. The session state records what Presence holds so a leaving path can release it without a viewer id it may no longer have, and calling the coordinator behind that record makes the two drift apart.

Viewing several conversations from one process

Presence bookkeeping is a set, not a slot. One process can hold an entry per conversation, which is what a split view, a dashboard with a row per running agent, or a panel of threads each backed by its own forked agent all need.

The generated LiveView helpers above model a socket showing one conversation, named by :conversation_id, and opening another replaces it. A host that shows several at once is not restricted to that. It works the set directly, on its own state map, through AgentSubscriberSession:

# Incremental.
changes = AgentSubscriberSession.add_tracked_viewer(state, conversation_id, user_id)
changes = AgentSubscriberSession.remove_tracked_viewer(state, conversation_id)
changes = AgentSubscriberSession.clear_tracked_viewers(state)

# Declarative: whatever this process was viewing, it is viewing exactly these now.
changes = AgentSubscriberSession.sync_tracked_viewers(state, %{
  main_id => user_id,
  note_a_id => user_id,
  note_b_id => user_id
})

Prefer sync_tracked_viewers/2 wherever the viewed set is derivable from state the host already keeps. The incremental pair can accumulate entries, because nothing in the library knows when one of the host's panels went away, and declaring the set removes the question. It is also idempotent: re-declaring the same set issues no calls at all, which matters because an untrack followed by a track of the same conversation is a leave broadcast, and an idle agent that acts on it schedules the very shutdown the entry exists to prevent.

One caveat for a host going down this road today: main-channel events are delivered as a bare {:agent, event} and do not name the agent that sent them, so two agents streaming into one mailbox are not separable. Until that changes, route each conversation's events through its own subscriber process (pass it to Sagents.Subscriber.subscribe_to_agent/4) or its own LiveView. Viewer presence has no such limit: entries are per topic and one process can hold as many as it likes.

Non-LiveView hosts use exactly these same functions on their own state map.

For lifecycle details (inactivity timeout, manual stop, grace periods), see lifecycle.md.

Event Ordering

Events are delivered in order per agent (each producer's send/2 is sequential), but:

• No ordering guarantee across different agents.
• Streaming deltas may batch multiple tokens.
• State updates are eventual (not transactional).

Recommended consumer patterns:

• Use {:llm_message, msg} for final message content rather than accumulating {:llm_deltas, _} (deltas are for live UI; the complete message is the source of truth).
• Treat {:todos_updated, todos} as a snapshot replacement.
• Always handle {:agent_shutdown, _} (and :DOWN if you installed your own monitor) so the UI can react to the agent going away.

Best Practices

Always check connected?/1 in LiveView mount

Subscriptions cost a GenServer.call to the producer. Don't pay for them during the disconnected initial render:

def mount(_params, _session, socket) do
  socket =
    if connected?(socket) do
      assign(socket, sagents_subs: Subscriber.subscribe_to_agent(%{}, agent_id))
    else
      assign(socket, sagents_subs: %{})
    end

  {:ok, socket}
end

Subscribe to presence_topic before subscribing to the agent

Order matters when an agent might be starting up concurrently. Subscribe to Sagents.Subscriber.presence_topic() first, then to the agent — that way, if the agent registers between your two calls, you still receive the presence_diff and can fulfill any :pending entry.

Let Sagents.Subscriber handle recovery for you

Don't roll your own Process.monitor + presence_diff loop. The handle_publisher_down/3 and handle_presence_diff/3 helpers already encode the :pending ↔ :subscribed state machine. Six lines of handle_info/2 dispatch is all the host needs.

Explicit unsubscription when scope changes

Subscriptions are cleaned up automatically when the subscriber process dies. But when a single LiveView switches between conversations, unsub the previous one explicitly to free the producer's bookkeeping slot sooner:

def handle_event("close_conversation", _params, socket) do
  new_subs =
    Subscriber.unsubscribe_from_agent(
      socket.assigns.sagents_subs,
      socket.assigns.agent_id
    )

  {:noreply, assign(socket, sagents_subs: new_subs, agent_id: nil)}
end