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 executes a fleet request against the captured registry callable. Protocol v3: exact execution only — one dispatch path (06 W1/W3), no provider-object topology.

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) 


304
305
306
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 304

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

.available_record!(snapshot, instance_key) ⇒ Object

Raises:



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

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:) ⇒ Object

The single dispatch entry (06 W1).



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 33

def call(envelope:, registry:)
  raise PolicyError, 'WorkerExecution.call requires registry' if registry.nil?

  envelope = FleetEnvelope.wrap(envelope)
  claims = nil
  idempotency_key = nil
  claims = validate_identity!(envelope)
  validate_policy!(envelope)
  idempotency_key = validate_idempotency!(envelope)
  response = dispatch!(envelope:, registry:)
  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:) ⇒ Object

The one dispatcher (06 W3): the exact resolution chain (W2). The claim's offering_id value is the lane's 5 tuple (field name kept for protocol continuity, D4); the claim resolves a lane exactly. The REQUESTED operation (envelope) is what gets dispatched — the lane constrains type/model/instance, the operation is a request property.



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 90

def dispatch!(envelope:, registry:)
  snapshot = registry.snapshot
  instance_key = IDENTITY::InstanceKey.new(
    provider_family: envelope.provider, instance_id: envelope.provider_instance
  )
  record = available_record!(snapshot, instance_key)
  operation = exact_operation(envelope)
  lane = exact_lane!(snapshot, record, envelope, operation)
  lease = registry.acquire(callable_handle: lane.callable_handle)
  begin
    dispatch_operation(lease.callable, operation, lane.model, exact_params(envelope))
  ensure
    lease.release
  end
end

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

Raises:



220
221
222
223
224
225
226
227
228
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 220

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

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

.dispatch_image(callable, model, params) ⇒ Object 



214
215
216
217
218
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 214

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

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



186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 186

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

.dispatch_speak(callable, model, params) ⇒ Object 



230
231
232
233
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 230

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

.evict_oldest_idempotency_entries! ⇒ Object 



274
275
276
277
278
279
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 274

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_lane!(snapshot, record, envelope, operation) ⇒ Object

The exact lane the claim names: the 5-tuple claim value must parse and validate, resolve in the snapshot, and agree with the activated instance, the envelope's operation type, and the envelope's model. The operation is a request property matched against the lane type — the lane's operation member is the representative of its type, not the identity.

Raises:



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

def exact_lane!(snapshot, record, envelope, operation)
  IDENTITY.validate_lane_id!(value: envelope.offering_id)
  lane = snapshot.lane(lane_id: envelope.offering_id)
  raise ERRORS::ExactOfferingMismatchError, 'lane_id not on the activated instance' if lane.nil?
  unless lane.instance_key == record.instance_key && lane.callable_handle.equal?(record.callable_handle)
    raise ERRORS::ExactOfferingMismatchError, 'lane does not belong to the activated instance'
  end

  require_matching_operation!(lane, operation)
  require_matching_model!(lane, envelope)
  lane
end

.exact_operation(envelope) ⇒ Object

06 §5: an unknown operation is a contract error at the worker.



133
134
135
136
137
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 133

def exact_operation(envelope)
  Legion::Extensions::Llm::Taxonomies.normalize_operation(value: envelope.operation)
rescue Legion::Extensions::Llm::Inventory::Errors::ValidationError
  raise ContractError, "unsupported exact operation: #{envelope.operation}"
end

.exact_params(envelope) ⇒ Object

W5 params strictness: no :model key (the Selection-derived model is untouchable), no duplicate param spellings. Shape violations are contract errors (06 §5), not offering mismatches.

Raises:



156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 156

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

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

  params
end

.idempotency_ttl_seconds ⇒ Object 



291
292
293
294
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 291

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

.mark_idempotency_success!(key) ⇒ Object 



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

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

.purge_idempotency_cache! ⇒ Object 



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

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

.rehydrate_wire_messages(value) ⇒ Object

W4 — the named rehydration boundary: the ONLY place serialized wire messages become Canonical::Message. Non-Hash/non-Message elements are contract errors.



173
174
175
176
177
178
179
180
181
182
183
184
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 173

def rehydrate_wire_messages(value)
  Array(value).map do |message|
    next message if message.is_a?(Canonical::Message)

    unless message.is_a?(Hash)
      raise ContractError,
            "fleet wire message must be a serialized Canonical::Message, got #{message.class}"
    end

    Canonical::Message.from_hash(message)
  end
end

.release_idempotency!(key) ⇒ Object 



252
253
254
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 252

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

.release_replay!(claims) ⇒ Object 



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

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

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

.require_matching_model!(lane, envelope) ⇒ Object

Raises:



146
147
148
149
150
151
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 146

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

  model
end

.require_matching_operation!(lane, operation) ⇒ Object

Raises:



139
140
141
142
143
144
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 139

def require_matching_operation!(lane, operation)
  lane_type = Legion::Extensions::Llm::Taxonomies.lane_type_for(operation: lane.operation)
  return if Legion::Extensions::Llm::Taxonomies.lane_type_for(operation: operation) == lane_type

  raise ERRORS::ExactOfferingMismatchError, "operation #{operation} does not match the claimed lane type"
end

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

Raises:



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

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

  params[key]
end

.reserve_idempotency_key!(key) ⇒ Object 



281
282
283
284
285
286
287
288
289
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 281

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 



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

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

.responder_setting(key, default:) ⇒ Object 



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

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

.validate_idempotency!(envelope) ⇒ Object

Raises:



72
73
74
75
76
77
78
79
80
 
# File 'lib/legion/extensions/llm/fleet/worker_execution.rb', line 72

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

  key = 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 



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

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

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

.validate_policy!(_envelope) ⇒ Object

W6 fail-closed: require_policy with no policy engine configured RAISES. The v2 warn-and-allow is deleted.

Raises:



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

def validate_policy!(_envelope)
  return true unless responder_setting(:require_policy, default: false)

  raise PolicyError, 'require_policy is enabled but no policy engine is configured'
end