Stream events to a live UI

You have an incident-response agent that investigates an alert, checks a few dashboards, and narrates what it finds. You want that narration on a live status page as it happens. The pattern is a small sink that receives every Mistri::Event and translates the ones the page cares about into your own frames.

The sink

A sink is any object with call(event). It pattern-matches on event.type and forwards a frame to your transport. There is no else: an event type the page does not render falls through and is ignored, so a new event type in a later mistri release is a no-op, never a crash.

class StatusPageSink
  def initialize(incident_id)
    @incident_id = incident_id
  end

  def call(event)
    case event.type
    when :text_start        then push(:update_started)
    when :text_delta        then push(:narration, text: event.delta)
    when :text_end          then push(:update_ended)
    when :thinking_delta    then push(:reasoning, text: event.delta)
    when :tool_result       then push(:action, tool: event.tool_call.name,
                                               seconds: event.duration,
                                               card: event.message&.ui)
    when :approval_needed   then push(:blocked, tool: event.tool_call.name)
    when :retry             then push(:reconnecting, attempt: event.attempt,
                                                     of: event.max_attempts,
                                                     retry_in: event.delay)
    when :done              then push(:resolved)
    when :error             then push(:failed, reason: event.error_message)
    end
  end

  def to_proc = method(:call).to_proc

  private

  # Your transport: an Action Cable broadcast, an SSE frame, a WebSocket send.
  def push(kind, **payload)
    StatusPage.broadcast(@incident_id, { kind: kind, **payload })
  end
end

Wiring it up

Wrap the sink in Coalesced so bursts of token deltas merge to a readable pace instead of one frame per token, and pass it as the run block:

sink = Mistri::Sinks::Coalesced.new(StatusPageSink.new(incident.id))

agent.run("Investigate the checkout latency alert and post what you find.", &sink)

Coalesced merges only the delta events (text_delta, thinking_delta, toolcall_delta) per content block. Every other event flushes the buffered text first and passes straight through, so a tool result never jumps ahead of the sentence that introduced it, and a turn always ends flushed.

What each branch is for

• Text is the narration. :text_start and :text_end bracket one message so the page can open and close an update bubble; the :text_delta between them carries the merged text in event.delta.
• :tool_result is an action the agent took. event.duration is its run time in seconds, and event.message&.ui is the tool’s UI channel: a compact card the model never paid to read, ideal for showing the dashboard snapshot behind a one-line summary.
• :approval_needed means a gated tool parked for a human. Show a “waiting on an on-call approval” state; the run has already returned. See Human approval.
• :retry fires before mistri waits out a transient provider failure. event.attempt, event.max_attempts, and event.delay are everything a “reconnecting, attempt 2 of 3” affordance needs. Clear it on the next narration frame.
• :done and :error are the only terminal events, and mistri holds them behind retries, so an :error that reaches the sink is final rather than a blip it will walk back.

Sub-agent lanes

If the agent fans work out to sub-agents, their events arrive tagged: event.origin is a label like log-scout#a41f. Key your frames by it and the page can render each worker in its own lane while the parent narrates the synthesis.

Every sink ships the same way. Mistri::Sinks::SSE.new(io) writes the events straight to an ActionController::Live stream, and Mistri::Sinks::ActionCable.new(channel) broadcasts them, both without a translating class when the default event.to_h shape is enough for your front end.