Class: Xeno::Event

Inherits:
ApplicationRecord show all
Defined in:
app/models/xeno/event.rb

Overview

The append-only session stream — also the audit log. index is the per-session cursor; consumers dedupe re-emitted events by (session, index).

Constant Summary collapse

TYPES =

The fixed vocabulary (docs/runtime.md § Events). Hooks may subscribe to any of these or "*".

%w[
  session.started turn.started message.received
  step.started step.completed step.failed
  actions.requested action.result input.requested
  reasoning.completed message.completed
  compaction.requested compaction.completed
  budget.exceeded
  turn.completed turn.failed turn.cancelled
  session.waiting session.completed session.failed
].freeze

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.append!(session, event_type, data = {}) ⇒ Object

Appends with a race-safe per-session index: concurrent writers collide on the unique index and retry with the next slot. The INSERT runs in its own savepoint because emit is routinely called inside enclosing transactions, and on Postgres a failed INSERT would otherwise abort the whole transaction.



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'app/models/xeno/event.rb', line 29

def self.append!(session, event_type, data = {})
  attempts = 0
  begin
    next_index = (where(session_id: session.id).maximum(:index) || -1) + 1
    transaction(requires_new: true) do
      create!(
        session_id: session.id,
        index: next_index,
        event_type: event_type,
        data: data,
        created_at: Time.current
      )
    end
  rescue ActiveRecord::RecordNotUnique
    # Every collision means a competing append succeeded, so retries are bounded by actual
    # contention. SQLite's file lock serializes writers; Postgres runs them concurrently, so a
    # burst can cost several rounds — back off with jitter instead of giving up early.
    attempts += 1
    raise if attempts >= 50

    sleep(rand * 0.002 * attempts)
    retry
  end
end

Instance Method Details

#envelope ⇒ Object

The wire envelope: { type, data, meta: { index, at } }.



55
56
57
 
# File 'app/models/xeno/event.rb', line 55

def envelope
  { "type" => event_type, "data" => data || {}, "meta" => { "index" => index, "at" => created_at.iso8601(3) } }
end