GenServer wrapper for Nous agents with PubSub integration.

This server:

• Wraps a Nous agent (standard or ReAct)
• Links to parent process (dies when parent dies)
• Subscribes to PubSub for incoming messages
• Publishes responses back via PubSub
• Maintains conversation context for multi-turn conversations

Context-Based State

Uses Nous.Agent.Context to maintain conversation state:

• Messages accumulate across turns
• Tool calls are tracked
• Usage is aggregated
• Callbacks forward to PubSub

Usage with LiveView

defmodule MyAppWeb.ChatLive do
  use MyAppWeb, :live_view

  def mount(_params, _session, socket) do
    # Start agent linked to this LiveView
    {:ok, agent_pid} = AgentServer.start_link(
      session_id: socket.assigns.session_id,
      agent_config: %{
        model: "lmstudio:qwen3-vl-4b-thinking-mlx",
        instructions: "You are a helpful assistant",
        tools: []
      }
    )

    # Subscribe to responses. Always build the topic with
    # Nous.PubSub.agent_topic/1 — it is what this server publishes to.
    Phoenix.PubSub.subscribe(MyApp.PubSub, Nous.PubSub.agent_topic(socket.assigns.session_id))

    {:ok, assign(socket, agent_pid: agent_pid, messages: [])}
  end

  def handle_event("send_message", %{"message" => msg}, socket) do
    AgentServer.send_message(socket.assigns.agent_pid, msg)
    {:noreply, socket}
  end

  # Receive streaming deltas
  def handle_info({:agent_delta, text}, socket) do
    # Append text to current response
    {:noreply, update(socket, :current_response, &(&1 <> text))}
  end

  # Receive complete response
  def handle_info({:agent_complete, result}, socket) do
    messages = socket.assigns.messages ++ [%{role: :assistant, content: result.output}]
    {:noreply, assign(socket, messages: messages, current_response: "")}
  end

  # Receive tool calls
  def handle_info({:tool_call, call}, socket) do
    # Show tool call in UI
    {:noreply, socket}
  end
end

Message Interruption

Calling send_message/2 while the agent is already processing a request automatically cancels the in-flight execution and starts a new one. The server uses an :atomics-based flag so the running task can detect cancellation without message-passing overhead. Both the interrupted user message and the new one are preserved in the conversation context, so no input is lost.

PubSub Events

Subscribers on the Nous.PubSub.agent_topic(session_id) topic (currently "nous:agent:<session_id>") receive the following messages:

{:session_event, _} is published by Nous.Agent.Context itself for every event committed to the session log, which is why nothing in this server broadcasts it: init/1 already points the run context at this topic. It is the message to render a transcript from, since it also carries the turn and step boundaries (:turn_start, :step_start, :step_end, :turn_end) that the ad-hoc callbacks above cannot express.

Steering and injection

send_message/2 is the interrupting path: it cancels whatever is in flight and starts over. The inbox paths never interrupt anything — they queue a message and let the run claim it at its next boundary:

steer/2 and inject/2 differ in exactly one bit, and it is the bit people get wrong. Worked example, on an agent that is idle after its last run:

# Nothing happens. No model request, no tokens, no run. The note is queued
# and will be handed to the next request the agent makes anyway.
AgentServer.inject(pid, "FYI: the staging deploy is frozen until 14:00")

# This starts a run. It claims the injected note *and* this message, in
# that order, so both are in the first request's message list.
AgentServer.steer(pid, "Which services are still pending release?")

Mid-run, the distinction is about latency rather than about starting work:

AgentServer.send_message(pid, "Audit every service for stale configs")
# ... the agent is three tool calls deep ...
AgentServer.steer(pid, "Actually, skip anything under /legacy")
# The NEXT model request of that same run sees the instruction. The tool
# call already in flight is neither interrupted nor cancelled, and the
# message is not claimed by the request that is already on the wire.

Both accept a binary or a Nous.Message, and both are safe to call whether or not a run is in flight.

Run state

The server keeps an explicit run_state of :idle or :running, and every transition into :idle goes through one function. That is what makes wakeup decidable: steer/2 on an idle agent must start work and inject/2 must not, and both answers depend on knowing — not inferring — whether a run is in flight.

The upstream implementation this was ported from needs a wakeRequested latch to cover the window where a message lands after the last claim of a run that is about to end. There is no latch here. The wake intent rides on the queued message itself (see Nous.Session.Inbox), so the :running -> :idle transition just asks the inbox whether a run is still owed. Enqueueing and finishing are both serialized through this process's mailbox, so exactly one of the two starts the run.

Lifecycle

Each server starts an inactivity timer (default 5 minutes, configurable via :inactivity_timeout). The timer resets on every send_message/2 call. When the timer fires, the server terminates with :normal.

If a :persistence backend is configured, the conversation context is automatically saved after each successful agent run and restored on start_link/1.