Module: Legion::Extensions::Llm::Discovery::Pipeline

Includes:
Helpers::Lex
Defined in:
lib/legion/extensions/llm/discovery/pipeline.rb

Overview

The shared discovery pipeline for EVERY lex-llm-* provider — the write half of the inventory interface. A provider mixes it into its own <Provider>::Runners::Discovery module and overrides ONLY the genuinely provider-specific methods (fetch_raw_models, check_health, build_offering_draft, build_callable, and — if its config keys differ — catalog_base_url / auth_token / derive_physical_id). Everything else (reconcile, claim, activate, probe, replace, weight publication, health display, dormant weight tracking) is inherited.

module Vllm
module Runners
  module Discovery
    extend self
    include Legion::Extensions::Llm::Discovery::Pipeline
    def fetch_raw_models(instance_cfg:) = ...
    def build_offering_draft(...)       = ...
    def build_callable(instance_cfg:)   = ...
  end
end
end

It is a STATELESS module in the LegionIO sense: it is not itself a live runner and it lives OUTSIDE the framework-scanned runners/ dir (a base in that dir is claimed by the builder as a live per-extension runner). Each provider's <Provider>::Runners::Discovery is the live, scanned runner; it is that module object which carries the per-instance working state (states — the claim handles + last built drafts + publish progress), one per provider. That state is the writer's working state, NOT a second inventory: the Inventory::Registry remains the single source of truth.

The base actor (Discovery::Actor) fires on the discovery interval and dispatches runner_class.refresh each tick (manual, in-process — the state and the Registry are both process-local, so the pipeline must run on the owning node, not as a remote Runner task).

WEIGHT IS NOT COMPUTED HERE. Drafts are built identity-weighted; Inventory::WeightReconciler recomputes the write-time weight from live settings at publish (commit/activate). One weight owner, no duplication.

Defined Under Namespace

Classes: CatalogFetchFailure

Instance Method Summary collapse

Instance Method Details

#activate_tracked_state(instance_id:, state:, probe_token:) ⇒ Object 



433
434
435
436
437
438
439
440
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 433

def activate_tracked_state(instance_id:, state:, probe_token:)
  Legion::Extensions::Llm::Inventory::WeightReconciler.activate_tracked!(
    settings: Legion::Settings, instance_id: instance_id, state_key: state.fetch(:name),
    state: state, states: states, mutex: state_mutex, probe_token: probe_token,
    activate: method(:activate_weight_snapshot),
    activation_sequence: ->(tracked) { tracked.fetch(:sequence) }
  )
end

#activate_weight_snapshot(instance_id:, state:, offerings:, sequence:, probe_token:) ⇒ Object 



426
427
428
429
430
431
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 426

def activate_weight_snapshot(instance_id:, state:, offerings:, sequence:, probe_token:)
  publisher.activate_instance_snapshot(
    instance_id: instance_id, publisher_token: state.fetch(:publisher_token),
    offerings: offerings, sequence: sequence, probe_token: probe_token
  )
end

#auth_token(instance_cfg:) ⇒ Object 



149
150
151
152
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 149

def auth_token(instance_cfg:)
  token = instance_cfg.dig(:credentials, :api_key) || instance_cfg[:api_key]
  token if token.is_a?(String) && !token.strip.empty?
end

#build_callable(instance_cfg:) ⇒ Object

Build the provider's inference callable captured into the registry. Abstract: a provider MUST implement how it executes inference.

Raises:

  • (NotImplementedError) 


137
138
139
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 137

def build_callable(instance_cfg:)
  raise NotImplementedError, "#{name} must implement #build_callable"
end

#build_offering_draft(instance_cfg:, instance_key:, model_id:, model_data:) ⇒ Object

Build the Inventory::OfferingDraft (evidence + metadata) for one model. NO weight — the reconciler computes it at publish. Abstract: a provider MUST implement its capability/operation evidence knowledge.

Raises:

  • (NotImplementedError) 


131
132
133
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 131

def build_offering_draft(instance_cfg:, instance_key:, model_id:, model_data:)
  raise NotImplementedError, "#{name} must implement #build_offering_draft"
end

#build_offerings(instance_cfg:, instance_key:) ⇒ Object

── Offering build (weight-free draft; reconciler weights at publish) ─



452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 452

def build_offerings(instance_cfg:, instance_key:)
  models = fetch_raw_models(instance_cfg: instance_cfg)
  models.filter_map do |model_data|
    model_id = model_id_from(model_data)
    next if model_id.empty?

    build_offering_draft(instance_cfg: instance_cfg, instance_key: instance_key, model_id: model_id, model_data: model_data)
  end
rescue StandardError => e
  raise e if discovery_programming_error?(e)
  raise e if e.is_a?(CatalogFetchFailure)

  handle_exception(e, level: :warn, handled: false, operation: "#{provider_family}.runner.discovery.build_offerings")
  raise CatalogFetchFailure, "catalog build failed (#{e.class.name})", cause: e
end

#build_probe_enqueue(instance_id:) ⇒ Object 



396
397
398
399
400
401
402
403
404
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 396

def build_probe_enqueue(instance_id:)
  proc do |request:|
    handle_reactive_probe(instance_id: instance_id, request: request)
    true
  rescue StandardError => e
    handle_exception(e, level: :warn, operation: "#{provider_family}.runner.discovery.probe_enqueue", instance_id: instance_id)
    false
  end
end

#catalog_base_url(instance_cfg:) ⇒ Object 



145
146
147
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 145

def catalog_base_url(instance_cfg:)
  normalize_api_base(instance_cfg[:base_url] || instance_cfg[:endpoint])
end

#check_health(instance_cfg:) ⇒ Object

Return an Inventory::ReadinessResult. Default is the OpenAI-compatible GET /health. A provider with a different readiness probe overrides it.



111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 111

def check_health(instance_cfg:)
  conn = build_connection(base_url: catalog_base_url(instance_cfg: instance_cfg), instance_cfg: instance_cfg, timeout: 5, open_timeout: 3)
  response = conn.get(health_path)
  Legion::Extensions::Llm::Inventory::ReadinessResult.new(
    ready: response.status == 200, reason: "#{health_path} returned #{response.status}",
    metadata: { status: response.status }
  )
rescue Faraday::ConnectionFailed => e
  handle_exception(e, level: :warn, handled: true, operation: "#{provider_family}.runner.discovery.health")
  readiness_failure(error: e)
rescue StandardError => e
  raise e if discovery_programming_error?(e)

  handle_exception(e, level: :warn, handled: true, operation: "#{provider_family}.runner.discovery.health")
  readiness_failure(error: e)
end

#claim_and_activate_instance(name:, instance_cfg:) ⇒ Object 



260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 260

def claim_and_activate_instance(name:, instance_cfg:)
  instance_id = name.to_s
  instance_key = Legion::Extensions::Llm::Inventory::Identity::InstanceKey.new(
    provider_family: provider_family, instance_id: instance_id,
    physical_id: derive_physical_id(instance_cfg: instance_cfg)
  )
  begin
    offerings = build_offerings(instance_cfg: instance_cfg, instance_key: instance_key)
  rescue CatalogFetchFailure
    log.debug { "#{provider_family} discovery: catalog fetch failed at claim for #{instance_id} — deferring readiness" }
    offerings = nil
  end
  callable = build_callable(instance_cfg: instance_cfg)
  probe_coordinator = Legion::Extensions::Llm::Inventory::ProbeCoordinator.new(
    instance_key: instance_key, enqueue: build_probe_enqueue(instance_id: instance_id)
  )
  publisher_token = publisher.claim_instance(
    instance_id: instance_id, physical_id: instance_key.physical_id,
    callable: callable, probe_request_handle: probe_coordinator
  )
  state = {
    name: instance_id, instance_key: instance_key, instance_cfg: instance_cfg,
    callable: callable, probe_coordinator: probe_coordinator,
    publisher_token: publisher_token, sequence: 0, offerings: offerings || []
  }
  Legion::Extensions::Llm::Inventory::WeightReconciler.track_initializing!(
    states: states, state_key: instance_id, state: state, mutex: state_mutex
  )
  perform_readiness(instance_id: instance_id, state: state, offerings: offerings) if offerings
end

#clear_settings_health(name:) ⇒ Object 



494
495
496
497
498
499
500
501
502
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 494

def clear_settings_health(name:)
  instance_settings = settings.dig(:instances, name.to_sym)
  return unless instance_settings.is_a?(Hash)

  instance_settings.delete(:health)
  instance_settings.delete(:capabilities)
rescue StandardError => e
  handle_exception(e, level: :warn, operation: "#{provider_family}.runner.discovery.clear_health", instance_name: name.to_s)
end

#derive_physical_id(instance_cfg:) ⇒ Object

── Physical identity (secondary; dedup/diagnostics only) ───────────



516
517
518
519
520
521
522
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 516

def derive_physical_id(instance_cfg:)
  host_port = extract_host_port(url: catalog_base_url(instance_cfg: instance_cfg))
  token = auth_token(instance_cfg: instance_cfg)
  return "#{host_port}/ak:#{::Digest::SHA256.hexdigest(token)[0, 6]}" if token

  host_port
end

#discover_instances ⇒ Object

The provider module's public instance catalog — the single source of configured instances, shared with the dispatch path (legion-llm Call::Providers reads the same method). A provider that defines no discover_instances does not populate the inventory.



197
198
199
200
201
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 197

def discover_instances
  return {} unless provider_module.respond_to?(:discover_instances)

  provider_module.discover_instances
end

#discovery_programming_error?(error) ⇒ Boolean

A programming bug in discovery must fail loud — swallowing it publishes ZERO offerings and makes an activated instance invisible.

Returns:

  • (Boolean) 


470
471
472
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 470

def discovery_programming_error?(error)
  error.is_a?(NameError) || error.is_a?(ArgumentError)
end

#dormant_weight_tracker ⇒ Object 



79
80
81
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 79

def dormant_weight_tracker
  @dormant_weight_tracker ||= Legion::Extensions::Llm::Inventory::DormantWeightTracker.new
end

#fetch_raw_models(instance_cfg:) ⇒ Object

Return an Array of raw model Hashes for this instance. Default is the OpenAI-compatible GET /v1/models -> body. A non-OpenAI provider overrides this entirely.

Raises:



94
95
96
97
98
99
100
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 94

def fetch_raw_models(instance_cfg:)
  conn = build_connection(base_url: catalog_base_url(instance_cfg: instance_cfg), instance_cfg: instance_cfg, timeout: 15, open_timeout: 5)
  response = conn.get(models_path)
  raise CatalogFetchFailure, "catalog fetch returned HTTP #{response.status}" unless response.status.between?(200, 299)

  Array(Legion::JSON.load(response.body).fetch(:data, []))
end

#handle_reactive_probe(instance_id:, request:) ⇒ Object 



376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 376

def handle_reactive_probe(instance_id:, request:)
  state = state_mutex.synchronize { states[instance_id] }
  return unless state

  coordinator = state[:probe_coordinator]
  return unless coordinator.begin_probe(request: request)

  probe_token = publisher.readiness_probe_started(instance_id: instance_id, publisher_token: state[:publisher_token])
  readiness = check_health(instance_cfg: state[:instance_cfg])
  coordinator.finish_probe(request: request)
  report_probe_result(instance_id: instance_id, state: state, probe_token: probe_token, readiness: readiness)
rescue StandardError => e
  begin
    coordinator.finish_probe(request: request)
  rescue StandardError => finish_e
    handle_exception(finish_e, level: :warn, handled: true, operation: "#{provider_family}.runner.discovery.reactive_probe.finish", instance_id: instance_id)
  end
  handle_exception(e, level: :warn, operation: "#{provider_family}.runner.discovery.reactive_probe", instance_id: instance_id)
end

#health_path ⇒ Object 



143
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 143

def health_path = '/health'

#model_id_from(model_data) ⇒ Object

The model's stable id within the catalog list. OpenAI-compatible /v1/models lists carry it under :id; Ollama's /api/tags carries it under :name.



105
106
107
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 105

def model_id_from(model_data)
  model_data[:id].to_s
end

#models_path ⇒ Object

OpenAI-compatible defaults — overridable per provider.



142
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 142

def models_path = '/v1/models'

#observe_dormant_weights ⇒ Object 



442
443
444
445
446
447
448
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 442

def observe_dormant_weights
  Legion::Extensions::Llm::Inventory::WeightReconciler.observe_dormant!(
    settings: Legion::Settings, provider_family: provider_family, states: states,
    mutex: state_mutex, tracker: dormant_weight_tracker,
    dormant_logger: ->(key) { log.info("[llm][#{provider_family}] action=dormant_weight weight_key=#{key.inspect} no_lane_published=true") }
  )
end

#offerings_equivalent?(previous, current) ⇒ Boolean

Returns:

  • (Boolean) 


474
475
476
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 474

def offerings_equivalent?(previous, current)
  offering_comparison_multiset(previous) == offering_comparison_multiset(current)
end

#perform_readiness(instance_id:, state:, offerings:) ⇒ Object 



291
292
293
294
295
296
297
298
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 291

def perform_readiness(instance_id:, state:, offerings:)
  reconcile_weight_snapshot(instance_id: instance_id, state: state, discovered_offerings: offerings)
  probe_token = publisher.readiness_probe_started(instance_id: instance_id, publisher_token: state[:publisher_token])
  readiness = check_health(instance_cfg: state[:instance_cfg])
  report_probe_result(instance_id: instance_id, state: state, probe_token: probe_token, readiness: readiness)
rescue StandardError => e
  handle_exception(e, level: :warn, operation: "#{provider_family}.runner.discovery.readiness", instance_id: instance_id)
end

#physical_id_changed?(state, instance_cfg) ⇒ Boolean

Returns:

  • (Boolean) 


234
235
236
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 234

def physical_id_changed?(state, instance_cfg)
  state[:instance_key].physical_id != derive_physical_id(instance_cfg: instance_cfg)
end

#provider_family ⇒ Object 



185
186
187
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 185

def provider_family
  @provider_family ||= provider_namespace.split('::').last.to_s.downcase.to_sym
end

#provider_module ⇒ Object 



189
190
191
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 189

def provider_module
  @provider_module ||= Kernel.const_get(provider_namespace)
end

#provider_namespace ⇒ Object

Resolved from the INCLUDING runner module's own name, because the pipeline is a mixed-in module: here self IS the provider's ::Runners::Discovery module (a Module), so its name is self.name, not self.class.name. "Legion::Extensions::Llm::Vllm::Runners::Discovery" -> "Legion::Extensions::Llm::Vllm"



181
182
183
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 181

def provider_namespace
  @provider_namespace ||= name.split('::')[0..-3].join('::')
end

#publisher ⇒ Object

── Claim / activate ────────────────────────────────────────────────



256
257
258
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 256

def publisher
  @publisher ||= Legion::Extensions::Llm::Inventory::Publisher.new(provider_family: provider_family)
end

#reconcile_instances ⇒ Object

── Reconcile (each tick) ───────────────────────────────────────────



205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 205

def reconcile_instances
  configured = discover_instances
  configured_ids = configured.keys.map(&:to_s)

  tracked_ids = state_mutex.synchronize { states.keys }
  tracked_ids.each do |instance_id|
    remove_instance_state(instance_id) unless configured_ids.include?(instance_id)
  end

  configured.each do |name, instance_cfg|
    update_instance(name: name.to_s, instance_cfg: instance_cfg)
  rescue StandardError => e
    handle_exception(e, level: :warn, operation: "#{provider_family}.runner.discovery.instance", instance_name: name.to_s)
  end
  observe_dormant_weights
end

#reconcile_weight_snapshot(instance_id:, state:, discovered_offerings:) ⇒ Object

── Weight publication (delegates to the shared reconciler) ─────────



408
409
410
411
412
413
414
415
416
417
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 408

def reconcile_weight_snapshot(instance_id:, state:, discovered_offerings:)
  Legion::Extensions::Llm::Inventory::WeightReconciler.commit_if_changed!(
    settings: Legion::Settings, instance_id: instance_id, state: state,
    discovered_offerings: discovered_offerings, mutex: state_mutex,
    equivalent: lambda do |previous, current|
      !states[instance_id].equal?(state) || offerings_equivalent?(previous, current)
    end,
    replace: method(:replace_weight_snapshot)
  )
end

#refresh ⇒ Object

── Entrypoint (called by the actor's manual dispatch) ───────────────



156
157
158
159
160
161
162
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 156

def refresh(**)
  reconcile_instances
  { success: true }
rescue StandardError => e
  handle_exception(e, level: :warn, operation: "#{provider_family}.runner.discovery.refresh")
  { success: false }
end

#refresh_instance(instance_id:, instance_cfg:) ⇒ Object

── Probe (refresh path) ────────────────────────────────────────────



332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 332

def refresh_instance(instance_id:, instance_cfg:)
  state = state_mutex.synchronize { states[instance_id] }
  return unless state

  status = publisher.snapshot.publication_status(instance_key: state[:instance_key])
  if status.state == :initializing
    begin
      offerings = build_offerings(instance_cfg: instance_cfg, instance_key: state[:instance_key])
    rescue CatalogFetchFailure
      log.debug { "#{provider_family} discovery: catalog fetch failed for #{instance_id} — staying :initializing" }
      return
    end
    perform_readiness(instance_id: instance_id, state: state, offerings: offerings)
    return
  end

  replace_if_changed(instance_id: instance_id, state: state, instance_cfg: instance_cfg)
  run_cadence_probe(instance_id: instance_id, state: state)
end

#remove_all_instances ⇒ Object 



164
165
166
167
168
169
170
171
172
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 164

def remove_all_instances(**)
  tracked = state_mutex.synchronize { states.keys }
  tracked.each { |instance_id| remove_instance_state(instance_id) }
  state_mutex.synchronize do
    states.clear
    dormant_weight_tracker.clear!
  end
  { success: true }
end

#remove_instance_state(instance_id) ⇒ Object 



238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 238

def remove_instance_state(instance_id)
  state = state_mutex.synchronize do
    tracked = states[instance_id]
    next unless tracked

    publisher.remove_instance(instance_id: instance_id, publisher_token: tracked[:publisher_token])
    states.delete(instance_id) if states[instance_id].equal?(tracked)
    tracked
  end
  return unless state

  clear_settings_health(name: state[:name])
rescue StandardError => e
  handle_exception(e, level: :warn, operation: "#{provider_family}.runner.discovery.remove_instance", instance_id: instance_id)
end

#replace_if_changed(instance_id:, state:, instance_cfg:) ⇒ Object 



352
353
354
355
356
357
358
359
360
361
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 352

def replace_if_changed(instance_id:, state:, instance_cfg:)
  begin
    new_offerings = build_offerings(instance_cfg: instance_cfg, instance_key: state[:instance_key])
  rescue CatalogFetchFailure
    log.debug { "#{provider_family} discovery: catalog fetch failed for #{instance_id} — keeping last snapshot" }
    return
  end
  changed = reconcile_weight_snapshot(instance_id: instance_id, state: state, discovered_offerings: new_offerings)
  write_instance_health(state) if changed
end

#replace_weight_snapshot(instance_id:, state:, offerings:, sequence:) ⇒ Object 



419
420
421
422
423
424
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 419

def replace_weight_snapshot(instance_id:, state:, offerings:, sequence:)
  publisher.replace_instance_snapshot(
    instance_id: instance_id, publisher_token: state.fetch(:publisher_token),
    offerings: offerings, sequence: sequence
  )
end

#report_probe_result(instance_id:, state:, probe_token:, readiness:) ⇒ Object 



300
301
302
303
304
305
306
307
308
309
310
311
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 300

def report_probe_result(instance_id:, state:, probe_token:, readiness:)
  committed = if readiness.ready? && tracked_unpublished?(instance_id: instance_id, state: state)
                activate_tracked_state(instance_id: instance_id, state: state, probe_token: probe_token)
              else
                report_tracked_readiness(instance_id: instance_id, state: state, probe_token: probe_token, readiness: readiness)
              end
  write_instance_health(state) if committed
  committed
rescue StandardError => e
  handle_exception(e, level: :warn, operation: "#{provider_family}.runner.discovery.report_probe", instance_id: instance_id)
  false
end

#report_tracked_readiness(instance_id:, state:, probe_token:, readiness:) ⇒ Object 



317
318
319
320
321
322
323
324
325
326
327
328
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 317

def report_tracked_readiness(instance_id:, state:, probe_token:, readiness:)
  state_mutex.synchronize do
    return false unless states[instance_id].equal?(state)

    if readiness.ready?
      publisher.readiness_succeeded(instance_id: instance_id, probe_token: probe_token)
    else
      publisher.readiness_failed(instance_id: instance_id, probe_token: probe_token, reason: readiness.reason)
    end
    true
  end
end

#reset_state! ⇒ Object

Test hook: drop this provider's working state for a fresh run.



84
85
86
87
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 84

def reset_state!
  @states = nil
  @dormant_weight_tracker = nil
end

#run_cadence_probe(instance_id:, state:) ⇒ Object 



363
364
365
366
367
368
369
370
371
372
373
374
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 363

def run_cadence_probe(instance_id:, state:)
  coordinator = state[:probe_coordinator]
  return unless coordinator.begin_probe

  probe_token = publisher.readiness_probe_started(instance_id: instance_id, publisher_token: state[:publisher_token])
  readiness = check_health(instance_cfg: state[:instance_cfg])
  coordinator.finish_probe
  report_probe_result(instance_id: instance_id, state: state, probe_token: probe_token, readiness: readiness)
rescue StandardError => e
  finish_probe_safely(coordinator)
  handle_exception(e, level: :warn, operation: "#{provider_family}.runner.discovery.cadence_probe", instance_id: instance_id)
end

#state_mutex ⇒ Object 



75
76
77
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 75

def state_mutex
  @state_mutex ||= Mutex.new
end

#states ⇒ Object

Per-provider working state, carried by the including runner module (one per provider). Lazy module-ivar accessors — a module has no constructor. The Inventory::Registry is the source of truth; this is only the writer's in-flight working state for this provider.



71
72
73
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 71

def states
  @states ||= Concurrent::Map.new
end

#tracked_unpublished?(instance_id:, state:) ⇒ Boolean

Returns:

  • (Boolean) 


313
314
315
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 313

def tracked_unpublished?(instance_id:, state:)
  state_mutex.synchronize { states[instance_id].equal?(state) && !state.fetch(:published) }
end

#union_capabilities(instance_key) ⇒ Object 



504
505
506
507
508
509
510
511
512
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 504

def union_capabilities(instance_key)
  capabilities = Set.new
  publisher.snapshot.lanes_for(instance_key: instance_key).each do |lane|
    lane.capability_evidence.each do |capability, evidence|
      capabilities << capability if evidence.supported?
    end
  end
  capabilities.to_a.sort
end

#update_instance(name:, instance_cfg:) ⇒ Object 



222
223
224
225
226
227
228
229
230
231
232
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 222

def update_instance(name:, instance_cfg:)
  state = state_mutex.synchronize { states[name] }
  if state && physical_id_changed?(state, instance_cfg)
    remove_instance_state(name)
    claim_and_activate_instance(name: name, instance_cfg: instance_cfg)
  elsif state
    refresh_instance(instance_id: name, instance_cfg: instance_cfg)
  else
    claim_and_activate_instance(name: name, instance_cfg: instance_cfg)
  end
end

#write_instance_health(state) ⇒ Object

── Health display (write-back to settings, post-commit) ────────────



480
481
482
483
484
485
486
487
488
489
490
491
492
 
# File 'lib/legion/extensions/llm/discovery/pipeline.rb', line 480

def write_instance_health(state)
  instance_key = state[:instance_key]
  snapshot = publisher.snapshot
  display = display_fact(instance_key: instance_key, snapshot: snapshot)

  instance_settings = settings.dig(:instances, state[:name].to_sym)
  return unless instance_settings.is_a?(Hash)

  instance_settings[:health] = health_hash(display)
  instance_settings[:capabilities] = union_capabilities(instance_key)
rescue StandardError => e
  handle_exception(e, level: :warn, operation: "#{provider_family}.runner.discovery.write_health", instance_id: state[:instance_key].instance_id)
end