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 (requires_new): emit is routinely called inside enclosing transactions, and on Postgres a failed INSERT otherwise aborts the whole transaction — the retry would raise PG::InFailedSqlTransaction instead of recovering.



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

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
    # (collisions are rare); Postgres runs them truly concurrently, so a
    # burst can cost more than a handful of 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 } }.



59
60
61
 
# File 'app/models/xeno/event.rb', line 59

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