Class: Xeno::Session

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

Overview

The durable conversation: lives for days or weeks, owns the transcript chat, the turn ledger, and the append-only event stream. Identified by id (inspect/stream handle) and at most one continuation_token (the channel-owned resume handle, unique among active sessions).

Constant Summary collapse

STATUSES = 
%w[running waiting completed failed].freeze

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.open!(message:, channel: nil, principal: nil, continuation_token: nil, definition: Xeno.definition) ⇒ Object

Opens a session for a first user message: builds the transcript chat per the resolved agent definition, stamps channel + principal, and stages turn 1. The caller enqueues the returned turn's job after the transaction commits (Session.start! does both).



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'app/models/xeno/session.rb', line 20

def self.open!(message:, channel: nil, principal: nil, continuation_token: nil, definition: Xeno.definition)
  transaction do
    chat = Chat.new
    options = definition.config.model_options
    chat.apply_runtime_options!(options)
    chat.provider = options[:provider].to_s if options[:provider]
    chat.model = definition.config.resolved_model
    chat.save!

    session = create!(
      agent: definition.name,
      chat: chat,
      channel: channel,
      principal: principal,
      continuation_token: continuation_token
    )
    session.emit("session.started", { agent: definition.name, channel: channel })
    session.stage_turn!(message, definition: definition)
    session
  end
end

.start!(message:, **options) ⇒ Object

Convenience for callers without transactional needs: open + enqueue.



43
44
45
46
47
 
# File 'app/models/xeno/session.rb', line 43

def self.start!(message:, **options)
  session = open!(message: message, **options)
  session.turns.last.enqueue!
  session
end

Instance Method Details

#active? ⇒ Boolean

Returns:

  • (Boolean) 


214
215
216
 
# File 'app/models/xeno/session.rb', line 214

def active?
  %w[running waiting].include?(status)
end

#drain_pending_messages!(definition: Xeno.definition) ⇒ Object

Folds every queued message into ONE next turn. Called when a turn parks or reaches a terminal status (and on messages to a parked session). Always defers the transcript: the drained turn may sit behind a parked turn whose tool calls are still unanswered, so its user rows are written only when the runner claims it — turn order and transcript order can never diverge. Row locks make a racing drain (park vs. incoming message) fold each message exactly once.



109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'app/models/xeno/session.rb', line 109

def drain_pending_messages!(definition: Xeno.definition)
  turn = nil
  transaction do
    queued = pending_messages.order(:id).lock.to_a
    next if queued.empty?

    contents = queued.map { |m| m.payload["content"] }
    turn = stage_turn!(contents, definition: definition, emit_received: false, defer_transcript: true)
    queued.each(&:destroy!)
  end
  turn
end

#emit(event_type, data = {}) ⇒ Object 



173
174
175
 
# File 'app/models/xeno/session.rb', line 173

def emit(event_type, data = {})
  Event.append!(self, event_type, data)
end

#finish!(new_status) ⇒ Object

Terminal states release the continuation token (a finished session must not squat a channel thread's token); a copy stays in metadata for audit.

Raises:

  • (ArgumentError) 


220
221
222
223
224
225
226
227
228
229
 
# File 'app/models/xeno/session.rb', line 220

def finish!(new_status)
  raise ArgumentError, "not a terminal status: #{new_status}" unless %w[completed failed].include?(new_status)

  transaction do
    audit = (metadata || {}).merge("released_continuation_token" => continuation_token)
    # State is working memory for the conversation; reset clears it.
    update!(status: new_status, continuation_token: nil, metadata: audit, state: {})
    emit("session.#{new_status}")
  end
end

#receive_message!(content) ⇒ Object

The channel entry point for follow-up messages. While a turn is active (one active turn per session) the message queues in pending_messages and is folded into the next turn; a PARKED session drains immediately (the message becomes a visible staged turn instead of sitting invisible until someone resolves the input — mvp-design's delivery semantics). An idle session stages and enqueues a turn right away.



87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'app/models/xeno/session.rb', line 87

def receive_message!(content)
  if turns.where(status: %w[pending running waiting]).exists?
    transaction do
      pending_messages.create!(payload: { "content" => content })
      emit("message.received", { content: content, queued: true })
    end
    drain_pending_messages! if reload.status == "waiting"
    nil
  else
    turn = stage_turn!(content)
    turn.enqueue!
    turn
  end
end

#settle_unanswered_tool_calls!(turn, reason:) ⇒ Object

Called before a turn dies with tool calls still unanswered (cancel while parked, schedule park failure): every dangling call gets a tool result in the transcript. Without this the assistant tool_call rows stay unanswered forever and every later generate is a provider 400 — the session is bricked. A completed action (e.g. an answered question whose resume never ran) injects its recorded output; everything else gets a denial, recorded on the action row as the audit trail.



184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
 
# File 'app/models/xeno/session.rb', line 184

def settle_unanswered_tool_calls!(turn, reason:)
  chat_record = Chat.find(chat_id)
  llm = chat_record.to_llm
  last_assistant = llm.messages.reverse.find { |m| m.role == :assistant }
  return unless last_assistant&.tool_call?

  answered = llm.messages.select { |m| m.role == :tool }.map(&:tool_call_id).compact.to_set
  last_assistant.tool_calls.values.reject { |call| answered.include?(call.id) }.each do |call|
    action = turn.actions.find_by(tool_call_id: call.id)
    if action&.completed?
      content = action.output&.fetch("content", nil).to_s
      status = "completed"
    else
      content = JSON.generate({ denied: true, reason: reason })
      status = "denied"
    end

    transaction do
      action.update!(status: "denied", output: { "content" => content }) if action && status == "denied"
      chat_record.add_message(role: :tool, content: content, tool_call_id: call.id)
    end
    emit("action.result", { turn_id: turn.id, call_id: call.id, tool: call.name, status: status })
  end
end

#stage_compaction_turn!(reason:, used: nil, limit: nil) ⇒ Object

Stages a compaction turn (summarize-and-replace, executed by the runner under a normal claim). It queues behind any active or parked turn via the ordinary session-ordering machinery — the queued-behind-the-active-turn semantics — and never appends a synthetic user message.



163
164
165
166
167
168
169
170
171
 
# File 'app/models/xeno/session.rb', line 163

def stage_compaction_turn!(reason:, used: nil, limit: nil)
  transaction do
    request = { "reason" => reason, "used" => used, "limit" => limit }.compact
    turn = Turn.append!(self, kind: "compaction", user_message: { "compaction" => request })
    emit("compaction.requested", { "turn_id" => turn.id }.merge(request))
    emit("turn.started", { turn_id: turn.id, sequence: turn.sequence })
    turn
  end
end

#stage_turn!(messages, definition: Xeno.definition, emit_received: true, defer_transcript: false) ⇒ Object

Stages the next turn for one or more messages: persists each user message into the transcript, refreshes instructions (each turn picks up edited instructions without a restart), and appends the turn row. Runs in the caller's process so TurnJob replays never double-add the user message.

defer_transcript: the turn row carries the messages but nothing is written into the transcript yet — the runner stages it when the turn is claimed. Required whenever an EARLIER turn may still be mid-flight (drained turns): appending user rows behind a parked turn's unanswered tool calls would wedge every later generate (providers demand tool results immediately after the assistant's tool_calls message).



61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'app/models/xeno/session.rb', line 61

def stage_turn!(messages, definition: Xeno.definition, emit_received: true, defer_transcript: false)
  contents = Array(messages)
  transaction do
    unless defer_transcript
      # Finding K: with_instructions triggers the first to_llm build; a
      # fresh chat record must get its runtime options back first.
      chat.apply_runtime_options!(definition.config.model_options)
      resolved = definition.instructions_for(session: self)
      chat.with_instructions(resolved) if resolved
      contents.each { |content| chat.add_message(role: :user, content: content) }
    end
    turn = Turn.append!(self, user_message: { contents: contents }, transcript_deferred: defer_transcript)
    if emit_received
      contents.each { |content| emit("message.received", { content: content, turn_id: turn.id }) }
    end
    emit("turn.started", { turn_id: turn.id, sequence: turn.sequence })
    turn
  end
end

#state ⇒ Object

The session-scoped KV store (see Xeno::SessionState).



210
211
212
 
# File 'app/models/xeno/session.rb', line 210

def state
  @state_handle ||= SessionState.new(self)
end

#steer!(content, definition: Xeno.definition) ⇒ Object

Steering (opt-in per message): stop what the agent is doing and make THIS message the next turn. The active turn dies safely — a parked or pending turn settles its dangling tool calls (recorded answers injected, unapproved gates denied — the H1 machinery) and cancels immediately; a running turn cancels cooperatively (the steer message queues and the runner's cancel path folds it into the next turn at the next step boundary). If the cancel loses the race with a completing turn, steering degrades to a normal follow-up — never an error.



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'app/models/xeno/session.rb', line 130

def steer!(content, definition: Xeno.definition)
  turn = turns.where(status: %w[pending running waiting]).order(:sequence).first

  case turn&.status
  when nil # idle: steer is just a message
    staged = stage_turn!(content, definition: definition)
    staged.enqueue!
    staged
  when "running"
    transaction do
      pending_messages.create!(payload: { "content" => content })
      emit("message.received", { content: content, queued: true, steer: true })
    end
    chat.cancel! # cooperative; the runner settles, cancels, and drains
    nil
  else # pending or waiting — no process owns it; replace it now
    transaction do
      settle_unanswered_tool_calls!(turn, reason: "steered by user")
      turn.update!(status: "cancelled")
      emit("turn.cancelled", { turn_id: turn.id, steer: true })
      update!(status: "running") if status == "waiting"
    end
    staged = stage_turn!(content, definition: definition)
    staged.enqueue!
    staged
  end
end