Using Sagents with a React Front End

Sagents was designed with LiveView in mind. Agent events are plain Elixir terms delivered by send/2 from an AgentServer to a subscriber pid, and a LiveView is a perfectly good pid: it receives {:agent, event}, folds it into assigns, and the framework pushes a diff to the browser.

None of that requires LiveView. It requires a process. If your front end is React, you supply that process yourself — a small, long-lived GenServer that subscribes on the client's behalf, keeps the same session state a LiveView would keep in assigns, and re-publishes serialized payloads onto a transport the browser can consume.

This guide describes that pattern — a subscription bridge — as a template you can adapt. The examples use Absinthe GraphQL subscriptions because that's a common React pairing, but nothing about the bridge is GraphQL-specific; the last section covers other transports.

Prerequisite reading: subscriptions_and_presence.md for the event catalog and the raw subscription APIs.

1. What LiveView was actually doing for you

It helps to be clear on the jobs before replacing them. A Sagents-backed LiveView does five things:

A React client can do #4 itself, and it will hold its own copy of #2 in a reducer. But #1, #3, and #5 have nowhere to live on the server, and #2 still needs a server-side copy — because the browser can disconnect and reconnect while the agent keeps running.

The bridge is that missing process.

flowchart LR
  subgraph lv["LiveView topology"]
    direction TB
    AS1["AgentServer"]
    LVP["LiveView process<br/>assigns = session state"]
    BR1["Browser DOM"]
    AS1 -- "agent event<br/>(Elixir term)" --> LVP
    LVP -- "HTML diff<br/>(LV socket)" --> BR1
  end

  subgraph rc["React topology"]
    direction TB
    AS2["AgentServer"]
    BG["Bridge GenServer<br/>session_state"]
    TP["Transport topic<br/>session-scoped"]
    RCC["React client<br/>reducer state"]
    AS2 -- "agent event<br/>(Elixir term)" --> BG
    BG -- "serialized JSON payload" --> TP
    TP -- "WebSocket" --> RCC
  end

The bridge is a headless LiveView. Same position in the topology, same responsibilities, no rendering.

2. The bridge's five jobs

flowchart TB
  AS["AgentServer<br/>(the agent's own process)"]

  subgraph B["Subscription Bridge GenServer — one per conversation"]
    direction TB
    J1["1 · Own the subscription<br/>AgentServer.subscribe/1 + <br/>monitor"]
    J2["2 · Fold events into <br/>session_state<br/>the same shape LiveView <br/>assigns would hold"]
    J3["3 · Serialize<br/>structs → <br/>transport-safe maps"]
    J4["4 · Publish to a stable topic<br/>stamped with <br/>conversation_id"]
    J5["5 · Snapshot on demand<br/>rehydrate a client that <br/>reconnected"]
    J1 --> J2 --> J3 --> J4
    J5 --> J3
  end

  T["Transport (GraphQL <br/>subscription / <br/>Channel / SSE)"]
  FE["React client"]

  AS -- "agent events" --> J1
  J4 --> T --> FE
  FE -. "reconnect, ask for snapshot" .-> J5
  FE -. "send message / answer HITL<br/>(mutation, not the bridge)" .-> AS

Note the asymmetry: events flow out through the bridge, commands flow directly to the AgentServer. The bridge is a one-way translator. Your mutations can call AgentServer.add_message/2 and AgentServer.resume/2 themselves, and the resulting events come back around through the bridge like any others. Don't route commands through the bridge — you'd be adding a queue in front of a process that already serializes its own calls.

3. Choosing a topic: session-scoped, not conversation-scoped

This is the decision most likely to bite you, so make it first.

The intuitive design is one transport topic per conversation: agent:<conversation_id>. It has a race in it. The conversation id often doesn't exist until the user sends their first message, so the client can't subscribe until after the mutation returns — and by then the agent may already have streamed its first deltas to nobody.

sequenceDiagram
  participant FE as React client
  participant API as Mutation
  participant AG as AgentServer
  participant BR as Bridge

  rect rgba(200,80,80,0.12)
    Note over FE,BR: Conversation-scoped topic — events lost
    FE->>API: sendMessage (no conversation yet)
    API->>AG: start agent, add_message
    API->>BR: start bridge
    AG-->>BR: llm_deltas
    BR-->>BR: publish to agent:conv-1 (no subscribers)
    API-->>FE: conversationId = conv-1
    FE->>BR: subscribe agent:conv-1
    Note right of FE: first deltas already gone
  end

Publish to a session-scoped topic instead — keyed by the authenticated user (or session, or tab), never by the conversation — and stamp every payload with conversation_id so the client can discriminate. The client opens exactly one subscription when the chat UI mounts, before any conversation exists.

sequenceDiagram
  participant FE as React client
  participant API as Mutation
  participant AG as AgentServer
  participant BR as Bridge

  rect rgba(80,160,110,0.12)
    Note over FE,BR: Session-scoped topic — stream is live first
    FE->>FE: chat panel mounts
    FE->>BR: subscribe agent_user:USER_ID
    FE->>API: sendMessage (no conversation yet)
    API->>AG: start agent, add_message
    API->>BR: start bridge
    AG-->>BR: llm_deltas
    BR-->>FE: streaming_delta for conv-1
    API-->>FE: conversationId = conv-1
  end

This mirrors the LiveView model exactly: the LiveView's subscription is process-scoped, alive from mount/3, not conversation-scoped. You get the same property, plus a free path to rendering multiple concurrent conversations over one socket.

The client then filters:

if (event.conversationId && event.conversationId !== activeConversationId) return

Authorize the topic at subscribe time, from server-trusted session context — never from a client-supplied argument. The topic string is the authorization boundary.

4. Session state: the bridge's assigns

The bridge holds the state a LiveView would hold. Concretely:

Two things make this state non-optional rather than a convenience:

1. Deltas are incremental. {:llm_deltas, deltas} carries fragments. Someone has to MessageDelta.merge_deltas/2 them. Doing it on the server means a reconnecting client can be handed the merged state instead of having missed half the fragments.
2. Interrupts are sticky. An agent sitting at :interrupted published its prompt once. A client that reloads the page has no way to learn about it from the event stream — the event already fired. The bridge still holds it, and can re-publish.

Share the transition logic with LiveView

If your app has both a LiveView and a React front end (common during a migration), don't write the state transitions twice. Put them in a host-agnostic module whose functions take a plain state map and return a changes map. Each host merges in its own idiom:

# LiveView
socket = assign(socket, changes)

# Bridge GenServer
state = %{state | session_state: Map.merge(state.session_state, changes)}

Sagents already ships the trickiest piece of this — the tool-call lifecycle:

defmodule MyApp.AI.AgentSession do
  @moduledoc "Host-agnostic session-state transitions for agent events."

  alias LangChain.MessageDelta
  alias Sagents.AgentUtils
  alias Sagents.StreamingSession

  def init_session_state do
    %{
      agent_status: :not_running,
      streaming_delta: nil,
      stream_id: nil,
      loading: false,
      todos: [],
      pending_tools: [],
      pending_question: nil,
      remaining_questions: [],
      question_responses: [],
      hitl_decisions: [],
      interrupt_data: nil
    }
  end

  def handle_status_running, do: %{agent_status: :running}

  def handle_status_idle,
    do: %{agent_status: :idle, loading: false, streaming_delta: nil, stream_id: nil}

  def handle_status_cancelled,
    do: %{agent_status: :cancelled, loading: false, streaming_delta: nil, stream_id: nil}

  def handle_status_error(_reason),
    do: %{agent_status: :error, loading: false, streaming_delta: nil, stream_id: nil}

  def handle_status_interrupted(nil), do: %{}

  def handle_status_interrupted(interrupt_data) do
    # AgentUtils turns the raw interrupt into presentable
    # pending_tools / pending_question / remaining_questions.
    Map.merge(
      %{agent_status: :interrupted, loading: false, interrupt_data: interrupt_data},
      AgentUtils.interrupt_session_changes(interrupt_data)
    )
  end

  # Stamp a stream_id when a new LLM call begins; keep it for every delta of
  # that call; clear it on completion. Lets consumers drop stale deltas.
  def handle_llm_deltas(state, deltas) do
    %{
      streaming_delta: MessageDelta.merge_deltas(state[:streaming_delta], deltas),
      stream_id: state[:stream_id] || Ecto.UUID.generate()
    }
  end

  def handle_llm_message_complete,
    do: %{streaming_delta: nil, stream_id: nil, loading: false}

  # Delegate the call_id-keyed tool lifecycle to Sagents so every host gets
  # identical semantics — notably: sibling tool calls in one assistant turn
  # share a single delta, and the delta is only cleared once *every* call in
  # it reaches a terminal status.
  defdelegate handle_tool_call_identified(state, tool_info), to: StreamingSession
  defdelegate handle_tool_execution_update(state, status, tool_info), to: StreamingSession
end

Sagents.StreamingSession is worth leaning on. Getting "two tools called in one turn, one finishes first" right by hand is fiddly — an :interrupted tool is paused, not terminal, and clearing the delta early makes in-flight sibling calls vanish from the UI.

5. Event → payload mapping

The bridge's translation table. Left column is what Sagents delivers; right column is a suggested wire shape. Every payload also carries conversation_id.

Serialization rules

Everything crossing the transport must be JSON-safe. Three rules cover most of it:

1. Never send a struct as-is. %MessageDelta{} and %LangChain.Message.ToolCall{} contain fields your client neither understands nor should depend on. Project them explicitly.
2. Never send a pid, ref, or function. These appear in raw agent state and in some interrupt payloads.
3. Have a fallback clause. Interrupt payloads are extensible; a catch-all that stringifies the unknown beats a crashing bridge.

defp serialize_streaming_delta(nil), do: nil

defp serialize_streaming_delta(%MessageDelta{} = delta) do
  %{
    content: MessageDelta.content_to_string(delta) || "",
    role: "assistant",
    tool_calls: Enum.map(delta.tool_calls || [], &serialize_tool_call/1)
  }
end

defp serialize_tool_call(%ToolCall{} = tc) do
  %{
    call_id: tc.call_id,
    name: tc.name,
    arguments: tc.arguments,
    display_text: tc.display_text,
    status: tc.status,
    execution_status: get_in(tc.metadata, ["execution_status"])
  }
end

defp serialize_interrupt_data(nil), do: nil
defp serialize_interrupt_data(data) when is_map(data), do: data
defp serialize_interrupt_data(data), do: %{raw: inspect(data)}

Serializing the delta's tool_calls is what lets the client render tool invocations inside the streaming assistant message rather than as detached rows — the same thing a LiveView gets for free by rendering the delta struct.

6. The bridge template

A complete, adaptable GenServer. Registered in Sagents.ProcessRegistry so it's discoverable and — when Horde is configured — cluster-aware.

defmodule MyApp.AI.AgentBridge do
  @moduledoc """
  Bridges Sagents AgentServer events to a React client's transport.

  One bridge per active conversation. Subscribes to the AgentServer
  directly, folds each event into a session state map, and re-publishes a
  serialized payload on a session-scoped topic.
  """
  use GenServer

  alias LangChain.Message.ToolCall
  alias LangChain.MessageDelta
  alias MyApp.AI.AgentSession
  alias Sagents.AgentServer
  alias Sagents.ProcessRegistry

  require Logger

  # =========================================================================
  # Public API
  # =========================================================================

  @doc "Start the bridge for `conversation_id` if one isn't already running."
  def ensure_bridging(conversation_id, agent_id, session) do
    case ProcessRegistry.lookup({:agent_bridge, conversation_id}) do
      [{_pid, _value}] ->
        :ok

      [] ->
        spec = {__MODULE__,
                conversation_id: conversation_id, agent_id: agent_id, session: session}

        case DynamicSupervisor.start_child(MyApp.AI.BridgeSupervisor, spec) do
          {:ok, _pid} -> :ok
          {:error, {:already_started, _pid}} -> :ok
          # init/1 returns :ignore when the agent is gone. Translate it so
          # callers' `with` chains don't have to match a bare atom.
          :ignore -> {:error, :agent_not_running}
          {:error, _reason} = error -> error
        end
    end
  end

  @doc """
  Ensure a client sees current state on its topic.

  No bridge running → start one (its `init/1` publishes a snapshot).
  Bridge already running → ask it to re-publish. Required after a client
  reconnect: the original broadcast went out before this client joined the
  topic.
  """
  def ensure_snapshot(conversation_id, agent_id, session) do
    case ProcessRegistry.lookup({:agent_bridge, conversation_id}) do
      [{pid, _value}] ->
        GenServer.cast(pid, :publish_snapshot)
        :ok

      [] ->
        ensure_bridging(conversation_id, agent_id, session)
    end
  end

  def start_link(opts) do
    conversation_id = Keyword.fetch!(opts, :conversation_id)
    GenServer.start_link(__MODULE__, opts, name: via(conversation_id))
  end

  defp via(conversation_id), do: ProcessRegistry.via_tuple({:agent_bridge, conversation_id})

  # =========================================================================
  # Lifecycle
  # =========================================================================

  @impl true
  def init(opts) do
    conversation_id = Keyword.fetch!(opts, :conversation_id)
    agent_id = Keyword.fetch!(opts, :agent_id)
    session = Keyword.fetch!(opts, :session)

    case AgentServer.subscribe(agent_id) do
      {:ok, _server_pid, _monitor_ref} ->
        state = %{
          conversation_id: conversation_id,
          agent_id: agent_id,
          topic: "agent_user:#{session.user_id}",
          session_state: AgentSession.init_session_state()
        }

        # Defer the snapshot until after init/1 returns, so the process is
        # registered and able to receive replies before we broadcast.
        {:ok, state, {:continue, :publish_initial_snapshot}}

      {:error, :process_not_found} ->
        Logger.warning("AgentBridge #{conversation_id}: agent #{agent_id} not running")
        :ignore
    end
  end

  @impl true
  def handle_continue(:publish_initial_snapshot, state) do
    {:noreply, publish_snapshot(state)}
  end

  @impl true
  def handle_cast(:publish_snapshot, state) do
    {:noreply, publish_snapshot(state)}
  end

  # =========================================================================
  # Event translation
  # =========================================================================

  @impl true
  def handle_info({:agent, {:llm_deltas, deltas}}, state) do
    state = apply_changes(state, AgentSession.handle_llm_deltas(state.session_state, deltas))
    publish_streaming_delta(state)
    {:noreply, state}
  end

  def handle_info({:agent, {:llm_message, _message}}, state) do
    stream_id = state.session_state.stream_id
    state = apply_changes(state, AgentSession.handle_llm_message_complete())

    publish(state, %{type: :streaming_complete, stream_id: stream_id, completed: true})
    {:noreply, state}
  end

  def handle_info({:agent, {:status_changed, status, data}}, state) do
    state = apply_changes(state, status_changes(status, data))
    publish_status(state, status)
    {:noreply, state}
  end

  def handle_info({:agent, {:tool_call_identified, tool_info}}, state) do
    changes = AgentSession.handle_tool_call_identified(state.session_state, tool_info)
    state = apply_changes(state, changes)

    publish(state, %{
      type: :tool_identified,
      call_id: tool_info.call_id,
      name: tool_info[:name],
      display_text: tool_info[:display_text],
      arguments: tool_info[:arguments]
    })

    # The delta now contains this tool call — re-publish so clients that
    # render tools inside the streaming message stay in sync.
    publish_streaming_delta(state)
    {:noreply, state}
  end

  def handle_info({:agent, {:tool_execution_update, status, tool_info}}, state) do
    changes = AgentSession.handle_tool_execution_update(state.session_state, status, tool_info)
    state = apply_changes(state, changes)

    publish(state, %{
      type: :tool_execution_update,
      call_id: tool_info.call_id,
      name: tool_info[:name] || "",
      display_text: tool_info[:display_text],
      status: status,
      result: tool_info[:result]
    })

    publish_streaming_delta(state)
    {:noreply, state}
  end

  def handle_info({:agent, {:display_message_saved, msg}}, state) do
    publish(state, %{
      type: :message_saved,
      stream_id: state.session_state.stream_id,
      message: serialize_display_message(msg)
    })

    {:noreply, state}
  end

  def handle_info({:agent, {:display_message_updated, msg}}, state) do
    publish(state, %{type: :message_updated, message: serialize_display_message(msg)})
    {:noreply, state}
  end

  def handle_info({:agent, {:todos_updated, todos}}, state) do
    state = apply_changes(state, %{todos: todos})
    publish(state, %{type: :todos, todos: Enum.map(todos, &Map.from_struct/1)})
    {:noreply, state}
  end

  # Primary shutdown signal.
  def handle_info({:agent, {:agent_shutdown, _data}}, state) do
    Logger.debug("AgentBridge #{state.conversation_id}: agent shutdown")
    {:stop, :normal, state}
  end

  # Safety net — the agent crashed without broadcasting a shutdown. Our
  # monitor ref (from AgentServer.subscribe/1) points at the agent, so when
  # it dies we go with it.
  def handle_info({:DOWN, _ref, :process, _pid, _reason}, state) do
    Logger.debug("AgentBridge #{state.conversation_id}: agent process down")
    {:stop, :normal, state}
  end

  # Ignore events this bridge doesn't translate.
  def handle_info({:agent, _other}, state), do: {:noreply, state}

  # =========================================================================
  # Snapshot
  # =========================================================================

  # Ask the agent for its live status and publish a single :status_changed,
  # so a freshly attached client sees current state without waiting for the
  # next transition.
  defp publish_snapshot(state) do
    case AgentServer.get_status(state.agent_id) do
      :not_running ->
        state

      status ->
        info = AgentServer.get_info(state.agent_id)
        data = snapshot_data_for_status(status, info)
        state = apply_changes(state, status_changes(status, data))

        publish_status(state, status)
        state
    end
  end

  defp snapshot_data_for_status(:interrupted, info), do: Map.get(info, :interrupt_data)
  defp snapshot_data_for_status(:error, info), do: Map.get(info, :error)
  defp snapshot_data_for_status(_status, _info), do: nil

  # =========================================================================
  # Helpers
  # =========================================================================

  defp apply_changes(state, changes) when is_map(changes) do
    %{state | session_state: Map.merge(state.session_state, changes)}
  end

  defp status_changes(:running, _data), do: AgentSession.handle_status_running()
  defp status_changes(:idle, _data), do: AgentSession.handle_status_idle()
  defp status_changes(:cancelled, _data), do: AgentSession.handle_status_cancelled()
  defp status_changes(:error, reason), do: AgentSession.handle_status_error(reason)
  defp status_changes(:interrupted, data), do: AgentSession.handle_status_interrupted(data)
  defp status_changes(_other, _data), do: %{}

  defp publish_status(state, status) do
    publish(state, %{
      type: :status_changed,
      status: status,
      interrupt_data: serialize_interrupt_data(state.session_state.interrupt_data),
      pending_tools: state.session_state.pending_tools,
      pending_question: state.session_state.pending_question
    })
  end

  defp publish_streaming_delta(state) do
    publish(state, %{
      type: :streaming_delta,
      stream_id: state.session_state.stream_id,
      streaming_delta: serialize_streaming_delta(state.session_state.streaming_delta)
    })
  end

  # Stamp conversation_id on every payload — the client needs it to
  # discriminate events from concurrently running conversations.
  defp publish(state, payload) do
    payload = Map.put(payload, :conversation_id, state.conversation_id)

    Absinthe.Subscription.publish(MyAppWeb.Endpoint, payload,
      subscribe_agent_events: state.topic
    )
  end

  # ... serialize_streaming_delta/1, serialize_tool_call/1,
  #     serialize_interrupt_data/1, serialize_display_message/1
end

Supervision

Start the bridges under a DynamicSupervisor. :transient is wrong here — a bridge that stops :normal because its agent shut down should stay stopped, and one that crashes will be restarted with a stale session state and no subscription. Prefer restart: :temporary and let the next ensure_bridging/3 create a fresh one:

# in your application supervision tree
{DynamicSupervisor, name: MyApp.AI.BridgeSupervisor, strategy: :one_for_one}

# in the bridge module
use GenServer, restart: :temporary

7. Wiring to GraphQL

Model the payloads as a union discriminated on :type. The client gets __typename for free, which is exactly the tag a TypeScript switch wants.

defmodule MyAppWeb.Schema.AgentTypes do
  use Absinthe.Schema.Notation

  object :agent_streaming_delta_event do
    field :conversation_id, non_null(:id)
    field :stream_id, :id
    field :streaming_delta, :agent_streaming_delta
  end

  object :agent_status_changed_event do
    field :conversation_id, non_null(:id)
    field :status, non_null(:agent_status)
    field :interrupt_data, :json
    field :pending_tools, :json
    field :pending_question, :agent_pending_question
  end

  # ... one object per payload type

  union :agent_conversation_event do
    types([
      :agent_streaming_delta_event,
      :agent_streaming_complete_event,
      :agent_status_changed_event,
      :agent_message_saved_event,
      :agent_message_updated_event,
      :agent_tool_identified_event,
      :agent_tool_execution_update_event,
      :agent_title_generated_event
    ])

    resolve_type fn
      %{type: :streaming_delta}, _ -> :agent_streaming_delta_event
      %{type: :streaming_complete}, _ -> :agent_streaming_complete_event
      %{type: :status_changed}, _ -> :agent_status_changed_event
      %{type: :message_saved}, _ -> :agent_message_saved_event
      %{type: :message_updated}, _ -> :agent_message_updated_event
      %{type: :tool_identified}, _ -> :agent_tool_identified_event
      %{type: :tool_execution_update}, _ -> :agent_tool_execution_update_event
      %{type: :title_generated}, _ -> :agent_title_generated_event
    end
  end

  object :agent_subscriptions do
    @desc """
    One persistent subscription per session, scoped to the authenticated
    user. Open it when the chat UI mounts — before any conversation exists —
    and filter incoming events by `conversationId`.
    """
    field :subscribe_agent_events, :agent_conversation_event do
      config(&subscription_config/2)
    end
  end

  # The topic is derived from server-trusted context, never from arguments.
  defp subscription_config(_args, %{context: %{user: %{id: user_id}}}) do
    {:ok, topic: "agent_user:#{user_id}"}
  end

  defp subscription_config(_args, _resolution), do: {:error, "Not authorized"}
end

The send-message mutation is where a turn actually begins. It starts the agent, guarantees a bridge, then hands the message to the agent:

def send_message(_parent, args, %{context: context}) do
  with {:ok, conversation} <- find_or_create_conversation(context, args[:conversation_id]),
       {:ok, agent_id} <- ensure_agent_running(conversation),
       :ok <- AgentBridge.ensure_bridging(conversation.id, agent_id, context.session),
       :ok <- AgentServer.add_message(agent_id, Message.new_user!(args.content)) do
    {:ok, %{conversation_id: conversation.id, status: :accepted}}
  end
end

Order matters: bridge before message. Reverse them and the first deltas are published before anything is subscribed to the agent.

Two smaller notes on shaping payloads for a typed client:

• Derive typed fields from raw ones with field resolvers. Publish interrupt_data as :json for escape-hatch access, but also expose a typed tool_approval / pending_question object resolved from it, so the client isn't hand-parsing a map whose shape can change.
• Keep stream_id on both streaming_delta and streaming_complete. It's what lets the client discard a delta that arrives after its stream closed — an ordinary occurrence when a tool result triggers a second LLM call immediately.

8. The React side

One subscription, held for the session's lifetime

The stream is session-scoped, so establish it once and don't tear it down when the conversation changes. Read the active conversation id and the event handler through refs, so changing either doesn't re-run the effect and drop the socket:

export const useAgentSubscription = ({
  conversationId,
  handleAgentEvent,
}: {
  conversationId: string
  handleAgentEvent: (event: AgentConversationEvent) => void
}) => {
  const conversationIdRef = useRef(conversationId)
  useEffect(() => {
    conversationIdRef.current = conversationId
  }, [conversationId])

  const handlerRef = useRef(handleAgentEvent)
  useEffect(() => {
    handlerRef.current = handleAgentEvent
  }, [handleAgentEvent])

  useEffect(() => {
    const subscription = subscribeToAgentEvents({
      onReconnected: handleReconnected,
      onUpdate: (event) => {
        if (!event) return
        // Session-scoped stream: ignore other conversations' events.
        if (
          event.conversationId &&
          event.conversationId !== conversationIdRef.current
        ) {
          return
        }
        handlerRef.current(event)
      },
    })
    return () => subscription.unsubscribe()
  }, [handleReconnected])
}

Dispatch on __typename

const handleAgentEvent = (event: AgentConversationEvent) => {
  switch (event.__typename) {
    case 'AgentStreamingDeltaEvent':
      handleStreamingDelta(event)
      break
    case 'AgentStreamingCompleteEvent':
      handleStreamingComplete(event)
      break
    case 'AgentStatusChangedEvent':
      handleStatusChanged(event)
      break
    case 'AgentMessageSavedEvent':
      handleMessageSaved(event)
      break
    case 'AgentToolIdentifiedEvent':
      handleToolIdentified(event)
      break
    case 'AgentToolExecutionUpdateEvent':
      handleToolExecutionUpdate(event)
      break
  }
}

Client-side state rules

Four rules that keep the UI honest and idempotent:

1. Streaming text is ephemeral; the saved message is the truth. Render deltas into a provisional bubble keyed by streamId, then reconcile against the message_saved event carrying the same streamId. Don't accumulate deltas into your permanent message list.
2. Drop deltas for closed streams. Keep a set of completed streamIds and ignore anything arriving for one.
3. Key everything tool-related by callId. Tool identified → execution update → result all share it. Upsert, never append.
4. Make interrupt rendering idempotent. The server may re-deliver a pending question after a reconnect (see below). Dedupe by toolCallId so a re-delivery upserts the existing card instead of stacking a second one.

Rule 4 is the one people skip, and it's the one that breaks after a network blip.

9. Reconnect and rehydration

A LiveView that loses its socket remounts and re-derives everything. A React client that loses its WebSocket reconnects to a topic and receives only future events — the agent's stream is not replayed. If the agent went :interrupted while the client was away, that prompt is gone forever unless the server re-publishes it.

sequenceDiagram
  participant FE as React client
  participant WS as Transport
  participant Q as Sync query/mutation
  participant BR as Bridge
  participant AG as AgentServer

  AG-->>BR: status_changed interrupted
  BR-->>WS: status_changed with pending_question
  Note over FE: disconnected — event missed

  FE->>WS: reconnect, rejoin session topic
  FE->>Q: re-sync active conversation
  Q->>BR: ensure_snapshot
  alt bridge alive
    BR->>AG: get_status, get_info
    AG-->>BR: interrupted, interrupt_data
  else bridge gone (agent slept)
    Q->>AG: wake agent, restore interrupt
    Q->>BR: start bridge, init publishes snapshot
  end
  BR-->>WS: status_changed with pending_question
  WS-->>FE: prompt re-rendered

Implement it in three parts:

1. The client detects reconnect (not initial connect) and fires a fire-and-forget re-sync for the active conversation. The recovered state arrives on the live stream, not as the query's response.
2. The server-side sync path calls ensure_snapshot/3, which either pokes the running bridge or starts one — and, if the agent itself has shut down since, wakes it first. {:agent_shutdown, data} carries interrupt_restorable, which tells you whether a pending interrupt can be rebuilt on the next boot; see subscriptions_and_presence.md.
3. The bridge re-publishes. Note that the snapshot path is deliberately less conditional than the live path: on a live interrupt you publish the prompt once, but on a snapshot you re-publish even if you already published it before, because you can't know what this client received. That's safe precisely because of client rule 4.

If your interrupt prompts are also persisted as display messages, the snapshot path should re-emit the persisted record rather than inventing a new one — otherwise a reconnect leaves a duplicate in the conversation history.

10. Inbound: messages, HITL, cancel

These bypass the bridge entirely. All AgentServer calls serialize on the agent's own process, so ordering is preserved without any coordination on your side.

# Send a message
AgentServer.add_message(agent_id, Message.new_user!(content))

# Cancel a run
AgentServer.cancel(agent_id)

# Answer a HITL interrupt or an AskUserQuestion
AgentServer.resume(agent_id, resume_data)

# Push client state into a middleware before the next model call
AgentServer.notify_middleware(agent_id, MyApp.Middleware.UserContext, {:context_changed, ctx})

Multi-part interrupts. A HITL interrupt may hold several tool calls, and an AskUserQuestion may hold several questions. The client answers them one at a time, but the agent resumes once, with all answers. Accumulate in the session state and only call resume/2 on the last one. Sagents.AgentUtils.advance_hitl_decisions/3 implements this for tool approvals — it returns {:more, changes} while decisions remain and {:resume, decisions, changes} when the set is complete.

The mirror-image pattern for questions:

def handle_question_response(state, response) do
  response = Map.put(response, :tool_call_id, state.pending_question.tool_call_id)
  accumulated = state.question_responses ++ [response]

  case state.remaining_questions do
    [] ->
      resume_data = if length(accumulated) == 1, do: hd(accumulated), else: accumulated

      {:resume, resume_data,
       %{pending_question: nil, remaining_questions: [], question_responses: [],
         interrupt_data: nil}}

    [next | rest] ->
      {:more,
       %{pending_question: next, remaining_questions: rest, question_responses: accumulated}}
  end
end

After a successful resume/2, merge %{agent_status: :running, loading: true} so the UI flips immediately rather than waiting for the status event.

One caveat on client-supplied context. Anything the browser sends — current page, selected record, form contents — is untrusted input. It is fine to feed into a prompt, but any identifier in it must be re-resolved server-side against the caller's authorization scope before a tool acts on it. The trust boundary is the resolver, not the middleware.

11. Presence and lifecycle

Sagents can shut an agent down when nobody is watching (see lifecycle.md and the viewer-presence section of subscriptions_and_presence.md). With LiveView, the LiveView is the viewer. With React, the bridge is the viewer — it is the only server-side process that exists for as long as the client is interested.

Track it on the conversation's viewer topic at init/1, untrack in terminate/2:

@viewer_prefix "bridge:"

# in init/1
Sagents.Presence.track(
  MyApp.Presence,
  "conversation:#{conversation_id}",
  @viewer_prefix <> conversation_id,
  %{type: :subscription_bridge}
)

@impl true
def terminate(_reason, state) do
  Sagents.Presence.untrack(
    MyApp.Presence,
    "conversation:#{state.conversation_id}",
    @viewer_prefix <> state.conversation_id
  )

  :ok
end

The resulting lifecycle:

stateDiagram-v2
  [*] --> NoBridge: conversation cold
  NoBridge --> Bridging: ensure_bridging <br/>(send message or re-sync)
  Bridging --> Bridging: translate and <br/>publish events
  Bridging --> Bridging: ensure_snapshot <br/>(client reconnect)
  Bridging --> NoBridge: agent_shutdown <br/>or agent DOWN
  NoBridge --> [*]

The bridge deliberately has no independent lifetime. It starts because an agent is running and a client cares; it dies when the agent does. It should never be the reason an agent stays alive, and it should never outlive one — a bridge holding a stale agent_id will silently publish nothing while looking perfectly healthy.

If you want the agent to sleep when the browser tab closes rather than when the bridge stops, track a second viewer keyed by the client session and untrack it on socket disconnect. That's a genuine choice: bridge-as-viewer keeps agents warm across brief reloads; session-as-viewer reclaims resources faster.

12. Checklist

Before shipping a bridge:

• [ ] Topic is session-scoped, authorized from server-trusted context,

  and every payload carries `conversation_id`.

• [ ] Client subscribes when the chat UI mounts, before the first

  message mutation.

• [ ] Mutation order is: ensure agent → ensure bridge → add message.
• [ ] init/1 returns :ignore when the agent isn't running, and callers

  translate that into an error.

• [ ] Initial snapshot is published from handle_continue, not init/1.
• [ ] Both {:agent_shutdown, _} and {:DOWN, ...} stop the bridge.
• [ ] A catch-all handle_info({:agent, _other}, state) clause exists — new

  Sagents events must not crash the bridge.

• [ ] No struct, pid, ref, or function reaches the serializer's output.
• [ ] stream_id is present on delta and complete events; the client drops

  deltas for closed streams.

• [ ] Reconnect triggers a re-sync that re-publishes the current snapshot.
• [ ] Interrupt cards dedupe by tool_call_id on the client.
• [ ] Multi-question / multi-tool interrupts resume exactly once.
• [ ] Bridge is restart: :temporary under a DynamicSupervisor.

13. Other transports

Nothing above depends on GraphQL. Only publish/2 and the payload schema change:

Phoenix Channels. Replace Absinthe.Subscription.publish/3 with MyAppWeb.Endpoint.broadcast(topic, event_name, payload) and use the event name where the union's :type was. Authorize in join/3. Channels give you a natural handle_in/3 for the inbound commands too, so a channel implementation can host both directions in one module — though the commands should still go straight to AgentServer.

Server-Sent Events. Keep the bridge exactly as written and have the SSE controller subscribe to a Phoenix.PubSub topic the bridge publishes to. Since SSE is one-way, commands go over ordinary HTTP endpoints. Note that SSE has no reconnect handshake you can hook cleanly — you'll want the client to re-issue the sync request on EventSource reopen.

Plain WebSocket / Phoenix.Socket.Transport. Same shape; you own the serialization format. This is the most work and the least benefit unless you already have a socket protocol.

In all cases the bridge stays the same process doing the same five jobs. The transport is the last ten lines.

Related documents

• subscriptions_and_presence.md — event catalog, Sagents.Subscriber, presence mechanics
• lifecycle.md — agent startup, inactivity, shutdown
• conversations_architecture.md — display messages and persistence
• clustering.md — ProcessRegistry with Horde, and what changes for a bridge in a multi-node deployment