class RubyLLM::AgUI::Rails::ActiveRecordStore

The durable run store: run identity in a table instead of a process.

The in-process store is correct for one process only. A host that runs more than one Puma worker, or that restarts, needs the reservation, the interrupts and the terminal status to outlive the worker that produced them, which is what this store is for.

Ordering is by row id, so “the latest run on the thread” means the run reserved last, not the run whose clock reading is largest. Two requests that arrive in the same millisecond still have an order.

RubyLLM::AgUI.configure { |config| config.run_store = ActiveRecordStore.new }

Public Instance Methods

find(scope:, thread_id:, run_id:) click to toggle source

# File lib/ruby_llm/ag_ui/rails/active_record_store.rb, line 37
def find(scope:, thread_id:, run_id:)
  row_for(scope, thread_id, run_id)&.as_run_record
end

find_interrupt(scope:, thread_id:, interrupt_id:) click to toggle source

# File lib/ruby_llm/ag_ui/rails/active_record_store.rb, line 59
def find_interrupt(scope:, thread_id:, interrupt_id:)
  raise InvalidInputError, 'interrupt_id is required.' if blank?(interrupt_id)

  row_for_interrupt(scope, thread_id, interrupt_id)&.as_run_record&.interrupt(interrupt_id)
end

latest_interrupted(scope:, thread_id:) click to toggle source

# File lib/ruby_llm/ag_ui/rails/active_record_store.rb, line 45
def latest_interrupted(scope:, thread_id:)
  interrupted = RunRecord::INTERRUPTED.to_s
  thread_rows(scope, thread_id).reverse.find { |row| row.status == interrupted }&.as_run_record
end

latest_run(scope:, thread_id:) click to toggle source

# File lib/ruby_llm/ag_ui/rails/active_record_store.rb, line 41
def latest_run(scope:, thread_id:)
  thread_rows(scope, thread_id).last&.as_run_record
end

redeem_interrupt(scope:, thread_id:, interrupt_id:, decision:, now:) click to toggle source

Records decision on the interrupt under a row lock, so two resumes that arrive together cannot both record a decision and run the tool.

# File lib/ruby_llm/ag_ui/rails/active_record_store.rb, line 67
def redeem_interrupt(scope:, thread_id:, interrupt_id:, decision:, now:)
  unless Interrupt::DECISIONS.include?(decision)
    raise ConfigurationError, "decision must be one of #{Interrupt::DECISIONS.join(', ')}"
  end

  row = row_for_interrupt(scope, thread_id, interrupt_id)
  raise ResumeRejectedError, "no interrupt #{interrupt_id}" if row.nil?

  row.with_lock { redeem(row.reload.as_run_record, interrupt_id, decision, now) }
end

reserve(scope:, thread_id:, run_id:, digest:) click to toggle source

Claims (scope, thread_id, run_id) for digest.

The unique index on the identity is the real guard: two workers that insert at the same time cannot both win, and the loser attaches to the row the winner wrote instead of executing the turn a second time.

# File lib/ruby_llm/ag_ui/rails/active_record_store.rb, line 26
def reserve(scope:, thread_id:, run_id:, digest:)
  scope_key(scope)
  record = RunRecord.new(
    scope: scope, thread_id: thread_id, run_id: run_id, digest: digest, created_at: now
  )
  Record.from_run_record(record).save!
  Reservation.new(record: record, status: Reservation::CREATED)
rescue ::ActiveRecord::RecordNotUnique
  attach(row_for(scope, thread_id, run_id), digest)
end

save(record) click to toggle source

# File lib/ruby_llm/ag_ui/rails/active_record_store.rb, line 50
def save(record)
  scope_key(record.scope)
  row = row_for(record.scope, record.thread_id, record.run_id)
  raise InvalidRunIdentityError, "run #{record.run_id} was never reserved" if row.nil?
  raise RunConflictError, conflict_message(record) if row.digest != record.digest

  row.write_run_record(record)
end