SuperPi.Event.Renderer.AGUI (super_pi v0.1.0)

Copy Markdown View Source

Renders events as AG-UI events, for SuperPi.Gateway.AGUI.

Each output item is a map with a "type" (RUN_STARTED, TEXT_MESSAGE_CONTENT, ...) and the fields of that event in camelCase, ready for Jason.encode!/1 and an SSE data: line.

Mapping

EventAG-UI
turn_startedRUN_STARTED
text_deltaTEXT_MESSAGE_START (once per message), TEXT_MESSAGE_CONTENT
thinking_deltaREASONING_START, REASONING_MESSAGE_START, REASONING_MESSAGE_CONTENT
message_completedTEXT_MESSAGE_END (and the reasoning end events)
toolcall_deltaTOOL_CALL_START (once per call), TOOL_CALL_ARGS
tool_startedTOOL_CALL_END - AG-UI means arguments complete by it, not tool complete
tool_completed, tool_blockedTOOL_CALL_RESULT
turn_completedRUN_FINISHED
turn_errorRUN_ERROR
turn_cancelledRUN_ERROR with code "cancelled" (AG-UI has no cancel event)
compaction_completedMESSAGES_SNAPSHOT, because the context really changed
state_changed, mode_changed, usage_updated, plan_updated, permission_request, ui_resolvedSTATE_DELTA
permission_request, ui_request, ui_resolved, notice, stream_reset, subagent_*CUSTOM

A bare run of deltas gets its TEXT_MESSAGE_START and TEXT_MESSAGE_END here, because the machine only emits deltas. A tool call that a provider did not stream gets TOOL_CALL_START, TOOL_CALL_ARGS and TOOL_CALL_END together when tool_started arrives.

State

The renderer keeps a small projection of the conversation (status, mode, model, usage, plan, pendingPermission). snapshot/1 gives STATE_SNAPSHOT for an opening or a reconnect. After a state event it recomputes the projection and emits STATE_DELTA with Jsonpatch.diff(previous, next). The durable document layer offers no patch stream for this: only pi.agent keeps ops, pi.live and pi.usage do not, and no write broadcasts. Doing it here is correct by construction and needs no change to document retention.

Options

  • :thread_id, :run_id - ids of the run. run_id defaults to the turn id.
  • :projection - initial projection (initial_projection/1 builds one from the documents).
  • :messages_fun - zero-arity function that returns the messages for MESSAGES_SNAPSHOT.

Coalescing

coalesce/1 merges consecutive TEXT_MESSAGE_CONTENT and TOOL_CALL_ARGS items of the same message or call. The transport calls it on the batch of events that arrived within about 16 ms. There is no ack window as in the RPC server: SSE is one-way and TCP is the backpressure.

Summary

Functions

coalesce(items)

Merges consecutive content items of the same message or tool call into one.

initial_projection(conversation_id)

The projection of a conversation, read from its documents.

snapshot(st)

STATE_SNAPSHOT of the current projection. Marks it as sent, so the next delta is relative to it.

Functions

coalesce(items)

@spec coalesce([map()]) :: [map()]

Merges consecutive content items of the same message or tool call into one.

initial_projection(conversation_id)

@spec initial_projection(integer()) :: map()

The projection of a conversation, read from its documents.

snapshot(st)

@spec snapshot(%SuperPi.Event.Renderer.AGUI{
  call_order: term(),
  calls: term(),
  message: term(),
  messages_fun: term(),
  projection: term(),
  reasoning: term(),
  run_id: term(),
  sent: term(),
  seq: term(),
  thread_id: term()
}) ::
  {[map()],
   %SuperPi.Event.Renderer.AGUI{
     call_order: term(),
     calls: term(),
     message: term(),
     messages_fun: term(),
     projection: term(),
     reasoning: term(),
     run_id: term(),
     sent: term(),
     seq: term(),
     thread_id: term()
   }}

STATE_SNAPSHOT of the current projection. Marks it as sent, so the next delta is relative to it.