Class: SolidLoop::Loop

Inherits:
ApplicationRecord show all
Defined in:
app/models/solid_loop/loop.rb

Constant Summary collapse

ACTIVE_STATUSES = 
%w[init queued running].freeze
STOPPABLE_STATUSES = 
(ACTIVE_STATUSES + [ "paused" ]).freeze
STEERABLE_STATUSES =

Steering — enqueue a user message onto a live loop, delivered to the model at the NEXT turn boundary.

A steering message is STAGED: role: user, steering: true, is_hidden: true, status: pending. Being hidden keeps it out of the in-flight turn's assembled conversation (MessageBuilding filters is_hidden: false); being steering

  • pending marks it as still-undelivered so the host app can show it as "pending" and so edit/cancel are allowed. MessageBuilding#build_messages materializes staged messages at the start of each turn (un-hide + mark status: success), which appends them AFTER the last resolved exchange and makes them immutable history. See drain_pending_steering!.

Positioning: the live conversation orders by COALESCE(conversation_order, id). An id alone is NOT enough — a steering message's id is frozen at ENQUEUE time, so a message enqueued mid-turn (after the assistant's tool_calls row is written but before its tool result rows are) sorts BETWEEN the assistant and its tool results, breaking the assistant/tool pairing invariant. Instead, delivery (see drain_pending_steering!) allocates a fresh, monotonic conversation_order value AT DELIVERY TIME, which sorts after the whole already-resolved batch. Ordinary messages leave conversation_order NULL and fall back to id unchanged.

%w[init queued running paused].freeze

Instance Method Summary collapse

Instance Method Details

#admin_frozen? ⇒ Boolean

Returns:

  • (Boolean) 


27
28
29
 
# File 'app/models/solid_loop/loop.rb', line 27

def admin_frozen?
  frozen_at.present?
end

#agent ⇒ Object 



10
11
12
13
 
# File 'app/models/solid_loop/loop.rb', line 10

def agent
  return nil unless agent_class_name.present?
  agent_class_name.constantize.new(self)
end

#cancel_pending_message(message_id) ⇒ Object

Cancel (delete) a still-pending steering message so it is never delivered. Rejected (raises) once delivered. Same fence as edit. Returns true.



160
161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'app/models/solid_loop/loop.rb', line 160

def cancel_pending_message(message_id)
  message = messages.find(message_id)

  message.with_lock do
    unless message.steering? && message.pending?
      raise SteeringError, "message #{message_id} is no longer pending; cannot cancel"
    end

    message.destroy!
  end

  true
end

#drain_pending_steering! ⇒ Object

Materialize the loop's staged steering messages into the live conversation. Called at the START of each turn's build_messages — a resolved boundary (the previous turn's tool results are all written; no half-open batch is created between here and assembly).

Exactly-once / crash-safety: each message is flipped pending -> success and un-hidden inside its own row lock, guarded by a pending? re-check. A retried turn-build (or a second concurrent build) re-reads the row as success and skips it — never re-materializing. A concurrent edit/cancel either won the lock first (message stays pending, gets materialized next) or lost it (sees success and is refused): no half-applied state.



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/solid_loop/loop.rb', line 185

def drain_pending_steering!
  # One committed, row-locked transaction. Lock the pending
  # steering rows FOR UPDATE in ID order and materialize them, then COMMIT —
  # all before the caller assembles the payload, so assembly reads a settled
  # conversation. Locking-and-reloading in one query closes the pluck-then-
  # find gap: a row cancelled/deleted between selection and lock simply is
  # not returned by the locking SELECT (tolerated). A concurrent edit/cancel
  # either won its own row lock first (row stays pending → materialized next
  # turn) or blocks here and then reads `success` and is refused.
  transaction do
    pending = pending_user_messages.lock("FOR UPDATE").to_a
    pending.each do |message|
      # Re-check under the held lock (defensive; the WHERE already filters).
      next unless message.steering? && message.pending?

      # Allocate the ordering value HERE, at delivery time — not the frozen
      # enqueue-time id — so the message sorts after every already-resolved
      # message written between enqueue and now (e.g. the tool results of a
      # batch the assistant opened mid-turn).
      message.update!(is_hidden: false, status: "success", conversation_order: next_conversation_order)
    end
  end
end

#edit_pending_message(message_id, content) ⇒ Object

Edit a still-pending steering message's content. Rejected (raises) once the message has been delivered (materialized) or was cancelled. The row lock + status re-check fences against a concurrent materialize: whoever takes the lock first wins; a message mid-/post-materialize reads status != pending and the edit is refused.



144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'app/models/solid_loop/loop.rb', line 144

def edit_pending_message(message_id, content)
  message = messages.find(message_id)

  message.with_lock do
    unless message.steering? && message.pending?
      raise SteeringError, "message #{message_id} is no longer pending; cannot edit"
    end

    message.update!(content: content)
  end

  message
end

#enqueue_user_message(content) ⇒ Object

Enqueue a free-text user message onto this loop. Returns the created (staged) Message, or raises if the loop is terminal.



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
 
# File 'app/models/solid_loop/loop.rb', line 118

def enqueue_user_message(content)
  with_lock do
    unless STEERABLE_STATUSES.include?(status)
      raise SteeringError, "cannot steer a #{status} loop"
    end

    messages.create!(
      role: "user",
      content: content,
      steering: true,
      is_hidden: true,
      status: "pending"
    )
  end
end

#pending_user_messages ⇒ Object

Still-pending (staged, not-yet-delivered) steering messages, in enqueue order.



135
136
137
 
# File 'app/models/solid_loop/loop.rb', line 135

def pending_user_messages
  messages.where(steering: true, status: "pending").order(:id)
end

#transition_status(from:, to:, expected_execution_token: :any, **attributes) ⇒ Object

CAS on status (and optionally the generation token) under the row lock. An optional block runs INSIDE the lock, after the checks pass and before the status write: use it for generation-owned side writes (message states, counters) so a losing CAS performs zero mutations.



35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'app/models/solid_loop/loop.rb', line 35

def transition_status(from:, to:, expected_execution_token: :any, **attributes)
  transitioned = false

  with_lock do
    next unless Array(from).map(&:to_s).include?(status)
    next if expected_execution_token != :any && execution_token != expected_execution_token

    yield if block_given?
    update!(**attributes, status: to)
    transitioned = true
  end

  transitioned
end

#unresolved_tool_calls ⇒ Object

Unresolved tool calls of the CURRENT conversation tail only. Recovery must never resurrect an obsolete tool turn, so selection starts from the newest visible user/assistant message and proceeds only when that tail is an assistant turn carrying tool calls.

TODO(backlog): resolution matches tool response messages by the tool_call_id string; a hard FK (+ unique index) from the tool response Message to its ToolCall row would make this join-able and typo-proof.



75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'app/models/solid_loop/loop.rb', line 75

def unresolved_tool_calls
  tail = messages
    .where(is_hidden: false, role: %i[user assistant])
    .order(id: :desc)
    .first
  return SolidLoop::ToolCall.none unless tail&.assistant?

  response_ids = messages
    .where(role: :tool)
    .where("id > ?", tail.id)
    .pluck(:tool_call_id)

  tail.tool_calls.ordered.where.not(tool_call_id: response_ids)
end

#with_generation(expected_execution_token) ⇒ Object

Fence for generation-owned writes that keep the loop running: takes the row lock, re-validates that the caller's generation still owns the loop, and only then yields. Returns true when the block ran; on a stale token (pause/stop/resume has moved the generation on) nothing is written.



54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'app/models/solid_loop/loop.rb', line 54

def with_generation(expected_execution_token)
  owned = false

  with_lock do
    next unless running? && execution_token == expected_execution_token

    yield
    owned = true
  end

  owned
end