Normandy.Coordination.AgentPool (normandy v2.2.0)

View Source

Pool manager for agent processes with automatic checkout/checkin.

Provides efficient pooling of identical agent processes with configurable size, overflow handling, and checkout strategies. Built on top of AgentSupervisor for fault tolerance.

Features

  • Fixed pool size with configurable overflow
  • Multiple checkout strategies: :lifo, :fifo
  • Automatic checkin on transaction completion
  • Built-in supervision and fault tolerance
  • Pool statistics and monitoring

Examples

# Start a pool of 10 agents
{:ok, pool} = AgentPool.start_link(
  name: :my_pool,
  agent_config: my_agent_config,
  size: 10,
  max_overflow: 5,
  strategy: :fifo
)

# Use transaction for automatic checkout/checkin
{:ok, result} = AgentPool.transaction(pool, fn agent_pid ->
  AgentProcess.run(agent_pid, input)
end)

# Manual checkout/checkin
{:ok, agent_pid} = AgentPool.checkout(pool)
result = AgentProcess.run(agent_pid, input)
:ok = AgentPool.checkin(pool, agent_pid)

# Get pool statistics
stats = AgentPool.stats(pool)
#=> %{size: 10, available: 7, in_use: 3, overflow: 0}

Summary

Types

pool_stat()
strategy()

Functions

checkin(pool, agent_pid)

Returns an agent to the pool.

checkout(pool, opts \\ [])

Checks out an agent from the pool.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

start_link(opts)

Starts an agent pool.

stats(pool)

Returns pool statistics.

stop(pool)

Stops the agent pool gracefully.

transaction(pool, opts \\ [], fun)

Executes a function with an agent from the pool.

Types

pool_stat()

@type pool_stat() :: %{
  size: non_neg_integer(),
  overflow: non_neg_integer(),
  available: non_neg_integer(),
  in_use: non_neg_integer(),
  max_overflow: non_neg_integer()
}

strategy()

@type strategy() :: :lifo | :fifo

Functions

checkin(pool, agent_pid)

@spec checkin(GenServer.server(), pid()) :: :ok

Returns an agent to the pool.

Only the process that checked the agent out can return it. A checkin from any other process, such as a late second checkin after the agent was lent again, is ignored with a warning; an agent whose holder exits comes back on its own.

Examples

{:ok, agent_pid} = AgentPool.checkout(pool)
# ... use agent ...
:ok = AgentPool.checkin(pool, agent_pid)

checkout(pool, opts \\ [])

@spec checkout(
  GenServer.server(),
  keyword()
) :: {:ok, pid()} | {:error, term()}

Checks out an agent from the pool.

Options

  • :timeout (timeout/0) - How long to wait for a free agent, in ms (blocking checkout). The default value is 5000.

  • :block (boolean/0) - false returns {:error, :no_agents} at once when every agent is in use and the overflow limit is reached. The default value is true.

Returns

  • {:ok, agent_pid}
  • {:error, :timeout} - blocking, and no agent became available within :timeout
  • {:error, :no_agents} - non-blocking, and every agent is in use with the overflow limit reached
  • {:error, :failed_to_start_agent} - the pool could not start an overflow agent

Examples

{:ok, agent_pid} = AgentPool.checkout(pool)
result = AgentProcess.run(agent_pid, input)
AgentPool.checkin(pool, agent_pid)

# Non-blocking checkout
case AgentPool.checkout(pool, block: false) do
  {:ok, agent_pid} -> # use agent
  {:error, :no_agents} -> # pool empty
end

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

start_link(opts)

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

Starts an agent pool.

Options

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

  • :agent_config (a map or keyword list of BaseAgent.init/1 options) - Required. The BaseAgent.init/1 config every pooled agent is built from.

  • :size (pos_integer/0) - Agents started up front. The default value is 10.

  • :max_overflow (non_neg_integer/0) - Extra agents started while every pooled agent is checked out. The default value is 5.

  • :strategy - Which free agent a checkout gets: the most (:lifo) or least (:fifo) recently returned. The default value is :lifo.

Examples

{:ok, pool} = AgentPool.start_link(
  name: :research_pool,
  agent_config: %{
    client: client,
    model: "claude-opus-5-5"
  },
  size: 10,
  max_overflow: 5,
  strategy: :fifo
)

stats(pool)

@spec stats(GenServer.server()) :: pool_stat()

Returns pool statistics.

Examples

stats = AgentPool.stats(pool)
#=> %{
  size: 10,
  available: 7,
  in_use: 3,
  overflow: 0,
  max_overflow: 5,
  waiting: 2
}

stop(pool)

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

Stops the agent pool gracefully.

Terminates all agent processes and shuts down the pool.

Examples

:ok = AgentPool.stop(pool)

transaction(pool, opts \\ [], fun)

@spec transaction(GenServer.server(), keyword(), (pid() -> term())) ::
  {:ok, term()} | {:error, term()}

Executes a function with an agent from the pool.

Automatically checks out an agent, executes the function, and checks it back in. The agent is returned to the pool even when the function raises.

Options

  • :timeout (timeout/0) - How long to wait for a free agent, in ms. The default value is 5000.

Examples

{:ok, result} = AgentPool.transaction(pool, fn agent_pid ->
  AgentProcess.run(agent_pid, "Analyze this data")
end)

# With timeout
{:ok, result} = AgentPool.transaction(pool, [timeout: 10_000], fn agent_pid ->
  AgentProcess.run(agent_pid, input)
end)

Returns

  • {:ok, result} - the function's return value, whatever it is: a function that returns {:error, reason} gives {:ok, {:error, reason}}
  • {:error, :timeout} - no agent became available within :timeout
  • {:error, :failed_to_start_agent} - the pool tried to start an overflow agent and could not

A function that raises or exits is not caught: the error propagates to the caller after the agent is checked back in.