ExAthena.Orchestrator.Coordinator (ExAthena v0.17.0)

Copy Markdown View Source

Per-run orchestration blackboard.

One Coordinator GenServer observes one top-level run: it ingests loop events for the main agent and every subagent (attributed via notify/3), folds them through the pure AgentInfo reducers, and broadcasts batched snapshots to subscribed hosts. It is observational only — control flow stays in the agent loop:

  • All loop → coordinator traffic is cast; a coordinator crash can never stall or kill a run.
  • Runs never read coordinator state; hosts do, via snapshot/1 / subscribe/2 (a real call API — never :sys.get_state/1).

Supervision & naming

Started on demand under ExAthena.Orchestrator.Supervisor (DynamicSupervisor) with restart: :temporary, named {:via, Registry, {ExAthena.Orchestrator.Registry, session_id}}.

Subscribers & batching

Subscribers are monitored pids receiving {:orchestrator_update, session_id, snapshot} via plain send/2 (Phoenix.PubSub only exists under mix athena.web; send keeps the coordinator host-agnostic). Event ingest marks the state dirty and arms a 100 ms flush timer — hosts see at most ~10 snapshots/s no matter how fast deltas arrive (per-delta fan-out freezes LiveView sockets).

Run lifecycle

attach_run/2 monitors the run process: on :DOWN, every non-terminal agent is marked :failed and a retention timer starts (default 5 min) so hosts can still inspect the final snapshot before the coordinator retires with :normal.

Summary

Functions

attach_run(coordinator, run_pid)

Monitor the run process so agent statuses track its fate.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

notify(coordinator, agent_id, event)

Ingest one event for agent_id (:main or a subagent id). Fire-and-forget cast — safe to call against a dead coordinator.

snapshot(coordinator)

Current snapshot of the run's observable state.

start_for(session_id, opts \\ [])

Start (or return) the coordinator for session_id.

start_link(opts)
subscribe(coordinator, subscriber_pid)

Subscribe pid to batched {:orchestrator_update, session_id, snapshot} messages. Subscribe-then-snapshot in one call, so no update can fall in between. The subscriber is monitored and dropped on exit.

whereis(session_id)

Pid of the coordinator for session_id, or nil.

Functions

attach_run(coordinator, run_pid)

@spec attach_run(pid() | String.t(), pid()) :: :ok | {:error, :not_found}

Monitor the run process so agent statuses track its fate.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

notify(coordinator, agent_id, event)

@spec notify(pid() | String.t(), term(), term()) :: :ok

Ingest one event for agent_id (:main or a subagent id). Fire-and-forget cast — safe to call against a dead coordinator.

Besides ExAthena.Loop.Events tuples, two coordinator-internal shapes are accepted: {:todos, list} (todo-list rewrite from a todo_writer side-channel) and {:agent_meta, %{prompt:, name:, linked_todo_id:}} (enrichment emitted by SpawnAgent before the subagent runs).

snapshot(coordinator)

@spec snapshot(pid() | String.t()) :: {:ok, map()} | {:error, :not_found}

Current snapshot of the run's observable state.

start_for(session_id, opts \\ [])

@spec start_for(
  String.t(),
  keyword()
) :: {:ok, pid()} | {:error, term()}

Start (or return) the coordinator for session_id.

start_link(opts)

subscribe(coordinator, subscriber_pid)

@spec subscribe(pid() | String.t(), pid()) :: {:ok, map()} | {:error, :not_found}

Subscribe pid to batched {:orchestrator_update, session_id, snapshot} messages. Subscribe-then-snapshot in one call, so no update can fall in between. The subscriber is monitored and dropped on exit.

whereis(session_id)

@spec whereis(String.t()) :: pid() | nil

Pid of the coordinator for session_id, or nil.