CrowdControl (crowd_control v0.1.1)

Copy Markdown View Source

Orchestrate many coding-agent CLI instances in parallel.

Claude Code, Open Code and omp are supported; pick one per session with :agent (see CrowdControl.Agent). Every agent reports through the same message contract, so a mixed fan-out collects uniformly.

See CrowdControl.CLI.build_command/1 (Claude Code / Open Code) and CrowdControl.Agent.Omp (omp) for the full list of session options.

Summary

Functions

Send the same prompt to all sessions.

Wait for all sessions to produce a result message.

Returns true if the session supervisor is alive and healthy.

Single-shot convenience: start a session, send prompt, collect result, stop.

Start N sessions with different options but the same prompt; collect all results.

Start a single session under the supervisor.

Start multiple sessions in parallel.

Stop all sessions in parallel.

Types

opts()

@type opts() :: keyword()

result_message()

@type result_message() :: {:result, String.t(), map()}

session()

@type session() :: pid()

Functions

broadcast(sessions, prompt)

@spec broadcast([session()], binary()) :: :ok

Send the same prompt to all sessions.

collect(sessions, timeout \\ 60000)

@spec collect([session()], pos_integer()) ::
  [{session(), result_message()}] | {:timeout, [{session(), result_message()}]}

Wait for all sessions to produce a result message.

Subscribes to each session and collects {:result, _, _} messages. Returns a list of {session_pid, result_message} tuples, or {:timeout, partial_results} if the deadline is reached first.

A session that broadcasts a terminal message instead of a result -- it exited, expired, or hit :line_too_large -- is dropped from the wait set rather than waited out, since it can never produce one. Such a session is simply absent from the returned list, so the list can be shorter than sessions.

Results are matched by turn. Session.subscribe/1 replays history, so a session already carrying a finished turn would otherwise hand the collector that old result the instant it attached -- returning stale data for a turn still in flight. Each session's Session.current_turn/1 is read before subscribing, and only results stamped with that turn or later are accepted.

healthy?()

@spec healthy?() :: boolean()

Returns true if the session supervisor is alive and healthy.

run(prompt, opts \\ [])

@spec run(binary(), opts()) :: result_message() | {:error, term()}

Single-shot convenience: start a session, send prompt, collect result, stop.

run_many(prompt, opts_list)

@spec run_many(binary(), [opts()]) ::
  [{session(), result_message()}]
  | {:timeout, [{session(), result_message()}]}
  | {:error, term()}

Start N sessions with different options but the same prompt; collect all results.

start_session(opts \\ [])

@spec start_session(opts()) :: {:ok, session()} | {:error, term()}

Start a single session under the supervisor.

Options are passed through to CrowdControl.Session and CrowdControl.CLI.

Additional option:

  • :prompt - initial prompt to send after the CLI starts

start_sessions(opts_list)

@spec start_sessions([opts()]) :: {:ok, [session()]} | {:error, [term()]}

Start multiple sessions in parallel.

Takes a list of option keyword lists. Returns {:ok, pids} if every session started, or {:error, errors} listing the failing entries.

Concurrency is clamped to the configured :max_sessions cap.

stop_all(sessions)

@spec stop_all([session()]) :: :ok

Stop all sessions in parallel.