Sagents.AgentServer (Sagents v0.11.1)

Copy Markdown

GenServer that wraps a DeepAgent and its State, managing execution lifecycle and broadcasting events via PubSub.

The AgentServer provides:

  • Asynchronous agent execution
  • State management and tracking
  • Event broadcasting for UI updates
  • Human-in-the-loop interrupt handling

Understanding agent_id

The agent_id is a runtime identifier used for process management and inter-process communication. It serves several critical purposes:

Process Registration

The agent_id is used to construct a Registry key via get_name(agent_id), which returns a :via tuple for GenServer registration:

  • Format: {:via, Registry, {Sagents.Registry, {:agent_server, agent_id}}}
  • Ensures only one AgentServer process exists per agent_id
  • Enables process lookup without maintaining PIDs

PubSub Topics

The agent_id forms the basis for PubSub topic construction:

  • Topic format: "agent_server:#{agent_id}"
  • External clients subscribe using: AgentServer.subscribe(agent_id)
  • Events broadcast include: status changes, LLM deltas, todos updates

Middleware Context

The agent_id is passed to middleware during initialization, enabling:

  • Coordination with agent-specific services (FileSystemServer, SubAgentsDynamicSupervisor)
  • Parent-child relationship establishment in SubAgent hierarchies
  • Per-agent resource isolation (virtual filesystems, etc.)

Supervision Tree Coordination

The agent_id flows through the entire supervision tree via AgentSupervisor, ensuring all child processes (FileSystemServer, AgentServer, SubAgentsDynamicSupervisor) are coordinated under the same agent context.

What agent_id IS NOT

Not Part of Conversation State: The agent_id is NOT included in serialized state (via export_state/1). It's a runtime identifier, not conversation data. This separation provides important benefits:

  • Flexibility: Restore the same conversation state under a different agent_id
  • State Cloning: Clone conversations for testing or forking scenarios
  • Clean Architecture: Clear separation between runtime identity and data

When restoring state via start_link_from_state/2, you must provide the agent_id as a parameter. This enables use cases like:

# Restore with same agent_id
AgentServer.start_link_from_state(saved_state, agent_id: "conversation-123")

# Clone conversation with different agent_id
AgentServer.start_link_from_state(saved_state, agent_id: "conversation-123-fork")

The agent_id can be any value that makes sense for your application:

  • Database conversation IDs: "conv_a1b2c3d4"
  • User-scoped identifiers: "user-#{user_id}-session-#{session_id}"
  • Randomly generated GUIDs: UUID.uuid4()
  • Application-defined values: "demo-agent-001"

Events

The server broadcasts events on the topic "agent_server:#{agent_id}".

All events are wrapped in an {:agent, event} tuple to help consumers identify and route events from AgentServer. This is similar to how FileSystemServer wraps its events in {:file_system, event}.

Event Format

Events are received in the format {:agent, event} where event is one of:

Todo Events

  • {:agent, {:todos_updated, todos}} - Complete snapshot of current TODO list

Status Events

  • {:agent, {:status_changed, :idle, nil}} - Server ready for work (also broadcast after successful execution completion)
  • {:agent, {:status_changed, :running, nil}} - Agent executing
  • {:agent, {:status_changed, :interrupted, interrupt_data}} - Awaiting human decision
  • {:agent, {:status_changed, :paused, pause_reason}} - Infrastructure pause; the payload is the cause the mode attached (nil when it attached none)
  • {:agent, {:status_changed, :cancelled, nil}} - Execution was cancelled by user
  • {:agent, {:status_changed, :error, reason}} - Execution failed

Node Transfer Events

  • {:agent, {:node_transferring, data}} - Agent is about to leave this node (broadcast during terminate/2)
    • data.from_node - The node the agent is leaving
  • {:agent, {:node_transferred, data}} - Agent has been restored on a new node (broadcast on startup after restore)
    • data.to_node - The node the agent has been restored to

Shutdown Events

  • {:agent, {:agent_shutdown, shutdown_data}} - Agent is shutting downA single shutdown delivers this event more than once (the trigger handler fires it, then terminate/2 fires it again). Every site emits the same map shape, so consumers can handle it with one idempotent clause:
    • shutdown_data.agent_id - The agent identifier
    • shutdown_data.reason - Shutdown reason (:inactivity | :no_viewers | the terminate/2 reason term)

    • shutdown_data.status - The server's status at shutdown
    • shutdown_data.interrupt_restorable - true when the server is shutting down :interrupted and the pending interrupt can be rebuilt on the next boot (see Sagents.State.interrupt_restorable?/2). A host showing an interrupt prompt can keep it on screen when this is true, because answering it will wake an agent that boots straight back into :interrupted. false whenever there is no interrupt, or the interrupt is process-bound (e.g. a sub-agent approval). See Sagents.AgentUtils.shutdown_session_changes/2.
    • shutdown_data.last_activity_at - DateTime of last activity
    • shutdown_data.shutdown_at - DateTime when shutdown was initiated

Tool Execution Events (consolidated)

  • {:agent, {:tool_execution_update, status, tool_info}} - Tool execution lifecycle update

    • status is one of: :executing, :completed, :failed
    • :executingtool_info contains :call_id, :name, :display_text, :arguments
    • :completedtool_info contains :call_id, :name, :result
    • :failedtool_info contains :call_id, :name, :error
  • {:agent, {:display_message_updated, display_msg}} - Tool status updated in DB (only when display_message_persistence is configured)

LLM Streaming Events

  • {:agent, {:llm_deltas, [%MessageDelta{}]}} - Streaming tokens/deltas received (list of deltas)
  • {:agent, {:llm_message, %Message{}}} - Complete message received and processed
  • {:agent, {:llm_token_usage, %TokenUsage{}}} - Token usage information

Message Persistence Events

  • {:agent, {:display_message_saved, display_message}} - Broadcast after message is persisted via display_message_persistence behaviour. The {:llm_message, ...} event is also broadcast alongside this event

Queued Message Events

  • {:agent, {:message_queued, %Message{}}} - A message arrived while a run was in flight and is being held for delivery at the next run boundary. See add_message/3 and queue_message_from_tool/3. Subscribe to this rather than reading a "queued" signal off add_message/2's return value, which stays :ok deliberately.

Note: File events are NOT broadcast by AgentServer. Files are managed by FileSystemServer which provides its own event handling mechanism.

Debug Events

When debug PubSub is configured, additional debug events are broadcast on the topic "agent_server:debug:#{agent_id}". These events provide deeper insight into agent execution for debugging and monitoring purposes.

Debug events are also wrapped in {:agent, {:debug, event}} for consistent routing with regular events.

Middleware Debug Events

  • {:agent, {:debug, {:agent_state_update, state}}} - Middleware state update with full state snapshot

Queued Message Debug Events

  • {:agent, {:debug, {:messages_drained, count}}} - A queued message was delivered and a follow-up run started. Note that no :idle status is broadcast between the two runs; this is the event for observers who need to know the extra run happened.
  • {:agent, {:debug, {:pending_message_held, :error}}} - A run failed while a message was queued. The message is kept, not delivered.
  • {:agent, {:debug, {:auto_execution_limit_reached, limit}}} - Too many consecutive runs were started by queued messages with no human input in between. The message was added to the conversation but no further run was started.

Usage

# Start a server
{:ok, agent} = Agent.new(
  agent_id: "my-agent-1",
  model: model,
  base_system_prompt: "You are a helpful assistant."
)

initial_state = State.new!(%{
  messages: [Message.new_user!("Write a hello world program")]
})

{:ok, _pid} = AgentServer.start_link(
  agent: agent,
  initial_state: initial_state,
  name: AgentServer.get_name("my-agent-1")
)

# Subscribe to events
AgentServer.subscribe("my-agent-1")

# Execute the agent
:ok = AgentServer.execute("my-agent-1")

# Cancel execution if needed
:ok = AgentServer.cancel("my-agent-1")

# Listen for events
receive do
  {:todos_updated, todos} -> IO.inspect(todos, label: "Current TODOs")
  {:status_changed, :idle, nil} -> IO.puts("Done!")
end

Human-in-the-Loop Example

# Configure agent with interrupts
{:ok, agent} = Agent.new(
  agent_id: "my-agent-1",
  model: model,
  interrupt_on: %{"write_file" => true}
)

{:ok, _pid} = AgentServer.start_link(
  agent: agent,
  initial_state: state,
  name: AgentServer.get_name("my-agent-1")
)

AgentServer.subscribe("my-agent-1")

# Execute
AgentServer.execute("my-agent-1")

# Wait for interrupt
receive do
  {:status_changed, :interrupted, interrupt_data} ->
    # Display interrupt_data.action_requests to user
    decisions = get_user_decisions(interrupt_data)
    AgentServer.resume("my-agent-1", decisions)
end

# Wait for completion
receive do
  {:status_changed, :idle, nil} -> :ok
end

Summary

Types

Status of the agent server

Functions

Add a message to the agent's state and transition to idle if completed.

Gets count of currently running agents.

Gets detailed information about a running agent.

Cancel a running LLM task.

Dismiss a terminal :halt interrupt and transition the server back to :idle.

Execute the agent.

Export the current conversation state to a serializable format.

Gets the agent configuration for the given agent.

Get the current inactivity status of an agent.

Get server info including status, state, and any error or interrupt data.

Gets metadata about the agent server including status and last activity.

Get the name of the AgentServer process for a specific agent.

Get the pid of the AgentServer process for a specific agent.

Get the current status of the server.

Gets all running agents matching a glob pattern.

Lists all currently running agent processes.

Send a targeted message to a specific middleware in a running AgentServer.

Hook fired after a subscriber is registered. Default no-op. Override to send a snapshot of the current state to the new subscriber so it can sync without polling.

Request the AgentServer to publish a specific debug PubSub message or event.

Request the AgentServer to publish an specific PubSub message or event.

Queue a message into the conversation from inside a running tool or middleware.

Reset the agent's state and filesystem to start fresh.

Restore agent state from a previously exported state.

Resume agent execution after an interrupt.

Check if an agent is running.

Persist a synthetic display message and broadcast it to subscribers.

Start an AgentServer.

Start a new AgentServer with restored state.

Stop the AgentServer.

Subscribe a process to events from this AgentServer.

Touch the agent to indicate activity and reset the inactivity timer.

Unsubscribe a process from events on the given channel.

Updates both the agent configuration and state.

Types

status()

@type status() :: :idle | :running | :interrupted | :paused | :cancelled | :error

Status of the agent server

Functions

add_message(agent_id, message, opts \\ [])

@spec add_message(String.t(), LangChain.Message.t(), keyword()) ::
  :ok | {:error, term()}

Add a message to the agent's state and transition to idle if completed.

This is useful for conversational interfaces where you want to add a new user message after the agent has completed a previous execution.

Returns :ok on success.

Adding a message while the agent is running

A :user message that arrives while the agent is :running is queued, not rejected. It is delivered at the next run boundary: appended to the conversation once the in-flight run completes successfully, which then starts a follow-up run. add_message/2 still returns :ok in that case.

This is deliberate and load-bearing: previously the message was written into the rolling state and then silently destroyed when the canonical state from Sagents.Agent.execute/3 replaced it wholesale, while the caller got a misleading {:error, "Cannot execute, server is in state: running"} despite the message having been accepted. Returning :ok is both more truthful and non-breaking for consumers that match only on :ok and {:error, reason}.

A consumer that wants to render a "queued" affordance should subscribe to the {:message_queued, message} event rather than read it off the return value.

Messages that arrive with a non-:user role while running are still rejected. Injecting an assistant turn mid-run is not a supported operation.

Options

  • :display - controls the transcript half of the message, independently of what the model sees. Three values:
    • absent (default) - existing behavior. The configured Sagents.MessagePreprocessor runs if one is set; with none, the display and LLM halves are the same message.
    • a %LangChain.Message{} - the caller supplies both halves directly. The configured preprocessor is not run; the caller has already made the decision the preprocessor exists to make.
    • :none - model-visible, display-invisible. Nothing reaches the transcript.

The two halves are independent messages, including their role. Because Sagents.DisplayMessagePersistence.save_message/3 derives the transcript entry's attribution from message.role, a message that is :user to the model can be :assistant in the transcript. That is the shape a tool-initiated injection needs: nobody typed it, so attributing it to the author would put words in their mouth.

Examples

# After agent completes
:ok = AgentServer.add_message("my-agent-1", Message.new_user!("What's next?"))
:ok = AgentServer.execute("my-agent-1")

# The author typed a shorthand; the model receives the expansion.
:ok = AgentServer.add_message("my-agent-1", Message.new_user!(expanded),
        display: Message.new_user!("/changelog for the latest release"))

# Model-visible, transcript-invisible.
:ok = AgentServer.add_message("my-agent-1", Message.new_user!(playbook),
        display: :none)

agent_count()

@spec agent_count() :: non_neg_integer()

Gets count of currently running agents.

Returns the total number of AgentServer processes registered in the Sagents.Registry.

Examples

AgentServer.agent_count()
# => 5

agent_info(agent_id)

@spec agent_info(String.t()) :: map() | nil

Gets detailed information about a running agent.

Returns a map with agent status and state information, or nil if the agent is not running.

Return Value

If the agent is running, returns a map containing:

  • :agent_id - The agent identifier
  • :pid - The process ID
  • :status - Current execution status (:idle, :running, :interrupted, etc.)
  • :state - Exported state snapshot
  • :message_count - Number of messages in the state
  • :has_interrupt - Boolean indicating if there's pending interrupt data

Examples

AgentServer.agent_info("conversation-1")
# => %{
#   agent_id: "conversation-1",
#   pid: #PID<0.1234.0>,
#   status: :idle,
#   state: %State{...},
#   message_count: 5,
#   has_interrupt: false
# }

AgentServer.agent_info("nonexistent")
# => nil

cancel(agent_id)

@spec cancel(String.t()) :: :ok | {:error, term()}

Cancel a running LLM task.

Stops the currently executing agent task and transitions the server to completed status. Returns {:error, reason} if the server is not running (no task to cancel).

Examples

:ok = AgentServer.cancel("my-agent-1")

dismiss_interrupt(agent_id)

@spec dismiss_interrupt(String.t()) :: :ok | {:error, term()}

Dismiss a terminal :halt interrupt and transition the server back to :idle.

Used to acknowledge a halt (e.g., user clicked "Dismiss" on the halt panel) without sending a new message. Clears interrupt_data, calls State.cancel_pending_interrupts/1, transitions status to :idle, and broadcasts the status change.

Only valid when status is :interrupted and the pending interrupt is a halt (or a :multiple_interrupts batch containing a halt). Other interrupt types — HITL approvals, ask_user_question — require an explicit response via resume/2 (which the relevant middleware interprets), so this function returns {:error, "interrupt requires explicit response (use resume)"} for those.

Examples

:ok = AgentServer.dismiss_interrupt("my-agent-1")

execute(agent_id)

@spec execute(String.t()) :: :ok | {:error, term()}

Execute the agent.

Starts agent execution asynchronously. The server will broadcast events as the agent runs. Returns :ok immediately.

Returns {:error, reason} if the server is not idle (already running, interrupted, etc.).

Examples

:ok = AgentServer.execute("my-agent-1")

export_state(agent_id)

@spec export_state(String.t()) :: map()

Export the current conversation state to a serializable format.

This can be persisted to a database and later used to restore the conversation state. The exported state uses string keys (not atoms) for compatibility with JSON/JSONB storage.

Returns a map with string keys containing:

  • "version" - Serialization format version
  • "state" - The conversation state (messages, todos, metadata)
  • "serialized_at" - ISO8601 timestamp

What is NOT included:

  • Agent configuration (middleware, tools, model) - must come from application code
  • agent_id - runtime identifier provided when restoring

This design allows you to restore the same conversation state under a different agent_id, enabling use cases like state cloning and conversation forking.

Examples

state = AgentServer.export_state("my-agent-1")
# Save to database
MyApp.Conversations.save_agent_state(conversation_id, state)

get_agent(agent_id)

@spec get_agent(String.t()) :: {:ok, Sagents.Agent.t()} | {:error, :not_found}

Gets the agent configuration for the given agent.

Returns

  • {:ok, Agent.t()} - The agent struct
  • {:error, :not_found} - Agent not found

Examples

{:ok, agent} = AgentServer.get_agent("my-agent-1")
# => {:ok, %Agent{agent_id: "my-agent-1", model: %ChatModel{...}, ...}}

get_inactivity_status(agent_id)

@spec get_inactivity_status(String.t()) :: map()

Get the current inactivity status of an agent.

Returns a map with:

  • :inactivity_timeout - Configured timeout in milliseconds (or nil/:infinity)
  • :last_activity_at - DateTime of last activity
  • :timer_active - Boolean indicating if timer is currently running
  • :time_since_activity - Milliseconds since last activity (or nil if no activity yet)

Examples

status = AgentServer.get_inactivity_status("my-agent-1")
# => %{
#   inactivity_timeout: 300_000,
#   last_activity_at: ~U[2025-11-06 10:15:30.123Z],
#   timer_active: true,
#   time_since_activity: 45_000
# }

get_info(agent_id)

@spec get_info(String.t()) :: map()

Get server info including status, state, and any error or interrupt data.

Returns a map with:

  • :status - Current status
  • :state - Current State
  • :interrupt_data - Interrupt data if status is :interrupted
  • :error - Error reason if status is :error

Examples

info = AgentServer.get_info("my-agent-1")

get_metadata(agent_id)

@spec get_metadata(String.t()) :: {:ok, map()} | {:error, :not_found}

Gets metadata about the agent server including status and last activity.

Returns

  • {:ok, metadata} - Map with agent metadata
  • {:error, :not_found} - Agent not found

Metadata Fields

  • :status - Current status atom (:idle, :running, :interrupted, :error, :cancelled)
  • :last_activity_at - DateTime of last activity (may be nil)
  • :conversation_id - Conversation ID (may be nil)
  • :node - The Erlang node where this agent is running

Examples

{:ok, metadata} = AgentServer.get_metadata("my-agent-1")
# => {:ok, %{status: :idle, last_activity_at: ~U[2024-01-01 12:00:00Z], conversation_id: "conv-123"}}

get_name(agent_id)

@spec get_name(String.t()) :: GenServer.name()

Get the name of the AgentServer process for a specific agent.

Examples

name = AgentServer.get_name("my-agent-1")
GenServer.call(name, :get_status)

get_pid(agent_id)

@spec get_pid(String.t()) :: pid() | {atom(), node()} | nil

Get the pid of the AgentServer process for a specific agent.

Examples

pid = AgentServer.get_pit("my-agent-1")
send(pid, message)

get_status(agent_id)

@spec get_status(String.t()) :: status() | :not_running

Get the current status of the server.

Returns one of:

  • :idle - Server ready for work
  • :running - Agent executing
  • :interrupted - Awaiting human decision
  • :cancelled - Execution was cancelled
  • :error - Execution failed
  • :not_running - Agent process does not exist

Examples

:idle = AgentServer.get_status("my-agent-1")
:not_running = AgentServer.get_status("non-existent-agent")

list_agents_matching(pattern)

@spec list_agents_matching(String.t()) :: [String.t()]

Gets all running agents matching a glob pattern.

Supports wildcard patterns using * which matches any sequence of characters.

Examples

# Get all conversation agents
AgentServer.list_agents_matching("conversation-*")
# => ["conversation-1", "conversation-2", "conversation-123"]

# Get all user agents
AgentServer.list_agents_matching("user-*")
# => ["user-42", "user-99"]

# Get specific prefix
AgentServer.list_agents_matching("demo-*")
# => ["demo-agent-001"]

list_running_agents()

@spec list_running_agents() :: [String.t()]

Lists all currently running agent processes.

Returns a list of agent_ids for all running AgentServer processes registered in the Sagents.Registry.

Examples

AgentServer.list_running_agents()
# => ["conversation-1", "conversation-2", "user-123"]

notify_middleware(agent_id, middleware_id, message)

@spec notify_middleware(String.t(), term(), term()) :: :ok

Send a targeted message to a specific middleware in a running AgentServer.

The message is routed through the middleware registry and delivered to the middleware's handle_message/3 callback. The middleware is identified by its ID (module name by default, or a custom string if configured via :id option).

This is a fire-and-forget operation — the caller does not wait for a response. If the AgentServer is not running, the message is silently dropped.

Use Cases

There are two primary use cases for this function:

1. External notifications (LiveViews, controllers, other processes)

Send context updates or configuration changes to a middleware from outside the agent system. The middleware decides how to handle the message — typically by updating state metadata that before_model/2 reads on the next LLM call.

# LiveView: user switched to editing a different blog post
AgentServer.notify_middleware(agent_id, MyApp.UserContext, {:post_changed, %{
  slug: "/blog/getting-started-with-elixir",
  title: "Getting Started with Elixir"
}})

# Controller: user changed a preference
AgentServer.notify_middleware(agent_id, MyApp.Preferences, {:preference_changed, :verbose, true})

2. Async task results (middleware sending messages to itself)

Middleware that spawns background tasks (e.g., title generation, embedding computation) uses this to send results back to the AgentServer for state updates.

# Inside an async task spawned by the middleware
AgentServer.notify_middleware(agent_id, middleware_id, {:title_generated, title})

Parameters

  • agent_id - The agent identifier (used to locate the AgentServer process)
  • middleware_id - The middleware ID to route the message to (module name or custom string)
  • message - Any term to be delivered to the middleware's handle_message/3 callback

Returns

  • :ok — always returns :ok, even if the AgentServer is not running

Examples

# Notify by module name (default middleware ID)
AgentServer.notify_middleware("conv-123", MyApp.Middleware.UserContext, {:post_changed, post})

# Notify by custom ID (when middleware was configured with `id: "english_title"`)
AgentServer.notify_middleware("conv-123", "english_title", {:regenerate, %{}})

on_subscribed(arg1, subscriber_pid, state)

Hook fired after a subscriber is registered. Default no-op. Override to send a snapshot of the current state to the new subscriber so it can sync without polling.

publish_debug_event_from(agent_id, event)

@spec publish_debug_event_from(String.t(), term()) :: :ok

Request the AgentServer to publish a specific debug PubSub message or event.

Designed to make it easier for middleware to publish debug messages to the Agent's debug PubSub. Debug events are useful for development and debugging but separate from user-facing events.

Standardized Middleware Action Pattern

Middleware should use the :middleware_action tuple pattern to avoid event proliferation:

{:middleware_action, middleware_module, action_data}

Where:

  • middleware_module - The middleware module (atom) that generated the event
  • action_data - Middleware-specific action tuple or data

This pattern allows the debug UI to handle all middleware events generically without needing to know about every possible middleware-specific event type.

Examples

# From ConversationTitle middleware
AgentServer.publish_debug_event_from(
  agent_id,
  {:middleware_action, Sagents.Middleware.ConversationTitle, {:title_generation_started, user_text}}
)

AgentServer.publish_debug_event_from(
  agent_id,
  {:middleware_action, Sagents.Middleware.ConversationTitle, {:title_generation_completed, title}}
)

# From custom middleware
AgentServer.publish_debug_event_from(
  agent_id,
  {:middleware_action, MyApp.CustomMiddleware, {:validation_started, params}}
)

publish_event_from(agent_id, event)

@spec publish_event_from(String.t(), term()) :: :ok

Request the AgentServer to publish an specific PubSub message or event.

Designed to make it easier for a middleware desiring to publish messages to the Agent's PubSub.

A PubSub message is only broadcast if the AgentServer is configured with PubSub.

queue_message_from_tool(agent_id, message, opts \\ [])

@spec queue_message_from_tool(String.t(), LangChain.Message.t(), keyword()) ::
  :ok | {:error, :no_server}

Queue a message into the conversation from inside a running tool or middleware.

This is the tool-facing door to the same queue add_message/3 uses. The message is held for the duration of the current run and delivered at the run boundary, arriving as an ordinary :user message at the head of a follow-up run. Nothing downstream needs to distinguish it from a message the user typed, because there is nothing to distinguish: it is an ordinary user message.

Use this when a tool needs to hand the model instructions rather than data. Models are trained to treat a tool result as data and a user turn as instruction, so returning a playbook as tool-result text is a functional downgrade, not a stylistic one.

Why this is a cast

handle_call(:cancel, ...) blocks the AgentServer in Task.shutdown(task, 2_000). Tool code runs inside that Task. A tool blocked in a GenServer.call back to its own server during a cancel is a mutual wait that only resolves when the 2s grace expires and the task is brutal-killed. Both existing tool-to-server helpers (publish_event_from/2, save_synthetic_message_from/2) are casts for the same reason.

The cost is that the caller cannot learn whether the queue accepted the message. The GenServer.whereis guard below covers the dominant failure mode: no server at all, which is unit tests and bare Sagents.Agent.execute/3. It returns {:error, :no_server} so a tool can fall back to inline content rather than silently doing nothing.

Options

Same :display option as add_message/3, with one difference: a configured Sagents.MessagePreprocessor is never run for a queued message. The behaviour is scoped to messages a human submitted; a tool-generated playbook is machine-generated and there is nothing coherent for a preprocessor to do with it.

Prefer display: :none. The model emits a narration turn after the tool result anyway, and that turn becomes its own display message. Supplying a display: acknowledgement as well produces two acknowledgements back to back. Reach for an explicit display: only when the wording, role, or guaranteed presence of the acknowledgement has to be deterministic.

Returns

  • :ok - the cast was sent
  • {:error, :no_server} - no AgentServer is running for this agent_id

Limitations

Sub-agents register under {:sub_agent, id}, not {:agent_server, id}, and block their own process for the whole of SubAgentServer.execute/1. A tool running inside a sub-agent therefore gets {:error, :no_server} and should take its fallback path. Queueing into the parent conversation is deliberately not offered.

Examples

def activate_command(%{"command" => name} = args, context) do
  playbook = render_playbook(name, args)

  case AgentServer.queue_message_from_tool(context.state.agent_id,
         Message.new_user!(playbook),
         display: :none
       ) do
    :ok ->
      {:ok, "The #{name} command has been initiated. Its instructions " <>
            "will arrive as the next message. Acknowledge briefly and stop."}

    {:error, :no_server} ->
      # Degraded path: no AgentServer (unit test, one-shot Agent.execute).
      # Return the playbook inline rather than silently doing nothing.
      {:ok, playbook}
  end
end

reset(agent_id)

@spec reset(String.t()) :: :ok | {:error, term()}

Reset the agent's state and filesystem to start fresh.

This clears:

  • All messages
  • All TODOs
  • Middleware state
  • Memory-only files (completely removed)
  • In-memory modifications to persisted files (discarded)

This preserves:

  • Metadata (configuration)
  • Persisted files (reverted to pristine state from storage)

Status transitions:

  • :completed, :error, or :cancelled:idle (ready for new execution)
  • Other statuses remain unchanged

Returns :ok on success.

Examples

# After agent completes or encounters error
:ok = AgentServer.reset("my-agent-1")
# Now you can execute again with clean state
:ok = AgentServer.execute("my-agent-1")

restore_state(agent_id, persisted_state)

@spec restore_state(String.t(), map()) :: :ok | {:error, term()}

Restore agent state from a previously exported state.

This updates an already-running agent to restore its state from a previously serialized format. The state should be a map with string keys (as returned by export_state/1).

Returns :ok on success or {:error, reason} on failure.

Examples

# Load from database
{:ok, persisted_state} = MyApp.Conversations.load_agent_state(conversation_id)

# Restore into existing agent
:ok = AgentServer.restore_state("my-agent-1", persisted_state)

resume(agent_id, resume_data)

@spec resume(String.t(), term()) :: :ok | {:error, term()}

Resume agent execution after an interrupt.

Parameters

  • agent_id - The agent identifier
  • resume_data - Data to resume with (polymorphic per middleware). For HITL: list of decision maps. For AskUserQuestion: response map.

Examples

# HITL resume
decisions = [
  %{type: :approve},
  %{type: :edit, arguments: %{"path" => "safe.txt"}},
  %{type: :reject}
]
:ok = AgentServer.resume("my-agent-1", decisions)

# AskUserQuestion resume
:ok = AgentServer.resume("my-agent-1", %{type: :answer, selected: ["PostgreSQL"]})

running?(agent_id)

Check if an agent is running.

save_synthetic_message_from(agent_id, attrs)

@spec save_synthetic_message_from(String.t(), map()) :: :ok

Persist a synthetic display message and broadcast it to subscribers.

Designed for middleware that needs to record user-facing transcript entries that do not correspond to an LLM message (for example, a user's answer to an ask_user question, or a "user cancelled" notification).

The attrs map should contain at minimum :message_type, :content_type, and :content. AgentServer routes the request to the configured Sagents.DisplayMessagePersistence implementation and broadcasts the saved record via {:display_message_saved, msg}.

No-op if the AgentServer was not configured with both display_message_persistence and conversation_id, or if the configured persistence module does not implement save_synthetic_message/3.

start_link(opts)

Start an AgentServer.

Options

  • :agent - The Agent struct (required)

  • :initial_state - Initial State (default: empty state)

  • :initial_subscribers - List of {channel, pid} tuples to enroll as subscribers before init/1 returns. Use this to atomically start the server and subscribe — every event broadcast (including the initial {:status_changed, :idle, nil} and any {:node_transferred, _} after a Horde restore) is delivered to listed pids. Channels are :main and :debug. Default: [].

  • :pubsub - PubSub configuration as {module(), atom()} tuple or nil (default: nil). Used only for presence wiring (subscribing to Phoenix.Presence diff broadcasts). Per-agent events are delivered directly to subscribers via Sagents.Publisher, no PubSub required.

  • :name - Server name registration (optional, defaults to get_name(agent.agent_id))

  • :inactivity_timeout - Timeout in milliseconds for automatic shutdown due to inactivity (default: 300_000 - 5 minutes) Set to nil or :infinity to disable automatic shutdown

  • :shutdown_delay - Delay in milliseconds to allow the supervisor to gracefully stop all children (default: 5000)

  • :conversation_id - Optional conversation identifier for message persistence (default: nil)

  • :agent_persistence - Module implementing Sagents.AgentPersistence for state snapshots (default: nil)

  • :display_message_persistence - Module implementing Sagents.DisplayMessagePersistence for display messages (default: nil)

  • :pending_resume - A resume payload to apply during boot, for an answer submitted while no process was alive to take it. Applied in handle_continue/2 before the initial status broadcast, so a server that boots :interrupted and consumes this announces :running once, rather than announcing an :interrupted snapshot it is about to leave. Discarded with a warning if the server boots into any other status (the interrupt was demoted as non-restorable, or answered elsewhere first). Prefer Sagents.Session.resume/4, which sets this for you and handles the live-agent and already-woken cases. Default: nil.

    Note it lives in the child spec, so a supervisor restart re-evaluates it against freshly loaded persisted state. If the resume already completed, that state no longer carries an interrupt and the payload is discarded; if the run died before completing, re-applying it is the correct recovery. The supervisor's restart intensity bounds a pathological loop.

Examples

# Start with automatic name (recommended)
{:ok, pid} = AgentServer.start_link(
  agent: agent,
  initial_state: state
)

# With PubSub enabled
{:ok, pid} = AgentServer.start_link(
  agent: agent,
  initial_state: state,
  pubsub: {Phoenix.PubSub, :my_app_pubsub}
)

# Start with explicit name (advanced use cases)
{:ok, pid} = AgentServer.start_link(
  agent: agent,
  initial_state: state,
  name: :my_custom_name
)

# With custom inactivity timeout
{:ok, pid} = AgentServer.start_link(
  agent: agent,
  inactivity_timeout: 600_000  # 10 minutes
)

# Disable automatic shutdown
{:ok, pid} = AgentServer.start_link(
  agent: agent,
  inactivity_timeout: nil
)

stop(agent_id)

@spec stop(String.t()) :: :ok

Stop the AgentServer.

Examples

:ok = AgentServer.stop("my-agent-1")

subscribe(agent_id, channel \\ :main, subscriber_pid \\ nil)

@spec subscribe(String.t(), :main | :debug, pid() | nil) ::
  {:ok, pid(), reference()} | {:error, :process_not_found}

Subscribe a process to events from this AgentServer.

Events on the :main channel are delivered as {:agent, event} messages; events on the :debug channel are delivered as {:agent, {:debug, event}} and provide additional insight into middleware state, sub-agent activity, and similar diagnostic data not surfaced on the main channel.

Delivery is via direct send/2. The producer monitors the subscriber so departure is cleaned up automatically — but subscribers should also Process.monitor/1 the returned server_pid to detect server death.

Arguments

  • agent_id — the agent's id.
  • channel:main (default) or :debug.
  • subscriber_pid — the pid to receive events. Defaults to self() when nil.

Returns {:ok, server_pid, monitor_ref} on success, or {:error, :process_not_found} if no AgentServer is running for agent_id.

Examples

# Most common: subscribe self() to the main channel.
{:ok, _pid, _ref} = AgentServer.subscribe("my-agent-1")

# Subscribe to debug events (used by sagents_live_debugger and similar).
{:ok, _pid, _ref} = AgentServer.subscribe("my-agent-1", :debug)

# Subscribe a foreign pid (e.g. a bridge GenServer that proxies events).
{:ok, _pid, _ref} = AgentServer.subscribe("my-agent-1", :main, bridge_pid)

touch(agent_id)

@spec touch(String.t()) :: :ok

Touch the agent to indicate activity and reset the inactivity timer.

This is useful when external events indicate activity (e.g., user viewing the agent in the UI, clicking tabs, etc.) to keep the agent alive and prevent automatic shutdown due to inactivity.

Returns :ok immediately (non-blocking).

Examples

:ok = AgentServer.touch("my-agent-1")

unsubscribe(agent_id, channel \\ :main, subscriber_pid \\ nil)

@spec unsubscribe(String.t(), :main | :debug, pid() | nil) :: :ok

Unsubscribe a process from events on the given channel.

Mirrors subscribe/3. Defaults channel to :main and subscriber_pid to self() (when nil). Always returns :ok.

update_agent_and_state(agent_id, agent, state)

Updates both the agent configuration and state.

This is the recommended way to restore a conversation:

  1. Create agent from current code using your agent factory
  2. Load state from database
  3. Call this function to update the running AgentServer

Parameters

  • agent_id - The agent server's identifier
  • agent - The new agent configuration (from current code)
  • state - The restored state (from database)

Examples

# Restore conversation
{:ok, agent} = MyApp.Agents.create_demo_agent(agent_id: "demo-123")
{:ok, state_data} = MyApp.Conversations.load_state(conversation_id)
{:ok, state} = Sagents.State.from_serialized(state_data["state"])

:ok = AgentServer.update_agent_and_state("demo-123", agent, state)

Returns

  • :ok - Agent and state updated successfully
  • {:error, reason} - If agent server is not running or update fails