Class: Xeno::Turn

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

Overview

One user message and all work until the agent responds, run as one ActiveJob. Correctness lives in these rows, not in the queue:

  1. Atomic claim (CAS on claim_token): one worker owns the turn.
  2. Heartbeat: a stale heartbeat lets the next attempt reclaim; checkpoint replay makes takeover safe.
  3. Fencing token: every write is conditioned on claim_token = mine, so a reaped zombie affects zero rows.
  4. Max attempts: poison turns fail instead of retrying forever.

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.append!(session, user_message:, transcript_deferred: false, kind: "message") ⇒ Object

Appends the next turn with a race-safely assigned sequence. The INSERT runs in its own savepoint because stage_turn! calls this inside a transaction, and on Postgres a failed INSERT would otherwise abort the whole transaction.



25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'app/models/xeno/turn.rb', line 25

def self.append!(session, user_message:, transcript_deferred: false, kind: "message")
  attempts = 0
  begin
    sequence = (where(session_id: session.id).maximum(:sequence) || 0) + 1
    transaction(requires_new: true) do
      create!(session: session, sequence: sequence, user_message: user_message,
              transcript_deferred: transcript_deferred, kind: kind)
    end
  rescue ActiveRecord::RecordNotUnique
    attempts += 1
    retry if attempts < 5
    raise
  end
end

Instance Method Details

#claim! ⇒ Object

The atomic claim. Returns the fencing token, or nil when another worker owns the turn or it isn't claimable. Claimable: pending, or running with a stale heartbeat. Poison turns fail here instead of retrying forever.

attempts counts failures, not claims: a stale-running reclaim is crash evidence and counts; a transient error counts in release_for_retry!; an approval or question resume counts in resumes, never against the poison ladder.



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'app/models/xeno/turn.rb', line 51

def claim!
  current = self.class.find(id)

  if current.attempts >= Xeno.config.max_turn_attempts
    poisoned = self.class.where(id: id, claim_token: current.claim_token)
                   .where.not(status: %w[completed failed cancelled])
                   .update_all(status: "failed", error: { "message" => "max attempts (#{current.attempts}) exhausted" })
    if poisoned == 1
      # A poisoned TURN leaves the SESSION running (idle, can take the next message); settle any
      # dangling tool calls so it stays usable.
      session.settle_unanswered_tool_calls!(self, reason: "turn failed (max attempts exhausted)")
      session.emit("turn.failed", { turn_id: id, error: "max attempts exhausted" })
    end
    return nil
  end

  stale_before = Xeno.config.turn_stale_after.ago
  earlier_active = self.class
    .where(session_id: session_id)
    .where("sequence < ?", sequence)
    .active

  claimed = self.class
    .where(id: id, claim_token: current.claim_token)
    .where("status = 'pending' OR (status = 'running' AND (heartbeat_at IS NULL OR heartbeat_at < ?))", stale_before)
    .where.not(earlier_active.arel.exists) # turns run in session order
    .update_all([
      "status = 'running', claim_token = ?, " \
      "attempts = attempts + (CASE WHEN status = 'running' THEN 1 ELSE 0 END), " \
      "heartbeat_at = ?",
      current.claim_token + 1, Time.current
    ])

  claimed == 1 ? current.claim_token + 1 : nil
end

#enqueue! ⇒ Object 



40
41
42
 
# File 'app/models/xeno/turn.rb', line 40

def enqueue!
  TurnJob.perform_later(id)
end

#fenced_update!(token, attributes) ⇒ Object

Raises:



103
104
105
106
107
108
 
# File 'app/models/xeno/turn.rb', line 103

def fenced_update!(token, attributes)
  rows = self.class.where(id: id, claim_token: token).update_all(attributes)
  raise Xeno::Fenced, "turn #{id}: claim #{token} was fenced out" unless rows == 1

  true
end

#heartbeat!(token) ⇒ Object

Fenced write: touches the heartbeat iff we still own the claim. Raises Xeno::Fenced when a zombie discovers it was reaped.



99
100
101
 
# File 'app/models/xeno/turn.rb', line 99

def heartbeat!(token)
  fenced_update!(token, heartbeat_at: Time.current)
end

#record_step!(token) ⇒ Object 



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

def record_step!(token)
  current = self.class.find(id)
  if current.steps_count >= Xeno.config.max_steps
    raise Xeno::MaxStepsExceeded, "turn #{id}: exceeded #{Xeno.config.max_steps} steps"
  end

  fenced_update!(token, steps_count: current.steps_count + 1, heartbeat_at: Time.current)
  current.steps_count + 1
end

#release_for_retry!(token, error) ⇒ Object

Transient failure: record the error, count the failure against the poison ladder, and hand the claim back (status pending) so the queue retry can claim immediately.



89
90
91
92
93
94
95
 
# File 'app/models/xeno/turn.rb', line 89

def release_for_retry!(token, error)
  current = self.class.find(id)
  fenced_update!(token,
    status: "pending",
    attempts: current.attempts + 1,
    error: { "class" => error.class.name, "message" => error.message })
end