Streaming Events

Watch an agent’s turn as it happens: the answer token by token, each tool call, each model call.

Table of contents

  1. Streaming a turn
  2. The events
  3. Durable vs live, and the content rule
  4. Attempts
  5. Subagents’ events
  6. stream* vs subscribe
  7. Falling behind
  8. Replay
    1. Tenants
  9. Subscriptions across runtimes
    1. Ordering across runtimes
  10. fs2, ZIO and Pekko Streams
    1. Typed streams of one run
  11. Java and Kotlin
  12. Your own graphs
    1. Typed handlers
  13. Limits

Streaming a turn

AgentBuilder.withStreaming() makes every model call of the agent stream its answer (the default is off: the agent calls complete). agent.stream(threadId, query)(listener) runs a turn and hands every event of it to listener, which receives the runtime’s StreamEvent. Match the agent’s events with the extractors in AgentEvents; they need no ujson:

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import org.llm4s.agent.Agent
import org.llm4s.agent.events.AgentEvents
import org.llm4s.agent.graph.ThreadId

for
  agent <- Agent.builder("assistant", client).withSystemPrompt("You are concise.").withStreaming().build()
  run <- agent.stream(ThreadId("t1"), "Explain monads in three sentences.") {
    case AgentEvents.TextDelta(d)          => print(d.text)
    case AgentEvents.ToolCallStarted(c)    => println(s"\n[${c.tool}]")
    case AgentEvents.ModelCallCompleted(m) => println(s"\n(${m.usage})")
    case _                                 => ()
  }
  result <- run.await()
yield result

stream returns the AgentRun at once; await() gives the AgentResult as for run, and returns only once the listener has returned from the run’s last event - so everything the listener collected is there when await returns. It waits at most 5 seconds for a listener that is still busy, then logs a WARN and returns. streamResume(threadId, answers) and streamRecover(threadId) are the streaming siblings of startResume and startRecover. A listener runs on its own dispatcher thread, so a slow listener never slows the run. Runnable version: StreamingAgentExample in modules/samples.

The events

Every event is an EventType in org.llm4s.agent.events.AgentEvents, named agent.<snake_case>, version 1.

Durable events are committed with the task that emits them, so they are stored, replayed, and never duplicated by a retry or a recover. They carry no message content.

Event Fields Sent
ModelCallCompleted agent, model, attempts, toolCalls, usage, estimatedCost after each provider call returns a completion - every attempt a retrying wrapper makes, numbered in attempts, and (attempts 0) each call a hook makes through ModelContext.accounted
ToolExecuted agent, toolCallId, tool, duration, outcome (Succeeded, Errored, Denied, Rejected, NeedsApproval, Asked) when a tool call’s outcome is recorded
HandedOff from, to when the model routes a handoff
GuardrailBlocked guardrail, phase (Input, Output) when a guardrail blocks a turn

Live events are for watching. They carry content, have no seq, and are never stored or replayed.

Event Fields Sent
ModelCallStarted agent, attempt before each attempt of a model call
TextDelta attempt, text per chunk of the answer (needs withStreaming())
ThinkingDelta attempt, text per chunk of the model’s reasoning (needs withStreaming())
ToolCallStarted toolCallId, tool, arguments after the arguments validate, before middleware runs
ToolCallResult toolCallId, content, isError when the call’s result is recorded

Notes on the durable events:

  • attempts is 0 when a model middleware answered without calling the model.
  • ToolExecuted.tool names the tool only when the agent has it; a call to a name the model invented is recorded as "<unknown>" (the live ToolCallResult still carries the call). Each non-handoff call of a batch that mixes a handoff with other calls runs nothing and is reported as Errored.
  • A tool call that is approved or edited produces two agent.tool_executed events for one call id, across two runs: NeedsApproval when it suspends, then the final outcome after resume.
  • agent.guardrail_blocked fires only for a guardrail’s block. Other middleware refusals (a blank query or answer, a custom Left) emit none.
  • The kernel’s own events (RunStarted, RunCompleted, RunFailed, …) arrive in the same stream, as StreamEvent.Durable.

Durable vs live, and the content rule

Durable events hold identifiers, names, usage, durations and outcomes, never message content: no assistant text, thinking, tool arguments, tool results or guardrail reasons. Content travels only in live events, and in AgentRunEnded.messages at the end of a traced run.

The reason is the guardrail Block guarantee. When an output guardrail blocks a turn, the runtime removes the blocked turn from the thread. If the log held the content, a Block would also have to redact the log. Replaying a run therefore shows its structure (which model calls, which tools, how long, with what usage), and a blocked answer is in no stored event.

Attempts

attempt counts the calls of the innermost model function within one task, from 1. A model wrapper that retries or falls back (wrapModelCall) calls it again, so a higher attempt on the same task means the text streamed so far for that task is discarded: clear what you have shown and start again. A task that recover runs again starts at attempt 1 under a new task id.

Subagents’ events

An agent with subagents sends SubagentEvents.TaskStarted (live) as each child starts and SubagentEvents.TaskEnded (durable, no content) when its task ends. With nested = true on stream* or AgentRun.subscribe, the listener also receives each child’s events as StreamEvent.Nested(child, event), matched with EventHandlers.onChild; see Delegation: following children.

stream* vs subscribe

  • Agent.stream, streamResume and streamRecover subscribe your listener during admission, before the run thread starts, so it sees every durable and every live event of the run, from the claim on.
  • AgentRun.subscribe(capacity = 1024)(listener) attaches later, to a run you started with start. It replays the run’s durable events from its start, then delivers new ones, but it misses live events sent before it attached. Live events sent after it returns are not lost while it replays: they are held and delivered in their place among the durable events. Up to capacity are held; beyond that they are dropped and counted, as for a full queue, and the listener gets a LiveGap(dropped) where they would have been.

Both are run-scoped: a listener sees only this run’s events, even when other runs share the thread, and ends after the run’s terminal event (RunCompleted, RunSuspended, RunFailed, RunCancelled, RunTimedOut), or after a Disconnected (Lagging, ListenerFailed, or - for subscribe, which replays - ReplayFailed). A run that crashes without a terminal event (or whose terminal commit fails) ends its stream as soon as the listener has returned from the run’s last event: as the run ends, a marker is queued behind its last event, and reaching it ends the subscription - also for a subscribe made after the run ended, once its replay is done. There is no fixed delay. Either way, await waits for the listener as described above.

Falling behind

Each listener has a bounded buffer (1024 events for stream*) for durable events and another of the same size for live ones, so a burst of live events never costs it a durable one:

  • If the live buffer is full, live events are dropped, and the listener gets StreamEvent.LiveGap(n) with the number lost, so a UI can note that text is missing. The final answer is still in the AgentResult.
  • If the durable buffer is full - the listener is that many durable events behind - it is disconnected with StreamEvent.Disconnected(lastSeq, DisconnectReason.Lagging). Durable events are in the log, so you can resubscribe from where you were with GraphRuntime.subscribe(threadId, afterSeq = lastSeq). That needs the agent’s runtime: build the agent withRuntime(runtime) and subscribe on that runtime. The subscription is thread-scoped, not run-scoped - it delivers every later run on the thread too - and does not end itself: cancel it when you are done.

A listener that throws is disconnected with ListenerFailed; the run is unaffected.

Replay

Durable events outlive the run. GraphRuntime.subscribe(threadId, afterSeq = 0) replays a thread’s events from the start, which gives the structure of every run without any content. See EventCollectionExample.

Tenants

A subscription reads the thread, so it is checked like history: pass the caller’s RunConfig, and a thread of another tenant is refused with GraphError.TenantMismatch before anything is delivered. TracingSubscriber.attach takes the same config. The default, RunConfig(), is the caller with no tenant, which is right only for threads whose runs set none.

1

runtime.subscribe(threadId, config = RunConfig().withTenantId(TenantId("acme")))(listener)

A thread can change hands under its id after you subscribe: one that does not exist yet is created by whichever tenant runs on it first, and one deleted through another runtime can be created again before your subscription learns of the delete. So a subscription screens each event against your tenant, and ends with Disconnected(lastSeq, TenantMismatch) - before the first event of another tenant’s thread, delivering none of it. RunHandle.subscribe and AgentRun.subscribe belong to a run, and use its tenant.

Subscriptions across runtimes

Several GraphRuntimes, in one process or several, can share one checkpoint store (Durable Checkpointers). A GraphRuntime.subscribe on any of them delivers the durable events of every commit on the thread, whichever runtime made it, with the same durable guarantees: in seq order, without gap or duplicate, and only after the commit. How those events interleave with live ones is the exception: see Ordering across runtimes.

  • The subscription’s own runtime hands it that runtime’s commits as they land. Other runtimes’ commits it reads from the store, on a second virtual thread (llm4s-watch-<threadId>) that runs while the subscription is live. A store that is told of commits delivers them at once: InMemoryCheckpointer, shared between runtimes in one process. A store that is not is read every pollInterval, 250 ms by default; the SQLite store is read this way.
  • Only durable events cross runtimes. Live progress is never stored, so another runtime’s TextDeltas and other live events never arrive.
  • Another runtime’s events wait in the store until the listener has room for them, so they never make a subscriber lag. The subscription’s own runtime’s commits are still delivered in their place. A commit that finds other runtimes’ events not yet read reads them first. If they do not fit, the subscriber lags as usual, and you resubscribe from lastSeq.
  • If a read of the store fails, the subscription delivers what it has queued and then ends with Disconnected(lastSeq, ReplayFailed(error)). Closing the store under it ends it this way too. Resubscribe with afterSeq = lastSeq, for example after a restart.
  • If the thread is deleted (GraphRuntime.deleteThread, Agent.forget), through this runtime or another, the subscription delivers what it has queued and then ends with Disconnected(lastSeq, ThreadDeleted): at once in the deleting runtime, and in another runtime at its next read of the store (its next poll, or, with the watch disabled, its runtime’s next commit on the thread). It never delivers a new thread created under the same id as a continuation of the deleted one; subscribe again from afterSeq = 0 to follow that one. A subscription in another runtime that has delivered no event yet cannot be told from a new one, and follows the new thread instead of ending - if it is your tenant’s; one of another tenant ends the subscription with TenantMismatch (see Tenants). A subscription that falls behind while its thread is deleted ends with ThreadDeleted, not Lagging: resubscribing from lastSeq would find the log gone.
1
2
3
4
5
6
7
8
9

import org.llm4s.agent.graph.*
import scala.concurrent.duration.*

// a process that only watches threads whose runs execute elsewhere, over the same SQLite file
val watcher = GraphRuntime(store, watch = WatchPolicy(pollInterval = 100.millis))
watcher.subscribe(threadId) {
  case StreamEvent.Durable(record) => println(s"${record.seq} ${record.event}")
  case _                           => ()
}

WatchPolicy.disabled turns the watch off, for a runtime that knows it is the store’s only user. Its subscriptions then see other runtimes’ commits only by subscribing again, which replays the log.

Agent.stream* and AgentRun.subscribe, and the fs2, ZIO, Java and Kotlin streams built on them, follow a run that their own runtime started, so they work as before. To follow a thread whose runs execute in another process, subscribe with GraphRuntime.subscribe on a runtime over the same store.

Ordering across runtimes

Concurrently active subscriptions to a thread that stay caught up see one interleaving of durable and live events within one runtime: every event handed to several of them has the same relative placement for each. That guarantee does not cover a subscription that attaches later, which misses earlier live events, or a slower one, which can lose live events (LiveGap) or be disconnected as Lagging. Durable seq order is preserved for every subscription. The guarantee holds when nothing outside this runtime commits to the thread. A commit made by another runtime or process reaches each subscription through its own store watch, through a hand-over’s gap read, or, for a subscription still in its initial replay, directly from that replay, so two subscriptions can place it differently; an external commit allows that, but does not necessarily make them differ.

If you need one order across runtimes, order only StreamEvent.Durable(record) by record.seq (or collect those events and sort by seq). There is no durable-only subscription option, so filter in the listener. Ignore StreamEvent.Live and StreamEvent.LiveGap: live events have no seq and are not stored, so they cannot be placed in a cross-runtime order. Treat StreamEvent.Disconnected as the end of the subscription, not as a record: it carries lastSeq and a reason, not an EventRecord. The structural fix, one ordered per-thread log shared by all subscriptions, is recorded as future work in the design record (#1920).

fs2, ZIO and Pekko Streams

llm4s-effect, llm4s-zio and llm4s-pekko expose the same stream as a value:

1
2
3
4
5
6
7
8
9
10
11
12

// cats-effect / fs2: Stream[IO, AgentStreamItem]
agentIO.stream(threadId, "Explain monads").evalMap {
  case AgentStreamItem.Event(AgentEvents.TextDelta(d)) => IO.print(d.text)
  case AgentStreamItem.Event(_)                       => IO.unit
  case AgentStreamItem.Done(result)                   => IO.println(s"\n${result.status}")
}.compile.drain

// ZIO: ZStream[Any, LLMError, AgentStreamItem]
agentZ.stream(threadId, "Explain monads").runForeach { /* same cases */ }

// Apache Pekko Streams: Source[AgentStreamItem, NotUsed]
agentPekko.stream(threadId, "Explain monads").runForeach { /* same cases */ }

AgentStreamItem is Event(StreamEvent) or Done(AgentResult), one enum per module. The stream ends after Done; a Left from admission or from the run fails the stream with that error. A run that ends without a terminal event (a crash) still ends the stream, as soon as the run’s last event has been delivered, and the stream then fails with the run’s error. streamResume and streamRecover exist on all three. A Pekko stream fails with an LLMException carrying the error; see Apache Pekko Integration. Interrupting the stream or stopping early (take(n), head) cancels the turn.

A slow consumer does not: the stream’s buffer never holds up the run’s subscription. Durable events are kept; live events beyond the buffer’s 256 are dropped, and the consumer receives one StreamEvent.LiveGap(n) with their count where they were dropped, then the rest of the run and Done. A Disconnected from the runtime fails the stream, and so does a consumer that has stopped draining altogether: once more than Agent.StreamCapacity (1024) durable events wait for it, as the runtime would disconnect a listener that far behind. Samples: AgentStreamIOExample, AgentStreamZIOExample.

Typed streams of one run

events(run, eventType) on AgentIO, AgentZ and AgentPekko is a stream of one event type’s decoded payloads, for a run you already hold (from agent.start); events(run, handlers) turns several types into one ADT of yours:

1
2
3
4
5
6
7
8
9
10
11
12
13
14

enum Progress:
  case Text(delta: String)
  case Checked(path: String)
  case Gap(lost: Int)

val handlers = EventHandlers
  .on(AgentEvents.TextDelta)(d => Progress.Text(d.text))
  .on(Checked)(c => Progress.Checked(c.path))
  .otherwise { case StreamEvent.LiveGap(n) => Progress.Gap(n); case _ => Progress.Gap(0) }

agentIO.events(run, Checked)          // Stream[IO, Checked]
agentIO.events(run, handlers)         // Stream[IO, Progress]
agentZ.events(run, handlers)          // ZStream[Any, LLMError, Progress]
agentPekko.events(run, handlers)      // Source[Progress, Future[Done]]

Merge typed streams with the library’s own combinators. Such a stream observes the run, through AgentRun.subscribe: it replays the run’s durable events from its start, then follows it live, and misses live events sent before it started. It ends with the run; it does not raise the run’s own error (await the run for that), and stopping it early cancels its subscription, never the run. Handlers run on the stream, so a slow or throwing handler slows or fails the stream only.

Java and Kotlin

The Java facade (llm4s-java-api) and the Kotlin API (modules/kotlin-api) stream a turn the same way, with the same buffer and the same rules: a slow consumer loses live events and gets one StreamEvent.LiveGap(n), never cancels the run; a run that ends without a terminal event still ends the stream with the run’s error; and cancelling the stream cancels the run and returns once it has ended, so the thread can be recovered at once. Both take the thread id as a String, as fs2 and ZIO take a ThreadId; streamResume and streamRecover exist on both. To read what a suspended turn waits for (result.status().pending(), or JAgent.pending and AgentKt.pending), or to resume or recover without a stream, see Suspended turns from Java and Kotlin.

Java. JAgent.stream(threadId, query, listener) returns an LlmResult<AgentStream> at once - a failed result for a refused start, and the listener hears nothing. An AgentStreamListener gets each event through onEvent, then exactly one of onComplete(JAgentResult) or onError(LlmException), all on the stream’s own thread. Only onEvent is abstract, so a lambda is a listener; StreamEvents.decode reads an event as an Optional:

1
2
3
4
5
6
7

// streaming = true: the agent's model calls stream, so the turn carries text deltas
JAgent agent = Llm4s.createAgent(client, ToolRegistry.empty(), true);
LlmResult<AgentStream> started = agent.stream(threadId, "Explain monads", event -> {
    StreamEvents.decode(AgentEvents.TextDelta(), event).ifPresent(d -> System.out.print(d.text()));
    if (event instanceof StreamEvent.LiveGap) System.out.print("[...]");
});
LlmResult<JAgentResult> result = started.get().await();   // or started.get().cancel()

AgentStream.await() returns once the listener has returned from its last call, with the same outcome. cancel() cancels the turn and returns once it has ended, or after 5 seconds with a WARN if it has not (see Threading and cancellation); the listener receives at most the event already being delivered, then onError with the cancellation - or onComplete, when the turn had already ended; an interrupt of the thread calling cancel() does not cut that wait short, and is kept. A listener that throws from onEvent - or sets its own thread’s interrupt flag - cancels the turn, and onError receives what it threw, or a CancelledError. Llm4s.createAgent(client) and createAgent(client, tools) build an agent that calls the model’s complete, so its stream carries no text deltas; pass streaming = true for them. Llm4s.wrapAgent(agent) takes an agent built with Agent.builder, for middleware, a system prompt or a runtime. streamResume takes a List<Answer>, built with Answer.approve(id), reject(id, reason), edit(id, argumentsJson) or reply(id, json). The Java sample in modules/samples/gradle-java streams a turn’s text.

Kotlin. AgentKt.stream(threadId, query) is a cold Flow<AgentStreamItem>: each collection runs the turn, emitting AgentStreamItem.Event(event) for each event, then AgentStreamItem.Done(result), a JAgentResult. A refused start or a failed run throws LLMException. Cancelling the collection - its scope, take(n), withTimeout - cancels the turn; starting and cancelling it run on Dispatchers.IO. A turn cancelled by anything but the collection fails it with LLMException, as in fs2 and ZIO. So does a delivery that dies of a fatal error (an OutOfMemoryError, a LinkageError): the Java listener then hears no terminal callback, and the flow - or run, continueConversation, resume or recover - ends with the facade’s report of it instead of suspending.

1
2
3
4
5
6
7

val agent = Llm4s.createAgent(client, ToolRegistry.empty(), streaming = true)
agent.stream(threadId, "Explain monads").collect { item ->
    when (item) {
        is AgentStreamItem.Event -> StreamEvents.decode(AgentEvents.TextDelta(), item.event).ifPresent { print(it.text()) }
        is AgentStreamItem.Done  -> println("\n${item.result.status().kind()}")
    }
}

The flow’s channel holds 64 items; past that, the stream’s 256-event buffer fills and live events are dropped.

Your own graphs

The same machinery serves graphs that are not agents. An EventType[A] names a payload once:

1
2
3
4
5
6

val Checked = EventType[Checked]("myapp.checked", 1)   // name matches [a-z0-9_.]{1,64}

Checked.emit(context, value)       // durable: stored with the task's commit
Checked.progress(context, value)   // live: sent now, never stored

listener { case Checked(v) => ... } // None for another name, version or an undecodable payload

RunContext.progress(name, version, payload) is the untyped form. A tool emits through its context’s run: Checked.progress(context.run, value).

Progress is sent only while the task’s run holds the thread. A node that keeps its RunContext, or a thread of its own, past the end of its run sends nothing: once the run has released the thread, or found its claim lost, progress is a no-op, so it never reaches a later run’s subscribers.

Typed handlers

EventHandlers packages typed handlers with their event types, so a selection of mixed types is a value, and each handler gets its own payload type:

1
2
3
4
5
6
7
8
9

val handlers = EventHandlers
  .on(AgentEvents.TextDelta)(d => print(d.text))
  .on(Checked)(c => println(s"checked ${c.path}"))
  .otherwise {
    case StreamEvent.LiveGap(n) => println(s"[$n live events lost]")
    case _                      => ()
  }

agent.stream(threadId, query)(handlers)   // also run.subscribe()(handlers), runtime.subscribe(...)(handlers)

An event is looked up once by name and version and its payload decoded once. Events no handler takes - kernel events, LiveGap, Disconnected, other names or versions, a payload that does not decode - go to otherwise, or are skipped without it. A second on for the same name and version replaces the first. handlers.dispatch(event) returns a handler’s result, for use in your own code.

Live events are at least once across recovery: a tool that recover runs again sends its progress again, to the recovered run’s stream. Durable events are not repeated in the log - a failed attempt’s are discarded with its task - but a handler with an effect of its own should key it by something stable carried in the payload, such as the tool call’s IdempotencyKey, never by run id.

Limits

  • Kernel failure messages. TaskFailed and RunFailed store an error message. If a guardrail’s or tool’s error quotes content (a guardrail reason that echoes the user’s text), that text reaches the log through the kernel event, not through an agent.* event.
  • Live events are not replayed, and a late subscribe misses earlier ones. A subscription receives every live event sent after subscribe returns, so GraphRuntime.subscribe followed by start sees all of the run’s - unless more arrive than its capacity holds, while it replays or before its listener drains them: those are dropped and reported as a LiveGap with their count.
  • Live events stay in their runtime. A subscription receives another runtime’s durable events, but never its live ones, which are not stored. With a polled store, such as SQLite, another runtime’s commits arrive up to WatchPolicy.pollInterval after they land.

See also the observability guide, the migration note, and Typed Agent Runtime §4.14.