@spec all([agent_or_pid()], term(), keyword()) ::
  {:ok, %{required(String.t()) => result()}} | result()

Waits for all agents to complete and returns all results.

Executes all agents concurrently and waits for all to finish, collecting both successes and failures.

Options

• :timeout (timeout/0) - How long to wait, in ms. The default value is 60000.

• :on_complete (function of arity 2) - Called as each agent finishes: (agent_id, result -> any).

• :max_concurrency (pos_integer/0) - Agents run at once (default: all of them).

• :fail_fast (boolean/0) - Stop at the first failure. The default value is false.

Examples

# Get all results
{:ok, results} = Reactive.all([agent1, agent2, agent3], input)
#=> {:ok, %{
       "agent_0" => {:ok, result1},
       "agent_1" => {:ok, result2},
       "agent_2" => {:error, reason}
     }}

# With concurrency limit
{:ok, results} = Reactive.all(many_agents, input, max_concurrency: 5)

# Fail fast on first error
{:error, reason} = Reactive.all(agents, input, fail_fast: true)

Returns

• {:ok, %{agent_id => {:ok, output} | {:error, exception}}} - every agent's result. An agent that runs past :timeout (the limit applies to each agent) is stopped and appears as {:error, %Normandy.TurnError{type: :task_down, reason: :timeout}}; one that exits, as that TurnError with the exit reason

• {:error, exception} - with fail_fast: true, the first failure

@spec map(agent_or_pid(), term(), (result() -> result())) :: result()

Executes an agent and applies a transformation if successful.

Useful for chaining operations based on agent responses.

Examples

result = Reactive.map(agent, input, fn
  {:ok, %{confidence: c}} when c > 0.8 ->
    {:ok, :high_confidence}

  {:ok, %{confidence: c}} when c < 0.5 ->
    {:ok, :low_confidence}

  {:ok, _} ->
    {:ok, :medium_confidence}

  error ->
    error
end)

@spec race([agent_or_pid()], term(), keyword()) :: result()

Races multiple agents and returns the first successful result.

Starts all agents concurrently and returns as soon as any agent succeeds. The agents still running are then killed (see "Telemetry" in the moduledoc).

Options

• :timeout (timeout/0) - How long to wait, in ms. The default value is 60000.

• :on_complete (function of arity 2) - Called as each agent finishes: (agent_id, result -> any).

Examples

# With agent structs
{:ok, fastest_result} = Reactive.race(
  [research_agent, search_agent, cached_agent],
  query,
  timeout: 5000
)

# With agent processes
{:ok, result} = Reactive.race([pid1, pid2, pid3], input)

# With callback
Reactive.race(agents, input,
  on_complete: fn agent_id, result ->
    Logger.info("Agent #{agent_id} completed: #{inspect(result)}")
  end
)

Returns

• {:ok, result} - First successful result
• {:error, :all_failed} - All agents failed
• {:error, :timeout} - Timeout reached before any success

@spec some([agent_or_pid()], term(), keyword()) ::
  {:ok, %{required(String.t()) => term()}} | result()

Waits for N successful results from agents.

Executes agents concurrently until the specified number of successful results is reached. Useful for quorum-based decisions.

Options

• :count (pos_integer/0) - Required. Successes needed.

• :timeout (timeout/0) - How long to wait, in ms. The default value is 60000.

• :on_complete (function of arity 2) - Called as each agent finishes: (agent_id, result -> any).

Examples

# Wait for 2 out of 3 agents to succeed (quorum)
{:ok, results} = Reactive.some(
  [agent1, agent2, agent3],
  input,
  count: 2,
  timeout: 10000
)
#=> {:ok, %{"agent_0" => result1, "agent_2" => result3}}

# Majority vote
agents = [voter1, voter2, voter3, voter4, voter5]
{:ok, votes} = Reactive.some(agents, question, count: 3)

Returns

• {:ok, %{agent_id => result}} - Map of N successful results
• {:error, :insufficient_successes} - Not enough agents succeeded
• {:error, :timeout} - Timeout reached before N successes
• Raises ArgumentError when :count is above the number of agents.

Executes an agent and conditionally executes another based on the result.

Examples

Reactive.when_result(agent1, input) do
  {:ok, %{needs_review: true}} ->
    AgentProcess.run(review_agent, "Please review")

  {:ok, %{confidence: c}} when c < 0.5 ->
    AgentProcess.run(fallback_agent, "Use fallback")

  {:ok, result} ->
    {:ok, result}

  error ->
    error
end