class RubyLLM::Browser::Rails::ActiveRecordOperationStore

Durable, multi-worker {RubyLLM::Browser::OperationStore} backed by ActiveRecord.

Supports SQLite and PostgreSQL. Reservations acquire a durable per-actor write lock, including when the actor has no existing operations. Worker writes check the exact fence and legal transition. No database transaction spans network I/O.

a duplicate reservation loses the unique-index race and is reported as {InvalidInput} instead of silently starting a second session
a duplicated worker that re-leases an operation bumps its fence, and any earlier writer is then refused as stale (fenced out)
a cancelled operation keeps its row and reservation until closure/expiry is proven; cancellation fences stale workers by bumping the fence

The store never interprets provider truth: it records the states the engine and reaper report.

Public Class Methods

new(record_class: OperationRecord) click to toggle source

Calls superclass method

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 22
def initialize(record_class: OperationRecord)
  super()
  @record_class = record_class
end

Public Instance Methods

attach_or_reserve(operation) click to toggle source

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 41
def attach_or_reserve(operation)
  @record_class.transaction do
    lock_actor!(operation.actor_id)
    existing = find_by_idempotency_key(operation.actor_id, operation.run_id, operation.idempotency_key)
    if existing
      unless existing.request_digest == operation.request_digest
        raise InvalidInput, "idempotency key belongs to a different request"
      end
      return existing.to_operation
    end
    raise InvalidInput, "operation is already reserved" if @record_class.exists?(operation.id)

    enforce_concurrency!(operation)
    insert!(operation)
  end
end

cancel(id, actor_id:, run_id:, now:) click to toggle source

Host-initiated cancellation. Marks the operation and fences stale workers.

@return [Operation] the cancelled (or already terminal) operation

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 138
def cancel(id, actor_id:, run_id:, now:)
  @record_class.transaction do
    record = locked!(id)
    assert_actor!(record, actor_id, run_id)
    current = record.to_operation
    return current if current.terminal? || current.cancel_requested?

    cancelled = current.with_fence(record.fence + 1, now: now).request_cancel(now: now)
    persist!(record, cancelled)
  end
end

claim_start(operation, now:) click to toggle source

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 89
def claim_start(operation, now:)
  @record_class.transaction do
    record = locked!(operation.id)
    current = record.to_operation
    operation.validate_update!(current)
    raise InvalidInput, "operation was already started" unless current.state == :reserved
    raise BudgetExhausted, "operation wall-clock budget exhausted" if current.expired?(now)

    persist!(record, current.advance(:starting, now: now))
  end
end

confirm_closed(operation, session_id:, now:) click to toggle source

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 114
def confirm_closed(operation, session_id:, now:)
  @record_class.transaction do
    record = locked!(operation.id)
    assert_actor!(record, operation.actor_id, operation.run_id)
    current = record.to_operation
    raise InvalidInput, "operation session cannot change" unless current.session_id == session_id

    target = current.cancel_requested? ? :cancelled : :closed
    persist!(record, current.advance(target, now: now))
  end
end

fetch(id) click to toggle source

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 58
def fetch(id)
  @record_class.find_by(id: id)&.to_operation
end

fetch!(id) click to toggle source

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 62
def fetch!(id)
  fetch(id) || raise(InvalidInput, "unknown operation #{id.inspect}")
end

find_by_idempotency_key(actor_id, run_id, key) click to toggle source

Finds an existing operation by its actor/run idempotency key.

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 159
def find_by_idempotency_key(actor_id, run_id, key)
  return nil if key.nil?

  @record_class.find_by(actor_id: actor_id, run_id: run_id, idempotency_key: key)
end

lease(id, actor_id:, run_id:, now:) click to toggle source

Claims the operation for one worker by bumping its fence token.

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 67
def lease(id, actor_id:, run_id:, now:)
  @record_class.transaction do
    record = locked!(id)
    assert_actor!(record, actor_id, run_id)
    if record.to_operation.terminal?
      raise InvalidInput, "operation #{id} is already #{record.state}"
    end

    leased = record.to_operation.with_fence(record.fence + 1, now: now)
    persist!(record, leased)
  end
end

mark_cleanup_pending(id, actor_id:, run_id:, error_code: nil, now:) click to toggle source

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 126
def mark_cleanup_pending(id, actor_id:, run_id:, error_code: nil, now:)
  @record_class.transaction do
    record = locked!(id)
    assert_actor!(record, actor_id, run_id)
    updated = record.to_operation.advance(:cleanup_pending, now: now, error_code: error_code)
    persist!(record, updated)
  end
end

open_for(actor_id) click to toggle source

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 150
def open_for(actor_id)
  @record_class.where(actor_id: actor_id).where(state: open_states).order(:created_at).map(&:to_operation)
end

open_operations() click to toggle source

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 154
def open_operations
  @record_class.where(state: open_states).order(:created_at).map(&:to_operation)
end

record_session(operation, session_id:, now:) click to toggle source

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 101
def record_session(operation, session_id:, now:)
  @record_class.transaction do
    record = locked!(operation.id)
    assert_actor!(record, operation.actor_id, operation.run_id)
    current = record.to_operation
    if current.session_id && current.session_id != session_id
      raise InvalidInput, "operation session cannot change"
    end
    target = current.fence == operation.fence ? :active : :cleanup_pending
    persist!(record, current.advance(target, session_id: session_id, now: now))
  end
end

reserve(operation) click to toggle source

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 27
def reserve(operation)
  @record_class.transaction do
    lock_actor!(operation.actor_id)
    if @record_class.exists?(operation.id)
      raise InvalidInput, "operation #{operation.id} is already reserved"
    end

    enforce_concurrency!(operation)
    insert!(operation)
  end
rescue ActiveRecord::RecordNotUnique
  raise InvalidInput, "operation #{operation.id} is already reserved"
end

update(operation) click to toggle source

Persists operation only when its fence is current.

# File lib/ruby_llm/browser/rails/active_record_operation_store.rb, line 81
def update(operation)
  @record_class.transaction do
    record = locked!(operation.id)
    operation.validate_update!(record.to_operation)
    persist!(record, operation)
  end
end