Module: Legion::Extensions::Llm::Fleet::WorkerExecution

Extended by:
Logging::Helper
Includes:
Logging::Helper
Defined in:
lib/legion/extensions/llm/fleet/worker_execution.rb

Overview

Applies responder-side policy and dispatches a fleet request to a local lex-llm provider.

Defined Under Namespace

Classes: PolicyError

Constant Summary collapse

ERRORS = 
Legion::Extensions::Llm::Inventory::Errors
IDENTITY = 
Legion::Extensions::Llm::Inventory::Identity
MAX_IDEMPOTENCY_ENTRIES = 
100_000

Class Method Summary collapse

Class Method Details

.auth_required? ⇒ Boolean

Returns:

  • (Boolean) 


325
326
327
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 325

def auth_required?
  Settings.value(:fleet, :auth, :require_signed_token, default: true) != false
end

.available_record!(snapshot, instance_key) ⇒ Object

Raises:



164
165
166
167
168
169
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 164

def available_record!(snapshot, instance_key)
  record = snapshot.instance(instance_key: instance_key)
  raise ERRORS::ExactOfferingMismatchError, 'instance is absent, initializing, or unavailable' unless record && record.availability.state == :available

  record
end

.call(envelope:, registry: nil, provider: nil) ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 29

def call(envelope:, registry: nil, provider: nil)
  validate_dispatch_target!(registry, provider)
  claims = nil
  idempotency_key = nil
  claims = validate_identity!(envelope)
  validate_policy!(envelope)
  idempotency_key = validate_idempotency!(envelope)
  response = dispatch!(envelope: envelope, registry: registry, provider: provider)
  mark_idempotency_success!(idempotency_key) if idempotency_key
  TokenValidator.mark_replay!(claims[:jti]) if claims.is_a?(Hash)
  response
rescue TokenError => e
  handle_exception(e, level: :warn, handled: false, operation: 'llm.fleet.worker_execution.identity')
  release_idempotency!(idempotency_key) if idempotency_key
  release_replay!(claims)
  raise PolicyError, e.message
rescue StandardError => e
  handle_exception(e, level: :warn, handled: false, operation: 'llm.fleet.worker_execution.call')
  release_idempotency!(idempotency_key) if idempotency_key
  release_replay!(claims)
  raise
end

.dispatch!(envelope:, registry:, provider:) ⇒ Object 



105
106
107
108
109
110
111
112
113
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 105

def dispatch!(envelope:, registry:, provider:)
  return dispatch_local_provider!(envelope: envelope, provider: provider) if provider

  if envelope_value(envelope, :execution_contract) == Protocol::EXACT_EXECUTION_CONTRACT
    exact_dispatch!(envelope: envelope, registry: registry)
  else
    legacy_registry_dispatch!(envelope: envelope, registry: registry)
  end
end

.dispatch_audio(callable, operation, model, params) ⇒ Object

Raises:



233
234
235
236
237
238
239
240
241
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 233

def dispatch_audio(callable, operation, model, params)
  require_param!(params, :audio_file, operation)
  raise ERRORS::ExactOfferingMismatchError, "#{operation} requires the language key" unless params.key?(:language)

  callable.public_send(
    operation, params[:audio_file], model: model, language: params[:language],
                                    **except(params, :audio_file, :model, :language)
  )
end

.dispatch_image(callable, model, params) ⇒ Object 



227
228
229
230
231
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 227

def dispatch_image(callable, model, params)
  require_param!(params, :prompt, :image)
  require_param!(params, :size, :image)
  callable.image(prompt: params[:prompt], model: model, **except(params, :prompt))
end

.dispatch_local_provider!(envelope:, provider:) ⇒ Object 



75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 75

def dispatch_local_provider!(envelope:, provider:)
  provider = provider.call(envelope) if provider.respond_to?(:call) && !provider.respond_to?(:chat)
  operation = envelope_value(envelope, :operation).to_sym
  params = normalize_hash(envelope_value(envelope, :params) || {})
  params = unpack_legacy_options(params)
  model = envelope_value(envelope, :model)

  case operation
  when :chat
    provider.chat(messages: params.fetch(:messages, []), model: model, **except(params, :messages))
  when :stream
    provider.stream_chat(messages: params.fetch(:messages, []), model: model, **except(params, :messages))
  when :embed
    provider.embed(text: params[:text], model: model, **except(params, :text))
  when :count_tokens
    provider.count_tokens(messages: params.fetch(:messages, []), model: model, **except(params, :messages))
  else
    raise PolicyError, "unsupported fleet operation: #{operation}"
  end
end

.dispatch_operation(callable, operation, model, params) ⇒ Object 



212
213
214
215
216
217
218
219
220
221
222
223
224
225
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 212

def dispatch_operation(callable, operation, model, params)
  case operation
  when :chat then callable.chat(messages: require_param!(params, :messages, operation), model: model, **except(params, :messages))
  when :stream_chat then callable.stream_chat(messages: require_param!(params, :messages, operation), model: model, **except(params, :messages))
  when :count_tokens then callable.count_tokens(messages: require_param!(params, :messages, operation), model: model, **except(params, :messages))
  when :embed then callable.embed(text: require_param!(params, :text, operation), model: model, **except(params, :text))
  when :image then dispatch_image(callable, model, params)
  when :transcribe then dispatch_audio(callable, :transcribe, model, params)
  when :translate then dispatch_audio(callable, :translate, model, params)
  when :speak then dispatch_speak(callable, model, params)
  when :moderate then callable.moderate(require_param!(params, :input, operation), model: model, **except(params, :input))
  else raise ERRORS::ExactOfferingMismatchError, "unsupported exact operation: #{operation}"
  end
end

.dispatch_speak(callable, model, params) ⇒ Object 



243
244
245
246
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 243

def dispatch_speak(callable, model, params)
  require_param!(params, :text, :speak)
  callable.speak(params[:text], model: model, voice: params[:voice], **except(params, :text, :model, :voice))
end

.envelope_value(envelope, key) ⇒ Object 



329
330
331
332
333
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 329

def envelope_value(envelope, key)
  return nil unless envelope.respond_to?(:key?)

  envelope[key] || envelope[key.to_s]
end

.evict_oldest_idempotency_entries! ⇒ Object 



295
296
297
298
299
300
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 295

def evict_oldest_idempotency_entries!
  sorted = @idempotency_keys.each_pair.sort_by { |_key, entry| entry[:expires_at] }
  sorted.first(@idempotency_keys.size - MAX_IDEMPOTENCY_ENTRIES).each_key do |key|
    @idempotency_keys.delete(key)
  end
end

.exact_dispatch!(envelope:, registry:) ⇒ Object

Exact path: resolve by signed offering_id; no model resolution, provider scan, :default, or first value.

Raises:



117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 117

def exact_dispatch!(envelope:, registry:)
  snapshot = registry.snapshot
  instance_key = exact_instance_key(envelope)
  record = available_record!(snapshot, instance_key)
  offering = record.offerings_by_id[envelope_value(envelope, :offering_id)]
  raise ERRORS::ExactOfferingMismatchError, 'offering_id not on the activated instance' if offering.nil?

  operation = exact_operation(envelope)
  model = require_matching_model!(offering, envelope)
  require_supported!(offering, operation)
  execute_via_lane(registry, snapshot,
                   { record: record, offering: offering, operation: operation, model: model, envelope: envelope })
end

.exact_instance_key(envelope) ⇒ Object 



158
159
160
161
162
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 158

def exact_instance_key(envelope)
  IDENTITY::InstanceKey.new(
    provider_family: envelope_value(envelope, :provider), instance_id: envelope_value(envelope, :provider_instance)
  )
end

.exact_operation(envelope) ⇒ Object 



171
172
173
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 171

def exact_operation(envelope)
  Legion::Extensions::Llm::Taxonomies.normalize_operation(value: envelope_value(envelope, :operation), allow_aliases: false)
end

.exact_params(envelope) ⇒ Object

Raises:



198
199
200
201
202
203
204
205
206
207
208
209
210
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 198

def exact_params(envelope)
  raw = envelope_value(envelope, :params) || {}
  params = {}
  raw.each do |key, value|
    sym = key.respond_to?(:to_sym) ? key.to_sym : key
    raise ERRORS::ExactOfferingMismatchError, "duplicate param spelling for #{sym}" if params.key?(sym)

    params[sym] = value
  end
  raise ERRORS::ExactOfferingMismatchError, 'params must not contain model' if params.key?(:model)

  params
end

.except(hash, *keys) ⇒ Object 



343
344
345
346
347
348
349
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 343

def except(hash, *keys)
  exclusions = keys.map(&:to_sym)
  hash.each_with_object({}) do |(key, value), result|
    normalized_key = key.respond_to?(:to_sym) ? key.to_sym : key
    result[normalized_key] = value unless exclusions.include?(normalized_key)
  end
end

.execute_via_lane(registry, snapshot, resolution) ⇒ Object 



148
149
150
151
152
153
154
155
156
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 148

def execute_via_lane(registry, snapshot, resolution)
  lane = matching_lane!(snapshot, resolution[:record], resolution[:offering], resolution[:operation], resolution[:model])
  lease = registry.acquire(callable_handle: lane.callable_handle)
  begin
    dispatch_operation(lease.callable, resolution[:operation], resolution[:model], exact_params(resolution[:envelope]))
  ensure
    lease.release
  end
end

.idempotency_ttl_seconds ⇒ Object 



312
313
314
315
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 312

def idempotency_ttl_seconds
  ttl = responder_setting(:idempotency_ttl_seconds, default: 600).to_i
  ttl.positive? ? ttl : 600
end

.legacy_registry_dispatch!(envelope:, registry:) ⇒ Object

Registry-backed v2 compatibility path for a migrated provider: execute only when (provider_family, instance, operation, model) resolves to exactly one supported local offering.

Raises:



134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 134

def legacy_registry_dispatch!(envelope:, registry:)
  snapshot = registry.snapshot
  instance_key = exact_instance_key(envelope)
  record = available_record!(snapshot, instance_key)
  operation = Legion::Extensions::Llm::Taxonomies.normalize_operation(value: envelope_value(envelope, :operation), allow_aliases: true)
  model = IDENTITY.normalize_text(value: envelope_value(envelope, :model), field: :model)
  matches = record.offerings_by_id.values.select { |o| o.model == model && o.operation_status(operation: operation) == :supported }
  raise ERRORS::ExactOfferingMismatchError, 'no matching local offering' if matches.empty?
  raise ERRORS::AmbiguousLegacyOfferingError, 'multiple matching local offerings' if matches.size > 1

  execute_via_lane(registry, snapshot,
                   { record: record, offering: matches.first, operation: operation, model: model, envelope: envelope })
end

.mark_idempotency_success!(key) ⇒ Object 



267
268
269
270
271
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 267

def mark_idempotency_success!(key)
  @idempotency_mutex.synchronize do
    @idempotency_keys[key.to_s] = { state: :complete, expires_at: Time.now.to_i + idempotency_ttl_seconds }
  end
end

.matching_lane!(snapshot, record, offering, operation, model) ⇒ Object

Raises:



188
189
190
191
192
193
194
195
196
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 188

def matching_lane!(snapshot, record, offering, operation, model)
  lane_id = IDENTITY.lane_id(instance_key: record.instance_key, operation: operation, model: model, offering_id: offering.offering_id)
  lane = snapshot.lane(lane_id: lane_id)
  valid = lane && lane.offering_id == offering.offering_id && lane.instance_key == record.instance_key &&
          lane.model == model && lane.operation == operation && lane.callable_handle.equal?(record.callable_handle)
  raise ERRORS::ExactOfferingMismatchError, 'no matching lane for the offering' unless valid

  lane
end

.normalize_hash(hash) ⇒ Object 



335
336
337
338
339
340
341
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 335

def normalize_hash(hash)
  return {} unless hash.respond_to?(:each)

  hash.each_with_object({}) do |(key, value), result|
    result[key.respond_to?(:to_sym) ? key.to_sym : key] = value
  end
end

.purge_idempotency_cache! ⇒ Object 



285
286
287
288
289
290
291
292
293
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 285

def purge_idempotency_cache!
  @idempotency_mutex.synchronize do
    now = Time.now.to_i
    @idempotency_keys.each_pair do |key, entry|
      @idempotency_keys.delete(key) if entry[:expires_at] <= now
    end
    evict_oldest_idempotency_entries! if @idempotency_keys.size > MAX_IDEMPOTENCY_ENTRIES
  end
end

.release_idempotency!(key) ⇒ Object 



273
274
275
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 273

def release_idempotency!(key)
  @idempotency_mutex.synchronize { @idempotency_keys.delete(key.to_s) }
end

.release_replay!(claims) ⇒ Object 



277
278
279
280
281
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 277

def release_replay!(claims)
  return unless claims.is_a?(Hash) && claims[:jti]

  TokenValidator.release_replay!(claims[:jti])
end

.require_matching_model!(offering, envelope) ⇒ Object

Raises:



175
176
177
178
179
180
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 175

def require_matching_model!(offering, envelope)
  model = IDENTITY.normalize_text(value: envelope_value(envelope, :model), field: :model)
  raise ERRORS::ExactOfferingMismatchError, 'model does not match the offering' unless offering.model == model

  model
end

.require_param!(params, key, operation) ⇒ Object

Raises:



248
249
250
251
252
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 248

def require_param!(params, key, operation)
  raise ERRORS::ExactOfferingMismatchError, "#{operation} requires the #{key} param" unless params.key?(key)

  params[key]
end

.require_supported!(offering, operation) ⇒ Object

Raises:



182
183
184
185
186
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 182

def require_supported!(offering, operation)
  return if offering.operation_status(operation: operation) == :supported

  raise ERRORS::ExactOfferingMismatchError, "operation #{operation} is not supported by the offering"
end

.reserve_idempotency_key!(key) ⇒ Object 



302
303
304
305
306
307
308
309
310
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 302

def reserve_idempotency_key!(key)
  @idempotency_mutex.synchronize do
    now = Time.now.to_i
    existing = @idempotency_keys[key]
    raise PolicyError, 'duplicate fleet idempotency key' if existing && existing[:expires_at] > now

    @idempotency_keys[key] = { state: :inflight, expires_at: now + idempotency_ttl_seconds }
  end
end

.reset_idempotency_cache! ⇒ Object 



262
263
264
265
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 262

def reset_idempotency_cache!
  @idempotency_keys = Concurrent::Map.new
  @idempotency_mutex = Mutex.new
end

.responder_setting(key, default:) ⇒ Object 



317
318
319
320
321
322
323
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 317

def responder_setting(key, default:)
  value = Settings.value(:fleet, :responder, key, default: nil)
  return auth_required? if key == :require_auth && value.nil?
  return default if value.nil?

  value
end

.unpack_legacy_options(params) ⇒ Object 



254
255
256
257
258
259
260
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 254

def unpack_legacy_options(params)
  options = params.delete(:options)
  return params unless options.is_a?(Hash)

  normalize_hash(options).each { |key, value| params[key] = value unless params.key?(key) }
  params
end

.validate_dispatch_target!(registry, provider) ⇒ Object

Raises:



99
100
101
102
103
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 99

def validate_dispatch_target!(registry, provider)
  return unless registry.nil? == provider.nil?

  raise PolicyError, 'WorkerExecution.call requires exactly one of registry or provider'
end

.validate_idempotency!(envelope) ⇒ Object

Raises:



65
66
67
68
69
70
71
72
73
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 65

def validate_idempotency!(envelope)
  return nil unless responder_setting(:require_idempotency, default: true)

  key = envelope_value(envelope, :idempotency_key)
  raise PolicyError, 'fleet idempotency_key is required' if key.to_s.empty?

  reserve_idempotency_key!(key.to_s)
  key.to_s
end

.validate_identity!(envelope) ⇒ Object 



52
53
54
55
56
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 52

def validate_identity!(envelope)
  return true unless responder_setting(:require_auth, default: true)

  TokenValidator.validate!(token: envelope_value(envelope, :signed_token), envelope: envelope)
end

.validate_policy!(_envelope) ⇒ Object

rubocop:disable Naming/PredicateMethod



58
59
60
61
62
63
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 58

def validate_policy!(_envelope) # rubocop:disable Naming/PredicateMethod
  return true unless responder_setting(:require_policy, default: false)

  log.warn('[fleet] require_policy is enabled but no policy engine is configured — allowing request')
  true
end