CrowdControl.Session (crowd_control v0.2.0)

GenServer managing a single coding-agent CLI instance.

Each session drives a CrowdControl.Backend — a local subprocess by default, or a container — plus a linked reader process that delivers stdout. The session buffers partial lines, decodes JSON messages, and broadcasts them to subscribers. Everything from line splitting onward is transport-agnostic: the session never learns which backend it is talking to.

Subscribers receive messages of the form {:crowd_control, session_pid, payload} where payload is one of CrowdControl.Protocol.message/0, {:exit, exit_status}, {:timeout, :session_expired}, or {:error, reason} (e.g. {:error, :line_too_large} when a single newline-free output line exceeds :max_line_bytes).

Options

Session-lifecycle options (CLI/argv options are forwarded to the agent adapter's build_command/1, e.g. CrowdControl.CLI.build_command/1):

:agent - CrowdControl.Agent adapter selecting the CLI dialect: :claude (default), :open_code, :omp, or a module. Inferred from :executable when omitted.
:backend - CrowdControl.Backend implementation, either a module or a {module, config} tuple whose config is merged into these opts. Defaults to CrowdControl.Backend.Local.
:prompt - initial prompt sent once the CLI starts (optional)
:timeout - ceiling in ms on a single turn, after which the session self-expires and broadcasts {:timeout, :session_expired}. The timer is armed at start and re-armed by send_prompt/2 — not by output, so a turn that streams for longer than this is still killed mid-flight. Size it against the slowest turn you expect, not the length of the conversation; long autonomous tasks against a self-hosted model routinely need more than the default. Use :infinity or nil to disable. Defaults to 300_000.
:max_prompt_size - reject prompts whose byte size exceeds this with {:error, :prompt_too_large} (optional; unbounded when unset)
:max_line_bytes - cap for a single newline-free output line. Exceeding it kills the subprocess and broadcasts {:error, :line_too_large} rather than buffering an unbounded remainder. Defaults to 1_048_576 (1 MiB).
:max_stream_bytes - cap on a session's total output. Exceeding it destroys the sandbox and broadcasts {:error, :stream_too_large}. Mainly for remote backends, whose output file grows without bound; unbounded when unset.
:max_messages - cap on messages retained for get_messages/1; oldest are dropped past the cap. Live subscribers are unaffected. Clamped to >= 0. Defaults to 10_000.

Summary

Types

session()

status()

t()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

current_turn(session)

The turn currently in flight, i.e. the number of prompts written so far.

get_messages(session)

Get accumulated messages in chronological order.

get_session_id(session)

Get the session ID (assigned by the CLI on init).

get_status(session)

Get the current session status.

send_prompt(session, prompt)

Send a user prompt to the CLI subprocess.

start_link(opts)

Start a session linked to the caller.

start_reattached(record)

Start a session that takes over a sandbox which outlived its previous session.

stop(session)

Gracefully stop the session.

subscribe(session)

Subscribe the calling process to receive session messages.

Types

session()

@type session() :: pid()

status()

@type status() :: :starting | :running | :completed | :error

t()

@type t() :: %CrowdControl.Session{
  agent: term(),
  agent_opts: term(),
  backend: term(),
  backend_state: term(),
  buffer: term(),
  byte_offset: term(),
  exit_status: term(),
  exited: term(),
  max_line_bytes: term(),
  max_messages: term(),
  max_prompt_size: term(),
  max_stream_bytes: term(),
  message_count: term(),
  messages: term(),
  opts: term(),
  owner: term(),
  persist?: term(),
  prompt_seq: term(),
  reader: term(),
  session_id: term(),
  status: term(),
  store_key: term(),
  subscribers: term(),
  timeout: term(),
  timeout_ref: term()
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

current_turn(session)

@spec current_turn(session()) :: non_neg_integer()

The turn currently in flight, i.e. the number of prompts written so far.

Every {:result, _, map} carries the same number under "turn". A collector that reads this before subscribing can tell a replayed result from an earlier turn apart from the one it is waiting for; see CrowdControl.collect/2. Returns 0 for a session that has not been prompted.

get_messages(session)

@spec get_messages(session()) :: [CrowdControl.Protocol.message()]

Get accumulated messages in chronological order.

Retention is capped at :max_messages (default 10000); once the cap is reached the oldest messages are dropped so the returned list is a bounded, newest-biased window. Live subscribers (see subscribe/1) receive every message regardless of this cap.

get_session_id(session)

@spec get_session_id(session()) :: String.t() | nil

Get the session ID (assigned by the CLI on init).

get_status(session)

@spec get_status(session()) :: status()

Get the current session status.

send_prompt(session, prompt)

@spec send_prompt(session(), binary()) :: :ok | {:error, atom()}

Send a user prompt to the CLI subprocess.

Returns :ok on success or {:error, reason} where reason is one of :invalid_prompt, :prompt_too_large, :completed, :error.

start_link(opts)

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

Start a session linked to the caller.

start_reattached(record)

@spec start_reattached(CrowdControl.Store.t()) :: GenServer.on_start()

Start a session that takes over a sandbox which outlived its previous session.

record comes from CrowdControl.Store. Used by CrowdControl.Reaper; you rarely call this directly.

stop(session)

@spec stop(session()) :: :ok

Gracefully stop the session.