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) 


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

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) 


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

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 



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

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) 


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

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 



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

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
      )
      offering_id = Identity.offering_id(
        instance_key: instance_key, provider_native_key: draft.provider_native_key
      )
      keys << canonical_key(provider_family, :offering, offering_id.to_s)
    end
  end
  keys
end

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



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

def rebuild_offerings(settings:, instance_key:, offerings:)
  offerings.map do |draft|
    inputs = WeightSchema.weight_inputs(
      settings: settings,
      instance_key: instance_key,
      provider_native_key: draft.provider_native_key,
      model: draft.model,
      tier: draft.tier
    )
    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.



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

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