Class: SolidLoop::ToolExecutionJob

Inherits:
ApplicationJob show all
Defined in:
app/jobs/solid_loop/tool_execution_job.rb

Instance Method Summary collapse

Instance Method Details

#perform(tool_call_id, execution_token = nil, reaper_lease_token = nil) ⇒ Object

reaper_lease_token (optional): when the reaper's case 2 re-enqueues a stalled tool call it FIRST CAS-claims the per-tool lease under the loop lock and hands its token here, so two racing reapers resolve to exactly one re-enqueue (the loser's CAS gets 0 rows). A normal dispatch omits it and the job self-claims a fresh lease below.



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
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/jobs/solid_loop/tool_execution_job.rb', line 10

def perform(tool_call_id, execution_token = nil, reaper_lease_token = nil)
  # A tool job must carry the generation token it was dispatched with.
  # A job without one (legacy/host-enqueued) is not part of any current
  # generation: fail closed instead of adopting the loop's token.
  return if execution_token.nil?

  # Pre-fetch basic objects to determine the agent class. The token fence
  # here is a plain generation check (no lease acquired yet); a stale job
  # no-ops silently — there is nothing to reconcile.
  tool_call = SolidLoop::ToolCall.find(tool_call_id)
  message = tool_call.message
  loop_model = message.loop
  loop_model.reload
  return unless loop_model.running? && loop_model.execution_token == execution_token

  # Reconciliation wrapper (findings 4/5/9): begin reconciliation immediately
  # after loading the loop/tool IDs and INCLUDE everything that used to sit
  # OUTSIDE the begin — agent resolution (a renamed/missing agent method),
  # sequential-mode evaluation, the lease claim aftermath, builder creation,
  # and `configure_tool_middlewares`. A non-graceful error at ANY of these
  # steps fails the loop VISIBLY under the fence instead of stranding it
  # `running` / reaping-and-retrying it forever. The context is built up
  # front (with lease_token nil until claimed) so a PRE-LEASE error still has
  # a generation-fenced failure path: `reconcile_failure!` transitions the
  # loop to failed under the execution_token even when no tool lease was held.
  # Hard kills (SIGKILL / an Exception outside StandardError) still fall
  # through to the reaper's case 2, which re-invokes with the SAME stable key.
  context = SolidLoop::Pipeline::ToolContext.new(tool_call_id: tool_call_id)
  context.tool_call = tool_call
  context.message = message
  context.loop = loop_model
  context.execution_token = execution_token
  context.lease_token = nil

  begin
    agent = loop_model.agent

    # Token fencing + the executed_at checkpoint keep every mode safe against
    # stale/duplicate jobs. The first-unresolved boundary is the *ordering*
    # invariant: it only applies when the agent opts into sequential
    # execution (parallel mode deliberately runs the turn's jobs in any
    # order). The check runs under the parent assistant message lock — the
    # same lock ResponseCreation resolves calls under — so it cannot race a
    # result landing.
    if agent.sequential_tool_calls?
      first_unresolved_id = nil
      message.with_lock do
        first_unresolved_id = loop_model.unresolved_tool_calls.first&.id
      end

      unless first_unresolved_id == tool_call.id
        # A stray/duplicate/out-of-order job must not stall the turn:
        # no-op, but re-dispatch the actual head so ordering self-heals. Route
        # through enqueue! so a silently-aborted re-dispatch surfaces (job
        # retries) instead of leaving the head-of-line call unrun.
        SolidLoop.enqueue!(self.class, first_unresolved_id, execution_token) if first_unresolved_id
        return
      end
    end

    # The per-tool_call lease (the SECOND fence token).
    # `execution_token` above is the generation fence (shared by every tool of
    # the turn); the `lease_token` here is per-tool, so the reaper can reclaim
    # ONE dead parallel tool without fencing its healthy siblings. The lease is
    # derived to strictly outlive the tool client timeout, so a healthy
    # long-running tool can never lose its lease to a reaper.
    lease_token =
      if tool_call.executed_at?
        # Already checkpointed but re-dispatched: the tool BODY must not run
        # again (executed_at is the terminal exactly-once guard). This path
        # exists so a call whose CONTINUATION (canonical response + successor)
        # was dropped — a crash after executed_at, reclaimed by reaper case 2 —
        # is completed: ToolExecution restores the checkpoint and ResponseCreation
        # commits the continuation (its own idempotent `response_message` check
        # makes a duplicate delivery harmless). No lease is needed or held.
        nil
      elsif reaper_lease_token
        # Reaper path: the lease was already CAS-claimed under the loop lock and
        # handed here. ADOPT it only if we STILL hold it (a worker that took over
        # after the reaper enqueued would have rotated it). No-op otherwise.
        return unless tool_call.lease_held_by?(reaper_lease_token)

        reaper_lease_token
      else
        # Normal dispatch: self-claim UNDER THE LOOP LOCK.
        #
        # SECONDARY race (clean-resume violation): the generation was validated
        # by the `reload` at the top of `perform`, but claiming the ToolCall
        # lease independently leaves a window in which a pause/resume can rotate
        # G1→G2 and revoke old leases BETWEEN the two. A stale worker would then
        # install a FRESH lease on the ToolCall without re-checking the loop,
        # and the legitimate G2 job could not claim until that lease expired (a
        # full tool-TTL delay before reaper case 2 repairs it). Fix: re-validate
        # `running + execution_token` under the loop row lock and claim the lease
        # in the SAME locked scope (loop → tool_call order). A rotation that
        # commits before we take the lock is now seen here → no-op; one that
        # arrives after blocks on the loop lock and rotates cleanly afterward.
        minted = claim_lease_under_loop_lock(loop_model, tool_call, execution_token, agent)
        if minted.nil?
          # Observability: the CAS minted nothing — either the
          # generation rotated away under us (stale worker no-op) or a DIFFERENT
          # live per-tool lease still holds this call (e.g. a host-side
          # paused→running resume that re-enqueued this job WITHOUT revoking the
          # gated tool_call's unexecuted lease — see SolidLoop::Base#resume!,
          # which NULLs lease_token/leased_until before re-enqueue). Without this
          # log the strand is invisible: no error, no event, the loop just waits
          # out the lease TTL / reaper. Reload so the logged lease reflects the
          # holder we lost to, not our stale in-memory copy — guarded so a
          # concurrent teardown of the row can't turn this pure-observability
          # path into a raise (a log must never change control flow).
          begin
            tool_call.reload
          rescue ActiveRecord::RecordNotFound
            nil
          end
          Rails.logger.warn(
            "ToolExecutionJob: tool lease claim lost — another live lease holds " \
            "this tool_call; job no-op (loop may strand until lease TTL/reaper). " \
            "tool_call_id=#{tool_call.id} loop_id=#{loop_model.id} " \
            "lease_token=#{tool_call.lease_token.inspect} " \
            "leased_until=#{tool_call.leased_until.inspect} " \
            "executed_at=#{tool_call.executed_at.inspect}"
          )
          return
        end

        minted
      end

    tool_call.reload
    context.tool_call = tool_call
    context.lease_token = lease_token

    builder = SolidLoop::Pipeline::Builder.new(SolidLoop.tool_middlewares.middlewares)
    agent.configure_tool_middlewares(builder)
    pipeline = SolidLoop::Pipeline.new(builder.middlewares)

    pipeline.call(context)
  rescue StandardError => e
    # If this generation (+ lease, when one was held) still owns the call under
    # the row lock we reconciled it to `failed` (durable) — swallow so ActiveJob
    # does not retry endlessly. If ownership was already rotated away
    # (pause/stop/reclaim), re-raise so a genuinely-unhandled error stays
    # visible. A pre-lease error takes the no-lease, generation-fenced path.
    raise unless reconcile_failure!(loop_model, tool_call, context, e)
  end
end