Raxol.AgentClientProtocol.Connection (Raxol Agent Client Protocol v0.1.0)

Copy Markdown View Source

ONE GenServer per peer link, either role — the correlation + dispatch brain of ACP.

The same module runs as :agent or :client; role only selects which MethodTable direction is inbound (side = role). The Connection:

  • never blocks — no GenServer.call to itself, no synchronous wait on tasks, no transport call that waits on peer progress;
  • never runs handler code in its own process — every inbound request/notification is dispatched to a Task.Supervisor.async_nolink task;
  • never creates an atom from wire input — all method→atom mapping is Router/MethodTable compile-time clauses.

Correctness spine

  • Outbound: async_request/6 is THE primitive; request/4 is a caller-parking wrapper on the same pending_out table. The Connection is the single timeout authority (Process.send_after) — no caller arms a parallel timer. Every pending_out entry leaves through exactly one of: correlated response, timeout, cancel_request/2, or reject_all_outgoing. No fifth path.
  • Inbound: dispatch via async_nolink; crash ⇒ -32603 with the id echoed at its exact wire type; notification crash ⇒ log only, never a frame. $/cancel_request and session/cancel are Connection-intercepted BEFORE Router.decode/4 (MethodTable rows carry the :protocol / :session_control layer tags).
  • Delegated reply: delegate_reply/3 + reply/3 + :deferred let the Session own WHEN the session/prompt response is sent while keeping updates and the response on ONE FIFO lane.
  • Response-count invariant: at most one response frame per inbound id; exactly one unless the peer $/cancel_request'd it first (then zero).

Implementation notes

  1. A pending outbound request's response is decoded through the Router.result_marker/1 marker stored at send time + the result module's from_json/1.
  2. Router.decode/4 returns the raw schema-decode error tuple; any non-:method_not_found decode error is mapped to -32602.
  3. The capability gate is located BEFORE decode with the correct -32601-beats--32602 ordering. capability_denied?/2 delegates to Capabilities.negotiated?/2, which fails closed on any missing/nil/false capability. The caps SNAPSHOT is the RECEIVER's OWN capability tree, not the whole initialize message. The agent snapshots agent_capabilities from the response it sends; the client snapshots client_capabilities from the request it sends (stashed in own_caps at outbound-submit time, committed at handshake), because the client's own caps are not present in the initialize RESPONSE it receives. See Capabilities + the MethodTable capability typedoc (ORACLE-DIVERGENCE note).
  4. transport_ref and the monitored carrier pid are both taken from the handle's :pid field (the Transport behaviour exposes no ref accessor; Transport.Paired and Transport.Stdio expose :pid and tag their inbound frames with that same pid).
  5. Telemetry events are emitted only when :telemetry is loaded (it is not a package dependency); otherwise Logger carries the signal.
  6. Sibling task_sup/session_sup may be injected directly in start_link/1 opts (test convenience) instead of parent_sup; with parent_sup, handle_continue/2 resolves them from which_children by child module/type (the Task.Supervisor child is task_sup, the other :supervisor child is session_sup).

Summary

Types

outcome()

Outcome delivered for an accepted outbound submission.

role()

Functions

async_request(conn, method, params, owner, tag, timeout_ms)

PRIMITIVE outbound submit. Non-blocking: returns :ok once the frame is accepted by the transport, and later delivers exactly one {:acp_result, tag, outcome} to owner — unless owner consumed the entry first via cancel_request/2 (then zero). Connection owns the timeout.

cancel_request(conn, tag)

Cancel a pending async_request by its {owner, tag}. Pops the entry, cancels its timer, best-effort $/cancel_request, delivers nothing. Unknown tag ⇒ :ok.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

close(conn)

Graceful close: reject outgoing, kill undelegated inbound tasks, close transport, stop :normal.

delegate_reply(conn, reply_ref, adopter)

Transfer an inbound request's reply obligation to adopter. MUST return before the handler returns :deferred. Connection monitors the adopter.

notify(conn, method, params)

Fire-and-forget notification: a GenServer.call for per-sender FIFO + backpressure; touches no pending table, arms no timer.

reply(conn, reply_ref, result_or_error)

Discharge a delegated reply obligation. Idempotent-safe: a consumed/unknown/cancelled reply_ref is a suppressed no-op ([:raxol, :acp, :dup_reply]) and still returns :ok.

request(conn, method, params, timeout_ms)

WRAPPER outbound request: parks the caller on pending_out and returns the outcome. Outer GenServer.call timeout is :infinity by design — the protocol timeout is the Connection's internal timer, the ONE authority.

start_link(opts)

Start a Connection. Opts: :role, :transport {mod, handle}, :handler, :handler_arg, :max_inbound (concurrent inbound-request cap, default 1000), and either :parent_sup or direct :task_sup/:session_sup.

update_watermark(conn, session_id)

Race-free base-ordinal accessor for a session's delivery stream. Because this is a GenServer.call (serialized with the demux point that assigns ordinals), a caller that subscribes to the broadcast path THEN calls this learns an exact base: every stamp assigned after this call returns is guaranteed to reach a live subscriber, and every stamp <= ordinal is pre-subscription. turn: nil means the session currently has no open session/prompt turn (the out-of-turn namespace applies). The primary direct-delivery path (Client.prompt/3, prompt_stream/4) does not need it at all.

Types

outcome()

@type outcome() ::
  {:ok, struct()}
  | {:ok, {:ext, map()}}
  | {:error,
     Raxol.AgentClientProtocol.Error.t()
     | :timeout
     | :connection_closed
     | {:result_decode, term()}}

Outcome delivered for an accepted outbound submission.

role()

@type role() :: :agent | :client

Functions

async_request(conn, method, params, owner, tag, timeout_ms)

@spec async_request(
  pid(),
  String.t(),
  struct() | map() | nil,
  pid(),
  term(),
  pos_integer() | :infinity
) ::
  :ok | {:error, :not_initialized | :connection_closed | {:transport, term()}}

PRIMITIVE outbound submit. Non-blocking: returns :ok once the frame is accepted by the transport, and later delivers exactly one {:acp_result, tag, outcome} to owner — unless owner consumed the entry first via cancel_request/2 (then zero). Connection owns the timeout.

cancel_request(conn, tag)

@spec cancel_request(pid(), term()) :: :ok

Cancel a pending async_request by its {owner, tag}. Pops the entry, cancels its timer, best-effort $/cancel_request, delivers nothing. Unknown tag ⇒ :ok.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

close(conn)

@spec close(pid()) :: :ok

Graceful close: reject outgoing, kill undelegated inbound tasks, close transport, stop :normal.

delegate_reply(conn, reply_ref, adopter)

@spec delegate_reply(pid(), reference(), pid()) :: :ok | {:error, :unknown_ref}

Transfer an inbound request's reply obligation to adopter. MUST return before the handler returns :deferred. Connection monitors the adopter.

notify(conn, method, params)

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

Fire-and-forget notification: a GenServer.call for per-sender FIFO + backpressure; touches no pending table, arms no timer.

reply(conn, reply_ref, result_or_error)

@spec reply(
  pid(),
  reference(),
  {:ok, struct()} | {:error, Raxol.AgentClientProtocol.Error.t()}
) :: :ok

Discharge a delegated reply obligation. Idempotent-safe: a consumed/unknown/cancelled reply_ref is a suppressed no-op ([:raxol, :acp, :dup_reply]) and still returns :ok.

request(conn, method, params, timeout_ms)

@spec request(pid(), String.t(), struct() | map() | nil, pos_integer()) ::
  {:ok, struct()} | {:ok, {:ext, map()}} | {:error, term()}

WRAPPER outbound request: parks the caller on pending_out and returns the outcome. Outer GenServer.call timeout is :infinity by design — the protocol timeout is the Connection's internal timer, the ONE authority.

start_link(opts)

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

Start a Connection. Opts: :role, :transport {mod, handle}, :handler, :handler_arg, :max_inbound (concurrent inbound-request cap, default 1000), and either :parent_sup or direct :task_sup/:session_sup.

update_watermark(conn, session_id)

@spec update_watermark(pid(), String.t()) ::
  {:ok, %{turn: reference() | nil, ordinal: non_neg_integer()}}

Race-free base-ordinal accessor for a session's delivery stream. Because this is a GenServer.call (serialized with the demux point that assigns ordinals), a caller that subscribes to the broadcast path THEN calls this learns an exact base: every stamp assigned after this call returns is guaranteed to reach a live subscriber, and every stamp <= ordinal is pre-subscription. turn: nil means the session currently has no open session/prompt turn (the out-of-turn namespace applies). The primary direct-delivery path (Client.prompt/3, prompt_stream/4) does not need it at all.