SuperPi.UI.Broker (super_pi v0.1.0)

Copy Markdown View Source

Carries a question from a running turn to a client that can answer it, and the answer back. Works across nodes: the machine that asks and the client that answers can be on different nodes.

A tool (ask_question) calls ask/3. The broker publishes a ui_request event on the conversation topic and blocks the caller until a client calls respond/2, the conversation is cancelled (cancel_pending/1), every responder has left, or the timeout passes.

ask/4 can publish a permission_request event instead of a ui_request (event: :permission_request). Either way, ui_resolved follows when the question ends, so a second watcher can dismiss a prompt that someone else answered, or that was cancelled or timed out.

Only a client that has registered with register_responder/1 can answer. The CLI chat and an RPC client that watches a conversation register. The HTTP gateway cannot prompt a user, so with no responder ask/3 returns {:error, :no_interactive_client} at once instead of waiting.

Membership lives in :pg process groups (scope :super_pi_pg). :pg spans the connected nodes and drops a process that exits, so no table needs cleaning.

Pending questions are in memory. If the node that asks stops mid-question, the task engine restarts the turn elsewhere and the model asks again. Nothing is answered by default.

Summary

Functions

Asks request (a map such as %{"kind" => "questionnaire", "questions" => [...]}) and waits.

Ends every question that waits on the conversation with {:error, :cancelled}.

The conversation a waiting question belongs to, or nil.

Registers the calling process as able to answer questions for conversation_id.

Delivers an answer. {:error, :unknown_request} when no question is waiting under that id.

True when at least one live process, on any node, can answer for the conversation.

Functions

ask(conversation_id, request, timeout, opts \\ [])

Asks request (a map such as %{"kind" => "questionnaire", "questions" => [...]}) and waits.

Returns {:ok, value}, {:error, :no_interactive_client}, {:error, :cancelled} or {:error, :timeout}.

Options: event: :permission_request publishes a permission_request event (the request map holds its fields, see SuperPi.Event.permission_request/3); turn_id: sets the turn id of the event.

cancel_pending(conversation_id)

Ends every question that waits on the conversation with {:error, :cancelled}.

conversation_of(request_id)

The conversation a waiting question belongs to, or nil.

register_responder(conversation_id)

Registers the calling process as able to answer questions for conversation_id.

respond(request_id, value)

Delivers an answer. {:error, :unknown_request} when no question is waiting under that id.

responder?(conversation_id)

True when at least one live process, on any node, can answer for the conversation.