ExAthena.Web.RunServer (ExAthena v0.17.0)

Copy Markdown View Source

Per-session owner of one in-flight agent run, decoupled from the LiveView.

The web run used to be a bare Task that sent events straight to the LiveView pid captured when the run started. A websocket reconnect spawns a new LiveView process (new pid), so the in-flight run kept sending to a dead process — the user saw streaming freeze and had to reopen the session.

RunServer fixes that by owning the run itself. It is named by the stable session id ({:via, Registry, ...}), survives LiveView reconnects, and:

  • owns the run Task and accumulates just enough live state to re-render a mid-run view (stream_text, current_action, a paused ask_user question, the streaming? flag);
  • lets any number of LiveViews attach/2 — subscribe-then-snapshot in one call, then receive the same {:athena, _} / {:athena_done, _} / {:athena_error, _} / {:athena_ask_user, _} messages the LiveView's existing handle_info clauses already expect;
  • routes a user's ask_user answer back into the blocked run via the stable channel (answer/3), so the reply survives a reconnect;
  • durably persists the final answer (Sessions.persist_run_result/3) the moment the run finishes — works even with zero LiveViews attached.

Supervision & naming

Started on demand under ExAthena.Web.RunSupervisor (a DynamicSupervisor) with restart: :temporary, named {:via, Registry, {ExAthena.Web.RunRegistry, session_id}}. Both the registry and supervisor are started by the mix athena.web task (web-only).

Mirrors the ExAthena.Orchestrator.Coordinator pattern; subscribe/snapshot fan-out is plain send/2 (host-agnostic — no Phoenix.PubSub dependency).

Summary

Functions

Route a user's ask_user answer back into the blocked run.

Attach pid to the live run for session_id. Subscribe-then-snapshot in one call (no event can fall between). Returns {:ok, snapshot} only while the run is still streaming; once finished there is nothing live to attach to and the caller should load the (durably persisted) session instead.

Returns a specification to start this module under a supervisor.

Whether a run for session_id is currently in flight (still streaming).

Start a run for session_id. Any previous (finished, lingering) server for the same session is replaced. spec carries

Stop the in-flight run for session_id (user hit stop).

Pid of the run server for session_id, or nil.

Functions

answer(session_id, tool_call_id, answer)

@spec answer(String.t(), String.t(), String.t()) :: :ok

Route a user's ask_user answer back into the blocked run.

attach(session_id, pid)

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

Attach pid to the live run for session_id. Subscribe-then-snapshot in one call (no event can fall between). Returns {:ok, snapshot} only while the run is still streaming; once finished there is nothing live to attach to and the caller should load the (durably persisted) session instead.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

running?(session_id)

@spec running?(String.t()) :: boolean()

Whether a run for session_id is currently in flight (still streaming).

start_link(spec)

start_run(session_id, spec)

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

Start a run for session_id. Any previous (finished, lingering) server for the same session is replaced. spec carries:

  • :run_opts — keyword list passed to ExAthena.run/2, without the pid-bound keys (:on_event, :assigns, :coordinator) — RunServer injects those so they target the stable server, not the LiveView.
  • :assistant_msg_id — id of the assistant turn (for durable persistence).
  • :run_sid — the orchestration/Overview scope id (echoed in snapshots).
  • :coordinator — optional Coordinator pid for the Overview tab.

stop_run(session_id)

@spec stop_run(String.t()) :: :ok

Stop the in-flight run for session_id (user hit stop).

whereis(session_id)

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

Pid of the run server for session_id, or nil.