The fundamentals · 09

Stream Events

When streaming responses, Riffer emits typed events that represent incremental updates from the LLM.

Using Streaming

Use stream instead of generate to receive events as they arrive:

agent = MyAgent.new

agent.stream("Tell me a story").each do |event|
  case event
  when Riffer::StreamEvents::TextDelta
    print event.content
  when Riffer::StreamEvents::TextDone
    puts "\n[Complete]"
  when Riffer::StreamEvents::ToolCallDelta
    # Tool call being built
  when Riffer::StreamEvents::ToolCallDone
    puts "[Tool: #{event.name}]"
  when Riffer::StreamEvents::WebSearchStatus
    puts "[Searching: #{event.status}]"
  when Riffer::StreamEvents::WebSearchDone
    puts "[Search complete: #{event.query}]"
  end
end

Event Types

TextDelta

Emitted when incremental text content is received:

event = Riffer::StreamEvents::TextDelta.new("Hello ")
event.role     # => :assistant
event.content  # => "Hello "
event.to_h     # => {role: :assistant, content: "Hello "}

Use this to display text in real-time as it streams.

TextDone

Emitted when text generation is complete:

event = Riffer::StreamEvents::TextDone.new("Hello, how can I help you?")
event.role     # => :assistant
event.content  # => "Hello, how can I help you?"
event.to_h     # => {role: :assistant, content: "Hello, how can I help you?"}

Fires once per model response, after its last TextDelta, and contains the response’s full text: every text block the provider returned, concatenated in order, so it always equals the joined TextDelta contents. A response that carries no text (for example, only tool calls) emits no TextDone.

ToolCallDelta

Emitted when tool call arguments are being streamed:

event = Riffer::StreamEvents::ToolCallDelta.new(
  item_id: "item_123",
  name: "weather_tool",
  arguments_delta: '{"city":'
)
event.role             # => :assistant
event.item_id          # => "item_123"
event.name             # => "weather_tool"
event.arguments_delta  # => '{"city":'

The name may only be present in the first delta. Accumulate arguments_delta to build the complete arguments.

ToolCallDone

Emitted when a tool call is complete:

event = Riffer::StreamEvents::ToolCallDone.new(
  item_id: "item_123",
  call_id: "call_456",
  name: "weather_tool",
  arguments: '{"city":"Tokyo"}'
)
event.role       # => :assistant
event.item_id    # => "item_123"
event.call_id    # => "call_456"
event.name       # => "weather_tool"
event.arguments  # => '{"city":"Tokyo"}'

Contains the complete tool call information.

ReasoningDelta

Emitted when reasoning/thinking content is streamed (OpenAI with reasoning enabled):

event = Riffer::StreamEvents::ReasoningDelta.new("Let me think about ")
event.role     # => :assistant
event.content  # => "Let me think about "

ReasoningDone

Emitted when one reasoning block is complete:

part = Riffer::Messages::Assistant::ReasoningPart.new(type: :text, text: "Let me think about this step by step...", format: "mock-v1")
event = Riffer::StreamEvents::ReasoningDone.new(part)
event.role       # => :assistant
event.part       # => the ReasoningPart
event.part.text  # => "Let me think about this step by step..."

part is the reasoning part the preceding ReasoningDelta events added up to, and the agent loop accumulates it onto the assistant message. Adapters that cannot yet replay their reasoning emit it as a :text part with no format, so it is stored for display but never sent back to the provider.

WebSearchStatus

Emitted during web search progress with status updates:

event = Riffer::StreamEvents::WebSearchStatus.new("searching", query: "Ruby language")
event.role    # => :assistant
event.status  # => "searching"
event.query   # => "Ruby language"
event.url     # => nil
event.to_h    # => {role: :assistant, status: "searching", query: "Ruby language"}

The url and query attributes are optional and only included in to_h when present. Status values include "in_progress", "searching", "completed", and "open_page".

# "open_page" status includes a url
event = Riffer::StreamEvents::WebSearchStatus.new("open_page", url: "https://example.com")
event.url   # => "https://example.com"
event.to_h  # => {role: :assistant, status: "open_page", url: "https://example.com"}

WebSearchDone

Emitted when web search completes:

event = Riffer::StreamEvents::WebSearchDone.new(
  "Ruby language",
  sources: [{title: "Ruby Programming", url: "https://ruby-lang.org"}]
)
event.role     # => :assistant
event.query    # => "Ruby language"
event.sources  # => [{title: "Ruby Programming", url: "https://ruby-lang.org"}]
event.to_h     # => {role: :assistant, query: "Ruby language", sources: [...]}

Contains the search query and an array of source hashes with title and url keys.

GuardrailTripwire

Emitted when a guardrail blocks execution during streaming:

agent.stream("Hello").each do |event|
  case event
  when Riffer::StreamEvents::GuardrailTripwire
    puts "Blocked by: #{event.guardrail_id}"
    puts "Reason: #{event.reason}"
    puts "Phase: #{event.phase}"  # :before or :after
  end
end

See Guardrails for more information.

GuardrailModification

Emitted when a guardrail transforms data during streaming:

agent.stream("Hello").each do |event|
  case event
  when Riffer::StreamEvents::GuardrailModification
    puts "Modified by: #{event.guardrail_id}"
    puts "Phase: #{event.phase}"              # :before or :after
    puts "Changed: #{event.message_indices}"  # Array of affected indices
  end
end

See Guardrails for more information.

SkillActivation

Emitted when a skill is activated during streaming. Fired when the LLM calls the skill activation tool:

agent.stream("Review this code").each do |event|
  case event
  when Riffer::StreamEvents::SkillActivation
    puts "[Skill activated: #{event.name}]"
  end
end

event = Riffer::StreamEvents::SkillActivation.new("code-review")
event.role  # => :system
event.name  # => "code-review"
event.to_h  # => {role: :system, name: "code-review"}

See Skills for more information.

Interrupt

Emitted when the agent loop is interrupted. This can happen in two ways:

  • An on_message callback calls agent.interrupt! or throw :riffer_interrupt (reason is a String or nil).
  • The max_steps limit is reached (reason is the Symbol :max_steps).

This is the streaming equivalent of response.outcome.reason == :interrupted (or :max_steps) in generate mode.

# Callback interrupt with a string reason
event = Riffer::StreamEvents::Interrupt.new(reason: "needs approval")
event.role    # => :system
event.reason  # => "needs approval"
event.to_h    # => {role: :system, interrupt: true, reason: "needs approval"}

# Max steps interrupt with a symbol reason
event = Riffer::StreamEvents::Interrupt.new(reason: :max_steps)
event.reason  # => :max_steps

The reason is nil when throw :riffer_interrupt is called without a second argument.

agent.stream("Hello").each do |event|
  case event
  when Riffer::StreamEvents::Interrupt
    puts "Loop was interrupted: #{event.reason}"
  end
end

After an interrupt, call stream again with a string to continue the loop. See Agent Lifecycle — Resuming an Interrupted Loop for details.

TokenUsageDone

Emitted when token usage data is available at the end of a response:

event = Riffer::StreamEvents::TokenUsageDone.new(token_usage: token_usage)
event.role                          # => :assistant
event.token_usage                   # => Riffer::Providers::TokenUsage
event.token_usage.input_tokens      # => 100
event.token_usage.output_tokens     # => 50
event.token_usage.total_tokens      # => 150
event.to_h                          # => {role: :assistant, token_usage: {input_tokens: 100, output_tokens: 50}}

Use this to track token consumption in real-time during streaming.

FinishReasonDone

Emitted once near the end of the stream with the normalized reason the model finished (no ordering guarantee relative to TokenUsageDone):

event = Riffer::StreamEvents::FinishReasonDone.new(finish_reason: :length, raw_finish_reason: "max_tokens")
event.role               # => :assistant
event.finish_reason      # => :length (see Messages — Finish Reasons for the vocabulary)
event.raw_finish_reason  # => "max_tokens" (the provider's raw wire value, or nil)
event.to_h               # => {role: :assistant, finish_reason: :length, raw_finish_reason: "max_tokens"}

The agent loop stamps this value onto the accumulated assistant message’s finish_reason.

Incomplete Streams

If a provider’s stream ends before its terminal event, the enumerator raises Riffer::IncompleteStreamError (a Riffer::Error subclass) instead of finishing normally, so a truncated or empty response is never returned as a complete message. Events already yielded before the raise were delivered as usual, but nothing from the failed step is added to the session: there is no partial assistant message to resume from, and messages from earlier completed steps (tool calls and their results) are untouched.

Supported on Amazon Bedrock, Anthropic, and OpenAI / Azure OpenAI.

The user prompt is added to the session before the run starts and stays there after the failure, so retry with agent.stream and no prompt. Passing the prompt again would add a second user turn.

attempts = 0
prompt = "Tell me a story"
begin
  agent.stream(prompt).each do |event|
    print event.content if event.is_a?(Riffer::StreamEvents::TextDelta)
  end
rescue Riffer::IncompleteStreamError => e
  warn "stream ended early: #{e.message}"
  prompt = nil # already in the session; re-run on the existing history
  retry if (attempts += 1) < 3
  raise
end

Streaming with Tools

When an agent uses tools during streaming, the flow is:

  1. TextDelta, ToolCallDelta and ToolCallDone events stream in, in whatever order the model produces them. Text can come before, between or after tool calls.
  2. If the response contained any text, one TextDone follows with the full text. A response that only calls tools has no TextDone.
  3. Agent executes tools internally
  4. Agent sends results back to LLM
  5. More text events stream in
  6. Repeat until no more tool calls

Note: When web_search is enabled, WebSearchStatus and WebSearchDone events may appear before text events as the provider performs a server-side search.

agent.stream("What's the weather in Tokyo?").each do |event|
  case event
  when Riffer::StreamEvents::TextDelta
    print event.content
  when Riffer::StreamEvents::ToolCallDone
    puts "\n[Calling #{event.name}...]"
  when Riffer::StreamEvents::TextDone
    puts "\n"
  end
end

Complete Example

class WeatherAgent < Riffer::Agent
  model 'openai/gpt-5-mini'
  instructions 'You are a weather assistant.'
  uses_tools [WeatherTool]
end

agent = WeatherAgent.new
text_buffer = ""

agent.stream("What's the weather in Tokyo and New York?").each do |event|
  case event
  when Riffer::StreamEvents::TextDelta
    print event.content
    text_buffer += event.content

  when Riffer::StreamEvents::TextDone
    # Final text available
    puts "\n---"
    puts "Complete response: #{event.content}"

  when Riffer::StreamEvents::ToolCallDelta
    # Could show "typing..." indicator

  when Riffer::StreamEvents::ToolCallDone
    puts "\n[Tool: #{event.name}(#{event.arguments})]"

  when Riffer::StreamEvents::ReasoningDelta
    # Show thinking process if desired
    print "[thinking] #{event.content}"

  when Riffer::StreamEvents::ReasoningDone
    puts "\n[reasoning complete]"

  when Riffer::StreamEvents::WebSearchStatus
    puts "[search: #{event.status}]"

  when Riffer::StreamEvents::WebSearchDone
    puts "[search complete: #{event.query}]"
    event.sources.each { |s| puts "  - #{s[:title]}: #{s[:url]}" }

  when Riffer::StreamEvents::Interrupt
    puts "\n[interrupted]"
  end
end

Base Class

All events inherit from Riffer::StreamEvents::Base:

class Riffer::StreamEvents::Base
  attr_reader :role

  def initialize(role: :assistant)
    @role = role
  end

  def to_h
    raise NotImplementedError
  end
end