Class: SolidLoop::LlmCompletionJob

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

Constant Summary collapse

READ_TIMEOUT =

10 minutes

600
EVENTS_SAVE_BEFORE_REQUEST =

two phase event save, good for slow models debugging

true
HTTP_STREAMING_UPDATE_MODEL_INTERVAL =

seconds - update model content every N seconds

2

Instance Method Summary collapse

Instance Method Details

#perform(loop_id) ⇒ Object 



9
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
 
# File 'app/jobs/solid_loop/llm_completion_job.rb', line 9

def perform(loop_id)
  execution_token = SecureRandom.uuid

  # Derive the lease from the agent's own read_timeout (+ margin) so it is
  # guaranteed to outlive the HTTP client — a healthy in-flight turn can then
  # never be reclaimed by the reaper. Resolved before the CAS so the same
  # atomic transition to `running` also stamps `lease_expires_at`.
  loop_for_lease = SolidLoop::Loop.find(loop_id)
  begin
    agent_for_lease = loop_for_lease.agent
  rescue NameError => e
    # A renamed/missing agent class: fail the still-`queued` loop
    # VISIBLY instead of raising forever on every retry. No token/lease is
    # held yet, so this is a plain queued→failed transition.
    Rails.logger.error "LlmCompletionJob: cannot resolve agent for loop #{loop_id}: #{e.class}: #{e.message}"
    # from: :queued only — a missing agent fails BEFORE any CAS claim, so no
    # worker can ever hold this loop in `running`; fencing to :queued keeps
    # this consistent with every other transition and can't stomp a sibling.
    loop_for_lease.transition_status(
      from: :queued,
      to: :failed,
      error_message: "#{e.class}: #{e.message}",
      execution_token: nil,
      lease_expires_at: nil
    )
    return
  end
  return unless agent_for_lease # nil agent_class_name: nothing to run

  lease_duration   = SolidLoop.config.llm_lease_duration(agent_for_lease)
  lease_expires_at = Time.current + lease_duration

  claimed = SolidLoop::Loop
    .where(id: loop_id, status: :queued)
    .update_all(
      status: :running,
      execution_token: execution_token,
      lease_expires_at: lease_expires_at,
      updated_at: Time.current
    )
  return if claimed == 0

  loop_record = nil

  # Reconciliation wrapper: wrap the ENTIRE post-claim job body —
  # the Context build, the second `agent` resolution, middleware config, and
  # the whole pipeline — so a non-graceful error (a NoMethodError from a
  # renamed agent, a bug in a middleware) fails the loop VISIBLY instead of
  # stranding it `running`. Hard kills still fall through to the reaper. The
  # inner ErrorHandling middleware still terminalizes its graceful allowlist
  # first; this only catches what escapes it.
  begin
    # Heartbeat the lease for the ENTIRE claimed interval.
    # The heartbeat wrapper opens IMMEDIATELY after the successful claim, so
    # it covers the CAS→`Context.new`→second-`agent`→config window too — making
    # "the entire claimed interval is heartbeated" literally true (previously
    # Context.new/agent/config ran BEFORE the heartbeat started, an unheartbeated
    # gap). It covers MCP init (AgentInitialization), the whole stream incl.
    # silent prefill (NetworkCalling), and finalization. `read_timeout` is an
    # inactivity timeout, not a total-duration bound, so the static claim-time
    # lease can expire on a HEALTHY worker; the heartbeat renews it at ~ttl/4
    # keyed on loop_id + execution_token, and flips a lost-lease flag if the
    # generation rotated away. The heartbeat still stops in `run`'s ensure
    # BEFORE any canonical completion.
    SolidLoop::LeaseHeartbeat.run(
      loop_id: loop_id,
      execution_token: execution_token,
      lease_duration: lease_duration,
      # The agent's own legal turn bound sizes the renewer's leak ceiling
      # (`max_duration + lease_leak_grace`), so a legitimate turn — capped at
      # max_duration by the agent's contract — is never dropped, while a genuine
      # leak self-expires shortly after max_duration.
      max_duration: agent_for_lease.max_duration,
      logger: Rails.logger
    ) do |heartbeat|
      context = SolidLoop::Pipeline::Context.new(loop_id: loop_id, execution_token: execution_token)
      loop_record = context.loop
      agent = loop_record.agent
      context.heartbeat = heartbeat

      builder = SolidLoop::Pipeline::Builder.new(SolidLoop.llm_middlewares.middlewares)
      agent.configure_llm_middlewares(builder)

      pipeline = SolidLoop::Pipeline.new(builder.middlewares)
      pipeline.call(context)
    end
  rescue SolidLoop::LostLease => e
    # The heartbeat lost the lease mid-turn (generation rotated away or the
    # reaper reclaimed us). We DON'T own the loop, so we write nothing and do
    # NOT fail it — recovery belongs to whoever now owns the generation (the
    # resume, or the reaper's re-enqueue). Swallow so ActiveJob doesn't retry.
    Rails.logger.info "LlmCompletionJob: lease lost for loop #{loop_id}, yielding turn: #{e.message}"
  rescue StandardError => e
    # If we owned the loop we reconciled it to `failed` (durable) — swallow so
    # ActiveJob does not retry endlessly (the CAS would no-op anyway). If we
    # did NOT own it (a concurrent pause/stop/reclaim already rotated the
    # token), re-raise so the genuinely-unhandled error stays visible.
    raise unless reconcile_failure!(loop_record, execution_token, e)
  end
end