class RubyLLM::AgUI::Store::MemoryStore

An in-process run store.

It is correct for a single process and a reasonable default in development and tests. It is not shared between processes, so a host that runs more than one Puma worker, or retries on a different worker, must use a durable store (the optional Rails adapter’s ActiveRecord store) instead. Nothing here pretends otherwise.

Every public method holds the store’s lock for its whole body, so the reservation and the interrupt redemption are atomic: the first caller to answer an interrupt is the only one that records a decision. The private helpers assume the lock is already held and never take it themselves.

Public Class Methods

new(clock: nil) click to toggle source

Calls superclass method RubyLLM::AgUI::Store::Base::new

# File lib/ruby_llm/ag_ui/store/memory_store.rb, line 19
def initialize(clock: nil)
  super
  @records = {}
  @threads = {}
  @lock = Mutex.new
end

Public Instance Methods

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

# File lib/ruby_llm/ag_ui/store/memory_store.rb, line 42
def find(scope:, thread_id:, run_id:)
  key = key_for(scope, thread_id, run_id)
  @lock.synchronize { @records[key] }
end

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

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

  @lock.synchronize { record_for_interrupt(scope, thread_id, interrupt_id)&.interrupt(interrupt_id) }
end

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

# File lib/ruby_llm/ag_ui/store/memory_store.rb, line 51
def latest_interrupted(scope:, thread_id:)
  @lock.synchronize { records_for(scope, thread_id).reverse.find(&:interrupted?) }
end

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

# File lib/ruby_llm/ag_ui/store/memory_store.rb, line 47
def latest_run(scope:, thread_id:)
  @lock.synchronize { records_for(scope, thread_id).last }
end

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

# File lib/ruby_llm/ag_ui/store/memory_store.rb, line 73
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

  @lock.synchronize { redeem(scope, thread_id, interrupt_id, decision, now) }
end

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

# File lib/ruby_llm/ag_ui/store/memory_store.rb, line 26
def reserve(scope:, thread_id:, run_id:, digest:)
  key = key_for(scope, thread_id, run_id)
  @lock.synchronize do
    existing = @records[key]
    next attach(existing, digest) if existing

    record = RunRecord.new(
      scope: scope, thread_id: thread_id, run_id: run_id, digest: digest, created_at: now
    )
    @records[key] = record
    thread = thread_key(scope, thread_id)
    @threads[thread] = (@threads[thread] || []) + [run_id]
    Reservation.new(record: record, status: Reservation::CREATED)
  end
end

save(record) click to toggle source

# File lib/ruby_llm/ag_ui/store/memory_store.rb, line 55
def save(record)
  key = key_for(record.scope, record.thread_id, record.run_id)
  @lock.synchronize do
    stored = @records[key]
    raise InvalidRunIdentityError, "run #{record.run_id} was never reserved" if stored.nil?
    raise RunConflictError, conflict_message(record) if stored.digest != record.digest

    @records[key] = record
  end
  record
end