Normandy.Coordination.AgentProcess (normandy v2.2.0)

View Source

GenServer wrapper for running BaseAgent instances as supervised processes.

AgentProcess enables agents to run as long-lived processes that can:

  • Maintain state across multiple invocations
  • Be supervised and restarted on failure
  • Receive messages asynchronously
  • Integrate with process registries

Example

# Start an agent process
{:ok, pid} = AgentProcess.start_link(
  agent: my_agent,
  name: :research_agent
)

# Execute synchronously
{:ok, result} = AgentProcess.run(pid, "Analyze this data")

# Execute asynchronously
:ok = AgentProcess.cast(pid, "Process in background", reply_to: self())

# Get current agent state
agent = AgentProcess.get_agent(pid)

Inline turns (the default)

With turn_engine: :inline, runs, casts and update_agent/2 calls go through one FIFO queue and execute one at a time, each turn in a worker process started from the config the previous one left, so a cast's turn is kept in memory like a run's. The process itself never waits on a turn: get_agent/1 (the config as of the last finished turn), get_id/1, get_stats/1 and approve/2 answer at once while a turn runs. A run whose caller died while it waited in the queue is skipped. The worker dies with the process (stop/1 kills a turn in flight); a worker killed from outside replies a :task_down Normandy.TurnError.

Pending runs

Runs and casts that were admitted and have not finished (queued or in flight, in either mode) are bounded by :max_pending_runs (default 100). One more is refused at once: run/3 returns, and cast/3 sends to :reply_to, {:error, %Normandy.TurnError{type: :queue_full, reason: limit}}. get_stats/1's run_count counts runs and casts that finished, failed ones included.

Durable turn engine (:server mode)

Pass turn_engine: :server to route turns through the durable Normandy.Agents.Turn.Session/Turn.Server engine (approval parking, passivation, persistence) instead of the default :inline path, which runs BaseAgent.run/3 in a worker process, one turn at a time. :inline is the default and is unchanged.

{:ok, pid} = AgentProcess.start_link(agent: config, turn_engine: :server)

In :server mode:

  • run/3/cast/3 route through Turn.Session; run/3 is non-blocking internally (the GenServer stays responsive while a turn is parked); approve/2 delivers human-approval decisions to a parked turn.
  • The SessionStore owns conversation memory: get_agent/1 reconstructs it from the store, and update_agent/2 updates only the config template (model/temperature/behaviours/tools) — memory mutations are ignored.
  • Session infra (:store, :registry, :supervisor) may be supplied via start_link; if omitted, the process starts and owns an in-memory store and a Turn.Supervisor that terminate with it, and registers its sessions under a scope of its own in a registry Normandy starts (no atom per process). start_link builds one Turn.Session from them and from :supervisor_mod, :template_provider, :template_id, :resume_policy, :subscriber, :handlers, :approval_timeout_ms, :on_approval_timeout and :idle_timeout_ms (see Turn.Session.new/1; start_link/1 checks them with its types); update_agent/2 gives it the new config for the next server start.

Summary

Types

agent_id()
run_opts()

Functions

approve(server, decisions)

Delivers approval decisions to a turn parked awaiting human approval (:server mode only). decisions maps tool_call_id to :approve | :reject; any parked id absent or :reject is treated as rejected (fail-closed).

cast(server, input, opts \\ [])

Executes the agent asynchronously.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

get_agent(server)

Returns the current agent state.

get_id(server)

Returns the agent ID.

get_stats(server)

Returns agent statistics and metadata.

run(server, input, opts \\ [])

Executes the agent synchronously.

start_link(opts)

Starts an AgentProcess GenServer.

stop(server)

Stops the agent process gracefully.

update_agent(server, update_fn, opts \\ [])

Updates the agent config: update_fn gets the current %Normandy.Agents.BaseAgentConfig{} and returns the new one.

Types

agent_id()

@type agent_id() :: String.t()

run_opts()

@type run_opts() :: [{:timeout, timeout()}]

Functions

approve(server, decisions)

@spec approve(GenServer.server(), %{optional(String.t()) => :approve | :reject}) ::
  :ok | {:error, Normandy.TurnError.t()} | {:error, :inline_mode}

Delivers approval decisions to a turn parked awaiting human approval (:server mode only). decisions maps tool_call_id to :approve | :reject; any parked id absent or :reject is treated as rejected (fail-closed).

Returns :ok; with no live parked session, {:error, %Normandy.TurnError{type: :session_unavailable, reason: :no_session}} (see Normandy.Agents.Turn.Session.approve/2); {:error, :inline_mode} when the process runs in :inline mode.

cast(server, input, opts \\ [])

@spec cast(GenServer.server(), term(), keyword()) :: :ok

Executes the agent asynchronously.

The agent runs in the background. If :reply_to is provided, sends {:agent_result, agent_id, result} when complete.

Options

  • :reply_to - Where to send {:agent_result, agent_id, result} when the run ends.

Example

:ok = AgentProcess.cast(pid, input, reply_to: self())

receive do
  {:agent_result, agent_id, result} ->
    IO.inspect(result)
end

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

get_agent(server)

@spec get_agent(GenServer.server()) :: struct()

Returns the current agent state.

Example

agent = AgentProcess.get_agent(pid)

get_id(server)

@spec get_id(GenServer.server()) :: agent_id()

Returns the agent ID.

Example

agent_id = AgentProcess.get_id(pid)
#=> "agent_1"

get_stats(server)

@spec get_stats(GenServer.server()) :: map()

Returns agent statistics and metadata.

Example

stats = AgentProcess.get_stats(pid)
#=> %{
  agent_id: "agent_1",
  run_count: 42,
  last_run: ~U[2024-01-15 10:30:00Z],
  total_runtime_ms: 15420
}

run(server, input, opts \\ [])

@spec run(GenServer.server(), term(), run_opts()) ::
  {:ok, term()} | {:error, Exception.t()}

Executes the agent synchronously.

Runs the agent with the given input and returns the result.

Options

  • :timeout (timeout/0) - How long the caller waits for the reply, in ms. A run is bounded by its retries, tool timeouts and max_tool_iterations, so the default waits for it, as Turn.Server.run/2 does. The default value is :infinity.

Example

{:ok, result} = AgentProcess.run(pid, "What is AI?")

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

Starts an AgentProcess GenServer.

Options

  • :agent (a %Normandy.Agents.BaseAgentConfig{}) - Required. The agent config the process runs.

  • :name (term/0) - A GenServer.name/0 for the process.

  • :agent_id (String.t/0) - The agent's id (default: a new UUID). In :server mode it is also the session id.

  • :context_pid - A Normandy.Coordination.StatefulContext process, kept in the agent's state.

  • :turn_engine - :inline runs BaseAgent.run/3 in a worker process, one turn at a time; :server routes turns through Turn.Session/Turn.Server (approval parking, passivation, persistence). The default value is :inline.

  • :store ({module, handle}) - :server mode: the SessionStore as {module, handle}. Without it the process starts and owns an in-memory store.

  • :registry ({module, handle}) - :server mode: the SessionRegistry as {module, handle}. Without it the process uses a scope of its own in a registry Normandy starts.

  • :supervisor (a pid or a supervisor name) - :server mode: the session supervisor. Without it the process starts and owns a Turn.Supervisor.

  • :supervisor_mod - :server mode: the module whose start_server/2 starts sessions (default Normandy.Agents.Turn.Supervisor).

  • :template_provider ({module, handle}) - :server mode: the AgentTemplate as {module, handle}.

  • :resume_policy - :server mode: the session's resume policy (default :lazy).

  • :template_id (String.t/0) - :server mode: the template id (default: the config's name).

  • :subscriber (function of arity 2) - :server mode: passed to Turn.Server.start_link/1.

  • :handlers (a Normandy.Agents.Turn.Driver.Handlers or nil) - :server mode: passed to Turn.Server.start_link/1.

  • :approval_timeout_ms (timeout/0) - :server mode: passed to Turn.Server.start_link/1.

  • :on_approval_timeout - :server mode: passed to Turn.Server.start_link/1.

  • :idle_timeout_ms (timeout/0) - :server mode: passed to Turn.Server.start_link/1.

  • :max_pending_runs (a positive integer or :infinity) - Runs and casts admitted and not finished (queued or in flight). One more is refused with a :queue_full Normandy.TurnError. The default value is 100.

Example

{:ok, pid} = AgentProcess.start_link(
  agent: my_agent,
  name: :my_agent,
  agent_id: "agent_1"
)

stop(server)

@spec stop(GenServer.server()) :: :ok

Stops the agent process gracefully.

Example

:ok = AgentProcess.stop(pid)

update_agent(server, update_fn, opts \\ [])

@spec update_agent(
  GenServer.server(),
  (Normandy.Agents.BaseAgentConfig.t() -> Normandy.Agents.BaseAgentConfig.t()),
  [{:timeout, timeout()}]
) :: :ok

Updates the agent config: update_fn gets the current %Normandy.Agents.BaseAgentConfig{} and returns the new one.

Raises ArgumentError when update_fn returns anything else, and in :server mode when the session refuses the new config (Normandy.Agents.Turn.Session.put_config/2: a session with a :template_provider needs a credential provider); the process keeps its config.

In :inline mode the update is queued behind the turn in flight and any runs before it (see "Inline turns"), so it sees their memory; the call waits for it.

Options

  • :timeout (timeout/0) - How long the caller waits for the reply, in ms. In :inline mode the update waits behind the turn in flight, so the default waits for it, as run/3 does. The bound is on the wait: an update the caller stopped waiting for still applies. The default value is :infinity.

Example

:ok = AgentProcess.update_agent(pid, fn agent ->
  %{agent | temperature: 0.2}
end)