class RubyLLM::AgUI::Emitter

Writes one run’s events and keeps its durable record in step.

The mapper decides which events a RubyLLM signal becomes; the emitter is what makes them durable and ordered. Every event advances the run record’s sequence, so a reconnect can order snapshots by it instead of by wall clock, and a terminal outcome is persisted before it is written: if the transport dies after the outcome is decided, the run still cannot be executed a second time.

The emitter decides nothing about the outcome and executes nothing. What it owns is the one ordering rule a caller should never have to remember: a run’s open text, tool calls and step are closed before the terminal event.

Attributes

record[R]

The run record, including the sequence numbers written so far.

Public Class Methods

new(writer:, mapper:, store:, record:, clock:, logger: nil) click to toggle source

rubocop:disable-next Metrics/ParameterLists – one keyword per collaborator the caller injects

# File lib/ruby_llm/ag_ui/emitter.rb, line 22
def initialize(writer:, mapper:, store:, record:, clock:, logger: nil)
  @writer = writer
  @mapper = mapper
  @store = store
  @record = record
  @clock = clock
  @logger = logger
end

Public Instance Methods

cancel() click to toggle source

Closes a run that was stopped before it completed. Never reported as a success: the outcome carries the protocol’s own “cancelled” discriminator, which 0.2.0 cannot express.

# File lib/ruby_llm/ag_ui/emitter.rb, line 65
def cancel
  @record = persist(@record.finish(RunRecord::CANCELLED, now: now))
  close(run_finished(Protocol.cancelled_outcome))
  @record
end

fail(error) click to toggle source

Closes the run as failed with a sanitized RUN_ERROR.

The record is persisted first, so a run that failed is not retried because reporting the failure failed too. The exception’s own message is never sent: it may carry prompts, keys or paths.

# File lib/ruby_llm/ag_ui/emitter.rb, line 85
def fail(error)
  message, code = ErrorSanitizer.sanitize(error)
  failure = { 'message' => message, 'code' => code }
  @logger&.warn("[ruby_llm_ag_ui] run #{@record.run_id} failed: #{error.class}")
  @record = persist(@record.finish(RunRecord::FAILED, now: now, failure: failure))
  close(::AgUiProtocol::Core::Events::RunErrorEvent.new(message: message, code: code))
  @record
end

interrupt(interrupts) click to toggle source

Closes the run as paused, recording interrupts so a later resume can answer them by id. At least one must be unresolved, which RunRecord enforces.

# File lib/ruby_llm/ag_ui/emitter.rb, line 74
def interrupt(interrupts)
  @record = persist(@record.interrupted(interrupts, now: now))
  close(run_finished(Protocol.interrupt_outcome(interrupts)))
  @record
end

replay(events) click to toggle source

Writes a reconnect’s events without advancing or persisting the record. The run already ended; a reconnect reports it rather than changing it.

# File lib/ruby_llm/ag_ui/emitter.rb, line 50
def replay(events)
  Array(events).each { |event| emit(event) }
  @record
end

start(event, state: nil) click to toggle source

Moves the run to started, records state as the host-owned state a reconnect serves, then opens the run on the wire with event.

# File lib/ruby_llm/ag_ui/emitter.rb, line 33
def start(event, state: nil)
  @record = persist(@record.start(now: now, snapshot: state))
  write(event)
  @record
end

succeed() click to toggle source

Closes the run as succeeded.

# File lib/ruby_llm/ag_ui/emitter.rb, line 56
def succeed
  @record = persist(@record.finish(RunRecord::SUCCEEDED, now: now))
  close(run_finished(Protocol.success_outcome))
  @record
end

write(events) click to toggle source

Writes events (one event or an array) and advances the run’s sequence by one per event.

# File lib/ruby_llm/ag_ui/emitter.rb, line 41
def write(events)
  Array(events).each do |event|
    emit(event)
    @record = @record.advance(now: now)
  end
end