Module: Legion::LLM::Inference::Executor::Escalation

Included in:
Legion::LLM::Inference::Executor
Defined in:
lib/legion/llm/inference/executor/escalation.rb

Overview

Escalation-area methods extracted from Executor verbatim (P4b §1.5, refactor-under-green). Owns the provider-call lifecycle (single + escalating, sync + stream + responses-API), error/retry classification, and the corresponding audit/metering emission.

Instance Method Summary collapse

Instance Method Details

#attempts_exhausted_rejection ⇒ Object 



87
88
89
90
91
92
93
94
95
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 87

def attempts_exhausted_rejection
  Legion::Extensions::Llm::Routing::Rejection.new(
    kind:                 :attempts_exhausted,
    reason:               "maximum attempts (#{@routing_requirements.maximum_attempts}) reached",
    inventory_generation: Legion::Extensions::Llm::Inventory::Registry.snapshot.generation,
    candidate_counts:     {},
    http_status:          503
  )
end

#client_stream_error?(err) ⇒ Boolean

Detect client-side stream errors (disconnects, broken pipes, socket timeouts) that originate from writing back to the HTTP client, not from the provider itself. Puma::ConnectionError is a RuntimeError (NOT an IOError), so it slips past the StreamAssembler's rescue IOError/EPIPE guards and reaches the executor raw — the exact class the production logs show tripping the vLLM circuit. StreamClosed is the assembler's own wrapper raised once the client socket is confirmed dead.

Returns:

  • (Boolean) 


277
278
279
280
281
282
283
284
285
286
287
288
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 277

def client_stream_error?(err)
  name = err.class.name.to_s
  msg  = err.message.to_s
  name.include?('Puma::ConnectionError') ||
    name.include?('StreamAssembler::StreamClosed') ||
    name.include?('Errno::EPIPE') ||
    (name.include?('IOError') && msg.include?('closed')) ||
    (name.include?('IOError') && msg.include?('already closed')) ||
    name.include?('EOFError') ||
    name.include?('Errno::ECONNRESET') ||
    name.include?('Errno::ECONNABORTED')
end

#execute_provider_request ⇒ Object

Raises:



235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 235

def execute_provider_request
  @timestamps[:provider_start] = Time.now
  @timeline.record(
    category: :provider, key: 'provider:request_sent',
    exchange_id: @exchange_id, direction: :outbound,
    detail: "calling #{@resolved_provider}",
    from: 'pipeline', to: "provider:#{@resolved_provider}"
  )

  raise Legion::LLM::ProviderError, "Native provider not registered: #{@resolved_provider}" unless fleet_dispatch? || use_native_dispatch?(@resolved_provider)

  execute_provider_request_native

  @timestamps[:provider_end] = Time.now
  record_provider_response
end

#execute_provider_request_native ⇒ Object 



252
253
254
255
256
257
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 252

def execute_provider_request_native
  result = execute_native_tool_loop
  merge_response_offering_metadata(result.metadata) if result.respond_to?(:metadata)
  @raw_response = result
  @tool_loop_messages = @last_tool_loop_messages if @last_tool_loop_messages
end

#execute_provider_request_stream ⇒ Object

Raises:



329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 329

def execute_provider_request_stream(&)
  @timestamps[:provider_start] = Time.now
  @timeline.record(
    category: :provider, key: 'provider:request_sent',
    exchange_id: @exchange_id, direction: :outbound,
    detail: "streaming from #{@resolved_provider}",
    from: 'pipeline', to: "provider:#{@resolved_provider}"
  )

  raise Legion::LLM::ProviderError, "Native provider not registered: #{@resolved_provider}" unless fleet_dispatch? || use_native_dispatch?(@resolved_provider)

  execute_provider_request_stream_native(&)

  @timestamps[:provider_end] = Time.now
  record_provider_response
end

#execute_provider_request_stream_native ⇒ Object 



346
347
348
349
350
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 346

def execute_provider_request_stream_native(&)
  result = execute_native_streaming_tool_loop(&)
  merge_response_offering_metadata(result.metadata) if result.respond_to?(:metadata)
  @raw_response = result
end

#internal_error?(err) ⇒ Boolean

G25 / B-H / PR #152 C5/C6: Internal errors (daemon NoMethodError/ArgumentError) come from shared daemon code — retrying on a different lane guarantees the same crash. Classified as terminal: raise immediately, never retry, never trip circuits, never push to tried_lanes.

Returns:

  • (Boolean) 


293
294
295
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 293

def internal_error?(err)
  err.is_a?(::NoMethodError) || err.is_a?(::ArgumentError) || err.is_a?(::NotImplementedError)
end

#non_provider_failure?(err) ⇒ Boolean

The circuit breaker answers exactly one question: is the upstream LLM PROVIDER itself broken/down? These three families are NOT provider failures and must never trip a circuit, report provider health, or escalate to another lane:

(1) client-side write/disconnect (client socket died) — client_stream_error?
(2) SSE assembly / canonical parse / translation (LegionIO's own bugs)
(3) daemon/programming errors (NoMethodError/ArgumentError) — internal_error?

Provider-agnostic: matches on exception family, never on provider name.

Returns:

  • (Boolean) 


315
316
317
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 315

def non_provider_failure?(err)
  client_stream_error?(err) || sse_translation_error?(err) || internal_error?(err)
end

#populate_ssot_v3_resolved_state(attempt_context) ⇒ Object

Populate resolved-state ivars from an AttemptContext so existing emit_* calls (metering / prompt-audit / tool-audit) see the correct provider/instance/model/tier/offering_id for this attempt.



23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 23

def populate_ssot_v3_resolved_state(attempt_context)
  sel = attempt_context.selection
  @resolved_provider     = sel.provider_family.to_sym
  @resolved_instance     = sel.instance_id.to_sym
  @resolved_model        = sel.model
  @resolved_tier         = attempt_context.lane.tier
  @resolved_offering_id  = sel.offering_id
  @resolved_offering_metadata = {}
  @current_attempt_context = attempt_context
  log.debug "[llm][executor] action=ssot_v3_resolved provider=#{@resolved_provider} " \
            "instance=#{@resolved_instance} model=#{@resolved_model}"
end

#record_provider_response ⇒ Object 



259
260
261
262
263
264
265
266
267
268
269
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 259

def record_provider_response
  duration_ms = ((@timestamps[:provider_end] - @timestamps[:provider_start]) * 1000).to_i
  log.debug("[pipeline][provider] action=response_received provider=#{@resolved_provider} model=#{@resolved_model} duration_ms=#{duration_ms}")
  @timeline.record(
    category: :provider, key: 'provider:response_received',
    exchange_id: @exchange_id, direction: :inbound,
    detail: 'response received',
    from: "provider:#{@resolved_provider}", to: 'pipeline',
    duration_ms: duration_ms
  )
end

#run_provider_call_engine ⇒ Object

SSOT v3 single-attempt sync path. Selects via RoutingSession, delegates dispatch+error-handling to run_provider_call_single (preserves existing ProviderError → 529/502 behavior), classifies success. SSOT v3 single engine (sync). One request-scoped RoutingSession owns selection, one-and-done consumed-attempt identity, and retry. Each attempt selects the exact provider+instance+model via Router.next_lane, dispatches that EXACT callable (SelectionDispatch, via ssot_v3_direct_dispatch), and classifies. A retriable outcome selects the next eligible lane (the failed identity can never reappear); a normalized instance_unavailable also marks that exact instance unavailable (probe-cleared recovery — the original-incident fix). A terminal outcome or attempt exhaustion raises RoutingRejected, which the maintained route maps to the dialect HTTP status. No legacy fallback, no HealthTracker mutation.



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 50

def run_provider_call_engine
  @routing_session = Legion::LLM::Inference::RoutingSession.new(
    request: @request, requirements: @routing_requirements
  )
  @routing_requirements.maximum_attempts.times do
    attempt = @routing_session.next_attempt!(
      snapshot: Legion::Extensions::Llm::Inventory::Registry.snapshot
    )
    populate_ssot_v3_resolved_state(attempt)
    result = ssot_v3_execute_attempt
    action = @routing_session.classify(dispatch_result: result, attempt_context: attempt)
    return if action.disposition == :success
    raise Legion::LLM::Errors::RoutingRejected.new(rejection: action.rejection) if action.disposition == :terminal
  end
  raise Legion::LLM::Errors::RoutingRejected.new(rejection: attempts_exhausted_rejection)
ensure
  @current_attempt_context = nil
end

#run_provider_call_ssot_v3_stream ⇒ Object

SSOT v3 §19 streaming dispatch + post-first-byte failover. Reuses the preflight RoutingSession/AttemptContext when present; otherwise selects inline (direct call_stream callers that did not preflight). On a retriable provider failure it preserves the existing StreamAssembler failover sequence: classify → retain consumed target + add justified exclusions → provider_failover_pending!(from:) → strip cross-provider thinking → next_attempt with a fresh snapshot → begin_dispatch_on(lane:) → continue the SAME client SSE session (no replay, no custom switch event). Terminal outcomes and exhaustion re-raise so the route emits the dialect terminal SSE error (headers are already committed).



146
147
148
149
150
151
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 146

def run_provider_call_ssot_v3_stream(&)
  session, attempt_context = ssot_v3_stream_session_and_attempt
  run_provider_call_ssot_v3_stream_loop(session: session, attempt_context: attempt_context, &)
ensure
  @current_attempt_context = nil
end

#run_provider_call_ssot_v3_stream_loop(session:, attempt_context:) ⇒ Object

Bounded failover loop (no loop do/retry). RoutingSession bounds the attempt count: next_attempt returns an attempts_exhausted Rejection once requirements.maximum_attempts distinct targets are consumed.



171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 171

def run_provider_call_ssot_v3_stream_loop(session:, attempt_context:, &)
  current = attempt_context
  while current
    populate_ssot_v3_resolved_state(current)
    begin
      execute_provider_request_stream(&)
      session.classify(
        dispatch_result: Legion::LLM::Call::SelectionDispatch::Result.success(value: @raw_response),
        attempt_context: current
      )
      return
    rescue StandardError => e
      current = ssot_v3_stream_handle_failure(error: e, session: session, attempt_context: current)
    end
  end
end

#sse_translation_error?(err) ⇒ Boolean

SSE assembly / canonical parse / translation errors originate inside LegionIO's own stream-assembly and translation layer, not from the upstream provider. Like daemon/programming errors, they must never trip a provider circuit or escalate to another lane — the upstream is healthy; the bug is ours. Matched by class name so this stays provider-agnostic (N×N invariant) and does not couple to lex-llm gems.

Returns:

  • (Boolean) 


302
303
304
305
306
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 302

def sse_translation_error?(err)
  name = err.class.name.to_s
  name.include?('JSON::ParseError') ||
    name.include?('JSON::ParserError')
end

#ssot_v3_execute_attempt ⇒ Object

Run one selected attempt through the exact callable. Returns a Phase 1 SelectionDispatch::Result (success value, or a normalized non-success ProviderOutcome). Client-write/disconnect and daemon/programming errors are NOT provider failures and propagate untouched (terminal).



73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 73

def ssot_v3_execute_attempt
  execute_provider_request
  Legion::LLM::Call::SelectionDispatch::Result.success(value: @raw_response)
rescue StandardError => e
  raise e if non_provider_failure?(e)

  outcome = @last_ssot_dispatch_outcome
  @last_ssot_dispatch_outcome = nil
  outcome ||= Legion::Extensions::Llm::Routing::ProviderOutcome.new(
    kind: :provider_error, reason: e.class.name.to_s
  )
  Legion::LLM::Call::SelectionDispatch::Result.failure(outcome: outcome)
end

#ssot_v3_stream_failover_outcome(error) ⇒ Object

Map a raised streaming provider error to a Phase 1 ProviderOutcome for classification. The exact outcome from SelectionDispatch is preserved losslessly by ssot_v3_direct_dispatch (@last_ssot_dispatch_outcome); fall back to a conservative :provider_error when it is absent.



225
226
227
228
229
230
231
232
233
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 225

def ssot_v3_stream_failover_outcome(error)
  outcome = @last_ssot_dispatch_outcome
  @last_ssot_dispatch_outcome = nil
  return outcome if outcome.is_a?(Legion::Extensions::Llm::Routing::ProviderOutcome)

  Legion::Extensions::Llm::Routing::ProviderOutcome.new(
    kind: :provider_error, reason: error.class.name.to_s
  )
end

#ssot_v3_stream_handle_failure(error:, session:, attempt_context:) ⇒ Object

Classify one streaming provider failure and either continue to the next eligible lane (returns the new AttemptContext) or re-raise (terminal, non-provider, or no replacement). Client-write/disconnect and daemon errors are never provider failures — they re-raise untouched.



192
193
194
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/llm/inference/executor/escalation.rb', line 192

def ssot_v3_stream_handle_failure(error:, session:, attempt_context:)
  raise error if non_provider_failure?(error)

  outcome = ssot_v3_stream_failover_outcome(error)
  action = session.classify(
    dispatch_result: Legion::LLM::Call::SelectionDispatch::Result.failure(outcome: outcome),
    attempt_context: attempt_context
  )
  raise error if action.disposition == :terminal

  # Preserve the existing StreamAssembler failover sequence (§19). The
  # assembler clears its partial canonical buffer and strips provider-bound
  # thinking/reasoning before the next provider renders a clean start.
  @stream_observer&.provider_failover_pending!(from: ssot_v3_stream_lane_hash(attempt_context))

  # The consumed target for the old provider is done; a re-selected lane on
  # a different callable acquires its own lease inside SelectionDispatch.
  release_preflight_lease

  snap = Legion::Extensions::Llm::Inventory::Registry.snapshot
  nxt = session.next_attempt(snapshot: snap)
  raise error if nxt.is_a?(Legion::Extensions::Llm::Routing::Rejection)

  @stream_observer&.begin_dispatch_on(lane: ssot_v3_stream_lane_hash(nxt))
  log.warn "[llm][executor] action=ssot_v3_stream_failover from_kind=#{outcome.kind} " \
           "to_lane=#{nxt.selection.lane_id}"
  nxt
end

#ssot_v3_stream_lane_hash(attempt_context) ⇒ Object

Lane Hash consumed by StreamAssembler (#initial_lane, #begin_dispatch_on, #provider_failover_pending!). Built from the exact Selection identity so the failover debug trailers report real lane IDs — NOT 'unknown:pending'.



125
126
127
128
129
130
131
132
133
134
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 125

def ssot_v3_stream_lane_hash(attempt_context)
  sel = attempt_context.selection
  {
    id:              sel.lane_id,
    provider_family: sel.provider_family,
    instance_id:     sel.instance_id,
    model:           sel.model,
    tier:            attempt_context.lane.tier
  }
end

#ssot_v3_stream_preflight ⇒ Object

SSOT v3 §19 streaming preflight body. Called from Executor#stream_preflight! (before the route opens SSE) once pre-provider steps have built the exact lane through a per-request RoutingSession and acquires that lane's DispatchLease, retaining both on the executor for the subsequent call_stream. A rejection propagates as Errors::RoutingRejected (via next_attempt!) so the route maps it to an HTTP status BEFORE headers — never an SSE server_error. Always selects (single engine) — an empty Registry yields a typed Rejection, never a legacy fallback.



106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 106

def ssot_v3_stream_preflight
  snap = Legion::Extensions::Llm::Inventory::Registry.snapshot
  @stream_session = Legion::LLM::Inference::RoutingSession.new(
    request: @request, requirements: @routing_requirements
  )
  attempt_context = @stream_session.next_attempt!(snapshot: snap)
  populate_ssot_v3_resolved_state(attempt_context)
  @preflight_lease = Legion::Extensions::Llm::Inventory::Registry.acquire(
    callable_handle: attempt_context.selection.callable_handle
  )
  lane = ssot_v3_stream_lane_hash(attempt_context)
  log.info "[llm][executor] action=ssot_v3_stream_preflight_selected lane=#{lane[:id]} " \
           "provider=#{@resolved_provider} model=#{@resolved_model}"
  lane
end

#ssot_v3_stream_session_and_attempt ⇒ Object

Resolve the (session, attempt_context) pair for streaming dispatch. Preflight (§19) already selected + acquired before SSE opened: reuse it. Otherwise select inline here (raises RoutingRejected → old-path fallback).



156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 156

def ssot_v3_stream_session_and_attempt
  return [@stream_session, @current_attempt_context] if @stream_session && @current_attempt_context

  snap = Legion::Extensions::Llm::Inventory::Registry.snapshot
  session = Legion::LLM::Inference::RoutingSession.new(
    request: @request, requirements: @routing_requirements
  )
  attempt_context = session.next_attempt!(snapshot: snap)
  populate_ssot_v3_resolved_state(attempt_context)
  [session, attempt_context]
end

#step_provider_call ⇒ Object

SSOT v3 single engine (sync). There is exactly one selector+executor path: the request-scoped RoutingSession loop. No gate, no legacy selector, no fallback.



16
17
18
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 16

def step_provider_call
  run_provider_call_engine
end

#step_provider_call_stream ⇒ Object

SSOT v3 single engine (streaming). Preflight (Executor#stream_preflight!) already selected + acquired the exact lane before SSE opened; this runs the dispatch + post-first-byte failover through the same RoutingSession and consumed-attempt set. Provider failures are classified inside the failover loop (no HealthTracker mutation); terminal/exhausted outcomes re-raise so the route emits the dialect terminal SSE error.



325
326
327
 
# File 'lib/legion/llm/inference/executor/escalation.rb', line 325

def step_provider_call_stream(&)
  run_provider_call_ssot_v3_stream(&)
end