Streaming

Copy Markdown View Source

Every turn can publish events as it runs. Use streaming when a UI needs partial assistant text, progress events, or a live activity feed.

Use This When

  • a UI needs to render partial assistant text as it is produced (chat UIs, CLIs, agent dashboards).
  • an operator console wants live progress events.
  • do not use streaming as durable telemetry. Live events are best-effort and per-request. For post-hoc inspection use Tracing And Events.

Prerequisites

  • A turn or session you can run with a live provider.
  • A consumer process (the caller itself, a GenServer, or a LiveView) ready to receive mailbox messages.
mix deps.get
mix test

Stream A Chat

For UI work, start the chat asynchronously and request streaming events.

{:ok, request} =
  Jidoka.chat_async(MyApp.SupportAgent, "Hi", stream: true)

stream = Jidoka.stream(request)

for event <- stream do
  if text = Jidoka.Stream.text_delta(event) do
    IO.write(text)
  end
end

{:ok, text} = Jidoka.await(request)

The stream yields Jidoka.Event values and stops when the turn finishes, fails, or hibernates.

Cancel an active request through its handle:

{:ok, %Jidoka.Cancellation{} = cancellation} =
  Jidoka.cancel(request, grace_ms: 100)

{:cancelled, ^cancellation} = Jidoka.await(request)

Cancellation first asks the runtime and active capabilities to stop. If they do not stop within :grace_ms, Jidoka stops the request process tree. The stream receives one terminal :turn_failed event with data.reason == :cancelled. It cannot later receive :turn_finished.

Concepts

Streaming is a per-request side channel. The runtime stays terminal-result oriented.

╭───────────────────────╮     ╭──────────────────────╮
│ Jidoka.turn(stream_to:│────▶│  Jidoka.Stream.emit/2│
│   pid, on_event: fun) │     ╰──────┬───────────────╯
╰───────────────────────╯            │
                                     ▼
                         ╭───────────────────────────╮
                         │  Caller mailbox:          │
                         │  {:jidoka_turn_event,     │
                         │   %Jidoka.Event{}}        │
                         ╰──────┬────────────────────╯
                                │
                                ▼
                         ╭───────────────────────────╮
                         │ Receive loop until        │
                         │ Stream.terminal?(event)   │
                         ╰───────────────────────────╯

Key facts:

How To

Step 1: Stream To A Pid

Pass the consumer process as :stream_to. The caller is the simplest consumer.

{:ok, _result} =
  Jidoka.turn(MyApp.SupportAgent, "Hello",
    stream_to: self()
  )

Inside a GenServer, set stream_to: self() from the handler that issued the turn; the events arrive in handle_info/2.

Step 2: Use The Receive Loop

The terminal-event contract makes the loop trivial.

def collect_events(request_id) do
  tag = Jidoka.Stream.message_tag()

  Stream.repeatedly(fn ->
    receive do
      {^tag, %Jidoka.Event{request_id: ^request_id} = event} -> event
    after
      5_000 -> :timeout
    end
  end)
  |> Enum.reduce_while([], fn
    :timeout, acc -> {:halt, Enum.reverse(acc)}
    event, acc -> if Jidoka.Stream.terminal?(event), do: {:halt, Enum.reverse([event | acc])}, else: {:cont, [event | acc]}
  end)
end

Always filter on request_id. The mailbox tag is shared across turns, and a parallel turn from the same caller will interleave events.

Step 3: Render Token Deltas

For chat UIs, the interesting event is :llm_delta. Use Jidoka.Stream.text_delta/1 to grab the content text.

def handle_info({:jidoka_turn_event, %Jidoka.Event{} = event}, state) do
  case Jidoka.Stream.text_delta(event) do
    text when is_binary(text) ->
      {:noreply, append_delta(state, text)}

    nil ->
      {:noreply, state}
  end
end

Jidoka.Stream.thinking_delta/1 is the matching helper for reasoning channels. Capabilities that emit :llm_delta directly should call Jidoka.Stream.emit/2 from inside the LLM function.

Step 4: Use An on_event Callback

:on_event is a 1-arity function. It runs inline before the next event is emitted; raised or thrown values are swallowed so the turn keeps running.

{:ok, _result} =
  Jidoka.turn(MyApp.SupportAgent, "Hello",
    on_event: fn event ->
      :telemetry.execute(
        [:my_app, :agent, :event],
        %{seq: event.seq},
        %{event: event.event, agent_id: event.agent_id}
      )
    end
  )

:stream_to and :on_event can be used together. The callback fires first; the mailbox delivery happens immediately after.

Step 5: Build A Lazy Enumerable

For consumers that prefer Enum.reduce/3, the Jidoka.Stream.events/2 helper wraps the receive loop.

Task.async(fn ->
  Jidoka.turn(MyApp.SupportAgent, "Hello",
    request_id: "req_demo",
    stream_to: self()
  )
end)

events =
  "req_demo"
  |> Jidoka.Stream.events(stream_event_timeout_ms: 5_000)
  |> Enum.to_list()

Enum.map(events, & &1.event)

The enumerable halts when it sees a terminal event for the request or when the per-event timeout fires.

Step 6: Stream Through A Session

Sessions accept the same options.

{:ok, _session, _text} =
  Jidoka.Session.chat(session_id, "Hi",
    store: store,
    stream_to: self(),
    on_event: &MyApp.Audit.publish/1
  )

The terminal events still apply: :turn_hibernated ends the stream for that turn even though the session may continue later.

Step 7: Cancel A Session Request

Start the session turn asynchronously, then cancel the returned handle.

{:ok, request} =
  Jidoka.Session.chat_async(session_id, "Prepare report",
    store: store,
    stream: true
  )

{:ok, cancellation} = Jidoka.Session.cancel(request)
{:cancelled, ^cancellation} = Jidoka.Session.await(request)

A persisted active session changes to :cancelled. The session stores the typed cancellation value in its error field. Cancellation does not roll back an external operation that already completed. Operations must use their idempotency keys and reconciliation policy for that case.

Common Patterns

  • Always filter on request_id. Concurrent turns share the mailbox tag. Filtering keeps consumers correct under load.
  • Render deltas, log lifecycle. UIs typically only care about :llm_delta; observability layers care about :turn_started, :turn_finished, :turn_failed.
  • Treat :turn_hibernated as a stream terminator. It is not an error; it is a normal pause. Resume separately.
  • Use on_event: for in-process side effects. Mailbox delivery is best for cross-process consumers; inline callbacks are best for telemetry inside the same process.
  • Do not lean on streaming for state. Use Turn.Result for the final truth; use streaming for UX.
  • Treat cancellation as a terminal result. Match on {:cancelled, %Jidoka.Cancellation{}}. Do not convert it to success.

Testing

The runtime's own tests use assert_receive against the mailbox tag.

test "stream_to publishes lifecycle events" do
  llm = fn _intent, _journal, _ctx ->
    {:ok, %{type: :final, content: "stream ok"}}
  end

  request = Jidoka.Turn.Request.new!(input: "Hello", request_id: "req_x")

  assert {:ok, %Jidoka.Turn.Result{content: "stream ok"}} =
           Jidoka.turn(MyApp.SupportAgent, request, llm: llm, stream_to: self())

  tag = Jidoka.Stream.message_tag()
  assert_receive {^tag, %Jidoka.Event{event: :turn_started, request_id: "req_x"}}
  assert_receive {^tag, %Jidoka.Event{event: :prompt_assembled, request_id: "req_x"}}
  assert_receive {^tag, %Jidoka.Event{event: :turn_finished, request_id: "req_x"} = terminal}

  assert Jidoka.Stream.terminal?(terminal)
end

For UI tests, prefer the events/2 enumerable so the test runs deterministically without polling.

Troubleshooting

SymptomLikely CauseFix
Mailbox receives no events:stream_to was not passed to the turn.Pass stream_to: pid (or {:pid, pid}) on the call.
Wrong events arriveConcurrent turns share the mailbox tag.Match on request_id in the receive clause.
:llm_delta events missingCapability did not call Jidoka.Stream.emit/2.Emit :llm_delta from the LLM function or use a streaming provider.
on_event: callback errors disappearJidoka ignores callback failures so the turn can finish.Log inside the callback before raising.
Loop never terminatesConsumer never saw a terminal event.Always check Jidoka.Stream.terminal?/1 and add an after timeout.
Jidoka.Stream.events/2 halts immediatelyDefault :stream_event_timeout_ms elapsed before the turn ran.Increase the timeout, or start the turn before subscribing.
Cancel returns :request_already_finishedA terminal event won the race before cancellation.Use the completed result. Jidoka never replaces it with cancellation.

Reference

Key modules touched in this guide: