Module: Legion::Extensions::Llm::Inventory::WeightReconciler

Defined in:
lib/legion/extensions/llm/inventory/weight_reconciler.rb

Overview

Shared atomic write-time weight publication for existing writer cadences.

Class Method Summary collapse

Class Method Details

.activate_tracked!(settings:, instance_id:, state_key:, state:, **activation) ⇒ Object

Initial and recovery activation both use this helper after readiness succeeds. Rebuild from current settings and publish under the writer's publication mutex, then mark the state published only after the publisher call succeeds.

Raises:

  • (ArgumentError) 


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
 
# File 'lib/legion/extensions/llm/inventory/weight_reconciler.rb', line 96

def activate_tracked!(settings:, instance_id:, state_key:, state:, **activation)
  unknown = activation.keys - %i[states mutex probe_token activate activation_sequence stable_signature]
  raise ArgumentError, "unknown activation keyword(s): #{unknown.join(', ')}" unless unknown.empty?

  states = activation.fetch(:states)
  mutex = activation.fetch(:mutex)
  probe_token = activation.fetch(:probe_token)
  activate = activation.fetch(:activate)
  activation_sequence = activation.fetch(:activation_sequence)
  stable_signature = activation[:stable_signature]
  mutex.synchronize do
    return false unless states[state_key].equal?(state)

    offerings = rebuild_offerings(
      settings: settings, instance_key: state.fetch(:instance_key),
      offerings: state.fetch(:offerings)
    )
    sequence = activation_sequence.call(state)
    activate.call(
      instance_id: instance_id, state: state, offerings: offerings,
      sequence: sequence, probe_token: probe_token
    )
    state[:sequence] = sequence
    cache_locked!(
      state: state, offerings: offerings, stable_signature: stable_signature
    )
    state[:published] = true
    true
  end
end

.commit_if_changed!(settings:, instance_id:, state:, discovered_offerings:, **publication) ⇒ Object

Periodic discovery calls this AFTER its existing network/discovery work. Read current Settings and perform weight rebuilding, comparison, sequence allocation, publish, and cache update under the same writer mutex. No Settings callback or separate reweight path exists.

Raises:

  • (ArgumentError) 


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
 
# File 'lib/legion/extensions/llm/inventory/weight_reconciler.rb', line 53

def commit_if_changed!(settings:, instance_id:, state:, discovered_offerings:, **publication)
  unknown = publication.keys - %i[mutex equivalent replace stable_signature]
  raise ArgumentError, "unknown publication keyword(s): #{unknown.join(', ')}" unless unknown.empty?

  mutex = publication.fetch(:mutex)
  equivalent = publication.fetch(:equivalent)
  replace = publication.fetch(:replace)
  stable_signature = publication[:stable_signature]
  mutex.synchronize do
    offerings = rebuild_offerings(
      settings: settings, instance_key: state.fetch(:instance_key),
      offerings: discovered_offerings
    )
    return false if equivalent.call(state.fetch(:offerings), offerings)

    if state.fetch(:published)
      commit_locked!(
        instance_id: instance_id, state: state, offerings: offerings,
        replace: replace, stable_signature: stable_signature
      )
    else
      cache_locked!(
        state: state, offerings: offerings, stable_signature: stable_signature
      )
    end
    true
  end
end

.configured_weight_keys(settings:, provider_family:) ⇒ Object 



169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
 
# File 'lib/legion/extensions/llm/inventory/weight_reconciler.rb', line 169

def configured_weight_keys(settings:, provider_family:)
  llm = config_hash(settings.dig(:extensions, :llm))
  provider = config_hash(SettingsCascade.lookup(llm, provider_family))
  keys = Set.new
  keys << canonical_key(provider_family, :provider) if weight_present?(provider)

  config_hash(SettingsCascade.lookup(provider, :models)).each do |model, config|
    keys << canonical_key(provider_family, :model, model.to_s) if weight_present?(config_hash(config))
  end
  config_hash(SettingsCascade.lookup(provider, :instances)).each do |instance, config|
    instance_config = config_hash(config)
    keys << canonical_key(provider_family, :instance, instance.to_s) if weight_present?(instance_config)
    config_hash(SettingsCascade.lookup(instance_config, :models)).each do |model, model_config|
      next unless weight_present?(config_hash(model_config))

      keys << canonical_key(
        provider_family, :instance, instance.to_s, :model, model.to_s
      )
    end
  end
  config_hash(SettingsCascade.lookup(provider, :offerings)).each do |offering_id, config|
    keys << canonical_key(provider_family, :offering, offering_id.to_s) \
      if weight_present?(config_hash(config))
  end
  keys
end

.observe_dormant!(settings:, provider_family:, states:, mutex:, **observation) ⇒ Object

Ordinary ticks call this after reconcile. Dormant observation has the same cadence as the existing writer; there is no reload callback path.

Raises:

  • (ArgumentError) 


129
130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/legion/extensions/llm/inventory/weight_reconciler.rb', line 129

def observe_dormant!(settings:, provider_family:, states:, mutex:, **observation)
  unknown = observation.keys - %i[tracker dormant_logger]
  raise ArgumentError, "unknown observation keyword(s): #{unknown.join(', ')}" unless unknown.empty?

  tracker = observation.fetch(:tracker)
  dormant_logger = observation.fetch(:dormant_logger)
  mutex.synchronize do
    observe_dormant_locked!(
      settings: settings, provider_family: provider_family, states: states,
      tracker: tracker, dormant_logger: dormant_logger
    )
  end
end

.published_weight_keys(provider_family:, states:) ⇒ Object 



196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/legion/extensions/llm/inventory/weight_reconciler.rb', line 196

def published_weight_keys(provider_family:, states:)
  keys = Set.new
  states.each_value do |state|
    next unless state.fetch(:published)

    offerings = Array(state[:offerings])
    next if offerings.empty?

    instance_key = state.fetch(:instance_key)
    keys << canonical_key(provider_family, :provider)
    keys << canonical_key(provider_family, :instance, instance_key.instance_id.to_s)
    offerings.each do |draft|
      keys << canonical_key(provider_family, :model, draft.model.to_s)
      keys << canonical_key(
        provider_family, :instance, instance_key.instance_id.to_s,
        :model, draft.model.to_s
      )
      draft.operation_evidence.each_value do |evidence|
        next unless evidence.supported?

        lane_id = Identity.compose_lane_id(
          tier: draft.tier, provider_family: instance_key.provider_family,
          instance_id: instance_key.instance_id,
          type: Taxonomies.lane_type_for(operation: evidence.operation), model: draft.model
        )
        keys << canonical_key(provider_family, :offering, lane_id)
      end
    end
  end
  keys
end

.rebuild_offerings(settings:, instance_key:, offerings:) ⇒ Object 



36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/legion/extensions/llm/inventory/weight_reconciler.rb', line 36

def rebuild_offerings(settings:, instance_key:, offerings:)
  offerings.map do |draft|
    inputs = WeightSchema.weight_inputs(
      settings: settings,
      instance_key: instance_key,
      model: draft.model,
      tier: draft.tier,
      operation_evidence: draft.operation_evidence
    )
    draft.with(weight_inputs: inputs, base_weight: WeightSchema.base_weight(inputs))
  end.freeze
end

.track_initializing!(states:, state_key:, state:, mutex:) ⇒ Object

Track a claimed but not-yet-activated instance before its readiness I/O. An ordinary writer pass may refresh its cached weights, but must never send a replacement for a publication that is still :initializing.



85
86
87
88
89
90
91
 
# File 'lib/legion/extensions/llm/inventory/weight_reconciler.rb', line 85

def track_initializing!(states:, state_key:, state:, mutex:)
  mutex.synchronize do
    state[:published] = false
    states[state_key] = state
  end
  state
end