An orchestrator should treat AgentHarness as an asynchronous state machine. Use subscriptions for turn progress, a native process monitor for session liveness, and supervised tasks for the few public calls that intentionally wait on external work.
Do not call await/2, enumerate stream/2, or synchronously open a CLI from a
GenServer callback. await/2 and stream/2 default to :infinity, while
session opening has a finite default but still waits for an external runtime.
Any of those waits blocks every other message handled by that GenServer.
Supervision setup
Give orchestration calls their own task supervisor:
children = [
{Task.Supervisor, name: MyApp.AgentTasks},
MyApp.AgentCoordinator
]
Supervisor.start_link(children, strategy: :one_for_one)AgentHarness supervises sessions and provider runtimes. MyApp.AgentTasks
keeps external waits out of the coordinator mailbox; it is not a pool of
fungible sessions.
A non-blocking consumer
The following worker starts the CLI in a task, then owns the session monitor and event subscription itself:
defmodule MyApp.AgentWorker do
use GenServer
alias AgentHarness.{Event, Request, Response}
def start_link(opts), do: GenServer.start_link(__MODULE__, opts)
@impl true
def init(opts) do
Process.flag(:trap_exit, true)
owner = Keyword.fetch!(opts, :owner)
prompt = Keyword.fetch!(opts, :prompt)
provider = Keyword.fetch!(opts, :provider)
session_opts = Keyword.get(opts, :session_options, [])
task =
Task.Supervisor.async(MyApp.AgentTasks, fn ->
AgentHarness.start_session(provider, session_opts)
end)
{:ok,
%{
owner: owner,
prompt: prompt,
startup_task: task,
work_task: nil,
session: nil,
session_monitor: nil,
turn: nil,
subscription: nil
}}
end
@impl true
def handle_info({ref, {:ok, session}}, %{startup_task: %Task{ref: ref} = task} = state) do
Process.demonitor(ref, [:flush])
Process.unlink(task.pid)
case AgentHarness.monitor(session) do
{:ok, session_monitor} ->
worker = self()
work_task =
Task.Supervisor.async(MyApp.AgentTasks, fn ->
start_work(session, state.prompt, worker)
end)
{:noreply,
%{
state
| startup_task: nil,
work_task: work_task,
session: session,
session_monitor: session_monitor
}}
error ->
send(state.owner, {:agent_start_failed, session, nil, error})
{:stop, :normal, %{state | startup_task: nil, session: session}}
end
end
def handle_info({ref, {:error, reason}}, %{startup_task: %Task{ref: ref} = task} = state) do
Process.demonitor(ref, [:flush])
Process.unlink(task.pid)
send(state.owner, {:agent_start_failed, nil, nil, reason})
{:stop, :normal, %{state | startup_task: nil}}
end
def handle_info(
{:DOWN, ref, :process, _task_pid, reason},
%{startup_task: %Task{ref: ref}} = state
) do
send(state.owner, {:agent_start_failed, nil, nil, {:startup_task_down, reason}})
{:stop, :normal, %{state | startup_task: nil}}
end
def handle_info(
{ref, {:ok, turn, subscription}},
%{work_task: %Task{ref: ref} = task} = state
) do
Process.demonitor(ref, [:flush])
Process.unlink(task.pid)
{:noreply,
%{state | work_task: nil, turn: turn, subscription: subscription}}
end
def handle_info(
{ref, {:error, turn, reason}},
%{work_task: %Task{ref: ref} = task} = state
) do
Process.demonitor(ref, [:flush])
Process.unlink(task.pid)
send(state.owner, {:agent_start_failed, state.session, turn, reason})
{:stop, :normal, %{state | work_task: nil, turn: turn}}
end
def handle_info(
{:DOWN, ref, :process, _task_pid, reason},
%{work_task: %Task{ref: ref}} = state
) do
send(
state.owner,
{:agent_start_failed, state.session, state.turn, {:work_task_down, reason}}
)
{:stop, :normal, %{state | work_task: nil}}
end
# Linked Tasks emit both EXIT and DOWN. Keep DOWN authoritative so a failed
# task is reported exactly once and its matching state is still available.
def handle_info({:EXIT, task_pid, _reason}, %{startup_task: %Task{pid: task_pid}} = state),
do: {:noreply, state}
def handle_info({:EXIT, task_pid, _reason}, %{work_task: %Task{pid: task_pid}} = state),
do: {:noreply, state}
def handle_info({:EXIT, _task_pid, :normal}, state), do: {:noreply, state}
def handle_info({:EXIT, _parent, :shutdown}, state), do: {:stop, :shutdown, state}
def handle_info({:EXIT, _pid, reason}, state), do: {:stop, reason, state}
def handle_info(
{:agent_harness, ref, %Event{type: :request_created, data: request}},
%{subscription: %{ref: ref}} = state
) do
response = response_for(request)
owner = state.owner
task_result =
Task.Supervisor.start_child(MyApp.AgentTasks, fn ->
result = AgentHarness.respond(request, response)
send(owner, {:agent_response_result, request.id, result})
end)
if match?({:error, _reason}, task_result) do
send(owner, {:agent_response_result, request.id, task_result})
end
{:noreply, state}
end
def handle_info(
{:agent_harness, ref, %Event{type: type} = event},
%{subscription: %{ref: ref}} = state
)
when type in [
:turn_completed,
:turn_failed,
:turn_cancelled,
:turn_interrupted
] do
send(state.owner, {:agent_terminal, state.session, state.turn, event})
Task.Supervisor.start_child(MyApp.AgentTasks, fn ->
AgentHarness.unsubscribe(state.subscription)
end)
{:noreply, %{state | subscription: nil}}
end
def handle_info(
{:agent_harness, ref, %Event{} = event},
%{subscription: %{ref: ref}} = state
) do
send(state.owner, {:agent_event, state.turn, event})
{:noreply, state}
end
def handle_info(
{:DOWN, ref, :process, _server, reason},
%{session_monitor: ref} = state
) do
send(state.owner, {:agent_session_down, state.session, reason})
{:stop, :normal, %{state | session_monitor: nil}}
end
# A demonitor/unlink race may leave the second half of a Task notification.
def handle_info({:DOWN, _ref, :process, _pid, _reason}, state), do: {:noreply, state}
defp response_for(%Request{kind: :question}) do
# Replace this with a user/UI round trip when policy cannot decide.
Response.answer("Postgres")
end
defp response_for(%Request{}) do
Response.deny("No unattended approval policy matched")
end
defp start_work(session, prompt, subscriber) do
with {:ok, turn} <- AgentHarness.start_turn(session, prompt) do
case AgentHarness.subscribe(turn, pid: subscriber, from: :start) do
{:ok, subscription} -> {:ok, turn, subscription}
error -> {:error, turn, error}
end
else
{:error, {:session_call_timeout, turn}} = error -> {:error, turn, error}
error -> {:error, nil, error}
end
end
endThis example is a one-turn event consumer. After the terminal event it removes the subscription but remains alive to own the session monitor. The owner receives any accepted turn handle along with the session and error, or receives both handles with the terminal event. It must deliberately reuse the session, stop it, or transfer monitor ownership before stopping this worker. That separation prevents an event consumer crash from silently closing a coding session.
The startup task is linked to the worker, which traps exits while distinguishing
the task from its supervisor. This matters because AgentHarness treats the
process calling start_session/2 as the startup owner until readiness is
acknowledged in both directions. If the worker dies, its linked startup task
dies too and the partially opened session is cleaned up instead of becoming an
idle orphan. A pending attempt marker also makes a Store commit reclaimable
after an untrappable SessionServer kill in that handoff window.
The important ownership detail is that AgentHarness.monitor/1 is called by
the GenServer. BEAM monitors notify the process that creates them. The worker
then delegates start_turn/3 and replay subscription to another task, passing
pid: worker so events and the SessionServer's subscriber monitor still belong
to the worker. This keeps provider admission and custom Store reads/writes out
of callbacks without moving lifecycle ownership to the task.
An uncertain admission failure is followed by SessionServer termination. Since
the worker keeps its session monitor after the terminal event, it reports the
subsequent :DOWN as a separate lifecycle signal. If the owner takes over the
session instead, it should call AgentHarness.monitor/1 itself before asking
the worker to exit; either {:error, :session_not_found} or a later :DOWN
then triggers reconciliation.
The subscription uses from: :start, so a fast provider cannot finish in the
gap between start_turn/3 and subscribe/2. Replay is installed before live
delivery, preserving event order. start_turn/3 itself returns after local
acceptance with a stable :starting turn; provider admission finishes
asynchronously.
The response call is also placed in a supervised task. Request ownership is
exactly once in the SessionServer, but the provider call can still take time.
Track {:agent_response_result, request_id, result} so rejected, expired, or
uncertain responses are visible to the orchestrator.
An in-flight duplicate response returns :response_in_progress; a later one
returns :already_resolved. A definite provider rejection leaves the request
pending, while {:provider_command_uncertain, reason} is followed by terminal
turn events and session :DOWN. Reconcile before retrying an uncertain answer.
Questions that require a human
Do not keep the GenServer callback waiting while a person decides. Store the
%Request{} by its ID, notify the UI or parent process, and return
{:noreply, state}. Later, a cast or info message can launch the same
asynchronous AgentHarness.respond/2 operation.
Pending requests expire when their turn ends. Treat :request_expired as the
signal to remove the request from UI state. A late response then returns an
error instead of being forwarded twice.
Completion and failure signals
These signals mean different things:
:turn_completed,:turn_failed,:turn_cancelled, and:turn_interruptedare authoritative turn outcomes.{:DOWN, session_monitor, :process, server, reason}means the session process disappeared. It does not prove whether already-started tools or filesystem changes ran.:transport_errorsays the provider runtime was lost; an active turn is also failed and the session becomes unavailable.- A timeout from a calling task is not evidence that a coding action did not start. Reconcile with the stable IDs, events, Store, and workspace before retrying.
AgentHarness.stop_session/2 and unsubscribe/1 are idempotent. Live-session
operations on a stopped handle return {:error, :session_not_found};
purge_session/2 intentionally operates only after the session is no longer
live.
Restart reconciliation
Stable IDs let a coordinator process that restarts inside a still-running application compare live ownership with history retained by the Store:
live_by_id = Map.new(AgentHarness.list_sessions(), &{&1.id, &1})
{:ok, stored} = AgentHarness.list_stored_sessions()
stale =
Enum.filter(stored, fn entry ->
not entry.live? and entry.snapshot.status not in [:closed]
end)The default Memory Store survives an individual coordinator restart, but it is lost when the AgentHarness application or VM stops. Reconciliation across a redeploy, application restart, or machine failure requires a durable custom Store. Pass the same Store owner used by the sessions when reading inventory:
store = {MyApp.AgentStore, MyApp.AgentStore}
{:ok, stored} = AgentHarness.list_stored_sessions(store: store)Use that same store option when starting, replacing, or purging those session
aggregates.
For each stale aggregate, inspect its last event, provider session ID, pending request records, and workspace before deciding what happened. AgentHarness does not automatically retry a turn whose side effects are uncertain.
The explicit recovery choices are:
- keep the aggregate for audit and start a differently named session;
- resume the provider conversation using its saved provider session/thread ID;
purge_session/2after determining the history is no longer needed;- start with
reuse: :closedfor a cleanly closed ID; - start with
reuse: :replaceto destructively replace a stale non-live ID.
Both reuse modes preserve the old aggregate until the replacement provider has opened successfully, then begin a fresh core journal. They do not reconstruct the old SessionServer state.
Use deterministic session IDs derived from work ownership when that helps a coordinator find duplicates after restart. Do not start the same deterministic ID independently on multiple BEAM nodes; v0.x inventory and Registry ownership are node-local.